教程

费用看板 Agent（Anthropic SDK 教程）

用 Anthropic SDK 在 SandBase 上构建自监控 Agent：追踪每次 API 调用的 token 用量和成本，产出日/周费用报告，实现预算控制。

SandBase Team 作者 SandBase Team
#anthropic-sdk #cost-tracking #ai-agent #tutorials #budget-control

结论先行 — 构建一个自监控 Agent：追踪自己的每次 API 调用、实时计算成本、强制预算限制、产出日/周支出报告。使用 Anthropic SDK 在 SandBase 上调 Claude。实现按次计费文章中的预算控制模式。含完整代码。

我们在构建什么

一个 Agent：

  1. 执行有用任务（回答问题、分析数据）
  2. 记录每次 API 调用的 token 数和成本
  3. 按日/周/月预算追踪支出
  4. 按需或按计划产出费用报告
  5. 接近预算上限时自动降级

这是”可观测性优先”的 Agent 模式——Agent 理解自己的经济账并做出成本感知的决策。

为什么自监控重要

多数 Agent 对成本是盲的。API 调用发出去、token 累积、月底账单到达。这带来问题：

  • 预算惊吓： 一个推理循环中的 Agent 可以在几分钟内烧掉 $50
  • 无成本归因： 哪个任务或用户驱动了支出？
  • 无自适应行为： Agent 不知道该不该用更便宜的模型
  • 无早期预警： 等你注意到时预算已经爆了

自监控 Agent 通过让成本成为决策一等输入来解决这四个问题。

前置要求

pip install anthropic pydantic

需要：

  • SandBase API key，有权访问 Claude 模型
  • Anthropic SDK（兼容 SandBase 的 Anthropic 兼容端点）

完整 Agent 代码

核心：成本追踪基础设施

"""
费用看板 Agent — Anthropic SDK on SandBase
自监控 Agent，追踪和报告自身 API 支出。
"""

import json
import time
from datetime import datetime
from dataclasses import dataclass
from typing import Optional
from anthropic import Anthropic

# --- 配置 ---
SANDBASE_API_KEY = "your-sandbase-api-key"
DAILY_BUDGET_USD = 5.00
WEEKLY_BUDGET_USD = 25.00
MONTHLY_BUDGET_USD = 80.00

# --- Anthropic 客户端（通过 SandBase） ---
client = Anthropic(
    base_url="https://api.sandbase.ai/anthropic",
    api_key=SANDBASE_API_KEY,
)

# --- 定价表（每百万 token，2026） ---
PRICING = {
    "claude-sonnet-5": {"input": 3.00, "output": 15.00},
    "claude-haiku-4": {"input": 0.80, "output": 4.00},
    "claude-opus-5": {"input": 15.00, "output": 75.00},
}

# --- 数据模型 ---
@dataclass
class APICallLog:
    timestamp: str
    model: str
    input_tokens: int
    output_tokens: int
    cache_read_tokens: int = 0
    cache_write_tokens: int = 0
    cost_usd: float = 0.0
    task_label: str = ""
    duration_ms: int = 0

@dataclass
class BudgetState:
    daily_spent: float = 0.0
    weekly_spent: float = 0.0
    monthly_spent: float = 0.0
    daily_limit: float = 5.00
    weekly_limit: float = 25.00
    monthly_limit: float = 80.00
    total_calls: int = 0
    last_reset_daily: str = ""
    last_reset_weekly: str = ""

# --- 成本计算器 ---
class CostCalculator:
    @staticmethod
    def calculate(model: str, input_tokens: int, output_tokens: int,
                  cache_read_tokens: int = 0, cache_write_tokens: int = 0) -> float:
        pricing = PRICING.get(model, PRICING["claude-haiku-4"])
        input_cost = (input_tokens / 1_000_000) * pricing["input"]
        output_cost = (output_tokens / 1_000_000) * pricing["output"]
        # 缓存读取比普通输入便宜 90%
        cache_read_cost = (cache_read_tokens / 1_000_000) * pricing["input"] * 0.1
        # 缓存写入比普通输入贵 25%
        cache_write_cost = (cache_write_tokens / 1_000_000) * pricing["input"] * 1.25
        return input_cost + output_cost + cache_read_cost + cache_write_cost

成本感知 Agent 封装

class CostDashboardAgent:
    """追踪自身支出并强制预算的 Agent。"""
    
    def __init__(self, model: str = "claude-sonnet-5"):
        self.model = model
        self.logs: list[APICallLog] = []
        self.budget = BudgetState(
            daily_limit=DAILY_BUDGET_USD,
            weekly_limit=WEEKLY_BUDGET_USD,
            monthly_limit=MONTHLY_BUDGET_USD,
            last_reset_daily=datetime.now().strftime("%Y-%m-%d"),
            last_reset_weekly=datetime.now().strftime("%Y-%W"),
        )
    
    def _check_budget(self) -> tuple[bool, str]:
        """检查是否在预算内。"""
        today = datetime.now().strftime("%Y-%m-%d")
        week = datetime.now().strftime("%Y-%W")
        
        if self.budget.last_reset_daily != today:
            self.budget.daily_spent = 0.0
            self.budget.last_reset_daily = today
        if self.budget.last_reset_weekly != week:
            self.budget.weekly_spent = 0.0
            self.budget.last_reset_weekly = week
        
        if self.budget.daily_spent >= self.budget.daily_limit:
            return False, f"日预算已用完 (${self.budget.daily_spent:.2f}/${self.budget.daily_limit:.2f})"
        if self.budget.weekly_spent >= self.budget.weekly_limit:
            return False, f"周预算已用完"
        if self.budget.monthly_spent >= self.budget.monthly_limit:
            return False, f"月预算已用完"
        return True, "OK"
    
    def _select_model(self, task_complexity: str = "normal") -> str:
        """根据预算压力动态选择模型。"""
        remaining_daily = self.budget.daily_limit - self.budget.daily_spent
        
        # 日预算用了 > 80%，降级到便宜模型
        if remaining_daily < self.budget.daily_limit * 0.2:
            print(f"  ⚠ 预算压力：切换到 claude-haiku-4")
            return "claude-haiku-4"
        
        # 简单任务始终用便宜模型
        if task_complexity == "simple":
            return "claude-haiku-4"
        
        return self.model
    
    def call(self, messages: list[dict], task_label: str = "",
             task_complexity: str = "normal", system: str = None) -> Optional[str]:
        """带成本追踪和预算强制的 API 调用。"""
        
        # 预算检查
        within_budget, reason = self._check_budget()
        if not within_budget:
            print(f"  🛑 已阻止：{reason}")
            return None
        
        # 模型选择
        model = self._select_model(task_complexity)
        
        # 发起调用
        start = time.time()
        kwargs = {"model": model, "max_tokens": 1024, "messages": messages}
        if system:
            kwargs["system"] = system
        
        response = client.messages.create(**kwargs)
        duration_ms = int((time.time() - start) * 1000)
        
        # 提取用量
        usage = response.usage
        input_tokens = usage.input_tokens
        output_tokens = usage.output_tokens
        cache_read = getattr(usage, 'cache_read_input_tokens', 0) or 0
        cache_write = getattr(usage, 'cache_creation_input_tokens', 0) or 0
        
        # 计算成本
        cost = CostCalculator.calculate(model, input_tokens, output_tokens, cache_read, cache_write)
        
        # 记录
        log = APICallLog(
            timestamp=datetime.now().isoformat(),
            model=model,
            input_tokens=input_tokens,
            output_tokens=output_tokens,
            cache_read_tokens=cache_read,
            cache_write_tokens=cache_write,
            cost_usd=cost,
            task_label=task_label,
            duration_ms=duration_ms,
        )
        self.logs.append(log)
        
        # 更新预算
        self.budget.daily_spent += cost
        self.budget.weekly_spent += cost
        self.budget.monthly_spent += cost
        self.budget.total_calls += 1
        
        # 实时成本输出
        print(f"  💰 ${cost:.4f} | {model} | {input_tokens}→{output_tokens} tok | {duration_ms}ms | {task_label}")
        
        return response.content[0].text
    
    def get_daily_report(self) -> dict:
        """生成今日支出报告。"""
        today = datetime.now().strftime("%Y-%m-%d")
        today_logs = [l for l in self.logs if l.timestamp.startswith(today)]
        
        model_costs = {}
        for log in today_logs:
            if log.model not in model_costs:
                model_costs[log.model] = {"calls": 0, "cost": 0.0, "tokens": 0}
            model_costs[log.model]["calls"] += 1
            model_costs[log.model]["cost"] += log.cost_usd
            model_costs[log.model]["tokens"] += log.input_tokens + log.output_tokens
        
        task_costs = {}
        for log in today_logs:
            label = log.task_label or "unlabeled"
            task_costs[label] = task_costs.get(label, 0) + log.cost_usd
        top_tasks = sorted(task_costs.items(), key=lambda x: x[1], reverse=True)[:5]
        
        total_cost = sum(l.cost_usd for l in today_logs)
        
        return {
            "period": f"日报 ({today})",
            "total_cost": total_cost,
            "total_calls": len(today_logs),
            "avg_cost_per_call": total_cost / len(today_logs) if today_logs else 0,
            "model_breakdown": model_costs,
            "top_tasks": [{"task": t, "cost": c} for t, c in top_tasks],
            "budget_remaining": {
                "daily": self.budget.daily_limit - self.budget.daily_spent,
                "weekly": self.budget.weekly_limit - self.budget.weekly_spent,
                "monthly": self.budget.monthly_limit - self.budget.monthly_spent,
            },
            "recommendations": self._generate_recommendations(today_logs),
        }
    
    def _generate_recommendations(self, logs: list[APICallLog]) -> list[str]:
        """生成成本优化建议。"""
        recs = []
        if not logs:
            return ["暂无数据，无法生成建议。"]
        
        expensive_calls = [l for l in logs if l.model == "claude-opus-5"]
        if expensive_calls:
            avg_output = sum(l.output_tokens for l in expensive_calls) / len(expensive_calls)
            if avg_output < 200:
                recs.append(f"建议对短回复使用 claude-haiku-4。{len(expensive_calls)} 次 Opus 调用平均仅 {avg_output:.0f} 输出 token。")
        
        total_input = sum(l.input_tokens for l in logs)
        total_cache_read = sum(l.cache_read_tokens for l in logs)
        if total_input > 10000 and total_cache_read < total_input * 0.1:
            recs.append("缓存利用率低 (<10%)。对重复系统提示启用 prompt caching 可降低最多 90% 缓存内容成本。")
        
        daily_pct = self.budget.daily_spent / self.budget.daily_limit
        if daily_pct > 0.7:
            recs.append(f"日预算已使用 {daily_pct*100:.0f}%。考虑推迟非紧急任务或切换到更便宜模型。")
        
        return recs if recs else ["支出在正常范围内。"]

运行 Agent

def demo():
    """演示费用看板 Agent。"""
    
    agent = CostDashboardAgent(model="claude-sonnet-5")
    
    print("🤖 费用看板 Agent — 演示")
    print(f"   日预算：${DAILY_BUDGET_USD}")
    print(f"   模型：claude-sonnet-5（压力下自动降级）")
    print()
    
    # 任务 1：简单问题（应使用便宜模型）
    print("📌 任务 1：简单问题")
    result = agent.call(
        messages=[{"role": "user", "content": "2+2 等于几？"}],
        task_label="simple_math",
        task_complexity="simple"
    )
    print(f"   → {result}\n")
    
    # 任务 2：复杂分析
    print("📌 任务 2：复杂分析")
    result = agent.call(
        messages=[{"role": "user", "content": 
            "分析运行 100 个 AI Agent 的成本影响，"
            "每个每天 50 次 API 调用，平均 $0.01/次。"
            "包含月度预估和优化策略。"}],
        task_label="cost_analysis",
        task_complexity="normal"
    )
    print(f"   → {result[:100]}...\n")
    
    # 任务 3：缓存系统提示
    system_prompt = "你是专注于 API 成本优化的财务分析 AI。始终提供具体金额和百分比。"
    
    print("📌 任务 3：缓存系统提示复用")
    for i, question in enumerate([
        "GPT-4.1 和 Claude Sonnet 5 的成本差异是什么？",
        "Prompt caching 在重复查询上能省多少钱？",
    ]):
        result = agent.call(
            messages=[{"role": "user", "content": question}],
            system=system_prompt,
            task_label=f"financial_q{i+1}",
            task_complexity="normal"
        )
        print(f"   → {result[:80]}...\n")
    
    # 生成报告
    print("\n" + "="*60)
    print("📊 日费用报告")
    print("="*60)
    report = agent.get_daily_report()
    print(f"  周期：{report['period']}")
    print(f"  总成本：${report['total_cost']:.4f}")
    print(f"  总调用：{report['total_calls']}")
    print(f"  平均成本/调用：${report['avg_cost_per_call']:.4f}")
    print(f"\n  模型分解：")
    for model, stats in report['model_breakdown'].items():
        print(f"    {model}：{stats['calls']} 次，${stats['cost']:.4f}")
    print(f"\n  按成本排序的任务：")
    for task in report['top_tasks']:
        print(f"    {task['task']}：${task['cost']:.4f}")
    print(f"\n  剩余预算：")
    for period, remaining in report['budget_remaining'].items():
        print(f"    {period}：${remaining:.2f}")
    print(f"\n  建议：")
    for rec in report['recommendations']:
        print(f"    → {rec}")

if __name__ == "__main__":
    demo()

预算控制的三个层级

Agent 实现了按次计费文章中的三层成本控制：

层级 1：硬预算限制

if self.budget.daily_spent >= self.budget.daily_limit:
    return None  # 拒绝发起调用

预算耗尽后 Agent 不再发起 API 调用。防止失控。

层级 2：自适应模型选择

if remaining_daily < self.budget.daily_limit * 0.2:
    return "claude-haiku-4"  # 比 Sonnet 便宜 4 倍

预算压力增大时自动降级到更便宜模型。质量优雅退化而非完全停止。

层级 3：任务感知路由

if task_complexity == "simple":
    return "claude-haiku-4"  # 简单任务不浪费昂贵模型

简单问题无论预算状态都路由到便宜模型。最便宜的优化——零实现成本，在简单调用上节省 60-80%。

缓存定价集成

Agent 追踪 Anthropic 缓存定价以展示真实节省：

# 缓存读取：比普通输入便宜 90%
cache_read_cost = (cache_read_tokens / 1_000_000) * pricing["input"] * 0.1

# 缓存写入：首次使用加 25% 溢价
cache_write_cost = (cache_write_tokens / 1_000_000) * pricing["input"] * 1.25

对有重复系统提示的 Agent（大多数生产 Agent），缓存利用可降低输入成本 90%。费用报告暴露此指标让你知道缓存是否生效。

有无自监控的成本对比

场景无监控有监控节省
正常一周$25.00$18.0028%（模型路由）
预算飙升（循环 bug）$150+（月底账单才发现）$5.00（硬限制触发）97%
低流量日$15.00（始终用同一模型）$4.00（自动降级）73%
重复查询$8.00（无缓存）$2.40（缓存命中）70%

最大节省来自防止失控支出和自动将简单任务路由到便宜模型。自监控本身的开销（每次追踪调用 ~$0.001）可忽略。

生产部署模式

import schedule

def run_production_agent():
    """带定时报告的生产部署。"""
    
    agent = CostDashboardAgent(model="claude-sonnet-5")
    
    schedule.every().day.at("09:00").do(
        lambda: send_report(agent.get_daily_report(), channel="slack")
    )
    schedule.every().monday.at("09:00").do(
        lambda: send_report(generate_weekly_report(agent), channel="email")
    )
    
    while True:
        schedule.run_pending()
        task = get_next_task()
        if task:
            agent.call(
                messages=[{"role": "user", "content": task.prompt}],
                task_label=task.label,
                task_complexity=task.complexity
            )
        time.sleep(1)

要点总结

  1. 成本是 Agent 一等输入。 Agent 应该知道自己花了多少并相应调整行为。
  2. 三层控制： 硬限制（防灾）、自适应模型（优雅降级）、任务路由（消除浪费）。
  3. 缓存可见性很重要。 不能测量缓存命中就不能优化。见 Anthropic 缓存定价深度解析。
  4. 报告闭环。 日/周报告让成本从惊吓变成受管理的指标。
  5. SandBase 上的 Anthropic SDK 与直连 Anthropic 工作完全一致——相同 SDK、相同模式、相同缓存行为——加上跨所有模型的统一计费。

猜你喜欢

Cover image for 用 OpenAI SDK 搭社媒监控 Agent
教程

用 OpenAI SDK 搭社媒监控 Agent

用 OpenAI SDK 做 LLM 分析 + requests 直接调 SandBase 社媒数据 API 搭建社交监控 Agent。完整代码教程。

Cover image for 多模态 Agent + Artifacts（E2B 沙盒）
教程

多模态 Agent + Artifacts（E2B 沙盒）

教程：构建多模态 Agent，用 Seedream 生成图片、写分析代码、在 E2B 沙盒中执行、产出报告 Artifact。展示 SandBase 生态组合：多模态模型 + 沙盒 + LLM 一体化工作流。

Cover image for RAG 成本结构拆解：嵌入 + 搜索 + LLM
教程

RAG 成本结构拆解：嵌入 + 搜索 + LLM

教程式 RAG 管道成本分解：嵌入、搜索和 LLM 三层组件的真实数字。100 万文档的计算、优化策略、以及 RAG 何时比长上下文更划算。