Skip to content

工程落地手册

从 Demo 到 Production 的关键差异

  • Demo 追求“能回答”
  • Production 追求“稳定、可控、可追责”

OpenAI 生产实践文档给出的重点是:组织权限、速率限制、时延优化、成本监控。

生产级架构建议

  1. 网关层
  • 鉴权
  • 限流
  • 请求审计
  1. 编排层
  • Prompt 组装
  • 工具路由
  • 策略切换(快速模型/高精度模型)
  1. 能力层
  • 模型服务
  • 检索服务
  • 外部工具调用
  1. 质量层
  • 实时监控
  • 评测回放
  • 告警与回滚

性能与成本优化

  • 优先减少无效输出 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 与优先级说明。

参考来源


生产最佳实践(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}")