跳转到主要内容

文档索引

获取完整文档索引： https://docs.crewai.com.cn/llms.txt

在深入了解之前，请使用此文件来浏览所有可用页面。

​
“流程优先”思维模式

在使用 CrewAI 构建生产级 AI 应用时，我们建议从“流程”（Flow）开始。 虽然可以运行独立的 Crew（工作组）或 Agent（智能体），但将其封装在 Flow 中可为构建稳健、可扩展的应用程序提供必要的结构。

​
为什么选择 Flow？

  1. 状态管理：Flow 提供了一种内置方式来管理应用程序各个步骤之间的状态。这对于在 Crew 之间传递数据、维护上下文以及处理用户输入至关重要。
  2. 控制能力：Flow 允许您定义精确的执行路径，包括循环、条件判断和分支逻辑。这对于处理边界情况并确保应用程序行为可预测是必不可少的。
  3. 可观测性：Flow 提供了清晰的结构，使跟踪执行情况、调试问题和监控性能变得更加容易。我们建议使用 CrewAI Tracing 来获取深入见解。只需运行 crewai login 即可启用免费的可观测性功能。

​
架构

一个典型的生产级 CrewAI 应用架构如下

​
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):
        # ... logic to get input ...
        pass

    @listen(gather_input)
    def run_research_crew(self):
        # ... trigger a Crew ...
        pass

​
2. 状态管理

使用 Pydantic 模型来定义状态。这可以确保类型安全，并明确在每个步骤中可以使用哪些数据。
  • 保持精简：仅存储需要在步骤之间持久化的数据。
  • 使用结构化数据：尽可能避免使用非结构化的字典。

​
3. 将 Crew 作为工作单元

将复杂任务委托给 Crew。每个 Crew 都应专注于特定目标（例如：“调研主题”、“撰写博客文章”）。
  • 不要过度设计 Crew：保持其专注度。
  • 显式传递状态：将必要的 Flow 状态数据传递给 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 的控制原语为您的 Crew 增加稳健性和控制力。

​
1. 任务护栏（Task Guardrails）

使用 任务护栏 在任务输出被接受之前对其进行验证。这确保了您的智能体能够产出高质量的结果。 
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）。这可以防止解析错误并确保类型安全。 
class ResearchResult(BaseModel):
    summary: str
    sources: List[str]

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

​
3. LLM 钩子（Hooks）

使用 LLM 钩子 在消息发送给 LLM 之前对其进行检查或修改，或用于对响应进行净化。 
@before_llm_call
def log_request(context):
    print(f"Agent {context.agent.role} is calling the LLM...")

​
部署模式

在部署 Flow 时，请考虑以下建议

​
CrewAI 企业版

部署 Flow 最简单的方法是使用 CrewAI 企业版。它为您处理基础设施、身份验证和监控。 查看 部署指南 以开始使用。
crewai deploy create

​
异步执行

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

​
持久化

使用 @persist 装饰器将 Flow 的状态保存到数据库中。这样，如果进程崩溃或需要等待人工输入，您可以恢复执行。 
@persist
class ProductionFlow(Flow[AppState]):
    # ...
默认情况下，当提供 kickoff(inputs={"id": <uuid>}) 时，@persist 会恢复流，并沿用同一个 flow_uuid 的历史记录。若要将持久化流分叉（fork）到新的谱系中（即从之前的运行中水合状态，但以新的 state.id 写入），请传入 restore_from_state_id。 
flow.kickoff(restore_from_state_id="<previous-run-state-id>")
新的运行会获得一个新的 state.id（自动生成，或者如果已固定则使用 inputs["id"]），因此其 @persist 写入不会扩展源数据的历史记录。将其与 from_checkpoint 结合使用会引发 ValueError；请选择一种水合来源。

​
总结

  • 从 Flow 开始。
  • 定义清晰的状态。
  • 使用 Crew 处理复杂任务。
  • 通过 API 和持久化机制进行部署。