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 结构的字典返回执行完的最终状态。如果你要流式看每个节点的输出用streamfor 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_length1, max_length4000) session_id: str Field(defaultdefault) class ChatResponse(BaseModel): reply: str session_id: strField里的约束不是摆设。min_length1挡住空消息max_length4000防止有人塞超长文本把你的模型调用打爆。这些校验在请求进入业务逻辑之前就完成了省得你在图里面还要判断。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_modelChatResponse) async def chat(req: ChatRequest): try: result await graph.ainvoke( {messages: [req.message]}, config{configurable: {thread_id: req.session_id}}, ) return ChatResponse( replyresult[messages][-1], session_idreq.session_id, ) except Exception as e: raise HTTPException(status_code500, detailstr(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(titleLangGraph Service, version1.0) app.include_router(chat.router) app.get(/health) async def health(): return {status: ok}启动用 uvicornuvicorn 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 的 RunnableLangGraph 编译后的图也是 Runnable快速变成 REST API。你不需要写路由、不需要定义 schema、不需要处理流式输出它全帮你包好了。一个最小的 LangServe 服务from fastapi import FastAPI from langserve import add_routes from graph import graph app FastAPI(titleLangGraph 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_code403, detailForbidden) 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里把版本钉死langgraph0.2.x langchain-core0.3.x langserve0.3.x fastapi0.115.x uvicorn0.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:6379depends_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 从零到上线的完整流程我把三条路径串成一个完整流程你可以照着走写graph.py把图定义好并模块级编译。写run_script.py本地验证逻辑确认输入输出符合预期。决定服务化方式内部 Demo 用 LangServe正式服务用 FastAPI。写schemas.py和路由定义接口契约。本地 uvicorn 跑通用 curl 或 Swagger 测接口。写 Dockerfile 和 docker-compose把环境固化。构建镜像、起容器验证容器内服务正常。加日志、健康检查、异常处理准备上线。每一步都验证通过再进下一步不要跳步。我见过太多人直接写 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 服务起来了但调用报 422422 是 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 TZAsia/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默认defaultstream默认False。这样前端可以先传最少参数跑通后面再逐步加功能不会因为接口变更导致联调反复。第四个是版本锁定比什么都重要。LangChain 生态迭代快小版本之间都可能有 breaking change。requirements.txt里所有包都钉死版本Docker 镜像打好 tag别用latest。我吃过一次亏线上服务某天突然挂了查半天发现是构建时拉了新版本依赖接口变了。从那以后所有生产镜像都用固定 tag。最后一个健康检查接口别省。/health返回 200容器编排和负载均衡靠它判断实例是否可用。没有健康检查实例挂了流量还在往上打用户看到的就是超时。这个接口实现成本极低但价值很高。这套东西跑下来一个 LangGraph 项目从脚本到线上服务快的话半天能搞定慢的话主要时间花在排查环境和依赖上。把上面这些坑提前避开能省下大量来回折腾的时间。