企业官网建设流程全解析

不用理论模型，就聊真实落地。最近半年我们团队在电力调度辅助决策系统里，用 LangGraph 把一套多智能体协作流程推上了生产环境。这中间经历了从迷恋 AutoGen 的群聊机制、到被复杂对话轮次折磨，再到回归 LangGraph 的显式图控制，最终稳定支撑每日数千次查询的过程。这篇文章不聊概念，只讲几个只要做多智能体项目就绕不开的工程问题：拓扑怎么选、状态怎么管、工具调用怎么不跑飞、断点续跑和并发怎么处理，以及凭什么敢让 AI 直接面对生产数据。

1. 为什么最终选型 LangGraph：被 AutoGen 的轮次机制坑过一次之后

先说结论：如果你的智能体之间不是平等的闲聊关系，而是有明确的主从、上下游、审批链，LangGraph 的显式图结构比 AutoGen 的自动对话轮次可控得多。

我们最早的原型用的是 AutoGen，看中的是它的 GroupChat 机制——多个 Agent 像会议室里的人一样自由发言，听起来很美好。但真跑到第二轮，问题就来了：轮次控制极其痛苦。你要么设置max_round让它自己聊，结果它可能在一个分支里反复横跳；要么手动干预发言顺序，那基本等于放弃了框架的全部优势。而且 AutoGen 的对话历史是线性累积的，一旦某个分支错误，整个上下文的可信度都受影响，没法精准回退。

LangGraph 不一样。它把整个协作过程建模成一个有向图，每个节点是一个处理函数，每条边是条件路由。你可以精确控制"谁在什么条件下调用谁"，每一步的输入输出都显式定义。这个特性对电网调度这种场景至关重要——我们不允许两个模型自由讨论后自行决定倒闸操作顺序，每一步必须可控、可审查、可回退。

另外一个实际原因是状态管理。LangGraph 的StateGraph围绕一个全局状态对象做流转，每个节点返回的 dict 会合并进状态。这意味着你可以把原始查询、中间检索结果、每个智能体的结论、最终响应全部保留在状态里，随时取用。AutoGen 在这方面的抽象要弱得多，你主要靠 conversation history 传递信息，结构化数据反而难放进去。

注意：选型时别只看 Demo 效果。AutoGen 的群聊在演示场景里非常惊艳，但惊艳恰恰来自它的自由度。生产系统要的是约束，不是自由。

2. 多智能体拓扑设计：我们最终选定的是"路由 + 执行 + 审查"三层结构

LangGraph 官方文档里画了 Supervisor、Hierarchical、Handoff 好几种拓扑，OpenAI 的 Swarm 还带了个 Agent 间交接的概念。但落到真实业务里，我建议你忘掉这些花哨名词，先想清楚一个问题：你的系统里，到底谁对最终结果负责？

我们第一版抄了"Supervisor + 多个 Worker"的官方示例，让一个调度员 Agent 决定把任务分给谁。很快发现问题：Supervisor 的路由决策本身就会出错，而且它错得毫无规律——有时候把设备状态分析任务分给了气象 Agent，有时候又把气象任务分给了拓扑分析 Agent，完全看模型当时的情绪。

后来改成三层结构，稳定下来了：

• 路由层（Router）：只做一件事，把用户查询分类到明确意图槽位。上下文只有一条系统提示 + 当前的用户原始输入，不携带其他任何 Agent 的中间结果，避免干扰。
• 执行层（Workers）：每个 Worker 是单一职责的专家型 Agent，只处理一个领域（设备台账、气象预警、拓扑分析、调度建议）。
• 审查层（Reviewer）：一个独立的质检 Agent，检查执行结果的完整性和内部一致性。

这个结构的核心思想是：路由决策不依赖上下文，执行过程不跨域，审查结果不参与执行。每一层都是独立可测的，出了问题你知道是哪个环节的锅。实际运行下来，Router 的准确率稳定在 97% 以上（在 300 条测试集上），Workers 各自只处理自己的领域，Prompt 可以写得很聚焦。

2.1 为什么不做 Chat Handoff

Handoff 模式在客服场景很好用，Agent 之间可以互相转单。但我们做下来发现这是把"上下文传递错误"问题从单 Agent 内部扩散到了整个系统——A 转给 B 时，A 的中间推理过程如果带了错误信息，B 无法察觉，因为它默认 A 是可信的。在电力这种容错率极低的行业，任何一环不可信都不可接受。宁可做显式的"重新检索 - 独立推理 - 交叉验证"，也不做隐式的信息交接。

2.2 状态字段按需最小化而不是把所有数据塞进去

LangGraph 的 State 是个类字典结构，很灵活，但也容易让人滥用。我们初版把所有 Agent 的输出都平铺进 state，后来状态里堆了二十多个字段，有 AssistantMessage、HumanMessage、Agent 专属记忆、工具返回的长文本。Prompt 组装时不仅要筛字段，还得注意消息顺序，维护成本直线上升。

现在我们的 State 只保留三类字段：

• query：原始查询，永不改变
• intermediate_results：检索、工具调用的结构化返回，按产生顺序追加
• agent_outputs：每个 Agent 的结论，以 Agent 名字为 key

消息列表不直接存全局 state，而是每个节点内部自行组装。这样状态对象体积可控，传给 LLM 的 token 消耗也降了大约 40%。

3. 工具调用（Tool Calling）在多智能体里的工程化：从裸调 Prompt 到受控执行

多智能体能落地，靠的不是 Agent 聊天能力，而是工具调用质量。LangGraph 本身不提供工具注册体系，它只是图编排引擎。你仍然需要自己管理 Function Calling 的 Tool Schema、执行过程、错误恢复。这一块我们踩的坑最深，也是回报率最高的优化点。

3.1 Tool Schema 必须用独立字段描述参数约束，别依赖模型"理解"

焱联网上很多 LangGraph 教程喜欢用@tool装饰器一把梭，函数签名直接给模型看。浅层 Demo 可以，生产环境不行——模型会自由发挥参数。我们的设备台账工具接受device_id，初版 schema 只写了"设备的唯一标识"，结果 model 传了设备名称、设备 IP、甚至经纬度，五花八门。

后来所有工具的描述改成三个强制字段：required、additionalProperties: false、以及每个参数的description都以枚举形式列出取值范围。OpenAI 和新浪的 Function Calling 都原生支持 JSON Schema，你只要认真写 schema，模型就不会乱填参数。实测 device_id 传错率从 18% 降到 2% 以内。

3.2 工具执行必须和 Agent 状态隔离

LangGraph 节点里执行工具时有个隐患：工具异常如果直接抛出，会污染整个 state。比如查询气象数据库超时，异常信息一旦被塞进 ChatHistory，模型会以为"工具调用了但没返回，说明该区域无气象数据"，然后基于这个错误认知继续推理，产出一个看似合理实则错误的结论。

我们的做法是给工具执行加一层 wrapper：超时、限流、数据结构校验异常全部转化为结构化错误码返回，错误详情单独存到intermediate_results的error字段，同时附带上一次成功调用的缓存结果（如果有）。模型看到的是"工具执行失败，原因 code_5002，近一次成功数据是 xxx"，它可以决定是否重试或换工具，但系统不会因为一次工具异常就丢失上下文。

3.3 并发工具调用：一次性并行 vs 逐步串行

LangChain 提供的bind_tools+tool_results支持一次返回多个工具调用请求，LangGraph 里可以并行执行。真实场景里我们推荐的做法是：同域类的工具并行，跨域类的工具串行。

比如查询同一个设备的台账、实时状态、历史检修记录，这三个是独立的，并行没问题，省了 2/3 的延迟。但"先查设备状态，再根据状态生成检修指导意见"这个链路，本质上有依赖关系，必须串行。你可以在 LangGraph 里用两个节点分步处理，或者在单个节点内控制工具执行顺序。我们统一在节点内部处理，这样状态流转的粒度不用拆得太碎，同时保持 LangGraph 图结构的清晰。

4. 状态持久化与断点续跑：Human-in-the-Loop 的真正工程实现

多智能体系统跑在你的服务器上，但最终决策可能落在人手上。电网调度场景里，Agent 生成的倒闸操作票、检修建议，绝对不允许自动执行，必须等值班调度员确认。这就是典型的 Human-in-the-Loop（HITL）需求。

LangGraph 的checkpointer提供了最基础的断点机制：图执行到任意节点时暂停，状态序列化到持久化存储（默认 SQLite，生产可换 Postgres），之后可以从暂停点继续执行。这个能力听起来简单，但真正要做到"人审的时候千万不能出岔子"，有几个问题必须单独处理。

4.1 Checkpoint 存储的是完整状态序列而不是当前快照

LangGraph 的默认MemorySaver只存在内存里，进程重启就丢，生产必换SqliteSaver或自己实现 Postgres 存储。而且你要理解它的机制：每次图节点的执行结果都会新增一个 checkpoint，这相当于状态序列，不只是最新状态的快照。好处是你可以回退到任意历史时刻重新执行，坏处是数据量增长很快。我们的经验是定期清理旧 checkpoint，只保留最近 N 次执行的和等待人工确认的活跃会话。

from langgraph.checkpoint.sqlite import SqliteSaver # 生产环境不要用 MemorySaver，重启即丢失 checkpointer = SqliteSaver.from_conn_string("postgresql://user:pass@host/langgraph")

4.2 人工确认节点要用 interrupt_before 而不是硬编码停住

我们第一版实现 HITL 是把"等待确认"写成一个节点，用一个 while 循环轮询数据库确认状态。这是灾难——图线程阻塞，所有并发请求都卡住。正确做法是使用图配置里的interrupt_before参数，让图在进入审查节点之前暂停并返回控制权，然后你的服务通过graph.invoke(..., config={"configurable": {"thread_id": "xxx"}})恢复执行。

恢复执行不是重新从头跑，而是从暂停的 checkpoint 继续。这意味着你必须保证在等待人工审查期间，上下游节点的 Prompt 和业务逻辑不能被修改，否则继续执行时可能出现"同一份状态、两个版本 Prompt"的隐性 bug。

4.3 人工确认的输入怎么进状态

LangGraph 2.x 里官方的做法是用Command(resume=value)把人工确认的结果注入状态。你可以在 next 节点里读取这个值，把它追加为 HumanMessage，或者直接覆盖某个中间变量。这里有个容易踩的坑：如果不显式用 Command(resume=...)，恢复执行时节点收到的输入里只有旧的intermediate_results，没有人工反馈，状态就断裂了。

4.4 超时和幂等：生产环境的隐藏要求

HITL 不光是暂停和恢复，你还得处理"人一直不确认"和"用户重复提交"两种情况。

• 超时：checkpoint 里要存created_at和expires_at，超过 24 小时未确认的会话自动标记为"已过期"，前端显示不可恢复。
• 幂等：用户点了两次"确认"，后端必须保证只 resume 一次。我们自己包了一层run_id的分布式锁，确认接口里先查run_id是否已消费，防止同一个人工输入被 LangGraph 执行两遍。

我这边的体会：HITL 不是 Agent 系统的附加功能，而是生产系统的骨架。如果你打算让 Agent 直接面向业务操作，从第一天就把"人审"设计进图里，而不是跑通了再补。

5. 流式输出与可观测性：从"黑盒调用"到"过程可回溯"

多智能体系统有个天然问题：单 Agent 你可以直接打日志看 Prompt 和 Completion，但多 Agent 之间的路由、工具调用、状态迁移，日志是分散的、交错在多个执行路径里的。没有可观测性，生产事故排查就是灾难。我们前期上线时最痛苦的就是"用户说回答错了，但不知道错在哪一层"。

5.1 LangGraph 的事件流 API 是你唯一需要关心的接口

LangGraph 的graph.stream()支持按事件、按节点、按更新三种模式。实测下来，最有用的是stream_mode="updates"——每次图流转到新节点，你就能拿到该节点返回的状态增量。我们把它接到一个事件总线，前端通过 WebSocket 实时展示"当前哪个 Agent 在做什么、检索了哪些数据、工具返回了什么"，用户不再觉得 AI 是个黑盒。

config = {"configurable": {"thread_id": "batch-20240112-001"}} async for event in graph.astream(input, config=config, stream_mode="updates"): for node_name, updated_state in event.items(): # 推送前端，同时落一份审计日志 logger.info(f"[{node_name}] {updated_state}")

5.2 钩子函数是追踪工具调用链的关键

LangGraph 的节点本身是普通 Python 函数，所以你可以在每个节点里手动埋点。但如果工具调用发生在节点内部的model.bind_tools(..., tool_choice=...)里，一个节点里可能发生多次 LLM 调用和工具执行，只打节点级日志会丢失细节。

我们统一在工具 wrapper 里埋了三个钩子：on_tool_start、on_tool_end、on_tool_error。每个钩子记录工具名、入参、出参摘要、耗时、错误码。LangSmith 也支持 tracing，但生产环境我会选择把关键链路打到自己的日志系统，因为 LangSmith 的采样率和大并发下的稳定性不如自建的日志可靠。

5.3 审计日志必须保留"原始查询 - 路由结果 - 工具入参 - 最终回答"的完整链路

多智能体系统一旦接入生产，你的日志就不是给开发者看的了，是给安全审计和业务复核看的。我们每条查询都会生成一个trace_id，贯穿 LangGraph 状态、日志、前端展示、数据库记录。审计界面提供"按 trace_id 重放执行过程"的功能——这比"看日志"好用的多，因为执行过程是结构化的状态变化，不是一行行文本。

6. 让它"下地干活"：FastAPI 集成与并发配置的几个硬经验

LangGraph 本身不关心你用什么 web 框架暴露接口，但一旦和 FastAPI 集成，有几个 LangGraph 特有的并发问题就出来了。我们踩过一个大坑，记录一下。

6.1 图实例是重资源，不能每个请求都重新构造

LangGraph 的编译图包含 PromptFactory、工具注册、模型客户端、checkpointer 连接池。如果在 FastAPI 的 request handler 里每次都调用builder.compile()，你的服务会在高并发时直接 OOM。

正确做法是把编译好的 graph 作为模块级单例，进程启动时构建一次，之后所有请求共享。如果不同业务线需要用不同的图，也建议用工厂函数 + LRU 缓存，而不是即时编译：

_graph_cache: dict[str, CompiledStateGraph] = {} def get_graph(domain: str) -> CompiledStateGraph: if domain not in _graph_cache: _graph_cache[domain] = build_domain_graph(domain) return _graph_cache[domain]

6.2 thread_id 是并发的天然隔离，但它不是并发控制

LangGraph 用thread_id隔离会话状态，不同 thread_id 的图执行互不干扰，这天然适合多用户并发。但要注意线程隔离不等于线程安全——同一个 thread_id 同时被两个请求 invoke，checkpoint 会相互覆盖，状态直接乱掉。

我们强制所有 WebSocket 请求在网关层按 thread_id 做互斥，一个 thread_id 同一时间只允许一个执行请求在途。这比在 LangGraph 应用层做锁更接近入口，能覆盖所有入口来源（同步 HTTP、异步回调、定时任务）。

6.3 FastAPI 的异步接口和 LangGraph 的 sync/async 要分清

如果你的业务大量依赖 LLM 调用和外部工具，async 是必要的——同步阻塞会让 worker 线程池迅速耗尽。LangGraph 提供ainvoke/astream异步接口，模型调用本身的耗时可以让出事件循环。但注意工具 wrapper 里如果用的是 sync 的 httpx 请求或 requests 库，会阻塞 event loop，必须用asyncio.to_thread包装：

def run_sync_query(sql: str) -> Result: ... res = await asyncio.to_thread(run_sync_query, sql)

6.4 模型上下文预算：多智能体放大了 Token 消耗

多智能体的 token 消耗不是单 Agent 的简单相加，Supervisor 路由会累积所有 Agent 的消息列表，Worker 之间传递结构化结果也会逐步膨胀。我们在 400 次压测里统计过，单轮完成一次"路由-检索-分析-审查"全链路，大约消耗 3000~5000 token（取决于中间检索的结果长度）。

省钱又保质量的两个措施：

1. 节点的 System Prompt 越短越好，领域知识走工具检索，不写死 Prompt。
2. 状态里只保留最新一轮的消息列表，中间轮次已经落库的不再反复发送。LangGraph 允许你在节点内自行控制传给 LLM 的 messages 子集，不必把整个 state 递给模型。

7. 让系统在真实数据上站稳：测试基线、重试策略与降级方案

多智能体系统上线不是写完图就完了，真正的工程实践在测试和运维阶段。我们的核心经验是：给每一种失败提前写好预案。

7.1 路由层的回归测试集是唯一的救命稻草

LLM 的非确定性让测试变得困难，但路由层一定是确定性最强的。我们建立了 300 条真实历史工单的回归集，每次修改路由层 Prompt，必须在回归集上跑一遍，准确率低于基线（97%）直接拒绝合并。这个数字听起来普通，但对我们而言是硬指标——路由错了，后面所有 Agent 的努力都白费。

执行层的测试更复杂，我们用"关键字段断言"而非"全文比对"：检查 Agent 输出里是否包含了正确的设备编号、是否存在幻觉结构、是否引用了检索结果中没有的数据。

7.2 LLM 调用必须封装重试，但重试要限流

大模型的 API 稳定性不代表 100%，生产环境必须考虑 429、超时、网络抖动。我们用tenacity做了一个统一的重试包装：首次失败等待 1s，第二次等待 3s，第三次直接放弃并把错误结构化返回。重试次数上限 3，防止雪崩。

但重试也会引入幂等问题——LLM 调用一次返回后网络断了，你以为失败重试，服务端实际已生成结果。这个问题在 LLM 层面没法完美解决，我们能做的只是：业务侧保证最终写入操作幂等，检索操作本身天然幂等，LLM 生成结果只做展示不落库两次。

7.3 降级策略：Agent 挂了，系统不能挂

单 Agent 挂了影响小，但多智能体是有依赖链的，上游挂了下游全部白干。我们给每个执行链路配备了降级路径：

• 路由层失败 → 回退到规则分类（基于关键词的意图匹配兜底）
• 某个 Worker 失败 → 跳过该环节，由审查层标记"信息缺失"并发给人工
• 审查层失败 → 允许执行结果以"未审查"状态进入人工队列

降级的关键不是设计得多优雅，而是要让最终用户明确看到"这个结果是降级产出的"。我们在前端对降级结果的标记是醒目的黄色警示条，避免 AI 结果被误当作全自动可靠结果。

8. 几个没写进标题但你早晚会撞上的细节

最后补充几个零碎的、但贯穿所有模块的判断，没有系统性的逻辑，却都是真实项目中验证过的。

8.1 LangGraph 版本升级要慎重

LangGraph 发展很快，0.x 到 1.x 的接口变化不小。如果你上了生产，不要追求最新版本，锁死一个版本，升级前必须跑完整回归。我们被 0.2.x 到 0.3.x 的StateGraphAPI 变更坑过一次，改起来不难，但排查成本很高。

8.2 企业级部署建议自己实现 checkpointer 存储

LangGraph 自带 SQLite 存储方便开发，但生产环境建议基于 PostgreSQL 实现。不要用自建的 JSON 文件快照方案，看起来方便，并发和一致性都会出问题。直接复用 SqliteSaver 源码思路，把存储层换成自己的连接池，十行代码的事，回报率很高。

8.3 "Agent 写代码"不如"代码写 Agent"

你不需要一个写代码的 Agent（Code Agent），你需要的是把业务规则写成代码，让 Agent 通过工具调用这些规则。这条判断我们在早期走了弯路——尝试让 LLM 自己生成业务判断代码，结果产出的代码经常是错的，而且比直接写死的规则还难维护。后来所有确定性逻辑全部下沉为 Python 函数，LLM 只负责：分类意图、抽取参数、选择函数、汇总结果。

这也带出一个多智能体设计的终极原则：不要让 AI 做它不擅长的事，只让它做好动词分类和上下文归纳。剩下的一切，交给工程。