网硕互联帮助中心在上一篇文章中，我们建立了 Agent 的可观测性与评估体系——你能看到推理链的每一步，能量化 Agent 的行为质量。但所有这些能力的前提是：Agent 已经作为一个生产级服务在运行。一个在笔记本上 python main.py 跑通的 Agent，和一个能承受真实流量、在故障时自动恢复、成本可控的服务之间，隔着工程化的最后一公里。Agent 服务与传统微服务在运维上有本质区别：Agent 的不确定性意味着“重试”可能产生不同的结果，“超时”可能意味着 Token 已经消耗但任务未完成，“扩缩容”需要基于队列深度而非 CPU 使用率。 本文从“从脚本到服务”的三条部署路径讲起，深入讲解 FastAPI 封装、容器化、Kubernetes 弹性伸缩、成本控制和幂等设计，并通过一个生产级模板帮你建立 Agent 运维的完整框架。
一、从脚本到服务：三条部署路径
LangGraph Agent 的部署有三条主流路径，选择取决于团队规模和运维偏好：
路径 3 是本文的重点——它也是大多数有生产级要求的团队最终会选择的路径。一个生产级 LangGraph 模板的典型架构是：
text
Client → FastAPI (auth · rate limit · validation)
→ PackRegistry / control_plane policies
→ domain_packs/* (LangGraph workflows)
→ agents/* (reusable agent nodes)
→ core/* (LLM · memory · security · cost · observability)
→ connectors/* (optional retrieval)
二、FastAPI 封装：Agent 的 HTTP 门面
FastAPI 是 Agent 服务化的标准选择。但 Agent 的 API 与传统 CRUD API 有几个关键区别。
2.1 SSE 流式响应
Agent 的执行是长时间运行的，用户需要实时看到进度。Server-Sent Events（SSE） 是标准方案：
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import json
app = FastAPI()
@app.post("/run/stream")
async def run_agent_stream(request: AgentRequest):
async def event_generator():
async for event in agent.astream({
"messages": [{"role": "user", "content": request.query}]
}):
yield f"data: {json.dumps(event)}\\n\\n"
return StreamingResponse(event_generator(), media_type="text/event-stream")
关键点：SSE 让前端能够逐步渲染 Agent 的思考过程、工具调用和最终答案，而不是等待整个任务完成。
2.2 幂等性：Agent 重试的安全网
Agent 的调用可能因为网络超时被客户端重试，但 Agent 的副作用（如创建订单、发送邮件）不应被重复执行。幂等键是解决方案：
@app.post("/packs/meeting_prep/run")
async def run_pack(
request: PackRequest,
idempotency_key: str = Header(..., alias="Idempotency-Key"),
):
cached = await idempotency_store.get(idempotency_key)
if cached:
return cached # 返回缓存结果，不重复执行
result = await pack.run(request)
await idempotency_store.set(idempotency_key, result, ttl=86400)
return result
生产级模板的行为：用相同的幂等键和请求体重试返回缓存响应；用相同的幂等键但不同的请求体返回 HTTP 409 Conflict。默认使用内存存储，多副本部署时切换到 Redis 后端。
2.3 输入验证与速率限制
Agent 的输入直接进入 LLM，输入验证是安全的第一道防线：
from pydantic import BaseModel, Field
class AgentRequest(BaseModel):
query: str = Field(min_length=1, max_length=2000)
conversation_id: str = Field(pattern=r"^[a-f0-9-]{36}$")
速率限制防止单个用户耗尽 Token 预算，通常用 slowapi 或 Redis 令牌桶实现。
三、容器化与 Kubernetes：让 Agent 弹性伸缩
3.1 多阶段 Docker 构建
dockerfile
FROM python:3.12-slim AS builder
WORKDIR /app
COPY requirements.txt .
RUN pip install –no-cache-dir -r requirements.txt
FROM python:3.12-slim
WORKDIR /app
COPY –from=builder /usr/local/lib/python3.12/site-packages /usr/local/lib/python3.12/site-packages
COPY . .
EXPOSE 8000
CMD [“uvicorn”, “api.main:app”, “–host”, “0.0.0.0”, “–port”, “8000”]
3.2 为什么 Agent 需要 KEDA 而非单纯 HPA
HPA 基于 CPU/内存伸缩，但 Agent 服务在等待 LLM 响应时 CPU 使用率很低——CPU 指标无法反映真实的负载。Agent 的负载信号是任务队列深度。
KEDA（Kubernetes Event-Driven Autoscaling） 基于外部事件源伸缩，天然适合 Agent 的任务积压场景：
队列深度驱动：当队列中有 50 个任务、每个 Pod 处理 5 个任务时，KEDA 伸缩到 10 个 Pod。
冷启动问题：Agent 的冷启动可能涉及加载模型、建立连接，需要配置最小热副本数来吸收突发流量。
HPA 与 KEDA 不能同时使用：KEDA 本身会创建和管理 HPA，两者同时使用会“打架”。
KEDA 配置示意：
apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
name: agent–worker–scaler
spec:
scaleTargetRef:
name: agent–worker
minReplicaCount: 2 # 最小热副本，吸收突发
maxReplicaCount: 20
triggers:
– type: redis
metadata:
address: redis:6379
listName: agent–tasks
listLength: "5" # 每 5 个积压任务扩 1 个 Pod
3.3 健康检查与优雅关闭
Agent 的 livenessProbe 不应只检查端口是否监听——它应该检查LLM 连接是否可用。readinessProbe 应该检查队列消费者是否就绪。
优雅关闭：收到 SIGTERM 后，停止接收新任务，等待进行中的任务完成（设置合理的超时），然后退出。这需要 FastAPI 的 lifespan 事件配合：
from contextlib import asynccontextmanager
@asynccontextmanager
async def lifespan(app: FastAPI):
worker.start()
yield
await worker.graceful_shutdown(timeout=30)
app = FastAPI(lifespan=lifespan)
四、成本控制：Agent 运维的“第一性”约束
Agent 的成本与传统微服务有本质区别：Token 消耗是变动成本，且与任务复杂度非线性相关。一个“慢”请求可能消耗 10 倍预算。
4.1 每运行成本上限
生产级模板的核心设计之一是每次运行美元预算上限：超过预算的请求返回 HTTP 402（Payment Required） ：
PACK_DEFAULT_BUDGET_USD = 0.50
def check_budget(run_cost: float, budget: float = PACK_DEFAULT_BUDGET_USD):
if run_cost > budget:
raise HTTPException(status_code=402, detail="Run exceeded cost budget")
成本参考：一次 research_analysis 运行（6 次 LLM 调用）在 Claude Sonnet 5 定价下约 $0.01–$0.05。
4.2 分层路由：主强子弱
一个经过验证的成本优化策略是主 Agent 用强模型、子 Agent 用便宜模型。清华开源的 PilotDeck 实践显示，采用“主 Sonnet 4.6 + 子 MiniMax-M2.7”的主强子弱架构，综合得分达到 70.6，成本仅 $3.15；而纯强模型方案成本更高，纯弱模型方案得分更低。
4.3 成本度量的正确单位
Token 价格不等于 Agent 成本。 正确的度量是每个成功完成任务的成本：
text
每次成功完成任务的成本 = (Token + 计算 + 审查时间 + 返工) ÷ 正确完成的任务数
这个公式迫使你同时关注质量和成本——一个便宜但频繁失败的 Agent，其“每次成功任务成本”可能远高于一个贵但可靠的 Agent。
五、实战：生产级 Agent 服务的部署清单
以下是将一个 LangGraph Agent 从脚本推向生产的最小可行清单：
5.1 持久化 Checkpointer
生产环境绝不能用内存 Checkpointer。LangGraph 提供 PostgresSaver 和 RedisSaver：
from langgraph.checkpoint.postgres import PostgresSaver
checkpointer = PostgresSaver.from_conn_string(
"postgresql://user:pass@postgres:5432/agentdb"
)
checkpointer.setup()
graph = builder.compile(checkpointer=checkpointer)
为什么必须持久化：服务重启、Pod 被驱逐、节点故障——Agent 的会话状态必须能在这些事件中存活。内存 Checkpointer 会让所有进行中的任务丢失状态。
5.2 Helm Chart 与 GitOps
生产级模板通常包含 Helm Chart，支持在 GKE/EKS/AKS 上部署，并通过 GitOps（ArgoCD/Flux）管理发布。金丝雀包版本控制（canary pack versioning）允许按流量权重逐步切换新版本的 Agent 工作流：
# Helm values 中的金丝雀配置
canary:
enabled: true
weight: 10 # 10% 流量走新版本
packVersion: v2
六、Agent 运维与传统微服务运维的本质区别
核心差异：传统微服务的运维对象是确定性的进程；Agent 服务的运维对象是概率性的决策系统。你无法用“重启”解决一个“提示词写错了”的问题，也无法用“加 CPU”解决一个“工具选择错误”的问题。可观测性和评估不是 Agent 运维的附加项，而是运维本身的一部分。
七、小结
三条部署路径：LangServe+FastAPI（轻量）、LangGraph Platform（托管）、FastAPI+Docker/K8s（自控）。
FastAPI 封装：SSE 流式响应、幂等键（相同键+相同体返回缓存，相同键+不同体返回 409）、Pydantic 验证、速率限制。
KEDA 而非 HPA：Agent 的负载信号是队列深度，不是 CPU。配置最小热副本吸收冷启动。
成本控制：每运行预算上限（超限返回 402）、主强子弱分层路由、每次成功任务成本才是正确度量。
持久化 Checkpointer：生产环境必须用 Postgres/Redis，内存存储在 Pod 驱逐时丢失状态。
运维本质差异：传统微服务运维确定性进程，Agent 运维概率性决策系统。可观测性和评估是运维的一部分。
评论前必须登录！
注册