企业官网建设流程全解析

接手一个有真实业务需求的多智能体任务时，很多人第一反应是让两个Agent“互相聊天”，理由是既然单个模型能力不够，那就让多个模型协作。这个起点其实是危险的——多智能体系统的复杂度从来不在“让多个模型对话”，而在状态管理、路由编排和异常恢复。LangGraph的出现恰好解决了这个问题：它把Agent的执行过程建模成一张有向状态图，每个节点是明确的动作，每条边是可控的流转条件，状态在节点之间显式传递，支持暂停、恢复、审计和返工。这套思路在生产落地时非常舒服。

这篇文章我会用一个“调研报告生成”的Demo作为主线，从环境准备开始，一步一步把核心架构搭出来。你会看到一个主管Agent如何调度三个不同角色的Worker，以及评审返工、条件路由、状态合并这些真正决定系统能不能跑起来的细节。文章里所有代码我都基于当前稳定版LangGraph API来写，换版本跑不通的时候，读代码注释也能快速定位问题。

1. 单智能体失灵的真实场景：为什么需要一张编排图

1.1 单Agent在多任务下为什么一定会崩

设想一个很常见的业务需求：用户说“帮我调研一下某个行业，写一份结构化分析报告，重点看市场玩家和未来趋势”。如果你只用一个Agent去处理，它内部要串行完成“检索资料→归纳整理→撰写报告→自我复查”。这个链路在Demo里往往能跑通，一旦任务变成“同时检索三个细分方向”“报告分为中英文两版”“最后要按指定模板输出”，单Agent的ReAct循环就会明显吃力。真正的瓶颈是记忆碎片化：工具调用的中间结果堆积在上下文里，到第三个子任务时，早期检索到的关键信息已经被模型“忘掉”了。

还有一个更隐蔽的问题：单Agent没有办法做真正的并行探索。它在一个上下文里依次调用工具，整个过程是串行的。多智能体架构的价值在于，可以把“检索A方向”和“检索B方向”拆给两个Worker并行执行，各自维护独立的工作上下文，最后再汇总到共享状态里。这种拆解方式，本质上是一次计算资源的重新分配，而不是简单的“多加一个模型调用”。

1.2 核心不是“多个模型对话”，而是“共享状态+控制流”

我对多智能体系统的理解是：它和微服务架构有相似之处——每个服务（Worker）只做一件事，服务之间通过明确的接口（共享状态）通信，由一个入口（Supervisor/Router）决定调用顺序。与其让Agent们自由地你一言我一语，不如让它们共享一张任务清单，各自只更新自己负责的那部分字段。这样系统的行为是确定的：某个节点执行后，状态里一定多出某个字段；某个条件不满足时，路由一定走回上一级。

这和“Agent对话式协作”是两种完全不同的设计哲学。对话式协作里，Agent之间的消息是自由文本，语义不稳定：一个Agent说“我做了”，另一个Agent可能理解为“做完了”，也可能理解为“正在做”。在图编排模式下，完成状态是显式的：状态里有一个results字段，Worker执行完就把产出写进去，主管节点只需要检查这个字段是否存在，语义不会产生歧义。

还有一个容易被忽略的点：图的每个节点都可以被打断和恢复。单Agent是一个不可分割的无状态循环，一旦中间某一步失败，要么重试整个循环，要么接受中间结果的丢失。而图结构天然支持“暂停在某个节点之前”“人工审批后再继续”“从最近一个检查点恢复”，这对生产系统是刚需。

1.3 LangGraph在同类框架里的位置

我实际对比过几种主流实现方式：

LangGraph和其他方案最大的区别是“显式控制流”。它不是让智能体自己决定下一步要调用谁，而是把路由规则、执行顺序、终止条件都写成代码。这看起来少了些“智能体自治”的浪漫感，但换来的东西更值钱：每一个决策步骤都可以被日志记录，每一条路径都可以被测试覆盖。

1.4 五个核心概念，用管道系统来理解

我习惯把LangGraph的运行机制比喻成一套水厂管道系统：

• State（状态）：相当于管道里的“水箱”，所有节点共享同一个水箱，节点往里面放水，也从里面取水。状态是TypedDict定义的，最简单的状态就是一个字典。
• Node（节点）：相当于一个“处理单元”，接收当前状态，做处理后把更新结果返回。在LangGraph里，节点就是一个接收state参数并返回dict的函数。
• Edge（边）：相当于“阀门”，决定执行顺序。固定边表示处理完A一定去B，不需要任何判断。
• Conditional Edge（条件边）：相当于“分流阀”，根据当前状态里的某个字段，从多个出口里选择一个。
• Checkpointer（检查点）：相当于在每个阀门处装了个“记录仪”，每次节点执行完毕，就把整个状态保存一份。这些记录意味着系统可以随时中断、恢复和回溯。

结束节点（END）是图里的特殊终点，当流程走到这里，一次invoke才算真正完成。另一个特殊起点是START，所有图都要从它开始连接第一个节点。

2. 最小骨架：半小时跑通第一个状态图

2.1 环境准备与版本说明

先确认Python版本，建议3.10及以上。安装依赖用pip一把梭：

pip install -U langgraph langchain-core langchain-openai

这里解释一下为什么要有langchain-core：LangGraph的状态结构参考了LangChain的BaseMessage概念，很多现成工具函数也依赖core里的抽象。实际安装时它会被自动带上，但显式安装能避免某些旧环境里依赖版本错乱的问题。

接下来配置模型接口。我建议在开发阶段用一个OpenAI兼容接口的模型，意味着只要设置base_url和api_key就能切到任意同类型服务，不必绑定具体厂商。本地开发时我也会在环境变量里放一个固定值，避免在代码里写死密钥：

import os os.environ["OPENAI_API_KEY"] = "sk-xxx" # os.environ["OPENAI_BASE_URL"] = "http://localhost:8000/v1"

如果你还没有可用的API Key，也可以先用一个假模型把图的逻辑跑通：写一个返回固定文本的节点函数，不调用真实LLM。图的正确性验证和模型能力验证是两件事，先用假模型验证前者，能省下不少调试时间。

2.2 第一个StateGraph：把“一步一步思考”变成图

先写一个最小图：输入一句话，模型节点回一句话。重点不是功能，是理解图的结构语法。

from typing import Annotated, TypedDict import operator from langgraph.graph import StateGraph, START, END class ChatState(TypedDict): messages: Annotated[list, operator.add] def model_node(state: ChatState): # 真实项目里这里会调用LLM user_input = state["messages"][-1] reply = f"收到你的消息：{user_input}" return {"messages": [reply]} builder = StateGraph(ChatState) builder.add_node("model_node", model_node) builder.add_edge(START, "model_node") builder.add_edge("model_node", END) graph = builder.compile() result = graph.invoke({"messages": ["你好，这是第一句话"]}) print(result["messages"])

跑完这段代码，你的result["messages"]里应该有两段消息：用户输入和模型回复。这里有两个新手最容易困惑的细节：

1. messages字段用了Annotated[list, operator.add]。它的意思是：每当节点返回这个字段时，新值和旧值不是互相覆盖，而是做列表拼接。我们看到的“两段消息”正是因为两次状态更新（用户输入一次、模型回复一次）合并进了同一个字段。
2. START和END是LangGraph内置的两个哨兵节点，不参与实际计算，只用来标注图的入口和出口。所有节点都通过add_edge(SRC, DST)连起来，构图完成后再compile()，编译这个动作会检查图的连通性和边的合法性。

2.3 invoke、stream、astream：三种执行方式

invoke是同步执行，适合脚本和测试。astream是异步流式执行，生产环境最常用，因为可以逐条拿到中间事件。实际开发时我几乎不用invoke，都用astream加日志：

async for event in graph.astream( {"messages": ["你好"]}, stream_mode="values", ): print(event.keys()) # 每次状态被更新时，都会输出当前完整状态

stream_mode="values"的含义是“每次状态更新后输出最新完整状态”。还有updates模式，只输出每个节点返回的增量更新，两者在排查问题时有不同用途：values看全貌，updates看“哪个节点改了什么”。

到这里，你已经有了一个能跑的图。接下来要做的，是把一个节点的图扩展成多个节点的多智能体协作图。

3. 核心架构从零构建：主管-工人模式的多智能体团队

3.1 目标场景与团队分工

我选了一个很典型的场景来说明：让多智能体团队自动完成一份行业调研报告。这个任务足够复杂，需要多步协作；同时结果容易评估，不会陷入“这报告到底好不好”的玄学。

我规划的团队结构是：

• 主管节点（Planner/Supervisor）：只有一个职责——看当前状态，决定下一步派谁干活。它不亲自检索，也不亲自写报告。
• 研究员Worker（Researcher）：执行检索动作，把调研摘要写进共享状态。
• 撰稿人Worker（Writer）：根据研究员的产出撰写报告段落。
• 评审Worker（Critic）：审查报告质量，输出“通过”或“需要修改”的意见。

如果你愿意，可以把场景换成任何其他多步骤任务，代码骨架几乎不用动。这正是图编排的好处：角色的职责边界一旦定义清楚，增删一个节点只是几分钟的事。

3.2 状态Schema设计：架构的第一步，也是最重要的一步

多智能体系统里，状态Schema设计得不好，后面所有节点都会在“读到什么、写到哪里”上反复打架。我的建议是：先画出“谁产出了什么数据，谁会消费什么数据”，再定义字段。

对于调研报告团队，状态字段由四类需求推导出来：

from typing import Annotated, Any, TypedDict import operator def merge_dict(dest: dict, src: dict) -> dict: """自定义状态合并函数：把两个字典合并，同名字段后者覆盖前者。""" return {**dest, **src} class TeamState(TypedDict): task: str # 全局任务描述，所有节点的输入依据 history: Annotated[list, operator.add] # 运行轨迹，记录每一轮调度/产出 next_agent: str # 主管刚做出的路由决策 results: Annotated[dict, merge_dict] # 各Worker产出，键为worker名 iteration: int # 当前循环轮数，防止无限返工 approved: bool # 报告是否通过评审

这里有两个字段的声明方式需要特别注意，因为它们直接决定了多智能体系统会不会出现数据互相覆盖的问题：

• history用了operator.add，适合追加日志。主管每做一次路由决定、每个Worker每完成一次工作，都会往这个列表里追加一条文本。它跟上一章里messages字段的用法是一样的。
• results用了自定义的merge_dict函数。这是为了处理“多个Worker分别往results里写自己的产出”的场景。如果不做这个声明，一个Worker返回{"results": {"researcher": "..."}}时，整个results字段会被覆盖，之前写入的其他Worker结果全部丢失。这是个非常隐蔽的问题，我单独写进第4章详述。

3.3 让三个Worker复用一套“Agent工厂”函数

如果三个Worker各写一个完整节点函数，代码量会膨胀且难以维护。典型的多智能体项目里，Worker之间的区别往往只在系统提示词和调用的工具，逻辑结构是高度相似的。所以我用一个工厂函数来创建Worker节点：

def make_worker_node(worker_name: str, system_prompt: str, tools: list[Any]): async def worker_node(state: TeamState): # 拼接消息时只取最近N条，避免上下文无限膨胀 recent_history = state["history"][-5:] # 调用LLM：传入系统提示词、工具定义、近期轨迹 # 伪代码，实际实现取决于你的模型客户端 response = await llm_with_tools.invoke([ {"role": "system", "content": system_prompt}, {"role": "user", "content": f"任务：{state['task']}\n近况：{recent_history}"}, ]) # 如果有工具调用，执行工具并收集结果 tool_output = "" if response.tool_calls: tool_result = await execute_tools(response.tool_calls, tools) tool_output = f"，工具返回：{tool_result}" return { "history": [f"{worker_name}完成一轮工作{tool_output}"], "results": {worker_name: response.content}, } return worker_node

这个工厂函数有几个设计意图：

• 每个Worker只往自己的results键里写数据。因为results声明了merge_dict合并逻辑，所以多个Worker即使执行顺序有交错，也不会互相覆盖。
• 传给LLM的上下文是“任务描述+最近几条日志”，不是全量历史。全量历史放进状态本身已经够重，如果再每次全量丢给模型，Token会被快速耗尽。
• 我在每个Worker结束时会往history追加一条执行日志。这是多智能体系统可观测性的第一道保障：出了问题，直接从history就能看到哪个节点跑过、大致做了什么。

在make_worker_node基础上创建三个具体节点：

builder = StateGraph(TeamState) # researcher负责检索并产出调研摘要 # writer基于摘要撰写报告，critic负责审核 builder.add_node("researcher", make_worker_node( "researcher", "你是行业研究员，负责检索和整理事实。输出结构化的调研摘要。", search_tools, )) builder.add_node("writer", make_worker_node( "writer", "你是报告撰稿人，基于研究员摘要撰写完整报告。", [], )) builder.add_node("critic", make_worker_node( "critic", "你是评审专家，检查报告是否完整、准确、结构清晰。输出通过或不通过。", [], ))

3.4 主管节点：让一个LLM发出路由指令

主管节点看起来只是“一个普通节点”，但它的关键不在代码复杂度，而在提示词设计。主管要做的唯一事情，是从["researcher", "writer", "critic", "finish"]里选一个输出，决定下一步让谁干活。

async def supervisor_node(state: TeamState): # 告诉主管只许输出一个合法节点名 prompt = f""" 你是任务主管。当前任务： {state["task"]} 团队执行情况： {state["history"]} 目前已产出摘要： {state["results"].get("researcher", "无")} 请决定下一步派哪个角色执行。只能输出以下名称之一： - researcher：需要检索/补充事实时 - writer：需要撰写/修改报告时 - critic：需要评审报告时 - finish：报告已通过评审，无需再改 直接输出角色名，不要输出任何其他内容。 """ decision = await llm.invoke(prompt).text # 做一次白名单校验，非法输出一律视为结束 if decision.strip() not in ["researcher", "writer", "critic", "finish"]: decision = "finish" return { "next_agent": decision, "history": [f"主管调度：{decision}"], }

我没有让主管一次性生成“整个计划的JSON”，而是让它每次只做一个决定。原因很简单：LLM的输出一旦变长，可靠性就指数下降。让它每次输出一个短字符串，哪怕它犯糊涂，后面也有校验兜底。这个设计在工程上叫“最小决策单元”，是多智能体路由的核心思想。

3.5 条件边与循环出口：把路由决策变成图结构

有了主管节点和三个Worker节点，接下来要把“根据next_agent把流程导向不同节点”这个逻辑落到图上。LangGraph里用条件边实现：

AGENT_MAP = { "researcher": "researcher", "writer": "writer", "critic": "critic", "finish": END, } def route_next_agent(state: TeamState) -> str: return AGENT_MAP.get(state["next_agent"], "researcher") builder.add_node("supervisor", supervisor_node) # 主管执行完后，根据next_agent字段选择去向 builder.add_conditional_edges( "supervisor", route_next_agent, AGENT_MAP, )

这段代码是整个多智能体系统最核心的“十字路口”。我强调一下它的运行逻辑：route_next_agent函数从当前状态读取next_agent字段，返回一个“路由键”；LangGraph拿着这个键去AGENT_MAP字典里查表，找到对应目标节点。如果没有路由键，我用get(..., "researcher")做了兜底，默认回到研究员——这比直接抛异常要好。

现在需要把图和主管连起来。整个团队的拓扑是：

• START→ 主管节点
• 主管 → 根据路由 → 任意Worker
• 任意Worker → 回到主管
• 主管 → 路由到finish→END

这个循环结构不是我用代码“写死”的，而是把条件边和固定边组合出来的：

builder.add_edge(START, "supervisor") # 三个Worker都固定回到主管，等待下一次调度 builder.add_edge("researcher", "supervisor") builder.add_edge("writer", "supervisor") builder.add_edge("critic", "supervisor")

注意：这里没有显式添加“主管→Worker”的边，因为条件边已经覆盖了。整个图会按照“主管→某Worker→主管→某Worker→……”的节奏运转，直到某个时刻主管输出finish。

3.6 加入评审返工与终止保障：让循环可结束

很多初版多Agent系统跑起来后最让人头疼的问题就是死循环：主管不断让Writer改稿，改到天荒地老。解决方式很简单，在状态里维护一个iteration计数，到限制后强制结束。

我在主管节点基础上加一段计数逻辑：

async def supervisor_node(state: TeamState): # ...原有调度逻辑... # 强制终止：达到最大轮次或报告已通过 if state["iteration"] >= 4 or state["approved"]: return { "next_agent": "finish", "history": [f"主管裁定：达到终止条件（iteration={state['iteration']}）"], } # ...正常调度逻辑...

然后所有Worker节点、包括主管节点在内，每次更新状态时都同步递增iteration。这里要特别小心：iteration是普通int字段，默认是覆盖式更新，多个节点同时返回同一个字段会“最后写入者胜出”。单线程执行时问题不大，但如果未来我们要并行跑多个Worker，就会遇到计数丢失。更多细节放在第4章讲。

评审返工的逻辑也可以加在critic节点内部：如果评审结论是“不通过”，就让approved保持False，主管自然会把流程导向Writer继续修改；如果评审通过，approved设为True，主管下一轮就会输出finish。

def critic_node(state: TeamState): # 调用评审模型，判断报告质量 # 伪代码 review_result = evaluate_report(state["results"].get("writer", "")) return { "approved": review_result.passed, "results": {"critic": review_result.comment}, }

到这里，一个完整的主管-工人多智能体团队已经成型。它的行为路径是：主管先派研究员检索，再派撰稿人写报告，再派评审人审稿；如果评审不通过，主管再派撰稿人改稿，如此循环，直到通过或轮次耗尽。整个过程完全由状态字段驱动，行为可预测、可审计。

4. 切到生产后，最先踩中的四个经典问题

4.1 多个Worker写同一个字典字段：last-write-wins的数据丢失

这是多智能体系统里出现频率最高的Bug，而且非常难一眼看出来。我在3.2里提到results字段要声明merge_dict，这里展开讲为什么。

假设你没有做合并声明，状态定义是：

class TeamState(TypedDict): task: str results: dict

然后研究员节点执行后返回：

{"results": {"researcher": "行业市场规模约300亿"}}

撰稿人节点执行后返回：

{"results": {"writer": "报告初稿..."}}

问题出现了：LangGraph的默认状态更新策略是整体覆盖。第二个返回执行完毕后，整个results字段变成{"writer": "报告初稿..."}，研究员辛辛苦苦检索到的摘要无影无踪。主管再往下一步走，看到的就是一份没有数据支撑的报告。

解决办法是给results字段加一个自定义reducer：

def merge_dict(dest: dict, src: dict) -> dict: return {**dest, **src} class TeamState(TypedDict): results: Annotated[dict, merge_dict]

加了声明后，每次节点返回{"results": {"writer": "..."}}，LangGraph都会把新字典与旧字典做合并，而不是覆盖。我强烈建议你在项目里把所有“键由不同Worker写的字典”都加上这个合并器，提前预防这类数据丢失。

4.2 条件边路由键对不上：InvalidUpdateError与fallback兜底

另一个高频问题发生在条件边配置阶段。某次我搭建团队时，主管输出角色名"researcher"，但条件边的映射表写成了"search_agent"，结果调用时直接抛出异常，错误信息形如InvalidUpdateError。这个错误表面上是状态更新失败，实际上是路由表里找不到对应节点。

更常见的版本是：LLM没按提示词输出，给了个"翻译一下"之类的自由文本，导致路由失败。我的经验是双保险：

首先是提示词强制收窄输出，见3.4里的“只许输出角色名”；其次在路由函数里做白名单兜底：

VALID_AGENTS = {"researcher", "writer", "critic", "finish"} def route_next_agent(state: TeamState) -> str: agent = state.get("next_agent", "researcher") if agent not in VALID_AGENTS: agent = "researcher" return agent

兜底的目标必须是个确定存在的节点，选researcher通常是安全的，因为它不依赖其他Worker的产出，风险最低。这个兜底策略让我在生产环境里少接了许多半夜告警电话。

4.3 检查点恢复的语义陷阱：thread_id决定“继续”还是“重来”

在LangGraph里，默认的compile()是不带检查点的。每次invoke都从START开始执行，执行结束状态就废弃了。一旦我们启用MemorySaver这类检查点机制，行为会变得非常不同：

from langgraph.checkpoint.memory import MemorySaver memory = MemorySaver() graph = builder.compile(checkpointer=memory)

现在执行同一张图两次：

config = {"configurable": {"thread_id": "session-001"}} graph.invoke({"task": "调研AI行业"}, config) graph.invoke({"task": "调研AI行业"}, config)

第二次调用不会从头开始，而是从session-001这个线程上一次执行到的位置继续。当然，在我们的主管-工人图里，上次执行已经走到了END，所以第二次invoke实际上是“没有新输入地重放一遍空循环”，结果会让人困惑。

正确理解：同一个thread_id代表同一个会话，跨调用共享状态。要开新任务，就要用一个新的thread_id：

graph.invoke(input_data, {"configurable": {"thread_id": "session-002"}})

这个特性真正发挥威力是在“人工审批”场景：图执行到某个高危节点前需要用户确认，系统会暂停；用户确认后，用同一个thread_id再次invoke时，流程会从暂停点继续，而不是从头跑一遍。这才是多智能体系统能在生产里落地的前提之一。

4.4 两个“小但不省心”的教训：消息全量塞入和工具异常被吞

第一个教训：把全量history无脑塞给LLM。我见过不少项目，状态里存了几百条消息，每个Worker被调用时都把这几百条消息全发给模型。结果是Token成本爆炸、响应延迟剧增。解决办法很简单：每个Worker只取自己需要的最近N条，长链结果存在results里，按需读取。

第二个教训：工具调用异常被吞掉。检索工具如果抛错，节点代码里如果写成：

try: result = await search_tool.invoke(...) except Exception as e: # 没有记录异常，也没有向上暴露 pass

那么下游节点会拿到一个空结果，然后生成一份看似正常但内容空洞的“报告”。这样的错误最难排查。我的习惯是工具异常一定要显式写进history，让主管的下一次决策能看到“刚才的工具调用失败了”，否则系统会把失败当成正常流程继续推进。

这里用一个表把这个章节的问题和应对方式汇总一下：

5. 从Demo到可运维：把多Agent图变成可观察、可恢复、可回归的系统

5.1 图结构可视化与debug模式

多智能体系统调试的第一步永远是“看清图长什么样”。LangGraph的编译结果自带图可视化能力。开发阶段我在Jupyter里最常用的一行代码：

print(graph.get_graph().print_ascii())

输出会以ASCII字符形式展示整张图，包含所有节点和边。节点名一旦多了，这张图就是排查数据流向的地图。如果部署在服务器环境，也可以把图导出成PNG，不过我更喜欢print_ascii，因为可以直接贴在Issue里。

在compile()时还有一个被低估的开关——debug=True。开启后，每次节点执行前后，框架会把输入状态和输出更新打印出来。第一次在多Agent项目里调试时，我靠这个开关定位了不下十个路由和合并问题。代价是日志量大，所以只建议在测试环境开启，生产环境用结构化日志替代。

5.2 把内存检查点换成持久化存储

上一章提到的MemorySaver只适合开发测试，进程一旦重启，所有状态归零。真实生产里，多Agent任务常常是长时间运行、需要跨进程恢复的，这时候要把检查点换成持久化实现。

以SQLite为例：

from langgraph.checkpoint.sqlite import SqliteSaver with SqliteSaver.from_conn_string("checkpoint.db") as saver: graph = builder.compile(checkpointer=saver)

流程再一次被打断后，我们只需要用同一个thread_id重新invoke，框架会从最近一个检查点继续执行。这种“断点续跑”能力，在人工审批、长时间检索、多轮返工的场景里价值极大。

有几点生产实践我建议直接照做：

• 每个会话/任务生成一个独立thread_id，不要复用。
• 关键节点的执行结果写入日志时带上thread_id，方便事后用同一ID恢复现场。
• 检查点数据属于敏感业务数据，注意备份和安全加密。

5.3 Mock工具回归测试：不烧Token就能验证流程正确性

多智能体系统最贵的测试成本是模型调用。每次调试都要消耗Token，而且模型输出随机，同一段代码可能这次通过、下次失败。所以我养成了一个习惯：先用Mock工具和固定回复把整张图跑通，再上真实模型。

具体做法：写一个FakeLLM，不管收到什么输入，都按预设规则返回固定文本。比如主管的假模型就固定返回"writer"，研究员的假模型固定返回"调研摘要"，撰稿人的假模型固定返回"报告正文"。只要假模型能让图跑通一次并进入END，说明图的拓扑、状态合并、边配置都正确；之后再把假模型换成真模型做行为测试。

class FakeLLM: def __init__(self, responses: dict): self.responses = responses async def invoke(self, messages): # 根据消息内容包含的角色关键词返回固定结果 text = str(messages) if "主管" in text: return {"content": self.responses.get("supervisor", "writer")} if "研究员" in text: return {"content": self.responses.get("researcher", "摘要")} ...

我把这个测试流程做成一个固定脚本，每次改完状态Schema或边配置，先跑一遍Mock全链路，再提交代码。这是多Agent项目和其他普通Python项目最大的不同：它多了一层“图正确性”需要验证，而图的Bug用真实模型几乎测不出来，因为模型会“聪明地”掩盖掉路由问题。

5.4 生产日志：看一眼当前进度

最后提一个非常简单但非常有效的习惯：在astream流式输出时，把所有状态更新点汇总成一个进度日志。

async for event in graph.astream(inputs, config, stream_mode="updates"): for node_name, update in event.items(): log.info("节点 %s 更新: %s", node_name, update.keys())

这样，每次主管调度、每个Worker执行，都会在日志里留下一条线索。在真实生产环境里，用户问“为什么报告还没出来”，我能直接查日志告诉他“系统已经完成检索，正在修改第二轮稿子”。这个体验和对着一个黑盒子猜完全不一样。

6. 多Agent模式选型：主管-工人之外，还有什么选择

6.1 三种常见模式横向对比

多Agent架构没有银弹，选型要看任务本身。我见过三种高频模式，分别适用于不同的业务形态：

主管-工人模式适合任务拆解不固定、需要逐步决策的场景。我们前面写的调研报告团队就是典型：先检索还是先写提纲，取决于当前任务状态，必须让主管动态决定。

6.2 手递手管线模式：固定流水线

手递手模式用固定边串起来，几乎没有条件边：

builder.add_node("extract", extract_node) builder.add_node("transform", transform_node) builder.add_node("load", load_node) builder.add_edge(START, "extract") builder.add_edge("extract", "transform") builder.add_edge("transform", "load") builder.add_edge("load", END)

适合的任务有明确的前后依赖关系，比如PDF解析→结构化→入库。这类任务的特点是没有“返工”需求，也不需要主管动态判断，加了主管反而画蛇添足。

6.3 层级团队模式：让一个子图成为另一个图的节点

如果任务复杂到一张图放不下，可以把子团队编译成一张子图，再把子图当作父图里的一个节点。这个模式特别适合大型组织型任务：比如高层主管拆出“市场调研团队子图”和“技术调研团队子图”，两个子团队并行推进，最后把结果合并。

research_team_graph = build_research_team_graph() tech_team_graph = build_tech_team_graph() builder.add_node("market_research", research_team_graph) builder.add_node("tech_research", tech_team_graph) builder.add_edge(START, "market_research") builder.add_edge(START, "tech_research") builder.add_edge("market_research", "merge_results") builder.add_edge("tech_research", "merge_results")

子图对于父图来说就是一个普通节点，父节点不需要关心子图内部的节点细节。这种分层抽象和代码模块化很类似，是大型多Agent系统里控制复杂度的核心手段。

6.4 什么时候应该放弃多Agent

最后说一点可能不太讨喜但非常重要的经验：多Agent不是越复杂越好。我的判断标准很简单——任务是否真的需要角色分离和状态隔离。

如果任务只是“检索几个资料然后写一段总结”，用单Agent+工具就足够高效，引入多Agent只会增加状态同步和路由失败的风险。多Agent真正适合的是那些需要不同角色长期协作、每个角色有独立上下文、需要评审返工的复杂任务。否则，一个带着工具列表的ReAct循环往往能做得更好，成本却低一个数量级。

我自己在项目里保留的路由原则是：先画任务流程图，如果能用一条直线画完，就用单Agent；如果出现循环、分支、返工，才考虑多Agent。把架构复杂度控制在问题复杂度的下一层，这个原则能省掉大量无意义的调试时间。

我实际用LangGraph构建多Agent系统的体会是：这个框架最大的价值不是帮你“造一个智能体”，而是帮你“默认就把数据结构、路由逻辑、状态恢复这些工程核心做对”。对比自研循环，你不需要自己写断点恢复；对比对话式多Agent，你不用靠解析对方的自由文本来判断下一步。当然它也有学习成本：状态合并策略、条件边映射、检查点语义都是必须理解的知识点。但一旦跨过这个门槛，多Agent系统会从“能跑通”进化到“敢上线”。