跳到主要内容

数据截至 (上游 commit 2ebb9390fe07)

02 · Crew 编排(多 Agent 怎么协作)

本章讲什么: 一支 Crew 怎么把若干 Task 和 Agent 组织起来——顺序跑还是交给经理调度、上下文怎么在任务间传、Agent 怎么互相委派、输出怎么校验和聚合。

2.1 它要解决的小问题​

单个 Agent(上一章)只能干一件事。真实项目要多步骤、多角色:研究员产出资料、写手据此成文、校对再润色。谁先谁后?后一步怎么拿到前一步的产物?谁来分配活?Crew 就是回答这些编排问题的层。

2.2 入口:kickoff → 选 process​

Crew.kickoff(inputs)(crew.py:992)是同步入口。核心分支极简(crew.py:1051-1057):

# crew.py:1051-1057(摘)
if self.process == Process.sequential:
    result = self._run_sequential_process()
elif self.process == Process.hierarchical:
    result = self._run_hierarchical_process()
else:
    raise NotImplementedError(...)

Process 只有两个成员:sequential 和 hierarchical(process.py,consensual 还是 TODO)。两种流程最终都调同一个 _execute_tasks(crew.py:1509-1516);层级模式只是先多造一个经理 Agent。

kickoff 还做了几件横切的事:开 OpenTelemetry baggage 追踪、进入事件总线运行时作用域、prepare_kickoff 插值 inputs、结束时排空记忆写入 + 清理文件(crew.py:1040-1086)。

2.3 顺序流程(sequential)​

_run_sequential_process 直接把任务列表交给 _execute_tasks(self.tasks)(crew.py:1509-1511)。_execute_tasks 是真正的编排循环(crew.py:1558-1627),逐个任务做:

  1. prepare_task_execution 选出该任务的 Agent、工具、是否跳过(crew.py:1583)。
  2. 若是 ConditionalTask,先判条件,不满足就跳过(crew.py:1589-1592,见 2.7)。
  3. 算出上下文 context = self._get_context(task, task_outputs)——即把已完成任务的输出喂给当前任务(crew.py:1614)。
  4. task.execute_sync(agent, context, tools) 同步执行(crew.py:1615)。
  5. 输出存进 task_outputs,记执行日志。

异步任务穿插: 若任务标了 async_execution,用 task.execute_async 拿到 Future 攒着(crew.py:1597-1606);遇到下一个同步任务前,先 _process_async_tasks 把攒着的 future 全部 join(crew.py:1608-1611)。这让相邻的异步任务能并行,但同步任务作为屏障。

2.4 层级流程(hierarchical):自动造一个经理​

_run_hierarchical_process 先 _create_manager_agent() 再跑同一个 _execute_tasks(crew.py:1513-1516)。经理 Agent 的创建逻辑(crew.py:1518-1543):

# crew.py:1533-1541(摘)— 自动合成的经理
manager = Agent(
    role=i18n.retrieve("hierarchical_manager_agent", "role"),
    goal=i18n.retrieve("hierarchical_manager_agent", "goal"),
    backstory=i18n.retrieve("hierarchical_manager_agent", "backstory"),
    tools=AgentTools(agents=self.agents).tools(),  # 关键:经理的工具 = 委派/提问
    allow_delegation=True,
    llm=self.manager_llm,
    verbose=self.verbose,
)

两个真实约束:

  • 经理不能自带普通工具:若你传的 manager_agent 带了 tools,先打一条警告并清空 tools,再抛 Exception("Manager agent should not have tools")(crew.py:1522-1528)。经理的「工具」只能是「把活派给手下」。
  • 层级模式下,任务执行用的 Agent 永远是经理(_get_agent_to_use 在 hierarchical 时返回 self.manager_agent,crew.py:1714-1717)。经理通过委派工具把活分给真正的工人 Agent。

2.5 Agent 之间怎么协作:委派与提问​

经理(以及任何 allow_delegation=True 的 Agent)拿到的特殊工具来自 AgentTools.tools()(tools/agent_tools/agent_tools.py:22),就两个:

工具作用文件
DelegateWorkTool把一个子任务整体派给某个「同事」Agent 去做tools/agent_tools/delegate_work_tool.py
AskQuestionTool向某个同事 Agent 提一个问题、拿回答tools/agent_tools/ask_question_tool.py

在顺序模式下,如果某个 Agent 开了 allow_delegation,_add_delegation_tools 会把「除自己外的其他 Agent」作为可委派对象注入委派工具(crew.py:1820-1830)。于是「委派」在实现上就是一个普通工具调用——复用了第 1 章的工具循环,优雅地把多 Agent 协作降维成了「Agent 调了个叫『派活给同事』的工具」。

_prepare_tools(crew.py:1645-1712)是工具装配的总闸:按 Agent 能力依次叠加委派、代码执行、多模态、平台(apps)、MCP、记忆、文件等工具。

2.6 上下文怎么传​

任务间的「接力」靠 _get_context:把前序任务的 TaskOutput 拼成文本,作为后一任务提示词的一部分(crew.py:1614、1598-1600)。顺序模式默认拿到目前所有产出;异步分支只拿最近的同步输出。Task 自身也可显式声明 context=[task_a, task_b] 来精确控制依赖(见 task.py)。

2.7 条件任务(ConditionalTask)​

_handle_conditional_task(crew.py:1629-1643)在执行前调 check_conditional_skip:根据前序输出判断该任务是否该跑。若跳过,产出一个「跳过」的占位 TaskOutput 并继续。注意它会先把攒着的异步 future 全部结清,因为条件判断可能依赖那些结果(crew.py:1637-1639)。

2.8 输出校验与聚合:guardrail → CrewOutput​

单个 Task 的执行核心在 Task._execute_core(task.py:806):它调 agent.execute_task(...) 拿到原始结果(task.py:850),包成 TaskOutput(带 raw / pydantic / json_dict,task.py:876-886),然后:

  • 跑 guardrail:self._guardrails 和 self._guardrail 逐个校验/修正输出(task.py:889-904)。guardrail 是「输出守门员」——可以是函数或 LLM 判定(LLMGuardrail),不通过可触发重试。
  • 触发 task / crew 级回调,可选写文件(task.py:922-948)。
  • emit TaskCompletedEvent(task.py:793-795)。

所有 TaskOutput 最后由 _create_crew_output(task_outputs) 聚合成一个 CrewOutput(crew.py:1627),result.raw 通常取最后一个任务的输出。kickoff 收尾还会跑 after_kickoff_callbacks 和计算 token 用量(crew.py:1060-1065)。

2.9 其他入口​

方法用途位置
kickoff_for_each(inputs: list)对一批输入各跑一次(每次 self.copy()),聚合 usagecrew.py:1091
kickoff_async把同步 kickoff 丢到线程跑;流式时返回 CrewStreamingOutputcrew.py:1127
akickoff原生异步:全程 async,不靠线程包装crew.py:1207
train / test / replay训练(带人类反馈迭代)、评测、从某任务重放crew.py:940 / 2227 / 2031

→ 下一章:另一套范式 Flow,它不靠「任务列表」而靠「方法间的事件触发」来编排。