LangGraph部署实战：从脚本内嵌到独立服务化的三条路径

1. 从脚本到服务为什么部署这一步最容易卡住人LangGraph 这个把状态机思路引入 LLM 应用编排的框架最近一年在 Python 圈子里热度一直往上走。很多人跟着官方示例跑通了本地脚本看着终端里一行行节点执行日志觉得“这东西我算是会了”。结果一到要把它变成别人能访问的服务问题就全冒出来了脚本里graph.invoke()一调就完事可一旦要处理并发请求、要保留会话状态、要暴露 HTTP 接口整个结构就得重新想。我自己第一次把 LangGraph 从脚本搬到线上踩的坑比写业务逻辑本身还多。最开始图省事直接写了个 FastAPI 的 POST 接口在函数里 new 一个 graph 然后 invoke本地测没问题一上量就发现每次请求都在重建编译图内存蹭蹭涨响应时间也不稳定。后来才慢慢摸清楚LangGraph 的部署其实有三条比较清晰的路径分别对应不同的场景和复杂度选错了不是不能用而是会在后期维护上付出成倍的代价。这篇内容就是把我这段时间的实践整理出来围绕LangGraph Python FastAPI LangServe Docker这套组合把三条部署路径讲透。不管你是刚跑通第一个 graph 的新手还是已经在做多轮对话服务的老手都能从中找到适合自己当前阶段的那条路。我会把每条路径的适用场景、核心代码结构、参数选择理由、以及实际部署时那些文档里不会写的坑都摊开来讲。先说清楚三条路径分别是什么。第一条是纯脚本内嵌把 graph 直接跑在 FastAPI 的请求处理函数里适合原型验证和内部工具第二条是LangServe 托管用 LangChain 官方提供的 LangServe 把 graph 快速包装成标准 API适合快速对外提供能力第三条是独立服务化把 graph 编译、状态管理、持久化全部拆成独立模块用 Docker 打包成可水平扩展的服务适合生产环境。这三条不是简单的“初级到高级”而是各有各的适用边界下面逐个拆。2. 三条部署路径的整体设计与选型逻辑2.1 先搞清楚 LangGraph 的运行时到底需要什么在讨论部署之前得先明白一个 LangGraph 应用在运行时到底依赖哪些东西。很多人部署出问题根源是对运行时结构没概念。一个典型的 LangGraph 应用包含这么几层状态定义StateGraph 的 schema、节点函数每个 node 的具体逻辑、边与条件路由决定执行流向、编译后的图对象compiled graph、以及运行时上下文checkpointer、config、thread_id 等。其中编译后的图对象是核心。graph.compile()返回的是一个可调用的 Runnable它本身是线程安全的可以被多个请求共享。这一点非常关键因为它直接决定了部署方式如果你在每次请求里都重新 compile那就是在浪费 CPU 和内存正确的做法是在应用启动时 compile 一次全局共享。另一个容易被忽略的是checkpointer。LangGraph 支持把每一步的状态持久化下来这样就能实现多轮对话、断点续跑、人工介入等功能。checkpointer 可以是内存版MemorySaver也可以是数据库版比如基于 SQLite、Postgres 的实现。内存版在单进程里够用但一旦你要多进程或者多容器部署就必须换成持久化版本否则每个进程的状态是割裂的。理解了这两点三条路径的差异就清楚了它们的本质区别在于图对象在哪里编译、状态存在哪里、以及请求怎么路由到图对象。2.2 三条路径的适用边界对比我把三条路径的关键维度整理成一张表方便你对照自己的场景快速判断。维度路径一脚本内嵌路径二LangServe 托管路径三独立服务化适用阶段原型验证、内部工具快速对外、Demo生产环境、多实例图编译时机请求内或模块级应用启动时应用启动时状态持久化内存为主内存或轻量持久化数据库持久化并发能力弱中强部署复杂度低低高可观测性靠日志内置部分需自行搭建扩展性差一般好这张表不是绝对的比如路径二在加了持久化之后也能撑一定并发但它的设计初衷就是快速暴露能力不是为高并发准备的。选路径的核心判断标准是你的服务是要给几个人用还是要给成百上千人用状态是可有可无还是业务核心。2.3 为什么我不建议一上来就上最重的方案见过不少团队项目刚起步就照着“生产级”的模板搭一套独立服务化架构结果业务逻辑还没跑通光在 Docker 编排、数据库连接、健康检查上就耗掉大半精力。我的建议是按需演进先用路径一把业务逻辑跑通确认 graph 的设计没问题需要对外提供接口时切到路径二快速验证等真正有并发和持久化需求了再迁移到路径三。这种演进式的好处是每一步的迁移成本都可控。路径一到路径二基本只是把 graph 的调用方式换一下路径二到路径三主要是把状态管理和部署配置重做。反过来如果一开始就上最重的方案后面发现业务逻辑要改改起来会非常痛苦因为你的业务代码和部署代码耦合太深了。3. 路径一脚本内嵌 FastAPI 的最小可用实现3.1 目录结构与核心代码路径一的目标是用最少的代码把 graph 跑起来并暴露成 HTTP 接口。我常用的目录结构是这样的project/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 入口 │ ├── graph.py # LangGraph 定义 │ └── schemas.py # 请求响应模型 ├── requirements.txt └── run.shgraph.py里定义状态和节点最后导出一个编译好的图对象from typing import TypedDict, Annotated from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages class State(TypedDict): messages: Annotated[list, add_messages] def chatbot(state: State): # 这里替换成你真实的 LLM 调用 return {messages: [(assistant, 收到你的消息了)]} def build_graph(): builder StateGraph(State) builder.add_node(chatbot, chatbot) builder.set_entry_point(chatbot) builder.add_edge(chatbot, END) return builder.compile() graph build_graph()注意这里graph是在模块级别编译的模块被导入时只编译一次之后所有请求共享这个对象。这是路径一里最重要的一个细节很多人写成在接口函数里build_graph()那就是每次请求都重新编译性能差得离谱。main.py里把它挂到 FastAPI 上from fastapi import FastAPI from pydantic import BaseModel from app.graph import graph app FastAPI() class ChatRequest(BaseModel): message: str thread_id: str default app.post(/chat) async def chat(req: ChatRequest): config {configurable: {thread_id: req.thread_id}} result graph.invoke( {messages: [(user, req.message)]}, configconfig, ) return {reply: result[messages][-1].content}3.2 参数选择与背后的考量这里有几个参数值得展开说。thread_id是 LangGraph 用来区分会话的标识配合 checkpointer 使用时同一个 thread_id 的多次调用会共享状态历史。路径一里如果没配 checkpointerthread_id 其实不起作用但提前把它放进接口设计里后面迁移到路径二、路径三时就不用改接口契约了。graph.invoke是同步调用在 FastAPI 的 async 函数里直接调用会阻塞事件循环。如果节点里有耗时的 LLM 调用建议用graph.ainvoke走异步或者把整个接口定义成普通 def 让 FastAPI 丢到线程池里跑。我实测下来节点里如果有网络 IO用ainvoke配合异步的 LLM 客户端吞吐能比同步版本高不少。注意路径一里如果用了MemorySaver做 checkpointer状态只存在当前进程内存里进程一重启就全没了。内部工具无所谓但别拿它当正式服务用。3.3 路径一的实操心得路径一最大的价值是快。我经常用它来做两件事一是验证 graph 的节点逻辑对不对二是给团队内部做个能点的 Demo。这两件事都不需要复杂的部署uvicorn app.main:app --reload一跑就能用。但路径一有两个硬伤。第一是没有内置的可观测性graph 执行到哪一步、每个节点的耗时、状态怎么变的全靠自己打日志。第二是并发能力弱因为状态在内存里多开几个 worker 就会出现会话串台的问题。所以路径一的天花板很明显业务一旦要给别人用就得考虑往路径二走。4. 路径二用 LangServe 把图快速变成标准 API4.1 LangServe 解决了什么问题LangServe 是 LangChain 官方出的一个库作用是把任意 Runnable包括编译后的 LangGraph 图自动包装成一套符合 REST 规范的 API还自带一个可交互的调试页面。它的核心价值在于省掉了手写接口和序列化的工作。用路径一的时候请求体、响应体、错误处理、流式输出这些都得自己写。LangServe 把这些都标准化了它提供/invoke、/batch、/stream、/stream_log等标准端点请求和响应的格式也是固定的。你只需要把图对象交给它剩下的它来处理。4.2 最小实现与目录结构路径二的目录结构比路径一还简单project/ ├── app/ │ ├── __init__.py │ ├── graph.py │ └── server.py ├── requirements.txt └── run.shserver.py是核心from fastapi import FastAPI from langserve import add_routes from app.graph import graph app FastAPI(titleLangGraph Service) add_routes( app, graph, path/chat, enable_feedback_endpointTrue, )就这么几行你的图就有了完整的 API。访问/chat/playground还能看到一个网页版的调试界面可以直接输入消息测试。这个 playground 在给非技术同事演示时特别好用不用写任何前端代码。4.3 LangServe 的输入输出格式与自定义LangServe 默认的输入输出格式是 LangChain 的消息格式。如果你想让接口更贴近业务可以传入自定义的输入输出 schemafrom pydantic import BaseModel class ChatInput(BaseModel): message: str class ChatOutput(BaseModel): reply: str add_routes( app, graph.with_types(input_typeChatInput, output_typeChatOutput), path/chat, )with_types这个方法很实用它让你在不改图逻辑的前提下把接口的输入输出契约固定下来。这样前端对接的时候就有明确的字段定义不用去猜 LangChain 内部的消息结构。4.4 路径二的坑与注意事项LangServe 虽然省事但有几个坑我踩过。第一个是版本兼容LangServe 和 LangChain、LangGraph 的版本耦合比较紧升级其中一个经常要连带升级其他的否则会出现序列化失败或者路由注册异常。我的做法是在requirements.txt里把这三个库的版本都锁死升级时一起升。第二个是流式输出的配置。LangServe 的/stream端点默认走的是stream_modevalues如果你想要更细粒度的 token 级流式需要在图里配置stream_modemessages并且确保你的 LLM 客户端支持流式。这块配置不对的话前端会看到一大坨一次性返回体验很差。第三个是生产环境的性能。LangServe 底层还是 FastAPI单进程跑没问题但它的 playground 和部分调试端点在生产环境应该关掉。可以通过add_routes的参数控制或者干脆在生产配置里不注册这些路由。提示LangServe 适合快速验证和中小规模使用。如果你的服务要扛高并发或者需要复杂的鉴权、限流、审计路径二会显得力不从心这时候该考虑路径三了。5. 路径三独立服务化的完整落地5.1 为什么要拆成独立服务路径三的核心思路是把 LangGraph 应用当成一个正经的后端服务来对待。这意味着图在应用启动时编译一次并全局共享状态通过数据库持久化服务用 Docker 打包可以水平扩展多个实例有健康检查、日志、指标。什么时候需要走到这一步我的判断标准是三条并发量上来了单进程扛不住、状态不能丢会话历史是业务核心、要部署到多台机器需要无状态或共享状态。满足任意一条路径二就不够用了。5.2 项目目录结构设计路径三的目录结构要清晰地区分业务逻辑、状态管理、接口层和部署配置project/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 入口 │ ├── api/ │ │ ├── __init__.py │ │ └── routes.py # 路由定义 │ ├── core/ │ │ ├── __init__.py │ │ ├── config.py # 配置管理 │ │ └── logging.py # 日志配置 │ ├── graph/ │ │ ├── __init__.py │ │ ├── state.py # 状态定义 │ │ ├── nodes.py # 节点函数 │ │ └── builder.py # 图构建与编译 │ ├── services/ │ │ ├── __init__.py │ │ └── chat_service.py # 业务服务层 │ └── persistence/ │ ├── __init__.py │ └── checkpointer.py # 持久化配置 ├── tests/ ├── Dockerfile ├── docker-compose.yml ├── requirements.txt └── .env.example这个结构看起来比前两条路径复杂但每一层都有明确职责。graph/只管图的定义services/管业务编排api/管 HTTP 契约persistence/管状态存储。这样拆的好处是任何一层要改都不影响其他层测试也好写。5.3 状态持久化的关键配置路径三里最重要的一环是 checkpointer。我一般用 Postgres 作为持久化后端因为它在事务和并发上比较可靠。LangGraph 提供了PostgresSaver配置方式大致如下from langgraph.checkpoint.postgres import PostgresSaver from app.core.config import settings def get_checkpointer(): return PostgresSaver.from_conn_string(settings.DATABASE_URL)这里有个细节要注意PostgresSaver需要在数据库里建表第一次使用前要调用一次checkpointer.setup()。这个操作我一般放在应用启动的 lifespan 里确保表结构就绪后再开始接收请求。from contextlib import asynccontextmanager from fastapi import FastAPI asynccontextmanager async def lifespan(app: FastAPI): checkpointer get_checkpointer() checkpointer.setup() app.state.graph build_graph(checkpointer) yield app FastAPI(lifespanlifespan)把图对象挂在app.state上路由里通过request.app.state.graph取用。这样图在启动时编译一次所有请求共享同时 checkpointer 也是全局唯一的。5.4 Docker 打包与多实例部署Dockerfile 我一般写成多阶段构建减小镜像体积FROM python:3.11-slim AS builder WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir --user -r requirements.txt FROM python:3.11-slim WORKDIR /app COPY --frombuilder /root/.local /root/.local COPY app ./app ENV PATH/root/.local/bin:$PATH EXPOSE 8000 CMD [uvicorn, app.main:app, --host, 0.0.0.0, --port, 8000, --workers, 4]这里--workers 4是关键。因为状态已经持久化到 Postgres多个 worker 之间不会串台可以放心开多进程。如果部署到多台机器只要它们连的是同一个数据库状态就是共享的。docker-compose 里把应用和数据库编排在一起services: app: build: . ports: - 8000:8000 environment: - DATABASE_URLpostgresql://user:passdb:5432/langgraph depends_on: - db db: image: postgres:16 environment: - POSTGRES_USERuser - POSTGRES_PASSWORDpass - POSTGRES_DBlanggraph volumes: - pgdata:/var/lib/postgresql/data volumes: pgdata:5.5 路径三的实操心得路径三最容易被低估的是数据库连接池的配置。LangGraph 的 checkpointer 在高并发下会频繁读写数据库如果连接池太小请求会排队等连接表现为响应时间忽高忽低。我一般把连接池的min_size设成 worker 数量的两倍max_size设成 worker 数量的四倍具体数值还要根据实际压测调整。另一个坑是状态对象的序列化。LangGraph 会把状态序列化后存进数据库如果你的状态里放了不可序列化的对象比如数据库连接、文件句柄存的时候就会报错。我的经验是状态里只放纯数据需要外部资源的时候通过 config 或者依赖注入传进去不要塞进 state。6. 三条路径的迁移与常见问题排查6.1 从路径一迁移到路径二迁移的核心改动是把手写的接口换成add_routes。业务逻辑graph 定义基本不用动主要改的是main.py。需要注意的是路径一里你可能自己定义了请求响应模型迁移后要么用 LangServe 的默认格式要么用with_types指定。接口路径也会变从你自定义的/chat变成 LangServe 的/chat/invoke前端要同步改。6.2 从路径二迁移到路径三这个迁移改动最大因为要引入持久化和服务分层。我的建议是分步走先把 checkpointer 从内存换成 Postgres验证状态持久化没问题再把目录结构按服务化重构最后加 Docker 和编排配置。每一步都单独测试不要一次性全改完再测否则出问题很难定位是哪一步引入的。6.3 常见问题速查表问题现象可能原因排查方向响应时间随请求量线性上升每次请求重新编译图检查图是否模块级或启动时编译多轮对话状态丢失checkpointer 是内存版或未配置确认 checkpointer 类型和 thread_id多 worker 下会话串台状态存在进程内存换持久化 checkpointer流式输出变成一次性返回stream_mode 配置不对检查图和服务端的流式配置Docker 容器启动后立即退出依赖服务未就绪加 depends_on 和健康检查数据库连接超时连接池太小或数据库负载高调整连接池参数压测验证LangServe 路由 404版本不兼容或路径拼写错误锁定版本检查 add_routes 的 path6.4 几个文档里不会写的避坑技巧第一个技巧是给图执行加超时。LLM 调用有时候会卡住如果不设超时请求会一直挂着占资源。我一般在节点函数里用asyncio.wait_for包一层或者在服务层设置整体超时。第二个技巧是日志里带上 thread_id 和节点名。LangGraph 执行链路比较长出问题时光看报错很难定位是哪个节点、哪个会话出的问题。在日志里把这两个信息带上排查效率能提升一大截。第三个技巧是压测时关注数据库而不是应用。路径三的性能瓶颈往往不在 Python 进程而在数据库的读写上。压测的时候盯着数据库的 QPS 和连接数比盯着应用的 CPU 更有用。7. 我在这三条路径上踩过的真实坑说几个具体的。有一次用路径二部署本地跑得好好的上了服务器之后/stream端点一直返回空。查了半天发现是服务器上的反向代理把流式响应缓冲了需要在代理配置里关掉缓冲。这个问题跟 LangServe 本身没关系但如果不了解流式输出的原理很容易误以为是代码问题。还有一次用路径三多实例部署之后发现同一个用户的会话有时候能接上、有时候接不上。最后定位到是负载均衡没有做会话粘性而 checkpointer 虽然持久化了但不同实例读到的状态版本不一致。解决办法是确保所有实例连同一个数据库并且用乐观锁或者版本号来处理并发写。这个坑让我意识到持久化只是第一步并发控制才是多实例部署的真正难点。最后一个坑是关于 Docker 镜像的。我一开始用python:3.11完整镜像构建出来快 1G推送到镜像仓库很慢。后来换成python:3.11-slim加多阶段构建体积降到 200M 左右部署速度快了很多。这个优化对路径三尤其重要因为多实例部署时镜像拉取时间是实打实的成本。这三条路径没有绝对的好坏关键是匹配你当前的阶段和需求。我个人的习惯是任何新项目都先用路径一把逻辑跑通需要给别人用时切路径二等真正有并发和持久化压力了再上路径三。这样每一步的投入都和收益匹配不会在还没验证业务的时候就陷进部署的复杂度里。