工程落地手册
从 Demo 到 Production 的关键差异
- Demo 追求“能回答”
- Production 追求“稳定、可控、可追责”
OpenAI 生产实践文档给出的重点是:组织权限、速率限制、时延优化、成本监控。
生产级架构建议
- 网关层
- 鉴权
- 限流
- 请求审计
- 编排层
- Prompt 组装
- 工具路由
- 策略切换(快速模型/高精度模型)
- 能力层
- 模型服务
- 检索服务
- 外部工具调用
- 质量层
- 实时监控
- 评测回放
- 告警与回滚
性能与成本优化
- 优先减少无效输出 token
- 用
streaming提升首 token 体验 - 对可并行任务做批处理
- 建立模型路由,把高成本模型留给高难任务
Agent 设计建议
Anthropic 的工程文章给出一个很实用的循环:
- gather context -> take action -> verify work -> repeat
你可以把它落成三类组件:
- Context agents:负责检索与上下文整理
- Action agents:负责工具调用与执行
- Verifier agents:负责结果校验与异常处理
多 Agent 协作建议
当多个 Agent 同时开发同一仓库时,建议把冲突治理放到工程流程里,而不是依赖“临场沟通”。
- 执行隔离:一 Agent 一分支 一 worktree。
- 所有权隔离:先登记文件锁,再开始改动。
- 契约优先:共享 API/类型先合并,再并行实现。
- 合并串行化:通过 merge queue 或严格顺序合并。
详细方法见:
发布检查清单
- 是否有回归评测门禁
- 是否有速率限制与重试策略
- 是否有可观测性(日志、指标、追踪)
- 是否定义了故障降级和回滚
- 是否完成安全评审
监控指标参考值
质量指标
| 指标 | 计算方式 | 建议阈值 | 说明 |
|---|---|---|---|
| 任务成功率 | 成功响应 / 总请求 | ≥ 95% | 核心业务指标 |
| 事实准确率 | 人工抽检正确数 / 抽检总数 | ≥ 90% | 需要定期评估 |
| 格式合规率 | 格式正确 / 总输出 | ≥ 99% | 结构化输出场景 |
| 拒答率 | 拒绝回答 / 总请求 | 5-15% | 过高可能是召回问题,过低可能是安全漏洞 |
性能指标
| 指标 | 建议阈值 | P95 目标 | 说明 |
|---|---|---|---|
| 首 Token 时延 (TTFT) | ≤ 2s | ≤ 3s | 影响用户等待体验 |
| 端到端时延 | ≤ 10s | ≤ 15s | 包含检索和生成 |
| 吞吐量 | 按业务规划 | - | 并发能力 |
成本指标
| 指标 | 计算方式 | 说明 |
|---|---|---|
| 单次请求成本 | (输入 token × 输入单价) + (输出 token × 输出单价) | 按模型定价计算 |
| 每千次调用成本 | 累计成本 / 请求数 × 1000 | 便于横向对比 |
| 检索成本占比 | 检索费用 / 总费用 | 优化空间识别 |
告警规则示例
yaml
alerts:
- name: high_error_rate
condition: error_rate > 5%
duration: 5m
severity: critical
action: page_oncall
- name: high_latency
condition: p95_latency > 15s
duration: 10m
severity: warning
action: notify_team
- name: cost_spike
condition: daily_cost > baseline * 1.5
severity: warning
action: notify_team
- name: low_quality_score
condition: quality_score < 0.85
duration: 1h
severity: warning
action: trigger_review角色动作卡
开发者
- 对关键链路建立 SLO,并配置报警策略。
- 实现模型路由和兜底路径,防止单点故障。
- 把上线开关、灰度比例和回滚做成标准化配置。
产品经理
- 把“用户体验可接受阈值”写成可量化指标。
- 在发布计划中预留灰度观察窗口和复盘窗口。
- 根据业务优先级分配模型预算与资源配额。
开发者与产品经理交接件
- 发布前检查表与签字记录。
- 运行看板与异常升级路径。
- 周迭代 backlog 与优先级说明。
参考来源
- https://developers.openai.com/api/docs/guides/production-best-practices
- https://developers.openai.com/api/docs/guides/structured-outputs
- https://claude.com/blog/building-agents-with-the-claude-agent-sdk
- https://cookbook.openai.com - OpenAI Cookbook 代码示例
生产最佳实践(2026 更新)
API 密钥安全管理
python
# ❌ 错误:硬编码 API 密钥
api_key = "sk-xxx" # 永远不要这样做!
# ✅ 正确:使用环境变量
import os
api_key = os.environ.get("OPENAI_API_KEY")
# ✅ 更好:使用密钥管理服务
# AWS Secrets Manager / Azure Key Vault / HashiCorp Vault
import boto3
def get_secret(secret_name):
client = boto3.client('secretsmanager')
response = client.get_secret_value(SecretId=secret_name)
return json.loads(response['SecretString'])组织和项目管理
text
最佳实践:
├── 组织 (Organization)
│ ├── 开发项目 (Dev Project) - 用于测试和开发
│ │ ├── 独立的 API 密钥
│ │ ├── 独立的速率限制
│ │ └── 独立的使用配额
│ │
│ └── 生产项目 (Prod Project) - 用于生产环境
│ ├── 严格权限控制
│ ├── 生产级速率限制
│ └── 监控和告警水平与垂直扩展策略
yaml
水平扩展 (Horizontal Scaling):
适用场景: 高并发请求
方案:
- 部署多个服务实例
- 使用负载均衡器分发请求
- 无状态设计,支持弹性伸缩
垂直扩展 (Vertical Scaling):
适用场景: 单请求资源密集
方案:
- 增加单实例 CPU/内存
- 优化单请求处理效率
- 适用于批处理场景
缓存策略 (Caching):
类型:
- 响应缓存: 相同请求直接返回缓存结果
- Embedding 缓存: 缓存文本向量
- 提示模板缓存: 预构建常用提示
实现方式:
- 内存缓存: Redis, Memcached
- 文件缓存: 本地文件系统
- 数据库缓存: PostgreSQL, MongoDB
负载均衡 (Load Balancing):
策略:
- 轮询 (Round Robin)
- 加权轮询 (Weighted Round Robin)
- 最少连接 (Least Connections)
- 一致性哈希 (Consistent Hashing)延迟优化技术
text
请求延迟分解:
┌─────────────────────────────────────────────────────┐
│ 请求生命周期 │
├─────────────────────────────────────────────────────┤
│ │
│ 网络传输 (可优化) │
│ ├── 用户 → API 服务器 │
│ └── API 服务器 → 用户 │
│ │
│ 服务端处理 │
│ ├── 处理提示词 Token (快) │
│ └── 生成输出 Token (慢) ← 主要延迟来源 │
│ │
└─────────────────────────────────────────────────────┘
优化策略:
1. 模型选择
- 简单任务用小模型 (GPT-3.5, Claude Haiku)
- 复杂任务用大模型 (GPT-4, Claude Sonnet/Opus)
2. Token 优化
- 减少 max_tokens 限制
- 使用 stop_sequences 提前终止
- 降低 n 和 best_of 参数
3. 流式响应
- 启用 stream=True
- 降低首字节延迟 (TTFT)
- 改善用户感知体验
4. 并行处理
- 独立请求并行执行
- 批处理 API 调用生产环境检查清单
markdown
## 上线前检查
### 安全
- [ ] API 密钥使用环境变量或密钥管理服务
- [ ] 实施请求速率限制
- [ ] 配置使用限额和告警阈值
- [ ] 敏感数据脱敏处理
- [ ] 审计日志记录
### 可靠性
- [ ] 实现重试机制 (指数退避)
- [ ] 配置超时和降级策略
- [ ] 设计故障转移路径
- [ ] 准备回滚方案
### 可观测性
- [ ] 监控关键指标 (延迟、错误率、成本)
- [ ] 配置告警规则
- [ ] 日志聚合和分析
- [ ] 分布式追踪
### 性能
- [ ] 启用流式响应
- [ ] 实现缓存策略
- [ ] 优化 Token 使用
- [ ] 模型路由配置
### 测试
- [ ] 单元测试覆盖
- [ ] 集成测试
- [ ] 负载测试
- [ ] 回归测试集 (金银测试集)成本监控仪表板
python
# 成本计算示例
class CostMonitor:
# 模型定价 (USD per 1K tokens)
PRICING = {
"gpt-4": {"input": 0.03, "output": 0.06},
"gpt-4-turbo": {"input": 0.01, "output": 0.03},
"gpt-3.5-turbo": {"input": 0.0005, "output": 0.0015},
"claude-3-opus": {"input": 0.015, "output": 0.075},
"claude-3-sonnet": {"input": 0.003, "output": 0.015},
"claude-3-haiku": {"input": 0.00025, "output": 0.00125},
}
def calculate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
pricing = self.PRICING.get(model, {"input": 0, "output": 0})
input_cost = (input_tokens / 1000) * pricing["input"]
output_cost = (output_tokens / 1000) * pricing["output"]
return input_cost + output_cost
def estimate_monthly_cost(self, daily_requests: int, avg_input: int, avg_output: int, model: str) -> float:
daily_cost = self.calculate_cost(model, avg_input, avg_output)
return daily_cost * daily_requests * 30
# 使用示例
monitor = CostMonitor()
cost = monitor.calculate_cost("gpt-4-turbo", input_tokens=1000, output_tokens=500)
print(f"单次请求成本: ${cost:.4f}")
monthly = monitor.estimate_monthly_cost(
daily_requests=10000,
avg_input=500,
avg_output=200,
model="gpt-3.5-turbo"
)
print(f"预估月成本: ${monthly:.2f}")