工具调用Token税：六层隐藏乘数全拆解

成本 = 单次调用Token数 × 单价

总Token = Σ(第i轮基础Token + 工具Schema定义Token × i + 已有工具结果Token × i)

# 原始Schema（约420 Token）
{
  "name": "query_database",
  "description": "Execute a SQL query against the configured MySQL database. Supports SELECT, INSERT, UPDATE, DELETE statements. Returns results as JSON array with column names as keys.",
  "parameters": {
    "type": "object",
    "properties": {
      "sql": {
        "type": "string",
        "description": "The SQL query to execute. Must be a valid SQL statement."
      },
      "database": {
        "type": "string",
        "description": "The database name to query. Defaults to the configured default database.",
        "enum": ["production", "staging", "analytics"]
      },
      "timeout": {
        "type": "integer",
        "description": "Query timeout in seconds. Default is 30.",
        "default": 30
      }
    },
    "required": ["sql"]
  }
}

# 压缩后Schema（约85 Token，-80%）
{
  "name": "query_database",
  "desc": "Execute SQL query on MySQL",
  "params": {"sql": "str (required)", "database": "enum[production,staging,analytics]", "timeout": "int default=30"}
}

graph TD
    A[Agent执行工具调用] --> B[工具返回原始结果]
    B --> C{结果Token数}
    C -->|小型结果<br>< 500 Token| D[直接注入上下文<br>成本可忽略]
    C -->|中型结果<br>500-5000 Token| E[选择性保留<br>摘要关键信息]
    C -->|大型结果<br>> 5000 Token| F[必须压缩<br>否则上下文溢出]
    E --> G[下一轮推理]
    F --> G
    D --> G
    G --> H{任务继续?}
    H -->|是| A
    H -->|否| I[任务完成]

    style F fill:#ff6b6b
    style E fill:#ffd93d
    style D fill:#6bff6b

# 级别1：工具结果清除（Tool-Result Clearing）
# 在工具结果被Agent消费后，用摘要替代原始结果
def clear_tool_result(context, tool_call_id):
    original = get_tool_result(context, tool_call_id)
    summary = llm_summarize(original, max_tokens=200)
    replace_tool_result(context, tool_call_id, summary)

# 级别2：选择性保留
# 只保留结构化信息，丢弃格式化文本
def selective_keep(raw_output):
    if is_file_content(raw_output):
        return extract_imports_and_signatures(raw_output)  # 保留接口，丢弃实现
    elif is_search_results(raw_output):
        return extract_top_3_with_relevance(raw_output)  # 只保留Top3
    elif is_error_message(raw_output):
        return raw_output  # 错误信息完整保留
    else:
        return llm_summarize(raw_output, max_tokens=300)

# 级别3：外部卸载（Offloading）
# 大型工具结果存储到外部，上下文只保留引用
def offload_result(raw_output, vector_store):
    doc_id = vector_store.store(raw_output, metadata={"tool": "read_file"})
    return f"[文件内容已卸载，doc_id={doc_id}，需要时可调用retrieve(doc_id)获取]"

T(i) = T_base + T_schema × N_tools + T_thinking(i) + Σ T_result(j) [j=1..i-1]

Total = Σ T(i) [i=1..N]
     ≈ N × T_base + N² × T_schema × N_tools / 2 + N × avg(T_result) × (N-1) / 2

class SlidingWindowAgent:
    def __init__(self, max_context=128000, working_ratio=0.6):
        self.max_context = max_context
        self.working_limit = int(max_context * working_ratio)
        self.long_term_memory = VectorStore()
        self.episode_summary = ""

    def after_tool_call(self, tool_result):
        current_usage = count_tokens(self.context)
        if current_usage > self.working_limit:
            # 将最早的工具结果卸载到长期记忆
            oldest = self.context.pop_oldest_tool_results(n=3)
            for result in oldest:
                self.long_term_memory.store(result)

            # 生成回合摘要替代原始上下文
            self.episode_summary = compress(
                self.episode_summary + summarize(oldest),
                max_tokens=500
            )
            self.context.inject(self.episode_summary, position="after_system_prompt")

# 一个2美元任务的典型重试风暴路径
第1轮：调用search_api → 返回空结果（-500 Token）
第2轮：调用search_api(query修改) → 返回相关结果（-800 Token）
第3轮：调用read_file(path错误) → FileNotFoundError（-300 Token）
第4轮：调用read_file(修正path) → 返回文件内容（-5,000 Token）
第5轮：调用edit_file(语法错误) → 编辑失败（-4,200 Token）
第6轮：调用edit_file(修正语法) → 成功（-4,200 Token）

实际有效Token：4,200（第6轮成功调用）
总消耗Token：15,000+
有效利用率：28%

class RetryBudgetLedger:
    def __init__(self, session_budget=500000, step_budget=3):
        self.session_budget = session_budget
        self.spent = 0
        self.step_retries = {}  # tool_name -> retry_count

    def pre_call(self, tool_name, estimated_tokens):
        # 检查会话预算
        if self.spent + estimated_tokens > self.session_budget:
            raise BudgetExhausted(
                f"Session budget exhausted: {self.spent}/{self.session_budget}"
            )

        # 检查步骤重试预算
        retries = self.step_retries.get(tool_name, 0)
        if retries >= self.step_budget:
            raise StepBudgetExhausted(
                f"Tool {tool_name} exceeded {self.step_budget} retries. "
                f"Escalating to alternative strategy."
            )

    def post_call(self, tool_name, success, tokens_used):
        self.spent += tokens_used
        if not success:
            self.step_retries[tool_name] = self.step_retries.get(tool_name, 0) + 1
        else:
            self.step_retries[tool_name] = 0  # 成功则重置

graph LR
    U[用户请求] --> TA[Tool Attention<br>中间件]
    TA --> ISO[Intent-Schema<br>Overlap评分]
    TA --> SM[Semantic<br>Matcher]
    TA --> HS[History<br>Selector]

    ISO --> TOP3[Top-K工具<br>候选集]
    SM --> TOP3
    HS --> TOP3

    TOP3 --> LLM[仅注入候选集<br>Schema到LLM]
    LLM --> CALL[工具调用]

    style TA fill:#4ecdc4
    style TOP3 fill:#ffe66d

# 原始方式：所有工具Schema全量注入
agent = Agent(mcp_servers=["mysql", "github", "slack", "filesystem"])
# 每轮注入 ~120,000 Token Schema定义

# Tool Attention方式：意图驱动的惰性加载
from tool_attention_mcp import ToolAttentionMiddleware

agent = Agent(
    mcp_servers=["mysql", "github", "slack", "filesystem"],
    middleware=[ToolAttentionMiddleware(top_k=5)]  # 每轮只注入Top-5相关工具
)
# 每轮注入 ~5,000 Token Schema定义，降低96%

Schema压缩：  减少 310,000 Token（34.8%贡献）
结果压缩：    减少 240,000 Token（27.0%贡献）  
惰性加载：    减少 160,000 Token（18.0%贡献）
协同效应：    额外减少 100,000 Token（非简单加总，存在协同）

# 最小化配置：Atlassian mcp-compressor
from mcp_compressor import compress_schemas

tools = compress_schemas(original_tools, mode="aggressive")
agent = Agent(tools=tools)

# 中等配置：Schema压缩 + 结果清除
from mcp_compressor import compress_schemas
from context_engineering import ToolResultClearer

tools = compress_schemas(original_tools, mode="balanced")
clearer = ToolResultClearer(strategy="summarize", max_tokens=200)
agent = Agent(tools=tools, context_manager=clearer)

# 全栈配置
from tool_attention_mcp import ToolAttentionMiddleware
from acon_compression import ObservationCompressor
from retry_budget import RetryBudgetLedger

middleware = [
    ToolAttentionMiddleware(top_k=8),
    ObservationCompressor(model="distilled-compressor-v2"),
    RetryBudgetLedger(session_budget=500000, step_budget=3)
]
agent = Agent(mcp_servers=all_servers, middleware=middleware)

# 多模型配置
from schema_adapter import CrossProviderSchemaAdapter

adapter = CrossProviderSchemaAdapter(
    providers=["openai", "anthropic", "google"],
    strategy="strictest_common"  # 采用最严格厂商的约束
)
unified_tools = adapter.adapt(original_tools)