主题
消费流
打印 LLM 分块，观察工具事件，并从 CrewAI 流中读取最终结果。
概述
当你想要订阅 CrewAI 流并在帧到达时打印或路由帧时，请使用本指南。
基本模式是：
python
stream = flow.stream_events(inputs={"topic": "AI agents"})
with stream:
for frame in stream:
...
result = stream.result始终在读取 stream.result 之前消费流。
打印 LLM 输出
如果你只关心 LLM 调用生成的文本，请订阅 llm 投影并打印 frame.content：
python
stream = flow.stream_events(inputs={"topic": "AI agents"})
with stream:
for frame in stream.llm:
print(frame.content, end="", flush=True)
print()
result = stream.resultframe.content 对于不携带可打印文本的帧是空字符串，因此这样也是安全的：
python
with flow.stream_events(inputs={"topic": "AI agents"}) as stream:
for frame in stream.events:
if frame.channel == "llm" and frame.content:
print(frame.content, end="", flush=True)
result = stream.result打印工具活动
工具事件到达 tools 通道。使用 frame.type 区分开始、完成和错误。
python
with flow.stream_events(inputs={"topic": "AI agents"}) as stream:
for frame in stream.events:
if frame.channel == "llm" and frame.content:
print(frame.content, end="", flush=True)
if frame.channel == "tools" and frame.type == "tool_usage_started":
print(f"\nTool started: {frame.event.get('tool_name')}")
if frame.channel == "tools" and frame.type == "tool_usage_finished":
print(f"\nTool finished: {frame.event.get('tool_name')}")
result = stream.resultframe.event 是源事件的结构化载荷。使用它获取元数据，例如工具名称、参数、消息角色和运行时标识符。
观察流程进度
流程生命周期和方法执行帧到达 flow 通道：
python
with flow.stream_events(inputs={"topic": "AI agents"}) as stream:
for frame in stream.flow:
print(frame.type, frame.namespace)
result = stream.result当你想要进度日志而不是 token 级输出时，请使用此方法。
交错选定通道
当你想要一个通道子集并保留它们的相对顺序时，请使用 interleave()：
python
with flow.stream_events(inputs={"topic": "AI agents"}) as stream:
for frame in stream.interleave(["llm", "tools"]):
if frame.channel == "llm":
print(frame.content, end="", flush=True)
elif frame.type == "tool_usage_started":
print(f"\nTool: {frame.event.get('tool_name')}")
result = stream.result流式输出直接 LLM 调用
直接调用 llm.call(...) 会返回最终组装结果。要流式传输直接的 LLM 调用，请使用 llm.stream_events(...)：
python
from crewai import LLM
llm = LLM(model="gpt-4o-mini")
stream = llm.stream_events("Explain streaming in one sentence.")
with stream:
for frame in stream.llm:
print(frame.content, end="", flush=True)
print()
result = stream.result流式传输一次对话轮次
对话流程为单条用户消息公开 stream_turn()：
python
stream = flow.stream_turn(
"What can you help me with?",
session_id="session-1",
)
with stream:
for frame in stream.interleave(["llm", "messages"]):
if frame.channel == "llm":
print(frame.content, end="", flush=True)
elif frame.channel == "messages":
print(f"\n{frame.event.get('role')}: {frame.event.get('content')}")
reply = stream.result异步消费者
异步流使用相同的通道投影：
python
stream = flow.astream(inputs={"topic": "AI agents"})
async with stream:
async for frame in stream.llm:
print(frame.content, end="", flush=True)
result = stream.result清理
尽可能将流用作上下文管理器。如果客户端断开连接，或者你提前停止消费，请关闭流：
python
stream = flow.stream_events(inputs={"topic": "AI agents"})
try:
for frame in stream.events:
print(frame.content, end="", flush=True)
finally:
if not stream.is_exhausted:
stream.close()对于异步流，请调用 await stream.aclose()。