L19 生产部署（上）：从 Demo 到可观测、可恢复的生产级 Agent

• 难度 ⭐⭐⭐⭐⭐（高级）
• 预计阅读 35 分钟
• 代码量 含完整可观测性 Agent 实现
• 先修 L8（元认知与评估）+ L16（MAF）

课程目标：理解 Agent 从 Demo 到生产需要跨越的工程鸿沟，掌握 OpenTelemetry 可观测性、LangSmith/Langfuse 监控、CI/CD 回归测试、生产部署最佳实践。

一、Demo 和生产之间隔着一道鸿沟#

你在本地跑通了一个 Agent：

agent.run("帮我订明天上午的会议室")
# 输出：好的，已帮你预订明天上午 9-10 点的会议室 A

Demo 成功。但部署到生产后：

• 怎么知道每天有多少人用？

• 怎么知道平均花费多少 token？

• 怎么知道 Agent 哪一步出错最多？

• 怎么知道用户最常问什么？

• Agent 出现循环调用怎么办？

• 新版本发布怎么知道没回归？

• 服务挂了怎么排查？

这一课我们跨过这道鸿沟。

二、用一个比喻：Agent 部署 = 开餐厅#

把 Agent 比作开餐厅：

没有监控的餐厅 = 没有可观测性的 Agent——你不知道发生了什么，只能事后救火。

三、生产环境的三大核心诉求#

诉求 1：可观测性（Observability）#

知道 Agent 在做什么——包括：

• 每次调用的完整轨迹（哪些 LLM 调用、哪些工具调用）

• 每次调用的成本和延迟

• 失败的原因和模式

• 用户使用模式

诉求 2：可恢复性（Reliability）#

Agent 出错时能恢复——包括：

• 自动重试（网络抖动、Rate Limit）

• 降级策略（LLM 挂了用更小的模型）

• 错误告警（异常率超过阈值通知）

• 限流保护（防止滥用）

诉求 3：可演进性（Evolvability）#

新版本不破坏老版本——包括：

• CI/CD 自动化部署

• A/B 测试新旧版本

• 回归测试（跑评测集确保不退步）

• 灰度发布（10% 流量 → 50% → 100%）

四、可观测性 1：OpenTelemetry 标准#

OpenTelemetry（OTel）是云原生时代的事实标准，所有框架都应该接入。

核心概念#

层级关系#

Trace: 用户问"北京天气"
├── Span: Agent 决策（200ms）
│   ├── Span: LLM 调用 gpt-4o-mini（150ms, 800 tokens）
│   ├── Span: 工具调用 get_weather（50ms, 返回 25°C）
│   └── Span: LLM 生成最终答案（80ms, 200 tokens）
└── 总计：480ms, 1000 tokens

用 LangSmith 实现（LangChain 官方方案）#

import os
os.environ["LANGSMITH_API_KEY"] = "lsv2_..."
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_PROJECT"] = "production-agent"

from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, MessagesState, START, END

llm = ChatOpenAI(model="gpt-4o-mini")

def agent(state: MessagesState):
    return {"messages": [llm.invoke(state["messages"])]}

graph = StateGraph(MessagesState).add_node("agent", agent)
graph.add_edge(START, "agent")
graph.add_edge("agent", END)
app = graph.compile()

# 每次 invoke 都会自动 trace
result = app.invoke({"messages": [{"role": "user", "content": "北京今天天气如何？"}]})

LangSmith 自动捕获：

• 每次调用的完整步骤

• 每次 LLM 调用的 prompt、response、token 数

• 每次工具调用的参数、结果、耗时

• 错误和异常

用 OpenTelemetry（开源、中立）#

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter

# 1. 设置 Tracer Provider
provider = TracerProvider()
processor = BatchSpanProcessor(OTLPSpanExporter(endpoint="http://localhost:4317"))
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)

tracer = trace.get_tracer(__name__)

# 2. 在 Agent 中添加自定义 span
@tracer.start_as_current_span("agent_step")
def traced_agent(state: MessagesState):
    span = trace.get_current_span()
    span.set_attribute("user.input", state["messages"][-1].content)
    span.set_attribute("model.name", "gpt-4o-mini")
    
    response = llm.invoke(state["messages"])
    
    span.set_attribute("llm.tokens", response.response_metadata["token_usage"]["total_tokens"])
    return {"messages": [response]}

用 Langfuse（开源、自托管友好）#

from langfuse.callback import CallbackHandler

handler = CallbackHandler(
    public_key="pk-...",
    secret_key="sk-...",
    host="https://cloud.langfuse.com"  # 或自托管地址
)

result = app.invoke(
    {"messages": [{"role": "user", "content": "北京天气？"}]},
    config={"callbacks": [handler]}
)

LangSmith vs Langfuse：

五、可观测性 2：关键指标监控#

必须监控的 5 个核心指标#

from dataclasses import dataclass
from collections import deque
import time

@dataclass
class AgentMetrics:
    """Agent 指标"""
    timestamp: float
    request_id: str
    user_id: str
    total_latency_ms: float
    llm_calls: int
    tool_calls: int
    total_tokens: int
    cost_usd: float
    success: bool
    error_type: str = None

class MetricsCollector:
    """指标收集器"""
    
    def __init__(self):
        self.metrics = deque(maxlen=10000)
        self.error_counts = {}  # error_type -> count
    
    def record(self, metric: AgentMetrics):
        self.metrics.append(metric)
        if not metric.success:
            self.error_counts[metric.error_type] = \
                self.error_counts.get(metric.error_type, 0) + 1
    
    def summary(self, window_minutes: int = 60) -> dict:
        """生成时间窗口内的汇总指标"""
        now = time.time()
        cutoff = now - window_minutes * 60
        recent = [m for m in self.metrics if m.timestamp > cutoff]
        
        if not recent:
            return {}
        
        total = len(recent)
        success_count = sum(1 for m in recent if m.success)
        
        return {
            "window_minutes": window_minutes,
            "total_requests": total,
            "success_rate": success_count / total,
            "p50_latency_ms": self._percentile([m.total_latency_ms for m in recent], 50),
            "p95_latency_ms": self._percentile([m.total_latency_ms for m in recent], 95),
            "p99_latency_ms": self._percentile([m.total_latency_ms for m in recent], 99),
            "avg_tokens": sum(m.total_tokens for m in recent) / total,
            "total_cost_usd": sum(m.cost_usd for m in recent),
            "top_errors": sorted(self.error_counts.items(), 
                                  key=lambda x: x[1], reverse=True)[:5]
        }
    
    def _percentile(self, values: list, p: int) -> float:
        sorted_values = sorted(values)
        idx = int(len(sorted_values) * p / 100)
        return sorted_values[min(idx, len(sorted_values)-1)]

# 使用
metrics_collector = MetricsCollector()

def wrapped_agent(state: MessagesState):
    request_id = str(uuid.uuid4())
    start = time.time()
    try:
        result = original_agent(state)
        latency = (time.time() - start) * 1000
        
        metrics_collector.record(AgentMetrics(
            timestamp=time.time(),
            request_id=request_id,
            user_id=state.get("user_id", "anonymous"),
            total_latency_ms=latency,
            llm_calls=result.get("llm_calls", 0),
            tool_calls=result.get("tool_calls", 0),
            total_tokens=result.get("total_tokens", 0),
            cost_usd=result.get("total_tokens", 0) * 0.00015 / 1000,
            success=True
        ))
        return result
    except Exception as e:
        metrics_collector.record(AgentMetrics(
            timestamp=time.time(),
            request_id=request_id,
            user_id="anonymous",
            total_latency_ms=(time.time() - start) * 1000,
            llm_calls=0,
            tool_calls=0,
            total_tokens=0,
            cost_usd=0,
            success=False,
            error_type=type(e).__name__
        ))
        raise

告警规则示例#

# Prometheus / Grafana 告警规则
alerts:
  - name: HighErrorRate
    condition: error_rate > 0.05  # 5%
    duration: 5m
    action: page_oncall
    
  - name: HighLatency
    condition: p95_latency_ms > 10000
    duration: 5m
    action: slack_notify
    
  - name: HighCost
    condition: avg_cost_per_request > $0.10
    duration: 1h
    action: email_team
    
  - name: LLMApiFailure
    condition: llm_api_error_rate > 0.02
    duration: 2m
    action: page_oncall + enable_fallback

六、可恢复性：错误处理与降级#

模式 1：自动重试 + 指数退避#

import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=1, max=10),
    retry_error_callback=lambda _: "服务暂时不可用，请稍后再试。"
)
async def call_llm_with_retry(messages):
    """带重试的 LLM 调用"""
    return await llm.ainvoke(messages)

模式 2：降级到更小的模型#

class TieredLLM:
    """分层 LLM - 失败时降级"""
    
    def __init__(self):
        self.primary = ChatOpenAI(model="gpt-4o")
        self.fallback = ChatOpenAI(model="gpt-4o-mini")
        self.emergency = ChatOpenAI(model="gpt-3.5-turbo")
    
    async def invoke(self, messages):
        for llm, name in [(self.primary, "primary"), 
                          (self.fallback, "fallback"),
                          (self.emergency, "emergency")]:
            try:
                response = await llm.ainvoke(messages, timeout=10)
                if name != "primary":
                    log.warning(f"使用降级模型 {name}")
                return response
            except Exception as e:
                log.error(f"{name} 模型失败: {e}")
                continue
        
        raise Exception("所有模型都不可用")

模式 3：循环检测与中断#

def detect_loop(state: MessagesState, max_repeats: int = 3) -> bool:
    """检测 Agent 是否陷入循环"""
    recent_tool_calls = [
        msg.tool_calls[0]["name"] 
        for msg in state["messages"][-10:] 
        if msg.tool_calls
    ]
    
    if len(recent_tool_calls) < max_repeats:
        return False
    
    # 检查最后 max_repeats 个调用是否相同
    last_n = recent_tool_calls[-max_repeats:]
    if len(set(last_n)) == 1:
        return True
    
    return False

def safe_agent(state: MessagesState):
    if detect_loop(state):
        return {"messages": [{"role": "assistant",
                              "content": "我陷入了循环，让我换个思路..."}]}
    return original_agent(state)

模式 4：Rate Limit 与成本控制#

from threading import Lock

class RateLimiter:
    """简单的速率限制器"""
    
    def __init__(self, max_per_minute: int = 60):
        self.max_per_minute = max_per_minute
        self.calls = deque()
        self.lock = Lock()
    
    def allow(self) -> bool:
        with self.lock:
            now = time.time()
            # 清理一分钟前的记录
            while self.calls and self.calls[0] < now - 60:
                self.calls.popleft()
            
            if len(self.calls) >= self.max_per_minute:
                return False
            
            self.calls.append(now)
            return True

class CostGuard:
    """成本守护 - 单请求 Token 上限"""
    
    def __init__(self, max_tokens_per_request: int = 50000):
        self.max_tokens = max_tokens_per_request
    
    def check(self, state: MessagesState) -> bool:
        current_tokens = sum(
            msg.response_metadata.get("token_usage", {}).get("total_tokens", 0)
            for msg in state["messages"] if hasattr(msg, "response_metadata")
        )
        return current_tokens < self.max_tokens

七、可演进性：CI/CD 与回归测试#

Agent 评测集：和单元测试一样重要#

# eval_cases.py - 评测集
EVAL_CASES = [
    {
        "input": "北京今天天气？",
        "expected_tools": ["get_weather"],
        "expected_keywords": ["北京", "温度", "天气"],
        "max_latency_ms": 5000,
        "max_cost_usd": 0.05
    },
    {
        "input": "公司地址是什么？",
        "expected_tools": ["get_company_info"],
        "expected_keywords": ["北京市", "海淀"],
        "max_latency_ms": 3000,
        "max_cost_usd": 0.02
    },
    # ... 100 个用例
]

def run_evaluation(agent) -> dict:
    """运行评测"""
    results = []
    for case in EVAL_CASES:
        start = time.time()
        try:
            response = agent.invoke({"messages": [{"role": "user", "content": case["input"]}]})
            latency = (time.time() - start) * 1000
            
            # 多维度评估
            tool_accuracy = check_tool_usage(response, case["expected_tools"])
            keyword_score = check_keywords(response["messages"][-1].content, 
                                            case["expected_keywords"])
            
            results.append({
                "case": case["input"],
                "success": keyword_score > 0.7,
                "tool_accuracy": tool_accuracy,
                "keyword_score": keyword_score,
                "latency_ms": latency,
                "within_sla": latency < case["max_latency_ms"]
            })
        except Exception as e:
            results.append({"case": case["input"], "success": False, "error": str(e)})
    
    return {
        "success_rate": sum(r["success"] for r in results) / len(results),
        "avg_tool_accuracy": sum(r.get("tool_accuracy", 0) for r in results) / len(results),
        "avg_keyword_score": sum(r.get("keyword_score", 0) for r in results) / len(results),
        "p95_latency_ms": percentile([r.get("latency_ms", 0) for r in results], 95),
        "sla_pass_rate": sum(r.get("within_sla", False) for r in results) / len(results)
    }

GitHub Actions CI/CD#

# .github/workflows/agent-ci.yml
name: Agent CI/CD

on:
  pull_request:
    branches: [main]
  push:
    branches: [main]

jobs:
  evaluate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      
      - name: Setup Python
        uses: actions/setup-python@v4
        with:
          python-version: "3.11"
      
      - name: Install dependencies
        run: pip install -r requirements.txt
      
      - name: Run evaluation
        env:
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
        run: |
          python -m pytest tests/test_agent_eval.py -v
      
      - name: Compare with baseline
        run: |
          python scripts/compare_metrics.py \
            --current metrics/current.json \
            --baseline metrics/baseline.json \
            --threshold 0.05
      
      - name: Deploy to staging
        if: github.event_name == 'push' && github.ref == 'refs/heads/main'
        run: |
          python scripts/deploy.py --env staging
      
      - name: Deploy to production
        if: github.event_name == 'push' && github.ref == 'refs/heads/main'
        run: |
          python scripts/deploy.py --env production --canary 10  # 10% 灰度

A/B 测试新旧版本#

import random

class ABTestRouter:
    """A/B 测试路由"""
    
    def __init__(self, agent_v1, agent_v2, traffic_split: float = 0.5):
        self.agent_v1 = agent_v1
        self.agent_v2 = agent_v2
        self.traffic_split = traffic_split
    
    def route(self, user_id: str):
        """根据 user_id 决定使用哪个版本"""
        # 用 hash 保证同一用户总是走同一版本
        if hash(user_id) % 100 < self.traffic_split * 100:
            return self.agent_v2, "B"
        return self.agent_v1, "A"

八、生产部署架构#

最小可用生产架构#

用户
   ↓
[CDN / API Gateway]
   ↓
[Agent Service (FastAPI / Flask)]
   ├─ Rate Limiter（限流）
   ├─ Auth Middleware（鉴权）
   ├─ Input Filter（注入检测）
   ├─ Agent Core（业务逻辑）
   ├─ Tracing（OpenTelemetry）
   └─ Metrics（Prometheus）
   ↓
[LLM API / Tool APIs]
   ↓
[Observability Stack]
   ├─ LangSmith / Langfuse
   ├─ Prometheus + Grafana
   └─ Sentry（错误追踪）
   ↓
[Storage]
   ├─ Postgres（用户、会话、Trace）
   └─ Redis（缓存、限流计数器）

FastAPI 部署示例#

from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel
import uuid

app = FastAPI(title="Production Agent API")

class AgentRequest(BaseModel):
    user_id: str
    message: str
    thread_id: str = None

class AgentResponse(BaseModel):
    request_id: str
    response: str
    cost_usd: float
    latency_ms: float

@app.post("/agent/invoke", response_model=AgentResponse)
async def invoke_agent(request: AgentRequest, 
                       user=Depends(verify_token)):
    request_id = str(uuid.uuid4())
    
    # 限流
    if not rate_limiter.allow():
        raise HTTPException(429, "Rate limit exceeded")
    
    # 注入检测
    if detect_injection(request.message):
        raise HTTPException(400, "Invalid input")
    
    # 调用 Agent
    thread_id = request.thread_id or str(uuid.uuid4())
    config = {"configurable": {"thread_id": thread_id}}
    
    start = time.time()
    try:
        result = await app_agent.ainvoke(
            {"messages": [{"role": "user", "content": request.message}]},
            config=config
        )
        latency = (time.time() - start) * 1000
        
        # 记录 metrics
        metrics_collector.record(AgentMetrics(
            timestamp=time.time(),
            request_id=request_id,
            user_id=request.user_id,
            total_latency_ms=latency,
            llm_calls=result.get("__llm_calls__", 0),
            tool_calls=result.get("__tool_calls__", 0),
            total_tokens=result.get("__total_tokens__", 0),
            cost_usd=result.get("__total_tokens__", 0) * 0.00015 / 1000,
            success=True
        ))
        
        return AgentResponse(
            request_id=request_id,
            response=result["messages"][-1].content,
            cost_usd=result.get("__total_tokens__", 0) * 0.00015 / 1000,
            latency_ms=latency
        )
    except Exception as e:
        # 错误处理 + 上报
        sentry_sdk.capture_exception(e)
        raise HTTPException(500, "Internal error")

@app.get("/health")
async def health():
    return {"status": "ok"}

@app.get("/metrics")
async def metrics():
    return metrics_collector.summary()

九、成本控制与优化#

成本模型#

成本 = (输入 token × $0.00015/1k) + (输出 token × $0.0006/1k) + 工具调用费

典型场景成本（gpt-4o-mini）：

成本优化策略#

策略 1：Token 控制

# 限制 max_tokens
llm = ChatOpenAI(model="gpt-4o-mini", max_tokens=500)

# 压缩上下文
def compress_context(messages: list, target_tokens: int = 2000) -> list:
    """压缩长上下文"""
    if estimate_tokens(messages) <= target_tokens:
        return messages
    
    # 保留 system message + 最近 N 条 + 摘要中间部分
    system = messages[0] if messages[0].type == "system" else None
    recent = messages[-10:]
    middle = messages[1:-10] if len(messages) > 11 else []
    
    if middle:
        summary = llm.invoke([
            {"role": "system", "content": "请用 200 字总结以下对话："},
            *middle
        ]).content
        middle = [{"role": "system", "content": f"对话摘要：{summary}"}]
    
    return [system] + middle + recent if system else middle + recent

策略 2：模型分层

# 简单任务用小模型
simple_llm = ChatOpenAI(model="gpt-4o-mini")

# 复杂任务用大模型
complex_llm = ChatOpenAI(model="gpt-4o")

# 路由
def route_query(query: str) -> ChatOpenAI:
    if is_complex_query(query):  # 多步推理、代码生成
        return complex_llm
    return simple_llm  # 默认 90% 场景

策略 3：缓存

from functools import lru_cache
import hashlib

class SemanticCache:
    """语义缓存：相似问题复用结果"""
    
    def __init__(self, similarity_threshold=0.95):
        self.cache = {}
        self.threshold = similarity_threshold
    
    def get(self, query: str, embedding_model) -> str:
        """如果缓存命中，返回缓存结果"""
        query_embedding = embedding_model.embed(query)
        
        for cached_query, (cached_embedding, cached_result) in self.cache.items():
            similarity = cosine_similarity(query_embedding, cached_embedding)
            if similarity > self.threshold:
                return cached_result
        
        return None
    
    def set(self, query: str, result: str, embedding_model):
        query_embedding = embedding_model.embed(query)
        self.cache[query] = (query_embedding, result)

十、和 L1-L9 的连接#

十一、动手时间：60 分钟部署一个生产 Agent#

步骤 1：加入 LangSmith 追踪（10 分钟）#

export LANGSMITH_API_KEY=lsv2_...
export LANGSMITH_TRACING=true

跑你的 Agent，登录 LangSmith 看 trace。

步骤 2：加指标收集（20 分钟）#

把上面的 MetricsCollector 接到你的 Agent，记录 10 次调用。

步骤 3：加 rate limit + 循环检测（15 分钟）#

测试：

• 1 秒内发 100 个请求，看 rate limit 是否生效

• 让 Agent 陷入循环，看是否能检测

步骤 4：写评测集 + CI（15 分钟）#

构造 20 个测试用例，写 CI workflow，每次 push 自动跑评测。

十二、生产检查清单#

上线前必查#

十三、小结#

十四、Read next#

下一节 L21 UX 设计原则，我们学习怎么让 Agent "用得爽"：

• 透明度设计：让用户知道 Agent 在做什么

• 流式响应：怎么提升响应速度感

• 错误恢复 UX：怎么优雅地处理失败

• HITL UX 设计：人工审批的用户体验

主要参考：

• LangSmith 文档 — LangChain 官方可观测性

• OpenTelemetry Python — OTel 官方文档

• Langfuse 文档 — 开源可观测性替代

十五、Sources#

• src-001 — Microsoft AI Agents for Beginners · T1 · 02-explore-agentic-frameworks

• LangSmith 官方文档 · T1 · LangSmith

• OpenTelemetry 官方文档 · T1 · OTel Python

• Langfuse 官方文档 · T1 · Langfuse

• HuggingFace Agents Course · T1 · Agents Course

💬 有疑问？ 试试问我："LangSmith 和 Langfuse 该选哪个？"、"我的 Agent 怎么压测？"、"生产事故怎么排查？"——我会用代码或比喻为你解答。