第 9 章 · 分层流程 hierarchical 与管理者
本章目标:掌握
Process.hierarchical的运行机制,学会配置manager_llm/manager_agent,能根据成本与可控性在时序与层级流程之间做正确选型,并掌握层级流程的调试技巧。
9.1 从"流水线"到"项目经理"
第 5 章的 sequential 流程像流水线:任务按预定顺序执行,上一棒输出喂给下一棒。hierarchical 流程则像公司组织:引入一个管理者(manager),由它动态决定"哪个任务交给谁做、做完要不要返工"。
官方文档对层级流程的定义是:任务不预先分配给具体 agent,管理者根据每个 agent 的能力进行规划、委派、审查产出并判断完成度。
import os
from crewai import Agent, Crew, Process, Task, LLM
llm = LLM(
model="openai/deepseek-chat",
base_url="https://api.deepseek.com/v1",
api_key=os.getenv("DEEPSEEK_API_KEY"),
temperature=0.7,
)
researcher = Agent(
role="市场研究员",
goal="收集并核实市场数据",
backstory="擅长快速检索与交叉验证数据的一线研究员。",
llm=llm,
)
analyst = Agent(
role="数据分析师",
goal="把原始数据转化为洞察",
backstory="精通统计与可视化的资深分析师。",
llm=llm,
)
writer = Agent(
role="报告撰写人",
goal="产出结构清晰的中文分析报告",
backstory="十年咨询行业报告写作经验。",
llm=llm,
)
# 层级流程下,任务只描述"要做什么",不指定 agent
research_task = Task(
description="调研 2025 年国内智能音箱市场的规模与主要玩家。",
expected_output="包含数据来源的市场规模摘要。",
)
analysis_task = Task(
description="基于调研结果分析增长驱动因素与风险。",
expected_output="3-5 条带论据的分析结论。",
)
report_task = Task(
description="撰写一份面向管理层的市场进入分析报告。",
expected_output="1000 字以内的结构化 Markdown 报告。",
)
crew = Crew(
agents=[researcher, analyst, writer], # 可用的人力池
tasks=[research_task, analysis_task, report_task],
process=Process.hierarchical,
manager_llm=llm, # 必填二选一:manager_llm 或 manager_agent
verbose=True,
)
result = crew.kickoff()manager_llm / manager_agent 必填
官方文档明确:使用 hierarchical 流程时,必须提供 manager_llm(字符串模型名或 LLM 实例)或 manager_agent 之一,否则 crew 无法启动。管理者由框架自动创建(或使用你指定的自定义管理者 agent)。
9.2 自定义管理者 agent
默认管理者是个"隐形人"。当你需要控制管理者的风格(比如更保守、更抠成本、要求中文沟通),可以显式创建 manager_agent:
manager = Agent(
role="项目经理",
goal="以最小的 token 成本按时交付高质量报告,严格把控每一步产出质量",
backstory=(
"资深项目经理。习惯先拆解任务再委派;"
"对含糊的产出一律打回重做;所有沟通使用简体中文。"
),
llm=llm,
allow_delegation=True, # 管理者必须允许委派
)
crew = Crew(
agents=[researcher, analyst, writer],
tasks=[research_task, analysis_task, report_task],
process=Process.hierarchical,
manager_agent=manager, # 用自定义管理者替换自动创建的默认管理者
)注意:使用 manager_agent 时,这个 agent 不要再放进 agents 列表——它是管理者,不是执行者。
9.3 层级 vs 时序:怎么选
| 维度 | sequential | hierarchical |
|---|---|---|
| 执行路径 | 固定(按 tasks 列表) | 动态(管理者规划+委派) |
| token 成本 | 低 | 高(管理者规划、审查、多轮沟通都耗 token) |
| 可预测性 | 高,易调试 | 低,同一输入路径可能不同 |
| 适用场景 | 步骤明确的标准流水线 | 任务分解本身需要智能判断的开放问题 |
选型经验法则:能用 sequential 说清楚的,就不要用 hierarchical。层级流程的价值在于"先做什么后做什么"本身需要推理的场景(如开放式调研、故障排查),而不是给普通流水线"升级"。
控制委派行为还有两个 agent 级开关:
allow_delegation=True/False:普通成员是否可以把工作转包给同事;- 管理者审查产出的严格程度,主要通过
manager_agent的 goal/backstory 来引导。
# 成员级委派控制:执行者只专注本职,不允许互相转包
researcher = Agent(
role="市场研究员",
goal="收集并核实市场数据",
backstory="擅长快速检索与交叉验证数据的一线研究员。",
llm=llm,
allow_delegation=False, # 关闭转包:避免成员间形成不受控的二次委派链
)把成员的 allow_delegation 统一设为 False、只保留管理者的委派权,是生产环境更可控的做法——所有调度决策都收敛到管理者一处,日志与成本也更容易归因。
9.4 调试层级流程
层级流程的不确定性让 verbose=True 成为刚需。推荐的三步调试法:
# 1. 打开详细日志,观察管理者的委派决策链
crew = Crew(..., process=Process.hierarchical, manager_llm=llm, verbose=True)
result = crew.kickoff()
# 2. 检查每个任务的实际执行者与产出(见第 12 章 CrewOutput 详解)
for i, task_output in enumerate(result.tasks_output, 1):
print(f"任务{i}: agent={task_output.agent!r}")
print(f" raw 前 80 字: {task_output.raw[:80]}")
# 3. 用 usage_metrics 评估层级开销是否值得(见第 12 章)
print(crew.usage_metrics)常见问题排查:
- 管理者反复委派同一任务:通常是成员的 role/goal 太相似,管理者无法区分能力边界——把角色职责写得互斥、具体;
- 产出质量不稳定:给每个 Task 写严格的
expected_output,管理者的审查标准会参考它; - 成本爆炸:先用 sequential 跑通业务逻辑验证产出质量,再评估是否有必要引入层级。
本章小结
- hierarchical 流程由管理者动态规划、委派、审查,任务不预先分配;
- 必须提供
manager_llm或manager_agent之一;manager_agent不要加入agents列表; - 层级流程 token 成本显著更高、路径不确定,只在"任务分解需要智能"时使用;
allow_delegation控制普通成员的转包能力;- 调试三板斧:verbose 日志、逐任务检查
tasks_output、用usage_metrics算成本账。
🧪 随堂测验
点击你认为正确的选项。答错时会展示正确答案与原因解析。
1. 使用 hierarchical 流程时,下列哪项是必须的?
2. 关于 sequential 与 hierarchical 的对比,正确的是?
3. 创建自定义 manager_agent 后,正确的做法是?
4. 层级流程中管理者反复把同一任务委派给不合适的成员,最可能的原因是?
🛠️ 动手实践
- 把第 5 章的 sequential 三人小队改造成 hierarchical 版本(先用
manager_llm),对比两者的usage_metricstoken 消耗差异。 - 再改用自定义
manager_agent,在 backstory 中要求"每个任务产出必须包含数据来源",观察产出质量变化。 - 故意把两个成员的 role 写成几乎一样,观察委派行为的不稳定性,然后修正并记录你的观察结论。
下一章解决"团队如何记住上次合作的经验":第 10 章 · 记忆系统。