大模型应用一旦上线,成本通常会经历一个规律:

  • PoC 期间“还好”,上线后“越用越贵”
  • 你以为是模型太贵,实际上往往是调用次数变多、上下文变长、重试放大、重复问题没缓存

这篇文章给一套工程化的“降本三板斧”,按优先级落到可执行动作:

  1. 缓存:减少“没必要的调用”
  2. 摘要/压缩:减少“没必要的 token”
  3. 多模型路由:把“没必要用大模型的请求”交给更合适的模型

并附一个 Python 骨架,帮助你把策略嵌进真实服务,而不是停留在 PPT。


0)TL;DR(先给结论)

  • 先做可观测:没有 tokens/latency/error/retry/cache_hit/route,降本就是伪进度。
  • 先止血再精修:先加 token 预算门禁与重试治理,再做缓存与路由。
  • 降本三板斧
    • 缓存:把“重复调用”干掉(最直接省钱)
    • 摘要/压缩:把“上下文膨胀”压回预算
    • 路由:把“简单请求”交给更便宜的模型,把“高风险请求”交给更稳的模型
  • 别只看平均值:用 P50/P95 做预算与门禁,平均值会骗人。

1)第 0 步:把成本变成可观测指标(否则你不知道省了多少)

建议你至少记录这些字段(能做看板 + 告警):

  • tokeninput_tokens/output_tokens/total_tokens(P50/P95)
  • 体验latency_ms(P50/P95)
  • 稳定error_rate/retry_rate/timeout_rate
  • 缓存cache_hit_rate(按 cache 类型拆:response/retrieval/tool 等)
  • 路由route_modelroute_reasonfallback_count

建议:把“提示词版本/策略版本/路由规则版本”也打进日志,否则你会解释不清 token 为什么漂移。


2)三板斧之一:缓存(少调一次,就少花一次钱)

缓存不是只有“Redis 存一下”,工程上至少要分清四类缓存:

2.1 Response Cache(响应缓存:最高 ROI)

适用:完全相同输入→完全相同输出的场景,例如:

  • 固定模板问答(业务规则、固定说明)
  • 后台运营/客服常见问题(文本高度重复)
  • 系统内部批处理(同一 prompt 多次调用)

关键点:

  • Key 设计:对 prompt、system、关键参数做规范化,再 hash
  • TTL:结合业务更新频率(规则变更要主动失效)
  • 安全:不同用户的上下文不要混用(必须做租户隔离/权限隔离)

2.2 Tool Cache(工具结果缓存:最容易被忽略)

适用:工具调用返回“稳定数据”的场景:

  • 商品信息/配置/政策条款/组织结构等结构化数据

策略:

  • 以工具入参做 key + TTL
  • 用“字段投影”返回必要字段,避免大段 JSON 进入上下文

2.3 Retrieval Cache(检索缓存:RAG 必备)

适用:RAG 中同一查询在短时间内大量重复出现。

缓存对象可以是:

  • 召回的 chunk_id 列表
  • 重排后的 top_n chunk_id 列表

收益:减少向量检索/重排开销,间接减少总延迟与成本。

2.4 Semantic Cache(语义缓存:省钱但要控风险)

思路:输入不完全相同,但语义相似就复用答案。
风险:可能在“细节差一点”的问题上答错,建议只用于:

  • 低风险任务(非金融/非政策/非强合规)
  • 有明确“相似度阈值 + 回退策略”

3)三板斧之二:摘要/压缩(少给 token,立刻省钱)

上线后 token 膨胀的三个高发点:

  1. history:对话历史无限叠加
  2. context:RAG TopK 越调越大 / 长文档整段塞
  3. tools:工具返回把全量日志/JSON 塞回去

对应的工程手段:

3.1 历史摘要(History Summarization)

做法:保留“最近 N 轮对话 + 一段长期摘要”。

  • 摘要只在超预算时触发(别每轮都总结)
  • 摘要可以用更便宜的模型生成(任务简单)

3.2 上下文预算(Context Budget)

给上下文一个硬上限,例如:

  • context_budget_tokens = 2000
  • history_budget_tokens = 800

然后做“按重要性截断/压缩”。工程上你可以先从最简单的规则开始:

  • 先砍 tools 返回的冗余字段
  • 再砍 history 的老轮次(用摘要替代)
  • 最后再砍 context(TopK 降级或压缩)

3.3 工具返回投影/截断(Tool Projection)

不要把“全量 JSON”塞进模型上下文,建议:

  • 只保留必要字段(投影)
  • 长数组截断(TopN)
  • 大文本摘要(保留关键条款 + 引用定位)

4)三板斧之三:多模型路由(用对模型,才是长期降本)

路由不是“便宜模型/贵模型二选一”,更像一个策略系统:

  • 任务分级:低风险/中风险/高风险
  • 资源约束:预算、P95 延迟、并发上限
  • 质量兜底:失败回退到更稳的模型

4.1 一套简单可用的路由规则(示例)

你可以先用“规则路由”落地,后面再升级:

  • 高风险(合规/财务/政策/支付):直接用强模型 + 严格输出校验
  • 需要引用证据:强模型(或强检索+中等模型),并要求引用格式
  • 低风险/格式化任务(改写、摘要、分类):便宜模型即可
  • 长上下文请求:先压缩/检索治理,再决定是否上强模型

4.2 回退策略(比路由更重要)

推荐至少有两级回退:

  1. 便宜模型失败 → 强模型重试
  2. 强模型仍失败 → 降级输出(模板/缓存/兜底提示)

并把回退次数记录到日志,否则你会以为“便宜模型省钱”,实际上一直在回退。


5)最小 Python 骨架:把“缓存 + 压缩 + 路由”串起来

下面代码是骨架(可直接拷到项目里改造),重点是结构与接口,而不是某个特定 SDK。

from dataclasses import dataclass
from typing import Any, Dict, List, Optional, Tuple
import hashlib
import json
import time


@dataclass
class ChatReq:
    tenant_id: str
    messages: List[Dict[str, str]]  # [{"role":"user","content":"..."}]
    task_tag: str                   # "rewrite"/"qa"/"policy"/"extract"...
    require_citation: bool = False
    max_output_tokens: int = 512


@dataclass
class ChatResp:
    content: str
    model: str
    cached: bool
    route_reason: str
    fallback_count: int = 0


class SimpleTTLCache:
    def __init__(self):
        self._store: Dict[str, Tuple[float, Any]] = {}

    def get(self, key: str) -> Optional[Any]:
        v = self._store.get(key)
        if not v:
            return None
        expire_at, payload = v
        if time.time() > expire_at:
            self._store.pop(key, None)
            return None
        return payload

    def set(self, key: str, value: Any, ttl_sec: int):
        self._store[key] = (time.time() + ttl_sec, value)


def stable_hash(obj: Any) -> str:
    raw = json.dumps(obj, ensure_ascii=False, sort_keys=True).encode("utf-8")
    return hashlib.sha256(raw).hexdigest()


def choose_model(req: ChatReq) -> Tuple[str, str]:
    """返回 (model_name, reason)"""
    if req.task_tag in {"policy", "payment", "legal"}:
        return "strong-model", "high_risk_task"
    if req.require_citation:
        return "strong-model", "need_citation"
    if req.task_tag in {"rewrite", "summarize", "classify", "extract"}:
        return "cheap-model", "low_risk_structured_task"
    return "mid-model", "default"


def compress_messages(messages: List[Dict[str, str]], budget_chars: int = 6000) -> List[Dict[str, str]]:
    """
    这里用字符预算做占位示例;真实工程里建议用 token 预算。
    最简单策略:保留末尾最近消息,前面的做摘要(摘要可用更便宜模型生成)。
    """
    total = sum(len(m.get("content", "")) for m in messages)
    if total <= budget_chars:
        return messages
    # TODO:替换为“摘要 + 最近N轮”
    tail = messages[-6:]  # 先保留最近 6 条
    summary = {"role": "system", "content": "对话摘要:前文较长,已省略(建议用摘要模型生成更准确的摘要)。"}
    return [summary] + tail


def call_llm(model: str, messages: List[Dict[str, str]], max_tokens: int) -> str:
    """
    这里是适配层:你可以接任何 OpenAI 兼容接口或自建服务。
    我自己用的是大模型中转平台147API。
    关键是把“模型名/消息/参数”做成统一函数,方便路由与A/B。
    """
    # TODO: 替换为真实调用
    return f"[stubbed] model={model}, reply=..."


def chat(req: ChatReq, cache: SimpleTTLCache) -> ChatResp:
    # 1) 输入压缩(控 token / 控上下文)
    messages = compress_messages(req.messages)

    # 2) 选择模型(路由)
    model, reason = choose_model(req)

    # 3) 响应缓存(按租户隔离)
    cache_key = stable_hash({
        "tenant": req.tenant_id,
        "model": model,
        "messages": messages,
        "max_output_tokens": req.max_output_tokens,
    })
    cached = cache.get(cache_key)
    if cached:
        return ChatResp(content=cached, model=model, cached=True, route_reason=reason)

    # 4) 调用 + 回退
    fallback_count = 0
    try:
        content = call_llm(model=model, messages=messages, max_tokens=req.max_output_tokens)
    except Exception:
        fallback_count += 1
        content = call_llm(model="strong-model", messages=messages, max_tokens=req.max_output_tokens)
        reason = f"{reason}+fallback"

    # 5) 写缓存(示例TTL:10分钟;需结合业务更新策略)
    cache.set(cache_key, content, ttl_sec=600)
    return ChatResp(content=content, model=model, cached=False, route_reason=reason, fallback_count=fallback_count)

这份骨架的重点是把“工程控制点”显式化:

  • compress_messages:上下文预算(控 token)
  • choose_model:路由策略(控单价)
  • cache:减少重复调用(控调用次数)
  • fallback:质量兜底(控事故)

6)落地 Checklist(按优先级)

6.1 第 1 天:先止血

  • 打点:token/延迟/错误率/重试率(P50/P95)
  • token 门禁:max_input/max_output/context_budget/history_window
  • 重试治理:可重试/不可重试分类 + 指数退避

6.2 第 1 周:上缓存与摘要

  • Response Cache:相同输入直接复用
  • Tool Cache:工具结果 TTL + 字段投影
  • History Summarization:摘要 + 最近 N 轮

6.3 第 1 月:上路由与门禁

  • 任务分级:低/中/高风险
  • 路由 + 回退:默认模型/回退模型/降级输出
  • 回归评测:降本不等于降正确率(建议固定样本集)
Logo

有“AI”的1024 = 2048,欢迎大家加入2048 AI社区

更多推荐