企业官网建设流程全解析

把 LangGraph 的 Agent 从本地脚本挪到生产环境，这一步我见过太多人卡住了。明明python agent.py跑得好好的，一接真实请求就各种幺蛾子：状态丢了、并发串了、工具调用超时了、用户等得想骂人了。LangGraph 的部署路径这件事，看起来是个运维话题，实际上决定的是整个 Agent 项目的架构走向。这篇就基于我用 FastAPI + LangChain + LangGraph 做 AI Agent 的实战经历，聊三条从脚本到服务的部署路径：直接脚本跑、自建 HTTP 服务、上托管平台。每条路我都给出可复现的代码和配置，再说说它们各自的边界和适用场景。

1. 为什么聊部署：脚本里能跑的 Agent，离生产还差多远

1.1 “能跑”和“能用”之间隔着什么

很多人对 LangGraph 的第一印象停留在“写一个状态图，然后.invoke()一把梭”。确实，你可以在 Notebook 里定义一个StateGraph，塞进去几个节点和条件边，把工具调用跑通，最后看着它一步步回答用户问题，感觉整个 Agent 已经完成了。

但生产环境不是这么回事。脚本模式是“进程内调用一次”，所有状态都活在内存里，进程退出就什么都没有。服务模式则要面对并发请求、失败重试、长连接超时、权限隔离、日志追踪、流量波动这些事。LangGraph 本身提供了一套很好的状态管理抽象，比如Checkpointer、thread_id、recursion_limit，但这些能力只有在正确的部署形态下才能真正发挥出来。

我见过一个项目，团队花了两周把 Agent 流程调得漂漂亮亮，然后直接用一个 Flask 接口包起来上线，结果第二天就出问题：用户 A 的对话上下文跑到了用户 B 的会话里。原因很简单，Agent 的状态存在全局变量里，根本没有按会话隔离。这不是 LangGraph 的锅，是部署时没把状态持久化设计进去。

所以聊部署路径，本质上是聊你怎么管理 Agent 的状态、并发、生命周期和可观测性。

1.2 三条部署路径的定位

这三条路径不是互相替代的关系，而是解决不同阶段的问题：

• 脚本化运行：面向开发调试和离线批处理，成本最低，适合验证流程。
• FastAPI 自建服务：面向私有化部署和定制接口，自己掌控一切，灵活性最高。
• 托管平台：面向生产级多 Agent 场景，平台帮你解决存储、监控、人审等问题。

我先把三条路的关键指标列个表，方便你心里有个底：

下面一条一条拆开讲。

2. 路径一：脚本化运行，搞定开发和批处理

2.1 最小可用的 LangGraph 脚本长什么样

不管最后走哪条部署路，脚本跑通是第一步。给你看一个最小但完整的例子，这个结构我用了很久，所有复杂 Agent 都是从这里长出来的：

from typing import TypedDict, Annotated from langgraph.graph import StateGraph, START, END from langgraph.prebuilt import ToolNode, tools_condition from langchain_openai import ChatOpenAI from langchain_core.tools import tool class AgentState(TypedDict): messages: Annotated[list, "append"] next_step: str @tool def get_weather(city: str) -> str: """查询指定城市的实时天气""" # 实际项目里替换成天气 API 调用 return f"{city} 晴，气温 26 摄氏度" tools = [get_weather] llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) llm_with_tools = llm.bind_tools(tools) def agent_node(state: AgentState): return {"messages": [llm_with_tools.invoke(state["messages"])]} graph = StateGraph(AgentState) graph.add_node("agent", agent_node) graph.add_node("tools", ToolNode(tools)) graph.add_edge(START, "agent") graph.add_conditional_edges("agent", tools_condition, {"tools": "tools", END: END}) graph.add_edge("tools", "agent") app = graph.compile() result = app.invoke({"messages": [("human", "杭州天气怎么样？")]}) print(result["messages"][-1].content)

这个脚本的关键在于两条边：tools_condition判断模型输出里有没有 tool_call，有就进工具节点，没有就直接走向 END。这就是 LangGraph 里最经典的 ReAct 循环。Annotated[list, "append"]保证每轮messages都会追加到已有状态里，而不是覆盖掉。

跑这个脚本的时候，哪个节点被调用了、模型输出了什么、工具返回了什么，全都直接打到终端里。这个体验对排查逻辑问题非常友好，比任何 Service 模式都直观。

2.2 把脚本当“调度任务”用

别急着说脚本模式没用，它在批处理场景里非常好使。比如说你要做一个每天早上 9 点自动汇总竞品信息的 Agent，脚本再合适不过了。

做法也简单，把 LangGraph 的app.invoke()包在一个函数里，然后用 cron 或 APScheduler 定时触发。这里有个经验：脚本的输入不要写死，从命令行参数和环境变量读。我一般会这样做：

import argparse import os import json parser = argparse.ArgumentParser() parser.add_argument("--task", required=True) args = parser.parse_args() # 从环境变量读取 API Key，不要在代码里硬编码 os.environ.get("OPENAI_API_KEY") result = app.invoke({ "messages": [("human", f"执行任务：{args.task}")], "next_step": "plan" }) with open(f"output_{args.task}.json", "w") as f: json.dump(result, f, ensure_ascii=False, indent=2)

这样做的价值在于，Agent 跑批处理任务的时候可以复用同一份 graph 定义，换任务只换参数。我还习惯把每次调用的输入输出落盘，后面调整 prompt 或工具时，可以拿历史数据做回归对比，比人脑记忆靠谱多了。

2.3 脚本模式的三条边界

但脚本模式有三个绕不开的硬伤，决定它不能直接当服务用：

第一，状态跨请求无法保留。脚本每次运行都是新进程，MemorySaver里的记录全没了。你没法让用户说一句“刚才那个问题再解释详细一点”，Agent 不知道“刚才”是什么。

第二，没有并发隔离。一旦同时进来两个请求，脚本只会按顺序处理一个，另一个排队排到天荒地老。就算你用多线程硬顶，共享状态会互相污染，出问题的时候极难排查。

第三，进程退出即丢。没有崩溃恢复、没有持久化，机器重启一下，所有运行中的 Agent 任务直接蒸发。

所以我的结论是：脚本模式是调试器和批处理工具，不是服务。你要让 AI Agent 真正“下地干活”，至少得走到第二条路。

3. 路径二：用 FastAPI 把 Agent 包成 HTTP 服务

3.1 为什么挑 FastAPI 而不是 Flask

把 Agent 暴露成 HTTP 接口，框架选择上我强烈建议 FastAPI。不是说 Flask 不行，而是 FastAPI 的异步支持、类型校验和自动文档这三个特性，和 LangGraph 的异步 API 简直绝配。

LangGraph 提供了ainvoke、astream_events这一整套异步方法，你拿 Flask 的同步模型去对接，Thread 调度会浪费掉大量 IO 等待时间。FastAPI 的async def直接把 event loop 打通了，Agent 在等 LLM 返回的时候，同一个进程还能处理其他请求。

而且 FastAPI 自带 OpenAPI 文档，接口调试不用另外装 Postman 之外的工具，浏览器打开/docs就能直接试。这对联调阶段的帮助特别大。

看一个最小实现：

from fastapi import FastAPI from pydantic import BaseModel from langgraph.checkpoint.memory import MemorySaver from langgraph.graph import StateGraph # 复用第 2 节里的 graph 定义 app = FastAPI() # 编译时挂上 checkpointer checkpointer = MemorySaver() graph_app = graph.compile(checkpointer=checkpointer) class ChatRequest(BaseModel): session_id: str message: str class ChatResponse(BaseModel): reply: str @app.post("/chat", response_model=ChatResponse) async def chat(req: ChatRequest): config = {"configurable": {"thread_id": req.session_id}} result = await graph_app.ainvoke( {"messages": [("human", req.message)]}, config ) return ChatResponse(reply=result["messages"][-1].content)

和脚本模式最大的区别就在checkpointer和thread_id。thread_id是会话的身份证，同一个 session 的请求会共享历史状态，不同 session 天然隔离。用户 A 的消息永远走 A 的线程，B 的线程不会串。

3.2 状态管理：从 MemorySaver 到 SqliteSaver

上面代码里用的是MemorySaver，这玩意在服务模式下只适合开发和压测，因为它把状态存在内存里，进程一重启全没了。生产环境至少要换成SqliteSaver。

from langgraph.checkpoint.sqlite import SqliteSaver # 注意 from_conn_string 返回的是一个上下文管理器 with SqliteSaver.from_conn_string("checkpoints.db") as checkpointer: graph_app = graph.compile(checkpointer=checkpointer)

用 SQLite 的好处是单文件、零运维，小规模部署完全够用。但如果你有多个 uvicorn worker，SQLite 的并发写会有锁竞争问题。这种情况我建议直接用 PostgreSQL，LangGraph 官方提供了langgraph-checkpoint-postgres包，用法差不多：

from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver async with AsyncPostgresSaver.from_conn_string( "postgresql://user:pass@localhost:5432/agent" ) as checkpointer: graph_app = graph.compile(checkpointer=checkpointer)

这里有个实际教训：graph.compile()的时机很重要。很多人在模块 import 的时候就把 graph 编译了，checkpointer 也随之初始化。如果这时数据库还没准备好，或者之后要切换数据库配置，你就得重启整个服务。我的做法是把编译逻辑放到 FastAPI 的 lifespan 钩子里，启动时再初始化，配置改起来方便得多。

3.3 核心接口与工具调用的透传

把 Agent 包成服务之后，接口设计就直接影响使用方的体验。我总结一个最简单的接口契约：session_id负责状态隔离，message负责用户输入，剩下的都交给 Agent 自己判断。不要在设计接口的时候把“调用哪个工具”暴露给调用方，否则你会陷入无休止的参数适配里。

但工具调用本身确实需要透传一些信息，比如用户在前端上传了一个文件，或者给了经纬度坐标。我的做法是给消息内容做结构化包装，而不是往messages列表里塞一个纯字符串：

class ChatRequest(BaseModel): session_id: str message: str metadata: dict = {} @app.post("/chat") async def chat(req: ChatRequest): user_message = req.message if req.metadata: user_message = f"{req.message}\n附加信息：{json.dumps(req.metadata, ensure_ascii=False)}"

说的直白一点，接口层只负责“接住”输入和“转交”给 Agent，具体工具调用的路由逻辑，LangGraph 的ToolNode和条件边已经处理好了。

3.4 流式响应和长任务

Agent 类接口最容易被吐槽的点就是“慢”。这不是接口实现的问题，是 LLM 首 token 延迟加上工具调用往返时间天然就高。如果你用普通的await graph_app.ainvoke()，用户会看到请求转圈十几秒，体验非常差。

解决办法是流式输出。FastAPI 配合 SSE（Server-Sent Events）可以做到用户侧像打字机一样逐字看到输出：

from fastapi.responses import StreamingResponse import json @app.post("/chat/stream") async def chat_stream(req: ChatRequest): config = {"configurable": {"thread_id": req.session_id}} async def event_generator(): async for event in graph_app.astream_events( {"messages": [("human", req.message)]}, config=config, version="v2" ): if event["event"] == "on_chat_model_stream": chunk = event["data"]["chunk"].content if chunk: yield f"data: {chunk}\n\n" elif event["event"] == "on_tool_start": yield f"data: {json.dumps({'tool': event['name'], 'status': 'start'})}\n\n" return StreamingResponse(event_generator(), media_type="text/event-stream")

这样做还有一个额外收益：你可以把工具调用的中间过程也推给前端，像“正在查询天气接口…”这类状态提示，用户就知道 Agent 在干活，而不是卡死了。

如果你不想搞流式，也不想让调用方长时间占着 HTTP 连接，那就把请求丢进消息队列，比如 Redis Stream 或 Celery，然后提供“任务提交”和“任务查询”两个接口。不过这种模式的实时性差一些，适合后台异步任务。

3.5 部署细节：uvicorn、超时、健康检查

FastAPI 服务本身部署起来不复杂，但有几个参数容易踩坑。

uvicorn 启动时，--workers大于 1 的时候，注意每个 worker 进程会各自初始化一份 checkpointer。如果用的是 SQLite 文件，多进程并发写会报database is locked。我实际部署时，小项目单 worker 加 SQLite 就够了，一旦需要多 worker，直接换 Postgres 存储，别在 SQLite 上死撑。

uvicorn main:app --host 0.0.0.0 --port 8000 --workers 2 --timeout-keep-alive 60

--timeout-keep-alive这个参数非常容易被忽视。默认值 5 秒会导致长连接在 Agent 处理过程中被断开。我一开始没调这个参数，前端经常报网络错误，排查了很久才发现是 keep-alive 超时。

再一个，一定要加健康检查接口。别小看这个，K8s 或 Docker Compose 的探针都依赖它：

@app.get("/healthz") async def healthz(): return {"status": "ok"}

4. 路径三：托管平台与生产级 Agent 服务

4.1 托管平台解决了哪些事

第三条路是直接把 Agent 部署到 LangGraph 的官方托管平台（LangGraph Platform 那一套）。注意，我不是推荐大家立刻迁移，而是告诉你什么时候值得考虑。

你自建 FastAPI 服务，状态、监控、权限、多 Agent 编排这些事全都得自己扛。托管平台把这些全部内置了：状态持久化不用自己配数据库，创作者可以通过平台管理多个 agent，每个请求都自带追踪日志，还提供人工审核节点给关键操作加一道闸。

我对托管平台最看重的其实是“人工介入”这件事。LangGraph 本身支持interrupt机制，在执行到某个节点前暂停，等人确认后再继续。这个能力在自建服务里要实现，需要你自己设计挂起状态、通知渠道、恢复接口，工作量不小。托管平台直接把这个做成平台级能力，业务方只需要在 graph 里插入一个 interrupt 节点。

4.2 接入一个托管平台要改什么

很多人以为上托管平台要重写代码，其实不用。Graph 定义还是本地那份代码，平台只是负责把代码跑起来，只是在接入方式上有一些约定。

第一步，在项目根目录放一个langgraph.json配置文件：

{ "dependencies": ["requirements.txt"], "graphs": { "agent": "./src/agent/graph.py:build_graph" }, "env": ".env" }

第二步，确保graph.py里有一个build_graph函数，返回编译后的 graph 或者 builder 对象。注意，不要在这里直接初始化数据库连接，平台会注入自己的存储方案。

def build_graph(): graph = StateGraph(AgentState) graph.add_node("agent", agent_node) graph.add_node("tools", ToolNode(tools)) graph.add_edge(START, "agent") graph.add_conditional_edges("agent", tools_condition, {"tools": "tools", END: END}) graph.add_edge("tools", "agent") return graph.compile()

接入平台之后，平台会为每个部署的环境生成一个 API 地址。你的调用方还是通过 HTTP 请求访问，只不过负载均衡、版本回滚、日志检索这些都有人管了。

4.3 选择托管平台的判断标准

我见过不少团队一上来就上托管平台，结果发现运维是省了，但定制化需求一大堆，平台反而成了限制。所以我把判断标准提炼成四条，满足两条以上再考虑：

• 你需要同时维护很多个 Agent，每个 Agent 的 prompt 和工具都不一样，自己写管理后台太累。
• 你的 Agent 流程里有需要人工确认的环节，比如交易确认、内容发布前审核。
• 你没有专职运维，也不想每天盯着日志看有没有报错。
• 你的业务需要外部客户直接访问 Agent API，而不是只在内网用。

反过来，如果只是公司内部一个辅助工具，日请求量几百，那托管平台属于过度设计，自建服务更轻量。

5. 三条路怎么选：我的选型清单

5.1 四个问题解决选择困难

我在带项目的时候，最常被问的就是“到底选哪条路”。我不会直接给答案，而是抛四个问题：

第一个问题：用户是谁？如果是内部工具，自建一个 FastAPI 服务绰绰有余；如果是外部客户，要考虑 API 稳定性、鉴权、限流，托管平台可能更省心。

第二个问题：QPS 多少？单机脚本顶多撑个位数并发；FastAPI 单实例能撑几十到一两百左右，再往上要么堆 worker，要么上平台自动扩容。

第三个问题：状态要不要长期保存？用户关了浏览器，下次回来还要能继续对话，那就必须上数据库持久化。MemorySaver 没法满足。

第四个问题：团队有没有人盯运维？没人盯就选托管平台，有人盯就自建。

这四个问题组合起来，答案基本就清晰了。

5.2 组合使用效果更好

三条路径不是非此即彼，我实际项目中经常组合着来。新功能开发阶段，先在脚本里把 graph 跑通，把工具调用、状态流转这些逻辑调对。然后封装成 FastAPI 服务，部署到测试环境让业务方体验。等某个 Agent 的流程相对稳定、需要正式上线了，把纯逻辑部分搬到托管平台跑，FastAPI 层保留内部管理接口和批处理任务。

有个项目我印象很深：一个是内部数据汇总 Agent，要求每天定时跑，这种始终留在脚本和 cron 里，稳定又省钱；另一个是客户咨询 Agent，需要保存历史会话、支持人工介入，后来就迁到了托管平台。自建的 FastAPI 服务作为导流层，把所有 Agent 的入口统一管理。这种分层设计，到现在运行得都很稳。

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

6.1 Agent 陷入工具调用死循环

症状是日志里 agent 节点和 tools 节点来回交替，输出永远不走向 END。排查方法很直接，先确认tools_condition的条件映射是否正确，再看模型是不是反复生成同一个 tool_call。

我踩过的坑是：工具返回的内容没有让模型“满意”，模型就一直尝试调用工具，直到逼近recursion_limit报错退出。解决方式是给工具加上详细的 docstring，返回结果尽量结构化，比如直接返回{"status": "ok", "data": ...}，模型拿到之后更倾向于整理答案而不是再次调用。

还可以在编译时显式设置：

app = graph.compile(checkpointer=checkpointer, interrupt_before=["tools"])

或用默认的recursion_limit兜底，但不要一刀切调太高，不然死循环的请求会拖着资源不放。

6.2 并发一上来状态就串线

这个我前面提过，最普遍的原因就是没有按thread_id隔离状态。另一个隐蔽原因是你把checkpointer写成了模块级单例，但不同请求复用了同一个 config 对象。

我的排查习惯是给每个请求打一个唯一的session_id，并且在日志里带上它：

logger.info("session=%s user_input=%s", req.session_id, req.message)

一旦出现串线，通过 session_id 能立刻定位是哪两个请求互相污染了。

6.3 工具调用结果解析失败

症状是模型生成了 tool_call，但ToolNode执行时报参数缺失或格式错误。大多数情况是因为工具的参数 schema 和实际实现不一致。

我一般会做两层防护：第一层，工具函数的 docstring写清楚每个参数含义，模型是根据这个来生成参数的；第二层，工具内部做容错，遇到异常不直接抛死，而是把错误信息返回给模型：

@tool def query_order(order_id: str) -> str: """根据订单号查询订单状态，参数 order_id 是字符串类型的订单编号。""" if not order_id.isdigit(): return "订单号格式错误，请确认后重试" return "订单已发货"

这样模型看到错误信息后，会自己修正参数，而不是整个流程崩溃。

6.4 部署后响应慢，甚至 504

我遇到过几次这种情况，一开始以为是 LangGraph 的问题，后来发现是网关层超时设置太短。Agent 一个完整流程要经过 LLM、工具调用、再让 LLM 总结，耗时十几秒很正常。

排查顺序是：先量 LLM 调用耗时，再量工具调用耗时，最后看 HTTP 层有没有提前断开。解决方式就三选一：做流式输出，把等待感降下来；调大网关超时；把任务改造成异步队列。除此之外，把不必要的工具调用去掉，或者用更快的模型版本，也能明显缩短链路耗时。

6.5 进程重启后历史会话丢失

症状是用户第二天回来发现自己和 Agent 的对话记录没了。这不用犹豫，就是MemorySaver导致的。只要换了 Postgres 或 SQLite 的 checkpoint，数据才会真正落盘。

这个知识点最简单，但也是生产事故最多发的点。我的建议是：一旦决定对外提供服务，第一时间把 MemorySaver 换掉，哪怕先用 SQLite 文件顶着，比内存强一百倍。

最后说点个人体会。部署 LangGraph 这件事，我踩过最大的坑不是技术不会，而是以为把 graph 编译出来就算部署完了。实际上从脚本到服务，核心工作全在状态管理、并发隔离和接口设计上。如果你现在正卡在这一步，别急着纠结选哪条路，先拿一个最小 agent 走通“脚本 → FastAPI → checkpointer”这条链路，跑通了再往上想托管平台的事。技术选型没法一步到位，但“先让它稳定跑起来”这个目标，任何时候都不会错。