生产部署建议
将 LangGraph Agent 部署到生产环境需要注意架构、性能和运维等多个方面。本文基于实际部署经验总结。
部署架构
基本架构（单机）
用户 → Nginx → LangGraph Agent (Python) → LLM API
→ 数据库/向量库
→ 缓存生产架构（分布式）
用户 → CDN → Nginx → Agent Server (多实例)
↓
Redis/Cache
↓
Postgres/SQLite (Checkpoint)
↓
LLM API | 向量库 | 外部服务Agent Server 部署
使用 LangGraph Agent Server
python
# app.py
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.postgres import PostgresSaver
# 定义图
graph = create_my_graph()
# 使用 Postgres 持久化
checkpointer = PostgresSaver.from_conn_string(
"postgresql://user:pass@localhost:5432/langgraph"
)
# 编译
app = graph.compile(checkpointer=checkpointer)启动服务：
bash
# LangGraph CLI 启动
langgraph dev
# 或 Docker
docker run -p 8123:8123 my-agent-image使用 FastAPI 自定义部署
python
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from langgraph.checkpoint.postgres import PostgresSaver
app = FastAPI()
checkpointer = PostgresSaver.from_conn_string(DB_URL)
graph = create_graph(checkpointer)
class Query(BaseModel):
message: str
thread_id: str = None
@app.post("/chat")
async def chat(query: Query):
config = {
"configurable": {
"thread_id": query.thread_id or str(uuid.uuid4())
}
}
result = graph.invoke(
{"messages": [{"role": "user", "content": query.message}]},
config
)
return {"response": result["messages"][-1].content, "thread_id": config["configurable"]["thread_id"]}
@app.post("/chat/stream")
async def chat_stream(query: Query):
config = {"configurable": {"thread_id": query.thread_id or str(uuid.uuid4())}}
async for event in graph.astream(
{"messages": [{"role": "user", "content": query.message}]},
config,
stream_mode="messages"
):
# SSE 流式输出
yield f"data: {event.content}\n\n"Nginx 配置
nginx
upstream agent_backend {
server 127.0.0.1:8000;
server 127.0.0.1:8001;
keepalive 64;
}
server {
listen 443 ssl;
server_name your-agent.com;
# SSE 需要关闭缓冲
location /chat/stream {
proxy_pass http://agent_backend;
proxy_http_version 1.1;
proxy_set_header Connection '';
proxy_buffering off;
proxy_cache off;
proxy_read_timeout 86400s;
}
location / {
proxy_pass http://agent_backend;
proxy_http_version 1.1;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection "upgrade";
}
}性能优化
Checkpoint 存储
python
# Postgres 连接池配置
from langgraph.checkpoint.postgres import PostgresSaver
checkpointer = PostgresSaver.from_conn_string(
DB_URL,
pool_size=10, # 连接池大小
max_overflow=20, # 最大额外连接
pool_timeout=30 # 等待连接超时
)缓存 LLM 调用
python
from functools import lru_cache
from langchain.chat_models import init_chat_model
@lru_cache(maxsize=100)
def get_model(model_name: str):
return init_chat_model(model_name)流式响应优先
python
# 优先使用 Streaming 而非等待完整响应
async for event in graph.astream(inputs, stream_mode="messages"):
await websocket.send_text(event.content)安全
速率限制
python
from slowapi import Limiter
from slowapi.util import get_remote_address
limiter = Limiter(key_func=get_remote_address)
@app.post("/chat")
@limiter.limit("10/minute") # 每分钟 10 次
async def chat(request: Request, query: Query):
...输入校验
python
from pydantic import BaseModel, Field
class SafeQuery(BaseModel):
message: str = Field(max_length=4000) # 限制长度
thread_id: str = Field(default=None, max_length=64)Token 限制
python
def token_limiter(state: State) -> State:
"""限制消息总量"""
max_tokens = 100_000
current_tokens = count_tokens(state["messages"])
if current_tokens > max_tokens:
# 压缩历史
state["messages"] = compress_messages(state["messages"])
return state监控
Prometheus 指标
python
from prometheus_client import Counter, Histogram
request_count = Counter('agent_requests_total', 'Total requests')
request_duration = Histogram('agent_request_duration_seconds', 'Request duration')
error_count = Counter('agent_errors_total', 'Total errors')
@app.middleware("http")
async def monitor(request: Request, call_next):
request_count.inc()
with request_duration.time():
response = await call_next(request)
if response.status_code >= 400:
error_count.inc()
return response日志
python
import structlog
logger = structlog.get_logger()
@app.post("/chat")
async def chat(query: Query):
logger.info("chat_request", message=query.message, thread=query.thread_id)
try:
result = graph.invoke(inputs, config)
logger.info("chat_success", thread=query.thread_id)
return result
except Exception as e:
logger.error("chat_error", error=str(e), thread=query.thread_id)
raise部署清单
- [ ] Checkpointer 配置（开发用 SQLite，生产用 Postgres）
- [ ] API 速率限制
- [ ] 输入校验和 Token 限制
- [ ] 超时配置
- [ ] 日志和监控
- [ ] 错误处理和重试策略
- [ ] 水平扩展（多实例）
- [ ] CI/CD 流水线
- [ ] HTTPS 证书
- [ ] 备份策略