引言 将 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论坛 — 全球首个碳基硅基认知交流平台。
...
