diff --git a/lib/crewai/src/crewai/crew.py b/lib/crewai/src/crewai/crew.py index fda456be40..3ce54b1b81 100644 --- a/lib/crewai/src/crewai/crew.py +++ b/lib/crewai/src/crewai/crew.py @@ -117,7 +117,11 @@ def get_supported_content_types(provider: str, api: str | None = None) -> list[s from crewai.tools.agent_tools.read_file_tool import ReadFileTool from crewai.tools.base_tool import BaseTool from crewai.types.callback import SerializableCallable -from crewai.types.streaming import CrewStreamingOutput +from crewai.types.streaming import ( + AsyncStreamSession, + CrewStreamingOutput, + StreamSession, +) from crewai.types.usage_metrics import UsageMetrics from crewai.utilities.constants import NOT_SPECIFIED, TRAINING_DATA_FILE from crewai.utilities.crew.models import CrewContext @@ -137,7 +141,10 @@ def get_supported_content_types(provider: str, api: str | None = None) -> list[s from crewai.utilities.rpm_controller import RPMController from crewai.utilities.streaming import ( create_async_chunk_generator, + create_async_frame_generator, create_chunk_generator, + create_frame_generator, + create_frame_streaming_state, register_cleanup, signal_end, signal_error, @@ -293,6 +300,17 @@ class Crew(FlowTrackable, BaseModel): default=False, description="Whether to stream output from the crew execution.", ) + stream_frames: bool = Field( + default=False, + description=( + "When True, kickoff yields StreamFrame events covering the full " + "execution (LLM tokens, tool calls, agent/task lifecycle) instead " + "of StreamChunk token chunks. Implies stream=True (streaming is " + "auto-enabled if not set). Returns a StreamSession / " + "AsyncStreamSession whose .subscribe() allows filtering by channel " + "(llm, tools, messages, flow, lifecycle)." + ), + ) max_rpm: int | None = Field( default=None, description=( @@ -692,6 +710,18 @@ def create_crew_knowledge(self) -> Crew: ) return self + @model_validator(mode="after") + def enforce_stream_frames_contract(self) -> Self: + """Ensure ``stream_frames`` implies ``stream``. + + ``stream_frames`` only takes effect when ``stream`` is True. Rather + than silently ignoring ``stream_frames=True`` with ``stream=False``, + enable streaming so the frame output is produced as documented. + """ + if self.stream_frames and not self.stream: + self.stream = True + return self + @model_validator(mode="after") def check_manager_llm(self) -> Self: """Validates that the language model is set when using hierarchical process.""" @@ -968,7 +998,7 @@ def kickoff( inputs: dict[str, Any] | None = None, input_files: dict[str, FileInput] | None = None, from_checkpoint: CheckpointConfig | None = None, - ) -> CrewOutput | CrewStreamingOutput: + ) -> CrewOutput | CrewStreamingOutput | StreamSession[CrewOutput]: """Execute the crew's workflow. Args: @@ -987,6 +1017,8 @@ def kickoff( get_env_context() if self.stream: enable_agent_streaming(self.agents) + if self.stream_frames: + return self._kickoff_frame_stream(inputs, input_files) ctx = StreamingContext() def run_crew() -> None: @@ -1098,7 +1130,7 @@ async def kickoff_async( inputs: dict[str, Any] | None = None, input_files: dict[str, FileInput] | None = None, from_checkpoint: CheckpointConfig | None = None, - ) -> CrewOutput | CrewStreamingOutput: + ) -> CrewOutput | CrewStreamingOutput | AsyncStreamSession[CrewOutput]: """Asynchronous kickoff method to start the crew execution. Args: @@ -1121,6 +1153,8 @@ async def kickoff_async( if self.stream: enable_agent_streaming(self.agents) + if self.stream_frames: + return await self._akickoff_frame_stream(inputs, input_files) ctx = StreamingContext(use_async=True) async def run_crew() -> None: @@ -1147,6 +1181,83 @@ async def run_crew() -> None: return await asyncio.to_thread(self.kickoff, inputs, input_files) + def _kickoff_frame_stream( + self, + inputs: dict[str, Any] | None, + input_files: dict[str, FileInput] | None, + ) -> StreamSession[CrewOutput]: + """Run kickoff and stream full execution frames. + + Yields ``StreamFrame`` events for the entire crew run (LLM tokens, + tool calls, agent/task lifecycle) via the existing frame streaming + pipeline, instead of the default ``StreamChunk`` token-only stream. + Consumers iterate the returned ``StreamSession`` (or filter with + ``.subscribe(channels=...)``) and access the final ``CrewOutput`` via + ``.result`` after exhaustion. Requires ``stream=True``; ``stream_frames`` + selects this path over chunk streaming. + + Args: + inputs: Optional input dictionary for task interpolation. + input_files: Optional dict of named file inputs for the crew. + + Returns: + A ``StreamSession`` yielding ``StreamFrame`` events. + """ + result_holder: list[CrewOutput] = [] + state = create_frame_streaming_state(result_holder, use_async=False) + output_holder: list[StreamSession[Any]] = [] + + def run_crew() -> CrewOutput | None: + self.stream = False + try: + crew_result = self.kickoff(inputs=inputs, input_files=input_files) + return crew_result if isinstance(crew_result, CrewOutput) else None + finally: + self.stream = True + + stream_session: StreamSession[CrewOutput] = StreamSession( + sync_iterator=create_frame_generator(state, run_crew, output_holder) + ) + output_holder.append(stream_session) + return stream_session + + async def _akickoff_frame_stream( + self, + inputs: dict[str, Any] | None, + input_files: dict[str, FileInput] | None, + ) -> AsyncStreamSession[CrewOutput]: + """Run kickoff asynchronously and stream full execution frames. + + Async counterpart of ``_kickoff_frame_stream``. Yields ``StreamFrame`` + events via the existing async frame streaming pipeline, instead of the + default ``StreamChunk`` token-only stream. Requires ``stream=True``; + ``stream_frames`` selects this path over chunk streaming. + + Args: + inputs: Optional input dictionary for task interpolation. + input_files: Optional dict of named file inputs for the crew. + + Returns: + An ``AsyncStreamSession`` yielding ``StreamFrame`` events. + """ + result_holder: list[CrewOutput] = [] + state = create_frame_streaming_state(result_holder, use_async=True) + output_holder: list[AsyncStreamSession[Any]] = [] + + async def run_crew() -> CrewOutput | None: + self.stream = False + try: + result = await asyncio.to_thread(self.kickoff, inputs, input_files) + return result if isinstance(result, CrewOutput) else None + finally: + self.stream = True + + stream_session: AsyncStreamSession[CrewOutput] = AsyncStreamSession( + async_iterator=create_async_frame_generator(state, run_crew, output_holder) + ) + output_holder.append(stream_session) + return stream_session + async def kickoff_for_each_async( self, inputs: list[dict[str, Any]], diff --git a/lib/crewai/tests/test_streaming.py b/lib/crewai/tests/test_streaming.py index e3d84f4c99..3c3ad5a65a 100644 --- a/lib/crewai/tests/test_streaming.py +++ b/lib/crewai/tests/test_streaming.py @@ -10,6 +10,7 @@ from crewai import Agent, Crew, Task from crewai.events.event_bus import crewai_event_bus from crewai.events.types.llm_events import LLMStreamChunkEvent, ToolCall, FunctionCall +from crewai.events.types.tool_usage_events import ToolUsageStartedEvent from crewai.flow.flow import Flow, start from crewai.state.checkpoint_config import CheckpointConfig from crewai.types.streaming import ( @@ -1011,3 +1012,168 @@ def emit_chunks(prefix: str, call_id: str) -> None: _unregister_handler(state_a.handler) _unregister_handler(state_b.handler) + + +# --------------------------------------------------------------------------- +# Frame-streaming (stream_frames) tests +# --------------------------------------------------------------------------- + + +class TestCrewFrameStreaming: + """Tests for Crew-level frame streaming (stream_frames=True).""" + + @staticmethod + def _fake_kickoff_returning( + output: Any, + events: list | None = None, + ) -> Any: + """Factory: returns a fake ``kickoff(self, ...)`` that emits *events* + on the crew bus and returns *output*.""" + + def fake_kickoff( + _self: Crew, + inputs: dict[str, Any] | None = None, + input_files: dict[str, Any] | None = None, + from_checkpoint: CheckpointConfig | None = None, + ) -> Any: + if events: + for event in events: + crewai_event_bus.emit(_self, event) + return output + + return fake_kickoff + + def test_frame_stream_yields_mixed_events( + self, researcher: Agent, simple_task: Task + ) -> None: + """When stream_frames=True, the StreamSession includes non-LLM events.""" + from crewai.crews.crew_output import CrewOutput + + crew = Crew( + agents=[researcher], + tasks=[simple_task], + verbose=False, + stream=True, + stream_frames=True, + ) + fake_output = MagicMock(spec=CrewOutput) + fake_output.raw = "mock output" + + fake_kickoff = self._fake_kickoff_returning( + fake_output, + events=[ + LLMStreamChunkEvent( + type="llm_stream_chunk", chunk="token", call_id="c-1" + ), + ToolUsageStartedEvent( + type="tool_usage_started", + tool_name="search", + tool_args={"q": "test"}, + ), + ], + ) + + with patch.object(Crew, "kickoff", fake_kickoff): + session = crew._kickoff_frame_stream(None, None) + frames = list(session) + + assert len(frames) >= 2, f"expected >=2 frames, got {len(frames)}" + + channels = {f.channel for f in frames} + assert "llm" in channels, "LLM chunks should produce 'llm' channel frames" + assert "tools" in channels, ( + "tool events should produce 'tools' channel frames — " + "this proves the frame path includes non-LLM events that the chunk " + "path drops" + ) + + assert session.result is fake_output + + def test_frame_stream_result_accessible_after_exhaustion( + self, researcher: Agent, simple_task: Task + ) -> None: + """StreamSession.result returns the CrewOutput after all frames are consumed.""" + from crewai.crews.crew_output import CrewOutput + + crew = Crew( + agents=[researcher], + tasks=[simple_task], + verbose=False, + stream=True, + stream_frames=True, + ) + fake_output = MagicMock(spec=CrewOutput) + + with patch.object(Crew, "kickoff", self._fake_kickoff_returning(fake_output)): + session = crew._kickoff_frame_stream(None, None) + list(session) + assert session.result is fake_output + assert session.is_completed + + def test_stream_frames_implies_stream( + self, researcher: Agent, simple_task: Task + ) -> None: + """stream_frames=True auto-enables stream=True (no silent ignore).""" + crew = Crew( + agents=[researcher], + tasks=[simple_task], + verbose=False, + stream_frames=True, + ) + assert crew.stream is True + assert crew.stream_frames is True + + def test_stream_frames_false_default_unchanged( + self, researcher: Agent, simple_task: Task + ) -> None: + """stream_frames=False (default) — the Crew field is False out of the box.""" + crew = Crew( + agents=[researcher], + tasks=[simple_task], + verbose=False, + stream=True, + ) + assert not crew.stream_frames + # The chunk path is still available (verified via existing tests). + + @pytest.mark.asyncio + async def test_async_frame_stream_yields_tool_events( + self, researcher: Agent, simple_task: Task + ) -> None: + """Async frame streaming via _akickoff_frame_stream yields tool frames.""" + from crewai.crews.crew_output import CrewOutput + + crew = Crew( + agents=[researcher], + tasks=[simple_task], + verbose=False, + stream=True, + stream_frames=True, + ) + fake_output = MagicMock(spec=CrewOutput) + fake_output.raw = "mock async" + + fake_kickoff = self._fake_kickoff_returning( + fake_output, + events=[ + LLMStreamChunkEvent( + type="llm_stream_chunk", + chunk="async-token", + call_id="c-async", + ), + ToolUsageStartedEvent( + type="tool_usage_started", + tool_name="fetch", + tool_args={"url": "/api"}, + ), + ], + ) + + with patch.object(Crew, "kickoff", fake_kickoff): + session = await crew._akickoff_frame_stream(None, None) + frames = [f async for f in session] + + assert len(frames) >= 2 + channels = {f.channel for f in frames} + assert "tools" in channels + assert session.result is fake_output