AI成本控制实战：8个Token降本策略

Token是AI服务的计费单位。输入（提示词、历史对话、检索结果、工具返回内容）和输出（模型回复）分别计费。多数厂商的输出单价高于输入，具体数字以各家官方价目表为准。

真正拉高成本的，通常不是单价，而是重复回传。每次调用都把历史对话、旧的检索结果和过期的中间步骤再发给模型，Token就在无效内容上反复计费。

先算清楚：Token花在哪里

优化之前要先量化。假设每轮对话新增 t 个Token，系统提示词长度为 S 个Token，共进行 N 轮：

• 第 n 轮的输入约为 S + n × t，随轮次线性增长。
• 累计输入为 N × S + t × N(N+1)/2，随轮次近似二次增长。

只有当Agent在一次任务中递归或分支调用，且每个分支都携带完整上下文时，成本才会接近指数增长。判断属于哪种情况，看日志里单次调用的输入Token是否随轮次持续上升即可。

单次任务成本的估算公式如下，价格单位为每百万Token的费用：

单次任务成本 = (未命中缓存的输入Token × 输入单价
              + 命中缓存的输入Token × 缓存单价
              + 输出Token × 输出单价) / 1,000,000
              + 工具调用与重试的额外成本
日成本 = 单次任务成本 × 日均任务数 × (1 + 重试率)

计算时还要注意几点：

• 工具返回的内容会计入下一轮的输入Token。
• 图片的Token折算规则因厂商而异，需按官方文档计算。
• 失败请求和重试同样消耗Token，要计入成本。
• 价格可能以人民币或美元标注，统一换算后再比较。

下面是一个简单的成本计算函数：

from dataclasses import dataclass
@dataclass
class Usage:
    input_tokens: int    # 未命中缓存的输入Token
    cached_tokens: int   # 命中缓存的输入Token
    output_tokens: int   # 输出Token
def task_cost(u: Usage, price_in: float, price_cache: float, price_out: float) -> float:
    """价格为每百万Token的费用，与厂商价目表保持一致。"""
    return (u.input_tokens * price_in
            + u.cached_tokens * price_cache
            + u.output_tokens * price_out) / 1_000_000
def daily_cost(per_task: float, tasks_per_day: int, retry_rate: float = 0.0) -> float:
    return per_task * tasks_per_day * (1 + retry_rate)

以一个客服Agent为例，处理一张工单可能需要调用模型十几次，分别用于理解意图、查询知识库、判断优先级、生成回复和检查格式。调用次数乘以工单量，日调用量很快达到百万级。因此降本的顺序是：先分层，再压缩，最后才考虑换更便宜的模型。

策略一：分层调用，大模型决策，小模型执行

旗舰模型负责任务规划、协调与最终判断。检索、审阅、格式转换等子任务交给轻量模型处理，并且可以并行执行。

落地时需要做三件事：

1. 为每类子任务定义能力下限，用一批带标注的样本离线评测，低于下限的模型不参与该类任务。
2. 用路由层按任务类型和复杂度分发请求，不让所有请求都进入旗舰模型。
3. 记录每层的准确率、人工复核率和单次成本，定期调整分层边界。

路由的最简版本如下：

def route(task_type: str, complexity: float) -> str:
    if task_type in {"complaint", "legal"} or complexity > 0.7:
        return "flagship"
    if task_type in {"faq", "format", "extract"}:
        return "small"
    return "medium"

路由规则本身也要评测。如果小模型在投诉类工单上漏判，多出来的人工成本可能超过节省的Token费用。

策略二：上下文压缩与摘要滚动

多轮对话是Token膨胀的主要来源。常见做法是保留最近的若干轮原文，把更早的对话压缩为结构化摘要。摘要中应包含用户意图、已确认的结论、关键实体（订单号、金额、日期等）和未完成事项，并剔除已经完成的中间步骤。

摘要长度可以先设为原文的10%到20%作为起点，再用评测集检查关键信息是否丢失。压缩比例没有通用答案，取决于对话的信息密度。

def compress_history(history, summarize, keep_last=6, max_summary_chars=800):
    """保留最近 keep_last 条消息原文，更早的消息压缩为摘要。"""
    if len(history) <= keep_last:
        return history
    old, recent = history[:-keep_last], history[-keep_last:]
    summary = summarize(old, max_chars=max_summary_chars)  # 可交给小模型生成
    return [{"role": "system", "content": f"历史摘要：\n{summary}"}] + recent

摘要模板的推荐字段：

意图：用户当前要解决的问题
已确认：已经得到确认的结论或数据
实体：订单号、金额、日期、产品型号等
待办：尚未完成的步骤
约束：用户提出的限制条件，例如"不要退款"

策略三：Prompt缓存与静态前缀复用

系统提示词、工具定义、知识库说明和Few-shot示例在每次调用中都相同，应放在Prompt的最前部，并启用厂商提供的前缀缓存。命中缓存的部分通常按折扣价计费，具体折扣与最短缓存长度、有效期等规则以厂商文档为准。

动态内容，例如用户问题和检索结果，放在后面。这样前缀能保持稳定，缓存命中率更高。

messages = [
    {"role": "system", "content": STATIC_SYSTEM_PROMPT},   # 固定不变
    {"role": "system", "content": FEW_SHOT_EXAMPLES},      # 固定不变
    {"role": "user", "content": f"用户问题：{question}\n检索结果：{chunks}"},  # 动态内容放最后
]

注意两个常见问题。第一，前缀中不能夹带时间戳、随机ID或请求编号，否则每次前缀都不同，缓存无法命中。第二，要在日志中记录每次调用的缓存命中Token数，否则无法判断缓存是否生效。

rag-">策略四：RAG检索结果精简

检索增强生成中最常见的浪费，是把整篇文档塞进上下文。精简可以从四个方面入手：

• 按相关性阈值过滤低分片段。
• 对片段做去重，去掉内容重复的部分。
• 截断过长片段，只保留与问题直接相关的段落。
• 控制Top-K，减少送入模型的片段数。

Top-K 的最优值取决于知识库结构和问题类型，不能直接套用经验值。建议在评测集上比较 K 为3、5、8、10 时的答案准确率与平均输入Token数，选择准确率达标、成本最低的组合。

def trim_chunks(chunks, min_score=0.35, top_k=5, max_chars=1200):
    seen, picked = set(), []
    for c in sorted(chunks, key=lambda x: x["score"], reverse=True):
        if c["score"] < min_score:
            break
        key = c["text"][:50]  # 简易去重，可换成哈希或向量相似度
        if key in seen:
            continue
        seen.add(key)
        picked.append(c["text"][:max_chars])
        if len(picked) == top_k:
            break
    return picked

策略五：约束输出长度

输出Token通常比输入贵。可以在Prompt中写明回答长度，例如”不超过200字”或”只输出JSON”，同时在接口参数中设置 max_tokens，并在后处理层截断超长输出。

结构化任务应使用JSON Schema或厂商提供的结构化输出功能，避免模型生成冗长的解释。

payload = {
    "model": "small-model",
    "max_tokens": 300,
    "messages": messages,
    "response_format": {"type": "json_object"},
}

需要注意，max_tokens 设得过低会导致回答被截断，进而触发重试，反而增加成本。上线前要统计截断率。

策略六：批处理与异步化

大量独立的小请求，例如批量分类、批量摘要，可以合并成一次批处理。系统提示词只需发送一次，固定开销被多条数据分摊。

对非实时场景，可以用异步队列收集请求，再通过厂商的批处理接口提交。部分厂商为批处理接口提供更低的单价，是否可用以官方说明为准。

批量大小会影响准确率。一次塞入过多条目，模型可能漏处理或错位。建议先用20条左右的小批量测试，再逐步增大。返回结果应包含每条数据的ID，方便对齐。

策略七：按有效成本选型

只看榜单分数选模型容易多花钱。更合理的做法是按任务类型计算综合成本：

每个合格结果的成本 = 单次任务成本 / 合格率

这个公式说明，单价低的模型如果合格率低，需要多次重试或人工复核，每个合格结果的成本反而可能更高。

选型流程可以这样做：

1. 用同一批样本，在候选模型上跑相同的任务。
2. 记录每个模型的合格率、平均输入Token、平均输出Token和单价。
3. 计算每个合格结果的成本，筛掉不满足能力下限的模型。
4. 在线上用小流量做A/B验证，确认效果和成本都符合预期。

模型价格和能力变化较快，可以通过各厂商官方价目表查询最新信息。如果同时接入多家模型，统一的API网关能简化A/B测试和切换。

策略八：调用监控与浪费审计

没有日志就无法判断成本花在哪里。每次请求至少应记录以下字段：

字段	说明
request_id	请求唯一标识，用于追踪
task_type	任务类型，如咨询、投诉、格式转换
model	实际调用的模型