1. 从脚本到服务,为什么部署这一步最容易卡住人
刚接触 LangGraph 的人,十有八九会经历同一个心理曲线:本地写个脚本,跑通一个带工具调用的 Agent,兴奋得不行,觉得自己已经摸到了智能体开发的门槛。然后下一步想把这东西变成一个别人也能用的服务,瞬间就懵了。脚本里graph.invoke()一行就出结果,可一旦要放到服务端,问题全冒出来了——状态存哪、多轮对话怎么续、并发请求会不会串、工具调用超时怎么办、日志怎么打、进程崩了重启后上下文还在不在。
我自己最开始就是把一个 LangGraph 的客服 Agent 直接塞进 Flask 的一个路由里,结果上线第一天就被并发打穿了。两个用户同时对话,状态互相污染,A 的订单信息出现在了 B 的回复里。那次事故让我彻底明白一件事:LangGraph 的部署,核心不是“怎么把图跑起来”,而是“怎么把图的状态管起来”。这也是为什么标题里专门点了 Checkpointer 这个词——它才是从脚本到服务这条路上真正的分水岭。
这篇内容我打算把 LangGraph 的部署路径拆成三条来讲,分别对应三种不同的落地阶段和团队规模。不是那种官方文档式的罗列,而是我自己踩过坑之后总结出来的选型逻辑:什么阶段该用哪条路,每条路的坑在哪,关键参数怎么定,出了问题怎么排查。如果你正在把 LangGraph 从 demo 往生产推,或者纠结要不要上 FastAPI、要不要接 LangSmith,这篇应该能帮你少走不少弯路。
三条路径大致是这样划分的:第一条是脚本直跑 + 本地持久化,适合验证和单机工具;第二条是FastAPI 服务化 + Checkpointer 状态管理,这是绝大多数团队真正落地时走的路;第三条是接入 LangSmith 做可观测 + 分布式状态后端,面向多实例、高并发的生产场景。下面逐条拆。
2. 三条部署路径的整体设计与选型逻辑
在动手之前,先把三条路径的定位讲清楚,不然很容易一上来就过度设计。我见过太多人,明明只是自己用的小工具,非要上分布式状态后端,结果维护成本比业务本身还高。选型的核心就一句话:你的状态需要活多久,你的服务需要扛多少并发。
2.1 路径一:脚本直跑,状态活在内存里
这是最原始的形态。你写一个main.py,里面构建 graph,然后graph.invoke(inputs),跑完打印结果,进程结束,状态消失。适合的场景很明确:一次性任务、批处理、本地调试、CI 里的自动化测试。
它的优势是零部署成本,python main.py就完事。但劣势也致命:没有跨请求的状态。也就是说,用户第一轮说了“我要查订单”,第二轮说“就是刚才那个”,你的脚本根本不知道“刚才那个”是什么。因为每次 invoke 都是全新的内存状态,进程一退,全没了。
所以路径一本质上不是“部署”,而是“运行”。它解决的是“让图跑起来”,不解决“让服务活下去”。我把它列进来,是因为很多人会卡在这一步很久,误以为这就是终点。实际上它只是起点,是你验证 graph 逻辑正确性的地方。
2.2 路径二:FastAPI 服务化,Checkpointer 接管状态
这是真正的分水岭。当你需要多轮对话、需要多个用户同时用、需要服务常驻,就必须把 graph 包进一个 Web 服务里。FastAPI 几乎是当前 Python 生态里做这件事的最优解,原因后面细讲。
这条路径的关键变化是:状态从内存搬到了 Checkpointer。LangGraph 的 Checkpointer 机制允许你在每一步之后把图的完整状态持久化下来,用thread_id作为索引。下一个请求带着同样的thread_id进来,Checkpointer 就能把之前的状态恢复出来,图从断点继续跑。
这就解决了路径一的所有问题:状态跨请求、跨进程、跨重启。用户 A 和用户 B 用不同的thread_id,天然隔离,不会串。服务重启了,只要 Checkpointer 的后端还在,对话上下文就还在。
2.3 路径三:LangSmith 可观测 + 分布式状态后端
当你的服务从单实例变成多实例,从几十个用户变成几千个用户,路径二会开始暴露新的瓶颈。单机的 SQLite Checkpointer 扛不住并发写,多个实例之间状态不共享,出了问题你根本不知道是哪一步、哪个工具调用慢。
这时候就得上路径三:状态后端换成 Postgres 或 Redis 这类支持并发的存储,同时接入 LangSmith 做全链路追踪。LangSmith 能让你看到每一次 graph 执行的完整轨迹——哪个节点耗时多久、工具调用的输入输出是什么、哪一步报错了。没有它,线上排查基本靠猜。
三条路径不是互斥的,而是递进的。你可以先在路径一验证逻辑,再迁到路径二上线,最后按需升级到路径三。下面我把每条路径的关键细节和实操要点展开讲。
3. 路径一实操:脚本直跑与状态验证的正确姿势
别看路径一简单,里面也有讲究。很多人脚本写得随意,导致后面迁移到服务化时几乎要重写。我的建议是:从第一天起就按“未来要服务化”的方式组织代码,这样迁移成本最低。
3.1 项目目录结构,一开始就别乱
我见过太多人把所有东西塞进一个main.py,几百行,graph 定义、工具函数、prompt、测试代码全混在一起。等到要拆成服务时,改到怀疑人生。哪怕现在只是脚本,也建议按下面的结构组织:
project/ ├── app/ │ ├── __init__.py │ ├── graph.py # graph 构建逻辑 │ ├── nodes.py # 各个节点函数 │ ├── tools.py # 工具定义 │ ├── state.py # State 类型定义 │ └── config.py # 配置读取 ├── scripts/ │ └── run_once.py # 一次性运行入口 ├── tests/ │ └── test_graph.py └── requirements.txt这个结构的好处是,graph.py里的构建函数可以被脚本调用,也可以被后面的 FastAPI 调用,一份逻辑两处复用。state.py单独抽出来定义TypedDict或 Pydantic 模型,是因为状态结构是 graph 的核心契约,改一处要动全身,必须集中管理。
3.2 State 定义,决定了后面状态能不能存
LangGraph 的 State 通常用TypedDict定义,里面每个字段就是图在节点间传递的数据。这里有个关键点:不是所有字段都需要被 Checkpointer 持久化。比如一些临时的中间变量、大对象(像原始文档内容),如果全塞进 State,Checkpointer 每次都要序列化一大坨,性能会很差。
我的做法是把 State 分成两类字段:核心对话状态(messages、thread 相关的元信息)必须持久化;临时计算数据尽量在节点内部处理完就丢掉,不进 State。举个例子:
from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] user_id: str thread_id: strmessages用add_messages这个 reducer,意味着新消息是追加而不是覆盖,这是多轮对话的基础。user_id和thread_id是业务标识,也存下来方便后续按用户查询。至于工具调用的中间结果,能不进 State 就不进。
提示:State 字段越精简,Checkpointer 的读写开销越小。我实测过一个项目,把 State 里一个存原始 PDF 文本的字段挪出去之后,单次状态写入从 200ms 降到了 30ms 左右。
3.3 脚本运行时的状态验证技巧
路径一虽然没有持久化,但验证逻辑时有个技巧很管用:用内存版 Checkpointer 模拟多轮。LangGraph 提供了MemorySaver,它把状态存在内存里,进程活着的时候可以跨 invoke 保持。这样你在脚本里就能模拟多轮对话,验证状态流转是否正确,而不需要真的起服务。
from langgraph.checkpoint.memory import MemorySaver checkpointer = MemorySaver() graph = builder.compile(checkpointer=checkpointer) config = {"configurable": {"thread_id": "test-1"}} graph.invoke({"messages": [("user", "你好")]}, config) graph.invoke({"messages": [("user", "刚才我说了什么")]}, config)两次 invoke 用同一个thread_id,第二次就能看到第一次的上下文。这个验证方式在路径一阶段极其重要,因为如果你连内存状态流转都没跑通,上了持久化只会更乱。我一般会在这个阶段把多轮、工具调用、异常分支都测一遍,确认 graph 逻辑没问题,再往服务化走。
4. 路径二实操:FastAPI 服务化与 Checkpointer 落地
这是绝大多数团队真正要走的路径,也是坑最多的一段。我把它拆成几个关键环节:服务框架选型、Checkpointer 后端选择、接口设计、并发处理。
4.1 为什么是 FastAPI,而不是 Flask
先说选型。Flask 当然能跑,我自己最早也是用 Flask。但 LangGraph 的场景有几个特点,让 FastAPI 明显更合适。
第一,LangGraph 的执行是异步友好的。很多节点涉及 LLM 调用、工具调用,都是 IO 密集操作。FastAPI 原生支持async/await,一个 worker 能扛住大量并发等待,而 Flask 的同步模型在这种场景下要么开很多线程,要么用 gevent 打补丁,都不优雅。
第二,流式输出。Agent 的回复经常需要流式吐给前端,FastAPI 的StreamingResponse配合 LangGraph 的astream用起来非常顺。Flask 做流式要绕不少弯。
第三,类型校验。FastAPI 基于 Pydantic,请求体、响应体的结构定义和校验是内置的。Agent 服务的输入输出结构往往比较复杂,有这层校验能省掉大量手写校验代码。
第四,自动文档。FastAPI 自带 Swagger UI,接口调试和对接前端时非常方便,不用额外维护接口文档。
所以我的建议很直接:新项目直接上 FastAPI,别在 Flask 上纠结。老项目如果已经是 Flask,也不是不能跑,但流式和并发这两块会比较难受。
4.2 Checkpointer 后端怎么选
这是路径二最关键的决策。LangGraph 支持多种 Checkpointer 后端,常见的有MemorySaver、SqliteSaver、PostgresSaver、RedisSaver。选哪个,取决于你的部署形态。
| 后端 | 适用场景 | 并发能力 | 持久化 | 我的评价 |
|---|---|---|---|---|
| MemorySaver | 本地调试、单进程测试 | 无 | 进程退出即丢 | 只用于开发 |
| SqliteSaver | 单机服务、小规模 | 弱,写锁竞争 | 有 | 单实例够用,多实例不行 |
| PostgresSaver | 多实例、生产环境 | 强 | 有 | 生产首选 |
| RedisSaver | 高并发、状态可容忍丢失 | 很强 | 可配置 | 适合会话型场景 |
我的实际经验是:单实例部署用 SqliteSaver 完全够,一旦要多实例或者并发上来了,直接上 PostgresSaver。SqliteSaver 的问题在于写操作会加锁,多个请求同时写状态时会排队,并发一高延迟就上去了。我压测过一个 SqliteSaver 的服务,20 个并发请求下,状态写入的平均延迟从 15ms 涨到了 200ms 以上,而换成 Postgres 之后基本稳定在 20ms 左右。
RedisSaver 适合那种状态可以容忍偶尔丢失、但对速度要求极高的场景,比如纯闲聊型 Agent。但如果是涉及订单、工单这类不能丢状态的业务,还是老老实实用 Postgres。
4.3 FastAPI 服务的最小可用骨架
下面是我常用的服务骨架,直接可以抄。核心是把 graph 的构建放在应用启动时做一次,全局复用,而不是每个请求都重新构建。
from contextlib import asynccontextmanager from fastapi import FastAPI from pydantic import BaseModel from langgraph.checkpoint.postgres import PostgresSaver graph = None @asynccontextmanager async def lifespan(app: FastAPI): global graph checkpointer = PostgresSaver.from_conn_string("postgresql://...") checkpointer.setup() graph = build_graph(checkpointer) yield app = FastAPI(lifespan=lifespan) class ChatRequest(BaseModel): thread_id: str message: str @app.post("/chat") async def chat(req: ChatRequest): config = {"configurable": {"thread_id": req.thread_id}} result = await graph.ainvoke( {"messages": [("user", req.message)]}, config ) return {"reply": result["messages"][-1].content}这里有几个细节值得说。lifespan是 FastAPI 推荐的启动/关闭钩子,graph 在这里构建一次,所有请求共享。checkpointer.setup()是 PostgresSaver 需要的,它会建表,第一次跑必须调,后面重复调也没事。接口用ainvoke而不是invoke,是为了不阻塞事件循环。
4.4 并发与 thread_id 隔离,最容易出事的地方
路径二上线后,我遇到最多的线上问题就是状态串了。根因几乎都是thread_id管理不当。有两种典型错误:
第一种,前端不传thread_id,后端用固定值或者随机生成。固定值会导致所有用户共享一个会话,随机生成会导致每次请求都是新会话,多轮对话直接失效。正确做法是前端为每个会话生成一个唯一thread_id,并在整个会话期间保持不变,通常存在浏览器的 localStorage 里。
第二种,thread_id用了用户 ID。这样同一个用户的所有会话会混在一起,用户开两个窗口聊天,内容会互相干扰。thread_id应该是会话级的,不是用户级的。
注意:
thread_id一旦确定,整个会话期间不能变。我见过有人为了“清空对话”而换thread_id,结果旧状态永远留在数据库里,越积越多。清空对话应该用专门的删除接口,而不是换 ID。
并发方面,FastAPI 默认用 uvicorn 跑,单 worker 是单进程。如果你的 graph 里有 CPU 密集的操作(比如本地跑 embedding),单 worker 会成为瓶颈,需要开多 worker。但多 worker 之后,MemorySaver 就不能用了,因为状态不共享,必须换成 Postgres 这类外部存储。这也是为什么我一直强调:只要考虑多实例,Checkpointer 就必须是外部后端。
5. 路径三实操:LangSmith 可观测与分布式状态
走到路径三,说明你的服务已经有了一定规模,开始需要“看得见”和“扛得住”。这一节讲两个核心:LangSmith 接入和分布式状态后端的注意事项。
5.1 LangSmith 接入,让黑盒变透明
LangGraph 和 LangSmith 是同一套体系,接入成本极低。你只需要设置两个环境变量:
export LANGCHAIN_TRACING_V2=true export LANGCHAIN_API_KEY=your_key export LANGCHAIN_PROJECT=your_project_name设置完之后,所有 graph 的执行会自动上报到 LangSmith。你打开控制台就能看到每一次执行的完整轨迹:每个节点的输入输出、每个工具调用的参数和返回、每一步的耗时、token 消耗、报错堆栈。
这个东西的价值,只有真正排查过线上问题的人才懂。我之前有个 Agent,用户反馈“有时候回复很慢”。没有 LangSmith 的时候,我只能加日志、猜、复现,折腾了两天。接入 LangSmith 之后,一眼就看出是某个工具调用在特定输入下会触发一个慢查询,耗时 8 秒。定位到问题,改起来就快了。
LangSmith 还有个我特别喜欢的用法:用它做回归测试。你可以把线上真实的执行轨迹标记为数据集,每次改完 prompt 或 graph 逻辑,跑一遍对比,看输出有没有变差。这比手写测试用例高效得多。
5.2 分布式状态后端的三个坑
上了多实例之后,状态后端从单机变成分布式,会冒出一些单机时不存在的问题。我踩过的有三个。
第一个是连接池配置。PostgresSaver 每个实例都会建连接,实例一多,连接数很容易打满数据库的上限。必须显式配置连接池大小,并且根据实例数算好总量。我一般会留 20% 的余量,比如数据库最大连接 100,4 个实例,每个实例的连接池就配 20 左右。
第二个是状态写入的时序。LangGraph 在每一步之后写状态,如果两个请求带着同一个thread_id并发进来,理论上会互相覆盖。正常业务里同一个会话不应该并发请求,但前端如果没做好防抖,用户狂点发送就会触发。我的处理方式是在服务层加一个基于thread_id的锁,同一个会话的请求串行处理。
第三个是状态数据的清理。Checkpointer 会一直累积状态,时间长了数据库会膨胀。必须定期清理过期会话。我一般会按业务定一个保留期,比如 30 天,写个定时任务删掉超期的 thread 数据。
5.3 三条路径的迁移成本对比
最后给个直观的对比,帮你判断现在该走哪条路,以及未来迁移的成本。
| 维度 | 路径一 | 路径二 | 路径三 |
|---|---|---|---|
| 状态持久化 | 无 | 单机持久化 | 分布式持久化 |
| 多轮对话 | 不支持 | 支持 | 支持 |
| 多实例 | 不支持 | 不支持 | 支持 |
| 可观测性 | 靠 print | 靠日志 | LangSmith 全链路 |
| 迁移到下一路径的成本 | 低(重构目录) | 中(换 Checkpointer + 加追踪) | - |
| 适合阶段 | 验证 | 上线 | 规模化 |
从路径一到路径二,主要成本在目录重构和接口设计;从路径二到路径三,主要成本在换 Checkpointer 后端和接入追踪。如果你一开始就按我上面说的目录结构组织代码,路径一到路径二的迁移会非常顺,基本就是加个 FastAPI 壳子的事。
6. 常见问题与排查技巧实录
这一节是我这些年攒下来的问题清单,基本都是线上真实遇到过的,按现象、原因、解决方式整理成表,方便你对照排查。
6.1 状态相关问题的速查表
| 现象 | 可能原因 | 排查方向 | 解决方式 |
|---|---|---|---|
| 多轮对话丢失上下文 | thread_id 每次请求都变 | 打印每次请求的 thread_id | 前端固定会话 ID |
| 不同用户对话串了 | thread_id 用了固定值或用户 ID | 检查 thread_id 生成逻辑 | 改为会话级唯一 ID |
| 服务重启后上下文没了 | 用了 MemorySaver | 检查 Checkpointer 类型 | 换成持久化后端 |
| 状态写入越来越慢 | State 字段太大 | 看单次写入耗时和 State 大小 | 精简 State,大对象移出 |
| 数据库连接打满 | 连接池未配置 | 看数据库连接数 | 配置连接池上限 |
6.2 工具调用相关的坑
LangGraph 的工具调用是另一个高频出问题的地方。最常见的现象是工具调用超时导致整个 graph 卡住。默认情况下,工具调用没有超时限制,如果某个外部 API 挂了,请求会一直挂着,占着 worker 不放。
我的做法是给所有工具调用加超时,用asyncio.wait_for包一层,超时就返回一个友好的错误信息,让 graph 继续往下走,而不是整个卡死。另外,工具调用的异常一定要在节点内部捕获,不要让异常冒泡到 graph 顶层,否则整个执行会中断,状态可能停在半路。
还有一个坑是工具返回的数据太大。比如一个搜索工具返回了几十 KB 的文本,直接塞进 messages,下一轮 LLM 调用的 token 会爆炸。我的处理方式是在工具内部做截断或摘要,只把关键信息返回给 graph。
6.3 日志与排查的实操心得
FastAPI 配合 uvicorn 时,有个经典问题:uvicorn 的访问日志和你的应用日志混在一起,排查时很难区分。我的做法是配置独立的 logger,给每个请求生成一个 request_id,在日志里带上 request_id 和 thread_id,这样一条请求的完整链路就能串起来。
另外,LangGraph 的节点执行默认不打日志。我会在每个节点的入口和出口加一行日志,记录节点名、耗时、关键状态字段。别小看这两行日志,线上出问题时,它能让你快速定位到是哪个节点慢、哪个节点报错。配合 LangSmith 的轨迹,基本能做到分钟级定位。
提示:日志里千万不要打完整的 messages 内容,尤其是涉及用户隐私的场景。我一般只打消息数量和最后一条的角色,具体内容需要时再去 LangSmith 里看。
7. 我个人的选型建议与踩坑体会
聊了这么多,最后说点掏心窝的。三条路径不是让你一步到位走到底,而是让你按需升级。我见过太多团队,demo 还没跑通就想着上分布式,结果基础设施搭了一堆,业务逻辑还没验证,纯属浪费。
我的建议是:先用路径一把 graph 逻辑彻底跑通,包括多轮、工具调用、异常分支;然后用路径二上线,单实例 + SqliteSaver 就能撑住早期用户;等并发真的上来了,再迁路径三。这个节奏最稳,每一步的投入都能立刻看到回报。
踩过的坑里,最让我印象深刻的是thread_id那个。当时线上用户反馈对话串了,我查了一整天才发现是前端在页面刷新时重新生成了thread_id。这种问题,文档里不会写,只有真踩过才知道。所以我现在做任何 Agent 服务,第一件事就是把thread_id的生命周期管理写进接口文档,前端必须严格遵守。
还有一个体会是:Checkpointer 的选择,本质是在持久性、并发性、成本之间做权衡。没有银弹,只有适合当前阶段的方案。Sqlite 简单但并发弱,Postgres 强但要多维护一个数据库,Redis 快但可能丢状态。想清楚你的业务能不能容忍状态丢失,答案自然就出来了。
最后分享一个小技巧:如果你不确定该用哪个 Checkpointer,就先用MemorySaver把服务跑起来,接口、并发、前端对接全部调通,最后再换后端。因为换 Checkpointer 在 LangGraph 里基本就是改一行代码的事,没必要一开始就纠结。把精力花在 graph 逻辑和接口设计上,那才是真正难改的部分。