LLM 成本优化实战：从 Token 预算到模型路由

from anthropic import Anthropic

client = Anthropic()

class ContextManager:
    def __init__(self, max_tokens: int = 8000, summary_threshold: int = 6000):
        self.max_tokens = max_tokens
        self.summary_threshold = summary_threshold
        self.messages = []
        self.summary = ""
    
    def add_message(self, role: str, content: str):
        self.messages.append({"role": role, "content": content})
        
        # 估算当前 token 数（粗略：4 字符 ≈ 1 token）
        total_chars = sum(len(m["content"]) for m in self.messages)
        estimated_tokens = total_chars // 4
        
        if estimated_tokens > self.summary_threshold:
            self._compress_history()
    
    def _compress_history(self):
        """把旧对话压缩成摘要"""
        # 保留最近 4 轮对话，其余压缩
        recent_messages = self.messages[-8:]
        old_messages = self.messages[:-8]
        
        if not old_messages:
            return
        
        # 用小模型（便宜）生成摘要
        summary_prompt = f"""请将以下对话历史压缩成 200 字以内的摘要，保留关键信息：

{chr(10).join(f'{m["role"]}: {m["content"][:500]}' for m in old_messages)}"""
        
        response = client.messages.create(
            model="claude-haiku-3-5-20241022",  # 用便宜模型做摘要
            max_tokens=300,
            messages=[{"role": "user", "content": summary_prompt}]
        )
        
        self.summary = response.content[0].text
        self.messages = recent_messages
        print(f"历史压缩：{len(old_messages)} 条 → 摘要 {len(self.summary)} 字")
    
    def get_messages_with_context(self) -> list[dict]:
        if self.summary:
            # 把摘要作为第一条系统消息注入
            return [
                {"role": "user", "content": f"[对话背景摘要]\n{self.summary}"},
                {"role": "assistant", "content": "已了解背景信息。"},
                *self.messages
            ]
        return self.messages

# 精简前：2100 tokens（每次都要付这 2100 的 input 费用）
VERBOSE_SYSTEM_PROMPT = """
你是一个专业的客服助手，名叫小智。你由我们公司的工程师精心打造，
具备丰富的产品知识和出色的沟通能力。你的性格友善、耐心、专业。

你的职责包括但不限于：
1. 回答用户关于产品功能的问题
2. 帮助用户解决使用过程中遇到的技术问题
3. 收集用户反馈并记录
4. 在必要时引导用户联系人工客服
...（继续写了 10 条）

当用户问到你的身份时，你应该这样回答：...
当用户情绪激动时，你应该这样处理：...
"""

# 精简后：280 tokens
CONCISE_SYSTEM_PROMPT = """你是客服助手小智。
职责：回答产品问题、解决技术问题、必要时转人工客服。
规则：只基于知识库内容回答；无法确定时说"我来帮您查一下"；保持友善简洁。"""

from anthropic import Anthropic

client = Anthropic()

# 系统提示词（很长，适合缓存）
LONG_SYSTEM_PROMPT = """你是一个专业的代码审查助手。以下是我们公司的代码规范：

## Python 规范
- 使用 Black 格式化，行长度 88
- 类型注解必须完整
- 所有公开函数必须有 docstring
- 禁止使用 global 变量
- 异步函数使用 asyncio，不使用 threading
...（500+ 行规范内容）

## Go 规范  
- 使用 gofmt 格式化
- 错误必须显式处理，不能忽略
- context 作为第一个参数
...（又 300 行）
"""

def review_code(code: str, language: str) -> str:
    response = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=1024,
        system=[
            {
                "type": "text",
                "text": LONG_SYSTEM_PROMPT,
                "cache_control": {"type": "ephemeral"}  # 标记为可缓存
            }
        ],
        messages=[
            {
                "role": "user",
                "content": f"请 review 以下 {language} 代码：\n\n```{language}\n{code}\n```"
            }
        ]
    )
    
    # 检查缓存命中情况
    usage = response.usage
    print(f"Input tokens: {usage.input_tokens}")
    print(f"Cache creation: {usage.cache_creation_input_tokens}")
    print(f"Cache read: {usage.cache_read_input_tokens}")
    
    return response.content[0].text

# 适合缓存的内容（放在 messages 靠前的位置，且需要 cache_control 标记）：

# 1. 长系统提示词（规范、角色描述）
# 2. RAG 检索到的文档（多个问题基于同一批文档）
# 3. Few-shot 示例（同类任务的示例集）
# 4. 工具定义（Agent 场景下的工具列表通常很长）

def rag_query(question: str, documents: list[str]) -> str:
    docs_content = "\n\n".join(f"文档{i+1}：\n{doc}" for i, doc in enumerate(documents))
    
    response = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=512,
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "type": "text",
                        "text": f"<knowledge_base>\n{docs_content}\n</knowledge_base>",
                        "cache_control": {"type": "ephemeral"}  # 缓存知识库文档
                    },
                    {
                        "type": "text", 
                        "text": f"\n根据以上文档回答：{question}"
                        # 问题不缓存，每次都变
                    }
                ]
            }
        ]
    )
    return response.content[0].text

from litellm import Router
import litellm

# 配置模型路由（使用 LiteLLM Router）
router = Router(
    model_list=[
        {
            "model_name": "fast-model",
            "litellm_params": {
                "model": "claude-haiku-3-5-20241022",
                "api_key": "your-key"
            }
        },
        {
            "model_name": "smart-model", 
            "litellm_params": {
                "model": "claude-sonnet-4-5",
                "api_key": "your-key"
            }
        },
        {
            "model_name": "cheap-model",
            "litellm_params": {
                "model": "deepseek/deepseek-chat",
                "api_key": "your-deepseek-key"
            }
        }
    ]
)

def classify_task_complexity(user_message: str) -> str:
    """
    用最小模型分类任务复杂度，决定路由到哪个主模型。
    分类本身的成本极低（Haiku ~0.5 cents/1000次）
    """
    classification_prompt = f"""将以下用户请求分类为：
- simple：简单问答、闲聊、基本信息查询
- medium：需要分析推理、代码生成、内容创作
- complex：需要深度推理、多步骤规划、专业领域复杂问题

只输出分类标签，不要解释。

用户请求：{user_message[:500]}"""
    
    response = router.completion(
        model="fast-model",
        messages=[{"role": "user", "content": classification_prompt}],
        max_tokens=10,
    )
    
    label = response.choices[0].message.content.strip().lower()
    return label if label in ["simple", "medium", "complex"] else "medium"


COMPLEXITY_TO_MODEL = {
    "simple": "fast-model",    # Haiku：$0.8/$4 per M
    "medium": "cheap-model",   # DeepSeek：$0.27/$1.1 per M
    "complex": "smart-model",  # Sonnet：$3/$15 per M
}

def smart_chat(user_message: str, conversation_history: list) -> str:
    complexity = classify_task_complexity(user_message)
    model = COMPLEXITY_TO_MODEL[complexity]
    
    response = router.completion(
        model=model,
        messages=conversation_history + [{"role": "user", "content": user_message}],
        max_tokens=1024,
    )
    
    # 记录路由决策，用于后续分析
    log_routing_decision(user_message[:100], complexity, model, response.usage)
    
    return response.choices[0].message.content

import anthropic
import json

client = anthropic.Anthropic()

def batch_analyze_feedback(feedback_list: list[str]) -> list[dict]:
    """批量分析用户反馈，使用 Batch API 节省 50% 成本"""
    
    # 构建批次请求
    requests = []
    for i, feedback in enumerate(feedback_list):
        requests.append({
            "custom_id": f"feedback-{i}",
            "params": {
                "model": "claude-haiku-3-5-20241022",
                "max_tokens": 100,
                "messages": [
                    {
                        "role": "user",
                        "content": f"""分析以下用户反馈，输出 JSON：
{{"sentiment": "positive/neutral/negative", "category": "product/service/pricing/other", "priority": "high/medium/low"}}

反馈：{feedback}"""
                    }
                ]
            }
        })
    
    # 提交批次
    batch = client.messages.batches.create(requests=requests)
    print(f"批次提交成功，ID: {batch.id}，共 {len(requests)} 条")
    
    # 轮询等待完成（实际生产中建议用 webhook 或定时任务）
    import time
    while True:
        batch_status = client.messages.batches.retrieve(batch.id)
        if batch_status.processing_status == "ended":
            break
        print(f"处理中... {batch_status.request_counts}")
        time.sleep(60)
    
    # 获取结果
    results = []
    for result in client.messages.batches.results(batch.id):
        if result.result.type == "succeeded":
            try:
                analysis = json.loads(result.result.message.content[0].text)
                results.append({
                    "id": result.custom_id,
                    "analysis": analysis
                })
            except json.JSONDecodeError:
                results.append({"id": result.custom_id, "error": "parse_failed"})
    
    return results


# OpenAI Batch API 类似
from openai import OpenAI
import json

openai_client = OpenAI()

def openai_batch_classify(texts: list[str]) -> str:
    """返回 batch job ID，后续轮询结果"""
    
    # 构建 JSONL 格式的批次文件
    batch_lines = []
    for i, text in enumerate(texts):
        batch_lines.append(json.dumps({
            "custom_id": f"item-{i}",
            "method": "POST",
            "url": "/v1/chat/completions",
            "body": {
                "model": "gpt-4.1-mini",
                "messages": [{"role": "user", "content": f"分类：{text[:200]}"}],
                "max_tokens": 50
            }
        }))
    
    # 上传文件
    import io
    file_content = "\n".join(batch_lines).encode()
    batch_file = openai_client.files.create(
        file=io.BytesIO(file_content),
        purpose="batch"
    )
    
    # 创建批次任务
    batch = openai_client.batches.create(
        input_file_id=batch_file.id,
        endpoint="/v1/chat/completions",
        completion_window="24h"
    )
    
    return batch.id

def should_self_host(
    monthly_tokens: int,      # 月均 token 消耗量
    api_price_per_m: float,   # API 价格（$/M tokens）
    gpu_monthly_cost: float = 3500,  # A100×4 的月租（AWS p4d.24xlarge 约 $3500/月）
    self_host_capacity_m: int = 500  # 自托管月处理能力（M tokens）
) -> dict:
    """简单的自托建 vs API 成本分析"""
    
    api_monthly_cost = (monthly_tokens / 1_000_000) * api_price_per_m
    self_host_unit_cost = gpu_monthly_cost / self_host_capacity_m
    self_host_monthly_cost = (monthly_tokens / 1_000_000) * self_host_unit_cost
    
    breakeven_tokens = gpu_monthly_cost / (api_price_per_m - self_host_unit_cost) * 1_000_000
    
    return {
        "api_monthly_cost": f"${api_monthly_cost:.0f}",
        "self_host_monthly_cost": f"${self_host_monthly_cost:.0f}",
        "recommended": "self_host" if monthly_tokens > breakeven_tokens else "api",
        "breakeven_m_tokens": f"{breakeven_tokens/1_000_000:.1f}M tokens/月"
    }

# 示例：使用 Claude Sonnet 处理 Output tokens（$15/M）
# vs 自托管 Qwen3-72B（A100×4，约 $0.5/M 等效成本）
print(should_self_host(
    monthly_tokens=100_000_000,  # 1 亿 tokens/月
    api_price_per_m=15,          # Sonnet output 价格
    gpu_monthly_cost=3500,
    self_host_capacity_m=200
))
# 输出：{'api_monthly_cost': '$1500', 'self_host_monthly_cost': '$1750', 'recommended': 'api', ...}
# 1亿 tokens/月这个量，自托管还没有 API 划算！

print(should_self_host(
    monthly_tokens=1_000_000_000,  # 10 亿 tokens/月
    api_price_per_m=15,
    gpu_monthly_cost=3500,
    self_host_capacity_m=200
))
# 这个量才开始值得自托管

from prometheus_client import Counter, Histogram, start_http_server
import time

# Prometheus metrics
llm_token_usage = Counter(
    'llm_token_total',
    'Total LLM token usage',
    ['model', 'token_type', 'feature', 'user_tier']
)
llm_cost_dollars = Counter(
    'llm_cost_dollars_total',
    'Total LLM cost in dollars',
    ['model', 'feature']
)
llm_request_duration = Histogram(
    'llm_request_duration_seconds',
    'LLM request latency',
    ['model', 'feature']
)

MODEL_PRICES = {
    "claude-haiku-3-5": {"input": 0.8/1e6, "output": 4.0/1e6},
    "claude-sonnet-4-5": {"input": 3.0/1e6, "output": 15.0/1e6},
    "deepseek-chat": {"input": 0.27/1e6, "output": 1.1/1e6},
}

def tracked_llm_call(
    model: str, 
    messages: list, 
    feature: str,
    user_tier: str = "standard",
    **kwargs
) -> object:
    start_time = time.time()
    
    response = router.completion(model=model, messages=messages, **kwargs)
    
    duration = time.time() - start_time
    usage = response.usage
    prices = MODEL_PRICES.get(model, {"input": 0, "output": 0})
    
    # 记录 metrics
    llm_token_usage.labels(model=model, token_type="input", feature=feature, user_tier=user_tier).inc(usage.prompt_tokens)
    llm_token_usage.labels(model=model, token_type="output", feature=feature, user_tier=user_tier).inc(usage.completion_tokens)
    
    cost = usage.prompt_tokens * prices["input"] + usage.completion_tokens * prices["output"]
    llm_cost_dollars.labels(model=model, feature=feature).inc(cost)
    llm_request_duration.labels(model=model, feature=feature).observe(duration)
    
    return response

# Prometheus alerting rule
- alert: LLMDailyCostHigh
  expr: increase(llm_cost_dollars_total[24h]) > 200
  annotations:
    summary: "LLM 日成本超过 $200，当前：{{ $value | printf \"%.2f\" }}"
    
- alert: LLMFeatureCostSpike  
  expr: |
    increase(llm_cost_dollars_total[1h]) 
    / increase(llm_cost_dollars_total[1h] offset 1d) > 3
  annotations:
    summary: "LLM 成本异常飙升（1小时成本是昨天同期的 3 倍）"