企业官网建设流程全解析

从年初开始，我在几个内部分析类项目里陆续把单Agent链路改造成了Multi-Agent架构。说实话，最开始我对"多Agent"这件事是有点怀疑的——一个长Prompt加一堆Tool调用，大部分任务不也能跑吗？直到实际动手用LangGraph做了一套带路由、检索、报告生成的编排系统，并且把它部署到生产环境稳定跑了几个月，我才真正理解Multi-Agent的价值在哪里，也彻底摸清了从架构设计到生产落地的完整流程。这篇就当是这段时间的系统性复盘：从架构选型为什么这么定开始，到LangGraph的具体实现代码，再到生产级部署要面对的稳定性、可观测性和成本问题，最后把我踩过的坑全部摊开。整个过程算不上教科书式的标准路径，但每一步都是真实项目里验证过的。

1. 先回答"为什么"：单 Agent 链路到底卡在哪里

1.1 一个反直觉的结论：多数场景根本不需要 Multi-Agent

可能很多人跟我最开始一样，看到"Multi-Agent"就觉得它是解决一切 LLM 应用问题的银弹：检索不够准？上多Agent。回答不够好？上多Agent。延迟高？再上多Agent。但实际踩过一遍之后，我可以很负责任地说：一个能用单Agent加上工具调用解决的问题，引入多Agent只会让系统更慢、更贵、更难排查。

举个实际例子。我之前接过一个内部运营平台的数据问答需求，最开始是用一个大Prompt实现的，里面同时写了"从数据库取数"、"计算指标"、"生成解释文案"三段指令，再配合几个SQL和计算工具。跑起来之后出现一个很典型的症状：当用户的问题同时涉及取数和文案时，模型经常会把工具输出直接当成答案返回给用户，少了一层"把结果翻译成人话"的加工。我尝试用单Agent加多轮Function Calling优化Prompt，调了很久，效果依然不稳定。

这就是典型的"单Agent职责过载"。但请注意，我并没有直接就上多Agent框架。正确答案是先想清楚：这个问题是"一个Agent内部的分工混乱"，还是"确实需要多个Agent各自持有不同的状态和工具"？如果是前者，优化Prompt往往会更划算；如果是后者，才值得引入多Agent体系。

1.2 真正需要多 Agent 的四个信号

根据我后来的实践，下面这四个信号只要出现两到三个，就可以认真考虑多Agent架构了。

第一个信号是职责冲突。一个Prompt里塞太多角色要求，比如同一个模型既要当"严谨的取数工程师"，又要当"善于煽动情绪的市场文案"，这两类指令放在同一个上下文里一定会互相污染。拆开之后，每个Agent的System Prompt可以写得非常长、非常具体，而彼此之间互不干扰。

第二个信号是工具链存在严格的质量门槛。比如生成代码前必须先检索知识库，而检索结果又需要经过一次校验，校验失败就重新检索。这种"一个人既干活又做质检"的单Agent链路，一旦失败率上去，你根本不知道是哪一步出了问题。拆成Agent之后，每一步的输入输出都成为可审计的中间状态。

第三个信号是有人工介入环节。比如报告类任务，中间结论需要人确认才能往下走。单Agent里做人工确认只能靠Prompt里的"请稍等用户确认"这类软性指令，非常脆弱；多Agent框架里，这直接是一个图上的节点暂停点。

第四个信号是组件复用和独立迭代。当你要同时服务多个业务线，每条业务线只想改其中一个角色的行为、不动其他逻辑时，把角色拆成独立节点，改单点即可，不会牵一发而动全身。

2. 架构设计：把拓扑图画清楚，再写一行代码

2.1 四种主流编排模式的取舍

设计多Agent系统，第一件事不是写代码，而是定编排拓扑。我这些年接触过的方案归纳起来主要是四种，各自的取舍非常不同：

我自己的经验是：生产系统里90%以上的场景，一个Supervisor加3到5个Worker就够用了。层级式看着高级，但每一层调度都引入一次LLM调用和一次延迟，层级越多，系统越像一个黑盒。Swarm模式写起来很自由，但一旦任务链路变长，消息流转路径就不直观了，排查问题非常痛苦。所以我后面所有的生产项目都默认走Supervisor模式。

2.2 Supervisor 模式的设计要点

Supervisor模式在实现上有一个关键问题：调度Agent怎么知道"任务已经完成了"？很多新手会忽略这个问题，结果就是Supervisor不停地把任务派给各个Worker，直到触发最大步数。我在状态里放了一个显式的"完成信号"字段，比如next，它的取值只能是某个Worker节点名或者finish。Supervisor每次只做一件事：根据当前对话判断下一步该去哪个节点。Worker执行完，必须回到Supervisor，由它决定继续还是结束。整个图的收敛条件是清晰的——只要Supervisor输出finish，图就结束。

另一个设计要点是Worker的幂等性。因为Agent节点可能被重试，Worker每次执行时不能假设"我上次没跑过"。比如检索Agent要写成"查得到就返回结果，查不到就明确返回'未找到'"，而不是抛异常。抛异常会让重试机制反复调用同一个Worker，问题依旧；而清晰的失败输出，才能让Supervisor有机会切换策略。这一点在后面的生产部署章节里还会继续展开。

2.3 State 设计：多 Agent 系统最容易翻车的地方

多Agent系统与传统程序最大的区别是状态管理。传统程序里"谁调用谁"写死了，数据流是明确的；多Agent里所有节点共享一个图状态，任何一个节点都可以写状态，而且写入方式会互相影响。这会带来两个问题，我在项目里都踩过。

第一个问题是消息列表无节制增长。最简单的State设计是messages: Annotated[list, add_messages]，这个reducer会把所有节点产生的消息都追加进去。任务轮次一多，上下文窗口直接爆掉。关于怎么用RemoveMessage修剪，我会在踩坑章节专门讲。

第二个问题是把临时计算结果和对话历史混在一起。比如分析Agent算出了一个结构化结果，如果把这个结果也塞进messages，下一个Agent会把数字当成"别人随口说的一句话"，语义上极其混乱。正确做法是把结构化字段和消息字段分开：messages只放需要LLM看到的对话内容，中间的量化结果放独立的字段（比如analysis_result），这样既方便最后的Writer读取，也降低LLM被冗余信息干扰的概率。这套状态设计原则，直接决定了后面的工作流能写多复杂。

3. LangGraph 实战：一套带路由、检索、分析与报告的完整实现

3.1 为什么最终选了 LangGraph

在确定LangGraph之前，我对比过几个方向：完全自研状态机、CrewAI、AutoGen。先说结论，从生产稳定性角度，LangGraph的StateGraph设计是这几个选择里最接近"程序员思维"的。

自研状态机当然最可控，但要维护节点注册、状态流转、断点恢复、流式输出，工程量非常大，而且很容易在状态分支上漏边。CrewAI和AutoGen用抽象封装了很多东西，上手快，但有一个问题很致命：它们把Agent之间的交互方式藏得太深，出了问题你很难看清楚"上一条消息到底从哪来、到哪去"。LangGraph则把编排模型退回到"图"这个最朴素的概念——节点负责做事，边负责流转，条件边负责决策。所有中间状态都可以从checkpointer里取出来看，这对调试是巨大的优势。

另一个现实原因是LangGraph的add_messagesreducer机制。它允许每个节点返回的消息自动合并进共享状态，这个"自动合并"看起来简单，实则是多Agent协作的基础。我在后面的代码里会具体展示它和普通列表追加的差别。

3.2 定义 State 与搭建图骨架

下面这个例子来自我项目中一个真实工作流的简化版：用户提出一个分析需求，系统内部有"检索知识库的检索Agent"、"做数据加工的计算Agent"、"写最终报告的报告Agent"，另外有一个Supervisor负责调度所有节点。代码里的配置全部走环境变量，生产环境不应该有任何硬编码的模型名和密钥。

import os from typing import Annotated, TypedDict from langchain_core.messages import SystemMessage, HumanMessage, AIMessage from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langgraph.checkpoint.memory import MemorySaver class WorkState(TypedDict): messages: Annotated[list, add_messages] next_agent: str final_report: str # 统一走一个模型通道，换成你实际接入的模型标识即可 llm = ChatOpenAI( model=os.getenv("MAIN_MODEL"), base_url=os.getenv("LLM_BASE_URL"), api_key=os.getenv("LLM_API_KEY"), temperature=0, )

State的定义里有两个关键设计。messages负责累积所有Agent之间的对话，是图里流动的主干；next_agent是Supervisor的输出，也是一个显式的完成信号；final_report则放最终报告，用于后续存档和前端展示。

3.3 Supervisor 与 Worker 节点的实现

先看Worker。一个Worker节点就是一个普通的Python函数，入参是当前状态，返回值是被合并进新状态的字典。这里的关键是：节点不能靠全局变量保存自己的中间结果，一切都要通过返回值写入state。

@tool def route_tool(next_agent: str) -> None: """选择下一个执行节点，可选值来自ROUTE_OPTIONS。""" pass ROUTE_OPTIONS = ["retriever", "analyzer", "writer", "finish"] def parse_route(res) -> str: """解析路由结果，解析失败时默认走finish，避免异常路由扩散。""" if res.tool_calls: return res.tool_calls[0]["args"]["next_agent"] return "finish" def supervisor_node(state: WorkState): route_llm = llm.bind_tools([route_tool]) res = route_llm.invoke([ SystemMessage(content="根据对话内容，从可选节点中选择下一步。报告已成型时选择finish。"), *state["messages"], ]) return {"next_agent": parse_route(res)} def retriever_node(state: WorkState): # 伪代码：调用内部检索服务，必须给出明确成功或失败标记 answer = search_knowledge_base(state["messages"][-1].content) if not answer: return {"messages": [AIMessage(content="知识库中没有找到相关信息。")]} return {"messages": [AIMessage(content=f"检索结果：{answer}")]} def analyzer_node(state: WorkState): # 伪代码：基于检索结果执行计算 result = run_analysis(state["messages"][-1].content) return { "messages": [AIMessage(content=f"分析结果：{result}")], "analysis_result": result, } def writer_node(state: WorkState): final = llm.invoke([ SystemMessage(content="你是报告整理员，把前面讨论整理成最终报告。"), *state["messages"], ]) return {"messages": [AIMessage(content=final.content)], "final_report": final.content}

图的核心是连接关系。我把retriever和analyzer都指回supervisor，这样Supervisor在每个环节之后都能重新评估下一步；writer则是出口节点，写完报告直接进入END：

builder = StateGraph(WorkState) builder.add_node("supervisor", supervisor_node) builder.add_node("retriever", retriever_node) builder.add_node("analyzer", analyzer_node) builder.add_node("writer", writer_node) builder.add_edge(START, "supervisor") builder.add_conditional_edges("supervisor", lambda s: s["next_agent"], { "retriever": "retriever", "analyzer": "analyzer", "writer": "writer", "finish": END, }) builder.add_edge("retriever", "supervisor") builder.add_edge("analyzer", "supervisor") builder.add_edge("writer", END)

这里有个容易出错的地方：条件边的path函数返回的是字典里的key名，字典再映射回目标节点名。这个映射关系一定要和ROUTE_OPTIONS对齐，否则运行时会出现"图上找不到节点"的错误。我把这个检查写成了单元测试，每次改节点名都不会漏。

3.4 断点续跑、Human-in-the-loop 与 Streaming

多Agent系统上生产，几乎必然遇到需要人工介入的场景。LangGraph 通过checkpointer加中断来实现：第一次执行会在writer节点前暂停，系统保留完整状态；人工审查通过之后，再从暂停点继续执行：

from langgraph.types import Command memory = MemorySaver() graph = builder.compile(checkpointer=memory, interrupt_before=["writer"]) user_input = {"messages": [HumanMessage(content="分析近三个月各区交付量变化，并给出建议")]} config = {"configurable": {"thread_id": "task-001"}} result = graph.invoke(user_input, config) # 运行到这里会停在writer之前 snapshot = graph.get_state(config) print(snapshot.next) # 可以看到下一步是 writer # 人工确认后恢复 graph.invoke(Command(resume=None), config)

流式输出方面，LangGraph 支持多种stream模式。我生产上主要用stream_mode="updates"，按节点粒度拿到每步的输出，直接推给前端当进度提示；如果要逐字输出，再用stream_mode="messages"取token级别的增量。这里有个小经验：先按节点推进度，再按token推文字，不要让用户对着空白页面等几十秒。

thread_id是这套机制的命脉。它既决定了checkpointer把状态存到哪个槽位，也是后续恢复执行的唯一钥匙。后面在部署章节我会强调它必须对每次任务严格唯一。

3.5 运行效果与一次完整任务追踪

跑起来之后，一次典型任务的状态流转大概是这个轮廓：

START -> supervisor(路由到retriever) -> retriever -> supervisor(路由到analyzer) -> analyzer -> supervisor(路由到writer) -> writer -> END

每一步结束后，messages里都多了当前节点的输出，next_agent被Supervisor重新赋值。这套结构的好处是：任何一步效果不对，都可以从checkpoint里把快照拿出来，单步重放，而不需要整个任务重跑。我在开发阶段就是这么排查问题的——把生产环境某个失败的thread_id拉出来，用同样的状态反复修改单个节点看效果，定位效率比"重新问一遍模型"高太多。

4. 生产级部署：稳定、可观测、可控成本

4.1 部署形态、持久化与线程隔离

跑通Demo和上生产是两道完全不同的门槛。Demo里用MemorySaver就够了，进程一重启状态就没了；生产环境至少要换成磁盘或数据库级别的checkpointer。我用的是SqliteSaver，单机场景足够；如果有多实例横向扩容，就要上Postgres版本的checkpointer，保证多个实例共享同一份任务状态。

from langgraph.checkpoint.sqlite import SqliteSaver with SqliteSaver.from_conn_string("data/checkpoints.db") as saver: graph = builder.compile(checkpointer=saver, interrupt_before=["writer"]) result = graph.invoke(user_input, {"configurable": {"thread_id": task_id}})

这里有一个很容易被忽略的坑：thread_id就是多Agent世界的用户会话标识，它必须对每次任务唯一，并且不能跨用户复用。否则A用户的任务会看到B用户的历史消息，这在合规上是严重事故。我在网关层给每个请求强制生成UUID再写入配置，绝不让上层业务自己传。

部署形态上，如果团队已经有一套LLM网关，我建议直接用FastAPI把编译好的graph包成/invoke和/stream两个接口，自己控制鉴权、限流和灰度。不要为了省事把所有逻辑塞进一个同步接口里——多Agent任务动辄几十秒，一定要走异步任务，或者至少是SSE流式返回。

4.2 重试、限流、超时与紧急兜底

生产环境里，模型服务商不会永远可靠。我遇到过模型接口间歇性超时、返回格式错乱、偶发限流等情况，处理思路分三层。

第一层是节点级重试。LangGraph提供RetryPolicy配置，可以对单个节点设置最大尝试次数和退避间隔。注意：重试节点必须是可重入的，如果节点有副作用（比如往外部系统写数据），重试会导致重复执行，这时候要把重试放在真正有副作用的调用之外。

第二层是整体超时。我会给一次完整的graph调用设置超时上限，比如90秒；同时限制图的递归深度（recursion_limit），防止意外死循环把请求卡死。这个参数平时用不上，但它是我在生产上救过命的配置项。

第三层是紧急兜底。当某个下游服务挂了，Worker不要立即抛异常让整条链路失败。更好的做法是Worker返回一个"当前不可用"的明确标记，让Supervisor走备选路径——比如检索失败就退化为直接回答，或者告诉用户稍后再试。把失败变成图里可处理的数据，而不是未捕获的异常，这是多Agent系统稳定性设计和普通分布式系统最大的区别。

4.3 Tracing 与指标监控

多Agent系统里，模型调用、工具调用、路由决策是交织的。传统按请求打点很难看清"这次回答变差到底是因为哪个Agent说了什么"。我的做法是给每个节点加上结构化日志，输出节点名、耗时、token用量、路由结论，然后汇总到一个集中的日志平台。

每一个日志条目至少包含这些字段：

• trace_id：一次任务的唯一编号，对应用户的thread_id
• node_name：当前执行节点
• model_name、prompt_tokens、completion_tokens
• duration_ms
• route_to：Supervisor这次决定的下一节点

有了这套日志之后，排查问题的思路就变成了：先按trace_id拉出整条链路的节点序列，看哪一步耗时异常、哪一步路由到了意外的分支，再针对该节点做单步重放。比对着模型输出猜原因有效率得多。如果不想自己造，也可以用LangGraph配套的追踪看板，但基本的字段设计思路是一样的。

4.4 成本优化三板斧

多Agent系统的成本比单Agent更猛，因为每次路由都是一次LLM调用。我实际压过一轮之后，发现优化空间主要在三块。

第一是模型分级。Supervisor这种只做决策、不生成长文本的节点，用便宜的小模型足够；真正生成报告、写长答案的节点才用大模型。我做了个按节点配置模型的机制，成本直接砍掉30%以上。

第二是上下文瘦身。前文说的RemoveMessage修剪，不只是为了窗口安全，也是为了省钱。每次把老消息压成摘要，能显著减少重复计费。

第三是缓存。对于稳定知识库里的经典问题，我会在检索层做语义缓存，命中缓存就直接返回结果，根本不需要走Agent链路。对于完全一致的用户请求，还可以在网关层做全链路响应缓存，不过要注意业务上是否接受复用旧答案。

5. 实测踩坑记录：循环、上下文爆炸与路由不稳定

5.1 无限循环与 recursion_limit

第一次上多Agent的时候，我犯过一个特别典型的错误：Supervisor路由到retriever，retriever返回没找到，Supervisor不死心又路由到retriever，再返回没找到……直到recursion_limit触发报错。这种循环的根因是Supervisor缺少失败后的"第二条路"。

我的修法是两处。一是在状态里加round_count，每经过一次Supervisor就自增，当轮次超过预设值（比如6轮）时，无条件路由到writer，让它基于现有信息先写一版报告，别空转。二是给路由加finish路径的偏好，让Supervisor在"低信息量"时更倾向于结束而不是继续试探。这个偏好可以通过System Prompt强调，也可以通过示例样本灌输。在生产上，我宁可系统给一个"不够完美但完整"的报告，也不希望它无限循环烧钱。

5.2 上下文无限膨胀与 RemoveMessage 修剪

add_messages的自动累积会让所有对话都堆在messages里。8轮任务之后，光历史文本就可能吃掉两万多token。用RemoveMessage可以主动删除不需要的消息，让后续节点只看到必要的尾部上下文：

from langchain_core.messages import RemoveMessage def writer_node(state: WorkState): # ...生成报告... trimmed = [RemoveMessage(id=m.id) for m in state["messages"][:-6]] # 只保留最近6条 return { "messages": [AIMessage(content=final.content), *trimmed], "final_report": final.content, }

这里有个细节：删除操作必须和新增操作一起返回，因为reducer按顺序处理消息；如果分两个节点做，删掉的中间消息可能还会被重新追加。我一开始没注意，导致剪了半天，状态里的messages一点没少。这个坑很隐蔽，如果只靠看最终结果大概率发现不了，得盯着checkpoint里的状态看才能定位。

5.3 路由结果不稳定与结构化输出约束

Supervisor的输出如果直接让模型打一段文字，经常会出现"返回了节点名但是大小写不对"或者"从选项里编出个不存在的节点名"的情况。生产上我从不指望模型"自觉遵守格式"，而是用工具调用强行约束。让Supervisor必须调用一个名为route_tool的工具，工具的参数只允许枚举那几个节点名，模型就没有自由发挥的空间了。

就算这样，仍然存在概率性的解析失败。我在parse_route里加了默认分支：解析失败就走finish，宁可在这一步结掉，也不让一个异常路由把整个任务带偏。这是多Agent系统里一个很重要的设计哲学——让失败尽快显性化，而不是让错误在系统里传播。

5.4 回归测试：用预置样例把系统"钉住"

多Agent系统比普通代码更难测试，因为同样的输入，模型输出会有抖动。我的做法是维护一个二三十条的样例集，覆盖典型场景和边界场景，每次改动节点或Prompt后完整跑一遍，人工快速过一遍流程，重点看两件事：路由是否符合直觉，最终报告是否包含了必要的结论。这个回归集不需要很大，但要足够稳定。

更进一步的方案是把回归结果量化。我后来给每条样例加了一个"是否成功产出报告"的通过指标，以及在报告中命中断言的命中率。跑完自动算分，低于阈值就拒掉发布。虽然这套东西做得比较粗糙，但在没有更成熟的评测体系之前，它至少能防止"改一个节点把整个系统改崩了"。我一直认为，评估体系是生产级多Agent系统里最值得投入的部分，它的完善程度直接决定了你敢不敢频繁迭代。

最后再分享一句我最近的体会：多Agent不是一种"技术时髦"，它本质上是一种把复杂职责拆开并显式管理状态流转的工程方法。架构选型时多问自己"单Agent真的不行吗"，实现时多考虑"状态怎么发展、失败怎么收敛"，部署时多考虑"怎么观测和兜底"。把这几个问题想清楚，LangGraph 也只是个工具，真正值钱的是你对自己业务里"分工"和"协作"边界的理解。