给 Agent 做 Token 成本治理：模型路由、提示缓存与上下文瘦身七步法

优化的前提是能归因。建议在调用封装层统一记录四类字段：模型、提示词模板、业务归属、调用阶段。没有这层标签，你无法回答「钱花在哪个环节」这个问题。

import time, json, logging

def call_llm(model, messages, *, stage, template, tenant, **kw):
    t0 = time.time()
    resp = client.chat.completions.create(model=model, messages=messages, **kw)
    u = resp.usage
    logging.info(json.dumps({
        "event": "llm_call",
        "model": model,
        "stage": stage,          # 例如 plan / tool / summarize / judge
        "template": template,    # 提示词模板标识
        "tenant": tenant,        # 业务归属
        "in_tokens": u.prompt_tokens,
        "out_tokens": u.completion_tokens,
        "latency_ms": int((time.time() - t0) * 1000),
    }, ensure_ascii=False))
    return resp

这一层的收益通常在 10%–15% 起，且不牺牲质量，属于「免费层」。三个动作按优先级排列：

# ① 工具输出：只留下一步真正需要的字段，不要把整块 JSON 回灌
def slim_tool_result(raw: dict, keep=("id", "status", "owner")):
    return {k: raw.get(k) for k in keep if k in raw}

# ② 历史：把中段对话压成一条摘要，只保留系统提示 + 最近若干轮
def trim_context(messages, max_recent=6, summary_model="gpt-4o-mini"):
    if len(messages) <= max_recent:
        return messages
    system = [m for m in messages if m["role"] == "system"]
    recent = messages[-max_recent:]
    middle = messages[len(system):-max_recent]
    summary = call_llm(summary_model, [{
        "role": "user",
        "content": "把以下对话压成给下一步用的要点，保留结论与约束，删掉过程：\n" + str(middle),
    }], stage="summarize", template="ctx_trim", tenant="internal")
    return system + [{"role": "user",
                      "content": "前情摘要：" + summary.choices[0].message.content}] + recent

③ 给输出设硬上限。会「自言自语」的 Agent 可能产出比最终答案多 3–10 倍的输出 token，而输出通常比输入贵 4–5 倍。为每个工作流单独设置输出上限，比全局设一个值更合理。

模型价格带大致分三档，最便宜与最贵之间可能相差百倍。把 60%–80% 的子任务路由到小模型，是收益最集中的一步。

def route(task_type: str) -> str:
    # 规则优先：能用规则判断的，不要交给模型判断
    if task_type in ("extract", "classify", "reformat"):
        return "gpt-4o-mini"
    if task_type in ("summarize", "draft"):
        return "gpt-4o"
    return "o-series-deep"   # 规划类，保持强模型

# 关键判断：这次失败会不会触发「带着大上下文重跑」？
# 会  → 一开始就用强模型，贵得值
# 不会 → 先让小模型试，失败再升级

提示缓存复用重复前缀背后的键值张量，稳定部分常能享受 50%–90% 的输入折扣。但缓存收益是有条件的，账要算清楚：写入通常按 1.25 倍计价、读取按 0.1 倍计价，因此需要至少两次读取才能回本。

Agent 会重复调用完全相同的工具：读同一个数据库 schema、抓同一份文档、查同一份配置。把这些结果缓存下来，命中时零边际成本。

import hashlib, time

_cache = {}

def cached_call(fn, *args, ttl=300, **kw):
    key = hashlib.sha256((fn.__name__ + repr(args) + repr(kw)).encode()).hexdigest()
    now = time.time()
    hit = _cache.get(key)
    if hit and now - hit["ts"] < ttl:
        return hit["value"]
    value = fn(*args, **kw)
    _cache[key] = {"value": value, "ts": now}
    return value

再往上一层是语义缓存：把高度相似的用户提问直接返回历史结果，不转发给模型。对于客服、知识问答这类重复度高的场景，通常能吸收 15%–30% 的流量。代价是缓存失效逻辑更复杂，需要有明确的过期与版本策略。

批处理接口对可容忍延迟的请求通常提供约 50% 的折扣。适合批处理的场景很明确：夜间报表生成、批量文档分类、历史数据富化、批量内容生成、非紧急文档分析。

成本应该和延迟、错误率放在同一块看板上，而不是月底才在财务报表里看到。核心指标只有一个：cost per successful task（每个成功任务的花费），并按模型与任务类型拆分。

这个指标能同时暴露三件事：某个模型在悄悄重试、某个工具返回了臃肿输出、某个子任务本该路由到更便宜的档位。当单任务成本突然抬升，那通常是一个 bug，而不是一张账单。

# 建议固定跟踪的三个信号
# 1) 缓存命中率：稳定提示词场景低于 60% 说明结构有问题
# 2) 每类任务的重试次数：每次重试都是整段上下文的重新发送
# 3) 每成功任务成本：按模型 × 任务类型拆分，观察趋势而非绝对值