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_agentcrew.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_toolscrew.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 参数区分派给谁。
数据结构:经理靠两个工具把下属当"工具"调 Manager Agent allow_delegation=True 🛠 DelegateWorkTool 派活 🛠 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 开跑前做的一长串准备。
← Day 20 顺序流程 Day 22 · kickoff 家族 →