AI 成本优化策略：从 Token 到基础设施的全链路省钱

# 冗长 Prompt（382 tokens）
prompt = """
你是一个专业的客服助手，你的主要职责是回答用户关于产品的问题。
你需要始终保持礼貌和专业，回答要简洁明了。
如果用户的问题超出了你的知识范围，请诚实告知用户你会将问题转接给人工客服。
在回答之前，请先理解用户的核心诉求，然后给出针对性的回答。
请使用以下格式回答：
1. 直接回答问题
2. 如果需要补充信息，请提供
3. 如果无法回答，请说明原因
"""

# 精简 Prompt（89 tokens）
prompt = "客服助手。简洁专业。不知则转人工。格式：直接回答→补充→说明原因。"

async def compressed_rag(query, documents):
    # 第一步：小模型快速摘要每个文档
    summaries = await asyncio.gather(*[
        small_model.summarize(doc, max_tokens=200)
        for doc in documents
    ])
    # 第二步：只将摘要 + 原始查询送入大模型
    context = "\n".join(summaries)
    response = await large_model.complete(
        f"基于以下信息回答：\n{context}\n\n问题：{query}"
    )
    return response

from sklearn.metrics.pairwise import cosine_similarity

def select_fewshot(query, examples, top_k=2):
    query_emb = embed_model.embed(query)
    scores = cosine_similarity([query_emb], examples["embeddings"])[0]
    top_indices = scores.argsort()[-top_k:][::-1]
    return [examples["data"][i] for i in top_indices]

class CostAwareRouter:
    def __init__(self):
        self.models = [
            # (model_name, input_price, output_price, capability_score)
            ("gpt-4o-mini",   0.15,  0.60, 0.65),
            ("claude-sonnet", 3.00, 15.00, 0.88),
            ("gpt-4o",        2.50, 10.00, 0.95),
        ]

    def route(self, task_complexity):
        candidates = []
        for name, in_p, out_p, cap in self.models:
            if cap >= task_complexity:
                candidates.append((name, in_p + out_p, cap))
        if not candidates:
            return self.models[-1][0]
        candidates.sort(key=lambda x: x[1])
        return candidates[0][0]

请求 → [L1 精确缓存] → [L2 语义缓存] → [L3 预计算] → LLM API
         (Redis)         (Vector DB)      (定时任务)

class TieredCache:
    def __init__(self, redis, vector_db, embed_model):
        self.redis = redis
        self.vector_db = vector_db
        self.embed_model = embed_model

    async def get(self, messages):
        # L1: 精确匹配
        key = self._hash(messages)
        if cached := await self.redis.get(key):
            return cached, "L1"
        # L2: 语义匹配
        query_emb = await self.embed_model.embed(self._flatten(messages))
        if results := await self.vector_db.search(query_emb, threshold=0.95):
            return results[0]["response"], "L2"
        return None, None

    async def set(self, messages, response, ttl=3600):
        key = self._hash(messages)
        await self.redis.setex(key, ttl, response)
        emb = await self.embed_model.embed(self._flatten(messages))
        await self.vector_db.upsert(emb, response, ttl=ttl)

import json, io

async def create_batch(tasks):
    requests = []
    for i, task in enumerate(tasks):
        requests.append({
            "custom_id": f"task-{i}",
            "method": "POST",
            "url": "/v1/chat/completions",
            "body": {
                "model": "gpt-4o",
                "messages": task["messages"],
                "max_tokens": 1000,
            }
        })
    file_content = "\n".join(json.dumps(r) for r in requests)
    batch_file = await openai.files.create(
        file=io.BytesIO(file_content.encode()),
        purpose="batch"
    )
    batch = await openai.batches.create(
        input_file_id=batch_file.id,
        endpoint="/v1/chat/completions",
        completion_window="24h",
    )
    return batch

# Batch API 比 实时 API 便宜 50%
# 实时: $2.50/1K input + $10.00/1K output
# Batch: $1.25/1K input + $5.00/1K output

# 启动 vLLM 服务
python -m vllm.entrypoints.openai.api_server \
  --model meta-llama/Llama-3-70B-Instruct \
  --tensor-parallel-size 4 \
  --max-model-len 8192

from openai import AsyncOpenAI

# 客户端兼容 OpenAI API 格式
vllm_client = AsyncOpenAI(
    base_url="http://vllm.internal:8000/v1",
    api_key="internal-key"
)
response = await vllm_client.chat.completions.create(
    model="meta-llama/Llama-3-70B-Instruct",
    messages=[{"role": "user", "content": query}],
)

@dataclass
class CostBreakdown:
    api_cost: float
    infra_cost: float
    embedding_cost: float
    storage_cost: float
    labor_cost: float

    @property
    def total(self):
        return sum([self.api_cost, self.infra_cost,
                    self.embedding_cost, self.storage_cost,
                    self.labor_cost])

def calc_roi(benefit: float, cost: CostBreakdown) -> dict:
    return {
        "monthly_cost": cost.total,
        "monthly_benefit": benefit,
        "net": benefit - cost.total,
        "roi": (benefit - cost.total) / cost.total * 100,
    }