大模型 API 成本优化 SOP：前缀缓存、模型路由与批量调用

import time
from collections import defaultdict


class CostTracker:
    """轻量调用成本追踪器：记录每次调用的 token 用量与费用"""

    def __init__(self):
        self.logs = []
        self.cost_by_model = defaultdict(float)

    def log(self, model, input_tokens, output_tokens,
            cached_tokens=0, cost=0.0):
        self.logs.append({
            "ts": time.time(),
            "model": model,
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "cached_tokens": cached_tokens,
            "cost": cost,
        })
        self.cost_by_model[model] += cost

    def summary(self):
        total_cost = sum(l["cost"] for l in self.logs)
        total_input = sum(l["input_tokens"] for l in self.logs)
        total_cached = sum(l["cached_tokens"] for l in self.logs)
        hit_rate = total_cached / total_input if total_input else 0
        return {
            "total_cost": round(total_cost, 4),
            "total_input_tokens": total_input,
            "cache_hit_rate": f"{hit_rate:.1%}",
            "cost_by_model": dict(self.cost_by_model),
        }

import anthropic

client = anthropic.Anthropic()  # 环境变量 ANTHROPIC_API_KEY

# 稳定的长文档/规则/示例放进 system，标记 cache_control
LONG_DOC = "（这里放几千 token 的文档内容、规则、few-shot 示例）"

response = client.messages.create(
    model="claude-sonnet-4-5-20250929",  # 以官网为准
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": f"你是文档助手。参考文档：\n{LONG_DOC}",
            "cache_control": {"type": "ephemeral"},
        }
    ],
    messages=[
        {"role": "user", "content": "总结文档要点。"}
    ],
)

# 响应 usage 里会返回缓存命中情况
print(f"缓存读取 tokens: {response.usage.cache_read_input_tokens}")
print(f"缓存写入 tokens: {response.usage.cache_creation_input_tokens}")
print(f"常规输入 tokens: {response.usage.input_tokens}")

def route_model(query: str) -> str:
    """按任务复杂度路由到不同模型（模型名以官网为准）"""
    q = query.lower()

    # 复杂推理：代码、数学、长文分析、多步推理 -> 强模型
    hard_keywords = [
        "代码", "编程", "debug", "数学", "推理", "分析",
        "code", "math", "reason", "analyze", "essay",
    ]
    if any(kw in q for kw in hard_keywords):
        return "qwen-max"  # 强模型

    # 简单任务：翻译、分类、摘要、格式转换 -> 轻量模型
    return "qwen-turbo"  # 便宜模型


def call_with_fallback(query: str) -> str:
    """带 fallback 的路由调用"""
    from openai import OpenAI
    client = OpenAI()  # OpenAI 兼容接口（Qwen/DeepSeek 均支持）

    primary = route_model(query)
    fallback = "qwen-plus"  # 主模型不可用时降级

    for model in [primary, fallback]:
        try:
            resp = client.chat.completions.create(
                model=model,
                messages=[{"role": "user", "content": query}],
                max_tokens=512,
            )
            return resp.choices[0].message.content
        except Exception as e:
            print(f"模型 {model} 调用失败: {e}，尝试 fallback")
    raise RuntimeError("所有模型均不可用")

import json
from openai import OpenAI

client = OpenAI()  # 环境变量 OPENAI_API_KEY

# 1. 把请求写成 JSONL 文件
texts = ["你好", "谢谢", "再见", "请问", "没问题"]
requests = [
    {
        "custom_id": f"req-{i}",
        "method": "POST",
        "url": "/v1/chat/completions",
        "body": {
            "model": "gpt-4o-mini",
            "messages": [{"role": "user",
                          "content": f"翻译成英文：{text}"}],
            "max_tokens": 200,
        },
    }
    for i, text in enumerate(texts)
]

with open("batch_input.jsonl", "w", encoding="utf-8") as f:
    for r in requests:
        f.write(json.dumps(r, ensure_ascii=False) + "\n")

# 2. 上传文件 + 创建 batch 任务
batch_file = client.files.create(
    file=open("batch_input.jsonl", "rb"),
    purpose="batch",
)

batch = client.batches.create(
    input_file_id=batch_file.id,
    endpoint="/v1/chat/completions",
    completion_window="24h",
)

print(f"Batch ID: {batch.id}, 状态: {batch.status}")

# 3. 轮询状态，完成后下载结果
# batch = client.batches.retrieve(batch.id)
# if batch.status == "completed":
#     result = client.files.content(batch.output_file_id)

def trim_context(messages: list, max_messages: int = 20) -> list:
    """保留 system + 最近 N 条，裁掉中间历史"""
    if len(messages) <= max_messages:
        return messages

    system_msgs = [m for m in messages if m["role"] == "system"]
    dialog = [m for m in messages if m["role"] != "system"]
    keep = max_messages - len(system_msgs)
    recent = dialog[-keep:] if keep > 0 else []

    return system_msgs + recent


def compress_with_summary(messages: list, llm_call) -> list:
    """超长对话：早期消息压成摘要，保留近期原文"""
    if len(messages) <= 20:
        return messages

    system_msgs = [m for m in messages if m["role"] == "system"]
    dialog = [m for m in messages if m["role"] != "system"]

    to_summarize = dialog[:-10]
    recent = dialog[-10:]

    # 早期对话让便宜模型压成摘要
    summary_input = "\n".join(
        f"{m['role']}: {m['content']}" for m in to_summarize)
    summary = llm_call(f"把以下对话压成 200 字摘要：\n{summary_input}")

    summary_msg = {
        "role": "system",
        "content": f"早期对话摘要：{summary}",
    }
    return system_msgs + [summary_msg] + recent

import hashlib
import time


class ResponseCache:
    """简单的响应缓存（只适合幂等请求）"""

    def __init__(self, ttl_seconds=300):
        self.cache = {}
        self.ttl = ttl_seconds

    def _key(self, model, messages):
        raw = f"{model}:{str(messages)}"
        return hashlib.md5(raw.encode()).hexdigest()

    def get(self, model, messages):
        key = self._key(model, messages)
        if key in self.cache:
            ts, result = self.cache[key]
            if time.time() - ts < self.ttl:
                return result
        return None

    def set(self, model, messages, result):
        key = self._key(model, messages)
        self.cache[key] = (time.time(), result)