CrewAI 的卖点是"用人类团队的语言描述 Agent 系统":角色、目标、背景故事、任务、编组。好处是十分钟能跑起一条多步内容流水线;代价是它把一部分控制流藏进了自然语言——谁先谁后、谁能派活,一部分由 Process 决定,一部分由模型自己决定。
所以它适合"谁的活谁干"本来清楚的场景,不适合需要精确分支与审计的状态机。下面把四个概念、两种流程、三件套、结构化输出和成本旋钮拆开。
1. 四个概念,各管一件事
| 概念 | 类 | 是什么 | 实际决定什么 |
|---|---|---|---|
| Agent 角色 | Agent |
带 role/goal/backstory 的执行者 |
输出的风格、判断倾向、愿不愿意自己查证 |
| Task 任务 | Task |
一次有描述、期望输出、可用工具的作业 | 具体产出物与验收标准 |
| Crew 编组 | Crew |
agents + tasks 的集合 | 谁在什么范围内能干什么 |
| Process 流程 | Process |
sequential 或 hierarchical |
执行顺序、由谁调度 |
关键认知:Task 才是函数,Agent 只是执行上下文。行为要稳定,约束得写进 description 与 expected_output,别指望 backstory 兜住一切。
2. 顺序流程与层级流程
from crewai import Agent, Task, Crew, Process
researcher = Agent(
role="资料研究员",
goal="为每个结论找到可引用的原始出处",
backstory="你只信一手文档与源码,遇到二手转述会回去核对原文。",
tools=[search_tool],
)
writer = Agent(
role="技术编辑",
goal="把研究结论写成 3000 字以内、有代码可跑的文章",
backstory="你删掉所有形容词,只保留可验证的事实和步骤。",
)
t1 = Task(description="围绕 {topic} 收集 5 条有出处的关键结论",
expected_output="编号列表,每条含结论、来源链接、可信度判断",
agent=researcher)
t2 = Task(description="基于给定结论写一篇技术文章",
expected_output="Markdown 正文,含至少 2 个代码块",
agent=writer, context=[t1])
crew = Crew(agents=[researcher, writer], tasks=[t1, t2],
process=Process.sequential, verbose=True)
print(crew.kickoff(inputs={"topic": "MCP 协议"}))
层级流程多一个调度者,它每轮都要做一次"下一步谁干"的判断:
manager = Agent(role="主编", goal="保证交付质量与风格一致",
backstory="你只关心最终能不能发出去,不合格就打回重做。",
allow_delegation=True)
crew = Crew(agents=[researcher, writer], tasks=[t1, t2],
process=Process.hierarchical, manager_agent=manager)
| 维度 | sequential |
hierarchical |
|---|---|---|
| 调度者 | 无,按 tasks 列表顺序 |
manager Agent(LLM 决策) |
| 请求成本 | 低,一步一次 | 高,每步多一次路由决策与复盘 |
| 可预测性 | 高,顺序写死在代码里 | 低,同样的输入可能走不同路径 |
| 适合 | 固定流水线:抓取→分析→写作→校对 | 边界模糊、需动态分工或返工 |
层级流程成本常是顺序流程的 2–3 倍,因为 manager 每次判断都是一次完整推理。没有返工需求就别开 hierarchical,它买的是灵活性,不是质量。
3. role / goal / backstory 三件套
| 字段 | 该写什么 | 写空的后果 |
|---|---|---|
role |
职能名词,如"资深后端评审" | 模型退化成通用助手,什么都答一点 |
goal |
可验收的目标,如"每条结论附来源链接" | 没有终止标准,Agent 不知道什么时候算完 |
backstory |
约束与偏好,如"只接受一手文档" | 同一 Crew 两次跑的结果差异很大 |
backstory 是最容易被写废的字段。写"你有 10 年经验、来自大厂"没有任何约束力;写"你会先质疑数据来源再谈结论,无法核实的数字直接标注'待确认'"就直接改变行为。三件套最终都拼进系统提示,它是提示工程的一个语法糖,不是人设表演。
4. tools 挂载:Agent 级与 Task 级
from crewai.tools import tool
@tool("check_endpoint")
def check_endpoint(url: str) -> str:
"""探测一个 HTTP 端点的状态码与响应时间。url 必须是完整 https 地址。"""
...
| 挂载粒度 | 写法 | 适合 |
|---|---|---|
| Agent 级 | Agent(tools=[...]) |
该角色的通用能力,多个任务共用 |
| Task 级 | Task(tools=[...]) |
只在这一步用一次的工具,避免误用 |
工具描述和 MCP 里一样,是给模型读的 prompt:写清"何时用、参数约束、返回什么形状",比写"这个函数做什么"有效。挂载面越大选错概率越高——挂 15 个工具通常不如挂 5 个。
5. context:只传该传的
| 依赖方式 | 写法 | 语义 |
|---|---|---|
| 列表顺序 | 默认 | 前序任务的输出会进入后续任务的上下文 |
显式 context |
Task(context=[t1]) |
只注入指定任务的输出,其他不可见 |
全局 inputs |
kickoff(inputs={...}) |
在描述里用 {topic} 插值 |
顺序流程里如果不写 context,上下文会随着任务数滚雪球——第 5 个任务看到的可能是前 4 个任务的全文。显式写 context 是控制 token 的主要手段,比事后压缩便宜得多。
6. 结构化输出
from pydantic import BaseModel
class Article(BaseModel):
title: str
sections: list[str]
word_count: int
t2 = Task(description="基于给定结论写文章大纲",
expected_output="一个 JSON 对象,含 title、sections、word_count",
agent=writer,
output_pydantic=Article)
out = crew.kickoff(inputs={"topic": "MCP"}).pydantic # Article 实例
| 输出方式 | 参数 | 拿到什么 |
|---|---|---|
| 纯文本 | 默认 | str,形状不可控 |
| Pydantic | output_pydantic=Model |
校验过的对象,校验失败会触发重试 |
| JSON | output_json=Model |
dict 形状,同样会校验 |
| 落盘 | output_file="out.json" |
文件,便于下游流水线消费 |
output_pydantic 的价值不只是类型安全:它把"格式不对"变成了一个可重试的错误,而不是一条悄悄流进下游的脏数据。代价是失败时会多烧几轮 token,所以 expected_output 里要把字段说清楚,别让模型猜。
6.1 一条完整的内容流水线
把上面几节拼起来就是一整条流水线。注意校对任务的 context 只接 t2,不接研究阶段的原始材料——那些内容已经体现在成稿里:
proofreader = Agent(role="事实校对", goal="核对数字与引用是否与来源一致",
backstory="你只做减法:删掉无法核实的断言,标出矛盾句。")
t3 = Task(description="校对正文中的事实性断言",
expected_output="问题清单;无问题时输出'无问题'",
agent=proofreader, context=[t2])
crew = Crew(agents=[researcher, writer, proofreader], tasks=[t1, t2, t3],
process=Process.sequential, memory=False)
memory=False 是隔离变量:开记忆会让上一轮的事实漂进下一轮。第一次调流水线先关掉,看清每个任务收了什么、吐了什么,再决定开不开。
三个阶段的传递形状:调研交的是"结论+来源",写作交的是"正文",校对只回"问题清单"。每多传一层全文,后面的任务就多背一份噪音。
7. 成本与轮次
| 参数 | 含义 | 调小的后果 | 调大的后果 |
|---|---|---|---|
max_iter |
单个 Agent 在单任务内的最大推理轮数 | 复杂任务半途而废 | 成本线性上涨,常见于死循环 |
max_rpm |
每分钟请求上限 | 触发限流原地等待 | 容易被供应商 429 |
max_execution_time |
整个 Crew 的时长上限 | 长任务被截断 | 挂死无人收尸 |
allow_delegation |
Agent 能否把活派给同伴 | hierarchical 下调度失效 |
互相甩锅、来回确认 |
max_iter 从 10 起步,看 verbose 日志里实际用了几轮,再按 P95 留 30% 余量;max_rpm 设成限速的 60%–70%。这两个参数不是"越大越聪明",超过某点只是把同一件事重复想。
8. 与 LangGraph 的定位差异
| 维度 | CrewAI | LangGraph |
|---|---|---|
| 表达方式 | 声明式:角色 + 任务 + 流程 | 命令式:节点 + 边 |
| 控制流 | Process + 模型判断 |
代码里显式写死 |
| 循环 | Agent 内部迭代 | 显式条件边 |
| 人工介入 | 自己包一层 | interrupt 原生支持 |
| 调试 | 看 verbose 日志 |
断点、checkpoint、可回放 |
| 适合 | 多步内容生产、角色分工清晰 | 审批流、强状态机、需要审计 |
一句话取舍:流程能用一张带分支和回退的流程图说清,用 LangGraph;流程是一条直线、每步需要不同"人格"的专业度,用 CrewAI。
两者不必二选一:常见形态是 LangGraph 管主流程与审批,某个节点内部开一个 Crew 完成"调研 + 写作"。图负责可审计,Crew 负责这一步的专业分工。
9. 排错
| 现象 | 原因 | 处理 |
|---|---|---|
| 输出跑题、风格每次不同 | backstory 太泛,没有拒绝规则 |
把约束改成可判定的句子 |
| 后一个任务重复前一个的活 | 没写 context,上下文互相污染 |
显式指定 context=[t1] |
| 结构化输出反复校验失败 | expected_output 没列字段 |
描述里逐字段说明类型与取值 |
| 任务卡在同一轮反复思考 | max_iter 太大 + 目标不可判定 |
收紧 goal,把 max_iter 降到 8–10 |
| 频繁 429 | max_rpm 高于供应商配额 |
降到配额 60%,加退避重试 |
| manager 把活派给自己 | 没设 allow_delegation 或角色重叠 |
manager 只做验收,不给工具 |
10. 边界
| 场景 | 该不该用 CrewAI | 理由 |
|---|---|---|
| 多步内容生产(调研→写作→校对) | 用 | 角色分工天然清晰 |
| 需要严格顺序与审批节点 | 不用 | 控制流应由代码掌握 |
| 单次工具调用 | 不用 | 一层包装的成本换不来收益 |
| 需要可回放重试 | 不用 | 没有 checkpoint,重跑等于重花钱 |
| 需要动态分工的探索型任务 | 用 hierarchical | 让模型决定下一步是谁 |
11. 自测
- Agent 与 Task 谁才是"函数",约束应该写在哪一个上?
hierarchical相比sequential多出来的成本花在哪一步?- 为什么
backstory写"大厂 10 年经验"没有效果? - 不写
context时,第 5 个任务的上下文里可能出现什么? output_pydantic校验失败时会发生什么,代价是什么?- 什么信号说明你该从 CrewAI 换到 LangGraph?