☰
LangGraph部署实战:从脚本内嵌到独立服务化的三条路径
2026/10/27 3:32:05 网站建设 项目流程

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.sh

graph.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)]}, config=config, ) return {"reply": result["messages"][-1].content}

3.2 参数选择与背后的考量

这里有几个参数值得展开说。thread_id是 LangGraph 用来区分会话的标识,配合 checkpointer 使用时,同一个 thread_id 的多次调用会共享状态历史。路径一里如果没配 checkpointer,thread_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 把图快速变成标准 API

4.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.sh

server.py是核心:

from fastapi import FastAPI from langserve import add_routes from app.graph import graph app = FastAPI(title="LangGraph Service") add_routes( app, graph, path="/chat", enable_feedback_endpoint=True, )

就这么几行,你的图就有了完整的 API。访问/chat/playground还能看到一个网页版的调试界面,可以直接输入消息测试。这个 playground 在给非技术同事演示时特别好用,不用写任何前端代码。

4.3 LangServe 的输入输出格式与自定义

LangServe 默认的输入输出格式是 LangChain 的消息格式。如果你想让接口更贴近业务,可以传入自定义的输入输出 schema:

from pydantic import BaseModel class ChatInput(BaseModel): message: str class ChatOutput(BaseModel): reply: str add_routes( app, graph.with_types(input_type=ChatInput, output_type=ChatOutput), path="/chat", )

with_types这个方法很实用,它让你在不改图逻辑的前提下,把接口的输入输出契约固定下来。这样前端对接的时候就有明确的字段定义,不用去猜 LangChain 内部的消息结构。

4.4 路径二的坑与注意事项

LangServe 虽然省事,但有几个坑我踩过。第一个是版本兼容,LangServe 和 LangChain、LangGraph 的版本耦合比较紧,升级其中一个经常要连带升级其他的,否则会出现序列化失败或者路由注册异常。我的做法是在requirements.txt里把这三个库的版本都锁死,升级时一起升。

第二个是流式输出的配置。LangServe 的/stream端点默认走的是stream_mode="values",如果你想要更细粒度的 token 级流式,需要在图里配置stream_mode="messages",并且确保你的 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(lifespan=lifespan)

把图对象挂在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 --from=builder /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_URL=postgresql://user:pass@db:5432/langgraph depends_on: - db db: image: postgres:16 environment: - POSTGRES_USER=user - POSTGRES_PASSWORD=pass - POSTGRES_DB=langgraph 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 的 path

6.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 左右,部署速度快了很多。这个优化对路径三尤其重要,因为多实例部署时镜像拉取时间是实打实的成本。

这三条路径没有绝对的好坏,关键是匹配你当前的阶段和需求。我个人的习惯是,任何新项目都先用路径一把逻辑跑通,需要给别人用时切路径二,等真正有并发和持久化压力了再上路径三。这样每一步的投入都和收益匹配,不会在还没验证业务的时候就陷进部署的复杂度里。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询