1. 从脚本到服务,为什么部署这一步值得单独拿出来聊
LangGraph 这个框架,我最早是在做多智能体编排的时候接触的。当时用 LangChain 的 Agent 做任务分发,链路一长就乱,状态管理全靠自己手写字典传来传去,调试的时候打印日志能刷满整个屏幕。后来切到 LangGraph,用图结构来定义节点和边,状态在节点之间流转,逻辑一下子清晰了很多。但问题也随之而来:本地跑通的脚本,怎么变成一个别人也能调用的服务?
这个问题看起来简单,实际上涉及的东西不少。脚本模式下,你只需要关心逻辑对不对;一旦要变成服务,就得考虑并发请求怎么处理、状态怎么持久化、多个会话之间怎么隔离、服务挂了之后能不能恢复、日志怎么打、接口怎么设计。这些在脚本阶段都可以忽略,但在服务阶段每一个都是必须解决的问题。
LangGraph 官方给出的部署路径大致可以分成三条:一是自己用 FastAPI 包一层,完全自托管;二是用 LangGraph Platform 的托管服务,省去运维;三是介于两者之间的混合方案,核心逻辑自己管,部分能力借助平台。这三条路径没有绝对的好坏,关键看你的团队规模、运维能力和业务对数据可控性的要求。
这篇文章我会把这三条路径逐一拆开,讲清楚每条路径适合什么场景、具体怎么落地、有哪些坑。如果你正在把 LangGraph 的原型往生产环境推,或者还在犹豫要不要上托管服务,下面的内容应该能帮你省不少时间。
2. 三条部署路径的整体设计与选型逻辑
2.1 先搞清楚三条路径的本质区别
很多人一上来就问“哪个方案最好”,这个问题本身就不太对。三条路径的区别不在于技术优劣,而在于你把哪些责任留给自己、哪些交给平台。
路径一:FastAPI 自托管。LangGraph 的图逻辑跑在你自己的进程里,用 FastAPI 暴露 HTTP 接口,状态持久化用 Checkpointer 接你自己的数据库。所有东西都在你的掌控之下,但运维、扩缩容、监控都得自己搞。
路径二:LangGraph Platform 托管。你把图定义好,通过 CLI 推到平台上,平台帮你处理 API 暴露、状态存储、任务队列、水平扩展。你只需要关心图逻辑本身,运维的事情基本不用管。
路径三:混合模式。核心的图逻辑和敏感数据留在自己的服务里,但借助平台的部分能力,比如用平台的 Checkpointer 做状态持久化,或者用 LangSmith 做链路追踪和调试。这种模式适合那些对数据可控性有要求、但又不想完全自己造轮子的团队。
选哪条路径,我一般会问三个问题:第一,你的数据能不能出你自己的网络边界?第二,你的团队有没有人能维护一套后端服务?第三,你的请求量级和并发要求是什么?这三个问题的答案基本就能定位到合适的路径。
2.2 选型时容易忽略的几个维度
除了上面三个问题,还有几个维度容易被忽略,但实际影响很大。
状态持久化的需求。LangGraph 的 Checkpointer 机制允许你在每个节点执行后保存状态,这样服务重启或者请求中断后可以从上次的状态继续。如果你做的是多轮对话或者长流程任务,Checkpointer 几乎是必须的。自托管的话,你需要自己选数据库、设计表结构、处理并发写入;托管方案通常内置了这些能力。
可观测性。LangGraph 的执行链路可能很长,一个请求经过十几个节点,每个节点的输入输出、耗时、异常都需要记录。LangSmith 在这方面做得比较成熟,能直接看到整个图的执行轨迹。自托管的话,你需要自己埋点、自己搭追踪系统,工作量不小。
冷启动和扩缩容。如果你的服务请求量波动大,自托管需要考虑进程池、容器编排、自动扩缩容。托管平台通常在这方面有现成的方案,但代价是你对底层资源的控制力会弱一些。
成本结构。自托管的成本主要是服务器和人力,托管方案通常是按调用量或资源占用计费。量小的时候托管可能更划算,量大的时候自托管的经济性会更好。这个需要根据实际业务量算一笔账。
2.3 一个实用的决策流程
我一般会按下面的流程来决策:
- 如果数据绝对不能出内网,直接选自托管,没有讨论余地。
- 如果团队里没有人能维护后端服务,优先考虑托管方案。
- 如果请求量很大且稳定,自托管的经济性通常更好。
- 如果还在快速迭代阶段,托管方案的迭代速度更快。
- 如果介于以上情况之间,考虑混合模式。
这个流程不是绝对的,但能帮你快速缩小选择范围。
3. FastAPI 自托管路径的核心细节与实操要点
3.1 项目结构怎么设计才不乱
自托管的第一步是把项目结构搭好。我见过太多项目一开始就一个main.py写到底,后面加功能的时候改得痛不欲生。LangGraph 的项目结构我建议按下面的方式来组织:
project/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 入口 │ ├── config.py # 配置管理 │ ├── graphs/ # LangGraph 图定义 │ │ ├── __init__.py │ │ ├── chat_graph.py │ │ └── task_graph.py │ ├── nodes/ # 节点函数 │ │ ├── __init__.py │ │ ├── llm_nodes.py │ │ └── tool_nodes.py │ ├── state/ # 状态定义 │ │ ├── __init__.py │ │ └── schemas.py │ ├── checkpointers/ # 持久化配置 │ │ └── __init__.py │ ├── api/ # 路由 │ │ ├── __init__.py │ │ ├── v1/ │ │ │ ├── __init__.py │ │ │ ├── chat.py │ │ │ └── tasks.py │ │ └── deps.py # 依赖注入 │ └── utils/ # 工具函数 │ ├── __init__.py │ └── logging.py ├── tests/ ├── .env ├── requirements.txt └── Dockerfile这个结构的好处是职责清晰:graphs放图定义,nodes放节点逻辑,state放状态 schema,api放路由。后面加新图或者新接口的时候,不会互相干扰。
注意:不要把图的定义和节点的实现混在一个文件里。图定义应该只关心节点怎么连、边怎么走,节点的具体逻辑放在
nodes目录下。这样改节点逻辑的时候不会影响到图的结构。
3.2 FastAPI 接口层怎么写才稳
FastAPI 的接口层是外部请求进入系统的入口,这里的设计直接影响服务的稳定性和可维护性。
请求模型的设计。每个接口都应该有明确的请求体和响应体模型,用 Pydantic 来定义。不要直接用dict接收请求,那样既没有类型检查,也没有文档。
from pydantic import BaseModel, Field from typing import Optional class ChatRequest(BaseModel): message: str = Field(..., description="用户输入的消息") session_id: str = Field(..., description="会话 ID,用于状态隔离") stream: bool = Field(default=False, description="是否流式返回") class ChatResponse(BaseModel): session_id: str reply: str node_trace: Optional[list] = None会话隔离。LangGraph 的 Checkpointer 用thread_id来区分不同的会话。在 FastAPI 层,你需要把外部传入的session_id映射到thread_id。我一般直接用session_id作为thread_id,但要注意做一层校验,防止用户传入非法字符。
流式返回。LangGraph 支持流式输出,FastAPI 可以用StreamingResponse来返回。流式返回的坑在于错误处理:如果流已经开始返回了,后面节点报错,你没法再改 HTTP 状态码。我的做法是在流开始之前先做一轮校验,确保图能正常启动,然后再开始流式输出。
from fastapi.responses import StreamingResponse @app.post("/chat/stream") async def chat_stream(request: ChatRequest): async def event_generator(): try: async for event in graph.astream( {"messages": [HumanMessage(content=request.message)]}, config={"configurable": {"thread_id": request.session_id}} ): yield f"data: {json.dumps(event)}\n\n" except Exception as e: yield f"data: {json.dumps({'error': str(e)})}\n\n" return StreamingResponse( event_generator(), media_type="text/event-stream" )超时和重试。LLM 调用可能很慢,节点执行可能超时。在 FastAPI 层要设置合理的超时时间,并且对可重试的错误做重试。我一般会在节点层面做重试,而不是在接口层,因为节点层面更清楚哪些错误可以重试。
3.3 Checkpointer 选型和配置
Checkpointer 是 LangGraph 自托管路径里最关键的一环。它决定了状态存在哪里、怎么存、能不能恢复。
内存 Checkpointer。MemorySaver是最简单的,状态存在进程内存里。适合开发和测试,但服务重启后状态就丢了,也不能多进程共享。生产环境不要用。
SQLite Checkpointer。SqliteSaver把状态存在 SQLite 文件里,适合单机部署。优点是零配置,缺点是并发写入性能有限,不适合高并发场景。
Postgres Checkpointer。PostgresSaver是最常用的生产级方案。支持并发写入、事务、连接池,能配合 Postgres 的高可用方案。配置的时候需要注意连接池大小和超时时间。
from langgraph.checkpoint.postgres import PostgresSaver from psycopg_pool import ConnectionPool pool = ConnectionPool( conninfo="postgresql://user:pass@localhost:5432/langgraph", min_size=2, max_size=10, timeout=30 ) checkpointer = PostgresSaver(pool) checkpointer.setup() # 首次运行需要建表注意:
checkpointer.setup()只需要在首次部署时执行一次,不要在每次服务启动时都调用。我见过有人在启动脚本里每次都调,结果每次重启都去检查表结构,白白增加启动时间。
Redis Checkpointer。如果你的状态需要快速读写,Redis 也是一个选择。但 Redis 的持久化能力不如 Postgres,需要根据业务对状态可靠性的要求来权衡。
选 Checkpointer 的时候,核心考虑三个因素:并发量、状态可靠性要求、运维复杂度。并发量小、可靠性要求不高,SQLite 就够;并发量大、要求高可用,上 Postgres;需要极速读写且能接受一定的数据丢失风险,考虑 Redis。
3.4 日志和可观测性怎么补
自托管路径下,日志和可观测性需要自己搞。LangGraph 的执行链路长,没有好的日志会很难排查问题。
结构化日志。不要用print,用logging模块,并且输出 JSON 格式的日志,方便后续采集和分析。
import logging import json class JsonFormatter(logging.Formatter): def format(self, record): log_data = { "timestamp": self.formatTime(record), "level": record.levelname, "message": record.getMessage(), "logger": record.name, } if hasattr(record, "extra_data"): log_data.update(record.extra_data) return json.dumps(log_data)节点级别的追踪。在每个节点的入口和出口打日志,记录节点名称、输入状态、输出状态、耗时。这样出问题的时候能快速定位到是哪个节点。
LangSmith 集成。即使自托管,也可以用 LangSmith 做追踪。只需要设置环境变量,LangGraph 会自动把执行轨迹上报到 LangSmith。
export LANGCHAIN_TRACING_V2=true export LANGCHAIN_API_KEY=your_api_key export LANGCHAIN_PROJECT=your_project这样你在 LangSmith 的界面上就能看到每次执行的完整轨迹,包括每个节点的输入输出和耗时。调试的时候非常有用。
4. LangGraph Platform 托管路径的落地细节
4.1 托管方案适合什么样的团队
LangGraph Platform 的托管方案,核心卖点是省去运维。你只需要定义图,用 CLI 推到平台上,平台帮你处理 API 暴露、状态存储、扩缩容、监控。
这个方案适合几类团队:一是小团队,没有专职的后端运维;二是快速迭代阶段,不想在基础设施上花时间;三是对弹性扩缩容有要求,但不想自己搞容器编排。
不太适合的情况:数据绝对不能出内网、请求量极大且稳定(自托管经济性更好)、需要深度定制底层行为。
4.2 从本地图到平台部署的完整流程
托管方案的部署流程大致分几步:
第一步,安装 CLI 并登录。
pip install langgraph-cli langgraph login第二步,准备项目结构。托管方案对项目结构有要求,核心是langgraph.json配置文件。
{ "dependencies": ["."], "graphs": { "chat": "./app/graphs/chat_graph.py:graph" }, "env": ".env" }这个配置告诉平台:依赖是什么、图定义在哪个文件的哪个变量、环境变量从哪里读。
第三步,本地测试。
langgraph dev这个命令会在本地启动一个开发服务器,模拟平台的行为。你可以在本地测试图能不能正常运行、接口能不能正常调用。
第四步,部署。
langgraph deploy部署完成后,平台会给你一个 API 地址,你可以直接调用。
4.3 托管方案的状态管理和并发处理
托管方案内置了 Checkpointer,你不需要自己配数据库。平台会根据你的图定义自动处理状态持久化。但有几个点需要注意:
thread_id 的管理。托管方案用thread_id来区分会话,你需要在调用的时候传入。平台会自动把状态和thread_id关联起来。
并发限制。托管方案通常有并发限制,具体取决于你的套餐。如果并发量超过限制,请求会排队。你需要根据业务量选择合适的套餐。
状态保留策略。平台通常会对状态设置保留期限,超过期限的状态会被清理。如果你的业务需要长期保留状态,需要确认平台的保留策略,或者自己定期导出状态。
4.4 托管方案的调试和监控
托管方案通常内置了监控面板,能看到请求量、延迟、错误率等指标。LangSmith 的集成也更顺畅,基本是开箱即用。
调试的时候,我一般会先在本地用langgraph dev跑通,再部署到平台。平台上的问题,大部分能在本地复现。如果本地没问题、平台上有问题,通常是环境变量或者依赖版本的问题。
注意:托管方案的环境变量管理和本地不一样。本地是从
.env文件读,平台上需要在控制台配置。部署前一定要确认所有需要的环境变量都配好了,否则会出现本地能跑、线上报错的情况。
5. 混合模式:自托管与托管的折中方案
5.1 什么情况下选混合模式
混合模式适合那些“既要又要”的场景:数据敏感的部分要自己管,但不想完全自己造轮子。
具体来说,几种典型的混合场景:
状态自托管、追踪用平台。图逻辑和状态存储都在自己的服务里,但用 LangSmith 做追踪。这样数据不出自己的边界,但调试体验和托管方案差不多。
核心图自托管、辅助图用平台。主业务流程的图自己部署,一些辅助性的图(比如内容审核、格式转换)用平台托管。这样核心逻辑可控,辅助逻辑省心。
开发用平台、生产用自托管。开发阶段用平台快速迭代,验证通过后再迁移到自托管的生产环境。这样开发效率高,生产环境也可控。
5.2 混合模式的技术实现要点
混合模式的技术难点在于两套系统的衔接。
状态同步。如果部分图在自托管服务、部分图在平台,状态怎么同步是个问题。我的做法是让自托管服务作为主状态存储,平台上的图通过 API 调用自托管服务来读写状态。
统一入口。外部请求应该有一个统一的入口,由入口来决定路由到自托管服务还是平台。这个入口可以是一个简单的反向代理,也可以是一个轻量的路由服务。
日志聚合。两套系统的日志需要聚合到一起,否则排查问题的时候要在两个地方来回看。我一般会用统一的日志采集方案,把两边的日志都送到同一个地方。
5.3 混合模式的运维复杂度
混合模式的运维复杂度比纯自托管低,但比纯托管高。你需要维护自托管的部分,同时也要管理平台上的配置。
我的经验是,混合模式适合有一定运维能力、但对数据可控性有要求的团队。如果团队完全没有运维能力,纯托管更合适;如果团队运维能力很强,纯自托管可能更简单,因为不用处理两套系统的衔接。
6. 常见问题与排查技巧实录
6.1 部署后接口超时怎么排查
接口超时是最常见的问题。排查思路按下面的顺序来:
先看是哪个节点慢。用 LangSmith 或者日志,找到耗时最长的节点。如果是 LLM 调用慢,考虑换更快的模型或者加缓存;如果是工具调用慢,考虑给工具调用加超时。
再看是不是并发问题。如果并发量上来之后才超时,可能是连接池不够或者数据库锁竞争。检查 Checkpointer 的连接池配置和数据库的并发能力。
最后看网络。如果节点本身不慢,但整体超时,可能是网络问题。检查服务到数据库、到 LLM API 的网络延迟。
下面是一个常见问题速查表:
| 现象 | 可能原因 | 排查方法 | 解决方案 |
|---|---|---|---|
| 接口超时 | 某节点执行慢 | 看 LangSmith 轨迹 | 优化慢节点或加超时 |
| 接口超时 | 连接池不够 | 看数据库连接数 | 增大连接池 |
| 状态丢失 | Checkpointer 配置错误 | 检查 thread_id 是否一致 | 修正 thread_id 映射 |
| 状态丢失 | 用了 MemorySaver | 检查 Checkpointer 类型 | 换成持久化 Checkpointer |
| 并发报错 | 数据库锁竞争 | 看数据库慢查询日志 | 优化写入或换数据库 |
| 流式输出中断 | 节点报错 | 看服务端日志 | 在流开始前做校验 |
| 部署后报错 | 环境变量缺失 | 对比本地和线上环境变量 | 补齐环境变量 |
| 部署后报错 | 依赖版本不一致 | 对比 requirements.txt | 锁定依赖版本 |
6.2 状态丢失的几种典型情况
状态丢失是自托管路径下最常见的问题之一。几种典型情况:
thread_id 不一致。每次请求传入的thread_id不一样,导致状态存在不同的地方。检查接口层是否正确传递了session_id。
Checkpointer 没初始化。PostgresSaver需要先调setup()建表,如果没建表,状态写入会失败。检查部署脚本里有没有执行setup()。
用了 MemorySaver。开发的时候用MemorySaver,部署的时候忘了换。检查生产环境的 Checkpointer 配置。
状态被清理。有些 Checkpointer 有 TTL 设置,状态过期后被清理。检查 TTL 配置是否符合业务需求。
6.3 性能优化的几个实操技巧
节点级别的缓存。如果某个节点的输入相同、输出也相同,可以加缓存。比如一些固定的工具调用或者 LLM 调用,可以用输入做 key 缓存结果。
并行执行。LangGraph 支持并行执行没有依赖关系的节点。如果图里有多个可以并行的分支,用并行执行能显著降低总耗时。
流式输出。对于 LLM 调用,流式输出能显著提升用户体验。用户不用等整个响应生成完,可以边生成边看。
连接池复用。数据库连接、HTTP 连接都要用连接池,避免每次请求都新建连接。连接池的大小要根据并发量来调,太小会排队,太大会浪费资源。
提示:性能优化之前先做 profiling,找到真正的瓶颈。我见过有人花了很多时间优化 LLM 调用,结果发现瓶颈在数据库写入。没有数据支撑的优化都是瞎猜。
6.4 部署上线的检查清单
上线之前,我一般会过一遍下面的清单:
- Checkpointer 是否用了持久化方案,
setup()是否执行过 - 环境变量是否都配置好了,特别是 API key 和数据库连接串
- 日志是否配置了结构化输出,日志级别是否合适
- 是否配置了健康检查接口
- 是否配置了超时和重试
- 是否配置了并发限制和限流
- LangSmith 追踪是否开启
- 数据库连接池大小是否合理
- 是否做了压力测试,知道服务的承载上限
- 是否有回滚方案
这个清单看起来简单,但每一条都是踩过坑之后总结出来的。特别是环境变量和 Checkpointer 这两项,出问题的概率最高。
7. 我个人在实际操作中的几点体会
三条路径我都实际用过,说几点个人体会。
自托管路径最灵活,但工作量也最大。如果你的团队有后端能力,自托管是长期来看最可控的方案。但要做好心理准备,运维的事情比想象中多,特别是状态管理和可观测性这两块。
托管路径最省心,但代价是数据要出你的边界,而且成本结构是持续性的。如果你的业务对数据可控性要求不高,托管方案能让你把精力集中在图逻辑本身。
混合模式听起来很美,但两套系统的衔接会带来额外的复杂度。如果团队规模不大,我一般建议要么纯自托管、要么纯托管,混合模式适合有明确边界划分的场景。
最后分享一个小技巧:不管选哪条路径,都先在本地用langgraph dev把图跑通,确认逻辑没问题再考虑部署。部署阶段的问题,大部分是环境和配置问题,图逻辑本身的问题在本地就能发现。把图逻辑和环境配置分开排查,能省很多时间。