跳转至

成本优化与缓存

概述

AI Agent的运行成本主要来自LLM API调用,优化成本是大规模部署的关键。本节介绍提示缓存、模型路由、Token优化等实用技术,帮助在保证质量的前提下显著降低运行成本。

提示缓存(Prompt Caching)

Anthropic 提示缓存

Anthropic 提供原生的提示缓存功能:

  • 缓存命中:输入价格降低90%
  • 缓存写入:首次缓存时价格增加25%
  • 缓存TTL:5分钟(每次命中刷新)

成本对比

操作 Claude Sonnet 正常价格 缓存命中价格 节省
输入 $3.00/1M $0.30/1M 90%
输出 $15.00/1M $15.00/1M 0%

使用场景

# 将系统提示和工具定义标记为可缓存
response = client.messages.create(
    model="claude-sonnet-4-20250514",
    system=[
        {
            "type": "text",
            "text": long_system_prompt,  # 大量系统指令
            "cache_control": {"type": "ephemeral"}
        }
    ],
    messages=[
        {"role": "user", "content": user_query}
    ]
)
# 相同系统提示的后续请求将命中缓存

OpenAI 自动缓存

OpenAI 自动缓存完全匹配的提示前缀:

  • 自动触发:无需额外配置
  • 折扣:缓存命中时输入价格减半
  • 适用模型:GPT-4o、GPT-4o mini

缓存策略设计

graph TD
    A[Agent请求] --> B{系统提示是否变化?}
    B -->|否| C[使用缓存前缀]
    B -->|是| D[更新缓存]
    C --> E[只发送新的用户消息]
    D --> E
    E --> F[LLM推理]
    F --> G[缓存新的前缀]

最大化缓存命中率

  1. 将不变的内容放在提示前部
  2. 系统提示 → 工具定义 → 少样本示例 → 对话历史 → 当前查询
  3. 保持系统提示和工具定义稳定不变

模型路由

智能路由策略

根据任务复杂度自动选择模型:

\[ \text{Model} = \begin{cases} M_{\text{cheap}} & \text{if difficulty}(task) \leq \theta_1 \\ M_{\text{medium}} & \text{if } \theta_1 < \text{difficulty}(task) \leq \theta_2 \\ M_{\text{expensive}} & \text{if difficulty}(task) > \theta_2 \end{cases} \]

路由实现

class ModelRouter:
    def __init__(self):
        self.models = {
            "simple": {"name": "gpt-4o-mini", "cost": 0.15},
            "medium": {"name": "claude-sonnet", "cost": 3.00},
            "complex": {"name": "claude-opus", "cost": 15.00},
        }

    def route(self, task):
        # 基于任务特征选择模型
        complexity = self.estimate_complexity(task)

        if complexity < 0.3:
            return self.models["simple"]
        elif complexity < 0.7:
            return self.models["medium"]
        else:
            return self.models["complex"]

    def estimate_complexity(self, task):
        """
        估计任务复杂度的因素:
        - 需要的推理步骤数
        - 是否需要工具使用
        - 是否涉及代码生成
        - 历史上类似任务的失败率
        """
        pass

级联策略(Cascade)

先用便宜模型尝试,失败时升级:

graph LR
    A[任务] --> B[小模型尝试]
    B --> C{成功?}
    C -->|是| D[返回结果]
    C -->|否| E[中等模型尝试]
    E --> F{成功?}
    F -->|是| D
    F -->|否| G[大模型处理]
    G --> D

期望成本

\[ E[C] = c_1 + (1-s_1) \cdot c_2 + (1-s_1)(1-s_2) \cdot c_3 \]

数值示例

假设 \(s_1 = 0.7, s_2 = 0.2, c_1 = \$0.01, c_2 = \$0.10, c_3 = \$0.50\)

\[ E[C] = 0.01 + 0.3 \times 0.10 + 0.3 \times 0.8 \times 0.50 = 0.01 + 0.03 + 0.12 = \$0.16 \]

直接使用大模型:$0.50。节省 68%。

Token优化

提示压缩

减少输入token数量的方法:

方法 压缩率 质量损失 适用场景
历史摘要 50-80% 长对话
工具输出截断 30-70% 低-中 大量工具输出
选择性上下文 40-60% 多文件引用
提示精简 10-30% 极低 所有场景

对话历史压缩

class ConversationCompressor:
    def compress(self, messages, max_tokens=4000):
        total_tokens = count_tokens(messages)

        if total_tokens <= max_tokens:
            return messages

        # 策略1: 保留最近N轮 + 摘要旧对话
        recent = messages[-6:]  # 最近3轮
        old = messages[:-6]

        summary = self.llm.summarize(old)

        return [
            {"role": "system", "content": f"对话历史摘要:{summary}"},
            *recent
        ]

输出截断

控制Agent每步的输出长度:

\[ \text{Output Budget} = \min(t_{\text{max}}, \frac{C_{\text{budget}} - C_{\text{current}}}{p_{\text{output}}}) \]

批处理

异步批处理

OpenAI和Anthropic都提供批处理API,价格减半:

提供商 批处理折扣 完成时间 适用场景
OpenAI Batch API 50% 24小时内 大规模离线处理
Anthropic Batch 50% 24小时内 批量分析
# OpenAI 批处理示例
batch = client.batches.create(
    input_file_id="file-xxx",
    endpoint="/v1/chat/completions",
    completion_window="24h"
)

适用场景

  • 大规模数据标注
  • 批量文档分析
  • 离线评估和测试
  • 不需要实时响应的任务

成本监控

监控指标

cost_metrics = {
    "per_task_cost": {
        "description": "每任务平均成本",
        "alert_threshold": "$1.00",
    },
    "daily_spend": {
        "description": "日总支出",
        "alert_threshold": "$100.00",
    },
    "cost_per_success": {
        "description": "每成功任务成本",
        "formula": "total_cost / successful_tasks",
    },
    "cache_hit_rate": {
        "description": "缓存命中率",
        "target": "> 60%",
    },
    "model_distribution": {
        "description": "各模型使用占比",
        "target": "cheap model > 70%",
    },
}

成本告警

\[ \text{Alert} = \begin{cases} \text{Warning} & \text{if daily\_cost} > 0.8 \times \text{budget} \\ \text{Critical} & \text{if daily\_cost} > \text{budget} \\ \text{Shutdown} & \text{if monthly\_cost} > \text{hard\_limit} \end{cases} \]

实践建议

  1. 先测量再优化:先建立成本基线,再有针对性地优化
  2. 缓存优先:提示缓存是最简单有效的优化手段
  3. 路由次之:模型路由可以显著降低平均成本
  4. 监控持续:成本监控应持续运行,及时发现异常
  5. 质量-成本平衡:不要为了降低成本过度牺牲质量

参考文献

  1. Anthropic. "Prompt Caching." 2024.
  2. OpenAI. "Batch API." 2024.
  3. Chen, L., et al. "FrugalGPT: How to Use Large Language Models While Reducing Cost." arXiv:2305.05176, 2023.
  4. Madaan, A., et al. "Automix: Automatically Mixing Language Models." arXiv:2310.12963, 2023.

交叉引用: - 成本分析 → 成本效益分析 - 部署架构 → 部署架构综述


评论 #