LangGraph 部署实战:从脚本到 FastAPI、LangServe 与 Docker 容器化

发布时间:2026/10/27 3:32:13
LangGraph 部署实战:从脚本到 FastAPI、LangServe 与 Docker 容器化

1. 从脚本到服务,为什么部署这一步总让人卡壳

LangGraph 这个把状态机思路引入大模型应用编排的框架,最近一年在圈子里讨论度一直不低。它把多轮对话、工具调用、条件分支、循环重试这些原本写成一坨 if-else 的逻辑,抽象成节点和边组成的图,代码可读性和可维护性确实上了一个台阶。但真正把项目从本地脚本推到线上服务,很多人会在这里卡住:本地python main.py跑得好好的,一换到服务化部署就各种问题,端口冲突、依赖缺失、日志丢失、容器里连不上数据库,一个接一个。

这篇内容就围绕 LangGraph 的三条典型部署路径展开,从最轻量的脚本直跑,到 FastAPI 包一层 HTTP 接口,再到 LangServe 快速暴露服务,最后用 Docker 把整个环境固化下来。我会把每条路径的适用场景、核心代码、参数选择理由、踩过的坑都摊开讲。如果你手头正好有一个 LangGraph 项目,纠结到底该用哪种方式上线,或者你刚接触 FastAPI、Docker 这些配套工具,想找一个能直接抄作业的完整方案,那这篇应该能帮你省下不少试错时间。

需要先说明一点:部署路径没有绝对优劣,只有匹配不匹配。一个内部跑批的脚本,硬上 Docker 加 K8s 就是过度设计;一个要给前端调用的对话服务,还靠python xxx.py裸跑就是给自己埋雷。下面三条路径,我会按复杂度从低到高讲,你可以对照自己的场景直接选。

2. 三条部署路径的整体设计与选型逻辑

2.1 先搞清楚三条路径各自解决什么问题

在动手之前,得先把这三条路径的定位理清楚,不然很容易选错方向。

第一条是脚本直跑。就是最原始的方式,python your_graph.py直接执行,LangGraph 的图在进程内构建、编译、调用。它适合什么?适合本地调试、一次性任务、定时批处理,或者你只是想把图跑通验证逻辑。它的优点是零额外依赖、启动快、改代码即时生效;缺点是没法对外提供稳定接口,进程挂了就挂了,也没法多并发。

第二条是FastAPI 包一层。LangGraph 本身是个库,不是服务。你要让它对外提供 HTTP 接口,就得有个 Web 框架来承接请求。FastAPI 是当前 Python 生态里做这件事最顺手的选择之一,异步支持好、类型校验强、自动生成接口文档。这一层做的事情是:接收请求参数,调用编译好的图,把结果按约定格式返回。适合需要给前端、给其他服务调用的场景。

第三条是LangServe 快速暴露。LangServe 是 LangChain 官方出的部署工具,专门用来把 LangChain/LangGraph 的 Runnable 快速变成 REST API。它帮你把路由、输入输出 schema、流式输出这些样板代码都封装好了,几行代码就能起一个服务。适合你想快速验证、做 Demo、或者团队内部临时对接的场景。但它的定制能力相对弱一些,复杂鉴权、特殊中间件还是得自己补。

至于Docker,它不是第四条路径,而是可以叠加在上面任意一条之上的环境固化手段。它的价值在于:把 Python 版本、依赖包、系统库、环境变量全部锁死在一个镜像里,换台机器照样跑。生产环境基本绕不开它。

2.2 选型时我实际会看的几个维度

选哪条路,我一般看这几个维度,你可以对照自己的情况打个分:

维度脚本直跑FastAPILangServe
对外提供 HTTP 接口不支持支持支持
定制鉴权/中间件无灵活有限
开发速度最快中等快
流式输出支持手动手动实现内置
适合生产否是视情况
学习成本低中低

我的经验是:验证阶段用脚本,对接阶段用 LangServe,正式上线用 FastAPI。Docker 则是在确定要长期运行之后再加,不要一上来就折腾容器,那会让你在还没跑通业务逻辑的时候就陷进环境问题里。

2.3 一个容易被忽略的前提:图要先编译好

不管走哪条路径,有个共同前提:LangGraph 的图要提前compile()好,并且尽量做成模块级单例。我见过不少人在每个请求里重新build_graph()再compile(),结果 QPS 一上来 CPU 直接飙满。图的构建是有开销的,节点、边、checkpointer 的初始化都不便宜。正确做法是在模块加载时编译一次,请求进来直接graph.invoke()或graph.ainvoke()。

# graph.py from langgraph.graph import StateGraph, END from typing import TypedDict class State(TypedDict): messages: list def call_model(state: State): # 你的模型调用逻辑 return {"messages": state["messages"] + ["reply"]} builder = StateGraph(State) builder.add_node("model", call_model) builder.set_entry_point("model") builder.add_edge("model", END) # 模块级编译,全局复用 graph = builder.compile()

这个graph对象就是后面三条路径都要用的核心。把它单独放一个模块,FastAPI 和 LangServe 都从这里 import,避免重复构建。

3. 路径一:脚本直跑,最快验证但别拿去上线

3.1 最小可运行脚本长什么样

脚本直跑的核心就是invoke。假设你已经有了上面那个graph,一个最小脚本是这样:

# run_script.py from graph import graph def main(): result = graph.invoke({"messages": ["你好"]}) print(result) if __name__ == "__main__": main()

就这么简单。invoke是同步调用,传进去一个符合 State 结构的字典,返回执行完的最终状态。如果你要流式看每个节点的输出,用stream:

for chunk in graph.stream({"messages": ["你好"]}): print(chunk)

stream会按节点逐步 yield 中间结果,调试的时候特别有用,能清楚看到图走到哪一步、每个节点返回了什么。

3.2 脚本直跑适合哪些真实场景

别小看脚本直跑,它在这些场景里其实是最优解:

  • 本地开发调试:改一行代码直接重跑,没有服务重启、没有容器构建的等待。
  • 定时批处理:比如每天凌晨跑一批数据清洗、报告生成,用 cron 或任务计划程序调起脚本就行,不需要常驻服务。
  • 一次性数据迁移:把历史对话按新逻辑重新过一遍图,跑完就完事。
  • CI 里的冒烟测试:在流水线里跑一个最小图,验证依赖和逻辑没坏。

我自己的习惯是,任何 LangGraph 项目都先写一个run_script.py,把核心链路跑通、结果符合预期之后,再考虑要不要包服务。这一步能帮你把业务逻辑的 bug 和环境问题分开,不然混在一起排查会很痛苦。

3.3 脚本直跑的坑和注意事项

第一个坑是相对路径。脚本里如果用了open("config.yaml")这种相对路径,换个目录执行就找不到文件。稳妥做法是用pathlib基于__file__拼绝对路径:

from pathlib import Path BASE_DIR = Path(__file__).resolve().parent config_path = BASE_DIR / "config.yaml"

第二个坑是环境变量。脚本直跑时环境变量来自当前 shell,很容易出现"我本地能跑,换台机器就报 key 缺失"。建议在脚本入口显式加载.env,用python-dotenv一行搞定:

from dotenv import load_dotenv load_dotenv()

第三个坑是异常吞掉。批处理脚本里如果某个节点抛异常,整个脚本挂掉,前面跑的结果可能没保存。建议在循环里包 try-except,记录失败项,跑完统一看日志。这个习惯在数据量大的时候能救命。

提示:脚本直跑永远不要用来对外提供服务。它没有并发控制、没有健康检查、没有优雅退出,进程一挂调用方直接超时。验证完就赶紧往服务化走。

4. 路径二:FastAPI 包一层,正式上线的常规选择

4.1 为什么是 FastAPI 而不是 Flask

很多人会问,Flask 也能做接口,为什么推荐 FastAPI。核心原因有三个:

第一,异步原生支持。LangGraph 的ainvoke、astream都是异步的,FastAPI 的 async 路由能直接 await,不用起线程池。Flask 是同步框架,要在里面跑异步得额外折腾,性能和代码整洁度都吃亏。

第二,类型校验和文档。FastAPI 基于 Pydantic,请求体定义成模型,字段类型、必填、默认值一目了然,还能自动生成 Swagger 文档。对接前端的时候,把/docs甩过去,对方自己看接口,省掉大量沟通。

第三,依赖注入。鉴权、数据库连接、配置这些横切关注点,用Depends组织得很干净,测试时也容易替换。

4.2 一个能直接用的项目目录结构

FastAPI 项目最怕的就是所有代码堆在一个main.py里。我常用的结构是这样:

project/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 实例和路由注册 │ ├── graph.py # LangGraph 图定义和编译 │ ├── schemas.py # Pydantic 请求/响应模型 │ ├── routers/ │ │ └── chat.py # 对话相关路由 │ └── config.py # 配置加载 ├── tests/ ├── .env ├── requirements.txt └── Dockerfile

这个结构的好处是职责清晰:graph.py只管图,schemas.py只管数据契约,routers只管 HTTP 层。后面要加新接口,新建一个 router 文件注册进去就行,不会动到核心逻辑。

4.3 请求响应模型怎么定义

schemas.py里定义输入输出,这是接口的契约:

from pydantic import BaseModel, Field class ChatRequest(BaseModel): message: str = Field(..., min_length=1, max_length=4000) session_id: str = Field(default="default") class ChatResponse(BaseModel): reply: str session_id: str

Field里的约束不是摆设。min_length=1挡住空消息,max_length=4000防止有人塞超长文本把你的模型调用打爆。这些校验在请求进入业务逻辑之前就完成了,省得你在图里面还要判断。

4.4 路由里怎么调用图

routers/chat.py是核心:

from fastapi import APIRouter, HTTPException from app.graph import graph from app.schemas import ChatRequest, ChatResponse router = APIRouter(prefix="/chat", tags=["chat"]) @router.post("/", response_model=ChatResponse) async def chat(req: ChatRequest): try: result = await graph.ainvoke( {"messages": [req.message]}, config={"configurable": {"thread_id": req.session_id}}, ) return ChatResponse( reply=result["messages"][-1], session_id=req.session_id, ) except Exception as e: raise HTTPException(status_code=500, detail=str(e))

这里有几个关键点。用ainvoke而不是invoke,因为路由是 async 的,同步调用会阻塞事件循环。config里的thread_id是给 checkpointer 用的,同一个 session_id 的对话能共享历史状态,这是 LangGraph 做多轮对话的关键。异常统一转成 HTTP 500,别把原始堆栈直接吐给调用方,那既不安全也不友好。

4.5 启动配置和 uvicorn 参数

main.py里创建应用并挂载路由:

from fastapi import FastAPI from app.routers import chat app = FastAPI(title="LangGraph Service", version="1.0") app.include_router(chat.router) @app.get("/health") async def health(): return {"status": "ok"}

启动用 uvicorn:

uvicorn app.main:app --host 0.0.0.0 --port 8000 --workers 2

--host 0.0.0.0是为了容器内能被外部访问,本地调试可以改127.0.0.1。--workers是进程数,一般设成 CPU 核数或核数的一半。但要注意,如果图里用了内存态的 checkpointer,多 worker 之间状态不共享,会出问题。这种情况要么用单 worker,要么把 checkpointer 换成 Redis 之类的共享存储。

4.6 uvicorn 日志丢失问题怎么解

热词里有个"uvicorn fastapi 日志丢失问题",这个坑我踩过。现象是:自己用logging打的日志,在 uvicorn 启动后看不到,或者格式乱掉。原因是 uvicorn 自己配置了 logging,会覆盖你之前的配置。

解决办法是在应用启动时重新配置 logging,并且禁用 uvicorn 的默认配置:

import logging.config LOGGING_CONFIG = { "version": 1, "disable_existing_loggers": False, "formatters": { "default": { "format": "%(asctime)s [%(levelname)s] %(name)s: %(message)s", }, }, "handlers": { "console": { "class": "logging.StreamHandler", "formatter": "default", }, }, "root": {"handlers": ["console"], "level": "INFO"}, } logging.config.dictConfig(LOGGING_CONFIG)

然后在启动命令里加--log-config指向一个配置文件,或者干脆用--no-access-log关掉访问日志,自己接管。关键是disable_existing_loggers设成False,不然你模块里已经创建的 logger 会被禁用掉。

5. 路径三:LangServe 快速暴露,Demo 和内部对接的利器

5.1 LangServe 到底帮你做了什么

LangServe 的定位很明确:把 LangChain 的 Runnable(LangGraph 编译后的图也是 Runnable)快速变成 REST API。你不需要写路由、不需要定义 schema、不需要处理流式输出,它全帮你包好了。

一个最小的 LangServe 服务:

from fastapi import FastAPI from langserve import add_routes from graph import graph app = FastAPI(title="LangGraph LangServe") add_routes(app, graph, path="/graph")

就这几行,服务就起来了。访问/graph/playground还能得到一个交互式调试页面,直接在浏览器里输入测试,特别适合做 Demo 给非技术同事看。

5.2 LangServe 自动生成的接口有哪些

add_routes挂载之后,会自动生成一组标准接口:

路径方法作用
/graph/invokePOST同步调用,返回完整结果
/graph/batchPOST批量调用
/graph/streamPOST流式返回
/graph/stream_logPOST流式返回中间步骤
/graph/playgroundGET交互式调试页
/graph/input_schemaGET输入 schema

这套接口的好处是标准化。前端对接的时候不用问你要字段名,直接看 schema 就行。流式接口也省了自己实现 SSE 的功夫。

5.3 LangServe 的定制和局限

LangServe 不是不能定制,add_routes支持传per_req_config_modifier来做请求级配置,也能加dependencies做鉴权:

from fastapi import Depends, HTTPException, Header async def verify_token(x_token: str = Header(...)): if x_token != "your-secret": raise HTTPException(status_code=403, detail="Forbidden") add_routes( app, graph, path="/graph", dependencies=[Depends(verify_token)], )

但它的局限也明显:路由结构是固定的,你想加个自定义的预处理、后处理逻辑,就得绕。复杂的中间件、特殊的响应格式、多图组合,LangServe 处理起来会别扭。所以我的判断是:内部工具、Demo、快速验证用 LangServe;对外正式服务、有复杂定制需求用 FastAPI 自己写。两者也可以混用,LangServe 挂一个调试路径,FastAPI 写正式接口。

5.4 版本兼容这个坑要提前防

LangServe 和 LangChain、LangGraph 的版本耦合比较紧,升级其中一个经常导致另一个报错。我遇到过add_routes突然不认graph对象的情况,排查半天发现是 LangGraph 升了个小版本,Runnable 接口变了。

建议在requirements.txt里把版本钉死:

langgraph==0.2.x langchain-core==0.3.x langserve==0.3.x fastapi==0.115.x uvicorn==0.32.x

具体版本号以你验证通过的组合为准,关键是不要用>=或浮动版本。生产环境里,依赖版本漂移是很多"昨天还好好的今天挂了"的元凶。

6. 用 Docker 把环境固化,让部署不再看机器脸色

6.1 为什么生产环境基本绕不开 Docker

脚本直跑和 FastAPI 本地跑,都有一个共同问题:依赖装在宿主机上,换台机器就得重装一遍,还经常因为系统库版本、Python 版本不一致出问题。Docker 把应用和它的运行环境一起打包成镜像,镜像里有什么,跑起来就是什么,跟宿主机无关。

对 LangGraph 项目来说,Docker 还额外解决一个问题:模型调用的网络配置、环境变量、密钥管理,都能通过容器编排统一处理,不用每台机器手动配。

6.2 一个精简的 Dockerfile 怎么写

FROM python:3.11-slim WORKDIR /app # 先拷依赖文件,利用镜像层缓存 COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt # 再拷代码 COPY app/ ./app/ EXPOSE 8000 CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

这里有个关键技巧:先 COPY requirements.txt 再 pip install,最后 COPY 代码。因为 Docker 是分层构建,只要 requirements.txt 没变,改代码重新构建时会复用依赖层,构建速度从几分钟降到几秒。很多人把COPY . .放最前面,结果每次改一行代码都要重装所有依赖,白白浪费时间。

6.3 镜像瘦身和构建加速

python:3.11-slim比完整版小很多,够用。如果还要更小,可以用alpine,但 Alpine 用的是 musl libc,某些带 C 扩展的包(比如 numpy、cv2)编译起来会踩坑,需要额外装构建工具。我的建议是:除非镜像大小是硬指标,否则用 slim 就够了,省下的那点体积不值得折腾编译问题。

构建时加.dockerignore,把不必要的东西排除掉:

__pycache__ *.pyc .env .git tests/ *.md

.env一定要排除,密钥不能打进镜像。运行时通过-e或--env-file注入。

6.4 docker-compose 编排多服务

实际项目里,LangGraph 服务往往还要连数据库、缓存。用 docker-compose 把这些串起来:

version: "3.9" services: app: build: . ports: - "8000:8000" env_file: - .env depends_on: - redis redis: image: redis:7-alpine ports: - "6379:6379"

depends_on只保证启动顺序,不保证 redis 已经 ready。稳妥做法是在应用里做重试连接,或者用 healthcheck 配合condition: service_healthy。这个细节不注意的话,容器起来时 redis 还没就绪,应用直接崩。

6.5 Docker 常见报错速查

报错原因解决
permission denied while trying to connect to the Docker API当前用户不在 docker 组把用户加入 docker 组后重新登录
容器内连不上宿主机的 MySQL用了 127.0.0.1改用宿主机内网 IP 或 host 网络模式
pip install 超时网络问题配置镜像源或增加超时重试
镜像构建后代码是旧的层缓存调整 COPY 顺序或加 --no-cache
容器启动即退出CMD 命令错误docker logs 看具体报错

注意:容器里的localhost指的是容器自己,不是宿主机。要访问宿主机的服务,得用宿主机的内网地址,或者用host.docker.internal(部分平台支持)。这个坑新手几乎必踩。

7. 三条路径的实操对比与我的选择建议

7.1 从零到上线的完整流程

我把三条路径串成一个完整流程,你可以照着走:

  1. 写graph.py,把图定义好并模块级编译。
  2. 写run_script.py,本地验证逻辑,确认输入输出符合预期。
  3. 决定服务化方式:内部 Demo 用 LangServe,正式服务用 FastAPI。
  4. 写schemas.py和路由,定义接口契约。
  5. 本地 uvicorn 跑通,用 curl 或 Swagger 测接口。
  6. 写 Dockerfile 和 docker-compose,把环境固化。
  7. 构建镜像、起容器,验证容器内服务正常。
  8. 加日志、健康检查、异常处理,准备上线。

每一步都验证通过再进下一步,不要跳步。我见过太多人直接写 Dockerfile,结果容器里报错,分不清是代码问题还是环境问题,排查成本翻倍。

7.2 性能相关的几个实测经验

关于并发,FastAPI 用 async 路由 +ainvoke,单 worker 在 IO 密集场景(等模型返回)下能扛住不错的并发。但如果图里有 CPU 密集的节点(比如本地跑 embedding),会阻塞事件循环,这时候要么把这类节点丢到线程池,要么多开 worker。

关于 worker 数量,不是越多越好。每个 worker 是独立进程,内存翻倍。如果图里加载了大模型或大索引,worker 开多了内存直接爆。我的经验是:内存允许的前提下,worker 数设为 CPU 核数;内存紧张就设 1-2 个,靠异步扛并发。

关于超时,模型调用可能很慢,接口层要设合理超时。uvicorn 默认没有请求超时,但反向代理(如 Nginx)通常有 60 秒默认值,长任务会被切断。要么调大代理超时,要么把长任务改成异步任务 + 轮询结果。

7.3 安全方面不能省的两件事

第一,密钥绝不进代码和镜像。API key、数据库密码全部走环境变量,.env加进.gitignore和.dockerignore。我见过把 key 硬编码提交到仓库的,后果很严重。

第二,接口要有基本防护。至少加个 token 鉴权,别让接口裸奔。请求体大小限制、频率限制也建议加上,防止被刷。FastAPI 可以用中间件做,LangServe 用dependencies做。

8. 常见问题排查实录

8.1 服务起来了但调用报 422

422 是 FastAPI 的请求校验失败。先看/docs里的 schema,确认字段名和类型对得上。常见原因是前端传的 JSON 字段名和 Pydantic 模型不一致,或者该传字符串传了数字。把detail字段打出来,里面会明确告诉你是哪个字段的问题。

8.2 图执行到一半卡住

如果图里有循环边,且没有正确的终止条件,会无限循环。LangGraph 支持设置recursion_limit,在 config 里传:

config = {"recursion_limit": 25}

超过限制会抛异常,而不是一直转。调试阶段把这个值设小一点,能快速发现循环逻辑的问题。

8.3 多轮对话状态丢失

大概率是 checkpointer 没配好,或者thread_id每次请求都变了。检查两点:一是编译图时是否传了 checkpointer,二是请求里的 session_id 是否稳定。用内存 checkpointer 时,多 worker 之间不共享,也会导致状态丢失,这种情况要换共享存储。

8.4 容器内时区不对

日志时间戳差 8 小时,是容器默认 UTC 时区。在 Dockerfile 里设置:

ENV TZ=Asia/Shanghai

或者在 compose 里加环境变量。这个不影响功能,但排查问题时时间对不上会很误导。

8.5 依赖装了但 import 报错

常见于本地和容器环境不一致。本地用 conda 装了包,容器里用 pip 装,包名或版本不同。解决办法是统一用requirements.txt管理,本地也用 venv + pip 安装,保证两边一致。pip freeze > requirements.txt能导出当前环境的精确版本。

9. 我在这几个项目里攒下的实操心得

先说一个最实在的:别在没跑通业务逻辑之前碰 Docker。我早期犯过这个错,图还没调通就急着写 Dockerfile,结果容器里报错,我花了两天在排查环境,最后发现是图本身的逻辑 bug。正确的顺序永远是先本地脚本跑通,再服务化,最后容器化。

第二个心得是日志要打够,但要打对地方。LangGraph 执行过程中,每个节点的输入输出、耗时、异常,都值得记录。但别用print,用logging分级别。DEBUG 级别记详细状态,INFO 记关键节点,ERROR 记异常。生产环境把级别调到 INFO,需要排查时临时调 DEBUG,不用改代码重启。

第三个是接口设计要留余地。请求体里除了必填字段,尽量给可选字段留默认值。比如session_id默认"default",stream默认False。这样前端可以先传最少参数跑通,后面再逐步加功能,不会因为接口变更导致联调反复。

第四个是版本锁定比什么都重要。LangChain 生态迭代快,小版本之间都可能有 breaking change。requirements.txt里所有包都钉死版本,Docker 镜像打好 tag,别用latest。我吃过一次亏,线上服务某天突然挂了,查半天发现是构建时拉了新版本依赖,接口变了。从那以后所有生产镜像都用固定 tag。

最后一个,健康检查接口别省。/health返回 200,容器编排和负载均衡靠它判断实例是否可用。没有健康检查,实例挂了流量还在往上打,用户看到的就是超时。这个接口实现成本极低,但价值很高。

这套东西跑下来,一个 LangGraph 项目从脚本到线上服务,快的话半天能搞定,慢的话主要时间花在排查环境和依赖上。把上面这些坑提前避开,能省下大量来回折腾的时间。