任务复杂度评估 → 匹配最低能力模型 → 验证质量达标 → 上线

简单任务（分类/提取/格式化）→ nano/flash 级
中等任务（内容生成/翻译）→ mini/sonnet 级
复杂任务（推理/分析/创作）→ 标准模型
关键任务（代码审核/决策）→ 旗舰模型

# ❌ 冗长的System Prompt（消耗 ~450 tokens）
system_bad = """
你是一个非常专业且经验丰富的客户服务代表，你需要用友好、耐心的方式
来回答用户提出的各种问题。请确保你的回答准确、完整，并且易于理解。
如果用户的问题你不确定，请诚实地告知用户你不太确定...
"""

# ✅ 精简版（消耗 ~120 tokens，节省73%）
system_good = "你是客服代表。友好、准确地回答问题。不确定时坦诚说明。"

# ❌ 让模型自由输出（输出可能500+ tokens）
prompt_bad = "分析这段文本的情感，详细解释你的推理过程"

# ✅ 指定JSON输出（输出约50 tokens）
prompt_good = """分析情感，返回JSON：
{"sentiment": "positive|negative|neutral", "confidence": 0.0-1.0}
文本：{text}"""

# ❌ 提供5个完整示例（~2000 tokens）
# ✅ 提供2个精简示例 + 1个边界case（~600 tokens）
# 节省70%的示例token，效果几乎无损

import tiktoken

def compress_prompt(prompt: str, max_tokens: int = 500) -> str:
    """当prompt超过阈值时自动截断低优先级部分"""
    enc = tiktoken.encoding_for_model("gpt-5")
    tokens = enc.encode(prompt)
    if len(tokens) <= max_tokens:
        return prompt
    return enc.decode(tokens[:max_tokens])

import anthropic

client = anthropic.Anthropic()

# 定义缓存内容（通常是长System Prompt或文档）
system_content = [
    {
        "type": "text",
        "text": "这里是你的长System Prompt或知识库内容...",
        "cache_control": {"type": "ephemeral"}  # 标记为可缓存
    }
]

# 首次请求：完整计费
response1 = client.messages.create(
    model="claude-sonnet-4-20250514",
    system=system_content,
    messages=[{"role": "user", "content": "问题1"}],
    max_tokens=1024
)

# 后续请求：缓存命中，输入token按90%折扣计费
response2 = client.messages.create(
    model="claude-sonnet-4-20250514",
    system=system_content,
    messages=[{"role": "user", "content": "问题2"}],
    max_tokens=1024
)

from openai import OpenAI
client = OpenAI()

# OpenAI自动缓存相同前缀的请求
# 当多个请求共享相同的system message时，自动享受50%折扣
response = client.chat.completions.create(
    model="gpt-5",
    messages=[
        {"role": "system", "content": "长系统提示词...（自动缓存）"},
        {"role": "user", "content": "用户问题"}
    ]
)

from openai import OpenAI
client = OpenAI()

# 准备批量请求文件 (JSONL格式)
batch_requests = [
    {
        "custom_id": "task-001",
        "method": "POST",
        "url": "/v1/chat/completions",
        "body": {
            "model": "gpt-5-mini",
            "messages": [{"role": "user", "content": "总结这段文本：..."}],
            "max_tokens": 500
        }
    },
    # ... 更多请求
]

# 写入JSONL文件
import json
with open("batch_input.jsonl", "w") as f:
    for req in batch_requests:
        f.write(json.dumps(req) + "\n")

# 上传并创建Batch任务
batch_file = client.files.create(file=open("batch_input.jsonl", "rb"), purpose="batch")
batch_job = client.batches.create(input_file_id=batch_file.id, endpoint="/v1/chat/completions", completion_window="24h")

print(f"Batch ID: {batch_job.id}, 状态: {batch_job.status}")
# 24小时内完成，享受50%折扣

import anthropic

client = anthropic.Anthropic()

batch = client.batches.create(
    requests=[
        {
            "custom_id": "task-001",
            "params": {
                "model": "claude-haiku-4-20250514",
                "max_tokens": 1024,
                "messages": [{"role": "user", "content": "翻译为英文：..."}]
            }
        }
        # ... 更多请求
    ]
)

import tiktoken

def count_tokens(text: str, model: str = "gpt-5") -> int:
    """计算文本的token数量"""
    enc = tiktoken.encoding_for_model(model)
    return len(enc.encode(text))

def estimate_cost(input_tokens: int, output_tokens: int, model: str) -> float:
    """估算API调用成本"""
    pricing = {
        "gpt-5":        {"input": 5.00, "output": 15.00},
        "gpt-5-mini":   {"input": 0.80, "output": 2.40},
        "gpt-5-nano":   {"input": 0.15, "output": 0.45},
        "claude-sonnet-4": {"input": 2.00, "output": 10.00},
        "claude-haiku-4":  {"input": 0.50, "output": 2.50},
        "deepseek-v3":  {"input": 0.14, "output": 0.28},
    }
    p = pricing.get(model, pricing["gpt-5-mini"])
    return (input_tokens * p["input"] + output_tokens * p["output"]) / 1_000_000

# 使用Prometheus + Grafana搭建监控
from prometheus_client import Counter, Histogram, start_http_server

TOKEN_USAGE = Counter('llm_tokens_total', 'Total tokens used', ['model', 'type'])
API_COST = Counter('llm_cost_dollars', 'Total API cost in dollars', ['model'])
API_LATENCY = Histogram('llm_latency_seconds', 'API call latency', ['model'])

def track_api_call(model: str, input_tok: int, output_tok: int, 
                   latency: float, cost: float):
    TOKEN_USAGE.labels(model=model, type='input').inc(input_tok)
    TOKEN_USAGE.labels(model=model, type='output').inc(output_tok)
    API_COST.labels(model=model).inc(cost)
    API_LATENCY.labels(model=model).observe(latency)

import re
from enum import Enum

class TaskComplexity(Enum):
    SIMPLE = "simple"       # 分类、提取、格式化
    MEDIUM = "medium"       # 翻译、摘要、问答
    COMPLEX = "complex"     # 推理、分析、创作
    CRITICAL = "critical"   # 代码审核、关键决策

# 模型路由映射
MODEL_ROUTING = {
    TaskComplexity.SIMPLE:  "gpt-5-nano",        # $0.15/M input
    TaskComplexity.MEDIUM:  "gpt-5-mini",         # $0.80/M input
    TaskComplexity.COMPLEX: "gpt-5",              # $5.00/M input
    TaskComplexity.CRITICAL:"gpt-5",              # $5.00/M input
}

# 简单的复杂度分类器（也可用LLM自身分类）
COMPLEXITY_KEYWORDS = {
    TaskComplexity.SIMPLE: ["分类", "提取", "格式化", "列表", "标签"],
    TaskComplexity.MEDIUM: ["翻译", "总结", "解释", "回答"],
    TaskComplexity.COMPLEX: ["分析", "推理", "比较", "评估", "设计"],
    TaskComplexity.CRITICAL: ["审核", "安全", "决策", "架构"],
}

def classify_task(query: str) -> TaskComplexity:
    """基于关键词的快速分类"""
    for complexity, keywords in COMPLEXITY_KEYWORDS.items():
        if any(kw in query for kw in keywords):
            return complexity
    return TaskComplexity.MEDIUM  # 默认中等

def route_request(query: str) -> str:
    """路由请求到最优模型"""
    complexity = classify_task(query)
    model = MODEL_ROUTING[complexity]
    return model

# 使用示例
query = "请将这段文本翻译成英文"
model = route_request(query)  # → gpt-5-mini（$0.80/M）
# 如果用gpt-5会花费$5.00/M，节省84%

async def smart_classify(query: str) -> TaskComplexity:
    """用gpt-5-nano做复杂度分类，成本几乎为零"""
    response = await client.chat.completions.create(
        model="gpt-5-nano",
        messages=[{
            "role": "user",
            "content": f"将以下任务分类为 simple/medium/complex/critical：\n{query}\n只回复分类名称。"
        }],
        max_tokens=10
    )
    label = response.choices[0].message.content.strip().lower()
    return TaskComplexity(label)

from openai import OpenAI

client = OpenAI()

def stream_response(prompt: str, model: str = "gpt-5-mini"):
    """流式输出，首字延迟降低80%"""
    stream = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        stream=True,  # 启用流式
        max_tokens=1024
    )
    
    full_response = ""
    for chunk in stream:
        if chunk.choices[0].delta.content:
            token = chunk.choices[0].delta.content
            full_response += token
            print(token, end="", flush=True)  # 实时输出
    
    return full_response

def break_even_analysis(
    few_shot_overhead_tokens: int,   # 每次请求的few-shot额外token
    requests_per_month: int,          # 月请求数量
    model_input_price: float,         # 输入价格 ($/M tokens)
    fine_tune_cost: float,            # 微调训练总成本
    fine_tune_inference_surcharge: float  # 微调模型推理加价
) -> dict:
    """计算微调的盈亏平衡点"""
    
    # Few-shot月度额外成本
    few_shot_monthly = (few_shot_overhead_tokens * requests_per_month 
                        * model_input_price) / 1_000_000
    
    # 微调月度额外成本（训练费摊销 + 推理加价）
    ft_monthly = (fine_tune_cost / 12 +  # 假设12个月摊销
                  fine_tune_inference_surcharge * requests_per_month / 1_000_000)
    
    months_to_break_even = fine_tune_cost / max(few_shot_monthly - ft_monthly, 0.01)
    
    return {
        "few_shot_monthly_cost": round(few_shot_monthly, 2),
        "fine_tune_monthly_cost": round(ft_monthly, 2),
        "monthly_savings": round(few_shot_monthly - ft_monthly, 2),
        "break_even_months": round(months_to_break_even, 1)
    }

# 示例：10万次/月，800 token few-shot开销
result = break_even_analysis(
    few_shot_overhead_tokens=800,
    requests_per_month=100_000,
    model_input_price=0.80,
    fine_tune_cost=200,
    fine_tune_inference_surcharge=0.40
)
# → few_shot_monthly: $64, fine_tune_monthly: $20.67, 盈亏平衡: 4.6个月

import hashlib
import json
import redis
from typing import Optional

class LLMResponseCache:
    def __init__(self, redis_url: str = "redis://localhost:6379"):
        self.redis = redis.from_url(redis_url)
        self.default_ttl = 3600 * 24  # 24小时
    
    def _make_key(self, model: str, messages: list, **kwargs) -> str:
        """生成缓存键"""
        content = json.dumps({
            "model": model,
            "messages": messages,
            **kwargs
        }, sort_keys=True)
        return f"llm:cache:{hashlib.sha256(content.encode()).hexdigest()}"
    
    def get(self, model: str, messages: list, **kwargs) -> Optional[str]:
        """查询缓存"""
        key = self._make_key(model, messages, **kwargs)
        result = self.redis.get(key)
        return result.decode() if result else None
    
    def set(self, model: str, messages: list, response: str, 
            ttl: int = None, **kwargs):
        """写入缓存"""
        key = self._make_key(model, messages, **kwargs)
        self.redis.setex(key, ttl or self.default_ttl, response)

# 使用示例
cache = LLMResponseCache()

def call_with_cache(messages: list, model: str = "gpt-5-mini", **kwargs):
    """带缓存的API调用"""
    # 1. 查缓存
    cached = cache.get(model, messages, **kwargs)
    if cached:
        return {"content": cached, "source": "cache", "cost": 0}
    
    # 2. 调API
    response = client.chat.completions.create(
        model=model, messages=messages, **kwargs
    )
    result = response.choices[0].message.content
    
    # 3. 写缓存
    cache.set(model, messages, result, **kwargs)
    return {"content": result, "source": "api", "cost": response.usage}

# 只需替换base_url，即可接入XiDao Gateway
from openai import OpenAI

client = OpenAI(
    api_key="your-xidao-api-key",
    base_url="https://api.xidao.online/v1"  # XiDao Gateway
)

# 调用任意模型，统一计量
response = client.chat.completions.create(
    model="gpt-5-mini",  # 也可用 claude-sonnet-4, gemini-2.5-pro 等
    messages=[{"role": "user", "content": "你好"}],
    extra_headers={
        "X-Team": "backend",       # 团队标签
        "X-Project": "chatbot",    # 项目标签
        "X-Budget-Limit": "100"    # 本次请求预算上限（美元）
    }
)

# 查看实时用量
# GET https://api.xidao.online/dashboard/costs?team=backend&period=month