用 CrewAI 构建多智能体 AI 系统

如何用 CrewAI 编排专业化的 AI 智能体——以及当 crew 真正跑在生产环境里时，会发生哪些变化。附两套我自己系统里的经验教训。

面对复杂的多步骤任务，单个 AI 智能体很快就会力不从心。顺理成章的解法：让多个专业化的智能体像一支团队那样协作。

CrewAI 正是为此而生的 Python 框架，截至 2026 年 10 月已发展到 1.15 版。我在自己的两套系统里用它：一是 Erst Recht，一个针对德国法律提供初步法律评估的 AI 平台；二是一个研究项目，其中一套自主的多智能体系统负责评估关于膳食补充剂的研究。本文先讲基础，再讲这些基础在生产环境里还剩下多少。一句话概括：少一点智能体魔法，多一点代码。

三个核心概念

与其打造一个无所不能的智能体，不如组建一支团队：角色、任务、流程。下面的示例展示一个通用的 crew，用来对收到的客服请求进行分拣。

1. 智能体（Agents）：专家

每个智能体都有一个角色（role）、一个目标（goal）和一段背景故事（backstory）。模型通过 LLM 类接入，模型名称前要带上服务商前缀。

from crewai import Agent, LLM

llm = LLM(
    model="provider/model-name",
    max_tokens=4096,
    timeout=120,  # hard ceiling per call
)

analyst = Agent(
    role="Support Analyst",
    goal="Classify requests and assess their urgency",
    backstory="""You analyze support requests for a B2B software product.
    Reply with the assessment only, no introduction.""",
    llm=llm,
    allow_delegation=False,
)

最佳实践：具体的角色

不要只写"分析师"，而要写目标清晰的"面向 B2B 软件的客服分析师"。同样重要的是：智能体不应该做什么。

2. 任务（Tasks）：具体的委派

任务定义了需要做什么。通过 context，一个任务可以拿到前一个任务的结果；通过 output_pydantic，你可以强制使用固定的输出格式，而不是自由文本。

from crewai import Task

task_assess = Task(
    description="Assess this request: {request}",
    expected_output="Category, priority, and a short rationale",
    agent=analyst,
    output_pydantic=TicketAssessment,  # schema instead of free text
    context=[task_classify],          # result of the previous agent
)

3. Crew：团队

crew 把智能体和任务组合到一起，并定义它们如何协作。

from crewai import Crew, Process

crew = Crew(
    agents=[classifier, analyst, writer],
    tasks=[task_classify, task_assess, task_reply],
    process=Process.sequential,
)

result = crew.kickoff(inputs={"request": text})
assessment = result.tasks_output[1].pydantic  # typed, no regex

另一种选择是 Process.hierarchical：由一个管理者 LLM 动态分配任务并审查结果。根据文档，这需要一个 manager_llm 或一个 manager_agent。在生产环境里，我依靠固定的执行顺序。一个事先就清楚的流程，才是我能测试、能调试的流程。

Crew 与 Flow：教程跳过的那部分

crew 是一组智能体。Flow 则是包裹在它外面的事件驱动流程：用 @start() 和 @listen() 标注的方法、用 @router() 实现的分支，再加上一个以 Pydantic 模型定义的类型化状态。Flow 的某一步可以运行一个 crew——也可以只是普通的 Python 代码。CrewAI 官方建议在生产环境中从 Flow 起步。在我的研究项目里，其原理大致如下（大幅简化）：

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

class RunState(BaseModel):
    topic_id: str = ""
    paper_ids: list[str] = []
    evaluation_ids: list[str] = []
    draft_id: str | None = None

class ResearchFlow(Flow[RunState]):
    @start()
    async def discover(self):
        # plain code: load and deduplicate studies
        self.state.paper_ids = await discover_papers(self.state.topic_id)

    @listen(discover)
    async def evaluate(self):
        # fixed scoring formula
        self.state.evaluation_ids = await evaluate_papers(self.state.paper_ids)

    @listen(evaluate)
    async def synthesize(self):
        # LLM step with a fixed output schema
        self.state.draft_id = await write_draft(self.state.evaluation_ids)

    # then: edit → fact-check → validate → publish

值得注意的是缺了什么：这套系统里一个 crew 都没有。我原本计划每个阶段用一个 crew。最终建成的是若干个阶段，都是直接调用模型的普通 async 函数，每个都有固定的输出 schema。在那里，我只用了 CrewAI 的 Flow：执行顺序、状态、定时启动。每个阶段都有清晰的输入和清晰的输出。留给一个自行规划的智能体去决定的东西，所剩无几。

两套系统一览

Erst Recht

多个专业化的智能体，由 CrewAI 编排，再加上一个法律知识库。产出是一份面向非法律专业人士的结构化初步法律评估。

膳食补充剂研究项目

一个由固定阶段组成的 Flow：查找研究、评分、起草文章、编辑、事实核查、验证证据等级、发布。最后一个阶段是不含 LLM 的纯代码。

生产环境中的五条规则

1. 过于严格的 schema 可能让整次运行中止

输出 schema 里的硬性取值约束（例如 Literal）是有风险的。模型只要返回一个近义词，Pydantic 就会抛出 ValidationError，视版本不同，CrewAI 会把它从 crew.kickoff() 里重新抛出。于是整次运行中止，已付费的 token 白白浪费，kickoff() 之后的兜底代码也不会执行。

解决办法：schema 只描述结构。允许的取值由代码在解析之后检查。以一个优先级字段为例：

# before: too strict, one synonym kills the whole run
class TicketAssessment(BaseModel):
    priority: Literal["high", "medium", "low"]

# after: schema for structure ...
class TicketAssessment(BaseModel):
    priority: str = Field(description="Allowed: 'high', 'medium', 'low'")

# ... and code for values, after parsing
ALLOWED = {"high", "medium", "low"}

assessment = result.tasks_output[1].pydantic
if assessment.priority.strip().lower() not in ALLOWED:
    flag_for_review(assessment)  # flag the value, the run is preserved

较新的 CrewAI 版本还提供任务护栏（Task Guardrails）：由一个函数检查输出，失败时把错误发回给智能体，默认最多重试三次（guardrail_max_retries）。对付格式错误，这种方式更干净。但每一次重试都是又一次付费的模型调用。

2. 智能体会编造来源

在研究项目里，文章中的每一条论断都必须有论文 ID 作为依据。提示词明确禁止编造 ID。但代码依然不会依赖这一点：

known_ids = set(paper_abstracts.keys())
for p in parsed.paragraphs:
    support = [pid for pid in p.supporting_paper_ids if pid in known_ids]
    if p.is_claim and not support:
        continue  # a claim without a real source is dropped
    paragraphs.append(...)

每一个被引用的 ID 都会与提示词里真正出现过的论文进行核对。之后还有更多硬性检查：如果某条论断无法追溯到真实来源，该阶段会直接失败、不做重试。事实核查则会核实被引用的 DOI 是否真实存在，以及论文是否真的支持该论断。

3. 没有哪个模型给自己打分

在研究项目里，负责撰写的模型和负责核查的模型来自不同的模型家族。一个审读自己文字的模型，很容易漏掉自己犯下的同样错误。这项核查是强制性的：如果它没有运行，就不发布，而不是悄悄降级为"未核实"。

4. 用户输入是数据，不是指令

凡是来自外部的东西都是数据，而不是命令：用户输入、上传的文档、来自数据库的摘要。这类文本在提示词里必须与任务指令明确隔开。在研究项目里，每一篇摘要在到达模型之前，还会由一个确定性的扫描器检查典型的注入模式。可疑的文本不会未经过滤就往下传。为什么这不是奢侈品，我在关于致命三要素的文章里讲过。

5. 成本和运行时间需要硬性上限

在研究项目里，每一次模型调用都要先经过一个成本管控器（Cost Governor），它对每次运行、每天和每月都设有固定上限。某次运行超出上限，就会被中止；当月超出上限，发布功能会自动关闭。

同样重要的是给每一次模型调用设置超时。一个挂起的调用不能堵住 worker。较长的 crew 运行也不该放在同步的 API 请求里；CrewAI 文档为此推荐使用 kickoff_async。如果你构建的是带工具的智能体，还应该有意识地设置单个智能体的限制：max_iter（默认 20）、max_execution_time、max_rpm。

代码还是智能体？我在哪里划线

两套系统给我的最重要一课：很大一部分工作根本不该交给智能体。

数据获取：在研究项目里，由普通的适配器从 PubMed 等数据库拉取研究，配有限流和去重。那里没有任何智能体自己去搜索。
提供知识：如果智能体要用到知识库，就要有意识地决定：是让智能体自己去搜，还是由代码预先取出相关段落。预先取出的上下文可复现，也可以记录日志。搜索工具给了智能体更多自由——也给了它更多迷路的机会。如何搭建这样的知识库，见我的 Docling 那篇文章。
评分：一项研究的质量分是一个固定公式，基于研究设计、样本量等经典标准。
证据等级：Validator 永远取模型建议值与计算值中较低的那个。不可能升级。
发布：不含 LLM 的纯代码，审批放行在多处加以保障。

经验法则

凡是有唯一正确答案的事情，都应该交给代码。智能体只负责需要判断力的步骤。即便如此，结果也要由代码来检查。

然后还有人。在最初的设计里，研究项目是"完全自主"的。最终建成的却带有一道审批关卡：最初的一批文章只有经过人工批准才能上线。这套系统如何把自己的失败转成 issue 和 pull request，我在《用循环取代提示》里写过。

什么时候你不需要 crew

步骤是固定的。如果每一步都有清晰的输入和固定的输出 schema，一个由普通函数组成的 Flow 就够了。我的研究项目就是这么跑的。
答案必须很快给出。每多一个顺序执行的智能体，就多一次 LLM 调用。对于没人实时等待的分析，这没问题；要达到聊天般的响应速度，就会显得拖沓。
为了多而多的智能体。每多一个智能体，都会增加 token 消耗和延迟。更复杂的拓扑能否带来更好的结果，要靠测试来证明，而不是靠架构图。
你无法检验结果。没有 schema、没有校验步骤、没有成本上限，crew 就不该放到用户面前。

第一个生产级 crew 的检查清单

从 Flow 起步。长期状态应放在数据库里，而不是 Flow 的状态里。
每个任务都配上 output_pydantic：schema 管结构，代码管取值。
在代码中把每一个来源、ID 和数字都与输入核对。
用与撰写者不同的模型家族来做核查。
在第一位用户到来之前，加上超时和成本上限。
把用户输入和外部文本明确隔开，并把它们当作数据处理。
起步阶段，在发布之前安排一个人把关。

结语

CrewAI 让入门变得很轻松：角色、任务、crew，搞定。到了生产环境，重心就转移了。对我来说，大部分工作都在智能体周围的代码里：不会让运行中止的 schema、筛掉编造来源的检查、给成本封顶的上限。智能体负责需要判断力的部分，其余一切交给代码。

用到的工具：

CrewAI CrewAI Docs DeepLearning.AI

正在规划一套多智能体系统，或者想让现有的 crew 达到生产就绪？联系我，我很乐意在架构设计和落地实现上帮上忙。

用 CrewAI 构建多智能体 AI 系统：团队协作，而非单打独斗