LLM Agent 成本控制：语义缓存与模型路由实战

大多数团队在算 Agent 成本时，只看 token 单价。但真正的浪费往往来自两个地方：同样的请求被反复调用，以及简单的任务被昂贵的模型处理。

一个典型的客服 Agent 每天处理 10 万次请求，其中 30% 是重复问题（"怎么退款""物流在哪看"）。如果每次都要调用 GPT-4o，仅这一项每月就能多烧几千美元。另一个常见问题是：分类任务用 Claude Sonnet，摘要任务用 GPT-4o，而实际上 Llama 3 70B 在相同任务上已经足够。

本文不谈"token 价格表"，而是给出一个可落地的成本控制框架：先量化浪费在哪里，再针对性部署缓存、路由和降级策略。

成本拆解：钱 actually 花在哪

在动手优化之前，先搞清楚你的 Agent 成本结构。一个 LLM 调用的总成本可以拆成三部分：

直接成本：API 调用费用。这是账单上最显眼的数字，但通常不是最大的浪费源。

间接成本：重试、超时、错误处理带来的额外调用。一个失败的请求可能触发 2-3 次重试，每次都是全额计费。

机会成本：用高能力模型处理低复杂度任务。用 GPT-4o 做简单的意图分类，相当于用法拉利送快递——能到目的地，但油费是问题。

from dataclasses import dataclass
from typing import Any


@dataclass
class CostBreakdown:
    total_requests: int = 0
    cache_hits: int = 0
    model_calls: dict[str, int] = None  # type: ignore

    def __post_init__(self):
        if self.model_calls is None:
            self.model_calls = {}

    @property
    def cache_hit_rate(self) -> float:
        if self.total_requests == 0:
            return 0.0
        return self.cache_hits / self.total_requests

    @property
    def cost_saved_by_cache(self) -> float:
        """估算缓存节省的成本（假设缓存命中完全避免 API 调用）"""
        # 简化模型：每次调用平均 $0.01
        return self.cache_hits * 0.01

    def report(self) -> dict[str, Any]:
        return {
            "total_requests": self.total_requests,
            "cache_hit_rate": f"{self.cache_hit_rate:.1%}",
            "cost_saved_monthly": f"${self.cost_saved_by_cache * 30:.2f}",
            "model_distribution": dict(self.model_calls),
        }

先跑一周的数据收集，你会发现 surprise：往往 20% 的请求模式贡献了 60% 的重复调用。

策略一：语义缓存——不只是键值匹配

传统的 HTTP 缓存基于 exact match（相同的 URL + 参数）。但 LLM 调用的输入是自然语言，用户 rarely 用完全相同的措辞重复提问。

语义缓存 的核心思想是：如果两个请求在语义上相似到足以产生相同响应，第二次请求应该直接返回缓存结果。

import hashlib
from dataclasses import dataclass, field
from typing import Any


@dataclass
class CacheEntry:
    key: str
    embedding: list[float]
    response: str
    model: str
    token_count: int
    created_at: str
    hit_count: int = 0
    similarity_threshold: float = 0.92


class SemanticCache:
    def __init__(self, embedding_fn, similarity_threshold: float = 0.92):
        self.embedding_fn = embedding_fn
        self.similarity_threshold = similarity_threshold
        self._store: dict[str, CacheEntry] = {}

    def _compute_key(self, text: str) -> str:
        return hashlib.sha256(text.encode()).hexdigest()[:16]

    def get(self, query: str) -> CacheEntry | None:
        key = self._compute_key(query)
        # 1. 先查 exact match
        if key in self._store:
            entry = self._store[key]
            entry.hit_count += 1
            return entry
        # 2. 再查语义相似
        query_emb = self.embedding_fn(query)
        for entry in self._store.values():
            sim = self._cosine_similarity(query_emb, entry.embedding)
            if sim >= self.similarity_threshold:
                entry.hit_count += 1
                return entry
        return None

    def put(self, query: str, response: str, model: str, token_count: int):
        key = self._compute_key(query)
        self._store[key] = CacheEntry(
            key=key,
            embedding=self.embedding_fn(query),
            response=response,
            model=model,
            token_count=token_count,
            created_at=__import__('datetime').datetime.now().isoformat(),
        )

    def invalidate(self, model: str, prompt_version: str):
        """当 prompt 或模型变更时，失效相关缓存"""
        to_remove = []
        for key, entry in self._store.items():
            if entry.model == model:
                to_remove.append(key)
        for key in to_remove:
            del self._store[key]

    @staticmethod
    def _cosine_similarity(a: list[float], b: list[float]) -> float:
        import math
        dot = sum(x * y for x, y in zip(a, b))
        mag_a = math.sqrt(sum(x * x for x in a))
        mag_b = math.sqrt(sum(x * x for x in b))
        return dot / (mag_a * mag_b) if mag_a and mag_b else 0.0

关键设计点：

• 双重查询策略：先 exact match（零延迟），再语义相似（覆盖同义重述）
• similarity_threshold 设为 0.92 左右——太低会返回错误答案，太高导致缓存命中率不足
• invalidate 方法确保 prompt 版本更新或模型切换时，旧缓存不会污染新响应
• 缓存 key 应该包含模型名称和 prompt 版本，避免不同配置下的响应混淆

适用场景：FAQ、常见问题、标准化流程（如退款查询、订单状态）。这些场景下用户提问高度重复，缓存命中率可达 30-50%。

工具参考：Pezzo 内置缓存层声称可节省高达 90% 的 LLM 成本和延迟；Helicone 提供代理级缓存，支持 OpenAI 兼容 API 的透明缓存；Bifrost 的语义缓存基于相似度匹配，能处理自然语言变体。

策略二：模型路由——按任务复杂度分配模型

不是所有请求都需要 GPT-4o。把任务按复杂度分层，每层分配合适的模型：

from dataclasses import dataclass
from typing import Literal


@dataclass
class ModelTier:
    name: str
    model_id: str
    cost_per_1k_tokens: float
    max_context: int
    best_for: list[str]


MODEL_TIERS: dict[str, ModelTier] = {
    "simple": ModelTier(
        name="simple",
        model_id="gpt-4o-mini",
        cost_per_1k_tokens=0.00015,
        max_context=128000,
        best_for=["intent_classification", "simple_qa", "formatting"],
    ),
    "medium": ModelTier(
        name="medium",
        model_id="gpt-4o",
        cost_per_1k_tokens=0.0025,
        max_context=128000,
        best_for=["rag_generation", "tool_planning", "multi_step_reasoning"],
    ),
    "complex": ModelTier(
        name="complex",
        model_id="claude-sonnet-4-20250514",
        cost_per_1k_tokens=0.003,
        max_context=200000,
        best_for=["complex_reasoning", "code_generation", "long_document"],
    ),
}


class ModelRouter:
    def __init__(self, classifier_fn):
        self.classifier_fn = classifier_fn
        self.call_stats: dict[str, int] = {}

    def route(self, request: dict) -> ModelTier:
        task_type = self.classifier_fn(request)
        tier = MODEL_TIERS.get(task_type, MODEL_TIERS["medium"])
        self.call_stats[tier.name] = self.call_stats.get(tier.name, 0) + 1
        return tier

    def estimate_cost(self, request: dict, estimated_tokens: int) -> float:
        tier = self.route(request)
        return (estimated_tokens / 1000) * tier.cost_per_1k_tokens

    def report(self) -> dict[str, Any]:
        total = sum(self.call_stats.values())
        return {
            tier: {
                "calls": count,
                "percentage": f"{count / total:.1%}" if total else "0%",
            }
            for tier, count in self.call_stats.items()
        }

关键设计点：

• 路由决策应该在发送请求之前做出，而不是等模型返回后再判断
• 分类器可以用一个轻量级模型（甚至规则引擎）实现，成本应该远低于被路由的请求本身
• 记录每次路由决策和实际使用的模型，用于后续优化分类器

适用场景：混合工作负载（同时有简单问答和复杂推理）、多模型环境、成本敏感的生产系统。

工具参考：Helicone 的 AI Gateway 提供智能路由和自动故障转移；Bifrost 连接 23+ LLM 提供商，支持自动负载均衡；Langfuse 的分析功能可以帮助你识别哪些请求适合路由到更便宜的模型。

策略三：Fallback Chain——成本与质量的动态平衡

有时候你不知道一个请求应该用哪个模型。这时候可以用 fallback chain：先用 cheapest 模型，如果置信度不够再升级。

from dataclasses import dataclass
from typing import Any


@dataclass
class FallbackConfig:
    primary_model: str
    fallback_model: str
    confidence_threshold: float = 0.8
    max_fallback_depth: int = 2


class FallbackChain:
    def __init__(self, primary_fn, fallback_fn, confidence_fn):
        self.primary_fn = primary_fn
        self.fallback_fn = fallback_fn
        self.confidence_fn = confidence_fn
        self.stats = {"primary_success": 0, "fallback_used": 0, "total": 0}

    def execute(self, request: dict) -> dict[str, Any]:
        self.stats["total"] += 1
        # 1. 主模型
        response = self.primary_fn(request)
        confidence = self.confidence_fn(response)
        if confidence >= self.fallback_config.confidence_threshold:
            self.stats["primary_success"] += 1
            return response
        # 2. 降级模型
        self.stats["fallback_used"] += 1
        return self.fallback_fn(request)

    def cost_analysis(self, primary_cost: float, fallback_cost: float) -> dict:
        primary_only = self.stats["total"] * primary_cost
        actual = (
            self.stats["primary_success"] * primary_cost
            + self.stats["fallback_used"] * fallback_cost
        )
        return {
            "hypothetical_primary_only": f"${primary_only:.2f}",
            "actual_with_fallback": f"${actual:.2f}",
            "savings": f"${primary_only - actual:.2f}",
            "fallback_rate": f"{self.stats['fallback_used'] / self.stats['total']:.1%}" if self.stats['total'] else "0%",
        }

关键设计点：

• confidence_fn 是核心——它需要快速、可靠地判断当前响应是否"足够好"。可以用长度、结构完整性、或一个小型分类器
• Fallback chain 的成本收益取决于 fallback 率。如果 80% 的请求在第一级就成功，你就能用 20% 的请求承担更贵的模型费用
• 避免无限 fallback——设置 max_fallback_depth，防止在多个模型间循环调用

适用场景：质量要求高但成本敏感的场景、模型能力不确定时的渐进式增强。

策略四：Prompt Caching——被忽视的成本杀手

很多团队没有意识到：system prompt 的重复传输是隐性成本。如果你的 system prompt 是 2000 tokens，每天 10 万次请求，那就是 2 亿 tokens 的重复传输。

from dataclasses import dataclass, field
from typing import Any


@dataclass
class PromptCacheConfig:
    system_prompt: str
    cache_ttl_seconds: int = 300
    max_cache_size: int = 1000


class PromptCache:
    def __init__(self, config: PromptCacheConfig):
        self.config = config
        self._cache: dict[str, Any] = {}
        self._hits = 0
        self._misses = 0

    def get_system_prompt(self, version: str) -> str:
        if version in self._cache:
            self._hits += 1
            return self._cache[version]
        self._misses += 1
        # 实际实现中这里应该从配置中心或数据库加载
        self._cache[version] = self.config.system_prompt
        return self._cache[version]

    def invalidate(self, version: str):
        self._cache.pop(version, None)

    def stats(self) -> dict[str, Any]:
        total = self._hits + self._misses
        return {
            "hits": self._hits,
            "misses": self._misses,
            "hit_rate": f"{self._hits / total:.1%}" if total else "0%",
            "tokens_saved": self._hits * len(self.config.system_prompt.split()),
        }

关键设计点：

• System prompt 通常比用户输入长得多，且变化频率低——是最值得缓存的部分
• 缓存 key 应该是 prompt 版本 hash，而不是请求 hash
• TTL 设置要平衡缓存命中率和 prompt 更新延迟

适用场景：高流量 Agent 系统、system prompt 较长（>1000 tokens）、prompt 版本管理规范。

工具参考：Langfuse 的提示词版本管理功能可以帮助你追踪 prompt 变更并配合缓存失效；Pezzo 的缓存层专门针对 prompt 重复传输优化。

成本优化决策矩阵

场景	首选策略	预期收益	实施复杂度
高重复率 FAQ	语义缓存	30-50% 调用减少	低
混合复杂度工作负载	模型路由	20-40% 成本降低	中
质量要求高但成本敏感	Fallback chain	15-30% 成本降低	中
高流量 + 长 system prompt	Prompt caching	10-20% token 节省	低
多种模型供应商	网关聚合	运维简化 + 故障转移	中

优先级建议：

1. 先部署语义缓存——最快见到效果，实施最简单
2. 再配置模型路由——需要一周数据收集来训练分类器
3. 然后加 fallback chain——在关键路径上渐进式增强
4. 最后优化 prompt caching——需要配合 prompt 版本管理

三个常见错误

错误一：缓存键太宽或太窄

缓存键如果只基于"用户 ID"，会把不同问题的回答混在一起；如果基于"完整请求文本"，又几乎没有命中。正确做法是基于"任务类型 + 核心实体"——比如 "order_status:ORD-12345" 而不是整个用户消息。

错误二：路由分类器过度复杂

很多人上来就用 LLM 做任务分类，结果分类器的成本接近被路由的请求本身。先用规则引擎（关键词、正则）做粗分类，只在边界 case 上调用轻量级模型。

错误三：忽略缓存失效

缓存最大的风险不是命中率低，而是返回过期答案。Prompt 更新、模型切换、业务规则变更时，必须有明确的缓存失效策略。建议在 CI/CD 流程中自动触发版本变更对应的缓存清理。

总结

• Agent 成本优化的第一步永远是量化。用 Langfuse 或 Helicone 追踪一周，你会发现 20% 的请求模式贡献了 60% 的浪费
• 语义缓存是投入产出比最高的策略。重复问题在客服、技术支持场景下占比可达 30-50%，缓存命中后完全避免 API 调用
• 模型路由不是"哪个便宜用哪个"，而是"哪个刚好够用用哪个"。简单任务用 gpt-4o-mini，复杂推理才上 Claude Sonnet
• Fallback chain 让质量有底线，成本有上限。先用便宜模型，置信度不够再升级，避免过度配置
• Prompt caching 是免费的午餐。System prompt 通常占 token 消耗的 40-60%，且变化频率低，缓存后几乎零成本
• Bifrost 和 Pezzo 提供了开箱即用的网关和缓存层，可以在一周内落地这些策略，不需要从零搭建

---

推荐结合 Langfuse（LLM 可观测性与成本分析）、Helicone（代理监控与缓存）、Pezzo（Prompt 管理与缓存层）和 Bifrost（LLM 网关与语义缓存）来搭建完整的 Agent 成本控制体系。