langgraph教程系列-11-从单兵到团队-多智能体模式
2026/8/6 0:44:19 网站建设 项目流程

本文是「LangGraph 教程系列」第 11 篇。写作时基于 langgraph 1.2.10、langchain 1.3.14、langchain-openai 1.4.1、Python 3.12+。配套代码仓库 https://github.com/wxj006007/deep-research-assistant ,本篇对应 tagv4

v3 已经把规划、研究和写作拆成了边界清晰的子图。可它们仍按固定顺序执行:先研究,再写作。资料缺了反方证据,谁来要求补查?资料足够时,谁来决定不要再浪费一次搜索?

拆图解决的是封装,团队化解决的是调度。本篇让研究助手拥有一名 Supervisor、研究员、审稿人和写作员。Supervisor 不亲自写答案,而是根据共享状态和专家报告决定下一位该工作的角色。

一、子图和 agent 不是一回事

子图是组合手段。v3 的 researcher 子图把审批、搜索和评估循环关在自己的职责边界内,父图不必知道它查了几轮。

agent 则多了角色、目标和调度权。研究员只补证据,审稿人只找覆盖缺口,写作员只基于已批准资料成稿。它们都可以用子图实现,但不会因为被装进子图就自动成为多智能体。

渲染错误:Mermaid 渲染失败: Parse error on line 11: ...摘要] persist --> end([END]) super ----------------------^ Expecting 'AMP', 'COLON', 'PIPE', 'TESTSTR', 'DOWN', 'DEFAULT', 'NUM', 'COMMA', 'NODE_STRING', 'BRKT', 'MINUS', 'MULT', 'UNICODE_TEXT', got 'end'

研究员内部仍是第 10 篇的子图,且每轮检索前仍会暂停等待人工审批。它不会因为变成一个专家,就得到绕过人类决定的权限。

二、三种拓扑,先选最小的一种

多智能体不是“角色越多越好”。常见形状有三种。

模式谁决定下一步适合什么
Supervisor一个中心调度者有统一交付物、需要审计和预算控制的研究任务
Hierarchical上级把工作继续委派给下级有稳定部门层级的大型组织流程
Swarm对等角色根据消息自行交接开放探索、角色可动态加入的任务

本篇选 Supervisor。用户只提出一个研究问题,最后也只需要一份答案;中心调度者能把轮数上限、取消和持久化条件放在一个看得见的位置。hierarchical 和 swarm 并不更高级,它们只是把控制权分散到更复杂的地方。

三、共享状态不是共享一切

专家应通过小而明确的字段契约协作,而不是互相读取提示词或私有变量。

classV4ResearchState(TypedDict,total=False):question:strdocs:Annotated[list[Doc],operator.add]expert_reports:Annotated[list[ExpertReport],operator.add]next_agent:Literal["researcher","critic","writer","finish"]supervisor_turns:intmax_supervisor_turns:intanswer:str

docs是累积的共享证据,expert_reports是同样累积的审计记录。报告必须带上作者和建议,这样 Supervisor 看见“需要继续研究”时,能知道它来自审稿人而不是猜测。

长期记忆、thread_idResearchContext沿用 v3。它们仍由父图统一管理:专家不能另开会话,也不能跨过写作节点直接保存研究摘要。

四、Supervisor 只分派工作

Supervisor 的输出不是自由文本,而是受限 JSON。它只可选择四个下一跳,并在分派给研究员时给出下一条查询。

defsupervisor_node(state:V4ResearchState)->dict[str,object]:response=get_llm().invoke([...])decision=_load_json(response.content)next_agent=str(decision.get("next_agent",""))ifnext_agentnotin{"researcher","critic","writer","finish"}:next_agent="researcher"ifnotstate.get("docs")else"critic"return{"next_agent":next_agent,"assignment":str(decision.get("assignment","")),"current_query":str(decision.get("query",""))orstate["question"],"supervisor_turns":state.get("supervisor_turns",0)+1,}

解析失败不能把图留在原地。示例采用保守回退:没有资料就派研究员,有资料就交给审稿人。写作员没有资料时也会被改派给研究员,避免它用空上下文编答案。

图的条件边只读取next_agent

builder.add_conditional_edges("supervisor",route_after_supervisor,{"researcher":"researcher","critic":"critic","writer":"writer","finish":"finish",},)

这让模型负责判断,代码负责边界。模型不能靠一句提示词跳进一个未注册节点,也不能自行执行持久化。

五、专家报告让返工可见

研究员完成其内部循环后,先写一份报告,再回到 Supervisor。审稿人检查全部资料并返回researcherwriter的建议。

{"agent":"critic","summary":"缺少不同持久化后端的取舍说明。","recommendation":"researcher",}

Supervisor 可以据此让研究员带着新的查询再跑一轮。这个返工不是把完整对话记录塞给每个模型,而是让下一位专家读共享资料和结构化结论。职责越清楚,后续接入来源核验员或合规审查员,越不容易把主图改成一团条件分支。

六、停止规则属于调度层

团队协作特别容易出现“审稿人说还不够,研究员又查一点”的无限回合。因此 v4 在父状态上保存supervisor_turnsmax_supervisor_turns,默认最多五次调度。

超过预算时,已有资料就交给写作员如实收尾;没有资料就进入finish节点,明确返回未收集到足够资料。人工拒绝仍由研究员子图返回cancelled,Supervisor 立即结束,且不会执行persist_memory

七、跑起来

代码在src/v4_multi_agent.py

python-msrc.v4_multi_agent

首次运行会在研究员的审批点暂停。用相同的thread_id传入Command(resume={"action": "approve"})后,图会从嵌套 checkpoint 恢复。演示会检查最终答案和调度次数;实际开发还应覆盖审稿返工、预算耗尽和人工拒绝三条路径。

到这里,研究助手已经会组织一个小团队,但它还只是本地 Python 进程。下一篇把这张图注册成服务,让客户端能创建会话、提交运行、流式读取结果,并在出问题时看到每个专家到底做了什么。

赞或收藏 关注 我们下次再见

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询