Agent省钱六栈：Token压缩工程从$47K到$450全拆解

Total Agent Cost = Model API (80-90%) + Infrastructure (5-10%) + Tool Calls (3-5%) + Storage (1-2%)

graph TB
    subgraph "Agent省钱六栈架构"
        A[用户请求] --> B["栈1: Provider缓存<br/>Prompt Caching<br/>降本10-90%"]
        B --> C["栈2: 语义缓存<br/>Semantic Cache<br/>降本20-62%"]
        C --> D["栈3: 模型路由<br/>Model Router<br/>降本30-70%"]
        D --> E["栈4: 上下文压缩<br/>Context Compression<br/>降本40-60%"]
        E --> F["栈5: 预算硬限<br/>Token Budget<br/>止损100%"]
        F --> G["栈6: 推测执行<br/>Speculative Exec<br/>加速2-3x"]
        G --> H[LLM调用]
    end

    subgraph "监控层"
        I[Token计数器] --> J[预算比较器]
        J --> K{超限?}
        K -->|是| L[熔断/降级]
        K -->|否| M[继续执行]
    end

    style B fill:#2ecc71,color:#000
    style C fill:#3498db,color:#fff
    style D fill:#9b59b6,color:#fff
    style E fill:#e67e22,color:#fff
    style F fill:#e74c3c,color:#fff
    style G fill:#1abc9c,color:#000

import anthropic

client = anthropic.Anthropic()

# 启用Prompt Caching——只需加cache_control参数
response = client.messages.create(
    model="claude-sonnet-4-6-20250514",
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": "你是一个专业的代码审查助手...",  # 大段固定System Prompt
            "cache_control": {"type": "ephemeral"}  # 关键：标记可缓存
        }
    ],
    messages=[{"role": "user", "content": "审查这段代码..."}]
)
# 缓存命中时input token按10%计费
# Anthropic: input $3/MTok → cache hit $0.30/MTok

from langcache import LangCache
import os

# Redis LangCache配置
cache = LangCache(
    redis_url=os.environ["REDIS_URL"],
    similarity_threshold=0.92,  # 相似度阈值——调高=更精确但命中率低
    ttl=3600,  # 缓存1小时
    embedding_model="text-embedding-3-small"
)

def agent_call_with_cache(query: str, system_prompt: str):
    # 先查缓存
    cached = cache.get(query=query)
    if cached:
        return cached  # 命中！跳过LLM调用

    # 未命中，调用LLM
    response = llm.invoke(system_prompt + query)

    # 写入缓存
    cache.set(query=query, response=response)
    return response

graph LR
    A[请求进入] --> B{路由策略}
    B -->|基于规则| C["简单关键词/长度匹配<br/>准确率70%<br/>延迟+0ms"]
    B -->|基于分类器| D["小模型分类<br/>准确率85-90%<br/>延迟+50-100ms"]
    B -->|基于级联| E["先试小模型<br/>不满意再升级<br/>准确率95%+<br/>延迟+0-2000ms"]

    C --> F["Haiku/Flash-Lite<br/>$0.10-1/MTok"]
    C --> G["Sonnet/Flash<br/>$1-5/MTok"]
    C --> H["Opus/GPT-4o<br/>$5-25/MTok"]

from enum import Enum
from dataclasses import dataclass

class Complexity(Enum):
    SIMPLE = "haiku"      # $0.25/MTok input
    MODERATE = "sonnet"   # $3/MTok input  
    COMPLEX = "opus"      # $15/MTok input

@dataclass
class RouterConfig:
    max_simple_tokens: int = 500      # 少于500 token → 简单
    tool_call_keywords: list = None   # 涉及工具调用关键词 → 中等
    multi_step_keywords: list = None  # 多步推理关键词 → 复杂

def route_request(query: str, history_len: int, tools_needed: int) -> str:
    """三级路由：Token长度 + 工具数 + 历史上下文"""
    score = 0
    if len(query.split()) > 200: score += 2
    if history_len > 10: score += 2
    if tools_needed > 3: score += 3

    if score <= 2: return Complexity.SIMPLE.value
    elif score <= 5: return Complexity.MODERATE.value
    else: return Complexity.COMPLEX.value

# 级联路由：先用Haiku试，不满意再升级
def cascade_route(query: str):
    haiku_result = call_llm("haiku", query)
    if confidence_score(haiku_result) > 0.85:
        return haiku_result  # 50-70%的请求在此截留

    sonnet_result = call_llm("sonnet", query)
    if confidence_score(sonnet_result) > 0.92:
        return sonnet_result  # 再截留20-30%

    return call_llm("opus", query)  # 只有10-15%需要最强模型

class ContextCompressor:
    """渐进式上下文压缩器"""

    def __init__(self, max_recent_turns: int = 5, summary_model: str = "haiku"):
        self.max_recent_turns = max_recent_turns
        self.summary_model = summary_model
        self.summary = ""
        self.recent_turns = []

    def add_turn(self, role: str, content: str):
        self.recent_turns.append({"role": role, "content": content})

        if len(self.recent_turns) > self.max_recent_turns:
            # 压缩旧轮次为摘要
            old_turns = self.recent_turns[:-self.max_recent_turns]
            self.summary = self._compress(old_turns)
            self.recent_turns = self.recent_turns[-self.max_recent_turns:]

    def _compress(self, turns: list) -> str:
        """用小模型做摘要——成本仅$0.001/次"""
        text = "\n".join(f"{t['role']}: {t['content']}" for t in turns)
        prompt = f"将以下对话历史压缩为关键信息摘要，保留决策、结论、数据点：\n{text}"
        return call_llm(self.summary_model, prompt)

    def get_context(self) -> list:
        """返回压缩后的上下文"""
        messages = []
        if self.summary:
            messages.append({"role": "system", "content": f"之前对话摘要: {self.summary}"})
        messages.extend(self.recent_turns)
        return messages

    def token_count(self) -> int:
        """估算当前Token数"""
        import tiktoken
        enc = tiktoken.encoding_for_model("gpt-4o")
        ctx = self.get_context()
        return sum(len(enc.encode(m["content"])) for m in ctx)

# 使用：20轮对话从200K Token压缩到30K
compressor = ContextCompressor(max_recent_turns=5, summary_model="haiku")

graph TB
    subgraph "Token预算硬限三层体系"
        A["L1: 单次调用预算<br/>Per-Action Cap<br/>每次LLM调用≤N tokens"]
        B["L2: 单Agent预算<br/>Per-Agent Budget<br/>每个Agent任务≤M tokens"]
        C["L3: 全局预算<br/>Fleet-Level Budget<br/>所有Agent总计≤K tokens/天"]
    end

    A -->|超出| D["截断输出/简化Prompt"]
    B -->|超出| E["强制切换到廉价模型"]
    C -->|超出| F["全局限流/暂停服务"]

    style D fill:#f39c12,color:#000
    style E fill:#e74c3c,color:#fff
    style F fill:#c0392b,color:#fff

import time
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class TokenBudget:
    """三级Token预算执行器"""
    per_action_limit: int = 4096       # 单次调用上限
    per_task_limit: int = 50000        # 单任务上限
    daily_limit: int = 2_000_000       # 日上限

    used_per_task: int = 0
    used_daily: int = 0

    def check_and_enforce(self, estimated_tokens: int) -> tuple[bool, str]:
        """返回(是否允许, 降级策略)"""
        # L3: 日预算检查
        if self.used_daily >= self.daily_limit:
            return False, "HARD_STOP: 日预算耗尽，暂停所有Agent"

        # L2: 任务预算检查
        if self.used_per_task + estimated_tokens > self.per_task_limit:
            remaining = self.per_task_limit - self.used_per_task
            if remaining < 1000:
                return False, "TASK_STOP: 任务预算耗尽，返回当前结果"
            return True, "DOWNGRADE: 切换到Haiku完成剩余工作"

        # L1: 单次调用检查
        if estimated_tokens > self.per_action_limit:
            return True, "TRUNCATE: 截断输入到{self.per_action_limit} tokens"

        return True, "OK"

    def record_usage(self, tokens: int):
        self.used_per_task += tokens
        self.used_daily += tokens

    def reset_task(self):
        self.used_per_task = 0

    def reset_daily(self):
        self.used_daily = 0

# 使用
budget = TokenBudget(per_action_limit=4096, per_task_limit=50000)
allowed, strategy = budget.check_and_enforce(estimated_tokens=6000)
if not allowed:
    handle_budget_exceeded(strategy)
elif "DOWNGRADE" in strategy:
    model = "haiku"  # 自动降级

import asyncio
from typing import Any

class SpeculativeExecutor:
    """推测执行器：并行运行最可能的下一步"""

    def __init__(self, confidence_threshold: float = 0.7):
        self.confidence_threshold = confidence_threshold
        self.waste_rate = 0  # 追踪浪费率

    async def execute_with_speculation(
        self, 
        current_step: Any, 
        predicted_next_steps: list[tuple[str, float]]  # (步骤, 置信度)
    ):
        # 执行当前步骤
        current_task = asyncio.create_task(self._run_step(current_step))

        # 同时启动高置信度推测步骤
        speculative_tasks = {}
        for step_name, confidence in predicted_next_steps:
            if confidence >= self.confidence_threshold:
                speculative_tasks[step_name] = asyncio.create_task(
                    self._run_step(step_name)
                )

        # 等待当前步骤完成
        result = await current_task

        # 检查推测是否命中
        next_step = determine_next_step(result)  # 根据结果确定真正的下一步
        if next_step in speculative_tasks:
            try:
                spec_result = await asyncio.wait_for(
                    speculative_tasks[next_step], timeout=0.1
                )
                self.waste_rate = 0  # 命中
                return result, spec_result  # 两步结果一起返回！
            except asyncio.TimeoutError:
                pass

        # 推测未命中——取消所有推测任务
        for task in speculative_tasks.values():
            task.cancel()
        self.waste_rate += 1

        return result, None

原始成本: $47,000/月
├── + Provider缓存 (82%↓): $47K → $8,460
│   ├── + 语义缓存 (45%↓): $8,460 → $4,653
│   │   ├── + 模型路由 (60%↓): $4,653 → $1,861
│   │   │   ├── + 上下文压缩 (50%↓): $1,861 → $931
│   │   │   │   ├── + 预算硬限 (止损): $931 → $931 (防止再涨)
│   │   │   │   │   ├── + 推测执行 (延迟-50%): Token成本+10% → $1,024

Agent经济健康度仪表盘:
├── Cost Per Successful Task (CPST) — 每成功任务成本
├── Token Efficiency Ratio — 有效Token / 总Token消耗
├── Cache Hit Rate — 缓存命中率（目标>60%）
├── Model Distribution — 各模型调用占比（目标: Haiku>50%）
├── Budget Utilization — 预算利用率（目标: 70-90%）
└── Waste Rate — 浪费率（推测失败、超限重试等，目标<15%）