生产架构

以 Flow 为先的思维方式

在使用 CrewAI 构建生产级 AI 应用时，我们建议先从 Flow 开始。

虽然你可以运行单独的 Crews 或 Agents，但将它们包裹在 Flow 中，会为构建稳健、可扩展的应用提供必要的结构。

为什么使用 Flows？

状态管理：Flows 提供内置方式来管理应用各步骤之间的状态。这对于在 Crews 之间传递数据、维护上下文以及处理用户输入至关重要。
控制能力：Flows 让你定义精确的执行路径，包括循环、条件分支和分流逻辑。这对于处理边界情况并确保应用行为可预测非常重要。
可观测性：Flows 提供清晰的结构，使追踪执行、调试问题和监控性能更容易。我们建议使用 CrewAI Tracing 获取详细洞察。只需运行 crewai login 即可启用免费的可观测性功能。

架构

一个典型的生产级 CrewAI 应用看起来像这样：

graph TD
    Start((Start)) --> Flow[Flow Orchestrator]
    Flow --> State{State Management}
    State --> Step1[Step 1: Data Gathering]
    Step1 --> Crew1[Research Crew]
    Crew1 --> State
    State --> Step2{Condition Check}
    Step2 -- "Valid" --> Step3[Step 3: Execution]
    Step3 --> Crew2[Action Crew]
    Step2 -- "Invalid" --> End((End))
    Crew2 --> End

1. Flow 类

你的 Flow 类是入口点。它定义状态 schema，以及执行逻辑的方法。

from crewai.flow.flow import Flow, listen, start
from pydantic import BaseModel

class AppState(BaseModel):
    user_input: str = ""
    research_results: str = ""
    final_report: str = ""

class ProductionFlow(Flow[AppState]):
    @start()
    def gather_input(self):
        # ... 获取输入的逻辑 ...
        pass

    @listen(gather_input)
    def run_research_crew(self):
        # ... 触发一个 Crew ...
        pass

2. 状态管理

使用 Pydantic 模型来定义你的状态。这既能保证类型安全，也能清楚地表明每一步可用的数据。

保持精简：只存储在步骤之间需要持久化的数据。
使用结构化数据：尽量避免使用无结构的字典。

3. 将 Crews 作为工作单元

把复杂任务委派给 Crews。一个 Crew 应该专注于一个明确的目标（例如“研究一个主题”、“撰写一篇博客”）。

不要过度设计 Crews：保持其专注。
显式传递状态：把 Flow state 中必要的数据传递给 Crew 输入。

    @listen(gather_input)
    def run_research_crew(self):
        crew = ResearchCrew()
        result = crew.kickoff(inputs={"topic": self.state.user_input})
        self.state.research_results = result.raw

控制原语

利用 CrewAI 的控制原语，为你的 Crews 增加健壮性和控制力。

1. Task Guardrails

使用 Task Guardrails 在接受 task 输出之前进行验证。这可以确保 agents 产出高质量结果。

def validate_content(result: TaskOutput) -> Tuple[bool, Any]:
    if len(result.raw) < 100:
        return (False, "Content is too short. Please expand.")
    return (True, result.raw)

task = Task(
    ...,
    guardrail=validate_content
)

2. 结构化输出

在 task 之间或向应用传递数据时，始终使用结构化输出（output_pydantic 或 output_json）。这可以防止解析错误并确保类型安全。

class ResearchResult(BaseModel):
    summary: str
    sources: List[str]

task = Task(
    ...,
    output_pydantic=ResearchResult
)

3. LLM Hooks

使用 LLM Hooks 在消息发送给 LLM 之前检查或修改它们，或者对响应进行清理。

@before_llm_call
def log_request(context):
    print(f"Agent {context.agent.role} is calling the LLM...")

部署模式

部署 Flow 时，请考虑以下事项：

CrewAI Enterprise

部署 Flow 最简单的方法是使用 CrewAI Enterprise。它会为你处理基础设施、认证和监控。

查看部署指南开始使用。

crewai deploy create

异步执行

对于长时间运行的任务，使用 kickoff_async 以避免阻塞你的 API。

持久化

使用 @persist 装饰器将 Flow 的状态保存到数据库中。这样当进程崩溃，或者你需要等待人工输入时，都可以恢复执行。

@persist
class ProductionFlow(Flow[AppState]):
    # ...

默认情况下，当提供 kickoff(inputs={"id": <uuid>}) 时，@persist 会恢复一个 flow，并延续同一个 flow_uuid 历史。要把一个已持久化的 flow 分叉成新的 lineage - 从之前的运行中填充状态，但写入一个新的 state.id - 请传入 restore_from_state_id：

flow.kickoff(restore_from_state_id="<previous-run-state-id>")

新运行会获得一个新的 state.id（自动生成，或者如果已固定则使用 inputs["id"]），因此它的 @persist 写入不会延续源历史。与 from_checkpoint 同时使用会引发 ValueError；请选择一种 hydration 来源。

总结

先从 Flow 开始。
定义清晰的 State。
用 Crews 处理复杂任务。
通过 API 和持久化进行部署。