3 天烧掉 $2,800 到日均 $87：GPT-5.6 成本失控治理方案

# 缓存命中率监控示例
# 命中率低于 50% 说明上下文设计有问题
curl -s https://api.openai.com/v1/usage?date=2026-07-10 \
  -H "Authorization: Bearer $OPENAI_API_KEY" | \
  python3 -c "
import json, sys
data = json.load(sys.stdin)
cache_hit = data.get('cached_tokens', 0)
total_input = data.get('input_tokens', 1)
hit_rate = cache_hit / total_input * 100 if total_input else 0
print(f'缓存命中率: {hit_rate:.1f}%')
if hit_rate < 50:
    print('⚠️  告警: 缓存命中率低于 50%，建议检查上下文设计')
"

1. Agent 决定调用 grep_log 工具查日志
2. 工具返回 5000 行日志（约 20000 token）
3. 20000 token 全部塞回模型上下文
4. 模型读完，决定再调用 stats 工具
5. stats 返回 200 行（约 800 token）
6. 800 token 又塞回上下文
7. 上下文已经 20800+ token，下一步继续累加

from openai import OpenAI
from datetime import datetime
import redis
import logging

logger = logging.getLogger(__name__)
redis_client = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)

classBudgetAwareOpenAIClient:
"""带预算熔断的 OpenAI 客户端"""

def__init__(self, daily_budget_usd: float = 100.0):
self.client = OpenAI()
self.daily_budget = daily_budget_usd
self.today_key = f"gpt56:cost:{datetime.now().strftime('%Y%m%d')}"

def_get_used_today(self) -> float:
        used = redis_client.get(self.today_key)
returnfloat(used) if used else0.0

def_add_cost(self, cost: float) -> None:
        redis_client.incrbyfloat(self.today_key, cost)
        redis_client.expire(self.today_key, 7 * 24 * 3600)

def_estimate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
        pricing = {
"gpt-5.6-sol":   {"input": 5.0,  "output": 30.0},
"gpt-5.6-terra": {"input": 2.5,  "output": 15.0},
"gpt-5.6-luna":  {"input": 1.0,  "output": 6.0},
        }
        p = pricing.get(model, pricing["gpt-5.6-sol"])
return (input_tokens * p["input"] + output_tokens * p["output"]) / 1_000_000

defresponses_create(self, model: str, input: str, **kwargs) -> any:
        est_input = int(len(input.split()) * 1.3)
        est_output = kwargs.get("max_output_tokens", 2000)
        est_cost = self._estimate_cost(model, est_input, est_output)

        used = self._get_used_today()
if used + est_cost > self.daily_budget:
            logger.warning(f"预算熔断! 已用=${used:.2f}, 预估=${est_cost:.4f}")
raise BudgetExceededException(
f"今日预算已用 ${used:.2f} / ${self.daily_budget:.2f}，已熔断"
            )

        response = self.client.responses.create(model=model, input=input, **kwargs)

ifhasattr(response, 'usage'):
            actual_cost = self._estimate_cost(
                model, response.usage.input_tokens, response.usage.output_tokens
            )
self._add_cost(actual_cost)

return response


classBudgetExceededException(Exception):
pass

# GPT-5.6 Multi-Agent 成本监控
# 文件: /etc/prometheus/rules/gpt56-cost-rules.yml
groups:
-name:gpt56_cost_monitoring
interval:60s
rules:
# 实时成本指标
-record:gpt56_daily_cost_usd
expr:|
          sum by (model) (
            rate(gpt56_input_tokens_total[1h]) * on(model) group_left()
            (
              gpt56_model_pricing{type="input"}
              + gpt56_model_pricing{type="output"} * 0.3
            )
          ) * 24

# 预算使用率
-record:gpt56_budget_usage_ratio
expr:|
          gpt56_daily_cost_usd_total / on() gpt56_daily_budget_usd

# 缓存命中率
-record:gpt56_cache_hit_rate
expr:|
          sum(rate(gpt56_cached_tokens_total[5m])) /
          sum(rate(gpt56_input_tokens_total[5m]))

# ============ 告警规则 ============

# 告警 1: 日成本超预算 80%
-alert:GPT56BudgetWarning
expr:gpt56_budget_usage_ratio>0.8
for:5m
labels:
severity:warning
team:ops
annotations:
summary:"GPT-5.6 日成本已达预算 80%"

# 告警 2: 日成本超预算 100%（P1）
-alert:GPT56BudgetExceeded
expr:gpt56_budget_usage_ratio>1.0
for:2m
labels:
severity:critical
team:ops
annotations:
summary:"GPT-5.6 日成本已超预算！已触发熔断"

# 告警 3: 缓存命中率低
-alert:GPT56LowCacheHitRate
expr:gpt56_cache_hit_rate<0.5
for:10m
labels:
severity:warning
team:dev
annotations:
summary:"GPT-5.6 缓存命中率低于 50%"

from openai import OpenAI
from pydantic import BaseModel, Field
from typing importLiteral
from prometheus_client import Counter, Gauge, Histogram, start_http_server
import logging, time

logger = logging.getLogger(__name__)
client = OpenAI()

# Prometheus 指标
INPUT_TOKENS = Counter('gpt56_input_tokens_total', '输入 Token', ['model'])
OUTPUT_TOKENS = Counter('gpt56_output_tokens_total', '输出 Token', ['model'])
SUBAGENT_STATUS = Gauge('gpt56_subagent_status', 'Agent 状态', 
                        ['coordinator_id', 'name', 'status'])

classSubAgentConfig(BaseModel):
    name: str
    role: str
    model: Literal["gpt-5.6-sol", "gpt-5.6-terra", "gpt-5.6-luna"] = "gpt-5.6-terra"
    max_input_tokens: int = 50000
    max_output_tokens: int = 10000

classMultiAgentOrchestrator:
def__init__(self, coordinator_id: str, total_budget_usd: float = 5.0):
self.coordinator_id = coordinator_id
self.total_budget = total_budget_usd
self.used_budget = 0.0
self.sub_agents: list[SubAgentConfig] = []

defadd_sub_agent(self, config: SubAgentConfig) -> None:
self.sub_agents.append(config)

defrun_sub_agent(self, config: SubAgentConfig, task: str) -> str:
# 标记运行中
        SUBAGENT_STATUS.labels(
            coordinator_id=self.coordinator_id,
            name=config.name, status="running"
        ).set(1)

        response = client.responses.create(
            model=config.model,
input=f"[角色: {config.role}]\n任务: {task}",
            max_output_tokens=config.max_output_tokens,
        )

ifhasattr(response, 'usage'):
            usage = response.usage
            INPUT_TOKENS.labels(model=config.model).inc(usage.input_tokens)
            OUTPUT_TOKENS.labels(model=config.model).inc(usage.output_tokens)

        SUBAGENT_STATUS.labels(
            coordinator_id=self.coordinator_id,
            name=config.name, status="completed"
        ).set(1)
return response.output_text

# 使用示例
orchestrator = MultiAgentOrchestrator(coordinator_id="code-review-001", total_budget_usd=5.0)

# 按任务复杂度分配模型（关键：不要全用 Sol）
orchestrator.add_sub_agent(SubAgentConfig(name="架构分析", role="分析整体架构", model="gpt-5.6-sol"))
orchestrator.add_sub_agent(SubAgentConfig(name="安全审计", role="检查安全问题", model="gpt-5.6-sol"))
orchestrator.add_sub_agent(SubAgentConfig(name="性能分析", role="识别瓶颈", model="gpt-5.6-terra"))
orchestrator.add_sub_agent(SubAgentConfig(name="文档检查", role="检查完整性", model="gpt-5.6-luna"))

✅ 适合 Agent 落地的场景特征
  □ 重复性高（每天执行多次）
  □ 规则明确（有清晰的输入输出）
  □ 容错可接受（失败可重试或人工兜底）
  □ 成本可量化（人工成本 vs API 成本可对比）

❌ 不适合 Agent 落地的场景
  □ 一次性任务（投入产出比低）
  □ 高风险操作（误操作代价极高）
  □ 实时性要求高（Agent 响应有延迟）
  □ 数据敏感（不适合发给外部 API）

#!/bin/bash
# GPT-5.6 试点阶段环境搭建
set -e

WORKSPACE="/workspace/gpt56-pilot"
BUDGET_DAILY="50"# 日预算 $50
BUDGET_MONTHLY="1000"# 月预算 $1000

echo"搭建 GPT-5.6 试点环境..."

# 1. 创建工作目录
mkdir -p "${WORKSPACE}"/{logs,reports,scripts,config}

# 2. 配置环境变量
cat > "${WORKSPACE}/config/env.sh" << EOF
export OPENAI_API_KEY="\${OPENAI_API_KEY:?请设置}"
export GPT56_DAILY_BUDGET=${BUDGET_DAILY}
export GPT56_DEFAULT_MODEL="gpt-5.6-terra"  # 默认 Terra
export GPT56_MAX_MODEL="gpt-5.6-sol"        # 最高 Sol
EOF

echo"✅ 试点环境搭建完成"
echo"日预算: \$${BUDGET_DAILY}"
echo"默认模型: gpt-5.6-terra"
echo"下一步: source ${WORKSPACE}/config/env.sh"

#!/bin/bash
# GPT-5.6 成本失控应急预案
# 触发条件：日成本超预算 100%

WEBHOOK_URL="${DINGTALK_ALERT_WEBHOOK}"

# 1. 立即熔断：禁用 Sol 模型
echo"[应急预案] 禁用 Sol 模型，强制降级到 Terra..."
redis-cli SET "gpt56:model:disable:gpt-5.6-sol""1" EX 86400

# 2. 通知所有在线 Agent 降级
redis-cli PUBLISH "gpt56:emergency""budget_exceeded:force_downgrade"

# 3. 发送钉钉告警
curl -s "${WEBHOOK_URL}" \
  -H 'Content-Type: application/json' \
  -d '{
    "msgtype": "text",
    "text": {
      "content": "【P0 告警】GPT-5.6 成本失控！\n
已禁用 Sol 模型，请运维立即处理。\n
时间：'$(date"+%Y-%m-%d %H:%M:%S")'"
    }
  }'

echo"$(date) - 成本失控应急预案已触发" >> /var/log/gpt56-emergency.log

最小可用方案：
  - API 接入：1 天
  - 成本监控脚本：1 天
  - 预算熔断中间件：2 天
  - Prometheus + Grafana 看板：1 天
总计：5 人天 + 月度 API 成本 $500-2000

✅ 模型选型
□ 复杂任务用 Sol，日常任务用 Terra，简单任务用 Luna
□ Sol 成本占比超过 70% 触发告警
□ 默认 reasoning 用 high，不要无脑 max

✅ 成本监控
□ 部署成本监控脚本，每天生成成本报告
□ 日预算 80% 预警，100% 熔断
□ 缓存命中率低于 50% 告警
□ Prometheus + Grafana 实时看板

✅ Multi-Agent 治理
□ 任务价值 > $5 再开 Ultra
□ 子 Agent 按任务复杂度分配模型
□ 每个 Sub Agent 设 token 上限
□ 总预算设熔断阈值

✅ 应急预案
□ 成本失控自动熔断
□ Sol 模型自动降级到 Terra
□ 钉钉/企业微信 P0 告警
□ 定期演练（每月一次）