用 LangChain 开发的智能体从“本地能跑“到“生产可用“

用 LangChain 开发的智能体从本地能跑到生产可用核心要解决的是服务化、状态持久化、可观测性和成本控制四个问题。下面按实际工程路径给你梳理一套完整的部署方案。一、部署前的必要准备1. 状态持久化最重要开发时你通常把对话历史存在内存里ConversationBufferMemory但生产环境一旦服务重启或扩容记忆就全丢了。上线前必须换成数据库持久化。推荐方案表格场景方案说明单会话短期记忆Redis速度快适合对话缓存TTL 自动过期多实例共享状态Postgres AsyncPostgresCheckpointerLangGraph 原生支持K8s 多副本共享状态长期知识记忆向量数据库Milvus/Pinecone/WeaviateRAG 场景跨会话检索如果你用LangGraph构建的 Agent强烈建议从第一天就接入PostgresCheckpointer这样即使 Pod 重启用户对话也能断点续传。2. 设置硬性边界Agent 最大的生产风险是无限循环和Token 费用失控。上线前必须加这三道保险Python# 1. 最大迭代次数防止死循环 agent_executor AgentExecutor( agentagent, toolstools, max_iterations15, # 超过15轮强制停止 max_execution_time30, # 超过30秒超时 early_stopping_methodgenerate ) # 2. 工具调用重试上限防止费用爆炸 from tenacity import retry, stop_after_attempt, wait_exponential retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10)) def call_external_api(...): ...3. 输出校验Agent 的返回结果不稳定下游如果期待 JSON 却收到一段自然语言整个链路就崩了。用 Pydantic 做结构化输出校验Pythonfrom pydantic import BaseModel, Field class AgentOutput(BaseModel): answer: str Field(description最终回答) sources: list[str] Field(description参考来源) confidence: float Field(description置信度, ge0, le1)二、服务化封装把 Agent 变成 APILangChain 本身不是 Web 框架你需要用FastAPI把它包装成 REST 服务。这是目前业界最主流的做法。最小可用示例Python# api.py from fastapi import FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import StreamingResponse from pydantic import BaseModel import uuid import uvicorn app FastAPI(titleLangChain Agent API) # CORS生产环境请限制具体域名 app.add_middleware( CORSMiddleware, allow_origins[*], allow_credentialsTrue, allow_methods[*], allow_headers[*], ) # 预加载核心组件服务启动时一次性加载 from langchain_openai import ChatOpenAI from langgraph.prebuilt import create_react_agent from langgraph.checkpoint.postgres import PostgresSaver import os llm ChatOpenAI(modelgpt-4o, temperature0) tools [...] # 你的工具集 # Postgres 持久化 conn_string os.getenv(POSTGRES_URI) checkpointer PostgresSaver.from_conn_string(conn_string) checkpointer.setup() # 初始化表结构 agent create_react_agent(llm, tools, checkpointercheckpointer) # API 接口 class ChatRequest(BaseModel): message: str session_id: str | None None app.post(/chat) async def chat(req: ChatRequest): session_id req.session_id or str(uuid.uuid4()) config {configurable: {thread_id: session_id}} try: response await agent.ainvoke( {messages: [(human, req.message)]}, configconfig ) return { session_id: session_id, response: response[messages][-1].content } except Exception as e: raise HTTPException(status_code500, detailstr(e)) app.post(/chat/stream) async def chat_stream(req: ChatRequest): 流式输出提升用户体验 session_id req.session_id or str(uuid.uuid4()) config {configurable: {thread_id: session_id}} async def event_generator(): async for chunk in agent.astream( {messages: [(human, req.message)]}, configconfig ): if agent in chunk and messages in chunk[agent]: yield fdata: {chunk[agent][messages][-1].content}\n\n yield data: [DONE]\n\n return StreamingResponse( event_generator(), media_typetext/event-stream ) app.get(/health) async def health(): return {status: ok, agent: ready} if __name__ __main__: uvicorn.run(app, host0.0.0.0, port8000)关键设计点预加载LLM 和 Agent 在启动时初始化不要每次请求都重新创建异步接口用ainvoke/astream而非invoke支持并发会话隔离通过thread_id区分不同用户配合 Postgres Checkpointer 实现有状态对话三、容器化Docker 部署把上面的服务打包成 Docker 镜像这是上云前的标准步骤。dockerfile# Dockerfile FROM python:3.11-slim WORKDIR /app # 安装依赖 COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt # 复制代码 COPY . . # 非 root 用户运行安全最佳实践 RUN useradd -m -u 1000 appuser chown -R appuser:appuser /app USER appuser EXPOSE 8000 CMD [uvicorn, api:app, --host, 0.0.0.0, --port, 8000, --workers, 4]txt# requirements.txt langchain0.3.0 langgraph0.2.0 langchain-openai0.2.0 fastapi0.115.0 uvicorn[standard]0.32.0 psycopg[binary]3.2.0 redis5.0.0 pydantic2.9.0 python-dotenv1.0.0 tenacity9.0.0构建并运行bashdocker build -t my-langchain-agent . docker run -d \ -p 8000:8000 \ -e OPENAI_API_KEYsk-xxx \ -e POSTGRES_URIpostgresql://user:passhost:5432/db \ my-langchain-agent四、生产部署的几种方式根据你的团队规模和流量选择不同方案方式 1单服务器直接部署适合初创/测试直接用 Docker Compose 在一台云服务器上跑yaml# docker-compose.yml version: 3.8 services: agent: build: . ports: - 8000:8000 environment: - OPENAI_API_KEY${OPENAI_API_KEY} - POSTGRES_URIpostgresql://postgres:postgresdb:5432/agent_db - REDIS_URIredis://redis:6379 depends_on: - db - redis db: image: postgres:15-alpine environment: POSTGRES_USER: postgres POSTGRES_PASSWORD: postgres POSTGRES_DB: agent_db volumes: - pgdata:/var/lib/postgresql/data redis: image: redis:7-alpine volumes: - redisdata:/data volumes: pgdata: redisdata:方式 2Kubernetes 部署适合中高流量yaml# k8s-deployment.yaml apiVersion: apps/v1 kind: Deployment metadata: name: langchain-agent spec: replicas: 3 selector: matchLabels: app: langchain-agent template: metadata: labels: app: langchain-agent spec: containers: - name: agent image: your-registry/my-langchain-agent:latest ports: - containerPort: 8000 env: - name: OPENAI_API_KEY valueFrom: secretKeyRef: name: api-secrets key: openai-key - name: POSTGRES_URI valueFrom: secretKeyRef: name: api-secrets key: postgres-uri resources: requests: memory: 1Gi cpu: 500m limits: memory: 2Gi cpu: 1000m --- apiVersion: v1 kind: Service metadata: name: langchain-agent-service spec: selector: app: langchain-agent ports: - port: 80 targetPort: 8000 type: ClusterIP配合HPA自动扩缩容bashkubectl autoscale deployment langchain-agent --cpu-percent70 --min3 --max10方式 3Serverless / 云函数适合低频、突发流量如果你的 Agent 调用量不大用 Serverless 更省钱表格平台方式Vercel前端 Next.js Edge Function 调用 Agent APIAWS Lambda用 Mangum 把 FastAPI 包成 Lambda 函数阿里云函数计算直接部署容器镜像腾讯云云函数类似方案Serverless 的注意事项冷启动时 LLM 初始化慢建议加预置并发状态必须用外部数据库Redis/Postgres不能存内存。方式 4LangGraph Platform官方托管如果你不想自己运维 K8sLangChain 官方提供了LangGraph Platform企业版支持一键部署 LangGraph 应用自带自动扩缩容、监控、断点续传等能力。五、生产环境必须做的五件事1. 可观测性接入 LangSmithAgent 的调试比传统软件难 10 倍——你不知道它为什么调了 3 次工具还没出结果。LangSmith 可以捕获每一次 Agent 的完整执行链路trace包括每次 LLM 调用、工具调用、Token 消耗。Pythonimport os os.environ[LANGCHAIN_TRACING_V2] true os.environ[LANGCHAIN_API_KEY] ls-xxx os.environ[LANGCHAIN_PROJECT] production-agent注意高流量生产环境200 req/day建议关闭 LangSmith 实时追踪或采样上报否则性能开销明显。2. 限流与成本控制Agent 的 Token 费用可能比你想象的高得多。在 API 网关层加限流Python# FastAPI 限流示例 from slowapi import Limiter, _rate_limit_exceeded_handler from slowapi.util import get_remote_address limiter Limiter(key_funcget_remote_address) app.state.limiter limiter app.post(/chat) limiter.limit(10/minute) # 每IP每分钟10次 async def chat(req: ChatRequest, request: Request): ...同时设置预算告警监控单日 Token 消耗超过阈值自动告警。3. 安全加固表格风险防护措施Prompt 注入输入过滤 输出校验用langchain_core.prompts的防御模板PII 泄露日志和 Trace 中脱敏手机号、身份证号正则替换API Key 泄露用 K8s Secret / 阿里云 KMS / AWS Secrets Manager 管理越权调用每个工具内部校验用户权限不要依赖 Agent 自觉4. 评估体系上线前准备一套测试集定期回归Python# 用 LangSmith 做批量评估 from langsmith import Client from langchain.smith import RunEvalConfig client Client() eval_config RunEvalConfig( evaluators[ qa, # 问答正确性 criteria, # 自定义标准 embedding_distance # 语义相似度 ] ) client.run_on_dataset( dataset_nameagent_test_set, llm_or_chain_factorylambda: agent, evaluationeval_config )5. 优雅降级当 LLM API 超时或限流时Agent 不能崩Pythonfrom tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type from openai import RateLimitError retry( retryretry_if_exception_type((RateLimitError, TimeoutError)), stopstop_after_attempt(3), waitwait_exponential(multiplier1, min2, max10) ) def safe_llm_call(...): return llm.invoke(...)六、部署 checklist表格检查项状态□ 状态持久化已接入 Postgres/Redis□max_iterations和max_execution_time已设置□ 输出已用 Pydantic 做结构化校验□ API 已用 FastAPI 封装支持异步和流式□ 已打包 Docker 镜像□ 环境变量/API Key 已用 Secret 管理□ LangSmith 追踪已接入采样模式□ 限流和预算告警已配置□ 测试集评估通过□ 健康检查/health接口已提供总结LangChain Agent 的部署没有银弹但有清晰的演进路径本地开发→ 用内存状态快速迭代测试部署→ Docker 单服务器 Postgres 持久化生产上线→ K8s 多副本 Redis/Postgres LangSmith 监控 限流降级大规模生产→ 考虑 LangGraph Platform 或自建更精细的调度系统核心原则就一条Agent 不是确定性程序而是有行为的系统——你需要的是可观测性、状态持久化和边界控制而不是传统软件的那套监控思维。