Crew 流式执行

简介

CrewAI 提供了在 crew 执行期间流式输出实时结果的能力，让你无需等待整个流程完成，就可以在结果生成时直接显示。这一功能对于构建交互式应用、向用户提供反馈以及监控长时间运行的流程尤其有用。

流式工作原理

启用流式输出后，CrewAI 会在 LLM 响应和工具调用发生时将其捕获，并将其打包为结构化块，其中包含当前正在执行的任务和代理的上下文。你可以实时遍历这些块，并在执行完成后访问最终结果。

启用流式输出

要启用流式输出，请在创建 crew 时将 stream 参数设为 True：

from crewai import Agent, Crew, Task

# 创建你的代理和任务
researcher = Agent(
    role="Research Analyst",
    goal="Gather comprehensive information on topics",
    backstory="You are an experienced researcher with excellent analytical skills.",
)

task = Task(
    description="Research the latest developments in AI",
    expected_output="A detailed report on recent AI advancements",
    agent=researcher,
)

# 启用流式输出
crew = Crew(
    agents=[researcher],
    tasks=[task],
    stream=True  # Enable streaming output
)

同步流式输出

当你在启用了流式输出的 crew 上调用 kickoff() 时，它会返回一个 CrewStreamingOutput 对象，你可以遍历它以接收逐步到达的块：

# 开始流式执行
streaming = crew.kickoff(inputs={"topic": "artificial intelligence"})

# 逐块遍历并处理
for chunk in streaming:
    print(chunk.content, end="", flush=True)

# 流式结束后访问最终结果
result = streaming.result
print(f"\n\nFinal output: {result.raw}")

流块信息

每个块都会为执行过程提供丰富上下文：

streaming = crew.kickoff(inputs={"topic": "AI"})

for chunk in streaming:
    print(f"Task: {chunk.task_name} (index {chunk.task_index})")
    print(f"Agent: {chunk.agent_role}")
    print(f"Content: {chunk.content}")
    print(f"Type: {chunk.chunk_type}")  # TEXT or TOOL_CALL
    if chunk.tool_call:
        print(f"Tool: {chunk.tool_call.tool_name}")
        print(f"Arguments: {chunk.tool_call.arguments}")

访问流式结果

CrewStreamingOutput 对象提供了几个有用的属性：

streaming = crew.kickoff(inputs={"topic": "AI"})

# 遍历并收集块
for chunk in streaming:
    print(chunk.content, end="", flush=True)

# 遍历完成后
print(f"\nCompleted: {streaming.is_completed}")
print(f"Full text: {streaming.get_full_text()}")
print(f"All chunks: {len(streaming.chunks)}")
print(f"Final result: {streaming.result.raw}")

异步流式输出

对于异步应用，你可以在 async 迭代中使用 akickoff()（原生异步）或 kickoff_async()（基于线程）：

使用 akickoff() 的原生异步

akickoff() 方法为整个链路提供真正的原生异步执行：

import asyncio

async def stream_crew():
    crew = Crew(
        agents=[researcher],
        tasks=[task],
        stream=True
    )

    # 开始原生异步流式执行
    streaming = await crew.akickoff(inputs={"topic": "AI"})

    # 对块进行异步迭代
    async for chunk in streaming:
        print(chunk.content, end="", flush=True)

    # 访问最终结果
    result = streaming.result
    print(f"\n\nFinal output: {result.raw}")

asyncio.run(stream_crew())

使用 kickoff_async() 的基于线程异步

适用于更简单的异步集成或向后兼容：

import asyncio

async def stream_crew():
    crew = Crew(
        agents=[researcher],
        tasks=[task],
        stream=True
    )

    # 开始基于线程的异步流式执行
    streaming = await crew.kickoff_async(inputs={"topic": "AI"})

    # 对块进行异步迭代
    async for chunk in streaming:
        print(chunk.content, end="", flush=True)

    # 访问最终结果
    result = streaming.result
    print(f"\n\nFinal output: {result.raw}")

asyncio.run(stream_crew())

Note

对于高并发工作负载，建议使用 akickoff()，因为它在任务执行、记忆操作和知识检索方面都使用原生异步。更多细节请参见 异步启动 Crew 指南。

配合 kickoff_for_each 使用流式输出

当使用 kickoff_for_each() 为多个输入执行 crew 时，流式输出的行为会因同步或异步而不同：

同步 kickoff_for_each

使用同步的 kickoff_for_each() 时，你会得到一个 CrewStreamingOutput 列表，每个输入对应一个对象：

crew = Crew(
    agents=[researcher],
    tasks=[task],
    stream=True
)

inputs_list = [
    {"topic": "AI in healthcare"},
    {"topic": "AI in finance"}
]

# 返回流式输出列表
streaming_outputs = crew.kickoff_for_each(inputs=inputs_list)

# 遍历每个流式输出
for i, streaming in enumerate(streaming_outputs):
    print(f"\n=== Input {i + 1} ===")
    for chunk in streaming:
        print(chunk.content, end="", flush=True)

    result = streaming.result
    print(f"\n\nResult {i + 1}: {result.raw}")

异步 kickoff_for_each_async

使用异步 kickoff_for_each_async() 时，你会得到一个单独的 CrewStreamingOutput，它会随着各个 crew 生成内容而并发地产生来自所有 crew 的块：

import asyncio

async def stream_multiple_crews():
    crew = Crew(
        agents=[researcher],
        tasks=[task],
        stream=True
    )

    inputs_list = [
        {"topic": "AI in healthcare"},
        {"topic": "AI in finance"}
    ]

    # 返回单个流式输出，覆盖所有 crew
    streaming = await crew.kickoff_for_each_async(inputs=inputs_list)

    # 来自所有 crew 的块会在生成时陆续到达
    async for chunk in streaming:
        print(f"[{chunk.task_name}] {chunk.content}", end="", flush=True)

    # 访问所有结果
    results = streaming.results  # List of CrewOutput objects
    for i, result in enumerate(results):
        print(f"\n\nResult {i + 1}: {result.raw}")

asyncio.run(stream_multiple_crews())

流块类型

块可以有不同类型，由 chunk_type 字段标识：

TEXT 块

来自 LLM 响应的标准文本内容：

for chunk in streaming:
    if chunk.chunk_type == StreamChunkType.TEXT:
        print(chunk.content, end="", flush=True)

TOOL_CALL 块

关于正在进行的工具调用的信息：

for chunk in streaming:
    if chunk.chunk_type == StreamChunkType.TOOL_CALL:
        print(f"\nCalling tool: {chunk.tool_call.tool_name}")
        print(f"Arguments: {chunk.tool_call.arguments}")

实用示例：构建带流式输出的 UI

下面是一个完整示例，展示如何使用流式输出构建交互式应用：

import asyncio
from crewai import Agent, Crew, Task
from crewai.types.streaming import StreamChunkType

async def interactive_research():
    # Create crew with streaming enabled
    researcher = Agent(
        role="Research Analyst",
        goal="Provide detailed analysis on any topic",
        backstory="You are an expert researcher with broad knowledge.",
    )

    task = Task(
        description="Research and analyze: {topic}",
        expected_output="A comprehensive analysis with key insights",
        agent=researcher,
    )

    crew = Crew(
        agents=[researcher],
        tasks=[task],
        stream=True,
        verbose=False
    )

    # Get user input
    topic = input("Enter a topic to research: ")

    print(f"\n{'='*60}")
    print(f"Researching: {topic}")
    print(f"{'='*60}\n")

    # Start streaming execution
    streaming = await crew.kickoff_async(inputs={"topic": topic})

    current_task = ""
    async for chunk in streaming:
        # Show task transitions
        if chunk.task_name != current_task:
            current_task = chunk.task_name
            print(f"\n[{chunk.agent_role}] Working on: {chunk.task_name}")
            print("-" * 60)

        # Display text chunks
        if chunk.chunk_type == StreamChunkType.TEXT:
            print(chunk.content, end="", flush=True)

        # Display tool calls
        elif chunk.chunk_type == StreamChunkType.TOOL_CALL and chunk.tool_call:
            print(f"\n🔧 Using tool: {chunk.tool_call.tool_name}")

    # Show final result
    result = streaming.result
    print(f"\n\n{'='*60}")
    print("Analysis Complete!")
    print(f"{'='*60}")
    print(f"\nToken Usage: {result.token_usage}")

asyncio.run(interactive_research())

使用场景

流式输出在以下场景中特别有价值：

• 交互式应用：在代理工作时向用户提供实时反馈
• 长时间运行的任务：展示研究、分析或内容生成的进度
• 调试与监控：实时观察代理行为和决策过程
• 用户体验：通过逐步结果降低感知延迟
• 实时仪表板：构建显示 crew 执行状态的监控界面

取消与资源清理

CrewStreamingOutput 支持优雅取消，因此当消费者断开连接时，进行中的工作会及时停止。

异步上下文管理器

streaming = await crew.akickoff(inputs={"topic": "AI"})

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

显式取消

streaming = await crew.akickoff(inputs={"topic": "AI"})
try:
    async for chunk in streaming:
        print(chunk.content, end="", flush=True)
finally:
    await streaming.aclose()  # async
    # streaming.close()       # sync equivalent

取消后，streaming.is_cancelled 和 streaming.is_completed 都会变为 True。aclose() 和 close() 都是幂等的。

重要说明

• 流式输出会自动为 crew 中的所有代理启用 LLM 流式输出
• 在访问 .result 属性之前，你必须先遍历完所有块
• 对于带流式输出的 kickoff_for_each_async()，请使用 .results（复数）来获取所有输出
• 流式输出的额外开销很小，甚至可以提升感知性能
• 每个块都包含完整上下文（任务、代理、块类型），适合丰富的 UI

错误处理

处理流式执行期间的错误：

streaming = crew.kickoff(inputs={"topic": "AI"})

try:
    for chunk in streaming:
        print(chunk.content, end="", flush=True)

    result = streaming.result
    print(f"\nSuccess: {result.raw}")

except Exception as e:
    print(f"\nError during streaming: {e}")
    if streaming.is_completed:
        print("Streaming completed but an error occurred")

通过利用流式输出，你可以使用 CrewAI 构建更具响应性和交互性的应用，让用户实时看到代理的执行过程和结果。