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(checkpointercheckpointer) 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 切到 SQLiteMemorySaver的问题是进程一退状态就没了。本地开发阶段如果想保留状态最轻量的方案是 SQLiteimport sqlite3 from langgraph.checkpoint.sqlite import SqliteSaver conn sqlite3.connect(checkpoints.db, check_same_threadFalse) checkpointer SqliteSaver(conn)注意check_same_threadFalse这个参数。SQLite 默认不允许跨线程共享连接而 LangGraph 在某些执行模式下会用到多线程不加这个参数会报ProgrammingError。这个坑我踩过报错信息还特别隐晦排查了半天。3.3 脚本阶段的实操心得脚本阶段虽然简单但有几个习惯建议早点养成thread_id 用有意义的命名规则比如f{user_id}-{session_id}别用随机数不然调试的时候根本对不上是哪次会话。把 graph 的编译和调用分开编译一次可以复用别每次请求都重新 compile那样开销很大。日志里打印 node 名称和耗时LangGraph 支持 stream 模式可以拿到每个节点的执行事件早期就接上对后面排查问题帮助极大。提示脚本阶段不要急着上异步。LangGraph 的同步接口在单机场景下完全够用异步的复杂度留到服务化阶段再引入。4. 路径二FastAPI 服务化的完整落地4.1 为什么选 FastAPI 而不是 FlaskLangGraph 的核心接口既有同步也有异步版本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_modelChatResponse) 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(replylast_msg.content, thread_idreq.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, versionv2 ): if event[event] on_chat_model_stream: chunk event[data][chunk].content if chunk: yield fdata: {json.dumps({content: chunk})}\n\n yield data: [DONE]\n\n return StreamingResponse(event_generator(), media_typetext/event-stream)versionv2这个参数别漏v1 的事件格式和 v2 差别挺大网上很多老教程还是 v1 的写法照抄会踩坑。4.4 Checkpointer 换成 Postgres服务化之后内存和 SQLite 都不够用了得上 Postgres。LangGraph 官方提供了PostgresSaverfrom langgraph.checkpoint.postgres import PostgresSaver from psycopg_pool import ConnectionPool DB_URI postgresql://user:passlocalhost:5432/langgraph pool ConnectionPool(conninfoDB_URI, max_size20) 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_V2true export LANGCHAIN_API_KEYyour_api_key export LANGCHAIN_PROJECTmy-langgraph-project设置完之后所有 LangGraph 的执行会自动上报代码一行都不用改。这个设计我觉得挺聪明观测能力不该侵入业务逻辑。5.2 什么阶段该接 LangSmith我的建议是路径二一开始就接。原因很简单服务化之后调用方变多问题复现难度上升没有 Trace 基本靠猜。而 LangSmith 的接入成本几乎为零早接早受益。路径一阶段可以不接本地调试 print 就够了。路径三本身就是围绕 LangSmith 构建的自然要接。5.3 自定义 Trace 与元数据默认的 Trace 已经很有用但加上自定义元数据会更好查。比如把 user_id、session_id 打进 metadataconfig { 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 换成 PostgresLangSmith 同步接上。这个阶段的核心是接口稳定和状态可靠性能优化可以往后放。等真的有并发压力了再考虑连接池调优、缓存、限流这些。平台托管阶段更多是团队协作的需求。当多人要改同一张图、要共享 Trace、要做权限管理时托管平台的价值才真正体现出来。单打独斗的话自建 FastAPI 完全够用。最后分享一个我踩过的坑有一次线上 Agent 突然开始返回重复内容查了半天发现是 Checkpointer 的表被误删了一部分导致状态读取异常。从那以后我养成了一个习惯给 Checkpointer 的数据库单独做备份策略并且监控表的大小变化。状态这东西平时不起眼出问题的时候是真要命。如果你现在正卡在某个阶段我的建议是先明确当前最痛的点是什么然后只解决那一个。部署路径不是一次选定的是随着项目长大慢慢演进的。 LangGraph 部署实战：从脚本到服务的三条路径与避坑指南
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(checkpointercheckpointer) 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 切到 SQLiteMemorySaver的问题是进程一退状态就没了。本地开发阶段如果想保留状态最轻量的方案是 SQLiteimport sqlite3 from langgraph.checkpoint.sqlite import SqliteSaver conn sqlite3.connect(checkpoints.db, check_same_threadFalse) checkpointer SqliteSaver(conn)注意check_same_threadFalse这个参数。SQLite 默认不允许跨线程共享连接而 LangGraph 在某些执行模式下会用到多线程不加这个参数会报ProgrammingError。这个坑我踩过报错信息还特别隐晦排查了半天。3.3 脚本阶段的实操心得脚本阶段虽然简单但有几个习惯建议早点养成thread_id 用有意义的命名规则比如f{user_id}-{session_id}别用随机数不然调试的时候根本对不上是哪次会话。把 graph 的编译和调用分开编译一次可以复用别每次请求都重新 compile那样开销很大。日志里打印 node 名称和耗时LangGraph 支持 stream 模式可以拿到每个节点的执行事件早期就接上对后面排查问题帮助极大。提示脚本阶段不要急着上异步。LangGraph 的同步接口在单机场景下完全够用异步的复杂度留到服务化阶段再引入。4. 路径二FastAPI 服务化的完整落地4.1 为什么选 FastAPI 而不是 FlaskLangGraph 的核心接口既有同步也有异步版本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_modelChatResponse) 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(replylast_msg.content, thread_idreq.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, versionv2 ): if event[event] on_chat_model_stream: chunk event[data][chunk].content if chunk: yield fdata: {json.dumps({content: chunk})}\n\n yield data: [DONE]\n\n return StreamingResponse(event_generator(), media_typetext/event-stream)versionv2这个参数别漏v1 的事件格式和 v2 差别挺大网上很多老教程还是 v1 的写法照抄会踩坑。4.4 Checkpointer 换成 Postgres服务化之后内存和 SQLite 都不够用了得上 Postgres。LangGraph 官方提供了PostgresSaverfrom langgraph.checkpoint.postgres import PostgresSaver from psycopg_pool import ConnectionPool DB_URI postgresql://user:passlocalhost:5432/langgraph pool ConnectionPool(conninfoDB_URI, max_size20) 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_V2true export LANGCHAIN_API_KEYyour_api_key export LANGCHAIN_PROJECTmy-langgraph-project设置完之后所有 LangGraph 的执行会自动上报代码一行都不用改。这个设计我觉得挺聪明观测能力不该侵入业务逻辑。5.2 什么阶段该接 LangSmith我的建议是路径二一开始就接。原因很简单服务化之后调用方变多问题复现难度上升没有 Trace 基本靠猜。而 LangSmith 的接入成本几乎为零早接早受益。路径一阶段可以不接本地调试 print 就够了。路径三本身就是围绕 LangSmith 构建的自然要接。5.3 自定义 Trace 与元数据默认的 Trace 已经很有用但加上自定义元数据会更好查。比如把 user_id、session_id 打进 metadataconfig { 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 换成 PostgresLangSmith 同步接上。这个阶段的核心是接口稳定和状态可靠性能优化可以往后放。等真的有并发压力了再考虑连接池调优、缓存、限流这些。平台托管阶段更多是团队协作的需求。当多人要改同一张图、要共享 Trace、要做权限管理时托管平台的价值才真正体现出来。单打独斗的话自建 FastAPI 完全够用。最后分享一个我踩过的坑有一次线上 Agent 突然开始返回重复内容查了半天发现是 Checkpointer 的表被误删了一部分导致状态读取异常。从那以后我养成了一个习惯给 Checkpointer 的数据库单独做备份策略并且监控表的大小变化。状态这东西平时不起眼出问题的时候是真要命。如果你现在正卡在某个阶段我的建议是先明确当前最痛的点是什么然后只解决那一个。部署路径不是一次选定的是随着项目长大慢慢演进的。