LLM 成本治理与 Token 优化

单次调用成本 = (输入 token × 输入单价)
             + (输出 token × 输出单价)
             - (缓存命中 token × 命中折扣)

批量成本 = Σ 单次调用成本

# cost_formula.py — 按计费模型估算单次调用成本（美元）
def call_cost(input_tokens, output_tokens, cached_tokens,
              price_in, price_out, cache_discount=0.1):
    cached_charge = cached_tokens * price_in * cache_discount
    fresh_input = max(0, input_tokens - cached_tokens) * price_in
    return fresh_input + cached_charge + output_tokens * price_out

请求 1: [System 角色][知识库 v3][工具说明][历史 H][问题 1]  ← 全价
请求 2: [System 角色][知识库 v3][工具说明][历史 H][问题 2]
             └─────────── 命中缓存 ──────────┘  ← 只对尾部计费

# cache_design.py — 缓存友好的上下文组装
def cacheable_context(static: dict, dynamic: dict) -> list[dict]:
    """静态前缀（角色/知识库/工具）+ 动态尾部（查询/临时检索）。"""
    prefix = [
        {"role": "system", "content": static["role"]},
        {"role": "system", "content": f"知识库：{static['kb']}"},
        {"role": "system", "content": f"工具：{static['tools']}"},
    ]
    tail = [{"role": "user", "content": dynamic["query"]}]
    return prefix + tail

def cache_hit_rate(stats) -> float:
    """缓存命中率 = 缓存 token / 总输入 token。健康值：> 60%。"""
    return stats["cached_input_tokens"] / stats["total_input_tokens"]

def kb_key(version: str) -> str:
    """知识库版本前缀：升级时只 bump 版本号，精确失效。"""
    return f"KB:{version}|"

def bump_kb(static: dict, new_version: str) -> dict:
    """升级知识库 → 返回新前缀，旧缓存自动失效。"""
    return {**static, "kb_version": new_version}

# routing.py — 复杂度分级路由
import json

def classify_complexity(query: str) -> str:
    """小模型分类器判复杂度：simple / medium / complex。"""
    resp = small_model.classify(query)
    return resp["label"]

def route(query: str) -> str:
    label = classify_complexity(query)
    return {
        "simple":  "gpt-4o-mini",   # 价格低 10-30 倍
        "medium":  "gpt-4o-mini",
        "complex": "gpt-4o",        # 只在难题上用
    }[label]

def monitor_routing(golden_cases, router, judge_fn) -> dict:
    """评估路由正确率：简单题是否被错误路由到贵模型。"""
    misroute_expensive = 0
    wrong_answers = 0
    for case in golden_cases:
        model = router(case["query"])
        if case["difficulty"] == "simple" and model == "gpt-4o":
            misroute_expensive += 1
        if case["difficulty"] == "complex" and model == "gpt-4o-mini":
            if judge_fn(case)["total"] < 3:
                wrong_answers += 1
    return {
        "expensive_misroute_pct": misroute_expensive / len(golden_cases),
        "wrong_answer_pct": wrong_answers / len(golden_cases),
    }

async def cascade_call(query: str, judge_fn) -> str:
    """先小模型，Judge 认为质量不足再升大模型。"""
    answer = await small_model.answer(query)
    if judge_fn(query, answer)["total"] >= 4:
        return answer, "small"          # 小模型已够好
    answer = await large_model.answer(query)
    return answer, "large"              # 升级兜底

# history_compress.py — 历史裁剪
def trim_history(messages, max_turns=6, summarizer=None):
    """超窗的旧消息压缩为摘要，保留关键决定。"""
    if len(messages) <= max_turns * 2:
        return messages
    keep = messages[-max_turns * 2:]          # 保留最近 N 轮
    older = messages[:-max_turns * 2]         # 更早的压缩
    summary = summarizer(older) if summarizer else "【历史省略】"
    return [{"role": "system", "content": f"早期对话摘要：{summary}"}] + keep

def compress_evidence(docs, budget_tokens) -> list[str]:
    """按 token 预算压缩检索片段：去重、截取、必要时摘要。"""
    docs.sort(key=lambda d: d["score"], reverse=True)
    result, used = [], 0
    for d in docs:
        snippet = trim_to_budget(d["text"], budget_tokens - used)
        if not snippet:
            break
        result.append(snippet)
        used += count_tokens(snippet)
    return result

# batching.py — 离线批处理调度
async def batch_offline_tasks(tasks, batch_size=20):
    """把 N 个小任务合并成大 batch，摊薄固定开销。"""
    for i in range(0, len(tasks), batch_size):
        batch = tasks[i:i + batch_size]
        # 一次调用处理多个输入，共享前缀与模型加载成本
        results = await model.batch_complete(
            prompts=[t["prompt"] for t in batch])
        yield from zip(batch, results)

def batch_economics(n, per_call_overhead, batch_size):
    """对比逐条与批处理的成本。"""
    single = n * (per_call_overhead + 1)
    batched = (n / batch_size) * (per_call_overhead + batch_size)
    saving = 1 - batched / single
    return {"single": single, "batched": batched,
            "saving_pct": saving * 100}

def constrained_answer(query: str, schema: dict) -> dict:
    """用结构化输出约束：只返回 schema 需要的字段。"""
    resp = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": query}],
        response_format={"type": "json_schema", "schema": schema},
        max_tokens=300,          # 硬性上限，防止发散
    )
    return json.loads(resp.choices[0].message.content)

def semantic_cache_lookup(query: str, threshold=0.95) -> str | None:
    """embedding 相似度命中历史答案，完全不再调用模型。"""
    emb = embed(query)
    hit = vector_db.search(emb, top_k=1)
    if hit and hit[0].score >= threshold:
        return hit[0].answer
    return None

# cost_dashboard.py — 成本看板核心指标
def cost_dashboard(daily_stats: list[dict]) -> dict:
    totals = sum_daily(daily_stats)
    return {
        "daily_cost": totals["cost"],
        "cost_by_feature": group_by_feature(daily_stats),
        "cost_by_model": group_by_model(daily_stats),
        "avg_cost_per_request": totals["cost"] / totals["requests"],
        "cache_hit_rate": totals["cached"] / totals["input_tokens"],
        "token_to_value_ratio": totals["value_score"] / totals["cost"],
    }

def cost_budget_check(daily: float, monthly_budget: float,
                      day_of_month: int, days_in_month: int) -> str:
    """月度预算按天线性预测，提前预警超支。"""
    projected = daily * days_in_month / max(day_of_month, 1)
    if projected > monthly_budget * 1.2:
        return "CRITICAL"
    if projected > monthly_budget:
        return "WARN"
    return "OK"

def auto_remediate(budget_status: str):
    """按预算健康度逐级降级：先路由、再限流、最后熔断。"""
    actions = {
        "WARN":     lambda: switch_to_mini_model(),      # 全量走小模型
        "CRITICAL": lambda: enable_semantic_cache_only(), # 优先命中缓存
        "OVERDUE":  lambda: reject_non_core_features(),   # 只保核心功能
    }
    return actions.get(budget_status, lambda: None)()

def cost_quality_gate(optimized, baseline, judge_fn, golden,
                      max_drop=0.03) -> bool:
    """成本优化方案与基线对比：质量下降超阈值则拒绝上线。"""
    saving = 1 - estimate_cost(optimized, golden) / estimate_cost(baseline, golden)
    drop = judge_avg(baseline, golden) - judge_avg(optimized, golden)
    print(f"节省 {saving:.0%}，质量下降 {drop:.3f}")
    assert drop <= max_drop, "成本优化牺牲质量，拒绝上线"
    return True

阶段一（1-2 周）        阶段二（2-4 周）          阶段三（1-2 月）
──────────────────  ────────────────────────  ────────────────────────
· 全量埋点成本数据     · 接入前缀缓存             · 成本×质量联合门禁
· 四维归因（功能/用户） · 引入复杂度路由 + 级联     · 自动降级动作链
· 建立月度基线         · 历史裁剪与压缩           · 语义缓存 / 批处理
   └ 能看见           └ 能省                    └ 能自动、能止损