1. 从本地脚本到 API 服务为什么这一步非走不可很多人写 LangChain 或 LangGraph 的脚本跑通一个链式调用之后就停下来了。终端里打印出结果心里觉得“这东西成了”。但真到要给别人用的时候问题就来了同事想试一下你得把代码发过去让他装 Python、配环境、填 Key折腾半天还不一定跑得起来前端想接个界面你总不能让人家直接调你的 Python 函数想统计一下每天被调用了多少次、哪个环节最耗时日志散落在控制台里根本没法看。这就是从“本地脚本”到“API 服务”这一步的核心驱动力把能力封装成标准接口让调用方不关心你内部用了什么框架、什么模型、什么提示词只关心输入和输出。这一步做完你的 LangGraph 工作流才算真正具备了被复用的可能。我自己的习惯是只要一个 LangGraph 工作流在本地跑通了三次以上就立刻着手把它包成 FastAPI 服务。原因很简单本地脚本的调试成本低但协作成本极高API 服务的初期搭建有一点门槛但一旦跑起来后面加功能、换模型、做监控都是顺水推舟的事。LangServe 是 LangChain 官方给出的一个快速方案它能把一个 Runnable 或 Chain 直接暴露成 HTTP 接口省掉不少样板代码。但如果你用的是 LangGraph或者需要更细粒度的控制比如自定义中间件、流式输出的分块逻辑、多路由组合那 FastAPI 手写反而是更稳的选择。这一篇要聊的就是这条路径上的三个台阶本地脚本怎么整理成可服务的结构、FastAPI 和 LangServe 怎么选、Docker 化之后生产环境该怎么摆。适合已经写过 LangChain/LangGraph 脚本、想把它变成真正能对外提供服务的读者。如果你还在纠结 LangChain 和 LangGraph 的区别那说明还没到这一步建议先把基础工作流跑顺。2. 方案选型LangServe、FastAPI 还是直接上 LangGraph Platform2.1 三个方案的本质差异先把这三个东西摆清楚很多人会把它们混在一起谈其实定位完全不同。LangServe是 LangChain 生态里的一个部署工具核心能力是把一个 Runnable 对象自动映射成 REST API。你写一个 chain调add_routes(app, chain, path/my-chain)它就帮你生成/invoke、/batch、/stream、/playground这几个端点。优点是快十分钟能出一个能用的接口缺点是灵活性差一旦你想在请求前后加自定义逻辑比如鉴权、限流、请求日志脱敏就得绕开它的默认行为反而更麻烦。FastAPI是一个通用的 Python Web 框架和 LangChain 没有绑定关系。你可以用它手写路由在路由函数里调用你的 LangGraph 工作流。优点是控制力强中间件、依赖注入、异常处理、OpenAPI 文档全都是你说了算缺点是要自己写不少样板代码尤其是流式输出streaming那块得手动处理StreamingResponse和分块编码。LangGraph Platform是 LangGraph 官方推出的托管方案适合团队不想自己维护服务器的情况。但它涉及托管服务和账号体系这里不展开只讨论自托管路线。2.2 选型判断表维度LangServeFastAPI 手写说明上手速度快10 分钟出接口中等半天到一天取决于是否需要流式灵活性低路由固定高完全自定义鉴权、限流、日志流式输出内置支持需手写 StreamingResponseLangGraph 场景常用多工作流组合一般每个 chain 一个路由灵活可自由编排复杂业务推荐 FastAPI调试体验自带 playground依赖 Swagger UIFastAPI 的 /docs 很好用生产适配需额外包一层直接可控推荐 FastAPI我自己的结论是如果只是 demo 或内部工具LangServe 够用只要涉及对外服务、多工作流、自定义鉴权直接上 FastAPI。LangGraph 的工作流尤其推荐 FastAPI因为 LangGraph 的astream和ainvoke本身就是异步的和 FastAPI 的 async 路由天然契合而 LangServe 对 LangGraph 的支持虽然也有但版本迭代中经常出现兼容性小问题踩过一次就不想再踩。2.3 为什么不是 Flask 或 Django有人会问Flask 也能写 APIDjango 也能为什么偏偏是 FastAPI。三个理由第一原生 async 支持LangGraph 的异步调用在 Flask 的同步模型里会阻塞 worker并发一高就排队第二Pydantic 模型校验请求体和响应体的类型检查是内置的省掉大量手写校验第三自动生成 OpenAPI 文档前端对接时直接把/docs甩过去沟通成本骤降。这三点在 LLM 应用场景里都是刚需所以 FastAPI 基本是这个场景的默认答案。3. FastAPI 项目目录结构别把代码堆在一个文件里3.1 推荐目录结构我见过太多人把整个服务写在一个main.py里三百行起步改一个路由要翻半天。下面这个结构是我在多个项目里沉淀下来的适配 LangGraph 工作流project/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 实例、中间件、路由注册 │ ├── config.py # 环境变量、配置项 │ ├── api/ │ │ ├── __init__.py │ │ ├── routes_chat.py # 对话相关路由 │ │ └── routes_health.py # 健康检查 │ ├── core/ │ │ ├── graph.py # LangGraph 工作流定义 │ │ └── llm.py # 模型客户端初始化 │ ├── models/ │ │ ├── request.py # 请求体 Pydantic 模型 │ │ └── response.py # 响应体 Pydantic 模型 │ └── utils/ │ └── logger.py # 日志配置 ├── tests/ ├── Dockerfile ├── docker-compose.yml ├── requirements.txt └── .env这个结构的核心思路是按职责分层api层只管收请求、调 core、返响应core层放业务逻辑LangGraph 的图定义、节点函数、模型客户端都在这里models层放数据结构请求和响应分开写方便单独演进。3.2 为什么这样分main.py只做三件事创建 FastAPI 实例、挂中间件、注册路由。它不应该包含任何业务逻辑。这样做的直接好处是当你需要加一个新工作流时只需要在core里加一个图在api里加一个路由文件然后在main.py里include_router一行改动面极小。config.py用 Pydantic 的BaseSettings来读环境变量这样 API Key、模型名称、超时时间这些都不硬编码在代码里。.env文件本地开发用生产环境通过容器环境变量注入这是十二要素应用的基本要求。core/llm.py里初始化模型客户端时要注意不要在模块顶层直接实例化而是封装成一个函数或懒加载的单例。原因是有些模型客户端在初始化时会做网络请求或读取环境变量如果放在模块顶层测试时导入模块就会触发很烦。我一般写成from functools import lru_cache from langchain_openai import ChatOpenAI from app.config import settings lru_cache(maxsize1) def get_llm(): return ChatOpenAI( modelsettings.model_name, api_keysettings.api_key, base_urlsettings.base_url, timeoutsettings.timeout, )lru_cache保证只初始化一次同时测试时可以清缓存替换。3.3 请求与响应模型的设计要点请求体模型不要直接用dict一定要用 Pydantic 模型。原因有三自动校验、自动文档、类型提示。一个典型的对话请求模型from pydantic import BaseModel, Field class ChatRequest(BaseModel): message: str Field(..., min_length1, max_length4000) session_id: str Field(defaultdefault) stream: bool Field(defaultFalse)Field里的约束会在请求进来时自动校验不满足直接返回 422不用你手写 if。响应模型同理把content、session_id、usage这些字段定义清楚前端对接时看/docs就够了。注意请求体里不要放api_key这种敏感字段。Key 应该由服务端统一管理通过环境变量注入绝不能由客户端传入。这既是安全要求也避免 Key 泄露后无法追责。4. LangGraph 工作流接入 FastAPI 的关键细节4.1 同步还是异步统一用 asyncLangGraph 的图编译后调用方式有invoke、ainvoke、stream、astream四种。在 FastAPI 里统一用异步版本。原因是 FastAPI 的 async 路由跑在事件循环里如果你在 async 路由里调同步的invoke它会阻塞整个事件循环其他请求全部排队。实测下来一个 3 秒的同步 LLM 调用在并发 10 的情况下第 10 个请求要等 30 秒才返回换成ainvoke之后10 个请求几乎同时返回。路由写法from fastapi import APIRouter from app.core.graph import get_graph from app.models.request import ChatRequest from app.models.response import ChatResponse router APIRouter() router.post(/chat, response_modelChatResponse) async def chat(req: ChatRequest): graph get_graph() result await graph.ainvoke( {messages: [{role: user, content: req.message}]}, config{configurable: {thread_id: req.session_id}}, ) return ChatResponse( contentresult[messages][-1].content, session_idreq.session_id, )thread_id是 LangGraph 做会话记忆的关键同一个thread_id的多次调用会共享状态。这里直接用session_id映射前端每次带同一个 session_id 就能保持上下文。4.2 流式输出怎么接LLM 应用不流式输出用户体验会差很多。LangGraph 的astream支持多种 stream_mode最常用的是messages模式它会逐 token 吐出消息块。在 FastAPI 里用StreamingResponse包装from fastapi.responses import StreamingResponse import json router.post(/chat/stream) async def chat_stream(req: ChatRequest): graph get_graph() async def event_generator(): async for chunk in graph.astream( {messages: [{role: user, content: req.message}]}, config{configurable: {thread_id: req.session_id}}, stream_modemessages, ): token, metadata chunk if token.content: yield fdata: {json.dumps({token: token.content})}\n\n yield data: [DONE]\n\n return StreamingResponse(event_generator(), media_typetext/event-stream)这里用的是 SSEServer-Sent Events格式前端用EventSource或 fetch 的 reader 都能接。注意几个坑第一media_type必须是text/event-stream第二每条消息以\n\n结尾少一个换行浏览器不认第三结束时要发一个明确的结束标记否则前端不知道什么时候停。实操心得SSE 在 Nginx 反代下默认会被缓冲导致流式效果失效。需要在 Nginx 配置里加proxy_buffering off;和proxy_cache off;否则你本地测试是流式的一上生产就变成一次性返回。4.3 超时与重试LLM 调用超时是常态尤其是复杂工作流。FastAPI 层面可以给路由加超时控制但更稳的做法是在模型客户端层面设置 timeout并在工作流里对关键节点做重试。LangGraph 的节点函数里可以用tenacity做重试from tenacity import retry, stop_after_attempt, wait_exponential retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, max10)) async def call_llm_with_retry(llm, messages): return await llm.ainvoke(messages)重试策略要区分错误类型网络超时、429 限流可以重试400 参数错误、401 鉴权失败重试没意义直接抛。tenacity可以配retry_if_exception_type来精细控制。5. Docker 化从能跑到能部署5.1 Dockerfile 怎么写才不臃肿一个常见的错误是直接用python:3.11基础镜像装完依赖镜像 1.5G 起步。推荐用python:3.11-slim再配合多阶段构建。下面这个 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 ENV PYTHONUNBUFFERED1 EXPOSE 8000 CMD [uvicorn, app.main:app, --host, 0.0.0.0, --port, 8000]几个关键点--user把依赖装到用户目录第二阶段只拷贝依赖不拷贝构建缓存PYTHONUNBUFFERED1让日志实时输出不然 Docker logs 看不到--host 0.0.0.0必须写否则容器外访问不到。5.2 docker-compose 编排单容器够用但一旦涉及 Redis做会话缓存或限流、Postgres做持久化就需要 compose。一个典型配置version: 3.9 services: api: build: . ports: - 8000:8000 env_file: - .env depends_on: - redis restart: unless-stopped redis: image: redis:7-alpine ports: - 6379:6379 volumes: - redis_data:/data restart: unless-stopped volumes: redis_data:restart: unless-stopped保证容器崩溃后自动拉起这是生产环境的基本要求。env_file把.env注入容器注意.env不要提交到 Git。5.3 常见 Docker 报错排查报错信息原因解决virtualization support not detectedBIOS 未开虚拟化进 BIOS 开启 VT-x/AMD-Vfailed to connect to the docker api at npipeDocker Desktop 未启动启动 Docker Desktop 等待就绪port is already allocated端口被占用改端口或杀掉占用进程容器启动后立即退出CMD 命令错误或依赖缺失docker logs container看日志镜像构建卡在 pip install网络问题换国内源或配代理镜像注意Windows 上装 Docker Desktop 报虚拟化错误九成是 BIOS 里虚拟化没开。重启进 BIOS找 Intel VT-x 或 AMD-V设为 Enabled保存重启即可。这个坑我踩过两次第一次折腾了一下午。6. 生产环境架构选型单机、多副本还是上编排6.1 三种典型架构单机单容器一台服务器一个容器Nginx 反代。适合日调用量几千以内、对可用性要求不高的场景。优点是简单缺点是挂了就全挂升级要停机。单机多副本一台服务器多个容器实例Nginx 做负载均衡。适合日调用量几万、需要滚动升级的场景。用 docker-compose 的--scale api3就能起三个副本Nginx 配 upstream 轮询。注意会话状态要外置到 Redis否则副本之间不共享上下文。多机编排多台服务器K8s 或 Docker Swarm 编排。适合日调用量十万以上、有专门运维的场景。LLM 应用上 K8s 的收益主要在弹性伸缩和故障自愈但复杂度陡增小团队不建议过早引入。6.2 选型判断我的经验判断线是日调用量 1 万以下单机单容器1 万到 10 万单机多副本加 Redis10 万以上再考虑编排。不要为了“看起来专业”过早引入 K8s运维成本会吃掉你所有开发时间。6.3 关键配置项不管哪种架构这几项必须配健康检查/health端点返回 200Docker 的HEALTHCHECK或 K8s 的 livenessProbe 都靠它。日志收集容器日志输出到 stdout由外部收集不要写文件到容器内。优雅关闭FastAPI 的lifespan里处理关闭逻辑等待进行中的请求完成。限流用 Redis 做令牌桶防止单个用户打爆服务。超时Nginx 的proxy_read_timeout要大于 LLM 最长响应时间否则长请求被切断。7. 常见问题与排查技巧实录7.1 API 调用类问题400 maximum context length 错误这是最常见的报错意思是输入 token 超过了模型上限。解决思路不是简单截断而是做上下文管理。LangGraph 里可以用trim_messages或自定义节点做滑动窗口保留最近 N 轮对话加系统提示。粗暴截断会丢失关键信息导致回答质量下降。401 api_key_requiredKey 没传或传错。检查.env是否被正确加载容器里docker exec -it container env | grep API看一眼。注意有些客户端读的是OPENAI_API_KEY有些读API_KEY名字要对上。429 限流加退避重试同时考虑多 Key 轮询或降级到更小的模型。7.2 部署类问题容器内时区不对基础镜像默认 UTC日志时间对不上。Dockerfile 里加ENV TZAsia/Shanghai并装tzdata。依赖装不上langchain和langgraph版本要匹配建议在requirements.txt里锁死版本号不要用。我吃过一次亏本地跑通构建时拉了新版本接口签名变了直接崩。内存溢出LLM 应用内存占用不小尤其是加载本地模型。Docker 默认不限制内存但宿主机内存有限。compose 里加deploy.resources.limits.memory: 2G做限制避免一个容器拖垮整机。7.3 排查速查表现象优先排查工具接口 500看容器日志docker logs接口超时看 LLM 响应时间、Nginx 超时日志 curl -w流式失效Nginx 缓冲配置proxy_buffering off并发上不去是否用了同步调用检查invokevsainvoke会话串了thread_id 是否唯一检查 session 生成逻辑实操心得排查线上问题时第一件事是看日志第二件事是复现。不要凭猜测改代码。我习惯在路由入口和出口各打一条日志记录 request_id、耗时、状态出问题时按 request_id 串起来看比什么都快。8. 我踩过的几个坑和最后的小建议第一个坑是过早优化。一开始就上 K8s、上服务网格结果光是维护编排配置就花了两周业务代码一行没写。后来退回来用 docker-compose两天上线跑得挺好。架构是长出来的不是设计出来的。第二个坑是忽略流式的反代配置。本地测试流式完美部署到服务器变成一次性返回查了半天以为是代码问题最后发现是 Nginx 默认缓冲。这个坑很隐蔽因为日志里看不出任何异常。第三个坑是Key 管理混乱。早期把 Key 写在代码里后来换 Key 要重新构建镜像。改成环境变量注入后换 Key 只需要改.env重启容器一分钟搞定。这个习惯越早养成越好。最后分享一个小技巧在main.py里加一个/version端点返回当前 Git commit hash 和构建时间。线上出问题时第一件事就是确认跑的是哪个版本避免“我明明改了怎么没生效”这种低级困惑。这个端点十行代码能省下大量扯皮时间。