主题
生产架构
使用 CrewAI 构建可用于生产的 AI 应用的最佳实践
流程优先的思维模式
在使用 CrewAI 构建生产级 AI 应用时，我们建议从 Flow 开始。
虽然可以运行单个团队或智能体，但将它们包裹在 Flow 中，才能为健壮、可扩展的应用提供必要的结构。
为什么选择 Flow？
- 状态管理：Flow 提供了内置的方式来管理应用不同步骤之间的状态。这对于在多个团队之间传递数据、维护上下文以及处理用户输入至关重要。
- 控制：Flow 允许你定义精确的执行路径，包括循环、条件判断和分支逻辑。这对于处理边界情况、确保应用行为可预测至关重要。
- 可观测性：Flow 提供了清晰的结构，使追踪执行过程、调试问题和监控性能更加容易。我们建议使用 CrewAI 追踪 获取详细洞察。只需运行
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 类是入口点。它定义了状态结构以及执行你的逻辑的方法。
python
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):
# ... logic to get input ...
pass
@listen(gather_input)
def run_research_crew(self):
# ... trigger a Crew ...
pass2. 状态管理
使用 Pydantic 模型定义状态。这能确保类型安全，并让每一步可用的数据清晰明了。
- 保持精简：只存储步骤之间需要持久化的内容。
- 使用结构化数据：尽可能避免使用非结构化的字典。
3. 团队作为工作单元
将复杂任务委派给团队。团队应聚焦于特定目标（例如“研究某个主题”、“撰写一篇博客文章”）。
- 不要过度设计团队：保持团队聚焦。
- 显式传递状态：将 Flow 状态中必要的数据传递给团队的输入。
python
@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 的控制原语，为你的团队增加健壮性与控制力。
1. 任务护栏
在接受任务输出之前，使用 任务护栏 对其进行校验。这能确保你的智能体产出高质量的结果。
python
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. 结构化输出
在任务之间或向你的应用传递数据时，始终使用结构化输出（output_pydantic 或 output_json）。这可以避免解析错误并确保类型安全。
python
class ResearchResult(BaseModel):
summary: str
sources: List[str]
task = Task(
...,
output_pydantic=ResearchResult
)3. LLM Hooks
使用 LLM Hooks 在消息发送给 LLM 之前检查或修改消息，或对响应进行清洗。
python
@before_llm_call
def log_request(context):
print(f"Agent {context.agent.role} is calling the LLM...")部署模式
部署你的 Flow 时，请考虑以下内容：
CrewAI 企业版
部署 Flow 最简单的方式是使用 CrewAI 企业版。它会为你处理基础设施、身份认证和监控。
请查看部署指南开始使用。
bash
crewai deploy create异步执行
对于长时间运行的任务，使用 kickoff_async 以免阻塞你的 API。
持久化
使用 @persist 装饰器将 Flow 的状态保存到数据库。这样，如果进程崩溃或你需要等待人工输入，就可以恢复执行。
python
@persist
class ProductionFlow(Flow[AppState]):
# ...默认情况下，当提供 kickoff(inputs=""id": <uuid>") 时，@persist 会恢复流程，延续同一个 flow_uuid 历史记录。若要将持久化的流程分叉为新的谱系——即从先前的运行中注入状态，但写入新的 state.id 下——请传入 restore_from_state_id：
python
flow.kickoff(restore_from_state_id="<previous-run-state-id>")新运行会获得一个新的 state.id（自动生成，或在固定时使用 inputs["id"]），因此它的 @persist 写入不会延续源的历史记录。与 from_checkpoint 组合使用会抛出 ValueError；请只选择一个状态注水来源。
总结
- 从 Flow 开始。
- 定义清晰的状态。
- 对复杂任务使用团队。
- 通过 API 和持久化进行部署。