企业官网建设流程全解析

1. 从脚本到服务，为什么部署这一步最容易卡住人

LangGraph 这个框架，写过 demo 的人都知道，本地跑起来是真舒服。一个StateGraph，几个节点函数，compile()一下，invoke()一调，状态在节点之间流转，图跑通了，逻辑也就验证完了。但问题往往出在下一步：怎么把这个跑在 Jupyter Notebook 或者python xxx.py里的东西，变成一个别人能访问、能持续运行、挂了能自动重启的服务？

我见过太多项目卡在这个坎上。脚本阶段一切正常，一旦要部署，问题就来了：状态怎么持久化？并发请求怎么处理？多个用户同时调用同一个图，状态会不会串？进程崩了怎么办？要不要上容器？上了容器之后 K8s 那套东西又该怎么配？

这些问题的本质，其实是从"能跑"到"能稳定对外服务"之间的鸿沟。LangGraph 本身是个编排框架，它不负责网络层、不负责进程管理、不负责容器编排。这些都得你自己选、自己搭。而选择多了，反而容易懵。

这篇文章就是想把这条路径讲清楚。我会按三条部署路径来拆：本地脚本直跑、FastAPI 封装成 HTTP 服务、Docker + K8s 容器化编排。每一条路径适合什么场景、核心要解决什么问题、具体怎么落地、有哪些坑，我都会结合自己的实操经验说透。不管你是刚写完第一个 LangGraph demo 想给别人试用，还是已经在做多租户的 Agent 平台，应该都能找到对应的参考。

先说清楚一个前提：LangGraph 的部署，核心矛盾从来不是"图怎么跑"，而是"图跑起来之后，状态、并发、生命周期这三件事怎么管"。三条路径的差异，本质上就是在这三件事上投入的复杂度不同。理解了这一点，后面的选型就不会盲目。

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

2.1 路径划分的底层依据：状态管理复杂度

为什么我把部署路径分成三条，而不是按"开发/测试/生产"这种常规维度分？因为对 LangGraph 这类有状态编排框架来说，决定部署复杂度的核心变量是状态管理的需求，不是环境本身。

一个无状态的 HTTP 接口，你随便怎么部署都行，多开几个实例做负载均衡就完事。但 LangGraph 的图是有状态的，StateGraph里的 state 在节点间传递，如果用了checkpointer（比如MemorySaver或SqliteSaver），状态还要跨请求持久化。这就带来几个连锁问题：

• 状态存在哪？内存里、SQLite 文件里、还是 Postgres 里？
• 多个实例部署时，状态怎么共享？
• 一个长流程跑到一半，进程重启了，状态还在不在？

这三个问题的答案，直接决定了你该走哪条路径。我整理了一张对照表，先把三条路径的定位说清楚：

这张表不是让你对号入座，而是帮你判断：你现在真正需要解决的是哪一层的问题。很多人一上来就想上 K8s，结果发现自己的状态管理根本没设计好，上了 K8s 反而更乱——因为多实例之间状态不共享，请求打到哪个 Pod 全看运气，用户体验直接崩掉。

2.2 为什么 FastAPI 是中间路径的最优解

三条路径里，FastAPI 这条是绝大多数项目的"甜点区"。原因有几个，我逐个说。

第一，LangGraph 是 Python 生态的，FastAPI 也是，两者天然亲和。你不需要跨语言做序列化，图的输入输出直接就是 Python 对象，Pydantic 模型一套，请求校验和响应格式化全搞定。相比之下，如果你用 Flask，异步支持弱，处理 LangGraph 里可能出现的异步节点（比如调用 LLM 的ainvoke）会很别扭；用 Django 又太重，为了一个图服务引入整个 ORM 和 admin，不划算。

第二，FastAPI 的异步模型和 LangGraph 的流式输出天然匹配。LangGraph 支持astream和astream_events，可以边跑边吐 token。FastAPI 的StreamingResponse配合async def生成器，能把流式输出直接透传给前端。这个组合我在实际项目里用过，SSE（Server-Sent Events）推流非常顺，用户能实时看到 Agent 的思考过程，体验比等一个完整响应好太多。

第三，FastAPI 自带 OpenAPI 文档。你部署完，访问/docs就能看到所有接口的交互式文档，调试和对接都省事。对于内部服务来说，这等于白送了一套接口说明。

当然，FastAPI 不是没有代价。它的异步模型要求你对async/await有基本理解，如果图里有阻塞操作（比如同步的数据库查询、CPU 密集计算），直接写在async def里会阻塞事件循环，整个服务的并发能力就废了。这个坑后面会专门讲怎么绕。

2.3 Docker 和 K8s 各自解决什么问题，别混为一谈

很多人把 Docker 和 K8s 当成一回事，其实它们解决的是完全不同层面的问题，我经常用这个类比来解释：

Docker 解决的是"这个服务在我机器上能跑，在你机器上也能跑"的问题，它管的是单个容器的打包和运行。K8s 解决的是"这个服务有 10 个副本，挂了 2 个要自动补上，流量要均匀分发"的问题，它管的是容器的编排和调度。

对 LangGraph 服务来说，Docker 的价值在于环境一致性。LangGraph 依赖的包不少——langgraph、langchain-core、各种 LLM SDK、可能还有向量库客户端，版本冲突是家常便饭。打成镜像之后，Python 版本、依赖版本、系统库全部锁定，换台机器docker run就能起，不用再折腾环境。

K8s 的价值则在于弹性伸缩和高可用。当你的 Agent 服务要面对不确定的流量，或者需要保证 99.9% 的可用性时，K8s 的 Deployment、Service、HPA（水平自动扩缩容）才有意义。但前提是，你的状态必须外置——因为 K8s 随时可能把 Pod 调度到别的节点，本地文件存储的状态会丢。

所以选型逻辑很清晰：先问状态存哪，再问要不要多实例，最后才决定上不上 K8s。状态还在进程内存里，就别急着上 K8s，先把状态外置这件事解决了再说。

3. 路径一：脚本直跑，最快验证但别拿去见客户

3.1 什么情况下脚本直跑就够了

脚本直跑不是"低级"方案，它在特定场景下是最优解。我列几个典型情况：

• 个人验证阶段：你刚写完图，想快速跑几个 case 看看逻辑对不对，这时候搞一套 FastAPI 纯属浪费时间。
• 内部演示：给同事或领导演示一下 Agent 能干什么，本地跑个脚本，终端里看输出，足够了。
• 批处理任务：图是用来处理一批数据的，跑完就结束，不需要常驻服务。比如批量给文档打标签、批量生成摘要。
• 定时任务：配合 cron 或系统的任务计划，每天跑一次，跑完退出。

这些场景的共同点是：没有并发需求，没有持续在线的需求，状态不需要跨请求共享。满足这三点，脚本直跑就是最省事的方案。

3.2 脚本直跑的核心写法与状态处理

脚本直跑的关键，是把图的编译和调用写清楚。我拿一个最简的 ReAct 风格 Agent 举例，说明几个容易忽略的点。

from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver from typing import TypedDict, Annotated import operator class AgentState(TypedDict): messages: Annotated[list, operator.add] next_step: str def planner_node(state: AgentState): # 这里通常是调用 LLM 做规划 return {"next_step": "tool"} def tool_node(state: AgentState): # 执行工具调用 return {"messages": [{"role": "tool", "content": "result"}]} def should_continue(state: AgentState): if state["next_step"] == "tool": return "tool" return END builder = StateGraph(AgentState) builder.add_node("planner", planner_node) builder.add_node("tool", tool_node) builder.add_edge(START, "planner") builder.add_conditional_edges("planner", should_continue, {"tool": "tool", END: END}) builder.add_edge("tool", "planner") # 关键点一：checkpointer 决定状态存哪 memory = MemorySaver() graph = builder.compile(checkpointer=memory) # 关键点二：config 里的 thread_id 是状态隔离的钥匙 config = {"configurable": {"thread_id": "user-001"}} result = graph.invoke({"messages": [{"role": "user", "content": "帮我查一下天气"}]}, config) print(result)

这段代码里有两个点值得展开。

第一，checkpointer的选择。MemorySaver把状态存在进程内存里，进程一退，状态全没。如果你需要状态在脚本多次运行之间保留，得换成SqliteSaver：

from langgraph.checkpoint.sqlite import SqliteSaver import sqlite3 conn = sqlite3.connect("checkpoints.db", check_same_thread=False) memory = SqliteSaver(conn)

SqliteSaver把状态写进本地 SQLite 文件，脚本重启后还能接着上次的thread_id继续跑。这个在批处理场景里很有用——比如处理到一半中断了，重跑时能跳过已完成的。

第二，thread_id的作用。它是状态隔离的标识，同一个thread_id的多次invoke会共享状态历史，不同thread_id之间互不干扰。脚本直跑时，如果你要处理多个用户的数据，记得给每个用户分配独立的thread_id，否则状态会串。

3.3 脚本直跑的注意事项

脚本直跑有几个坑，我踩过，这里提醒一下。

注意：MemorySaver在多进程环境下完全失效。如果你用multiprocessing或者gunicorn多 worker 跑脚本，每个进程有自己独立的内存，状态不共享。这时候必须换SqliteSaver或外部存储。

另一个坑是异常处理。脚本直跑时，图里某个节点抛异常，整个脚本就挂了。如果你在跑批处理，一个 case 失败导致整批中断，很烦。建议在调用层包一层 try-except，把失败的 case 记下来，继续跑下一个：

for item in batch: try: result = graph.invoke({"messages": [item]}, config) except Exception as e: print(f"处理 {item} 失败: {e}") continue

还有一点，脚本直跑时日志要打全。因为没有服务层的监控，出问题只能靠日志排查。建议在关键节点加日志，记录输入、输出、耗时。LangGraph 本身支持通过callbacks接入 LangSmith 之类的追踪工具，但如果你不想引入外部依赖，自己用logging模块打点也够用。

4. 路径二：FastAPI 封装，把图变成能对外服务的 API

4.1 FastAPI 项目目录结构怎么设计才不乱

从脚本到服务，第一件事是把代码组织好。我见过太多项目，所有代码堆在一个main.py里，几百行，改一处牵全身。LangGraph 服务因为涉及图定义、节点逻辑、工具、API 路由、配置，更需要清晰的目录结构。

我常用的结构是这样的：

langgraph-service/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 入口，挂载路由 │ ├── config.py # 配置管理（环境变量、常量） │ ├── graph/ │ │ ├── __init__.py │ │ ├── builder.py # 图的构建逻辑 │ │ ├── state.py # State 定义 │ │ └── nodes.py # 各节点函数 │ ├── tools/ │ │ └── __init__.py # 工具定义 │ ├── api/ │ │ ├── __init__.py │ │ ├── routes.py # 路由定义 │ │ └── schemas.py # Pydantic 请求/响应模型 │ └── services/ │ └── agent_service.py # 业务逻辑，封装图的调用 ├── tests/ ├── requirements.txt ├── Dockerfile └── .env

这个结构的分层逻辑是：graph 层只管图怎么建，api 层只管请求怎么接，services 层把两者粘起来。这样改图逻辑不影响 API，改 API 不影响图，测试也好写。

config.py里集中管理配置，比如 LLM 的 API key、模型名、checkpointer 的连接串。用pydantic-settings从环境变量读，本地开发用.env，部署时用环境变量注入，一套代码两种环境都能跑。

4.2 用 FastAPI 封装 LangGraph 的核心代码

先看 Pydantic 模型，定义请求和响应：

from pydantic import BaseModel, Field from typing import Optional class ChatRequest(BaseModel): message: str = Field(..., description="用户输入") thread_id: str = Field(..., description="会话标识，用于状态隔离") stream: bool = Field(False, description="是否流式返回") class ChatResponse(BaseModel): thread_id: str reply: str status: str = "ok"

再看路由，核心是把图的调用封装成 HTTP 接口：

from fastapi import APIRouter, HTTPException from fastapi.responses import StreamingResponse from app.api.schemas import ChatRequest, ChatResponse from app.services.agent_service import AgentService router = APIRouter(prefix="/api/v1") agent_service = AgentService() @router.post("/chat", response_model=ChatResponse) async def chat(req: ChatRequest): try: reply = await agent_service.run(req.message, req.thread_id) return ChatResponse(thread_id=req.thread_id, reply=reply) except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @router.post("/chat/stream") async def chat_stream(req: ChatRequest): async def event_generator(): async for chunk in agent_service.stream(req.message, req.thread_id): yield f"data: {chunk}\n\n" return StreamingResponse(event_generator(), media_type="text/event-stream")

AgentService是关键，它把图的调用包起来，对外暴露简单的run和stream方法：

from app.graph.builder import build_graph from langgraph.checkpoint.memory import MemorySaver class AgentService: def __init__(self): self.graph = build_graph(checkpointer=MemorySaver()) async def run(self, message: str, thread_id: str) -> str: config = {"configurable": {"thread_id": thread_id}} result = await self.graph.ainvoke( {"messages": [{"role": "user", "content": message}]}, config ) return result["messages"][-1]["content"] async def stream(self, message: str, thread_id: str): config = {"configurable": {"thread_id": thread_id}} async for event in self.graph.astream_events( {"messages": [{"role": "user", "content": message}]}, config, version="v2" ): if event["event"] == "on_chat_model_stream": chunk = event["data"]["chunk"].content if chunk: yield chunk

这里有几个设计决策值得说。

为什么用ainvoke而不是invoke？因为 FastAPI 的路由是async def，在异步上下文里调用同步的invoke会阻塞事件循环。ainvoke是异步版本，能让出控制权，其他请求可以并发处理。这个区别在高并发下非常明显，我实测过，同步调用在 10 个并发请求时就开始排队，异步调用能轻松扛住几十个。

为什么用astream_events而不是astream？astream返回的是每个节点执行完后的状态快照，粒度太粗。astream_events能拿到更细粒度的事件，包括 LLM 的 token 流，适合做打字机效果。version="v2"是必须的，v1 的事件格式不一样，容易踩坑。

4.3 状态持久化：从 MemorySaver 到 Redis

MemorySaver在单实例 FastAPI 里能用，但有两个问题：一是进程重启状态丢失，二是多 worker 部署时状态不共享。生产环境得换外部存储。

LangGraph 官方支持多种 checkpointer，常用的有SqliteSaver、PostgresSaver、RedisSaver。选哪个看你的场景：

• 单机小服务：SqliteSaver够用，零依赖，一个文件搞定。
• 多实例部署：PostgresSaver或RedisSaver，状态集中存储，所有实例共享。
• 需要 TTL 自动过期：RedisSaver更合适，可以给会话状态设过期时间。

以RedisSaver为例，配置大概是这样：

from langgraph.checkpoint.redis import RedisSaver import redis redis_client = redis.Redis(host="localhost", port=6379, db=0) checkpointer = RedisSaver(redis_client) graph = builder.compile(checkpointer=checkpointer)

注意：换 checkpointer 之后，thread_id的语义不变，但状态的存储位置变了。如果你之前用MemorySaver跑了一些会话，换到 Redis 后这些会话的历史就找不回来了。迁移时要么接受历史丢失，要么写个脚本把内存状态导出再导入。

4.4 FastAPI 部署的实操心得与常见坑

FastAPI 封装这条路径，我踩过的坑主要集中在几个地方，整理成速查表：

关于日志丢失这个问题，我单独说一下。uvicorn 启动时会自己配置 logging，如果你在代码里用logging.basicConfig()或者自定义 logger，很容易被 uvicorn 的配置覆盖。正确的做法是在启动时传入log_config：

import uvicorn if __name__ == "__main__": uvicorn.run( "app.main:app", host="0.0.0.0", port=8000, log_config="log_config.yaml" # 自定义日志配置 )

log_config.yaml里定义好 formatter、handler、logger，这样 uvicorn 和你的应用日志能统一格式，排查问题时不会一个有一个没有。

还有一个实操技巧：用lifespan管理图的初始化。图的构建、checkpointer 的连接，这些应该在服务启动时做一次，而不是每个请求都做。FastAPI 的lifespan上下文管理器正好干这个：

from contextlib import asynccontextmanager from fastapi import FastAPI @asynccontextmanager async def lifespan(app: FastAPI): # 启动时初始化 app.state.agent_service = AgentService() yield # 关闭时清理 await app.state.agent_service.close() app = FastAPI(lifespan=lifespan)

这样图只构建一次，checkpointer 连接也只建一次，请求来了直接用，性能好很多。

5. 路径三：Docker + K8s，多实例编排的完整落地

5.1 Dockerfile 怎么写才又小又快

LangGraph 服务的镜像，我建议用多阶段构建，把构建依赖和运行时依赖分开，镜像能小不少。基础镜像选python:3.11-slim，比完整版小几百 MB。

# 构建阶段 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"]

几个优化点：

• --no-cache-dir不缓存 pip 下载的包，镜像更小。
• --user把包装到用户目录，方便从构建阶段复制。
• 只复制app目录，测试、文档、.env都不进镜像。
• 用CMD而不是ENTRYPOINT，方便运行时覆盖命令。

.dockerignore也要配好，把__pycache__、.git、tests、*.md排除掉，构建上下文小，构建快：

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

注意：不要把 API key 写进 Dockerfile 或镜像里。用环境变量在运行时注入，或者用 K8s 的 Secret 挂载。镜像一旦推到仓库，里面的内容就藏不住了。

5.2 K8s 部署清单：Deployment、Service、ConfigMap

K8s 部署 LangGraph 服务，核心是三个资源：Deployment 管 Pod，Service 管网络，ConfigMap/Secret 管配置。

Deployment 的清单：

apiVersion: apps/v1 kind: Deployment metadata: name: langgraph-agent spec: replicas: 3 selector: matchLabels: app: langgraph-agent template: metadata: labels: app: langgraph-agent spec: containers: - name: agent image: your-registry/langgraph-agent:v1.0.0 ports: - containerPort: 8000 env: - name: REDIS_HOST valueFrom: configMapKeyRef: name: agent-config key: redis_host - name: LLM_API_KEY valueFrom: secretKeyRef: name: agent-secret key: llm_api_key resources: requests: memory: "512Mi" cpu: "250m" limits: memory: "1Gi" cpu: "1000m" livenessProbe: httpGet: path: /health port: 8000 initialDelaySeconds: 30 periodSeconds: 10 readinessProbe: httpGet: path: /health port: 8000 initialDelaySeconds: 10 periodSeconds: 5

这里有几个关键配置。

replicas: 3表示起 3 个副本。但前提是状态已经外置到 Redis 或 Postgres，否则 3 个副本各自有各自的内存状态，请求打到哪个 Pod 状态就不一样，用户体验会崩。

resources的 requests 和 limits 要设。requests 是调度依据，limits 是硬上限。LangGraph 服务如果调 LLM，内存占用主要在等待响应时的连接和缓冲，512Mi 到 1Gi 是常见范围。CPU 方面，如果图里有本地计算，给足；纯调 API 的话，250m 到 500m 够用。

livenessProbe和readinessProbe是保命的。liveness 探针失败，K8s 会重启 Pod；readiness 探针失败，K8s 会把 Pod 从 Service 的端点里摘掉，不再转发流量。所以你的服务要提供一个/health接口，返回 200 表示健康。这个接口最好能检查关键依赖（比如 Redis 连接），而不只是返回一个静态的 ok。

Service 的清单：

apiVersion: v1 kind: Service metadata: name: langgraph-agent-svc spec: selector: app: langgraph-agent ports: - port: 80 targetPort: 8000 type: ClusterIP

ClusterIP是集群内访问，如果要对集群外暴露，用NodePort或LoadBalancer，或者通过 Ingress 转发。生产环境一般用 Ingress，配合域名和证书。

5.3 多实例下的状态一致性怎么保证

这是 K8s 部署 LangGraph 最核心的问题。3 个副本，用户请求可能打到任意一个，如果状态存在 Pod 本地，就会出现"这次对话在这个 Pod，下次对话在另一个 Pod，历史全没了"的情况。

解决方案只有一个：状态必须外置，所有 Pod 共享同一个存储。具体做法：

• checkpointer 用RedisSaver或PostgresSaver，连接串通过环境变量注入。
• 所有 Pod 连同一个 Redis/Postgres 实例（或集群）。
• thread_id作为状态隔离的 key，无论请求打到哪个 Pod，都能从共享存储里读到同一个会话的状态。

这样配置之后，Pod 是无状态的，可以随意扩缩容、重启、迁移，状态始终一致。

注意：Redis 单点也是故障点。生产环境建议用 Redis 集群或哨兵模式，避免 Redis 挂了整个服务不可用。如果对持久化要求高，用 Postgres 更稳，但延迟比 Redis 高一些。

5.4 K8s 部署的常见故障与排查

K8s 部署的坑不少，我挑几个高频的说。

Pod 一直CrashLoopBackOff。最常见的原因是启动命令报错，或者依赖连不上。排查步骤：kubectl logs <pod-name>看日志，kubectl describe pod <pod-name>看事件。如果是连不上 Redis，检查 ConfigMap 里的地址对不对，网络策略有没有放行。

the api server is not healthy after 4m。这个报错在初始化 K8s 控制节点时出现，通常是网络插件没装好，或者容器运行时配置有问题。检查kubelet状态、containerd状态，确认网络插件（如 Calico、Flannel）的 Pod 是否正常运行。

Service 访问不通。先确认 Pod 的 readiness 探针是否通过，kubectl get endpoints <svc-name>看有没有端点。如果端点是空的，说明没有 Pod 通过就绪检查。再检查 Service 的selector和 Pod 的labels是否匹配，这个经常因为拼写错误导致。

镜像拉取失败ImagePullBackOff。私有仓库要配imagePullSecrets，把仓库的认证信息做成 Secret，在 Deployment 里引用。另外确认镜像 tag 存在，别写了个不存在的版本。

排查 K8s 问题，我习惯按这个顺序：kubectl get pods看状态 →kubectl describe pod看事件 →kubectl logs看应用日志 →kubectl exec进容器手动验证。大部分问题在前两步就能定位。

6. 三条路径的选型决策与迁移时机

6.1 什么时候该从脚本升级到 FastAPI

脚本直跑够不够用，判断标准很简单：有没有"别人要调用"的需求。如果只有你自己跑，脚本就行；一旦有第二个人要用，或者要接入前端、接入其他系统，就该上 FastAPI 了。

具体信号包括：

• 需要 HTTP 接口，让前端或其他服务调用。
• 需要并发处理多个请求。
• 需要流式输出，做打字机效果。
• 需要统一的鉴权、限流、日志。

这些需求出现任何一个，脚本直跑就不够了。升级到 FastAPI 的成本其实不高，核心工作就是把图的调用包一层 HTTP，代码量不大，但收益明显。

6.2 什么时候该从 FastAPI 升级到 K8s

FastAPI 单实例能扛的量其实不小，我实测过一个 4 核 8G 的机器，跑 LangGraph 服务，QPS 能到几十（取决于图里 LLM 调用的耗时）。所以不是一上来就得上 K8s。

该上 K8s 的信号：

• 需要高可用：单实例挂了服务就断了，业务不能接受。
• 需要弹性伸缩：流量波动大，高峰期要自动扩容，低谷期要缩容省钱。
• 需要多环境隔离：开发、测试、生产要用同一套部署方式，环境之间隔离。
• 团队规模上来了：多个人协作，需要标准化的部署流程和资源管理。

如果只是内部小工具，日活几十人，FastAPI 单实例加个systemd守护进程就够了，上 K8s 是过度设计。K8s 的学习成本和运维成本都不低，别为了技术而技术。

6.3 迁移过程中的状态迁移策略

从一条路径迁到另一条，最大的风险是状态丢失。我建议的迁移策略是双写过渡：

1. 新路径部署好，checkpointer 指向新的存储。
2. 旧路径暂时保留，但把新请求都导到新路径。
3. 旧路径上的存量会话，等自然结束后下线。
4. 确认新路径稳定后，彻底关掉旧路径。

如果存量会话很重要，不能等它自然结束，那就写个迁移脚本，把旧存储的状态读出来，写到新存储。LangGraph 的 checkpointer 接口是统一的，get和put方法都有，写迁移脚本不难，但要注意状态格式的兼容性——不同 checkpointer 的序列化方式可能不一样，迁移前先在测试环境验证。

7. 实操中踩过的坑与独家经验

7.1 流式输出在生产环境的三个陷阱

流式输出在本地跑得好好的，一上生产就出问题，我遇到过三次，每次原因都不一样。

第一次是 Nginx 缓冲。Nginx 默认会缓冲响应，导致 SSE 的 chunk 攒够一批才发出去，用户看到的还是"卡一下出一大段"。解决方法是加响应头X-Accel-Buffering: no，或者在 Nginx 配置里对 SSE 的 location 关掉proxy_buffering。

第二次是 uvicorn 的 worker 数。uvicorn 多 worker 模式下，SSE 连接可能被分配到不同 worker，如果状态在内存里，流式过程中读不到之前的上下文。这个的根因还是状态没外置，换成 Redis 之后就好了。

第三次是客户端超时。SSE 连接如果长时间没有数据，某些客户端或中间层会主动断开。解决方法是加心跳，每隔 15 秒发一个空注释: heartbeat\n\n，保持连接活跃。

7.2 资源限制设多少才合理

K8s 里resources设多少，很多人拍脑袋。我的经验是先观察再设。本地跑的时候用docker stats看内存和 CPU 占用，取峰值上浮 50% 作为 limits，取平均值的 1.2 倍作为 requests。

LangGraph 服务的资源占用，主要看图的复杂度：

• 纯 LLM 调用，无本地计算：内存 512Mi 够，CPU 250m 够。
• 有向量检索、本地 embedding：内存 1Gi 起，CPU 500m 起。
• 有本地模型推理：那得按模型大小算，另当别论。

注意：limits 设太低会导致 Pod 被 OOMKilled，设太高会浪费资源。建议先用宽松的 limits 跑一段时间，看监控数据再收紧。

7.3 日志和监控怎么配才够用

服务上线后，出问题第一件事是看日志。LangGraph 服务的日志，我建议分三层：

• 访问日志：记录每个请求的thread_id、耗时、状态码。用 FastAPI 的中间件统一打。
• 业务日志：记录图的关键节点执行情况，比如"进入 planner 节点"、"工具调用返回"。
• 错误日志：记录异常堆栈，带上thread_id方便定位是哪个会话出的问题。

监控方面，至少要有 QPS、响应时间、错误率、Pod 数量这几个指标。Prometheus + Grafana 是标配，FastAPI 可以用prometheus-fastapi-instrumentator快速接入，几行代码就能暴露指标。

7.4 常见问题速查表

8. 写在最后的一点个人体会

三条路径走下来，我最大的感受是：部署这件事，复杂度不在技术本身，而在对需求的判断。很多人一上来就想上最重的方案，结果发现根本用不上，反而被 K8s 的运维拖累。也有人一直用脚本直跑，等到业务量上来了才手忙脚乱地重构。

我的建议是按需演进，别提前优化。先用脚本验证逻辑，逻辑通了上 FastAPI 对外服务，服务稳定了、有高可用需求了再上 K8s。每一步的迁移成本都不高，因为 LangGraph 的图定义和状态管理是解耦的，换部署方式不用改图本身。

最后分享一个小技巧：不管走哪条路径，把图的构建逻辑和部署逻辑彻底分开。图定义放在graph/目录，部署相关的代码放在api/、deploy/目录。这样你从 FastAPI 迁到 K8s 时，图代码一行不用动，只改部署配置。这个习惯能帮你省下大量重构时间。