流式传输运行时契约

从流程、直接 LLM 调用和对话轮次流式输出有序的运行时帧。

概述

CrewAI 提供了一种基于帧的流式传输契约，适用于需要的不只是纯文本块的运行时。该契约会为流程生命周期事件、直接 LLM token、工具活动、对话消息和自定义事件发出有序的 StreamFrame 对象。

当你正在构建 UI、服务桥接、终端应用或部署运行时，需要在流程、对话轮次或直接 LLM 调用运行期间获得稳定的结构化事件流时，请使用此 API。

StreamFrame

每个帧都使用相同的外层结构：

from crewai.types.streaming import StreamFrame

frame.id           # unique frame id
frame.seq          # execution-local order, when available
frame.type         # source event type, such as "flow_started"
frame.channel      # "llm", "flow", "tools", "messages", "lifecycle", or "custom"
frame.namespace    # source/runtime namespace
frame.timestamp    # event timestamp
frame.parent_id    # parent event id, when available
frame.previous_id  # previous event id, when available
frame.data         # event payload
frame.event        # alias for frame.data
frame.content      # printable text for token-like frames, otherwise ""

channel 字段是消费者路由帧最快的方式：

通道	包含内容
llm	来自 LLM 流式事件的 token 和思考块
flow	流程生命周期、方法执行、路由以及暂停/恢复事件
tools	工具使用事件
messages	对话记录事件
lifecycle	不属于其他通道的运行时生命周期事件
custom	无法映射到内置通道的事件

frame.type 保留源事件类型，因此消费者可以处理某个通道内的特定事件。

流式输出流程

在 Flow 上设置 stream=True，让 kickoff() 返回一个流式会话：

from crewai.flow import Flow, start

class ReportFlow(Flow):
    @start()
    def generate(self):
        return "done"

flow = ReportFlow(stream=True)
stream = flow.kickoff()

with stream:
    for chunk in stream:
        print(chunk.content, end="", flush=True)
        if chunk.type == "tool_usage_started":
            print(chunk.event["tool_name"])

result = stream.result

你必须先消费完流，再读取 stream.result。提前访问结果会引发 RuntimeError，以免消费者误把未完成的运行当作已完成。

如果你只想为单次调用启用流式输出，而不想在 Flow 实例上设置 stream=True，也可以直接调用 flow.stream_events(...)。

按通道过滤

StreamSession 提供通道投影，在所选通道内保持全局帧顺序：

stream = flow.stream_events()

with stream:
    for frame in stream.llm:
        print(frame.content, end="", flush=True)

result = stream.result

可用的投影包括：

投影	帧
stream.events	所有帧
stream.llm	LLM 帧
stream.messages	对话消息帧
stream.flow	流程帧
stream.tools	工具帧
stream.interleave([...])	选定的一组通道

当消费者只需要部分通道但仍需保持它们的相对顺序时，使用 stream.interleave(["flow", "llm", "messages"])。

异步流式传输

异步消费者请使用 astream()：

flow = ReportFlow()
stream = flow.astream()

async with stream:
    async for chunk in stream.events:
        print(chunk.channel, chunk.type, chunk.content)

result = stream.result

异步会话拥有与同步会话相同的投影。

流式输出直接 LLM 调用

llm.call(...) 仍返回最终组装后的结果。当你希望在各分块到达时逐个迭代，同时保留结构化事件载荷，请使用 llm.stream_events(...)：

from crewai import LLM

llm = LLM(model="gpt-4o-mini")
stream = llm.stream_events(
    messages=[
        {
            "role": "user",
            "content": "Explain CrewAI streaming in two short sentences.",
        }
    ]
)

with stream:
    for chunk in stream:
        print(chunk.content, end="", flush=True)

result = stream.result

llm.stream_events(...) 会临时为被包装的调用启用流式传输，并在之后恢复 LLM 先前的 stream 设置。各服务商集成会继续发出底层 LLM 流事件；该辅助函数为所有 LLM 服务商提供基于这些事件的统一迭代器 API。

对话轮次

Conversational Flows 可以使用 stream_turn() 流式输出单个用户轮次：

from crewai import Flow
from crewai.flow import ConversationConfig, ConversationState

@ConversationConfig(llm="gpt-4o-mini", defer_trace_finalization=True)
class ChatFlow(Flow[ConversationState]):
    conversational = True

flow = ChatFlow()
stream = flow.stream_turn("What can you help me with?", session_id="session-1")

with stream:
    for frame in stream.events:
        if frame.channel == "llm" and frame.type == "llm_stream_chunk":
            print(frame.content, end="", flush=True)

reply = stream.result

在 stream_turn() 期间，内置的对话回答路径会为该轮次启用 LLM token 流式传输，并在之后恢复 LLM 先前的 stream 设置。创建自有智能体或 LLM 实例的自定义路由处理器如需 token 级输出，应自行将这些 LLM 配置为流式传输。

清理

尽可能将会话用作上下文管理器。如果客户端在流被耗尽前断开连接，请显式关闭会话：

stream = flow.stream_events()

try:
    for frame in stream.events:
        print(frame.type)
finally:
    if not stream.is_exhausted:
        stream.close()

对于异步流，使用 await stream.aclose()。

旧版分块流式传输

使用 stream=True 的团队流式传输仍返回面向分块的 CrewStreamingOutput API，如 流式团队执行 所述。直接的 llm.call(...) 仍返回最终 LLM 结果。帧契约面向需要在流程、直接 LLM 调用、对话轮次、工具和消息之间获得稳定事件封装的运行时。