Day 21 / 共 60 天 · 阶段4 Crew 与流程
Process.hierarchical:给团队配一个"经理"来调度
Day 20 的顺序流程是"死板的流水线"——任务顺序写死、agent 提前指定好。今天的层级流程换一种玩法:加一个经理 agent(manager),由它来决定"这个任务派给谁、要不要追问、拿到结果满不满意"。神奇的是,实现只多了一步 _create_manager_agent()(crew.py:1484),核心秘密在于——经理靠两个"委派工具"把下属当工具来调。今天把这一步拆到底。
📍 你在 60 天里的位置(阶段4 Crew 与流程 · 共 8 天)
D19 Crew 全字段→
D20 顺序流程→
D21 层级+manager→
D22 kickoff 家族→
D23 规划→
D24 训练+replay→
D25 记忆开关→
D26 事件系统
💡 先用一个类比兜住今天
层级流程就像一个项目经理带一队专家:你把任务交给经理,经理自己不写代码也不查资料,它只做一件事——"这活该派给谁?"然后把任务委派给对应的下属,或者追问下属细节,收到结果后决定接受还是重派。而在 CrewAI 里,"委派"和"追问"被实现成经理手里的两个工具(DelegateWorkTool / AskQuestionTool)——每个下属就成了经理能"调用"的对象。这就是层级流程的全部魔法。
L01
痛点:任务多、谁干不确定,怎么办?
🤔 痛点顺序流程要求你提前想清楚:每个任务派给哪个 agent、按什么顺序。可现实里常是"我有一堆专家(写手、程序员、分析师),有个模糊的大目标,但具体拆成什么活、派给谁、要不要来回问,我事先不知道"。这时候你需要一个"会看菜下饭"的调度者——它根据任务动态决定派谁。这就是层级流程要解决的。
💡 一句话本质
层级流程 = 顺序流程 + 一个"经理 agent" + 两个委派工具。执行引擎还是那个
_execute_tasks(Day 20),但多了个 _create_manager_agent():它造一个只会调度、不带业务工具的经理,并把所有下属 agent 包装成 DelegateWorkTool(派活)和 AskQuestionTool(追问)两个工具塞给经理。于是经理在自己的 ReAct 循环里,就能"调用工具"= "把活委派给某个下属"。下属被降维成了经理的工具。大白话你可能以为层级流程有一套复杂的"调度算法"。其实没有——它复用了 Agent 已有的工具调用能力(Day 08 那个 ReAct 循环)。经理就是个普通 agent,只不过它的"工具箱"里装的不是搜索、计算,而是"把任务丢给张三"、"问李四一个问题"。委派 = 调工具,就这么朴素。
L02
分流入口 + 构造期的前置校验
入口和顺序流程对称,只多一行(crew.py:1479):
# crew.py:1479
def _run_hierarchical_process(self) -> CrewOutput:
"""Creates and assigns a manager agent to complete the tasks."""
self._create_manager_agent() # ★唯一的区别:先造经理
return self._execute_tasks(self.tasks) # 之后完全复用 Day 20 的引擎
而"必须有经理"这条规则,早在构造 Crew 时就被校验了(crew.py:696):
# crew.py:696
@model_validator(mode="after")
def check_manager_llm(self) -> Self:
if self.process == Process.hierarchical:
if not self.manager_llm and not self.manager_agent:
raise PydanticCustomError("missing_manager_llm_or_manager_agent",
"Attribute `manager_llm` or `manager_agent` is required "
"when using hierarchical process.", {})
if (self.manager_agent is not None) and (self.agents.count(self.manager_agent) > 0):
raise PydanticCustomError("manager_agent_in_agents",
"Manager agent should not be included in agents list.", {})
return self
_create_manager_agent()层级流程比顺序流程只多这一步。造完经理后,执行引擎、流水线逻辑全和 Day 20 一模一样。必须有 manager_llm 或 manager_agent★二选一:给 manager_llm(框架自动造经理)或给 manager_agent(你自己造好的经理)。两个都没有 → 构造期报错。manager 不能在 agents 里经理不能同时列进 agents——否则它既是调度者又是下属,逻辑打架。构造期就拦下。对比 Day 20 你会记得,顺序流程要求"每个 task 都有 agent"(
validate_tasks),层级流程没有这条要求——因为派活是经理运行时决定的,用户不必提前指定。两种流程的校验规则正好互补。L03
_create_manager_agent:自动造一个经理
核心方法(crew.py:1484),先看"自动造"分支:
# crew.py:1484
def _create_manager_agent(self) -> None:
if self.manager_agent is not None:
... # 用户自定义经理(L04)
else:
self.manager_llm = create_llm(self.manager_llm) # 把字符串/配置变成真 LLM
i18n = get_i18n(prompt_file=self.prompt_file)
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,
)
self.manager_agent = manager
manager.crew = self # 经理认得自己属于哪个 crew
create_llm(self.manager_llm)把用户传的 manager_llm(可能是 "gpt-4o" 字符串)实例化成真正的 LLM 对象。i18n.retrieve("hierarchical_manager_agent", ...)★经理的 role/goal/backstory 不用你写,从内置多语言模板里取。框架预置了一套"我是项目经理,负责调度"的人设。tools=AgentTools(...).tools()★最关键一行:把 self.agents(所有下属)打包成委派工具列表塞给经理(L05 详解)。allow_delegation=True经理必须允许委派——不然它拿着委派工具也不会用。manager.crew = self回填引用:经理知道自己在哪个 crew 里,才能访问 crew 的记忆、上下文等。自动造和自定义两条分支最后都执行这一句。💡 为什么经理的人设用 i18n 内置而不让用户写?因为"当好一个调度经理"是通用能力——它跟你的具体业务无关,写法还很讲究(要引导模型学会拆解、委派、验收)。框架把这套经过打磨的 prompt 预置进多语言模板,小白
manager_llm="gpt-4o" 一填就得到一个专业经理,不用自己琢磨 prompt。想深度定制的人再走 L04 的自定义分支。L04
自定义 manager_agent 分支
如果你自己传了 manager_agent(crew.py:1485):
# crew.py:1485
if self.manager_agent is not None:
self.manager_agent.allow_delegation = True # 强制打开委派
manager = self.manager_agent
if manager.tools is not None and len(manager.tools) > 0:
self._logger.log("warning", "Manager agent should not have tools",
color="bold_yellow")
manager.tools = [] # 清空工具
raise Exception("Manager agent should not have tools") # ★并直接报错
else:
... # 自动造(L03)
manager.crew = self
allow_delegation = True不管你原来设成啥,框架强行打开委派——层级经理不委派就没意义。manager.tools 不为空 → 报错★你自定义的经理不能带业务工具。带了会先告警、清空,然后 raise Exception 中止。(为什么?L07 详解)没有塞 AgentTools注意:自定义分支没有那句 tools=AgentTools(...)!委派工具是在别处注入的(_update_manager_tools,crew.py:1814),执行时才补上。⚠️ 边界:自定义经理带工具会直接崩
这段代码有个耐人寻味的细节:明明已经
manager.tools = [] 清空了,为什么还要 raise Exception?因为框架宁可让你显式失败,也不悄悄"帮你修好"。如果它默默清空继续跑,你可能一直以为"经理带着我的工具在干活",结果行为和预期完全不符、极难排查。直接 raise 逼你把工具从经理身上拿掉、想清楚职责边界。这是"不要静默纠正用户的错误配置"的设计立场。L05
AgentTools:把每个下属包装成"委派工具"
魔法核心(tools/agent_tools/agent_tools.py:16):
# tools/agent_tools/agent_tools.py:16
class AgentTools:
"""Manager class for agent-related tools"""
def __init__(self, agents: Sequence[BaseAgent]) -> None:
self.agents = agents
def tools(self) -> list[BaseTool]:
coworkers = ", ".join([f"{agent.role}" for agent in self.agents]) # 下属名单
delegate_tool = DelegateWorkTool( # 工具①:把整个任务派给某个下属
agents=self.agents,
description=I18N_DEFAULT.tools("delegate_work").format(coworkers=coworkers))
ask_tool = AskQuestionTool( # 工具②:只问某个下属一个问题
agents=self.agents,
description=I18N_DEFAULT.tools("ask_question").format(coworkers=coworkers))
return [delegate_tool, ask_tool]
coworkers = ", ".join(roles)把所有下属的 role 拼成一份名单,写进工具描述。这样经理调用工具时知道有哪些人可派。DelegateWorkTool★"委派工作"工具:经理调用它、指定 coworker=某角色 + 任务内容,框架就让那个下属 agent 去执行并返回结果。AskQuestionTool★"提问"工具:经理只想问下属一个问题(不是派整个活)时用。比如"你确认这个数据的来源吗?"。返回两个工具就这俩。所有下属共享这两个工具,靠 coworker 参数区分派给谁。图注:两个委派工具是"经理→下属"的桥。经理调工具时给出 coworker=角色名,工具内部路由到对应下属执行。
L06
为什么层级流程下 task 不必指定 agent
回到 _execute_tasks(Day 20),层级流程下每个 task 用哪个 agent 由这里决定(crew.py:1675):
# crew.py:1675
def _get_agent_to_use(self, task: Task) -> BaseAgent | None:
if self.process == Process.hierarchical:
return self.manager_agent # ★层级流程:所有 task 都先交给经理
return task.agent # 顺序流程:用 task 自己指定的 agent
# crew.py:1814 执行前给经理补上"委派工具 + task 允许的工具"
def _update_manager_tools(self, task: Task, tools: list[BaseTool]) -> list[BaseTool]:
if self.manager_agent:
if task.agent:
tools = self._update_manager_tools_for_agent(task, tools, self.manager_agent)
else:
tools.extend(self.manager_agent.tools or []) # 把委派工具并进来
return tools
_get_agent_to_use★层级流程下,不管 task 有没有指定 agent,都先交给经理。经理再用委派工具把它分给合适的下属。return task.agent(顺序)对比:顺序流程直接用 task 自己绑的 agent。这就是 Day 19 为什么顺序流程校验"每个 task 必须有 agent",层级不用。_update_manager_tools执行每个 task 前,把经理的委派工具并进本次可用工具集,确保经理手里始终有"派活"的能力。📝 例子:一个层级 crew 的运行
你配
运行时:
Crew(agents=[写手, 程序员], tasks=[t1], manager_llm="gpt-4o", process=hierarchical),t1="做一个能显示天气的网页"(没指定 agent)。运行时:
t1 交给经理 → 经理在自己的 ReAct 循环里想"这需要写代码" → 调 DelegateWorkTool(coworker="程序员", task="写天气网页代码") → 程序员执行、返回代码 → 经理觉得还差文案 → DelegateWorkTool(coworker="写手", ...) → 综合后给出 Final Answer。整个分派是经理运行时决定的,你没写一行"谁干什么"。L07
取舍:经理为什么不能带业务工具
💡 设计取舍①:为什么强制"经理不带业务工具"?
如果经理自己也能查资料、写代码,它就会倾向于"自己动手"而不是委派——那下属就成摆设,层级结构名存实亡。而且经理的 role/goal 是为"调度"调教的,让它兼职干专业活,两头都做不好。所以源码硬性剥夺经理的业务工具(自定义经理带工具直接 raise),逼它专注于"派活 + 验收"。这是"职责单一"原则在多 agent 系统里的体现:调度者只调度。
💡 设计取舍②:层级 vs 顺序,怎么选?
结论:能用顺序就别用层级。层级更灵活但更贵、更不可控——经理可能反复委派、绕圈子。只有当"任务拆分本身就需要智能决策"时,层级的额外成本才值得。
| 顺序流程 | 层级流程 | |
|---|---|---|
| 谁决定分派 | 你(写死) | 经理(运行时) |
| 成本 | 低(无经理开销) | 高(经理多轮 LLM 调度) |
| 可预测性 | 强(步骤固定) | 弱(经理自由发挥) |
| 适合 | 线性、明确的流程 | 目标模糊、需动态拆解 |
👶 小白:经理也是个 agent,那它会不会也陷入无限委派?
👨🏫 老师:会有风险,但经理和普通 agent 一样受 max_iter(Day 08)约束——转够圈数会被强制收尾。所以就算它反复纠结派谁,也不会真的无限循环。不过这也提醒你:层级流程要给经理配一个够聪明的 manager_llm,弱模型当经理容易乱派、绕圈、烧钱。调度是个高认知任务。
L08
今日小结
💡 一图记住层级流程
_run_hierarchical_process = _create_manager_agent() + _execute_tasks()。造经理 = 取内置人设 + 用
AgentTools 把下属打包成 DelegateWorkTool/AskQuestionTool 两个工具 + allow_delegation=True。运行时所有 task 都先交给经理,经理靠"调用委派工具"= "把活分给下属"来完成调度。执行引擎复用 Day 20 的流水线。🧠 今天你应该能回答
- 层级流程比顺序流程在代码上只多了哪一步?
- 经理"调度"的本质是什么?(调用委派工具)
AgentTools.tools()返回哪两个工具?各干什么?- 为什么层级流程下 task 不必指定 agent?由谁决定用哪个 agent?
- 为什么框架强制"经理不能带业务工具",还宁可 raise?
- 什么时候该用层级、什么时候该用顺序?
✋ 10 分钟动手
P=lib/crewai/src/crewai
sed -n '1479,1510p' $P/crew.py # _create_manager_agent
sed -n '1675,1680p' $P/crew.py # _get_agent_to_use
cat $P/tools/agent_tools/agent_tools.py # 两个委派工具
# 跑一个层级 crew,看经理如何委派(verbose)
python -c "
from crewai import Agent, Task, Crew, Process
w=Agent(role='写手', goal='写文案', backstory='资深文案', verbose=True)
c=Agent(role='程序员', goal='写代码', backstory='全栈', verbose=True)
t=Task(description='做一句 Python 的宣传语并给一行示例代码', expected_output='文案+代码')
crew=Crew(agents=[w,c], tasks=[t], manager_llm='gpt-4o', process=Process.hierarchical, verbose=True)
print(crew.kickoff())
"
明日预告 · Day 22:两种流程都读完了,明天回到"怎么把 crew 跑起来"——
kickoff 家族。同步的 kickoff、批量的 kickoff_for_each(每条输入复制一份 crew)、异步的 kickoff_async / akickoff,它们的关系、copy() 为什么关键、prepare_kickoff 开跑前做的一长串准备。