推理成本控制：Budget Forcing 与智能路由

成本构成深度分析

Budget Forcing：动态思考预算分配

import anthropic
import re
from typing import Optional

client = anthropic.Anthropic()

class ThinkingBudgetOptimizer:
    """根据问题难度动态调整 thinking budget，平衡质量与成本"""

    # 难度 → budget 映射（单位：tokens）
    BUDGET_MAP = {
        "trivial":  0,      # 禁用思考（极简问答、格式转换）
        "easy":    1000,   # 轻量推理（基础计算、简单比较）
        "medium":  5000,   # 中等推理（多步骤问题、代码审查）
        "hard":    12000,  # 深度推理（数学竞赛、复杂架构设计）
        "expert":  25000,  # 专家级（博士级科学、定理证明）
    }

    def classify_difficulty(self, question: str) -> str:
        """基于规则的难度分类（可替换为小模型分类器）"""
        q = question.lower()
        word_count = len(question.split())

        # 极简问题特征（不需要推理）
        trivial_patterns = ['翻译', 'translate', '什么是', 'how to spell']
        if word_count < 8 or any(p in q for p in trivial_patterns):
            return "trivial"

        # 专家级问题特征
        expert_keywords = ['证明', 'prove', 'theorem', '定理', '∫', '∑', '∀', '∃']
        if any(k in question for k in expert_keywords) and word_count > 30:
            return "expert"

        # 难问题特征
        hard_keywords = ['算法复杂度', '系统设计', '优化', '竞赛', '推导', '比较分析']
        has_code = '```' in question or word_count > 100
        if any(k in q for k in hard_keywords) or (has_code and word_count > 50):
            return "hard"

        # 中等问题（默认）
        if word_count > 20:
            return "medium"
        return "easy"

    def ask(self, question: str, force_difficulty: Optional[str] = None) -> dict:
        """智能提问：自动分级，返回答案和成本信息"""
        difficulty = force_difficulty or self.classify_difficulty(question)
        budget = self.BUDGET_MAP[difficulty]

        thinking_config = (
            {"type": "enabled", "budget_tokens": budget}
            if budget > 0
            else {"type": "disabled"}
        )

        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=budget + 2000,
            thinking=thinking_config,
            messages=[{"role": "user", "content": question}]
        )

        # 提取思考 token 数量
        thinking_tokens = sum(
            len(b.thinking.split())
            for b in response.content
            if hasattr(b, 'thinking')
        )

        answer = next(b.text for b in response.content if b.type == "text")
        return {
            "answer": answer,
            "difficulty": difficulty,
            "budget": budget,
            "actual_thinking_tokens": thinking_tokens,
            "total_output_tokens": response.usage.output_tokens,
        }

Prompt Caching：降低输入成本

class CachedReasoningService:
    """使用 Prompt Caching 降低重复系统提示的成本"""

    def __init__(self, system_prompt: str, budget: int = 5000):
        self.budget = budget
        # 系统提示封装为可缓存格式
        # 缓存有效：系统提示必须 ≥ 1024 tokens（Claude 的最低缓存阈值）
        self.system = [{
            "type": "text",
            "text": system_prompt,
            "cache_control": {"type": "ephemeral"}  # 标记为可缓存（5分钟TTL）
        }]

    def query(self, question: str) -> dict:
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=self.budget + 2000,
            thinking={"type": "enabled", "budget_tokens": self.budget},
            system=self.system,
            messages=[{"role": "user", "content": question}]
        )

        usage = response.usage
        # 缓存命中时，cache_read_input_tokens 费率是普通输入的 10%
        # 第一次调用：cache_creation_input_tokens > 0（写入缓存，略贵）
        # 后续调用：cache_read_input_tokens > 0（命中缓存，大幅省钱）
        is_cache_hit = usage.cache_read_input_tokens > 0

        answer = next(b.text for b in response.content if b.type == "text")
        return {
            "answer": answer,
            "cache_hit": is_cache_hit,
            "cache_read_tokens": usage.cache_read_input_tokens,
            "cache_write_tokens": usage.cache_creation_input_tokens,
            "normal_input_tokens": usage.input_tokens,
            "output_tokens": usage.output_tokens,
        }

# 成本节省估算：
# 系统提示 = 2000 tokens，每日 100 次查询
# 不缓存：100 × 2000 × $3/M = $0.60/天
# 缓存命中（90%）：9 × 2000 × $3/M（写入）+ 91 × 2000 × $0.3/M（读取）
#                = $0.054 + $0.055 = $0.109/天
# 节省：约 82%

智能模型路由：三层路由策略

from enum import Enum
from dataclasses import dataclass

class RouteConfig:
    """模型路由配置：任务类型 × 复杂度 → (模型, 思考配置)"""

    # 规则：行 = 任务类型，列 = 复杂度
    ROUTES = {
        # 对话类（不需要推理）
        ("chat",       "low"):    ("claude-haiku-4-5-20251001", None),
        ("chat",       "medium"): ("claude-sonnet-4-6", None),
        ("chat",       "high"):   ("claude-sonnet-4-6", None),

        # 推理类（需要 Extended Thinking）
        ("reasoning",  "low"):    ("claude-sonnet-4-6",
                                      {"type": "enabled", "budget_tokens": 2000}),
        ("reasoning",  "medium"): ("claude-sonnet-4-6",
                                      {"type": "enabled", "budget_tokens": 8000}),
        ("reasoning",  "high"):   ("claude-sonnet-4-6",
                                      {"type": "enabled", "budget_tokens": 20000}),

        # 数学类（使用最强的推理配置）
        ("math",       "low"):    ("claude-sonnet-4-6",
                                      {"type": "enabled", "budget_tokens": 5000}),
        ("math",       "high"):   ("claude-opus-4-6",
                                      {"type": "enabled", "budget_tokens": 32000}),

        # 代码类（中等推理）
        ("code",       "low"):    ("claude-sonnet-4-6", None),
        ("code",       "high"):   ("claude-sonnet-4-6",
                                      {"type": "enabled", "budget_tokens": 10000}),
    }

    def route(self, task_type: str, complexity: str) -> tuple:
        key = (task_type, complexity)
        return self.ROUTES.get(key,
                               ("claude-sonnet-4-6",
                                {"type": "enabled", "budget_tokens": 5000}))

# 使用示例
router = RouteConfig()
model, thinking = router.route("math", "high")
print(f"使用模型: {model}, 思考预算: {thinking}")

budget_tokens vs 质量的权衡曲线

通过实验数据，我们可以看到思考预算与质量之间的关系：

成本监控与告警

from datetime import datetime, timedelta
from collections import defaultdict

class CostMonitor:
    """推理成本实时监控"""

    # 2025 年 claude-sonnet-4-6 估算费率（美元/百万 token）
    RATES = {
        "input": 3.0,
        "cache_write": 3.75,
        "cache_read": 0.30,
        "output": 15.0,  # thinking token 也按此费率
    }

    def __init__(self, daily_budget_usd: float = 10.0):
        self.daily_budget = daily_budget_usd
        self.daily_cost = 0.0
        self.request_count = 0
        self.reset_time = datetime.now() + timedelta(days=1)

    def record_usage(self, usage) -> dict:
        """记录一次请求的用量，返回成本信息"""
        # 重置每日计数器
        if datetime.now() > self.reset_time:
            self.daily_cost = 0.0
            self.request_count = 0
            self.reset_time = datetime.now() + timedelta(days=1)

        # 计算本次成本
        cost = (
            usage.input_tokens * self.RATES["input"] +
            usage.cache_creation_input_tokens * self.RATES["cache_write"] +
            usage.cache_read_input_tokens * self.RATES["cache_read"] +
            usage.output_tokens * self.RATES["output"]
        ) / 1_000_000

        self.daily_cost += cost
        self.request_count += 1

        # 成本告警
        budget_usage = self.daily_cost / self.daily_budget
        if budget_usage > 0.9:
            print(f"警告：已使用 {budget_usage:.0%} 日预算")
        elif budget_usage > 0.7:
            print(f"提示：已使用 {budget_usage:.0%} 日预算")

        return {
            "request_cost_usd": cost,
            "daily_total_usd": self.daily_cost,
            "budget_usage_pct": budget_usage,
            "requests_today": self.request_count,
        }