企业官网建设流程全解析

今年做企业级项目，我发现单 Agent 已经越来越难撑起真实业务：一个 Agent 既要理解用户意图，又要调数据库、检索知识库，还要决定什么时候问人、什么时候给结论，所有 Prompt 塞在一起，状态一乱就是灾难。所以我转向 LangGraph，用有向图的方式把多个专职 Agent 组合成一套可观测、可恢复的多智能体系统。这篇文章是我从零搭完一个生产可用的多智能体系统后，把架构选型、核心概念、代码实现、FastAPI + Docker 部署，以及边缘设备推理扩展完整梳理一遍的记录。代码可以直接抄，但我觉得架构思路比代码本身更值得看。

如果你已经用 LangChain 写过几个 Agent，但觉得它们“跑起来容易、控住难”，或者你正在为多智能体系统做技术选型，这篇很适合你。我会尽量说人话，把每个关键选择的“为什么”也讲清楚，避免你照完代码还是不知道为什么这么搭。

1. 先想清楚再动手：多智能体系统的架构选型

1.1 为什么我选 LangGraph 而不是自己写状态机

先说结论：LangGraph 本质上是一个“有向图状态机”，不是又一个 Agent 封装框架。它把一次任务拆成节点（Node）和边（Edge），节点执行动作，边决定流转，所有中间数据放进一个全局状态（State）。这个设计和我们平时手写if/else状态机很像，但它把“流程控制”和“业务逻辑”彻底分开了。

很多人会把 LangChain 和 LangGraph 放一起比较，实际上两者不是同一层的东西。LangChain 提供的是模型调用、Prompt 模板、工具封装、检索链路这些“零件”；LangGraph 则是把零件按图组装起来的“流水线”。我在实际项目中通常两个一起用：用 LangChain 的create_react_agent封装子 Agent，用 LangGraph 的StateGraph做顶层编排。如果只用 LangChain 的 AgentExecutor，流程一复杂，分支一多，代码就会堆满回调函数，排查问题时很难看清是哪一步出了问题。

相比之下，LangGraph 有几个明显优势：状态集中管理，每个节点只要关心自己读写的那部分状态；支持条件边，路由逻辑可以独立成一个函数；自带 Checkpointer，可以记录每一步执行快照，做到断点恢复和人工干预。这几个能力几乎就是企业级多智能体系统的刚需。

1.2 企业级多智能体的两种协作模式

我见过比较多的多智能体架构，归纳起来就是两张组织方式：一种是“中央调度”，另一种是“流水线接力”。

中央调度模式里有一个 Supervisor Agent 负责理解用户请求、拆解任务、然后决定把任务交给哪个子 Agent，最后汇总结果。它的优点是决策集中，权限边界清晰，适合企业里那种“入口统一、后端分工明确”的场景，比如工单系统、客服助理。

流水线接力模式则更像一条产线，每个 Agent 只负责一道工序，处理完把结果交给下一个 Agent。优点是链路简单、每个节点职责单一，但缺点是一旦中间某一步需要根据结果动态分流，代码就会变得比较复杂。LangGraph 的 Conditional Edge 对这种动态分支支持得很好，所以我一般在架构上采用“Supervisor + 流水线混合”：入口用一个 Supervisor 做粗粒度路由，执行阶段用多个按序或按条件触发的子 Agent。

这个设计的核心是为了控制复杂度。如果所有 Agent 全都互相可调用，看起来灵活，实际运行时会陷入循环调用，token 消耗成倍增长，问题定位也非常困难。企业级系统要的不是“理论上能解决所有问题”，而是“可预期、可观测、可降级”。

1.3 先画架构图再写代码：分层与角色划分

我强烈建议写代码前先用一张架构图把角色、边界、数据流定下来。下面这个分层是我在项目里实际使用的，你可以直接当模板：

接入层 FastAPI / WebSocket / 企业IM回调 ------------------------------------------------ 编排层 LangGraph Supervisor Graph（状态机 + Checkpointer） ------------------------------------------------ 执行层 SQL Agent 知识库 Agent 审批 Agent ------------------------------------------------ 数据层 PostgreSQL 向量数据库 对象存储

架构图不是画给别人看的，是给自己理清数据的流向。每个 Agent 的输入是什么、输出什么、写哪些状态、失败后谁来兜底，这些都得在图上标清楚。画图工具方面，我习惯用 Excalidraw 画快速草图，正式文档用 draw.io，两个都是免费的，导出 SVG 放在企业 Wiki 里也很好维护。

很多项目失败不是因为代码写不出来，而是因为没想清楚边界。画图时如果发现某个 Agent 要同时“查数据库、读文件、发通知、做总结”，说明职责过重，需要继续拆分。多智能体系统的粒度不是越细越好，我的标准是：每个 Agent 的 Prompt 能在一屏内写完，工具不超过 5 个，职责描述不超过三句话。

2. 环境准备与 LangGraph 核心概念实战

2.1 安装 LangGraph 与依赖

LangGraph 的安装很简单，直接在 Python 3.11 环境里执行：

pip install langgraph langchain-openai langchain-community

如果你要用 Postgres 做检查点存储，还需要langgraph-checkpoint-postgres。我建议一开始只用默认的内存检查点或者 SQLite，跑通流程后再上 Postgres。还需要准备一个大模型的 API Key。如果只是本地验证，可以先用langchain-openai配置 OpenAI 兼容接口，后面我会讲怎么切到本地模型。

我踩过的一个小坑是版本兼容性。LangGraph 迭代比较快，某些 API 在老版本里叫StateGraph.add_conditional_edges，新版本里仍然保留，但参数形式有所调整。所以我建议固定版本，用pip freeze > requirements.txt锁定环境，别随意升级。安装完后可以用下面这段代码验证：

from langgraph.graph import StateGraph, START, END print(StateGraph, START, END)

能正常打印出来就说明环境没问题。

2.2 必须理解的 5 个核心概念

多智能体系统开发中最容易混淆的是 LangGraph 的五个概念：State（状态）、Node（节点）、Edge（边）、Conditional Edge（条件边）、Checkpointer（检查点）。

State 是一个全局数据对象，可以是TypedDict、Pydantic 模型或 LangChain 的MessagesState。各个节点都从这个对象读数据、往这个对象写数据。Node 是一个普通函数，接收当前 State，返回一个字典，返回的字典会被合并进 State。Edge 定义节点之间的静态流转关系；Conditional Edge 是根据状态内容动态决定走哪条分支的函数。

Checkpointer 是 LangGraph 区别于普通状态机的关键能力。它会把每一步执行完之后的完整状态快照保存下来，可以用于断点续跑、人工介入、失败重试。我给客户演示时最常用的一个例子是：让系统先走几步，停在人工审批节点，人工修改某个字段后，从停下的位置继续执行，而不是从头重跑。

节点: router -> sql_agent -> final_answer 状态: user_query / analysis / answer / next_step 条件边: router 根据分析结果选择 sql_agent 或 knowledge_agent 检查点: SqliteSaver 保存每一步状态，支持中断恢复

2.3 最小可运行示例：两条边也能跑通流程

我先把一个最小图跑通，让你对整体代码结构有体感。下面这个图有两个节点：一个负责接收消息，一个负责返回结果，中间用条件边做路由。

from typing import TypedDict from langgraph.graph import StateGraph, START, END class DemoState(TypedDict): user_query: str answer: str def entry_node(state: DemoState): # 模拟简单的意图判断 return {"answer": f"收到：{state['user_query']}，进入处理流程"} def final_node(state: DemoState): return {"answer": f"最终结果：已处理 {state['user_query']}"} def route(state: DemoState): # 这里只是演示，实际可以按关键词判断或交给 LLM return "final_node" graph = StateGraph(DemoState) graph.add_node("entry", entry_node) graph.add_node("final", final_node) graph.add_edge(START, "entry") graph.add_conditional_edges("entry", route, {"final_node": "final"}) graph.add_edge("final", END) app = graph.compile() result = app.invoke({"user_query": "查询上季度销售数据"}) print(result)

这段代码虽然简单，但是包含了 LangGraph 的核心骨架：定义状态、增加节点、串联边、编译成可执行图。后面所有复杂系统都是在这个骨架上长出来的。把这个跑通之后再去看多智能体代码，思路就会清晰很多。

3. 从单智能体到多智能体：企业工单系统代码实战

3.1 业务场景与角色划分

用实际业务来拆解：我要做一个企业内部的智能工单系统，员工提一个问题，系统需要判断它是「数据查询类」还是「制度文档类」还是「综合问答类」，然后分给不同的子 Agent 处理。如果是数据查询，SQL Agent 去查数据库；如果是制度文档，知识库 Agent 去检索文档库；如果是敏感操作，还要先经过人工审批。

角色划分如下：

这里的核心原则是“每个 Agent 只做一件事”。SQL Agent 不需要知道文档库有什么，Knowledge Agent 也不需要知道数据库表结构。这样划分之后，训练和调试成本都会大幅降低，而且某个 Agent 挂了不会拖垮整个链路。

3.2 定义全局状态和节点函数

多智能体系统的状态设计非常关键。我的习惯是：全局状态放“用户请求、路由结果、中间数据、最终答案、是否需要人工审批”，不要把对话历史全部塞在里面，否则状态会越来越大，排查问题也很困难。

from typing import TypedDict, Literal from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.sqlite import SqliteSaver class AgentState(TypedDict): user_query: str # 原始用户请求 route: str # 路由结果：sql / knowledge / general sql_result: str # SQL 子 Agent 返回的数据 doc_result: str # 知识库子 Agent 返回的内容 final_answer: str # 最终回复 needs_approval: bool # 是否需要人工审批

节点函数的输入输出都是这个 State 的子集。每个节点只更新自己关心的字段，返回一个字典。这种方法比传一个大对象到处改变量要安全得多，尤其多智能体并发运行的时候，不会有多个 Agent 同时写同一个字段的问题——只要它们各自写不同字段。

3.3 Supervisor 路由：用规则还是用 LLM 决定下一步

路由是整个多智能体系统里最关键的节点。它决定了一个请求会被送去哪个子 Agent。我用过两种方案：纯规则路由和 LLM 路由，各有适用场景。

纯规则路由适合请求模式比较固定的内部系统，例如包含“查询、统计、数据”这些关键词就走 SQL Agent，包含“制度、流程、政策”就走 Knowledge Agent。规则路由的好处是稳定、零额外成本、好测试，缺点是泛化能力弱，用户换个说法就分错了。

LLM 路由是用一个轻量模型判断用户意图，输出一个结构化路由结果。我通常让模型输出 JSON，然后用一个函数解析 JSON 里的字段，决定走哪条条件边。下面是我的实现方式：

from langchain_openai import ChatOpenAI router_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) def supervisor_node(state: AgentState): prompt = f"""判断用户的请求类型，只输出 JSON： {{"route": "sql"}} 或 {{"route": "knowledge"}} 或 {{"route": "general"}} 用户请求：{state['user_query']}""" resp = router_llm.invoke(prompt) route = parse_route(resp.content) return {"route": route} def route_after_supervisor(state: AgentState) -> Literal["sql_agent", "knowledge_agent", "final_node"]: return { "sql": "sql_agent", "knowledge": "knowledge_agent", "general": "final_node", }.get(state["route"], "final_node")

这里有个经验：路由模型不要用和主模型一样大的参数，因为路由任务非常简单，用大模型纯属浪费。我在生产环境用的是gpt-4o-mini这类轻量模型，一次路由调用不到 1 秒，成本也低很多。

3.4 子 Agent 实现与工具注册

子 Agent 我用 LangChain 的create_react_agent配合自定义工具来写，这样既不用重复造轮子，又能利用 LangGraph 的图结构控制生命周期。下面以 SQL Agent 为例：

from langchain_core.tools import tool from langchain.agents import create_react_agent @tool def query_sales_data(date: str) -> str: """查询指定日期的销售数据，返回原始记录。""" # 这里是模拟，实际环境下替换为数据库连接 return f"当日销售额 120 万，订单 350 单" tools = [query_sales_data] sql_agent_prompt = PromptTemplate.from_template( "你是企业数据查询助手，只能使用查询工具回答，不要编造数据。" "用户问题：{input}" ) sql_agent_runnable = create_react_agent(router_llm, tools, sql_agent_prompt) def sql_agent_node(state: AgentState): result = sql_agent_runnable.invoke({"input": state["user_query"]}) return {"sql_result": result["output"]}

create_react_agent内部会自动处理工具调用的循环，相当于把 LangGraph 的节点内部做成了一个单 Agent 子图。子 Agent 之间不会互相看见对方，状态完全隔离，这对多智能体的故障隔离很有帮助。

Knowledge Agent 的结构类似，只是工具换成向量库检索函数。重要的点是每个子 Agent 的错误处理：如果工具抛异常，不要把原始异常直接返回给用户，而是返回一个可读性强的错误信息，同时记录日志。错误信息也可以写进 State，让上层 Supervisor 决定是否需要换一个 Agent 处理。

3.5 人工审批：用 interrupt 实现 Human-in-the-loop

多智能体系统在企业落地时，最难却最必要的就是“人机协同”。不是所有操作都让 Agent 自动执行，比如批量删除数据、修改权限、发送外部通知，这些操作必须有审批环节。

LangGraph 的 Checkpointer 天然支持 Human-in-the-loop。我的做法是在敏感操作节点之前插入一个中断点，图中断后返回给调用方，等待人工确认后继续：

graph = StateGraph(AgentState) # ... add nodes graph.add_edge("sql_agent", "approval_node") graph.add_edge("approval_node", "final_node") compiled = graph.compile( checkpointer=SqliteSaver.from_conn_string("checkpoints.db"), interrupt_before=["approval_node"] # 执行到 approval_node 前暂停 ) # 第一次执行，会暂停在审批环节 config = {"configurable": {"thread_id": "ticket-1001"}} app.invoke({"user_query": "删除测试数据"}, config) # 人工审批通过后，继续执行 app.invoke(Command(resume={"approved": True}), config)

这个机制的生产价值很高：既保留 Agent 的自动化效率，又把关键决策权牢牢握在人类手里。刚开始做多智能体系统的同学往往忽略这一点，我看过太多全自动 Agent 系统上线后因为不可控被叫停，加上人工审批环节之后反而顺利推开。

4. 部署指南：从开发机到生产环境

4.1 用 FastAPI 封装成 HTTP 服务

LangGraph 图本身是一个 Python 对象，生产环境里我通常用 FastAPI 包一层 HTTP 接口。这样前端、企业微信机器人、调度平台都能统一接入，同时我们可以把鉴权、限流、日志这些横切逻辑放在这一层集中处理。

from fastapi import FastAPI, HTTPException, Header from pydantic import BaseModel from typing import Optional from agent_graph import build_graph app = FastAPI(title="Multi-Agent API") graph = build_graph() class ChatRequest(BaseModel): message: str thread_id: str user_id: str = "anonymous" def verify_token(authorization: str): if authorization != "Bearer prod-token-2024": raise HTTPException(status_code=401, detail="invalid token") @app.post("/chat") async def chat(req: ChatRequest, authorization: str = Header(default="")): verify_token(authorization) config = {"configurable": {"thread_id": req.thread_id}} result = await graph.ainvoke({"user_query": req.message}, config=config) return {"answer": result["final_answer"], "thread_id": req.thread_id}

这里特别强调thread_id的设计。多智能体系统每个对话会话必须有一个唯一会话 ID，LangGraph 靠它分隔不同会话的检查点状态。如果多个用户共用一个thread_id，状态会串，这是我在联调时踩过的最严重的坑之一。

4.2 Docker 镜像与 docker-compose 编排

生产部署我用两部分：Dockerfile 构建应用镜像，docker-compose 编排应用和依赖组件。多智能体系统的状态持久化不能依赖内存，我建议用 SQLite 先跑单机，等真的有水平扩展需求了再切 Postgres。

FROM python:3.11-slim WORKDIR /app ENV PYTHONUNBUFFERED=1 COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . EXPOSE 8000 CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

配套的docker-compose.yml可以这样写：

services: api: build: . ports: - "8000:8000" environment: - OPENAI_API_KEY=${OPENAI_API_KEY} - LANGCHAIN_TRACING_V2=true - LANGCHAIN_API_KEY=${LANGCHAIN_API_KEY} volumes: - ./checkpoints.db:/app/checkpoints.db restart: unless-stopped

这里有个细节：SQLite 文件一定要挂载到宿主机卷。如果不挂载，容器一重启所有会话状态就全丢了，用户会发现自己聊到一半，系统把前面的上下文忘得干干净净。

4.3 可观测性：没有日志就别谈企业级

多智能体系统比单 Agent 系统难调的地方在于，问题可能出现在任意一个节点。所以从第一天就要把链路日志设计好。我每个节点都记录四个字段：节点名、耗时、输入摘要、输出摘要、异常信息。

LangSmith 是 LangChain 官方的可观测平台，我直接在环境变量里开启：

export LANGCHAIN_TRACING_V2=true export LANGCHAIN_API_KEY=ls_xxx

开启之后，LangGraph 的每一次图执行都会被记录下来，包括每个节点调用了什么模型、消耗了多少 token、耗时多久、在哪一步失败。排查“为什么这个请求走了错误分支”的时候，直接在 trace 面板里看每个节点的输入输出，比打印日志效率高十倍。

如果企业的数据合规要求不能把 trace 传到第三方平台，那就用 OpenTelemetry 接入自建监控，至少保证节点级日志和耗时指标能采集上来。没有可观测性的多智能体系统，上线后就是你拿着放大镜在几万行日志里捞针，那种体验我不想再有第二次。

4.4 并发、限流与生产安全

多智能体系统上线后的第一波事故通常来自并发问题。LangGraph 的 Checkpointer 是有锁的，同一个thread_id如果有两个请求同时进来，后写会覆盖先写，状态就乱了。我在接口层的解法是给同一个thread_id的请求加互斥锁，同时用 Redis 做分布式锁，防止多副本场景下的并发写。

限流这块，我建议分两层。第一层用 Nginx 或者 API 网关做 IP 级限流，第二层在 FastAPI 里基于用户维度做限流，核心原因是大模型调用很贵，不加限制会被内部用户一个循环调用打到爆。特别是多智能体系统，一次请求可能要调用几十次模型，成本是单 Agent 的好几倍。

安全上最基本的两个点：API 鉴权必须统一走 Header 里的 Token，不要把密钥放进 URL 参数；模型输出的内容要做敏感信息过滤再加一层白名单校验，尤其 SQL Agent 很容易被诱导生成“删除表”这类危险 SQL，我在工具函数里做了严格的关键词拦截。

4.5 边缘部署扩展：Jetson AGX Orin 上跑本地 LLM

多智能体系统还有一个被很多人忽略的落地场景：模型推理放在边缘设备。有些企业的数据策略非常严格，业务数据不允许出内网，这时候可以把推理端落到本地设备上。Jetson AGX Orin 是我在对比之后选出来的设备，算力足够，能跑 7B 到 13B 的量化模型，而且功耗比 x86 + GPU 方案低很多。

部署方式是先用 llama.cpp 把量化模型跑成一个 OpenAI 兼容的 HTTP 服务，LangGraph 侧不需要改任何逻辑，只要把模型的base_url指向边缘设备的地址：

from langchain_openai import ChatOpenAI local_llm = ChatOpenAI( model="qwen2.5-7b-instruct-q4_k_m", base_url="http://192.168.1.100:8080/v1", api_key="not-needed", temperature=0.2, )

这样 LangGraph 的 Supervisor 和子 Agent 全部可以复用，只是底层模型从云端换成本地，数据不出内网。我在实际测试中，7B Q4 量化模型跑意图路由和知识库总结完全够用，但跑复杂 SQL 生成偶尔会出错，所以路由模型放边缘，重逻辑 Agent 仍走云端，是一个比较务实的混合方案。

5. 常见问题与排查技巧实录

5.1 状态被并发请求写坏

这是多智能体系统上线初期最高频的问题。现象是用户 A 的问题，回答里出现了用户 B 的业务数据。排查思路是先确认接口层是否把thread_id正确隔离，再看 Checkpointer 用的是内存还是持久化存储。如果是多个副本部署，强烈建议把检查点切到 PostgreSQL，用langgraph-checkpoint-postgres配合事务锁。

5.2 上下文越滚越大，token 成本失控

多智能体对话跑了几轮之后，State 里的消息列表会越来越大，每轮都要把全部历史送给模型，token 消耗直线上升。我的做法是在进入子 Agent 之前做消息裁剪：只保留最近 10 轮对话，更早的上下文交给一个“摘要节点”压缩成一段话放进 State。这个做法能省 60% 以上的 token，对回答质量影响很小。

5.3 子 Agent 调用超时

LLM 接口偶尔会超时，尤其是网络不稳定或者模型负载高的时候。我在两个层面处理：给模型调用设置timeout参数，时间到了直接抛异常；工具函数里做重试，最多重试 2 次，每次退避 1 秒。如果重试仍然失败，就在节点里返回结构化错误信息，交给上层决定是否走降级链路。

我整理了一张速查表，覆盖我碰到的排障场景：

5.4 模型从云端切到本地后的兼容性问题

从 OpenAI 切到 llama.cpp 或 Ollama 本地模型，最常见的问题是 tool calling 支持程度不同。OpenAI 的工具调用有严格 schema，本地模型格式不一定完全兼容。我发现比较稳妥的方式是：路由和总结这类任务用普通文本输出，再写一个 JSON 解析器提取结果，不要过度依赖本地模型的 tool calling。SQL Agent 这类必须用工具调用的场景，就继续走云端模型。

还有一个小细节：不同模型的temperature默认值不一样，有些本地模型默认就是 0.8，导致路由结果忽左忽右。按我的经验，路由节点一律设成 0，主回答节点可以稍微高一点，但也不要超过 0.7，否则企业场景里会频繁出现“很有创意但完全不可用”的回答。

说到最后，分享一个我踩过最深的一次坑。刚开始搭多智能体系统时，我把所有子 Agent 的 Prompt 写得特别详细，恨不得把用户画像和业务背景全塞进去，结果路由经常错乱，子 Agent 之间还互相干扰。后来我把每个 Agent 的 Prompt 砍到只剩“你是谁、你能做什么、你绝对不能做什么”三句话，准确率反而明显上来。多智能体系统不是模型越强或者 Prompt 越长越好，而是边界越清晰越好，每一个 Agent 只需要对自己那一小块领域负责并做到极致，剩下的交给 Supervisor 去协调。现在我在尝试的方向是把每个子 Agent 独立部署成微服务，中间通过消息队列通信，让整条链路可以独立扩容和降级。这个方向踩坑肯定不少，等跑出稳定效果后再写一篇详细的拆解出来。