2756 字
约 9 分钟
18
S21-X06 · CrewAI:角色、任务与协作流程的工程化

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. 自测

  1. Agent 与 Task 谁才是"函数",约束应该写在哪一个上?
  2. hierarchical 相比 sequential 多出来的成本花在哪一步?
  3. 为什么 backstory 写"大厂 10 年经验"没有效果?
  4. 不写 context 时,第 5 个任务的上下文里可能出现什么?
  5. output_pydantic 校验失败时会发生什么,代价是什么?
  6. 什么信号说明你该从 CrewAI 换到 LangGraph?
S21-X06 · CrewAI:角色、任务与协作流程的工程化
https://www.clxhxhhr.top/posts/5689/
作者
clxstart
发布于
2024-12-31
许可协议
CC BY-NC-SA 4.0
评论
0 条
还没有评论,先写一条吧。