第十一章 生产部署与最佳实践
11.1 LangGraph Platform / Cloud
你的 Agent 在本地跑得很好,但如何让它服务于真实用户?LangGraph 提供了两种部署方案。
LangGraph Platform(自托管)
LangGraph Platform 是一个自托管的服务器框架,把你的图暴露为 API:
python
# langgraph.json —— 项目配置文件
{
"graphs": {
"my_agent": "./agent.py:app"
},
"dependencies": ["langchain-deepseek", "langgraph"],
"env": ".env"
}bash
# 安装 CLI
pip install langgraph-cli
# 本地开发服务器
langgraph dev
# 构建 Docker 镜像
langgraph build -t my-agent:latest
# 运行容器
docker run -p 8123:8000 my-agent:latestLangGraph Platform 的架构:
┌─────────────────────────────────────────┐
│ LangGraph Platform │
│ │
│ ┌──────────┐ ┌──────────────────┐ │
│ │ 你的图 │ │ 内置 API │ │
│ │ (Agent) │──→│ /runs/stream │──→ 客户端
│ │ │ │ /threads │ │
│ └──────────┘ │ /store │ │
│ └──────────────────┘ │
│ │
│ ┌──────────────────────────────────┐ │
│ │ Checkpointer(PostgreSQL) │ │
│ └──────────────────────────────────┘ │
└─────────────────────────────────────────┘LangGraph Cloud(托管服务)
不想管服务器?LangGraph Cloud 是 LangChain 提供的全托管方案:
自托管 vs 托管:
LangGraph Platform(自托管)
✅ 完全控制
✅ 数据不出你的服务器
❌ 需要自己管理服务器
LangGraph Cloud(托管)
✅ 零运维
✅ 自动扩缩容
❌ 数据经过 LangChain 服务器
❌ 付费服务| 方案 | 适合谁 | 费用 |
|---|---|---|
| 本地开发 | 个人学习/原型 | 免费 |
| Platform(自托管) | 企业/隐私敏感 | 服务器成本 |
| Cloud(托管) | 快速上线/小团队 | 按用量付费 |
建议:学习和原型阶段用本地开发。需要上线时,优先考虑 LangGraph Platform 自托管——你已经有了 Docker,部署并不复杂。
11.2 性能优化:异步、缓存、批处理
异步是基础
Agent 的大部分时间花在等待 I/O——等 LLM 回复、等 API 返回、等数据库查询。用异步可以大幅提升吞吐量:
python
# ❌ 同步:每次 LLM 调用阻塞整个线程
def agent(state):
response = llm.invoke(state["messages"])
return {"messages": [response]}
# ✅ 异步:等待 LLM 时可以处理其他请求
async def agent(state):
response = await llm.ainvoke(state["messages"])
return {"messages": [response]}
# 异步运行
result = await app.ainvoke(input_data, config)LLM 缓存
相同的输入不必每次都调 LLM——用缓存可以节省时间和费用:
python
from langchain_core.globals import set_llm_cache
from langchain_community.cache import SQLiteCache
# 设置 SQLite 缓存
set_llm_cache(SQLiteCache(database_path="llm_cache.db"))
# 之后所有 LLM 调用自动缓存
# 相同输入第二次调用 → 直接返回缓存结果,不调 API并行节点执行
LangGraph 自动并行执行没有依赖关系的节点:
python
# A → B 和 A → C 会并行执行
graph.add_edge("A", "B")
graph.add_edge("A", "C")
graph.add_edge("B", "D")
graph.add_edge("C", "D")性能优化速查
| 优化项 | 方法 | 效果 |
|---|---|---|
| 异步执行 | ainvoke / astream | 吞吐量提升 3-5x |
| LLM 缓存 | SQLiteCache | 重复查询零延迟 |
| 并行节点 | Fan-out 图结构 | 独立任务同时执行 |
| 流式输出 | stream() | 首字延迟降低 |
| 模型选择 | 简单任务用小模型 | 成本降低 50%+ |
11.3 可观测性:LangSmith 集成
Agent 在生产环境跑起来后,你怎么知道它在"想什么"?出了问题怎么排查?答案是可观测性。
LangSmith —— LangChain 官方的追踪平台
bash
# 设置环境变量即可启用
export LANGCHAIN_TRACING_V2=true
export LANGCHAIN_API_KEY="your-api-key"
export LANGCHAIN_PROJECT="my-agent-project"启用后,LangGraph 执行的每一步都会被自动记录:
LangSmith 记录的信息:
┌─ Run: "用户问北京天气" ──────────────────────┐
│ │
│ ├─ Node: agent (1.2s) │
│ │ ├─ LLM Call: deepseek-chat (1.1s) │
│ │ │ Input: [HumanMessage("北京天气")] │
│ │ │ Output: tool_calls=[get_weather] │
│ │ │ Tokens: 45 in / 12 out │
│ │ │ Cost: $0.003 │
│ │ └─ │
│ │ │
│ ├─ Node: tools (0.5s) │
│ │ ├─ Tool: get_weather("北京") → "晴,25°C" │
│ │ └─ │
│ │ │
│ ├─ Node: agent (0.8s) │
│ │ ├─ LLM Call: deepseek-chat (0.7s) │
│ │ │ Output: "北京今天晴朗..." │
│ │ └─ │
│ │ │
│ Total: 2.5s | Tokens: 120 | Cost: $0.008 │
└───────────────────────────────────────────────┘关键监控指标
| 指标 | 说明 | 预警阈值 |
|---|---|---|
| 延迟 | 每次 invoke 的总耗时 | > 10s 预警 |
| Token 用量 | 每次调用的 Token 消耗 | 日均超预算预警 |
| 错误率 | 工具调用 / LLM 调用失败率 | > 5% 预警 |
| 循环次数 | ReAct 循环了多少轮 | > 5 轮预警 |
| 用户满意度 | 人工反馈或自评分 | < 3 分预警 |
生产必备:不开 LangSmith 就上生产 = 闭着眼开车。Agent 的行为不确定性远高于传统 API,可观测性是你唯一的"安全带"。
11.4 安全与限流
Agent 有调工具的能力——这意味着它可能执行危险操作。生产环境必须设置安全边界。
循环次数限制
防止 Agent 陷入无限循环(ReAct 不停调工具):
python
# 方法 1:compile 时设置 recursion_limit
app = graph.compile(
checkpointer=memory,
)
# invoke 时限制递归深度
result = app.invoke(
input_data,
config={"configurable": {"thread_id": "xxx"}, "recursion_limit": 25}
)
# → 超过 25 轮循环自动停止,抛出 GraphRecursionError工具权限控制
不是所有工具都应该随便调:
python
SAFE_TOOLS = {"search", "read_file", "calculator"}
DANGEROUS_TOOLS = {"delete_file", "send_email", "execute_sql"}
def gated_tool_node(state: MessagesState):
"""带权限控制的工具执行"""
last_msg = state["messages"][-1]
for tc in last_msg.tool_calls:
if tc["name"] in DANGEROUS_TOOLS:
# 危险工具 → 拒绝或走审批
raise ValueError(f"工具 '{tc['name']}' 未经授权")
# 安全工具正常执行
return ToolNode(tools).invoke(state)输入输出过滤
python
def sanitize_input(state: MessagesState):
"""过滤用户输入中的危险内容"""
user_msg = state["messages"][-1].content
# 检测 Prompt Injection
if any(kw in user_msg.lower() for kw in ["忽略指令", "ignore instructions"]):
return {"messages": [AIMessage(content="检测到潜在的提示注入,已拒绝处理。")]}
return state安全 Checklist
| 安全项 | 措施 |
|---|---|
| 循环次数 | recursion_limit=25 |
| 工具权限 | 白名单 / 黑名单 |
| 输入过滤 | Prompt Injection 检测 |
| 输出过滤 | 敏感信息脱敏 |
| 速率限制 | 每用户 QPS / 每日调用上限 |
| API Key | 环境变量 + 密钥管理服务 |
11.5 从原型到生产的 Checklist
以下是把 LangGraph Agent 从"能跑"变成"能用于生产"的完整检查清单:
🏗️ 架构
- [ ] State 设计合理,字段不多不少
- [ ] 循环有兜底(
recursion_limit/retry_count) - [ ] 关键操作有 Human-in-the-Loop
- [ ] 复杂逻辑拆分为子图
💾 持久化
- [ ] 使用 PostgresSaver(不是 MemorySaver!)
- [ ] thread_id 命名规范化
- [ ] 定期清理过期的 checkpoint 数据
⚡ 性能
- [ ] 节点函数使用 async
- [ ] 配置 LLM 缓存
- [ ] 面向用户的接口使用 stream
- [ ] 简单任务用小模型降本
🔒 安全
- [ ] 设置 recursion_limit
- [ ] 危险工具需人工审批
- [ ] 输入过滤(Prompt Injection / 敏感内容)
- [ ] 输出过滤(脱敏 / 内容审核)
- [ ] API Key 放在环境变量或 Secrets Manager
📊 可观测性
- [ ] 启用 LangSmith 追踪
- [ ] 设置延迟 / 错误率 / Token 用量告警
- [ ] 记录关键节点的日志
- [ ] 定期分析失败 case
🚀 部署
- [ ] 使用 LangGraph Platform 或容器化部署
- [ ] 配置健康检查 / 就绪探针
- [ ] 设置自动重启策略
- [ ] 负载均衡(多实例)
最终建议:不用一次做完。从最小可用版本开始,每次迭代补全一项。但 持久化、循环限制、可观测性 这三项应该在第一天就配好。
本章小结
| 知识点 | 要点 |
|---|---|
| LangGraph Platform | 自托管,Docker 部署,完全控制 |
| LangGraph Cloud | 全托管,零运维,按量付费 |
| 异步 | ainvoke / astream 提升吞吐 |
| LLM 缓存 | SQLiteCache 减少重复调用 |
| LangSmith | 环境变量启用,全链路追踪 |
| 安全限流 | recursion_limit + 工具白名单 + 输入过滤 |
| 生产 Checklist | 架构 / 持久化 / 性能 / 安全 / 可观测 / 部署 |
教程完结:恭喜你读完了全部 11 章!你已经掌握了 LangGraph 从入门到生产部署的全部核心知识。附录提供了 API 速查表和常见问题解答,方便日常查阅。