作者：PySuper | 来源：zhengxingtao.com | 更新日期：2026-12-01
关联阅读：[第02篇 vLLM 部署] [第06篇链路追踪] [第07篇语义缓存] [第19篇状态机基础] [第39篇持久化与人在回路] [第40篇子图与多Agent编排]
写 Demo 是快乐的——graph.invoke() 一跑，Agent 就活了。但把 Agent 推上生产线？那是另一个游戏。
你在本地跑得好好的 Agent，一上线就各种翻车：Checkpoint 写入超时、LLM 调用烧钱、工具重复执行、状态膨胀到数据库撑不住……这些问题，Demo 阶段根本遇不到。
这篇我们就聊最后一公里：怎么把 LangGraph Agent 从 Demo 变成能在生产环境稳定运行的系统。内容覆盖架构设计、Checkpoint 配置、流式输出、LLM 优化、错误处理、可观测性、成本控制，以及 9 个真实踩坑记录。
一、生产环境的挑战
1.1 Demo → 生产的 Gap 分析
先看一张对比表，感受一下落差：
表格
1.2 核心指标
生产环境的 Agent 系统，至少要达到这些硬指标：
表格
1.3 企业刚性要求 vs LangGraph 原生不足
LangGraph 是个优秀的编排框架，但它不是为生产部署直接设计的。以下是企业环境中的硬性要求，LangGraph 原生要么不支持，要么需要大量补丁：
表格
一句话总结：LangGraph 解决了编排问题，但运维问题得你自己扛。
二、部署架构设计
2.1 生产架构全景图
plaintext
┌─────────────────────────────────────────┐
│ 流量入口 (Nginx / ALB) │
│ SSL 终结 · 限流 · 负载均衡 · 健康检查 │
└─────────────────┬───────────────────────┘
│
┌─────────────────▼───────────────────────┐
│ API 层 (FastAPI) │
│ 请求校验 · 认证鉴权 · 限流 · 路由分发 │
└─────────────────┬───────────────────────┘
│
┌───────────────────────┼───────────────────────┐
│ │ │
┌─────────▼──────────┐ ┌────────▼─────────┐ ┌─────────▼──────────┐
│ LangGraph 编排层 │ │ LangGraph 编排层 │ │ LangGraph 编排层 │
│ (Worker Pod 1) │ │ (Worker Pod 2) │ │ (Worker Pod N) │
│ StateGraph 编译 │ │ StateGraph 编译 │ │ StateGraph 编译 │
│ 节点调度 · 状态机 │ │ 节点调度 · 状态机 │ │ 节点调度 · 状态机 │
└─────────┬──────────┘ └────────┬─────────┘ └─────────┬──────────┘
│ │ │
└───────────────────────┼───────────────────────┘
│
┌───────────────────────┼───────────────────────┐
│ │ │
┌─────────▼──────────┐ ┌────────▼─────────┐ ┌─────────▼──────────┐
│ 状态持久化 - PG │ │ 状态缓存 - Redis │ │ 监控告警层 │
│ Checkpoint 持久化 │ │ 热数据 · 语义缓存 │ │ Prometheus │
│ 连接池 · Pipeline │ │ 向量检索 · TTL │ │ Grafana + AlertMgr │
└────────────────────┘ └──────────────────┘ │ LangSmith / OTel │
└────────────────────┘
┌───────────────────────┼───────────────────────┐
│ │ │
┌─────────▼──────────┐ ┌────────▼─────────┐ ┌─────────▼──────────┐
│ LLM 接入层 │ │ 工具服务层 │ │ 对象存储 │
│ 模型路由 · 熔断 │ │ 搜索 · 数据库查询 │ │ S3 / MinIO │
│ 语义缓存 · 降级 │ │ API 调用 · 代码执行│ │ 大文件 · 文档内容 │
│ GPT-4o / Claude │ │ 幂等性 · 超时控制 │ │ Checkpoint 外挂 │
│ Qwen / Llama (小) │ │ │ │ │
└────────────────────┘ └────────────────────┘ └─────────────────────┘
2.2 Docker Compose 配置
开发和小规模部署用 Docker Compose 足够：
yaml
# docker-compose.yml
version: "3.9"
services:
# ---- API 网关 ----
api:
build: .
ports:
- "8000:8000"
environment:
- DATABASE_URL=postgresql://langgraph:secret@postgres:5432/langgraph
- REDIS_URL=redis://redis:6379/0
- OPENAI_API_KEY=${OPENAI_API_KEY}
- LANGSMITH_API_KEY=${LANGSMITH_API_KEY}
- LANGSMITH_TRACING=true
- LANGSMITH_PROJECT=production-agent
depends_on:
postgres:
condition: service_healthy
redis:
condition: service_healthy
restart: unless-stopped
deploy:
resources:
limits:
memory: 2G
cpus: "2.0"
# ---- PostgreSQL ----
postgres:
image: postgres:16-alpine
environment:
POSTGRES_USER: langgraph
POSTGRES_PASSWORD: secret
POSTGRES_DB: langgraph
volumes:
- pgdata:/var/lib/postgresql/data
- ./init.sql:/docker-entrypoint-initdb.d/init.sql
healthcheck:
test: ["CMD-SHELL", "pg_isready -U langgraph"]
interval: 5s
timeout: 3s
retries: 5
restart: unless-stopped
command: >
postgres
-c shared_buffers=256MB
-c work_mem=16MB
-c max_connections=200
-c checkpoint_completion_target=0.9
# ---- Redis ----
redis:
image: redis:7-alpine
command: >
redis-server
--maxmemory 512mb
--maxmemory-policy allkeys-lru
--save 60 1000
--appendonly yes
volumes:
- redisdata:/data
healthcheck:
test: ["CMD", "redis-cli", "ping"]
interval: 5s
timeout: 3s
retries: 5
restart: unless-stopped
# ---- 监控 ----
prometheus:
image: prom/prometheus:v2.52.0
volumes:
- ./prometheus.yml:/etc/prometheus/prometheus.yml
ports:
- "9090:9090"
restart: unless-stopped
grafana:
image: grafana/grafana:11.0.0
ports:
- "3000:3000"
environment:
- GF_SECURITY_ADMIN_PASSWORD=admin
volumes:
- grafanadata:/var/lib/grafana
restart: unless-stopped
volumes:
pgdata:
redisdata:
grafanadata:
2.3 K8s 部署配置要点
生产环境上 K8s，几个关键点要注意：
yaml
# k8s/deployment.yaml — 关键配置片段
apiVersion: apps/v1
kind: Deployment
metadata:
name: langgraph-agent
spec:
replicas: 3 # 至少 3 副本，保证滚动更新不停服
strategy:
type: RollingUpdate
rollingUpdate:
maxUnavailable: 1
maxSurge: 1
template:
spec:
# 优雅关闭：给正在运行的图 30s 完成
terminationGracePeriodSeconds: 30
containers:
- name: agent
image: langgraph-agent:latest
ports:
- containerPort: 8000
resources:
requests:
memory: "1Gi"
cpu: "500m"
limits:
memory: "2Gi"
cpu: "2000m"
env:
- name: DATABASE_URL
valueFrom:
secretKeyRef:
name: langgraph-secrets
key: database-url
# 就绪探针：确保 DB 连接已建好
readinessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 10
periodSeconds: 10
# 存活探针：检测死锁
livenessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 30
periodSeconds: 30
# 生命周期钩子：关闭前排空连接
lifecycle:
preStop:
exec:
command: ["sh", "-c", "sleep 5"]
K8s 部署特别注意：
HPA 自动扩缩容：基于 CPU 和自定义指标（如活跃线程数）配置 HPA
PDB 中断预算：确保滚动更新时始终有足够副本在线
Affinity 反亲和：副本分布到不同节点，避免单节点故障
Init Container：启动前等 PostgreSQL 和 Redis 就绪
三、Checkpoint 生产配置
Checkpoint 是 LangGraph 生产化最关键的一环。选错存储后端，轻则性能拉胯，重则数据丢失。
3.1 PostgreSQL Checkpoint 详细配置
PostgreSQL 是生产环境的推荐选择。LangGraph 0.2 把 langgraph-checkpoint-postgres 单独拆包，做了大量优化：
Write 侧：使用 PostgreSQL Pipeline 模式减少数据库往返
增量存储：每个 Channel 值单独存储和版本化，新 Checkpoint 只存变化的字段
Read 侧：使用游标高效分页获取长线程历史
异步连接池配置
python
import os
import asyncio
from contextlib import asynccontextmanager
from psycopg_pool import AsyncConnectionPool
from psycopg.rows import dict_row
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
# ──────────────────────────────────────────
# 生产级 PostgreSQL Checkpoint 配置
# ──────────────────────────────────────────
DB_URI = os.environ["DATABASE_URL"]
# 连接池参数
pool_config = {
"conninfo": DB_URI,
"max_size": 20, # 最大连接数，按并发量调整
"min_size": 5, # 最小空闲连接，减少冷启动
"max_idle": 300.0, # 空闲连接最大存活时间（秒）
"max_lifetime": 1800.0, # 连接最大生命周期（秒），防止长连接问题
"kwargs": {
"autocommit": True,
"prepare_threshold": 0, # 禁用预备语句缓存，避免连接池复用问题
"row_factory": dict_row,
},
}
@asynccontextmanager
async def get_checkpointer():
"""生产级 Checkpointer 上下文管理器"""
async with AsyncConnectionPool(**pool_config) as pool:
checkpointer = AsyncPostgresSaver(pool)
# 首次使用必须调用 setup()，创建表和索引
await checkpointer.setup()
yield checkpointer
# ──────────────────────────────────────────
# 带重试的 Checkpointer（推荐）
# ──────────────────────────────────────────
import logging
from typing import Any, AsyncIterator
import psycopg
from langgraph.checkpoint.base import CheckpointTuple
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
from psycopg_pool import AsyncConnectionPool
logger = logging.getLogger(__name__)
# 连接错误模式，需要重试
CONNECTION_ERROR_PATTERNS = [
"ssl error", "ssl connection has been closed",
"connection lost", "connection closed", "connection reset",
"server closed the connection unexpectedly",
"connection pool exhausted", "connection timeout",
]
class ResilientPostgresCheckpointer(AsyncPostgresSaver):
"""带连接恢复能力的 PostgreSQL Checkpointer"""
def __init__(self, conn_pool: AsyncConnectionPool, *, max_retries: int = 3):
super().__init__(conn_pool)
self.max_retries = max_retries
def _is_connection_error(self, exc: Exception) -> bool:
if isinstance(exc, (psycopg.OperationalError, psycopg.InterfaceError)):
error_str = str(exc).lower()
return any(p in error_str for p in CONNECTION_ERROR_PATTERNS)
if isinstance(exc, (ConnectionError, OSError, TimeoutError)):
return True
return False
async def _with_retry(self, operation_name: str, func, *args, **kwargs) -> Any:
last_exc = None
for attempt in range(self.max_retries + 1):
try:
result = func(*args, **kwargs)
if asyncio.iscoroutine(result):
return await result
return result
except Exception as e:
last_exc = e
if self._is_connection_error(e) and attempt < self.max_retries:
wait_time = 0.5 * (2 ** attempt) # 指数退避
logger.warning(
f"[{operation_name}] 连接错误，{wait_time:.1f}s 后重试 "
f"(attempt {attempt + 1}/{self.max_retries}): {e}"
)
await asyncio.sleep(wait_time)
else:
raise
raise last_exc
async def aput(self, config, checkpoint, metadata, new_versions):
return await self._with_retry("aput", super().aput, config, checkpoint, metadata, new_versions)
async def aget_tuple(self, config):
return await self._with_retry("aget_tuple", super().aget_tuple, config)
Checkpoint TTL 过期清理
python
# 定期清理过期 Checkpoint（Cron Job 或后台任务）
async def cleanup_expired_checkpoints(
pool: AsyncConnectionPool,
retention_days: int = 30,
batch_size: int = 1000,
):
"""清理超过保留期的 Checkpoint 数据"""
async with pool.connection() as conn:
# 删除过期的 checkpoint 记录
result = await conn.execute(
"""
DELETE FROM checkpoints
WHERE thread_id IN (
SELECT DISTINCT thread_id FROM checkpoints
WHERE created_at < NOW() - INTERVAL '%s days'
LIMIT %s
)
""",
(retention_days, batch_size),
)
deleted = result.rowcount if result else 0
logger.info(f"清理过期 Checkpoint: 删除 {deleted} 条记录")
return deleted
3.2 Redis Checkpoint 配置
Redis 适用于高频读写、临时状态的场景——比如实时对话中的热数据缓存。
python
import json
import redis.asyncio as aioredis
from langgraph.checkpoint.base import BaseCheckpointSaver
class RedisCheckpointCache:
"""Redis 热数据缓存层，搭配 PostgreSQL 使用"""
def __init__(
self,
redis_url: str = "redis://localhost:6379/0",
ttl: int = 3600, # 默认 1 小时过期
):
self.redis = aioredis.from_url(redis_url, decode_responses=True)
self.ttl = ttl
def _key(self, thread_id: str) -> str:
return f"lg:checkpoint:{thread_id}"
async def get(self, thread_id: str) -> dict | None:
"""读取缓存"""
data = await self.redis.get(self._key(thread_id))
if data:
return json.loads(data)
return None
async def set(self, thread_id: str, checkpoint: dict):
"""写入缓存，设置 TTL"""
await self.redis.set(
self._key(thread_id),
json.dumps(checkpoint, ensure_ascii=False),
ex=self.ttl,
)
async def delete(self, thread_id: str):
"""删除缓存"""
await self.redis.delete(self._key(thread_id))
Redis vs PostgreSQL 选型：
表格
3.3 Checkpoint 存储优化
Checkpoint 膨胀是生产环境的隐形杀手。根据 LangGraph 官方数据，默认的全量快照模式下，存储增长是 O(N²) ——一个 200 轮的对话会产生 5.3GB 的 Checkpoint 数据。
优化策略一：State 只存必要字段
python
from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
# ❌ 不好的设计：把所有东西都塞进 State
class BadState(TypedDict):
messages: Annotated[list, add_messages]
full_document_content: str # 几十 KB 的文档全文
search_results: list[dict] # 搜索结果原文
tool_response_raw: str # 工具原始返回
# ✅ 好的设计：只存引用，大对象放外部
class GoodState(TypedDict):
messages: Annotated[list, add_messages]
document_id: str # 文档 ID，内容存 S3
search_result_ids: list[str] # 结果 ID，内容存 Redis
tool_response_summary: str # 只存摘要
优化策略二：大对象只存 ID
python
import boto3
import json
s3_client = boto3.client("s3")
BUCKET = "agent-checkpoint-assets"
async def store_large_object(data: dict, thread_id: str, key: str) -> str:
"""将大对象存入 S3，返回引用 ID"""
object_key = f"{thread_id}/{key}"
s3_client.put_object(
Bucket=BUCKET,
Key=object_key,
Body=json.dumps(data, ensure_ascii=False).encode("utf-8"),
)
return object_key # 只存这个 key 到 State
async def load_large_object(object_key: str) -> dict:
"""从 S3 加载大对象"""
response = s3_client.get_object(Bucket=BUCKET, Key=object_key)
return json.loads(response["Body"].read().decode("utf-8"))
优化策略三：增量 Channel（langgraph 1.2+）
如果你用 langgraph 1.2+，可以使用 DeltaChannel，它只存每步的变化量，存储从 O(N²) 降到 O(N)：
python
# langgraph 1.2 新增的 DeltaChannel
from langgraph.channels.delta import DeltaChannel
from langgraph.graph.message import add_messages
from typing import TypedDict, Annotated
class State(TypedDict):
# 之前：每步存完整消息列表，O(N²)
# messages: Annotated[list, add_messages]
# 现在：只存增量，O(N)
messages: Annotated[list, DeltaChannel(add_messages)]
实测数据：200 轮对话，全量快照 5.3GB → DeltaChannel 129MB，降幅 40 倍。（据 Delta Channels: Evolving our Runtime for Long-Running Agents）
四、流式输出（Streaming）
用户等不了 30 秒才看到第一个字。流式输出是 Agent 产品化的刚需。
4.1 四种流式模式
plaintext
┌─────────────────────────────────────────────────────────────────┐
│ LangGraph 流式模式 │
├────────────┬────────────────────────────────────────────────────┤
│ values │ 每步输出完整 State 快照 │
│ │ 适合：调试、全量状态追踪 │
│ │ 开销：大（每步完整序列化） │
├────────────┼────────────────────────────────────────────────────┤
│ updates │ 每步输出增量更新（只含当前节点的返回值） │
│ │ 适合：进度展示、节点级日志 │
│ │ 开销：小（只传变化部分） │
├────────────┼────────────────────────────────────────────────────┤
│ messages │ 逐 Token 输出 LLM 响应 │
│ │ 适合：打字机效果、实时对话 │
│ │ 输出：2-tuple (token_chunk, metadata) │
├────────────┼────────────────────────────────────────────────────┤
│ custom │ 自定义进度推送 │
│ │ 适合：工具执行进度、多阶段任务状态 │
│ │ 通过 get_stream_writer() 在节点内推送 │
└────────────┴────────────────────────────────────────────────────┘
4.2 与 FastAPI 集成的 SSE 推送
完整的流式 API 实现：
python
import json
import asyncio
from typing import AsyncGenerator
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from langgraph.config import get_stream_writer
from langgraph.graph import StateGraph, START, END, MessagesState
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
# ──────────────────────────────────────────
# 工具定义
# ──────────────────────────────────────────
@tool
def search_knowledge_base(query: str) -> str:
"""搜索知识库"""
# 模拟搜索
return f"关于「{query}」的搜索结果：..."
# ──────────────────────────────────────────
# Graph 构建
# ──────────────────────────────────────────
model = ChatOpenAI(model="gpt-4o-mini", streaming=True)
tools = [search_knowledge_base]
def build_graph(checkpointer: AsyncPostgresSaver):
graph = create_react_agent(model, tools, checkpointer=checkpointer)
return graph
# ──────────────────────────────────────────
# FastAPI 应用
# ──────────────────────────────────────────
app = FastAPI(title="LangGraph Agent API")
class ChatRequest(BaseModel):
message: str
thread_id: str = "default"
@app.post("/chat/stream")
async def chat_stream(request: ChatRequest):
"""SSE 流式接口 — 支持 messages + updates 双模式"""
config = {"configurable": {"thread_id": request.thread_id}}
async def event_generator() -> AsyncGenerator[str, None]:
async with get_checkpointer() as checkpointer:
graph = build_graph(checkpointer)
# 同时获取 Token 流和节点更新
async for event in graph.astream(
{"messages": [("user", request.message)]},
config=config,
stream_mode=["messages", "updates"],
):
# stream_mode=["messages", "updates"] 返回 (mode, data) 元组
mode, data = event
if mode == "messages":
token, metadata = data
if token.content: # 只推送有内容的 chunk
yield f"data: {json.dumps({
'type': 'token',
'content': token.content,
'node': metadata.get('langgraph_node', ''),
}, ensure_ascii=False)}\n\n"
elif mode == "updates":
# 节点级更新
yield f"data: {json.dumps({
'type': 'update',
'data': str(data),
}, ensure_ascii=False)}\n\n"
# 结束标记
yield f"data: {json.dumps({'type': 'done'})}\n\n"
return StreamingResponse(
event_generator(),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"X-Accel-Buffering": "no", # Nginx 不缓冲
},
)
# ──────────────────────────────────────────
# Custom 流式模式（进度推送）
# ──────────────────────────────────────────
from langgraph.config import get_stream_writer
from typing import TypedDict
class ProgressState(TypedDict):
query: str
answer: str
def research_node(state: ProgressState):
"""研究节点 — 带自定义进度推送"""
writer = get_stream_writer()
writer({"progress": "正在检索相关文档...", "step": 1})
# ... 执行检索 ...
writer({"progress": "正在分析文档内容...", "step": 2})
# ... 执行分析 ...
writer({"progress": "正在生成回答...", "step": 3})
# ... 生成回答 ...
return {"answer": "分析完成"}
# 使用 custom 模式接收进度
@app.post("/chat/progress")
async def chat_with_progress(request: ChatRequest):
async def event_generator():
# ... graph 构建省略 ...
async for chunk in graph.astream(
{"query": request.message},
stream_mode="custom",
):
yield f"data: {json.dumps(chunk, ensure_ascii=False)}\n\n"
yield f"data: {json.dumps({'type': 'done'})}\n\n"
return StreamingResponse(event_generator(), media_type="text/event-stream")
4.3 Nginx SSE 配置
SSE 流式推送在 Nginx 层需要禁用缓冲：
nginx
location /chat/ {
proxy_pass http://langgraph_backend;
proxy_http_version 1.1;
# 禁用所有缓冲，SSE 必须实时推送
proxy_buffering off;
proxy_cache off;
# 不设超时，让长连接保持
proxy_read_timeout 300s;
# 必须的 SSE 头
proxy_set_header Connection '';
proxy_set_header X-Accel-Buffering no;
}
五、LLM 调用优化
LLM 调用占 Agent 系统成本的 70%+。优化好这一层，成本直接砍半。
5.1 语义缓存（Semantic Cache）
精确匹配缓存（hash key）对 LLM 没用——用户问"怎么退款"和"退款流程是什么"，hash 完全不同，但意思一样。
语义缓存的思路：把请求转成向量，跟缓存里的向量做余弦相似度比较，超过阈值就命中。
plaintext
┌──────────────────────────────────────────────────────────────┐
│ 语义缓存流程 │
│ │
│ 用户请求 ──→ Embedding ──→ Redis 向量检索 ──→ 相似度 > 0.9? │
│ │ │
│ ┌────┴────┐ │
│ │ Yes │ No │
│ ▼ ▼ │
│ 返回缓存结果 调用 LLM │
│ │ │
│ 存入缓存 ◄──────┘ │
└──────────────────────────────────────────────────────────────┘
Redis 向量检索实现
python
import json
import hashlib
import time
import numpy as np
import openai
import redis.asyncio as aioredis
class SemanticCache:
"""基于 Redis 向量检索的语义缓存"""
def __init__(
self,
redis_url: str = "redis://localhost:6379/0",
similarity_threshold: float = 0.92,
ttl: int = 3600,
embedding_model: str = "text-embedding-3-small",
):
self.redis = aioredis.from_url(redis_url, decode_responses=True)
self.threshold = similarity_threshold
self.ttl = ttl
self.embedding_model = embedding_model
self.client = openai.AsyncOpenAI()
async def _get_embedding(self, text: str) -> list[float]:
"""获取文本向量"""
response = await self.client.embeddings.create(
input=text,
model=self.embedding_model,
)
return response.data[0].embedding
@staticmethod
def _cosine_similarity(a: list[float], b: list[float]) -> float:
"""余弦相似度"""
a_np, b_np = np.array(a), np.array(b)
return float(np.dot(a_np, b_np) / (np.linalg.norm(a_np) * np.linalg.norm(b_np)))
async def get(self, query: str) -> str | None:
"""查询缓存"""
query_embedding = await self._get_embedding(query)
# 从 Redis 获取所有缓存的 key
keys = []
async for key in self.redis.scan_iter("sem_cache:*"):
keys.append(key)
if not keys:
return None
# 逐个比较相似度（生产环境应使用 Redis 的向量搜索功能）
best_score = 0.0
best_result = None
for key in keys:
cached = await self.redis.hgetall(key)
if not cached:
continue
cached_embedding = json.loads(cached["embedding"])
score = self._cosine_similarity(query_embedding, cached_embedding)
if score > best_score and score >= self.threshold:
best_score = score
best_result = cached["response"]
return best_result
async def set(self, query: str, response: str):
"""写入缓存"""
embedding = await self._get_embedding(query)
cache_key = f"sem_cache:{hashlib.md5(query.encode()).hexdigest()}"
await self.redis.hset(cache_key, mapping={
"query": query,
"response": response,
"embedding": json.dumps(embedding),
"created_at": str(int(time.time())),
})
await self.redis.expire(cache_key, self.ttl)
# ──────────────────────────────────────────
# 集成到 LangGraph 节点
# ──────────────────────────────────────────
cache = SemanticCache(
similarity_threshold=0.92,
ttl=3600, # 1 小时过期
)
async def cached_llm_node(state):
"""带语义缓存的 LLM 调用节点"""
query = state["messages"][-1].content
# 1. 查缓存
cached = await cache.get(query)
if cached:
return {"messages": [("assistant", f"[缓存命中] {cached}")]}
# 2. 缓存未命中，调用 LLM
response = await model.ainvoke(state["messages"])
# 3. 写入缓存
await cache.set(query, response.content)
return {"messages": [response]}
实战经验：客服问答类场景，语义缓存命中率可达 40%-60%，直接砍掉近半 LLM 调用。FAQ 场景更高，能达到 90%。
5.2 模型路由
不是每个请求都需要 GPT-4o。简单分类用小模型，复杂推理才上大模型：
plaintext
┌──────────────────────────────────────────────────────────┐
│ 模型路由决策树 │
│ │
│ 用户请求 │
│ │ │
│ ┌─────▼─────┐ │
│ │ 意图分类 │ (小模型 / 规则) │
│ └─────┬─────┘ │
│ │ │
│ ┌──────────────┼──────────────┐ │
│ │ │ │ │
│ ┌────▼────┐ ┌────▼────┐ ┌────▼────┐ │
│ │ 简单任务 │ │ 中等任务 │ │ 复杂任务 │ │
│ │ 分类 │ │ 总结 │ │ 推理 │ │
│ │ 提取 │ │ 润色 │ │ 代码 │ │
│ │ 格式化 │ │ 翻译 │ │ 多步规划│ │
│ └────┬────┘ └────┬────┘ └────┬────┘ │
│ │ │ │ │
│ ┌────▼────┐ ┌────▼────┐ ┌────▼────┐ │
│ │ Qwen-7B │ │ GPT-4o │ │ Claude │ │
│ │Llama3-8B│ │ mini │ │ Opus │ │
│ │ $0.01/1K│ │ $0.15/1K│ │ $15/1K │ │
│ └─────────┘ └─────────┘ └─────────┘ │
│ │
│ 成本占比：60%请求 → 小模型 → 仅占 5% 总成本 │
└──────────────────────────────────────────────────────────┘
python
from enum import Enum
from langchain_openai import ChatOpenAI
from langchain_community.chat_models import ChatOllama # 本地模型
class TaskComplexity(Enum):
SIMPLE = "simple" # 分类、提取、格式化
MODERATE = "moderate" # 总结、翻译、润色
COMPLEX = "complex" # 推理、代码、多步规划
# 模型池
MODELS = {
TaskComplexity.SIMPLE: ChatOpenAI(
model="qwen2.5-7b-instruct", # 或本地 Ollama
base_url="http://localhost:11434/v1",
temperature=0,
max_tokens=1000,
timeout=10,
),
TaskComplexity.MODERATE: ChatOpenAI(
model="gpt-4o-mini",
temperature=0,
max_tokens=2000,
timeout=30,
),
TaskComplexity.COMPLEX: ChatOpenAI(
model="gpt-4o",
temperature=0,
max_tokens=4000,
timeout=60,
),
}
# 路由规则
ROUTING_KEYWORDS = {
TaskComplexity.SIMPLE: [
"分类", "提取", "格式化", "标签", "判断是否",
"classify", "extract", "format", "label",
],
TaskComplexity.COMPLEX: [
"分析", "推理", "编写代码", "规划", "设计",
"analyze", "reason", "code", "plan", "design",
],
}
def classify_complexity(query: str) -> TaskComplexity:
"""根据关键词初步分类任务复杂度"""
query_lower = query.lower()
for keyword in ROUTING_KEYWORDS.get(TaskComplexity.COMPLEX, []):
if keyword in query_lower:
return TaskComplexity.COMPLEX
for keyword in ROUTING_KEYWORDS.get(TaskComplexity.SIMPLE, []):
if keyword in query_lower:
return TaskComplexity.SIMPLE
return TaskComplexity.MODERATE # 默认中等
class ModelRouter:
"""模型路由器 — 带降级策略"""
def __init__(self, models: dict, fallback_model=None):
self.models = models
self.fallback_model = fallback_model or models[TaskComplexity.MODERATE]
async def route(self, query: str):
"""路由到合适的模型"""
complexity = classify_complexity(query)
model = self.models.get(complexity, self.fallback_model)
return model, complexity
async def invoke_with_fallback(self, query: str, messages: list):
"""调用模型，失败自动降级"""
model, complexity = await self.route(query)
try:
return await model.ainvoke(messages)
except Exception as e:
# 复杂任务失败 → 降级到中等模型
if complexity == TaskComplexity.COMPLEX:
logger.warning(f"复杂模型调用失败，降级到中等模型: {e}")
return await self.models[TaskComplexity.MODERATE].ainvoke(messages)
# 其他失败 → 用兜底模型
logger.warning(f"模型调用失败，使用兜底: {e}")
return await self.fallback_model.ainvoke(messages)
# ──────────────────────────────────────────
# 集成到 LangGraph
# ──────────────────────────────────────────
router = ModelRouter(MODELS)
async def routed_llm_node(state):
"""带模型路由的 LLM 节点"""
query = state["messages"][-1].content
model, complexity = await router.route(query)
# 按复杂度限制 Token 上限
max_tokens = {
TaskComplexity.SIMPLE: 500,
TaskComplexity.MODERATE: 2000,
TaskComplexity.COMPLEX: 4000,
}[complexity]
response = await model.ainvoke(
state["messages"],
max_tokens=max_tokens,
)
return {"messages": [response]}
成本对比：假设 1000 次请求，60% 简单 + 30% 中等 + 10% 复杂。全用 GPT-4o 月费7,650，降幅 83%。
5.3 上下文压缩（LLMLingua）
长上下文是 Token 消耗的大头。微软的 LLMLingua 可以压缩 Prompt，最高 20 倍压缩，性能损失极小：
python
from llmlingua import PromptCompressor
# 初始化压缩器
compressor = PromptCompressor(
model_name="microsoft/llmlingua-2-xlm-roberta-large-meetingbank",
use_llmlingua2=True,
)
async def compress_context(messages: list[dict], target_token: int = 2000) -> list[dict]:
"""压缩对话上下文"""
# 只压缩历史消息，保留最近 2 轮原文
if len(messages) <= 4:
return messages
# 提取历史消息文本
history_text = "\n".join(
f"{m.get('role', 'user')}: {m.get('content', '')}"
for m in messages[:-4] # 排除最近 2 轮
)
# 压缩
compressed = compressor.compress_prompt(
history_text,
rate=0.5, # 50% 压缩率
force_tokens=["!", "?", "."], # 保留标点
)
# 重建消息列表
compressed_messages = [
{"role": "system", "content": f"[历史对话摘要] {compressed['compressed_prompt']}"}
] + messages[-4:] # 最近 2 轮保持原文
return compressed_messages
六、错误处理与重试
6.1 节点级重试：tenacity 装饰器
python
import openai
from tenacity import (
retry,
stop_after_attempt,
wait_exponential,
wait_fixed,
retry_if_exception_type,
before_sleep_log,
)
import logging
logger = logging.getLogger(__name__)
# ──────────────────────────────────────────
# 自定义错误类型
# ──────────────────────────────────────────
class AgentError(Exception):
"""Agent 错误基类"""
code: int = 0
retryable: bool = False
class LLMTimeoutError(AgentError):
code = 1001
retryable = True
class LLMRateLimitError(AgentError):
code = 1002
retryable = True
class ToolTimeoutError(AgentError):
code = 2001
retryable = True
class InvalidInputError(AgentError):
code = 3001
retryable = False
class DeadLoopError(AgentError):
code = 4001
retryable = False
# ──────────────────────────────────────────
# LLM 调用重试策略
# ──────────────────────────────────────────
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=2, max=10),
retry=retry_if_exception_type((LLMRateLimitError, LLMTimeoutError)),
before_sleep=before_sleep_log(logger, logging.WARNING),
reraise=True,
)
async def llm_call_with_retry(messages, **kwargs):
"""LLM 调用 — 带指数退避重试"""
try:
return await model.ainvoke(messages, **kwargs)
except openai.APITimeoutError as e:
raise LLMTimeoutError("LLM 调用超时") from e
except openai.RateLimitError as e:
raise LLMRateLimitError("LLM 调用被限流") from e
except openai.APIError as e:
raise AgentError(f"LLM API 错误: {e}") from e
# ──────────────────────────────────────────
# 工具调用重试策略
# ──────────────────────────────────────────
@retry(
stop=stop_after_attempt(2),
wait=wait_fixed(1),
retry=retry_if_exception_type(ToolTimeoutError),
reraise=True,
)
async def tool_call_with_retry(tool_func, *args, **kwargs):
"""工具调用 — 带固定间隔重试"""
try:
return await tool_func.ainvoke(*args, **kwargs)
except TimeoutError as e:
raise ToolTimeoutError("工具调用超时") from e
except Exception as e:
raise AgentError(f"工具调用错误: {e}") from e
6.2 工具幂等性：request_id 去重
python
import hashlib
import redis.asyncio as aioredis
redis_client = aioredis.from_url("redis://localhost:6379/1")
def make_request_id(func_name: str, args: tuple, kwargs: dict) -> str:
"""生成请求幂等 ID"""
raw = f"{func_name}:{args}:{sorted(kwargs.items())}"
return hashlib.md5(raw.encode()).hexdigest()
async def idempotent_tool(func):
"""工具幂等性装饰器"""
async def wrapper(*args, **kwargs):
request_id = make_request_id(func.__name__, args, kwargs)
# 检查是否已执行
cached = await redis_client.get(f"idempotent:{request_id}")
if cached:
return json.loads(cached)
# 执行并缓存结果
result = await func(*args, **kwargs)
await redis_client.set(
f"idempotent:{request_id}",
json.dumps(result, ensure_ascii=False),
ex=86400, # 24 小时过期
)
return result
return wrapper
# 使用示例
@idempotent_tool
async def create_order(product_id: str, quantity: int) -> dict:
"""创建订单 — 幂等性保证"""
# 相同参数不会重复创建
return {"order_id": "ORD-123", "status": "created"}
6.3 死循环检测
python
from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated
import operator
class SafeState(TypedDict):
messages: Annotated[list, add_messages]
step_count: int
last_response_hash: str # 上一步响应的 hash
MAX_STEPS = 25 # 全局递归上限
SIMILARITY_THRESHOLD = 0.9 # 连续相似响应阈值
def should_continue(state: SafeState) -> str:
"""死循环检测 + 路由决策"""
# 检测 1：步数上限
if state["step_count"] >= MAX_STEPS:
return "end"
# 检测 2：连续相似响应（Agent 在反复说一样的话）
current_hash = hashlib.md5(
str(state["messages"][-1].content).encode()
).hexdigest()
if current_hash == state.get("last_response_hash", ""):
# 连续两次相同响应，强制终止
return "end"
# 正常路由
last_msg = state["messages"][-1]
if getattr(last_msg, "tool_calls", None):
return "tools"
return "end"
6.4 优雅降级：本地模型兜底
python
from langchain_community.chat_models import ChatOllama
# 本地兜底模型
FALLBACK_MODEL = ChatOllama(
model="qwen2.5:7b",
base_url="http://localhost:11434",
temperature=0,
)
async def resilient_llm_node(state):
"""带降级的 LLM 节点"""
try:
response = await llm_call_with_retry(state["messages"])
return {"messages": [response]}
except AgentError as e:
logger.warning(f"云端模型全部失败，启用本地兜底: {e}")
try:
response = await FALLBACK_MODEL.ainvoke(state["messages"])
return {"messages": [response]}
except Exception as fallback_err:
logger.error(f"本地兜底也失败了: {fallback_err}")
return {
"messages": [("assistant", "抱歉，服务暂时不可用，请稍后再试。")],
}
七、可观测性
出了问题你得能查到。可观测性三驾马车：Metrics + Logs + Traces。
plaintext
┌─────────────────────────────────────────────────────────────┐
│ 可观测性三驾马车 │
│ │
│ ┌─────────────┐ ┌─────────────┐ ┌──────────────────┐ │
│ │ Metrics │ │ Logs │ │ Traces │ │
│ │ Prometheus │ │ ELK/Loki │ │ LangSmith / OTel │ │
│ │ │ │ │ │ │ │
│ │ · QPS │ │ · 结构化日志 │ │ · 节点耗时 │ │
│ │ · 延迟 P99 │ │ · 请求追踪 │ │ · LLM 调用链 │ │
│ │ · 错误率 │ │ · 错误堆栈 │ │ · Token 消耗 │ │
│ │ · Token 消耗│ │ · 业务埋点 │ │ · Checkpoint 延迟│ │
│ └─────────────┘ └─────────────┘ └──────────────────┘ │
│ │
│ ┌──────────────────────┐ │
│ │ Grafana Dashboard │ │
│ │ 统一可视化 + 告警 │ │
│ └──────────────────────┘ │
└─────────────────────────────────────────────────────────────┘
7.1 LangSmith 集成
LangSmith 是 LangChain 官方的可观测性平台，集成最简单：
python
import os
# 环境变量配置（零代码侵入）
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "ls-..."
os.environ["LANGSMITH_PROJECT"] = "production-agent"
# 就这么简单，LangGraph 自动上报 Traces
graph = build_graph(checkpointer)
result = await graph.ainvoke(
{"messages": [("user", "hello")]},
config={"configurable": {"thread_id": "t1"}},
)
# LangSmith 自动记录：节点耗时、LLM 调用、Token 消耗
7.2 OpenTelemetry 集成（开源方案）
不想用 LangSmith？OpenTelemetry 是开源替代：
python
# 安装依赖
# pip install opentelemetry-sdk opentelemetry-exporter-otlp
# pip install opentelemetry-instrumentation-langchain
import os
from opentelemetry import trace as trace_api
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.instrumentation.langchain import LangchainInstrumentor
# ──────────────────────────────────────────
# 1. 配置 OTel Provider
# ──────────────────────────────────────────
provider = TracerProvider()
provider.add_span_processor(
BatchSpanProcessor(
OTLPSpanExporter(
endpoint=os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4317"),
)
)
)
trace_api.set_tracer_provider(provider)
# ──────────────────────────────────────────
# 2. 自动注入 LangChain/LangGraph
# ──────────────────────────────────────────
LangchainInstrumentor().instrument()
# 之后所有 graph.invoke / graph.astream 都会自动上报 Trace
7.3 关键监控指标
python
import time
import logging
from dataclasses import dataclass, field
from typing import Callable
from langchain_core.runnables import RunnableConfig
logger = logging.getLogger("langgraph.metrics")
# ──────────────────────────────────────────
# 指标收集器（对接 Prometheus）
# ──────────────────────────────────────────
from prometheus_client import Counter, Histogram, Gauge
# 请求计数
REQUEST_TOTAL = Counter(
"langgraph_request_total",
"Total requests",
["status", "node"],
)
# 节点执行耗时
NODE_DURATION = Histogram(
"langgraph_node_duration_seconds",
"Node execution duration",
["node_name"],
buckets=[0.1, 0.5, 1, 2, 5, 10, 30, 60],
)
# LLM Token 消耗
LLM_TOKENS = Counter(
"langgraph_llm_tokens_total",
"LLM token consumption",
["model", "token_type"], # token_type: input / output
)
# Checkpoint 读写延迟
CHECKPOINT_DURATION = Histogram(
"langgraph_checkpoint_duration_seconds",
"Checkpoint read/write duration",
["operation"], # operation: read / write
buckets=[0.001, 0.005, 0.01, 0.05, 0.1, 0.5],
)
# 活跃线程数
ACTIVE_THREADS = Gauge(
"langgraph_active_threads",
"Number of active threads",
)
# 错误计数
ERROR_TOTAL = Counter(
"langgraph_error_total",
"Total errors",
["error_type", "node"],
)
# ──────────────────────────────────────────
# 监控装饰器
# ──────────────────────────────────────────
def monitored_node(node_name: str):
"""节点监控装饰器"""
def decorator(func: Callable):
async def wrapper(state, config: RunnableConfig = None):
start = time.monotonic()
try:
result = await func(state, config)
duration = time.monotonic() - start
NODE_DURATION.labels(node_name=node_name).observe(duration)
REQUEST_TOTAL.labels(status="success", node=node_name).inc()
return result
except Exception as e:
duration = time.monotonic() - start
NODE_DURATION.labels(node_name=node_name).observe(duration)
REQUEST_TOTAL.labels(status="error", node=node_name).inc()
ERROR_TOTAL.labels(
error_type=type(e).__name__,
node=node_name,
).inc()
raise
return wrapper
return decorator
# 使用示例
@monitored_node("llm_call")
async def llm_node(state):
response = await model.ainvoke(state["messages"])
# 记录 Token 消耗
if hasattr(response, "usage_metadata"):
LLM_TOKENS.labels(model="gpt-4o", token_type="input").inc(
response.usage_metadata.get("input_tokens", 0)
)
LLM_TOKENS.labels(model="gpt-4o", token_type="output").inc(
response.usage_metadata.get("output_tokens", 0)
)
return {"messages": [response]}
7.4 告警规则
yaml
# prometheus/alert_rules.yml
groups:
- name: langgraph_alerts
rules:
# 节点执行超时
- alert: NodeExecutionSlow
expr: histogram_quantile(0.99, langgraph_node_duration_seconds) > 30
for: 5m
labels:
severity: warning
annotations:
summary: "节点执行 P99 超过 30s"
description: "{{ $labels.node_name }} P99 延迟 {{ $value }}s"
# 错误率过高
- alert: HighErrorRate
expr: |
rate(langgraph_error_total[5m])
/ rate(langgraph_request_total[5m]) > 0.05
for: 3m
labels:
severity: critical
annotations:
summary: "错误率超过 5%"
# Checkpoint 写入延迟
- alert: CheckpointWriteSlow
expr: histogram_quantile(0.95, langgraph_checkpoint_duration_seconds{operation="write"}) > 0.05
for: 5m
labels:
severity: warning
annotations:
summary: "Checkpoint 写入 P95 超过 50ms"
# LLM Token 消耗异常
- alert: TokenSpendAnomaly
expr: |
increase(langgraph_llm_tokens_total{token_type="output"}[1h]) > 1000000
for: 10m
labels:
severity: warning
annotations:
summary: "1 小时输出 Token 超过 100 万"
八、成本控制
LLM 调用是 Agent 系统最大的成本项，不控制好账单能吓死人。
8.1 单请求 Token 上限
python
# 全局 Token 限制配置
TOKEN_LIMITS = {
"max_input_tokens": 4000, # 单次输入上限
"max_output_tokens": 2000, # 单次输出上限
"max_total_per_thread": 50000, # 单线程总上限
}
async def token_guarded_llm_node(state):
"""带 Token 上限的 LLM 节点"""
messages = state["messages"]
# 检查输入 Token 数（粗估：1 中文字 ≈ 1.5 token）
estimated_tokens = sum(
len(m.get("content", "")) * 1.5 for m in messages
)
if estimated_tokens > TOKEN_LIMITS["max_input_tokens"]:
# 截断历史消息
messages = messages[:2] + messages[-4:] # 保留系统消息 + 最近 2 轮
response = await model.ainvoke(
messages,
max_tokens=TOKEN_LIMITS["max_output_tokens"],
)
return {"messages": [response]}
8.2 日费用告警
python
import redis.asyncio as aioredis
from datetime import date
redis_client = aioredis.from_url("redis://localhost:6379/2")
# 模型单价（$/1K tokens）
MODEL_PRICING = {
"gpt-4o": {"input": 0.0025, "output": 0.01},
"gpt-4o-mini": {"input": 0.00015, "output": 0.0006},
"claude-3-opus": {"input": 0.015, "output": 0.075},
}
DAILY_BUDGET = 100.0 # 日预算 $100
async def track_cost(model: str, input_tokens: int, output_tokens: int):
"""追踪单次调用成本"""
pricing = MODEL_PRICING.get(model, {"input": 0.001, "output": 0.003})
cost = (
pricing["input"] * input_tokens / 1000
+ pricing["output"] * output_tokens / 1000
)
today = date.today().isoformat()
daily_cost = await redis_client.incrbyfloat(f"cost:{today}", cost)
if daily_cost > DAILY_BUDGET * 0.8:
logger.warning(f"日费用已达预算 80%: ${daily_cost:.2f} / ${DAILY_BUDGET}")
if daily_cost > DAILY_BUDGET:
logger.error(f"日费用超预算: ${daily_cost:.2f} / ${DAILY_BUDGET}")
# 触发降级：后续请求走小模型
await redis_client.set(f"degraded:{today}", "1", ex=86400)
return cost
8.3 闲时降级策略
python
from datetime import datetime
def should_use_small_model() -> bool:
"""闲时降级：非工作时间使用小模型"""
now = datetime.now()
hour = now.hour
# 工作时间 9:00-18:00 用大模型，其他时间用小模型
is_work_hours = 9 <= hour < 18
# 周末也降级
is_weekend = now.weekday() >= 5
# 预算超限也降级
is_over_budget = bool(
await redis_client.get(f"degraded:{date.today().isoformat()}")
)
return not is_work_hours or is_weekend or is_over_budget
8.4 成本审计表
表格
九、踩坑记录
下面是 9 个真实踩过的坑，每个都是血的教训。
坑 1：本地 Checkpoint 在分布式部署下的灾难
现象：开发环境用 MemorySaver 跑得好好的，上了 K8s 多副本后，用户反馈"对话经常从中间断掉"。
原因：MemorySaver 只存内存，每个 Pod 各自一份。请求被负载均衡分发到不同 Pod，状态互不可见。
解决：
python
# ❌ 开发环境
from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()
# ✅ 生产环境
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
from psycopg_pool import AsyncConnectionPool
async with AsyncConnectionPool(conninfo=DB_URI, max_size=20) as pool:
checkpointer = AsyncPostgresSaver(pool)
await checkpointer.setup()
教训：从第一天就用 PostgreSQL Checkpointer，即使开发环境也用 Docker 起一个 PG。别想着"后面再换"，后面换的成本是 10 倍。
坑 2：工具未实现幂等性的重复调用
现象：用户反馈"我下了一个订单，结果系统给我创建了两个"。查日志发现 LLM 两次调用同一个工具。
原因：LLM 有时会重复调用工具（尤其是流式输出中），如果工具不是幂等的，就会产生重复副作用。
解决：所有写操作工具必须实现幂等性，参考前面 6.2 节的 @idempotent_tool 装饰器。
坑 3：未设置 max_steps 导致死循环
现象：Agent 跑着跑着就不回来了，日志显示一直在两个节点之间来回跳。LLM 偶尔会"犯傻"，反复调用同一个工具或陷入循环推理。
原因：LangGraph 默认 recursion_limit=25，但对于某些场景（如 Agent 和工具之间循环），25 步可能还不够——关键是你的条件边没有正确的终止条件。
解决：
python
# 编译时设置更保守的限制
graph = builder.compile(
checkpointer=checkpointer,
)
# 运行时限制步数
result = await graph.ainvoke(
input_data,
config={
"recursion_limit": 15, # 根据业务调整
},
)
# 更好的方案：在条件边中加循环检测
def should_continue(state):
if state.get("step_count", 0) >= 10:
return "end" # 强制结束
# ... 正常路由逻辑
坑 4：敏感数据存入 State
现象：安全审计发现用户密码、API Key 等敏感信息出现在 PostgreSQL Checkpoint 表里。
原因：Agent 在对话中收集到敏感信息后，直接存入 State，而 State 被 Checkpointer 全量持久化。
解决：
python
# 1. State 设计时就排除敏感字段
class SafeState(TypedDict):
messages: Annotated[list, add_messages]
user_id: str # ✅ 只存 ID
# ❌ 不要存: api_key, password, token
# 2. 敏感信息脱敏后存储
import re
def sanitize_message(content: str) -> str:
"""脱敏处理"""
# 脱敏 API Key
content = re.sub(r'sk-[a-zA-Z0-9]{20,}', 'sk- ***REDACTED***', content)
# 脱敏邮箱
content = re.sub(r'\b[\w.]+@[\w.]+\.\w+\b', ' ***@** *. ***', content)
# 脱敏手机号
content = re.sub(r'\b1[3-9]\d{9}\b', '1XX****XXXX', content)
return content
坑 5：全异步改造的坑
现象 ：把所有节点改成 async 后，某些工具执行报错 no running event loop。
原因 ：Python 3.10 及以下版本中，LangGraph 的 get_stream_writer() 在异步节点内不工作。另外，混合使用同步和异步工具也会出问题。
解决 ：
python
# Python 3.11+ 可以用 get_stream_writer()
from langgraph.config import get_stream_writer
async def my_node(state):
writer = get_stream_writer()
writer({"progress": "处理中..."})
# ...
# Python 3.10 需要手动传 writer 参数
async def my_node(state, *, writer):
writer({"progress": "处理中..."})
# ...
建议 ：生产环境用 Python 3.11+，省很多坑。
坑 6：冷启动优化
现象 ：K8s 新 Pod 启动后第一个请求延迟特别高（5-10 秒），触发超时告警。
原因 ：Graph 编译 + LLM 客户端初始化 + DB 连接建立，叠加起来很慢。
解决 ：
python
from contextlib import asynccontextmanager
from fastapi import FastAPI
# 全局变量，启动时初始化
_graph = None
_checkpointer = None
@asynccontextmanager
async def lifespan(app: FastAPI):
"""应用生命周期：启动时预加载资源"""
global _graph, _checkpointer
# 1. 预建连接池
pool = AsyncConnectionPool(conninfo=DB_URI, min_size=5, max_size=20)
await pool.open()
# 2. 预初始化 Checkpointer
_checkpointer = AsyncPostgresSaver(pool)
await _checkpointer.setup()
# 3. 预编译 Graph
_graph = build_graph(_checkpointer)
# 4. 预热 LLM（发一个 dummy 请求，建立连接）
try:
await model.ainvoke([("user", "ping")], max_tokens=1)
except Exception:
pass
logger.info("✅ 资源预加载完成")
yield
# 关闭时清理
await pool.close()
app = FastAPI(lifespan=lifespan)
坑 7：状态膨胀
现象 ：运行一段时间后 PostgreSQL 磁盘占用飙升，Checkpoint 表体积巨大。
原因 ：LangGraph 默认的全量快照模式下，messages 通道每步都存完整历史，存储增长 O(N²)。
解决 ：参考 3.3 节的三种优化策略。核心是：
State 只存必要字段，大对象只存 ID
定期清理过期 Checkpoint
如果用 langgraph 1.2+，开启 DeltaChannel
python
# 状态瘦身前后对比
class FatState(TypedDict):
messages: Annotated[list, add_messages]
search_results: list[dict] # 每条 5KB，10 条 = 50KB
document_text: str # 100KB 文档全文
class LeanState(TypedDict):
messages: Annotated[list, add_messages]
search_result_ids: list[str] # 10 个 ID = 500B，缩 100 倍
document_id: str # 1 个 ID = 36B，缩 2800 倍
坑 8：并发竞争
现象 ：多个请求同时修改同一个 thread_id 的状态，导致数据不一致。
原因 ：LangGraph 的 Checkpoint 读写不是原子操作。两个请求同时读最新 Checkpoint，各自修改后写入，后写的会覆盖先写的。
解决 ：
python
import asyncio
# 方案 1：应用层锁（简单但限制并发）
_thread_locks: dict[str, asyncio.Lock] = {}
def get_thread_lock(thread_id: str) -> asyncio.Lock:
if thread_id not in _thread_locks:
_thread_locks[thread_id] = asyncio.Lock()
return _thread_locks[thread_id]
async def safe_invoke(graph, input_data, thread_id: str):
async with get_thread_lock(thread_id):
return await graph.ainvoke(
input_data,
config={"configurable": {"thread_id": thread_id}},
)
# 方案 2：使用 LangGraph Platform 的任务队列（推荐）
# LangGraph Platform 内置了线程级任务队列，自动串行化同 thread 请求
坑 9：监控盲区
现象 ：系统"看起来"正常运行，但用户投诉"回复越来越慢"。等发现时已经影响了大量用户。
原因 ：只监控了 API 层的延迟和错误率，没有监控 Agent 内部：节点耗时分布、Token 消耗趋势、Checkpoint 延迟。
解决 ：
plaintext
┌────────────────────────────────────────────────────┐
│ 监控盲区 Checklist │
│ │
│ API 层（通常已覆盖） │
│ ├── 请求延迟 P50 / P95 / P99 │
│ ├── 错误率 4xx / 5xx │
│ └── QPS / 并发数 │
│ │
│ Agent 层（容易遗漏！） │
│ ├── 每个节点的执行耗时 ← 关键 │
│ ├── LLM 调用次数 & Token 消耗 ← 关键 │
│ ├── Checkpoint 读写延迟 ← 关键 │
│ ├── 工具调用成功/失败率 │
│ └── 递归深度分布 │
│ │
│ 资源层 │
│ ├── PostgreSQL 连接池使用率 │
│ ├── Redis 内存占用 │
│ ├── 进程内存 / CPU │
│ └── 活跃线程数 / 僵尸线程数 │
│ │
│ 业务层 │
│ ├── 日费用趋势 ← 成本失控的早期信号 │
│ ├── 语义缓存命中率 │
│ └── 模型路由分布 │
└────────────────────────────────────────────────────┘
总结
把 LangGraph Agent 从 Demo 推上生产线，核心是回答三个问题：
1. 状态怎么不丢？ → PostgreSQL Checkpoint + 连接池 + 重试 + TTL 清理
2. 成本怎么控制？ → 语义缓存 + 模型路由 + Token 上限 + 上下文压缩
3. 出问题怎么查？ → Prometheus 指标 + 结构化日志 + LangSmith/OTel 追踪
一句话：** LangGraph 解决了编排问题，但运维问题得你自己扛 **。架构设计、Checkpoint 配置、流式推送、错误处理、可观测性——这些一个都不能少。
Demo 验证想法，生产验证工程。希望能帮你少踩几个坑。
关联阅读：
第02篇：vLLM 部署 — 本地模型的部署与推理优化
第06篇：链路追踪 — OpenTelemetry 全链路追踪实战
第07篇：语义缓存 — Redis 向量检索实现 LLM 缓存
第19篇：状态机基础 — LangGraph 核心概念入门
第39篇：持久化与人在回路 — Checkpoint 深入与 HITL
第40篇：子图与多Agent编排 — 复杂 Agent 架构设计
评论区