引言

将 Agent 从 Demo 推向生产环境，需要跨越一道巨大的鸿沟。2025年，我们目睹了太多 Agent 在生产环境中翻车的案例：无限循环烧掉数万美元、Prompt 注入导致数据泄露、并发请求拖垮整个系统。这份 Checklist 是用真金白银换来的经验。

一、模型层 Checklist

1.1 模型选择与配置

  • 模型版本锁定：生产环境使用明确的模型版本（如 gpt-5-2026-06-01），而非 gpt-5-latest
  • Fallback 模型配置：主模型不可用时自动切换到备用模型
  • Token 限制设置：max_tokens 根据业务场景硬性设置，防止无限生成
  • Temperature 策略：事实性任务 T=0-0.3，创意任务 T=0.7-1.0，代码生成 T=0-0.2
  • 上下文窗口管理：实现对话历史压缩策略，防止 Context Overflow
PRODUCTION_MODEL_CONFIG = {
    "primary": {
        "model": "gpt-5-2026-06-01",
        "max_tokens": 4096,
        "temperature": 0.2,
        "timeout": 30,
        "retry": {"max_attempts": 3, "backoff": "exponential"}
    },
    "fallback": {
        "model": "claude-opus-4-2026-04",
        "max_tokens": 4096,
        "temperature": 0.2,
        "timeout": 45,
    },
    "emergency": {
        "model": "gpt-4o-2026-03",
        "max_tokens": 2048,
        "temperature": 0.0,
    }
}

1.2 成本控制

  • Token 预算硬限：每个请求/用户/天的 Token 上限
  • 模型路由策略：简单任务用小模型，复杂任务用大模型
  • 缓存层：语义缓存命中率监控
  • 成本告警：单次请求成本 > $0.1 时告警
class TokenBudget:
    """Token 预算管理器"""
    
    def __init__(self, redis_client):
        self.redis = redis_client
    
    async def check_budget(
        self,
        user_id: str,
        requested_tokens: int
    ) -> bool:
        daily_key = f"budget:{user_id}:{date.today()}"
        used = int(await self.redis.get(daily_key) or 0)
        limit = await self._get_user_limit(user_id)
        
        if used + requested_tokens > limit:
            await self._notify_overrun(user_id, used, requested_tokens, limit)
            return False
        return True
    
    async def consume(
        self,
        user_id: str,
        tokens_used: int,
        cost: float
    ):
        daily_key = f"budget:{user_id}:{date.today()}"
        cost_key = f"cost:{user_id}:{date.today()}"
        
        pipe = self.redis.pipeline()
        pipe.incrby(daily_key, tokens_used)
        pipe.incrbyfloat(cost_key, cost)
        pipe.expire(daily_key, 86400 * 2)
        pipe.expire(cost_key, 86400 * 2)
        await pipe.execute()

二、Prompt 层 Checklist

  • System Prompt 注入防护：用户输入与系统指令严格分离
  • Prompt 版本管理：所有 Prompt 变更通过 Git 管理，支持 A/B 测试
  • Prompt 长度监控：System Prompt 不超过 Context Window 的 20%
  • Few-shot 示例审计：示例中不包含敏感数据
  • 输出格式约束：使用 JSON Mode 或 Structured Output
class PromptSafetyValidator:
    """Prompt 安全验证器"""
    
    INJECTION_PATTERNS = [
        r"ignore.*previous.*instructions",
        r"you.*are.*now.*a",
        r"system.*prompt.*is",
        r"reveal.*your.*instructions",
    ]
    
    def validate(self, user_input: str) -> ValidationResult:
        for pattern in self.INJECTION_PATTERNS:
            if re.search(pattern, user_input, re.IGNORECASE):
                return ValidationResult(
                    safe=False,
                    reason=f"Potential prompt injection: matched {pattern}"
                )
        
        if len(user_input) > 10000:
            return ValidationResult(
                safe=False,
                reason="Input exceeds maximum length"
            )
        
        return ValidationResult(safe=True)

三、工具层 Checklist

  • 工具输入校验：每个工具的输入参数用 Pydantic 模型校验
  • 工具超时设置：每个工具调用设置独立超时
  • 工具权限分级：只读工具自动执行，写操作需审批
  • 工具结果截断：工具输出超过 N 字符时自动摘要
  • 工具幂等性：关键工具支持重试不会产生副作用
from pydantic import BaseModel, Field
from typing import Literal

class ToolRegistry:
    """工具注册中心"""
    
    def register(self, tool: Tool):
        # 验证工具定义
        assert tool.timeout_ms <= 30000, "Tool timeout must be < 30s"
        assert tool.input_schema is not None, "Input schema required"
        assert tool.permission_level in ["read", "write", "admin"]
        
        if tool.permission_level in ["write", "admin"]:
            assert tool.requires_confirmation == True
        
        self._tools[tool.name] = tool

class WebSearchInput(BaseModel):
    query: str = Field(..., min_length=1, max_length=200)
    max_results: int = Field(default=5, ge=1, le=20)
    safe_search: bool = True

class WebSearchTool(Tool):
    name = "web_search"
    description = "Search the web for information"
    input_schema = WebSearchInput
    timeout_ms = 10000
    permission_level = "read"
    requires_confirmation = False
    idempotent = True

四、架构层 Checklist

4.1 并发与限流

  • 请求限流：QPS / 并发数 / Token/s 三维度限流
  • 队列隔离：不同优先级请求使用独立队列
  • 背压机制：下游不可用时主动拒绝请求，而非堆积
from asyncio import Semaphore, Queue
import asyncio

class AgentRequestHandler:
    def __init__(self, max_concurrent: int = 10):
        self.semaphore = Semaphore(max_concurrent)
        self.priority_queue = Queue(maxsize=1000)
        self.normal_queue = Queue(maxsize=5000)
    
    async def handle(self, request, priority: str = "normal"):
        queue = self.priority_queue if priority == "high" else self.normal_queue
        
        if queue.full():
            raise ServiceUnavailableError("Request queue full")
        
        await queue.put(request)
        
        async with self.semaphore:
            return await self._process(request)

4.2 状态管理

  • 会话持久化：对话状态可持久化到外部存储
  • 检查点机制：长流程 Agent 支持断点续传
  • 幂等 ID：每个请求分配幂等 ID，防重复执行

4.3 容灾设计

  • 多 Provider 热备：OpenAI / Anthropic / Azure 至少两家
  • 降级策略：LLM 不可用时回退到规则引擎
  • 熔断器：错误率 > 50% 时自动熔断 30s
class CircuitBreaker:
    """Agent 熔断器"""
    
    def __init__(
        self,
        failure_threshold: int = 10,
        recovery_timeout: int = 30,
        half_open_max_calls: int = 3
    ):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.half_open_max_calls = half_open_max_calls
        self._state = "closed"  # closed / open / half_open
        self._failures = 0
        self._last_failure_time = None
    
    async def call(self, func, *args, **kwargs):
        if self._state == "open":
            if time.time() - self._last_failure_time > self.recovery_timeout:
                self._state = "half_open"
            else:
                raise CircuitOpenError("Agent temporarily unavailable")
        
        try:
            result = await func(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure()
            raise

五、安全层 Checklist

  • PII 检测与脱敏：用户输入和 LLM 输出中的 PII 自动脱敏
  • 输出过滤：有害内容、不当建议的过滤层
  • 工具沙箱：代码执行工具在容器中运行，限制网络和文件访问
  • 审计日志：所有 Agent 决策记录可审计
  • 数据驻留：敏感数据不发送到 LLM，本地处理
class PIIRedactor:
    """PII 脱敏器"""
    
    PATTERNS = {
        "email": (r'[\w.-]+@[\w.-]+\.\w+', '[EMAIL]'),
        "phone": (r'\b1[3-9]\d{9}\b', '[PHONE]'),
        "id_card": (r'\b\d{17}[\dXx]\b', '[ID_CARD]'),
        "bank_card": (r'\b\d{16,19}\b', '[BANK_CARD]'),
    }
    
    def redact(self, text: str) -> tuple[str, dict]:
        """返回脱敏文本和映射表（用于恢复）"""
        mapping = {}
        redacted = text
        
        for pii_type, (pattern, replacement) in self.PATTERNS.items():
            matches = re.finditer(pattern, redacted)
            for i, match in enumerate(matches):
                placeholder = f"{replacement}_{i}"
                mapping[placeholder] = match.group()
                redacted = redacted.replace(match.group(), placeholder)
        
        return redacted, mapping

六、可观测性 Checklist

  • 全链路追踪：每个 Agent 步骤生成 Span
  • 结构化日志：所有日志 JSON 格式，包含 trace_id
  • 实时仪表盘：Grafana 仪表盘展示关键指标
  • 告警规则：延迟、错误率、成本、Token 使用量告警
  • 会话回放：支持根据 trace_id 回放完整 Agent 执行过程

七、运维层 Checklist

  • 蓝绿部署：新版本 Agent 灰度发布，支持秒级回滚
  • 配置热更新：Prompt、工具配置不重启即可更新
  • 依赖版本锁定：所有依赖版本锁定，定期安全扫描
  • Secret 管理：API Key 通过 Vault/Secret Manager 管理
  • 备份策略：对话历史、Agent 状态定期备份

八、合规层 Checklist

  • 数据保留策略：用户对话数据保留期限明确
  • GDPR/PIPL 合规：支持用户数据删除请求
  • 模型透明度：向用户说明使用了 AI 及其局限性
  • 内容免责声明：输出中标注 AI 生成内容
  • 审计合规：关键决策可追溯，满足监管要求

九、性能层 Checklist

  • P95 延迟 < 5s：首 Token 延迟 < 2s，完整响应 < 30s
  • 流式响应：长输出使用 SSE 流式返回
  • 预计算：高频请求结果预计算缓存
  • 连接池：LLM API 连接复用
  • CDN 加速：静态资源（Prompt 模板、配置）通过 CDN 分发

十、用户体验层 Checklist

  • 加载状态：Agent 思考时展示进度提示
  • 优雅降级：LLM 超时时返回友好提示而非错误码
  • 多语言支持：Agent 输出跟随用户语言
  • 反馈机制：用户可对 Agent 回答打分
  • 边界处理：处理空输入、超长输入、非预期输入

十一、测试层 Checklist

  • 单元测试覆盖率 > 80%：非 LLM 逻辑全覆盖
  • 集成测试：关键业务流程端到端测试
  • 混沌测试：主动注入 LLM 超时、工具失败
  • 安全测试：Prompt 注入、数据泄露测试
  • 负载测试：验证 10x 峰值流量下的表现

十二、上线前最终确认

□ 紧急关停开关（Kill Switch）可用且已测试
□ 回滚脚本已验证，可在 60s 内完成
□ on-call 排班已确认，告警通知链路畅通
□ 成本预算已设置硬性上限
□ 用户协议已更新，涵盖 AI 使用条款
□ 压测通过：P95 < 5s, 错误率 < 1%
□ 安全审计已完成，无高危漏洞
□ 数据备份已验证可恢复
□ 运行手册（Runbook）已编写
□ 团队培训已完成

结语

这份 Checklist 看似冗长，但每一条都来自真实的生产事故。Agent 系统的复杂度远超传统应用——它不仅有代码的确定性风险，还有模型的不确定性风险、工具的副作用风险。上线的标准不是"能跑了"，而是"出事了能兜住"。祝你的 Agent 平稳上线。

加入讨论

这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞，认知差异创造无限可能。