3.4 Token 经济学与成本控制 | Loop Engineering Token 预算管理与成本优化实战

导读

Token 成本是 Loop Engineering 最大的运营风险。一个失控的自主循环可能在数小时内消耗掉数千美元的 API 调用费用。Addy Osmani 明确警告："usage patterns can vary wildly if you are token rich or poor"——Token 的充裕程度会根本性地改变 Agent 的使用模式。本节系统讲解 Token 预算策略、成本优化技巧和效率度量体系。

学习目标

• 分析 Loop Engineering 循环中 Token 消耗的构成与热点
• 实现三种预算策略（固定预算、自适应预算、分级预算）的 Token 预算管理器
• 掌握上下文压缩、Skills 复用、精确工具选择等成本优化技术
• 建立以 Token/任务完成率为核心的效率度量体系

核心概念

Token 消耗模型

Loop Engineering 的循环本质是一个 Token 消耗放大器。每次 discover-plan-execute-verify 四阶段循环，Agent 都需要读取整个上下文窗口并生成新的输出。关键洞察：Token 消耗不是线性的，而是随迭代次数指数增长——因为每次迭代都会向上下文追加新内容。

Token 消耗增长曲线

三种预算策略

环境准备

前置知识

• 理解 LLM Token 计费模型（输入/输出分别计价）
• 了解主流模型定价（Claude、GPT-4、Gemini）
• 已完成 3.1-3.3 节学习

环境要求

• Python 3.10+
• 依赖包：pip install matplotlib（可选，用于可视化）

分步实战

步骤一：构建 Token 预算管理器

实现核心的预算管理器，支持三种策略和实时监控。

"""token_budget_manager.py — Token 预算管理器"""
import time, json, math
from dataclasses import dataclass, field
from enum import Enum

class BudgetStrategy(Enum):
    FIXED = "fixed"          # 固定预算
    ADAPTIVE = "adaptive"    # 自适应预算
    TIERED = "tiered"        # 分级预算

@dataclass
class TokenUsage:
    """单次调用的 Token 使用记录"""
    timestamp: float = field(default_factory=time.time)
    input_tokens: int = 0
    output_tokens: int = 0
    stage: str = ""           # discover/plan/execute/verify
    agent_id: str = "main"
    
    @property
    def total(self): return self.input_tokens + self.output_tokens

@dataclass
class BudgetTier:
    """分级预算中的单个层级"""
    name: str
    budget_limit: int         # Token 上限
    priority: int             # 优先级 (1=最高)
    used: int = 0
    
    @property
    def remaining(self): return max(0, self.budget_limit - self.used)
    @property
    def utilization(self): 
        return self.used / self.budget_limit if self.budget_limit > 0 else 0

class TokenBudgetManager:
    """Token 预算管理器"""
    
    def __init__(self, strategy: BudgetStrategy, total_budget: int,
                 input_price=3.0, output_price=15.0, safety_margin=0.1):
        self.strategy = strategy
        self.total_budget = total_budget
        self.input_price = input_price / 1e6   # 美元/token
        self.output_price = output_price / 1e6
        self.safety_margin = safety_margin
        self.effective_budget = int(total_budget * (1 - safety_margin))
        self.used = 0
        self.history: list[TokenUsage] = []
        self.tiers: list[BudgetTier] = []
        self.is_exhausted = False
        self.warnings_sent = 0
    
    def record_usage(self, usage: TokenUsage) -> dict:
        """记录 Token 使用并返回预算状态"""
        self.used += usage.total
        self.history.append(usage)
        
        status = self._check_budget()
        return {
            "usage": usage.total,
            "total_used": self.used,
            "remaining": max(0, self.effective_budget - self.used),
            "utilization": self.used / self.effective_budget,
            "cost_usd": self._calc_cost(),
            "status": status,
            "exhausted": self.is_exhausted
        }
    
    def _check_budget(self) -> str:
        """检查预算状态"""
        util = self.used / self.effective_budget
        
        if util >= 1.0:
            self.is_exhausted = True
            return "exhausted"
        elif util >= 0.9:
            self.warnings_sent += 1
            return "critical"
        elif util >= 0.7:
            return "warning"
        elif util >= 0.5:
            return "moderate"
        return "healthy"
    
    def _calc_cost(self) -> float:
        """计算累计成本(美元)"""
        input_total = sum(u.input_tokens for u in self.history)
        output_total = sum(u.output_tokens for u in self.history)
        return input_total * self.input_price + output_total * self.output_price
    
    def get_stage_breakdown(self) -> dict:
        """按循环阶段分解 Token 使用"""
        breakdown = {}
        for usage in self.history:
            if usage.stage not in breakdown:
                breakdown[usage.stage] = {"input": 0, "output": 0, "count": 0}
            breakdown[usage.stage]["input"] += usage.input_tokens
            breakdown[usage.stage]["output"] += usage.output_tokens
            breakdown[usage.stage]["count"] += 1
        return breakdown
    
    def estimate_remaining_iterations(self, avg_tokens_per_iter: int) -> int:
        """估算剩余可执行迭代次数"""
        remaining = max(0, self.effective_budget - self.used)
        return int(remaining / avg_tokens_per_iter) if avg_tokens_per_iter > 0 else 0
    
    def adaptive_adjust(self, progress: float) -> int:
        """自适应调整预算 (0-1的进度值)"""
        if progress < 0.3:
            # 进度不足30%但已用50%预算 → 缩减
            if self.used / self.effective_budget > 0.5:
                return int(self.effective_budget * 1.3)  # 放宽20%
        elif progress > 0.7:
            # 进度超70% → 放宽预算允许完成
            return int(self.effective_budget * 1.5)
        return self.effective_budget
    
    def set_tiers(self, tier_configs: list[dict]):
        """设置分级预算"""
        self.tiers = [BudgetTier(**cfg) for cfg in tier_configs]
    
    def allocate_to_tier(self, tier_name: str, tokens: int) -> bool:
        """从指定层级分配 Token"""
        for tier in self.tiers:
            if tier.name == tier_name:
                if tier.remaining >= tokens:
                    tier.used += tokens
                    return True
                return False
        return False


# 演示
if __name__ == "__main__":
    mgr = TokenBudgetManager(BudgetStrategy.FIXED, total_budget=100_000)
    
    # 模拟循环中的 Token 使用
    stages = ["discover","plan","execute","verify"] * 8  # 8轮循环
    for i, stage in enumerate(stages, 1):
        usage = TokenUsage(
            input_tokens=2000 + i*200,  # 上下文逐渐增长
            output_tokens=500 + i*50,
            stage=stage)
        result = mgr.record_usage(usage)
        
        print(f"迭代{i//4+1}.{stage}: +{usage.total:,} tokens | "
              f"累计:{result['total_used']:,} | "
              f"剩余:{result['remaining']:,} | "
              f"${result['cost_usd']:.4f} | [{result['status']}]")
        
        if result["exhausted"]:
            print(f"\n🛑 预算耗尽! 停止循环")
            break
    
    print(f"\n📊 阶段分解:")
    for stage, data in mgr.get_stage_breakdown().items():
        total = data["input"] + data["output"]
        print(f"  {stage}: {total:,} tokens ({data['count']}次调用)")
​

步骤二：实现成本优化策略

在预算管理的基础上，实现具体的成本优化技术。

"""cost_optimizer.py — Token 成本优化器"""
import json, re
from collections import Counter

class ContextCompressor:
    """上下文压缩器: 减少发送给LLM的Token数量"""
    
    def __init__(self, max_context_tokens: int = 8000):
        self.max_tokens = max_context_tokens
    
    def compress_history(self, messages: list[dict]) -> list[dict]:
        """压缩对话历史,保留关键信息"""
        if not messages:
            return []
        
        total = sum(self._estimate_tokens(m["content"]) for m in messages)
        if total <= self.max_tokens:
            return messages
        
        # 策略: 保留最近N条完整 + 历史摘要
        recent_count = max(2, len(messages) // 3)
        recent = messages[-recent_count:]
        older = messages[:-recent_count]
        
        # 对历史消息生成摘要
        summary = self._summarize_messages(older)
        summary_msg = {"role": "system", "content": 
            f"[历史摘要: {len(older)}条消息已压缩]\n{summary}"}
        
        return [summary_msg] + recent
    
    def _estimate_tokens(self, text: str) -> int:
        """粗略估算Token数 (中文约1.5字/token)"""
        return len(text) // 2 + len(re.findall(r'[\u4e00-\u9fff]', text)) // 2
    
    def _summarize_messages(self, messages: list[dict]) -> str:
        """提取历史消息的关键信息"""
        summaries = []
        for m in messages[-10:]:  # 只摘要最近10条
            content = m["content"]
            if len(content) > 200:
                summaries.append(content[:200] + "...")
            else:
                summaries.append(content)
        return "\n".join(summaries)


class SkillReuseOptimizer:
    """Skills 复用优化器: 避免重复加载相同的Skill内容"""
    
    def __init__(self):
        self._skill_cache: dict[str, str] = {}
        self._load_count: Counter = Counter()
    
    def load_skill(self, skill_name: str, loader_fn) -> str:
        """加载Skill, 使用缓存避免重复加载"""
        if skill_name in self._skill_cache:
            self._load_count[skill_name] += 1
            return self._skill_cache[skill_name]
        
        content = loader_fn()
        self._skill_cache[skill_name] = content
        self._load_count[skill_name] = 1
        return content
    
    def get_savings_report(self) -> dict:
        """报告复用节省的Token"""
        total_savings = 0
        details = []
        for name, count in self._load_count.items():
            if count > 1:
                content_len = len(self._skill_cache[name])
                saved = content_len * (count - 1)
                total_savings += saved
                details.append({"skill": name, "loads": count,
                               "tokens_saved": saved})
        return {"total_tokens_saved": total_savings, "details": details}


class ToolSelectionOptimizer:
    """精确工具选择: 只暴露当前阶段需要的工具"""
    
    # 按循环阶段推荐的工具集
    STAGE_TOOLS = {
        "discover": ["read", "search", "list_files"],
        "plan": ["read", "search"],
        "execute": ["read", "write", "edit", "exec"],
        "verify": ["read", "exec", "search"],
    }
    
    def get_tools_for_stage(self, stage: str) -> list[str]:
        """获取指定阶段的最小工具集"""
        return self.STAGE_TOOLS.get(stage, [])
    
    def estimate_tool_savings(self, all_tools: list[str], 
                               active_tools: list[str],
                               avg_tool_desc_tokens: int = 200) -> int:
        """估算工具选择节省的Token"""
        excluded = set(all_tools) - set(active_tools)
        return len(excluded) * avg_tool_desc_tokens


# 演示优化组合
if __name__ == "__main__":
    # 1. 上下文压缩
    compressor = ContextCompressor(max_context_tokens=5000)
    fake_messages = [{"role": "user", "content": f"消息{i}: " + "测试内容" * 50} 
                      for i in range(20)]
    compressed = compressor.compress_history(fake_messages)
    original_tokens = sum(compressor._estimate_tokens(m["content"]) for m in fake_messages)
    compressed_tokens = sum(compressor._estimate_tokens(m["content"]) for m in compressed)
    print(f"上下文压缩: {original_tokens:,} → {compressed_tokens:,} tokens "
          f"(节省{(1-compressed_tokens/original_tokens):.0%})")
    
    # 2. Skills复用
    skill_opt = SkillReuseOptimizer()
    for _ in range(5):
        skill_opt.load_skill("python-coding", lambda: "# Python编码规范...")
    report = skill_opt.get_savings_report()
    print(f"Skills复用节省: {report['total_tokens_saved']:,} tokens")
    
    # 3. 精确工具选择
    tool_opt = ToolSelectionOptimizer()
    all_tools = ["read","write","edit","exec","search","list_files",
                 "browser","upload","tts"]
    active = tool_opt.get_tools_for_stage("execute")
    saved = tool_opt.estimate_tool_savings(all_tools, active)
    print(f"工具精简(execute阶段): {len(all_tools)}→{len(active)}工具, "
          f"节省~{saved:,} tokens")
​

步骤三：建立效率度量体系

以 Token/任务完成率为核心指标，量化评估循环的经济效率。

"""efficiency_metrics.py — 循环效率度量体系"""
import time, json
from dataclasses import dataclass, field

@dataclass
class EfficiencyMetrics:
    """效率度量收集器"""
    task_name: str
    start_time: float = field(default_factory=time.time)
    total_tokens: int = 0
    total_cost_usd: float = 0.0
    iterations: int = 0
    tasks_completed: int = 0
    tasks_total: int = 1
    errors: int = 0
    
    @property
    def elapsed_minutes(self): return (time.time() - self.start_time) / 60
    
    @property
    def tokens_per_task(self):
        return self.total_tokens / self.tasks_completed if self.tasks_completed > 0 else 0
    
    @property
    def cost_per_task(self):
        return self.total_cost_usd / self.tasks_completed if self.tasks_completed > 0 else 0
    
    @property
    def completion_rate(self):
        return self.tasks_completed / self.tasks_total if self.tasks_total > 0 else 0
    
    @property  
    def tokens_per_iteration(self):
        return self.total_tokens / self.iterations if self.iterations > 0 else 0
    
    def record(self, tokens: int, cost: float, completed: bool = False):
        self.total_tokens += tokens
        self.total_cost_usd += cost
        self.iterations += 1
        if completed: self.tasks_completed += 1
    
    def summary(self) -> str:
        return (f"任务: {self.task_name}\n"
                f"  迭代: {self.iterations} | 完成: {self.tasks_completed}/{self.tasks_total}\n"
                f"  Token消耗: {self.total_tokens:,} ({self.tokens_per_task:,.0f}/任务)\n"
                f"  成本: ${self.total_cost_usd:.4f} (${self.cost_per_task:.4f}/任务)\n"
                f"  Token/迭代: {self.tokens_per_iteration:,.0f}\n"
                f"  完成率: {self.completion_rate:.0%} | 错误: {self.errors}\n"
                f"  耗时: {self.elapsed_minutes:.1f}分钟")

# 演示
if __name__ == "__main__":
    metrics = EfficiencyMetrics("API测试套件开发", tasks_total=4)
    # 模拟4个任务的执行
    for i in range(1, 5):
        tokens = 3000 + i * 2000  # 越往后Token越多
        cost = tokens * 5e-6
        metrics.record(tokens, cost, completed=True)
        print(f"任务{i}完成: {metrics.summary()}")
​

完整示例：带预算控制的循环引擎

将 Token 预算管理集成到循环引擎中，实现自动化的成本控制。

"""budget_aware_loop.py — 带预算控制的循环引擎"""
from token_budget_manager import TokenBudgetManager, TokenUsage, BudgetStrategy
from efficiency_metrics import EfficiencyMetrics
import time

class BudgetAwareLoop:
    def __init__(self, task_name: str, budget: int = 100_000,
                 max_iterations: int = 20):
        self.budget_mgr = TokenBudgetManager(BudgetStrategy.FIXED, budget)
        self.metrics = EfficiencyMetrics(task_name)
        self.max_iterations = max_iterations
    
    def run(self, step_fn):
        """step_fn(iteration) -> (stage, input_tokens, output_tokens, done)"""
        for i in range(1, self.max_iterations + 1):
            # 检查预算
            remaining = self.budget_mgr.effective_budget - self.budget_mgr.used
            if remaining < 2000:
                print(f"\n🛑 预算不足({remaining:,} tokens), 停止循环")
                break
            
            stage, in_tok, out_tok, done = step_fn(i)
            usage = TokenUsage(input_tokens=in_tok, output_tokens=out_tok, stage=stage)
            status = self.budget_mgr.record_usage(usage)
            cost = (in_tok + out_tok) * 5e-6
            self.metrics.record(in_tok + out_tok, cost, completed=done)
            
            print(f"  迭代{i}[{stage}]: {in_tok+out_tok:,} tok | "
                  f"累计:{status['total_used']:,} | "
                  f"${status['cost_usd']:.4f} | [{status['status']}]")
            
            if status["exhausted"]: break
            if done: 
                print(f"✅ 任务完成!")
                break
        
        print(f"\n{self.metrics.summary()}")
        return {"iterations": self.metrics.iterations,
                "total_tokens": self.metrics.total_tokens,
                "total_cost": f"${self.metrics.total_cost_usd:.4f}"}

if __name__ == "__main__":
    def simulate_step(i):
        stages = ["discover","plan","execute","verify"]
        stage = stages[(i-1) % 4]
        in_tok = 2000 + i * 300  # 上下文增长
        out_tok = 800 + i * 100
        done = i >= 12
        return stage, in_tok, out_tok, done
    
    loop = BudgetAwareLoop("用户认证开发", budget=80_000, max_iterations=15)
    result = loop.run(simulate_step)
    print(json.dumps(result, ensure_ascii=False))
​

常见问题FAQ

Q1：Loop Engineering 中 Token 消耗为什么不是线性的？如何预估总成本？

Token 消耗非线性的根本原因是上下文累积。每次迭代，Agent 不仅生成新的输出，还要读取之前所有的对话历史。假设单次迭代的输入是 N tokens、输出是 M tokens，那么 K 次迭代的总消耗约为 K×N + K×M + (K²/2)×M（输出成为下一次的输入）。预估方法：先用小规模试运行 3-5 次迭代，测量平均每迭代的 Token 增长率，然后按公式推算。建议预算至少设为预估值的 1.5 倍作为安全系数。

Q2：Token 贫穷时如何优化 Agent 的使用模式？Osmani 的警告具体指什么？

Osmani 的警告核心是：Token 预算的充裕程度会改变 Agent 的行为模式。Token 充裕时，你可以让 Agent 做详尽的探索、生成冗余的代码、进行多轮自我审查。Token 紧张时，你必须让 Agent 更精确——明确指定工具、限制输出长度、减少探索性步骤。具体策略：（1）上下文压缩，用摘要替代完整历史；（2）按阶段精简工具集，避免工具描述占用 Token；（3）Skills 复用，避免重复加载相同指令；（4）设定更紧凑的 max_tokens 限制。

Q3：如何对比不同模型的成本效益？是不是越便宜的模型越适合循环工程？

不一定。循环工程中模型的决策质量直接影响迭代次数——一个更贵的模型可能因为更准确的规划和更少的错误重试，总成本反而更低。建议的评估方法是：记录"任务完成所需的总 Token 数"而非"单次调用的 Token 数"。如果 Claude 3.5 Sonnet 需要 5 次迭代完成（总计 50K tokens），而更便宜的 Haiku 需要 12 次迭代（总计 80K tokens），Sonnet 反而更经济。关键指标是 Token/任务完成率。

最佳实践与避坑

✅ 最佳实践

循环开始前设定明确预算上限。 永远不要在没有预算限制的情况下启动自主循环。即使是测试环境也要设上限——失控的测试循环同样会烧钱。使用安全系数（通常 1.5x）应对不确定性。

按阶段精简工具描述。 工具的系统提示在每次 API 调用中都会发送。如果 Agent 在 execute 阶段不需要浏览器工具，就不要在工具列表中包含它。这能节省数千 tokens/次。

监控 Token/迭代比率的变化趋势。 如果这个比率在持续上升（每次迭代消耗越来越多），说明上下文在膨胀，可能需要触发上下文压缩或重新锚定。

使用分级预算管理复杂项目。 将总预算的 60% 分配给核心功能开发，25% 给测试和优化，15% 留作应急。防止早期过度消耗导致后期关键工作无法执行。

⚠️ 避坑指南

不要只看单次 API 调用成本。 循环工程的成本是累积的。一个"每次只花 0.01 美元"的循环，运行 1000 次就是 10 美元。评估总成本而非单次成本。

不要为了省 Token 而过度压缩上下文。 压缩过度会导致信息丢失，增加 Agent 犯错概率，反而因为需要更多迭代来修复错误而消耗更多 Token。找到平衡点。

不要忽视输出 Token 的成本。 在大多数模型中，输出 Token 的价格是输入的 3-5 倍。限制 Agent 的 max_tokens 和避免冗长输出比减少输入更有效。

本节小结

本节从 Token 消耗模型入手，建立了完整的 Token 经济学管理体系。我们实现了三种预算策略的管理器，掌握了上下文压缩、Skills 复用、工具精简等成本优化技术，并以 Token/任务完成率为核心建立了效率度量体系。

核心要点：Token 消耗随迭代指数增长；预算管理是循环工程的必修课；成本优化不是省钱而是让有限预算产生更多价值；效率指标应关注总成本而非单次成本。

下一节将探讨当循环中发生错误时如何恢复——错误恢复与自愈循环设计，确保循环在面对各种故障时仍能可靠推进。

延伸阅读

1. Addy Osmani, "AI Engineering: Token Economics in Autonomous Loops" — Token 经济学深度分析
2. Claude Pricing Guide, "Understanding Input/Output Token Costs" — 定价模型详解
3. OpenClaw Docs, "Cost Control and Budget Settings" — OpenClaw 的成本控制配置
4. Research: "Efficient Token Management for LLM Agents" (2025) — 学术研究
5. Boris Cherny, "Claude Code: Cost-Effective Agent Loops" — Claude Code 的成本优化实践

关键词：Token经济学、Token预算、成本控制、Loop Engineering、上下文压缩、Skills复用、工具选择优化、效率度量、自适应预算、分级预算、API成本
难度：⭐⭐⭐⭐ 中高级
预计阅读时间：25 分钟