无缓存: 50步 × 3K × $3/M = $0.45
有缓存: 1次写入($3/M × 3K × 1.25) + 49次读取($0.30/M × 3K × 49) = $0.053
节省: $0.40/循环，即系统提示成本降低88%

# Anthropic SDK 提示缓存实现
import anthropic

client = anthropic.Anthropic()

response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": "你是一个代码审查专家...",  # 3K token 系统提示
            "cache_control": {"type": "ephemeral"}  # 标记缓存
        }
    ],
    tools=[...],  # 工具定义也标记缓存
    messages=[{"role": "user", "content": "审查这个PR"}]
)

# 模型分层路由实现
MODEL_TIERS = {
    "orchestrator": "claude-opus-4-7",    # 复杂规划
    "specialist":   "claude-sonnet-4-6",  # 领域工作
    "formatter":    "claude-haiku-4-5",   # 格式化/提取
    "classifier":   "claude-haiku-4-5",   # 路由决策
}

def route_model(task_type: str, complexity: float) -> str:
    if complexity > 0.8:
        return MODEL_TIERS["orchestrator"]
    elif complexity > 0.4:
        return MODEL_TIERS["specialist"]
    return MODEL_TIERS["formatter"]

def prune_context(full_context: str, max_tokens: int = 10000) -> str:
    estimated_tokens = len(full_context) // 4
    if estimated_tokens <= max_tokens:
        return full_context
    # 用廉价模型做摘要压缩
    response = client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=max_tokens,
        messages=[{
            "role": "user", 
            "content": f"在{max_tokens//2}Token内总结关键事实:\n\n{full_context}"
        }]
    )
    return response.content[0].text

def smart_route(request):
    # 确定性规则优先（零Token成本）
    if "翻译" in request.intent:
        return translation_agent
    if request.file_ext in [".py", ".js"]:
        return code_review_agent
    if request.confidence < 0.7:
        return llm_router  # 仅边缘情况用LLM
    return default_agent

# 每用户预算硬上限
from datetime import datetime, timedelta

class UserBudget:
    def __init__(self, daily_limit: float = 5.0):
        self.daily_limit = daily_limit
        self.spent = {}

    def check(self, user_id: str) -> bool:
        today = datetime.now().strftime("%Y-%m-%d")
        spent_today = self.spent.get(f"{user_id}:{today}", 0)
        return spent_today < self.daily_limit

    def record(self, user_id: str, cost: float):
        today = datetime.now().strftime("%Y-%m-%d")
        key = f"{user_id}:{today}"
        self.spent[key] = self.spent.get(key, 0) + cost

import asyncio

async def optimized_fanout(task, agents, max_parallel=3):
    """控制并行度，避免Token爆炸"""
    semaphore = asyncio.Semaphore(max_parallel)

    async def bounded_run(agent):
        async with semaphore:
            return await agent.run(task)

    results = await asyncio.gather(*[
        bounded_run(agent) for agent in agents
    ])
    return results

# 多Agent编排成本估算器
base_tokens_per_call = 2_000
num_agents = 5
context_compounding = 1.4   # 每步累积~40%更多上下文
daily_tasks = 1_000

tokens_per_task = base_tokens_per_call * num_agents * context_compounding
monthly_tokens = tokens_per_task * daily_tasks * 30

# Sonnet 4.6 定价: $3输入/$15输出 每百万Token
input_cost = (monthly_tokens * 0.7 / 1_000_000) * 3
output_cost = (monthly_tokens * 0.3 / 1_000_000) * 15
raw_inference = input_cost + output_cost

# 应用1.85倍现实乘数
realistic_monthly = raw_inference * 1.85

print(f"每任务Token: {tokens_per_task:,.0f}")
print(f"月Token量: {monthly_tokens:,}")
print(f"原始推理: ${raw_inference:,.0f}")
print(f"现实月费: ${realistic_monthly:,.0f}")
# 输出: 每任务Token: 14,000 / 月Token量: 420,000,000
# 原始推理: $4,410 / 现实月费: $8,159

graph TD
    A[任务进入] --> B{任务可分解为<br/>独立子任务?}
    B -->|是| C{子任务数量>3?}
    B -->|否| D{需要专业领域<br/>切换?}

    C -->|是| E[并行扇出模式<br/>成本:2-3× 延迟:0.8×]
    C -->|否| F[顺序流水线<br/>成本:1.5-2× 延迟:2-4×]

    D -->|是| G{需要迭代<br/>优化?}
    D -->|否| H[单Agent+工具<br/>成本:1× 基准线]

    G -->|是| I[评估器-优化器<br/>成本:4-8× 延迟:可变]
    G -->|否| J[动态路由器<br/>成本:1.1× 延迟:~1×]

    E --> K{Token预算<br/>能承受?}
    F --> K
    I --> K
    J --> K

    K -->|否| L[降级:单Agent+<br/>提示工程优化]
    K -->|是| M[✅ 投入生产<br/>+缓存+分层路由]

    style H fill:#4CAF50,color:#fff
    style L fill:#FF9800,color:#fff
    style M fill:#2196F3,color:#fff
    style I fill:#F44336,color:#fff