Agent at Scale 成本怎么算：Claude API 团队级预算管理完整指南

月度预算 =
  日任务量
× 平均每个任务的调用次数
× 平均每次调用的输入 / 输出 Token
× 模型路由占比
× 30 天
× 安全系数

import anthropic
import time

client = anthropic.Anthropic(
    api_key="YOUR_API_KEY",
    base_url="https://gw.claudeapi.com"
)

RETRYABLE_STATUS = {529, 503, 502, 500}
MAX_RETRIES = 4
TASK_TOKEN_BUDGET = 100_000  # 单任务 token 上限

def call_with_budget(messages, model="claude-sonnet-4-6", task_tokens_used=0):
    for attempt in range(MAX_RETRIES):
        try:
            response = client.messages.create(
                model=model,
                max_tokens=4096,
                messages=messages
            )
            used = response.usage.input_tokens + response.usage.output_tokens
            task_tokens_used += used
            if task_tokens_used > TASK_TOKEN_BUDGET:
                raise RuntimeError(f"任务已消耗 {task_tokens_used} tokens，超出预算，强制终止")
            return response, task_tokens_used
        except anthropic.APIStatusError as e:
            if e.status_code not in RETRYABLE_STATUS:
                raise  # 不可重试错误直接抛出
            wait = 2 ** attempt
            print(f"[Retry {attempt+1}] status={e.status_code}，等待 {wait}s")
            time.sleep(wait)
    raise RuntimeError("超过最大重试次数")

response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=4096,
    system=[
        {
            "type": "text",
            "text": "你是一个专业的数据分析 Agent，以下是完整的工具说明和背景文档...\n" + long_system_doc,
            "cache_control": {"type": "ephemeral"}  # 标记为可缓存
        }
    ],
    messages=messages
)

def trim_messages(messages, max_turns=10):
    """保留最后 max_turns 轮对话，避免上下文无限膨胀"""
    if len(messages) > max_turns * 2:
        # 保留第一条（通常是任务初始指令）+ 最近 N 轮
        return [messages[0]] + messages[-(max_turns * 2 - 1):]
    return messages

def route_model(task_type: str) -> str:
    routing = {
        "classify": "claude-haiku-4-5-20251001",
        "summarize": "claude-haiku-4-5-20251001",
        "code_gen": "claude-sonnet-4-6",
        "planning": "claude-sonnet-4-6",
        "deep_analysis": "claude-opus-4-7",
    }
    return routing.get(task_type, "claude-sonnet-4-6")  # 默认 Sonnet

import asyncio
from asyncio import Semaphore

async def run_agent_pool(tasks, max_concurrent=5):
    """控制并发数量，避免触发限速"""
    sem = Semaphore(max_concurrent)
    
    async def run_one(task):
        async with sem:
            return await process_task(task)
    
    return await asyncio.gather(*[run_one(t) for t in tasks])

import logging
from dataclasses import dataclass, field
from typing import List

@dataclass
class AgentCallLog:
    task_id: str
    model: str
    input_tokens: int
    output_tokens: int
    cache_read_tokens: int = 0
    cache_write_tokens: int = 0
    retry_count: int = 0
    error: str = ""

    def cost_usd(self) -> float:
        """估算本次调用基础成本（美元），缓存读写以控制台账单为准"""
        price_map = {
            "claude-opus-4-7":           (4.0, 20.0),
            "claude-sonnet-4-6":         (2.4, 12.0),
            "claude-haiku-4-5-20251001": (0.8, 4.0),
        }
        inp, out = price_map.get(self.model, (2.4, 12.0))
        return (self.input_tokens * inp + self.output_tokens * out) / 1_000_000

# 使用示例
def log_call(response, task_id, model, retry_count=0):
    log = AgentCallLog(
        task_id=task_id,
        model=model,
        input_tokens=response.usage.input_tokens,
        output_tokens=response.usage.output_tokens,
        cache_read_tokens=getattr(response.usage, "cache_read_input_tokens", 0),
        cache_write_tokens=getattr(response.usage, "cache_creation_input_tokens", 0),
        retry_count=retry_count,
    )
    logging.info(f"[AgentCall] {log}")
    return log