Flow 流式执行

简介

CrewAI Flows 支持流式输出，让你在 flow 执行过程中接收实时更新。这一功能使你能够构建响应迅速的应用，逐步展示结果、提供实时进度更新，并为长时间运行的工作流带来更好的用户体验。

Flow 流式工作原理

当 Flow 启用流式输出后，CrewAI 会捕获并流式传输 flow 中任意 crew 或 LLM 调用的输出。随着执行推进，流会传递包含内容、任务上下文和代理信息的结构化块。

启用流式输出

要启用流式输出，请在 Flow 类上将 stream 属性设为 True：

from crewai.flow.flow import Flow, listen, start
from crewai import Agent, Crew, Task

class ResearchFlow(Flow):
    stream = True  # 为整个 flow 启用流式输出

    @start()
    def initialize(self):
        return {"topic": "AI trends"}

    @listen(initialize)
    def research_topic(self, data):
        researcher = Agent(
            role="Research Analyst",
            goal="Research topics thoroughly",
            backstory="Expert researcher with analytical skills",
        )

        task = Task(
            description="Research {topic} and provide insights",
            expected_output="Detailed research findings",
            agent=researcher,
        )

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

        return crew.kickoff(inputs=data)

同步流式输出

当你在启用了流式输出的 flow 上调用 kickoff() 时，它会返回一个可以遍历的 FlowStreamingOutput 对象：

flow = ResearchFlow()

# 开始流式执行
streaming = flow.kickoff()

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

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

流块信息

每个块都会提供其在 flow 中来源的上下文：

streaming = flow.kickoff()

for chunk in streaming:
    print(f"Agent: {chunk.agent_role}")
    print(f"Task: {chunk.task_name}")
    print(f"Content: {chunk.content}")
    print(f"Type: {chunk.chunk_type}")  # TEXT or TOOL_CALL

访问流式属性

FlowStreamingOutput 对象提供有用的属性和方法：

streaming = flow.kickoff()

# 遍历并收集块
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"Total chunks: {len(streaming.chunks)}")
print(f"Final result: {streaming.result}")

异步流式输出

对于异步应用，请使用 kickoff_async() 并配合异步迭代：

import asyncio

async def stream_flow():
    flow = ResearchFlow()

    # 开始异步流式执行
    streaming = await flow.kickoff_async()

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

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

asyncio.run(stream_flow())

在多步骤 Flow 中使用流式输出

流式输出可以无缝贯穿多个 flow 步骤，包括执行多个 crew 的 flow：

from crewai.flow.flow import Flow, listen, start
from crewai import Agent, Crew, Task

class MultiStepFlow(Flow):
    stream = True

    @start()
    def research_phase(self):
        """First crew: Research the topic."""
        researcher = Agent(
            role="Research Analyst",
            goal="Gather comprehensive information",
            backstory="Expert at finding relevant information",
        )

        task = Task(
            description="Research AI developments in healthcare",
            expected_output="Research findings on AI in healthcare",
            agent=researcher,
        )

        crew = Crew(agents=[researcher], tasks=[task])
        result = crew.kickoff()

        self.state["research"] = result.raw
        return result.raw

    @listen(research_phase)
    def analysis_phase(self, research_data):
        """Second crew: Analyze the research."""
        analyst = Agent(
            role="Data Analyst",
            goal="Analyze information and extract insights",
            backstory="Expert at identifying patterns and trends",
        )

        task = Task(
            description="Analyze this research: {research}",
            expected_output="Key insights and trends",
            agent=analyst,
        )

        crew = Crew(agents=[analyst], tasks=[task])
        return crew.kickoff(inputs={"research": research_data})


# 贯穿两个阶段的流式输出
flow = MultiStepFlow()
streaming = flow.kickoff()

current_step = ""
for chunk in streaming:
    # 追踪当前正在执行的 flow 步骤
    if chunk.task_name != current_step:
        current_step = chunk.task_name
        print(f"\n\n=== {chunk.task_name} ===\n")

    print(chunk.content, end="", flush=True)

result = streaming.result
print(f"\n\nFinal analysis: {result}")

实用示例：进度仪表板

下面是一个完整示例，展示如何使用流式输出构建进度仪表板：

import asyncio
from crewai.flow.flow import Flow, listen, start
from crewai import Agent, Crew, Task
from crewai.types.streaming import StreamChunkType

class ResearchPipeline(Flow):
    stream = True

    @start()
    def gather_data(self):
        researcher = Agent(
            role="Data Gatherer",
            goal="Collect relevant information",
            backstory="Skilled at finding quality sources",
        )

        task = Task(
            description="Gather data on renewable energy trends",
            expected_output="Collection of relevant data points",
            agent=researcher,
        )

        crew = Crew(agents=[researcher], tasks=[task])
        result = crew.kickoff()
        self.state["data"] = result.raw
        return result.raw

    @listen(gather_data)
    def analyze_data(self, data):
        analyst = Agent(
            role="Data Analyst",
            goal="Extract meaningful insights",
            backstory="Expert at data analysis",
        )

        task = Task(
            description="Analyze: {data}",
            expected_output="Key insights and trends",
            agent=analyst,
        )

        crew = Crew(agents=[analyst], tasks=[task])
        return crew.kickoff(inputs={"data": data})


async def run_with_dashboard():
    flow = ResearchPipeline()

    print("="*60)
    print("RESEARCH PIPELINE DASHBOARD")
    print("="*60)

    streaming = await flow.kickoff_async()

    current_agent = ""
    current_task = ""
    chunk_count = 0

    async for chunk in streaming:
        chunk_count += 1

        # Display phase transitions
        if chunk.task_name != current_task:
            current_task = chunk.task_name
            current_agent = chunk.agent_role
            print(f"\n\n📋 Phase: {current_task}")
            print(f"👤 Agent: {current_agent}")
            print("-" * 60)

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

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

    # Show completion summary
    result = streaming.result
    print(f"\n\n{'='*60}")
    print("PIPELINE COMPLETE")
    print(f"{'='*60}")
    print(f"Total chunks: {chunk_count}")
    print(f"Final output length: {len(str(result))} characters")

asyncio.run(run_with_dashboard())

带状态管理的流式输出

流式输出与 Flow 的状态管理自然配合：

from pydantic import BaseModel

class AnalysisState(BaseModel):
    topic: str = ""
    research: str = ""
    insights: str = ""

class StatefulStreamingFlow(Flow[AnalysisState]):
    stream = True

    @start()
    def research(self):
        # 流式执行期间可以访问状态
        topic = self.state.topic
        print(f"Researching: {topic}")

        researcher = Agent(
            role="Researcher",
            goal="Research topics thoroughly",
            backstory="Expert researcher",
        )

        task = Task(
            description=f"Research {topic}",
            expected_output="Research findings",
            agent=researcher,
        )

        crew = Crew(agents=[researcher], tasks=[task])
        result = crew.kickoff()

        self.state.research = result.raw
        return result.raw

    @listen(research)
    def analyze(self, research):
        # 访问更新后的状态
        print(f"Analyzing {len(self.state.research)} chars of research")

        analyst = Agent(
            role="Analyst",
            goal="Extract insights",
            backstory="Expert analyst",
        )

        task = Task(
            description="Analyze: {research}",
            expected_output="Key insights",
            agent=analyst,
        )

        crew = Crew(agents=[analyst], tasks=[task])
        result = crew.kickoff(inputs={"research": research})

        self.state.insights = result.raw
        return result.raw


# 使用流式输出运行
flow = StatefulStreamingFlow()
streaming = flow.kickoff(inputs={"topic": "quantum computing"})

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

result = streaming.result
print(f"\n\nFinal state:")
print(f"Topic: {flow.state.topic}")
print(f"Research length: {len(flow.state.research)}")
print(f"Insights length: {len(flow.state.insights)}")

使用场景

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

多阶段工作流：展示研究、分析和综合阶段的进度
复杂流水线：为长时间运行的数据处理 flow 提供可见性
交互式应用：构建显示中间结果的响应式 UI
监控与调试：实时观察 flow 执行和 crew 交互
进度追踪：向用户展示工作流当前正在执行的阶段
实时仪表板：为生产环境中的 flow 创建监控界面

流块类型

与 crew 流式输出类似，flow 的块也可以有不同类型：

TEXT 块

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

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

TOOL_CALL 块

关于 flow 内部工具调用的信息：

for chunk in streaming:
    if chunk.chunk_type == StreamChunkType.TOOL_CALL and chunk.tool_call:
        print(f"\nTool: {chunk.tool_call.tool_name}")
        print(f"Args: {chunk.tool_call.arguments}")

错误处理

在流式执行期间优雅地处理错误：

flow = ResearchFlow()
streaming = flow.kickoff()

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

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

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

取消与资源清理

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

异步上下文管理器

streaming = await flow.kickoff_async()

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

显式取消

streaming = await flow.kickoff_async()
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() 都是幂等的。

重要说明

流式输出会自动为 flow 中使用的任何 crew 启用 LLM 流式输出
在访问 .result 属性之前，你必须先遍历完所有块
流式输出既适用于结构化也适用于非结构化的 flow 状态
Flow 流式输出会捕获 flow 中所有 crew 和 LLM 调用的输出
每个块都包含生成它的代理和任务的上下文
流式输出对 flow 执行的额外开销很小

结合 Flow 可视化

你可以把流式输出与 flow 可视化结合起来，以获得完整视图：

# 生成 flow 可视化
flow = ResearchFlow()
flow.plot("research_flow")  # Creates HTML visualization

# 使用流式输出运行
streaming = flow.kickoff()
for chunk in streaming:
    print(chunk.content, end="", flush=True)

result = streaming.result
print(f"\nFlow complete! View structure at: research_flow.html")

通过利用 flow 流式输出，你可以构建复杂、响应迅速的应用，为用户提供复杂多阶段工作流的实时可见性，让你的 AI 自动化更透明，也更具交互性。