1. 从脚本到服务，LangGraph 部署到底在解决什么问题
很多人第一次接触 LangGraph，都是在 Jupyter Notebook 或者一个单文件 Python 脚本里跑通一个带工具调用的 Agent，看着它一步步推理、调用搜索、返回结果，感觉挺爽。但爽完之后问题就来了：这东西怎么给别人用？怎么让前端调？怎么保证多轮对话的状态不丢？怎么在服务器上稳定跑着而不是每次手动python main.py？
这就是“从脚本到服务”的核心命题。脚本阶段你关心的是逻辑能不能跑通，服务阶段你关心的是并发、状态持久化、可观测性、部署形态这四件事。LangGraph 本身是一个编排框架，它不负责帮你起 HTTP 服务，也不负责帮你管理进程，所以“部署路径”这件事必须结合外围工具来谈。
我前后用 LangGraph 做过几个不同规模的项目，从个人玩具到小团队内部工具都有，踩过的坑集中在三个地方：一是状态存哪（Checkpointer 选型），二是怎么暴露接口（FastAPI 还是别的），三是怎么观测和调试（LangSmith 要不要接）。这三件事组合起来，基本就决定了你的部署路径。
这篇文章我会把三条常见的部署路径拆开讲：纯脚本 + 本地持久化、FastAPI 服务化 + 数据库 Checkpointer、平台托管 + LangSmith 全链路观测。每条路径我都会说清楚它适合什么场景、核心配置怎么写、参数为什么这么选，以及我在实际项目里踩过的具体坑。如果你现在手里有一个跑通的 LangGraph 脚本，正发愁怎么把它变成别人能用的服务，这篇应该能帮你少走点弯路。
2. 三条部署路径的整体设计与选型逻辑
2.1 为什么是三条路径而不是一条
部署这件事没有银弹，因为不同阶段的需求差异太大。我见过有人一上来就上 K8s + 消息队列 + 分布式 Checkpointer，结果项目就他一个人用，纯属给自己找罪受。也见过有人把生产环境的 Agent 还跑在nohup python main.py &上，进程一挂全完蛋。
所以我把路径按复杂度和适用阶段分成三档，你可以对号入座：
| 路径 | 核心组成 | 适用阶段 | 状态存储 | 观测能力 |
|---|---|---|---|---|
| 路径一：脚本直跑 | LangGraph + 内存/SQLite Checkpointer | 本地验证、Demo | 内存或本地文件 | 靠 print 和日志 |
| 路径二：FastAPI 服务化 | LangGraph + FastAPI + 数据库 Checkpointer | 内部工具、小规模线上 | Postgres/Redis | 日志 + 自定义埋点 |
| 路径三：平台托管 | LangGraph Platform + LangSmith | 团队协作、正式产品 | 平台托管 | 全链路 Trace |
这个分档不是绝对的，路径二里你也可以接 LangSmith，路径三里你也可以自己写 FastAPI。分档的目的是让你先想清楚当前阶段最缺什么，而不是盲目堆技术。
2.2 选型背后的三个核心判断
第一个判断：状态要不要跨进程存活。如果你的 Agent 是无状态的（每次请求都是独立的一问一答，不需要记住上一轮），那 Checkpointer 用内存的MemorySaver就够了，进程重启丢状态也无所谓。但只要涉及多轮对话、人工审核中断（human-in-the-loop）、长时间运行的任务，状态就必须落到进程外，这时候数据库 Checkpointer 就是刚需。
第二个判断：调用方是谁。如果只有你自己在命令行里跑，脚本就够了。如果前端、其他服务、甚至第三方要调，那就必须有一个稳定的接口层，FastAPI 是目前 Python 生态里最顺手的选择，异步支持好、和 LangGraph 的 async 接口天然契合。
第三个判断：出问题了你怎么查。这是最容易被忽略的一点。Agent 的执行链路很长，一次请求可能经过七八个节点、好几次工具调用，出了问题光看日志根本定位不到是哪一步的 prompt 出了问题。LangSmith 的价值就在这里，它把整条链路可视化，每个节点的输入输出、耗时、token 消耗都清清楚楚。
2.3 一个常见的选型误区
我见过不少人纠结“要不要一开始就上最完整的方案”。我的建议是按需演进，但预留接口。具体来说：Checkpointer 从一开始就用抽象接口（LangGraph 本身就做了这层抽象，MemorySaver和PostgresSaver的调用方式几乎一样），这样从内存切到数据库只需要改一行初始化代码。FastAPI 的接口层也可以先写薄一点，后面加鉴权、加限流都是增量的事。
真正不建议提前做的是分布式和微服务拆分。LangGraph 的图本身就是一个执行单元，把它拆成多个服务反而增加了状态同步的复杂度，得不偿失。
3. 路径一：脚本直跑与本地持久化实操
3.1 最小可运行结构
先看最朴素的形态。一个 LangGraph 脚本通常长这样：
from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver from typing import TypedDict, Annotated from langgraph.graph.message import add_messages class State(TypedDict): messages: Annotated[list, add_messages] def chatbot(state: State): return {"messages": [("ai", "收到你的消息了")]} builder = StateGraph(State) builder.add_node("chatbot", chatbot) builder.add_edge(START, "chatbot") builder.add_edge("chatbot", END) checkpointer = MemorySaver() graph = builder.compile(checkpointer=checkpointer) config = {"configurable": {"thread_id": "user-001"}} result = graph.invoke({"messages": [("user", "你好")]}, config) print(result)这段代码里最关键的是thread_id。它是 LangGraph 区分不同会话的标识，同一个thread_id的多次调用会共享状态历史。很多人第一次用会忘记传 config，结果发现多轮对话记不住上下文，其实就是没指定 thread_id。
3.2 从 MemorySaver 切到 SQLite
MemorySaver的问题是进程一退状态就没了。本地开发阶段如果想保留状态，最轻量的方案是 SQLite：
import sqlite3 from langgraph.checkpoint.sqlite import SqliteSaver conn = sqlite3.connect("checkpoints.db", check_same_thread=False) checkpointer = SqliteSaver(conn)注意check_same_thread=False这个参数。SQLite 默认不允许跨线程共享连接，而 LangGraph 在某些执行模式下会用到多线程，不加这个参数会报ProgrammingError。这个坑我踩过，报错信息还特别隐晦，排查了半天。
3.3 脚本阶段的实操心得
脚本阶段虽然简单，但有几个习惯建议早点养成：
- thread_id 用有意义的命名规则，比如
f"{user_id}-{session_id}"，别用随机数，不然调试的时候根本对不上是哪次会话。 - 把 graph 的编译和调用分开，编译一次可以复用，别每次请求都重新 compile，那样开销很大。
- 日志里打印 node 名称和耗时，LangGraph 支持 stream 模式，可以拿到每个节点的执行事件，早期就接上对后面排查问题帮助极大。
提示：脚本阶段不要急着上异步。LangGraph 的同步接口在单机场景下完全够用，异步的复杂度留到服务化阶段再引入。
4. 路径二：FastAPI 服务化的完整落地
4.1 为什么选 FastAPI 而不是 Flask
LangGraph 的核心接口既有同步也有异步版本，ainvoke、astream这些异步方法在 FastAPI 里可以直接 await，而在 Flask 里要么用线程池包一层，要么放弃异步。Agent 的执行往往涉及大量 IO 等待（调模型、调工具），异步能显著提升并发能力。
另外 FastAPI 自带 Pydantic 校验和 OpenAPI 文档，接口定义清晰，前端对接成本低。我试过用 Flask 写同样的接口，代码量大概多三成，还得自己处理请求体校验。
4.2 项目目录结构
一个能上生产的 FastAPI + LangGraph 项目，我一般这么组织：
app/ ├── main.py # FastAPI 入口 ├── graph/ │ ├── builder.py # 图定义与编译 │ ├── nodes.py # 各节点逻辑 │ └── state.py # State 定义 ├── api/ │ ├── routes.py # 路由 │ └── schemas.py # 请求/响应模型 ├── core/ │ ├── config.py # 配置管理 │ └── checkpointer.py # Checkpointer 初始化 └── utils/ └── logger.py这个结构的好处是图逻辑和接口逻辑解耦。图怎么改不影响接口，接口怎么改不影响图。我见过把所有东西塞一个main.py的写法，超过 500 行之后基本没法维护。
4.3 核心接口实现
最关键的两个接口是发起对话和流式返回。先看发起对话：
from fastapi import FastAPI from pydantic import BaseModel from app.graph.builder import get_graph app = FastAPI() graph = get_graph() class ChatRequest(BaseModel): message: str thread_id: str class ChatResponse(BaseModel): reply: str thread_id: str @app.post("/chat", response_model=ChatResponse) async def chat(req: ChatRequest): config = {"configurable": {"thread_id": req.thread_id}} result = await graph.ainvoke( {"messages": [("user", req.message)]}, config ) last_msg = result["messages"][-1] return ChatResponse(reply=last_msg.content, thread_id=req.thread_id)流式接口用astream_events，可以拿到 token 级别的输出：
from fastapi.responses import StreamingResponse import json @app.post("/chat/stream") async def chat_stream(req: ChatRequest): config = {"configurable": {"thread_id": req.thread_id}} async def event_generator(): async for event in graph.astream_events( {"messages": [("user", req.message)]}, config, version="v2" ): if event["event"] == "on_chat_model_stream": chunk = event["data"]["chunk"].content if chunk: yield f"data: {json.dumps({'content': chunk})}\n\n" yield "data: [DONE]\n\n" return StreamingResponse(event_generator(), media_type="text/event-stream")version="v2"这个参数别漏，v1 的事件格式和 v2 差别挺大，网上很多老教程还是 v1 的写法，照抄会踩坑。
4.4 Checkpointer 换成 Postgres
服务化之后，内存和 SQLite 都不够用了，得上 Postgres。LangGraph 官方提供了PostgresSaver：
from langgraph.checkpoint.postgres import PostgresSaver from psycopg_pool import ConnectionPool DB_URI = "postgresql://user:pass@localhost:5432/langgraph" pool = ConnectionPool(conninfo=DB_URI, max_size=20) checkpointer = PostgresSaver(pool) checkpointer.setup() # 首次运行需要建表setup()只需要跑一次，它会创建checkpoints、checkpoint_writes等表。生产环境建议把这一步放到独立的迁移脚本里，别每次启动都跑。
连接池的max_size要根据并发量调。我一般按预期并发数 × 1.5来估，20 个连接大概能撑住几十 QPS 的对话场景。太小会排队，太大浪费数据库连接。
4.5 服务化阶段的避坑清单
- uvicorn 日志丢失问题：用
uvicorn app.main:app --log-config指定日志配置，或者在代码里显式配置 logging，否则 LangGraph 内部的日志可能被吞掉。 - Windows 打包：如果要在 Windows 上打包成 exe，注意
psycopg的二进制依赖，建议用pyinstaller的--collect-all参数把依赖全带上。 - 超时控制：Agent 执行可能很久，FastAPI 默认没有超时，建议在反向代理层（如 Nginx）设置
proxy_read_timeout，否则长任务会被掐断。 - 并发写同一 thread：同一个 thread_id 并发请求会导致状态冲突，建议在业务层加锁，或者用不同的 thread_id 隔离。
5. 路径三：LangSmith 接入与全链路观测
5.1 LangSmith 到底解决什么问题
前面说过，Agent 的链路很长，出了问题难定位。LangSmith 的核心价值是把每次执行的完整 Trace记录下来：哪个节点、什么输入、什么输出、耗时多少、token 消耗多少、有没有报错。它相当于给 Agent 装了一个黑匣子。
接入方式简单到有点反直觉，只需要设置环境变量：
export LANGCHAIN_TRACING_V2=true export LANGCHAIN_API_KEY=your_api_key export LANGCHAIN_PROJECT=my-langgraph-project设置完之后，所有 LangGraph 的执行会自动上报，代码一行都不用改。这个设计我觉得挺聪明，观测能力不该侵入业务逻辑。
5.2 什么阶段该接 LangSmith
我的建议是路径二一开始就接。原因很简单：服务化之后调用方变多，问题复现难度上升，没有 Trace 基本靠猜。而 LangSmith 的接入成本几乎为零，早接早受益。
路径一阶段可以不接，本地调试 print 就够了。路径三本身就是围绕 LangSmith 构建的，自然要接。
5.3 自定义 Trace 与元数据
默认的 Trace 已经很有用，但加上自定义元数据会更好查。比如把 user_id、session_id 打进 metadata：
config = { "configurable": {"thread_id": req.thread_id}, "metadata": { "user_id": req.user_id, "env": "production" }, "run_name": "chat_endpoint" }这样在 LangSmith 后台可以按 user_id 过滤，排查某个用户的具体问题特别方便。run_name则让同一次请求的多个节点归到一组，看起来更清晰。
5.4 平台托管形态的取舍
LangGraph 官方有托管平台，把图部署上去之后，接口、状态、观测都是现成的。适合团队协作和正式产品，省去了自己维护 FastAPI 和数据库的功夫。
但它也有代价：灵活性下降。比如你想在接口层做一些特殊的鉴权逻辑，或者想混用其他框架，托管平台就不太方便。我的经验是，如果团队里没有专门的运维，托管平台能省很多事；如果有能力自己维护，自建 FastAPI 的可控性更高。
6. 常见问题与排查技巧实录
6.1 状态相关的问题
问题：多轮对话记不住上下文。排查顺序：先确认 config 里有没有传 thread_id，再确认 Checkpointer 是不是每次请求都重新初始化了。我见过有人在路由函数里checkpointer = MemorySaver()，每次请求都是新的，状态自然存不住。Checkpointer 应该是全局单例。
问题：切换 Checkpointer 后旧数据读不出来。不同 Checkpointer 的表结构不一样，内存和 SQLite 的数据没法直接迁移。切换时要么接受历史丢失，要么写迁移脚本。生产环境切换建议在低峰期做，并提前备份。
6.2 接口相关的问题
问题：流式接口前端收不到数据。大概率是响应头的问题。SSE 需要Content-Type: text/event-stream，并且要禁用缓冲。Nginx 下还要加proxy_buffering off，否则数据会被攒着一起发。
问题：并发请求下状态串了。检查 thread_id 的生成逻辑，确保每个会话唯一。如果多个用户共用一个 thread_id，状态会互相覆盖。这个 bug 很隐蔽，因为单用户测试时完全正常。
6.3 排查速查表
| 现象 | 可能原因 | 排查方向 |
|---|---|---|
| 状态不持久 | Checkpointer 未全局单例 | 检查初始化位置 |
| 流式无输出 | 响应头或代理缓冲 | 检查 Content-Type 和 Nginx 配置 |
| 并发状态错乱 | thread_id 冲突 | 检查 ID 生成规则 |
| 日志缺失 | uvicorn 日志配置 | 显式配置 logging |
| 连接池耗尽 | max_size 太小 | 按并发量调整 |
| Trace 不上报 | 环境变量未生效 | 检查 LANGCHAIN_ 开头的变量 |
6.4 几条独家避坑经验
第一，thread_id 的命名规则要能反查。我习惯用{业务前缀}-{用户ID}-{时间戳}，出问题时一眼能看出是哪个用户哪个时段。
第二，Checkpointer 的 setup 别放在应用启动流程里。多实例部署时会有并发建表的竞争，放到独立的初始化脚本里更稳。
第三，LangSmith 的采样率要控制。全量上报在高并发下会产生大量数据，可以在 config 里按比例采样，或者只对报错的请求上报。
第四，本地开发和生产用不同的 LangSmith project。混在一起查起来很乱，用LANGCHAIN_PROJECT环境变量区分开。
7. 三条路径的演进节奏与我的实际体会
回到最开始的问题：从脚本到服务，到底该怎么走。我的实际体会是，别跳步，但每一步都留好升级的口子。
脚本阶段把图逻辑跑通，Checkpointer 用内存的，重点是把节点划分清楚、状态定义合理。这个阶段图设计得好不好，直接决定了后面服务化顺不顺。我见过图设计得一团糟的，服务化之后每个接口都要写一堆适配代码。
服务化阶段把 FastAPI 接上，Checkpointer 换成 Postgres，LangSmith 同步接上。这个阶段的核心是接口稳定和状态可靠，性能优化可以往后放。等真的有并发压力了，再考虑连接池调优、缓存、限流这些。
平台托管阶段更多是团队协作的需求。当多人要改同一张图、要共享 Trace、要做权限管理时，托管平台的价值才真正体现出来。单打独斗的话，自建 FastAPI 完全够用。
最后分享一个我踩过的坑：有一次线上 Agent 突然开始返回重复内容，查了半天发现是 Checkpointer 的表被误删了一部分，导致状态读取异常。从那以后我养成了一个习惯，给 Checkpointer 的数据库单独做备份策略，并且监控表的大小变化。状态这东西平时不起眼，出问题的时候是真要命。
如果你现在正卡在某个阶段，我的建议是先明确当前最痛的点是什么，然后只解决那一个。部署路径不是一次选定的，是随着项目长大慢慢演进的。
