摘要
给一个智能体一项任务，它能完成得不错。给它三项任务，第二次交接就可能出问题：它会忘记第一步找到的内容，会慷慨地给自己的草稿打高分，会在输出仍未完成时宣布成功。多智能体系统通过专业化分工解决这一困境。本文系统梳理2026年Python多智能体编排的四大主流框架选型逻辑，深入解析Supervisor、Handoff、Network等核心协作模式，并以LangGraph 1.0为技术底座，从零构建一个可运行的“研究员+写作者+审稿人”三角色协作流水线，包含条件重试边与状态管理机制，帮助读者完成从单智能体思维到多智能体协作思维的跨越。
关键词
LangGraph、多智能体协作、Supervisor模式、框架选型、条件边、状态管理
一、框架选型：2026年Python多智能体编排格局
2026年，Python多智能体编排领域由四个框架主导：Microsoft Semantic Kernel、LangGraph、AutoGen和CrewAI。对于零基础学习者，选择第一个框架比同时学习多个更重要。
1.1 LangChain + LangGraph：主线选择
LangChain 1.0是核心框架，负责模型抽象、提示词管理、工具定义和RAG组件。LangGraph 1.0是官方编排运行时，将工作流建模为有向图，节点代表处理步骤，边代表控制流。两者的关系是：LangChain提供构建块，LangGraph决定这些构建块如何组织和流转。
LangGraph的核心能力在于条件分支、循环、并行等复杂控制流，并支持状态持久化、断点续跑和人工介入。智能体的标准入口是create_agent（即create_react_agent的正式版本），旧的AgentExecutor已废弃。推荐的学习顺序是：ChatModel → Prompt → Tool → RAG → LCEL → LangGraph → create_agent。
1.2 CrewAI：轻量补充
CrewAI是四个编排框架中设置成本最低的。声明角色（如researcher、writer、reviewer），选择协作模式即可跑起来，适合快速原型验证。但它的控制粒度不如LangGraph精细，不适合需要复杂条件逻辑的生产系统。
1.3 选型建议
以LangGraph为主线，CrewAI为补充。LangGraph的StateGraph模型让你能够像搭建流程图一样构建多智能体工作流，状态管理和条件边的能力是构建可靠多智能体系统的关键。
二、多智能体协作的四种核心模式
在动手写代码之前，需要理解多智能体系统中几种基本的协作架构。
2.1 Supervisor（监督者模式）
一个中心监督者负责任务分解、分配和进度决策，多个专业Agent各司其职，完成后向监督者反馈。监督者通过条件判断决定下一步由谁执行。优势是集中控制、流程清晰、易调试；劣势是监督者是单点瓶颈，且每次路由都需要一次LLM调用。适用场景：研究报告生成、代码审查等多步骤流程任务。
2.2 Handoff（交接模式）
Agent之间直接交接控制权，没有中间协调层。这是OpenAI Agents SDK的核心设计模式。优势是去中心化、通信延迟低；劣势是交接链易形成环路、调试困难。适用场景：客服、用户对话等需要动态任务流转的场景。
2.3 Network（网络模式）
智能体间可以直接通信，形成网状拓扑。灵活但复杂度高，适合小规模、强交互的场景。
2.4 本教程的选择：图结构流水线
本文采用显式图结构的方式：不依赖模型自行决定下一步，而是在代码中预先声明节点、边和共享状态。研究流程是确定的——先研究，再写作，最后审稿，不合格则退回修改。这种流水线模式最适合用图结构来表达。
三、实战项目：研究员+写作者+审稿人流水线
3.1 系统架构设计
整个系统的结构如下：
text
START → researcher → writer → reviewer ──(条件边)──→ END（通过） ↑ │ └──────────────────────┘ (不通过，退回修改)
三个命名节点、一条通过边和一条重试边，以及一个汇集topic、notes、draft、verdict的共享状态对象。
3.2 定义共享状态
所有智能体共享一个类型化的状态对象，这是它们之间传递信息的唯一通道：
python
from typing import TypedDict, List, Annotated from langgraph.graph.message import add_messages class ResearchState(TypedDict): topic: str # 研究主题 notes: List[str] # 研究员收集的原始笔记 draft: str # 写作者产出的草稿 verdict: str # 审稿人的判定：PASS 或 REWORK feedback: str # 审稿人的修改建议 revision_count: int # 修改次数（防止无限循环） messages: Annotated[list, add_messages] # 对话历史
状态中的所有字段都是可读写的，但不同的节点负责更新不同的字段。revision_count是防止无限循环的关键——当修改次数达到上限时，即使审稿人不通过也强制结束。
3.3 创建专业智能体
使用create_react_agent创建三个专业Agent。每个Agent有独立的系统提示词和工具集：
python
from langgraph.prebuilt import create_react_agent from langchain_openai import ChatOpenAI model = ChatOpenAI(model="gpt-4o", temperature=0.1) # 研究员：负责收集和分析资料 researcher = create_react_agent( model, tools=[web_search, arxiv_search], prompt="你是一个资深研究员，专注于收集和分析技术资料。" "根据给定的研究主题，收集3-5个关键信息点，" "每个信息点包含来源和核心内容。" ) # 写作者：负责将研究笔记转化为文章 writer = create_react_agent( model, tools=[write_document, format_markdown], prompt="你是一个技术写作专家，擅长将复杂概念转化为清晰文章。" "基于研究员提供的笔记，撰写一篇结构完整的文章。" "如果收到审稿意见，根据意见修改草稿。" ) # 审稿人：负责质量把关 reviewer = create_react_agent( model, tools=[check_facts, grammar_check], prompt="你是一个严格的审稿人，确保内容的准确性和质量。" "评估文章是否满足以下标准：" "1. 事实准确，无编造内容" "2. 结构清晰，逻辑连贯" "3. 语言流畅，无语法错误" "输出JSON格式：{\"verdict\": \"PASS\" 或 \"REWORK\", \"feedback\": \"具体修改建议\"}" )注意create_react_agent的prompt参数——它定义了Agent的角色和行为准则。审稿人的提示词中明确要求输出JSON格式，这是后续条件边能够正确解析判定的前提。
3.4 编写节点函数
每个节点是一个Python函数，接收状态对象，返回要更新的状态字段：
python
def researcher_node(state: ResearchState): """研究员节点：根据主题收集信息""" result = researcher.invoke({ "messages": [{"role": "user", "content": f"研究主题：{state['topic']}\n请收集相关信息。"}] }) return { "notes": [result["messages"][-1].content], "revision_count": 0 # 初始化修改次数 } def writer_node(state: ResearchState): """写作者节点：基于笔记撰写或修改草稿""" if state.get("feedback"): # 有审稿反馈，是修改场景 prompt = f"""研究主题：{state['topic']} 研究笔记：{state['notes']} 上一版草稿：{state['draft']} 审稿意见：{state['feedback']} 请根据审稿意见修改草稿。""" else: # 首次写作 prompt = f"""研究主题：{state['topic']} 研究笔记：{state['notes']} 请基于以上笔记撰写一篇结构完整的文章。""" result = writer.invoke({"messages": [{"role": "user", "content": prompt}]}) return {"draft": result["messages"][-1].content} def reviewer_node(state: ResearchState): """审稿人节点：评估草稿质量，输出PASS或REWORK""" prompt = f"""请审阅以下文章： 主题：{state['topic']} 草稿：{state['draft']} 请评估文章质量，输出JSON格式的审稿结果。""" result = reviewer.invoke({"messages": [{"role": "user", "content": prompt}]}) import json content = result["messages"][-1].content # 从模型输出中提取JSON try: parsed = json.loads(content) except json.JSONDecodeError: # 处理Markdown代码块包裹的情况 import re match = re.search(r'\{.*\}', content, re.DOTALL) parsed = json.loads(match.group(0)) if match else {"verdict": "PASS", "feedback": ""} return { "verdict": parsed.get("verdict", "PASS"), "feedback": parsed.get("feedback", ""), "revision_count": state.get("revision_count", 0) + 1 }3.5 定义条件边
条件边是流水线的核心控制逻辑。审稿人返回结果后，系统根据verdict字段决定下一步：
python
def route_after_review(state: ResearchState) -> str: """审稿后的路由决策""" # 达到最大修改次数，强制通过 if state.get("revision_count", 0) >= 3: return "finish" # 审稿通过 if state.get("verdict") == "PASS": return "finish" # 审稿不通过，退回写作者修改 return "rework"最大修改次数限制是必须的。没有这个限制，审稿人可能永远不满意，导致无限循环和Token浪费。
3.6 构建图
将所有节点和边组装成完整的图：
python
from langgraph.graph import StateGraph, START, END # 创建图，指定状态类型 workflow = StateGraph(ResearchState) # 添加节点 workflow.add_node("researcher", researcher_node) workflow.add_node("writer", writer_node) workflow.add_node("reviewer", reviewer_node) # 添加边：线性流程 workflow.add_edge(START, "researcher") workflow.add_edge("researcher", "writer") workflow.add_edge("writer", "reviewer") # 添加条件边：审稿后的分支 workflow.add_conditional_edges( "reviewer", route_after_review, { "rework": "writer", # 退回写作者修改 "finish": END # 审稿通过，结束 } ) # 编译图 app = workflow.compile()条件边的三个参数分别是：源节点名称、路由函数、目标映射字典。路由函数返回映射中的键，LangGraph据此决定下一步走向哪个节点。
3.7 运行完整流水线
python
result = app.invoke({ "topic": "LangGraph在2026年的多智能体协作能力", "notes": [], "draft": "", "verdict": "", "feedback": "", "revision_count": 0, "messages": [] }) print(f"最终草稿：\n{result['draft']}") print(f"\n修改次数：{result['revision_count']}") print(f"审稿结果：{result['verdict']}")运行后，如果审稿人判定为REWORK，图会自动回到writer节点，携带feedback状态，写作者据此修改草稿，然后再次进入reviewer节点。这个循环最多执行3次，之后强制结束。
四、状态管理与持久化
4.1 为什么需要持久化
上面的示例使用内存中的状态，进程结束后所有状态丢失。生产级系统需要Checkpointer来持久化状态，支持跨会话恢复和断点续跑。
python
from langgraph.checkpoint.sqlite import SqliteSaver # 使用SQLite作为持久化存储 memory = SqliteSaver.from_conn_string("checkpoints.db") app = workflow.compile(checkpointer=memory) # 运行时指定thread_id config = {"configurable": {"thread_id": "research_001"}} result = app.invoke(initial_state, config)SqliteSaver将每一步的状态快照写入SQLite数据库，进程重启后可以通过相同的thread_id恢复会话。
4.2 消息历史的独立管理
在多智能体系统中，每个Agent可能需要独立的对话历史，而不是共享全部消息。LangGraph的子图（Subgraph）机制支持这一点：每个Agent可以作为子图运行，拥有自己的消息历史，只向父图暴露必要的输出。
五、常见问题与优化建议
5.1 无限循环
最常见的问题。审稿人永远不满意，写作者反复修改但无法达标。解决方案：在状态中记录修改次数，达到上限后强制通过。推荐上限为3次。
5.2 工具选择困难
给单个Agent配备过多工具会导致“选择恐惧症”。解决方案：让每个Agent只持有与其角色相关的工具。研究员只需要搜索工具，写作者只需要写作工具，审稿人只需要校验工具。
5.3 状态膨胀
随着修改次数增加，状态对象中的草稿和反馈会不断累积，导致上下文膨胀。解决方案：在写作者节点中，不将全部历史草稿传给LLM，只传上一版草稿和当前反馈。
5.4 成本控制
每次审稿和修改都是一次完整的LLM调用。优化方案：使用模型路由——研究员用轻量模型（如glm-4-flash）做并行检索，写作者和审稿人用高质量模型保证输出质量。
六、总结
本文从框架选型出发，系统梳理了LangGraph多智能体协作的核心概念与实践方法。关键要点可以归纳为五条：
框架选择：以LangGraph为主线，
create_react_agent创建专业Agent，StateGraph组织流程。协作模式：Supervisor适合集中控制，Handoff适合动态流转，显式图结构适合确定性流水线。
状态管理：所有Agent共享类型化状态对象，通过节点函数读写状态字段。
条件边：用路由函数决定下一步走向，实现“审稿不通过则退回修改”的循环逻辑。
防循环：在状态中记录修改次数，达到上限后强制结束，避免Token浪费。
从单Agent循环到多Agent图结构，是从“模型自行判断路径”到“代码显式控制流程”的转变。这个转变让你的Agent系统从不可预测的黑箱，变成可以在 diff 中读出的、可控可调试的工程结构。
