跳到主要内容

第 2 章 · 多 Agent 协作与计划编排

第 1 章讲的是"一个 Agent"。但复杂数据任务往往要分工:有人规划、有人写 SQL、有人调工具。本章讲 DB-GPT 怎么把多个 Agent 编成一个团队,按计划把活分下去。

2.1 它要解决的小问题​

"分析上季度各区销售并预警异常"这种任务,单个 Agent 一口吃不下。需要:先有人把它拆成有序的子任务(先查数据、再算指标、再判异常),再有人决定每一步派给谁(查数据派给 DataScientist,异常判断派给另一个),还要管好依赖(第 3 步要用第 1 步的结果)。

2.2 思路/直觉:Manager + Planner + Plan 存储​

DB-GPT 的多 Agent 用一个管理者 Agent(ManagerAgent)统筹,它本身也是个 ConversableAgent(所以也有五步循环),但它的 act 被改写成"推进计划":

  任务来了
     │
     ▼
┌──────────────────────────────────────────────┐
│ AutoPlanChatManager.act  循环 max_round 次:    │
│                                                │
│   plans = 读 plans_memory                       │
│      │ 没有计划?                                │
│      ├──► 叫 PlannerAgent 拆出计划,存进 memory   │
│      │ 有计划:                                  │
│      ├──► 取第一个待办子任务 now_plan            │
│      ├──► select_speaker 选谁来干                │
│      ├──► 把依赖步骤的结果拼进 prompt            │
│      ├──► speaker.generate_reply 干这一步        │
│      └──► 成功:complete_task / 失败:更新状态返回 │
│                                                │
│   所有子任务做完 → 返回最终结果                  │
└──────────────────────────────────────────────┘

怎么读: 这是一个"取待办 → 派活 → 标完成"的工作队列循环,计划本身存在记忆里(plans_memory),所以可以跨轮、跨 Agent 共享。

2.3 团队的骨架:Team 与 ManagerAgent​

团队抽象在 dbgpt/agent/core/base_team.py。Team(base_team.py:68-116)管一组 agents,提供 hire(招人)、agent_by_name(按名找人)、select_speaker(选发言人,抽象)。

ManagerAgent(base_team.py:119-172)同时继承 ConversableAgent 和 Team——它既是个能跑循环的 Agent,又管着一个团队。一个细节很说明设计意图:它的 thinking 被改写成"不额外思考"(base_team.py:143-158),因为管理者按流程和计划办事,不需要自己再发挥,只把收到的消息透传。

2.4 真实实现:AutoPlanChatManager.act​

核心在 AutoPlanChatManager.act(dbgpt/agent/core/plan/team_auto_plan.py:154-317)。拆开看:

没有计划 → 叫 Planner 拆(team_auto_plan.py:182-208):

# team_auto_plan.py:189-205  真实片段(精简)
planner = (await PlannerAgent().bind(self.memory).bind(self.agent_context)
           .bind(self.llm_config).bind_agents(self.agents).build())
plan_message = await planner.generate_reply(
    received_message=AgentMessage.from_llm_message(
        {"content": message.content, "rounds": rounds}),
    sender=self, reviewer=reviewer,
)

PlannerAgent 自己也是个普通 Agent,跑五步循环产出一份计划,计划被存进 plans_memory(SQL/内存表)。注意 .bind_agents(self.agents)——Planner 知道队里有哪些角色,才能在拆任务时"预分配"执行人。

有计划 → 取待办、派活(team_auto_plan.py:209-269):

# team_auto_plan.py:210-241  真实片段(精简)
todo_plans = [p for p in plans
              if p.state in [Status.TODO.value, Status.RETRYING.value]]
if not todo_plans:
    return ActionOutput(is_exe_success=True, content=final_message)  # 全做完了
now_plan = todo_plans[0]
# 选下一个发言人:计划里预分配了就用 pre_allocated,否则让 LLM 选
speaker, model = await self.select_speaker(
    speaker, self, now_plan.sub_task_content, now_plan.sub_task_agent)

处理依赖(team_auto_plan.py:53-86、242-251):如果 now_plan.rely 标了依赖步骤号,process_rely_message 会把那些步骤的子任务内容 + 执行结果拼成 rely_messages,塞进当前子任务的上下文——这就是"第 3 步用第 1 步结果"的实现。

派活并记结果(team_auto_plan.py:259-302):

# team_auto_plan.py:259-288  真实片段(精简)
agent_reply_message = await speaker.generate_reply(
    received_message=current_goal_message, sender=self, reviewer=reviewer,
    rely_messages=AgentMessage.from_messages(rely_messages))
if agent_reply_message.success:
    plan_result = agent_reply_message.action_report.content
    self.memory.plans_memory.complete_task(
        conv_id, now_plan.sub_task_num, plan_result)   # 标完成,存结果
else:
    self.memory.plans_memory.update_task(..., Status.FAILED.value, ...)
    return ActionOutput(is_exe_success=False, content=plan_result)  # 失败即停

关键:派出去的 speaker.generate_reply 就是第 1 章那套五步循环——子 Agent 自己会 think→act→verify→纠错。所以多 Agent 不是另起一套逻辑,而是"用 Manager 反复调用单 Agent 循环"。

2.5 怎么选下一个发言人​

select_speaker(team_auto_plan.py:99-152)两条路:

  • 预分配:计划里 now_plan.sub_task_agent 指定了执行人,直接用(team_auto_plan.py:109-112)。
  • 自动选:没指定时,让 selector(就是 Manager 自己)thinking 一把,prompt 是"读任务内容,从这些角色里选合适的来干"(team_auto_plan.py:115-130),再用 mentioned_agents 从模型回答里解析出唯一被提到的角色名(team_auto_plan.py:138-145)。

这是个"LLM 当调度器"的设计:用一次轻量的 LLM 调用,把自然语言的任务描述映射到团队里的某个角色。

2.6 关键细节 / 坑​

  • Manager 不重试:ManagerAgent.max_retry_count = 1(base_team.py:134-136),因为真正的纠错重试发生在被派的子 Agent 内部,Manager 层再重试是浪费。
  • 计划失败即停:某子任务失败,act 直接返回失败(team_auto_plan.py:300-302),不会硬着头皮往下做依赖它的步骤——避免错误传播。
  • Planner 拆计划也会重试:act 里若连续多轮建不出有效计划,试到第 4 次就放弃报错(team_auto_plan.py:182-188)。
  • round 上限保护:整个推进循环受 self.max_round 限制(team_auto_plan.py:172、314-317),防止子任务反复 RETRYING 把会话拖到天荒地老。
  • system prompt 里有硬约束:默认人设模板明确写了"严禁在计划里直接调 tool,所有工具调用必须经 ToolExpert 代理"(profile/base.py:32-35)——这是把"规划"和"执行工具"职责分离的设计约定。

2.7 代码地图​

主题文件路径(相对 packages/dbgpt-core/src/)关键符号
团队抽象dbgpt/agent/core/base_team.pyTeam, ManagerAgent
计划编排主体dbgpt/agent/core/plan/team_auto_plan.pyAutoPlanChatManager.act
选发言人dbgpt/agent/core/plan/team_auto_plan.pyAutoPlanChatManager.select_speaker
处理依赖步骤dbgpt/agent/core/plan/team_auto_plan.pyprocess_rely_message
Plannerdbgpt/agent/core/plan/planner_agent.pyPlannerAgent
计划存储dbgpt/agent/core/memory/gpts/GptsPlansMemory, GptsPlan
角色解析工具dbgpt/agent/core/agent_manage.pymentioned_agents, participant_roles