【学习笔记】成本优化实战——用最少的钱驾驭最强的 Agent- 13/15

xian_wwq

88人浏览 · 2026-07-27 09:05:12

xian_wwq · 2026-07-27 09:05:12 发布

先说一个真实案例。

Peter Steinberger，PSPDFKit 的创始人，把 Claude API 接进自己的日常开发工作流，用于代码审查、文档生成、调试辅助。一个月后收到账单：$20,000。

没有做什么异常的事，只是让 Agent 持续跑，每次调用都用最好的模型，上下文不加控制，工具输出不做优化。结果就是这样。

这不是极端案例。是 Agent 成本不加控制的自然结果。

一、190 倍的价格差距

在说优化策略之前，先理解成本结构。

当前主流模型的价格（per million tokens）大约是：

模型	输入价格	输出价格
Claude Opus 4.7	$15	$75
Claude Sonnet 4.6	$3	$15
Claude Haiku 4.5	$0.08	$0.4

Opus 和 Haiku 的输入价格差：187 倍。输出价格差：187 倍。

这不是说 Haiku 比 Opus 差 187 倍——对于简单任务，Haiku 的质量完全够用。差的是模型能力，不是「简单任务的完成质量」。

一个典型的 Agent 工作流，如果盲目用 Opus 做所有步骤，可能花 $5。用模型路由后，只有关键步骤用 Opus，其余用 Haiku，可能花 $0.3。节省 94%，质量无损。

二、策略一：模型路由（最高杠杆）

模型路由是成本优化里杠杆最高的一个——一行配置可能带来 80% 的成本降低。

核心思路：按任务复杂度分配模型。简单的用小模型，复杂的才用大模型。

任务复杂度分类框架：

from enum import Enum
from anthropic import Anthropic

class TaskComplexity(Enum):
    SIMPLE = "simple"      # Haiku
    MEDIUM = "medium"      # Sonnet
    COMPLEX = "complex"    # Opus

def classify_task(task: str) -> TaskComplexity:
    """用 Haiku 本身来分类任务复杂度——分类成本极低"""
    client = Anthropic()
    
    response = client.messages.create(
        model="claude-haiku-4-5-20251001",  # 用最便宜的模型做分类
        max_tokens=10,
        messages=[{
            "role": "user",
            "content": f"""判断以下任务的复杂度，只回答 simple/medium/complex：

simple: 格式转换、简单问答、代码补全、信息提取
medium: 代码审查、文档撰写、逻辑分析、多步推理
complex: 架构设计、复杂调试、跨模块重构、需要深度推理

任务：{task}

只回答一个词："""
        }]
    )
    
    complexity = response.content[0].text.strip().lower()
    return TaskComplexity(complexity) if complexity in ["simple", "medium", "complex"] else TaskComplexity.MEDIUM

def route_model(complexity: TaskComplexity) -> str:
    return {
        TaskComplexity.SIMPLE: "claude-haiku-4-5-20251001",
        TaskComplexity.MEDIUM: "claude-sonnet-4-6",
        TaskComplexity.COMPLEX: "claude-opus-4-7",
    }[complexity]

def run_task(task: str) -> str:
    complexity = classify_task(task)
    model = route_model(complexity)
    
    # 实际执行任务
    client = Anthropic()
    response = client.messages.create(
        model=model,
        max_tokens=4096,
        messages=[{"role": "user", "content": task}]
    )
    return response.content[0].text

注意：分类本身也有成本，用 Haiku 做分类，每次分类大约花 0.001 美元——可以忽略不计。

预设路由规则：如果你的 Agent 工作流步骤固定，可以在 Harness 层直接预设每个步骤的模型，不需要动态分类：

# 三 Agent 代码审查流水线（第11篇案例）的模型配置
PIPELINE_MODEL_CONFIG = {
    "code_writer": "claude-sonnet-4-6",      # 代码生成：Sonnet（质量和成本平衡）
    "test_engineer": "claude-haiku-4-5-20251001",  # 测试生成：Haiku（模式化任务）
    "security_reviewer": "claude-opus-4-7",   # 安全审查：Opus（关键，不省）
}

何时不路由：安全审查、架构决策、需要深度推理的关键步骤，固定用最强模型，不要为了省钱在关键节点降级。

三、策略二：Prompt 缓存

Anthropic的 Prompt Caching 可以缓存系统提示和重复出现的上下文，命中缓存后价格降低 90%（Claude 3 系列），延迟降低约 75%。

基本原理：第一次调用时，Anthropic 缓存你标记的内容块。后续调用如果命中缓存，这部分内容不重新计算，只收取缓存读取费（约为正常价格的 10%）。

接入方式：

from anthropic import Anthropic

client = Anthropic()

# 系统提示缓存——这部分内容在多次调用间不变
SYSTEM_PROMPT = """你是一个代码审查专家...（通常 1000-3000 tokens 的详细规则）"""

response = client.messages.create(
    model="claude-opus-4-7",
    max_tokens=4096,
    system=[
        {
            "type": "text",
            "text": SYSTEM_PROMPT,
            "cache_control": {"type": "ephemeral"}  # 标记为可缓存
        }
    ],
    messages=[
        {
            "role": "user",
            "content": [
                {
                    "type": "text",
                    "text": code_to_review,
                    # 如果代码库背景也是固定的，也可以缓存
                    "cache_control": {"type": "ephemeral"}
                }
            ]
        }
    ]
)

# 查看缓存效果
usage = response.usage
print(f"缓存命中: {usage.cache_read_input_tokens} tokens")
print(f"新计算: {usage.cache_creation_input_tokens} tokens")

最大化缓存命中率的关键：缓存的内容必须完全一致（字符级别）。实践中常见的破坏缓存的原因：

每次调用在系统提示里插入时间戳（别这样做）
系统提示里有动态生成的内容
消息顺序不一致

缓存适用场景：

固定的详细系统提示（最典型）
每次调用都包含的大型参考文档
固定的代码库背景信息

不适合缓存：用户输入、每次都变化的任务上下文。

四、策略三：Prompt 压缩

当上下文里有大量冗余内容时，压缩可以显著降低 Token 消耗。

LLMLingua 是目前最成熟的 Prompt 压缩工具，可以在保留语义的前提下压缩 Prompt，最高实现 20x 压缩（但激进压缩会损失质量，实际生产中 3-5x 更稳妥）：

from llmlingua import PromptCompressor

compressor = PromptCompressor(
    model_name="microsoft/llmlingua-2-bert-base-multilingual-cased-meetingbank",
    use_llmlingua2=True
)

# 原始上下文：可能包含大量工具输出、历史对话
original_context = """
[大量工具输出、文档内容、历史对话...]
"""

# 压缩
compressed = compressor.compress_prompt(
    original_context,
    rate=0.5,          # 压缩到原始大小的 50%
    force_tokens=["\n", "。", "："]  # 保留这些 token
)

print(f"原始: {len(original_context)} chars")
print(f"压缩后: {len(compressed['compressed_prompt'])} chars")
print(f"压缩率: {compressed['ratio']:.1f}x")

压缩适用场景：

工具输出特别长（日志文件、搜索结果、数据库查询结果）
历史对话积累了大量冗余信息
参考文档里大部分内容和当前任务无关

不适合压缩的内容：

代码（压缩会破坏语法）
结构化数据（JSON、YAML）
需要精确引用的内容

更实用的替代方案——工具输出卸载（Tool Output Offloading，第4篇讲过）：工具输出不放进上下文，只保留摘要，全量内容写到文件，Agent 需要时再读取。这比压缩更简单，也更可靠：

def run_tool_with_offload(tool_name: str, args: dict) -> str:
    """运行工具，长输出卸载到文件"""
    result = run_tool(tool_name, args)
    
    if len(result) > 2000:  # 超过阈值就卸载
        # 写到文件
        output_path = f"/tmp/tool_output_{tool_name}_{timestamp()}.txt"
        with open(output_path, "w") as f:
            f.write(result)
        
        # 返回摘要 + 文件路径
        return f"""[工具输出已卸载到文件]
路径: {output_path}
摘要: {result[:500]}...
如需完整内容，读取上述文件。"""
    
    return result

五、策略四：子 Agent 隔离

这是策略一的延伸，但关注点不同：不是路由到不同模型，而是通过独立上下文控制每个子任务的 Token 规模。

问题：单 Agent 长时间运行，上下文不断积累。一个任务从 2000 tokens 的初始上下文跑到 80,000 tokens，每次 LLM 调用都要付 80,000 tokens 的钱，哪怕大部分内容和当前步骤无关。

解决方案：把大任务拆成独立的子 Agent，每个子 Agent 只获取完成子任务所需的最小上下文：

def run_subtask(subtask: str, relevant_context: str) -> str:
    """子 Agent 在干净的上下文里运行，只包含当前子任务的相关信息"""
    client = Anthropic()
    
    response = client.messages.create(
        model="claude-haiku-4-5-20251001",  # 子任务通常用小模型
        max_tokens=2048,
        messages=[{
            "role": "user",
            "content": f"""上下文：
{relevant_context}

任务：
{subtask}"""
        }]
    )
    return response.content[0].text

# 主 Agent 拆分任务，分发给子 Agent
def run_with_isolation(big_task: str, codebase_info: dict) -> str:
    subtasks = decompose_task(big_task)  # 分解任务
    results = []
    
    for subtask in subtasks:
        # 只传递和这个子任务相关的上下文
        relevant_context = extract_relevant_context(subtask, codebase_info)
        result = run_subtask(subtask, relevant_context)
        results.append(result)
    
    # 汇总结果（也可以用小模型做）
    return summarize_results(results)

成本对比：

单 Agent 处理 5 个子任务，上下文累积到 60,000 tokens：5 次调用 × 60,000 tokens = 300,000 tokens
子 Agent 隔离，每个子任务 5,000 tokens 上下文：5 次调用 × 5,000 tokens = 25,000 tokens
节省 92%

六、策略五：上下文管理

上下文管理是贯穿整个 Harness 的基础策略，前面几篇都提过，这里从成本视角重新整理。

Progressive Disclosure（渐进式披露）：系统提示里只包含 Agent 完成当前任务所需的信息，不把所有可能用到的内容一次性塞进去：

❌ 错误做法：把 100 个工具的完整文档都放进系统提示
✓ 正确做法：系统提示只包含工具目录（每个工具一行描述），Agent 按需读取完整文档

Compaction 时机优化：不要等上下文快溢出才压缩（那时候已经花了很多 token）。设置一个更早的阈值触发压缩：

def should_compact(usage: dict) -> bool:
    """在使用 60% 时就触发压缩，而不是等到 90%"""
    context_limit = 200000  # Claude 的上下文限制
    return usage["input_tokens"] > context_limit * 0.6

工具描述精简：工具描述越长，每次调用的 Token 成本越高。精简工具描述，去掉不必要的例子和解释：

# 冗长的工具描述（每次调用增加 300 tokens）
TOOL_READ_VERBOSE = {
    "name": "read_file",
    "description": "读取指定路径的文件内容。支持文本文件和代码文件。可以指定起始行和结束行来读取部分内容。如果文件不存在会返回错误。建议在读取大文件时指定行范围...",
}

# 精简后（每次调用只增加 50 tokens）
TOOL_READ_CONCISE = {
    "name": "read_file",
    "description": "读取文件内容。参数：path（必填），start_line/end_line（可选，指定行范围）。",
}

七、成本监控与预算控制

优化策略落地之后，还需要监控系统来确认效果，并防止成本失控。

7.1 实时成本追踪：每次 LLM 调用后记录成本，按任务类型和模型归因：

import json
from datetime import datetime

def track_cost(model: str, usage, task_type: str):
    # 当前价格（per million tokens）
    PRICING = {
        "claude-opus-4-7":         {"input": 15.0,  "output": 75.0},
        "claude-sonnet-4-6":       {"input": 3.0,   "output": 15.0},
        "claude-haiku-4-5-20251001": {"input": 0.08, "output": 0.4},
    }
    
    prices = PRICING.get(model, {"input": 3.0, "output": 15.0})
    
    # 计算成本（缓存命中按 10% 计算）
    cache_read = getattr(usage, "cache_read_input_tokens", 0)
    normal_input = usage.input_tokens - cache_read
    
    cost = (
        (normal_input / 1_000_000) * prices["input"] +
        (cache_read / 1_000_000) * prices["input"] * 0.1 +  # 缓存读取 10%
        (usage.output_tokens / 1_000_000) * prices["output"]
    )
    
    # 写日志
    log_entry = {
        "timestamp": datetime.now().isoformat(),
        "model": model,
        "task_type": task_type,
        "input_tokens": usage.input_tokens,
        "output_tokens": usage.output_tokens,
        "cache_read_tokens": cache_read,
        "cost_usd": round(cost, 6),
    }
    
    with open("cost_log.jsonl", "a") as f:
        f.write(json.dumps(log_entry) + "\n")
    
    return cost

7.2 预算硬顶告警：设置每日 / 每周预算上限，超限停止调用：

def check_budget(daily_limit_usd: float = 10.0) -> bool:
    """检查今日成本是否超限"""
    today = datetime.now().strftime("%Y-%m-%d")
    today_cost = 0.0
    
    try:
        with open("cost_log.jsonl") as f:
            for line in f:
                entry = json.loads(line)
                if entry["timestamp"].startswith(today):
                    today_cost += entry["cost_usd"]
    except FileNotFoundError:
        return True  # 没有日志，允许继续
    
    if today_cost >= daily_limit_usd:
        print(f"⚠️ 今日成本 ${today_cost:.2f} 已达上限 ${daily_limit_usd}，停止执行")
        return False
    
    return True

7.3 每周成本报告：汇总按模型和任务类型的成本分布，找出最贵的调用类型：

# 简单的 jq 分析
cat cost_log.jsonl | jq -r '[.model, .task_type, .cost_usd] | @csv' | \
  awk -F',' '{sum[$1","$2] += $3; count[$1","$2]++} 
  END {for (k in sum) print k, sum[k], count[k]}' | \
  sort -t' ' -k3 -rn | head -20

八、延迟优化：成本和速度的双赢

成本和延迟往往可以同时优化——用小模型不只省钱，也更快。

Claude Haiku vs Opus 的延迟差距：Haiku 的首 token 时间约 200-400ms，Opus 约 1000-2000ms。在工具密集的 Agent 工作流里，如果有 20 次 LLM 调用，这个差距会被放大 20 倍。

并行化：如果多个子任务没有依赖关系，并行运行而不是顺序运行。第11篇里 TestEngineer 和 SecurityReviewer 并行跑，不等待彼此——这不只省时间，也在单位时间内完成了更多工作：

import asyncio
from anthropic import AsyncAnthropic

async def run_parallel_agents(tasks: list[dict]) -> list[str]:
    client = AsyncAnthropic()
    
    async def run_one(task: dict) -> str:
        response = await client.messages.create(
            model=task["model"],
            max_tokens=task.get("max_tokens", 2048),
            messages=[{"role": "user", "content": task["prompt"]}]
        )
        return response.content[0].text
    
    # 并行执行所有任务
    results = await asyncio.gather(*[run_one(t) for t in tasks])
    return list(results)

# 使用
results = asyncio.run(run_parallel_agents([
    {"model": "claude-haiku-4-5-20251001", "prompt": "生成测试用例..."},
    {"model": "claude-opus-4-7", "prompt": "安全审查..."},
]))

快速路由：用 Haiku 做任务分类（< 500ms），决定用哪个模型执行——路由本身的延迟可以接受，换来的是后续调用的大幅提速。