LLM 成本优化策略 · 运维手册

适用场景：生产环境 LLM 应用成本控制，平衡质量与成本，建立可持续的 AI 服务体系。

---

1. 成本构成分析

1.1 Token 计费模型

模型	输入价格 ($/1M)	输出价格 ($/1M)	适用场景
GPT-4o	2.50	10.00	复杂推理
GPT-4o-mini	0.15	0.60	简单任务
Claude 3.5 Sonnet	3.00	15.00	长文档处理
Claude 3 Haiku	0.25	1.25	快速响应
Local Qwen 7B	0	0	私有化部署
Local Llama 70B	0	0	私有化部署

1.2 成本优化框架

总成本 = Σ(Token_i × Price_i) + Infrastructure Cost

优化方向：
├─ 减少 Token 消耗
│   ├─ 输入压缩（Prompt 精简）
│   ├─ 输出控制（限制长度）
│   └─ 缓存复用（相同请求）
├─ 降低单价
│   ├─ 模型分级（合适而非最强）
│   ├─ 批量优惠（预留实例）
│   └─ 本地部署（边际成本趋零）
└─ 提高利用率
    ├─ 并发优化（GPU 利用率）
    ├─ 批处理（合并请求）
    └─ 负载均衡（避免热点）

---

2. Token 消耗优化

2.1 Prompt 压缩

def compress_prompt(original: str, max_tokens: int = 2000) -> str:
    """压缩 Prompt 减少 Token"""
    # 策略 1: 移除空白和注释
    cleaned = re.sub(r'\s+', ' ', original).strip()

    # 策略 2: 保留关键部分
    lines = cleaned.split('\n')
    important = [l for l in lines if any(kw in l.lower() for kw in ['question', 'context', 'important'])]

    # 策略 3: 截断超长段落
    result = '\n'.join(important)
    if len(result) > max_tokens * 4:  # 粗略估算：1 token ≈ 4 chars
        result = result[:max_tokens * 4] + "...[truncated]"

    return result

2.2 输出长度控制

from openai import OpenAI

client = OpenAI()

response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "简短回答：Kubernetes 是什么？"}],
    max_tokens=50,  # 限制输出长度
    temperature=0.3  # 降低随机性
)

2.3 批量请求合并

async def batch_requests(prompts: list[str]) -> list[str]:
    """合并批量请求节省开销"""
    # 使用 batch API（价格降低 50%）
    batch_input = [
        {"custom_id": str(i), "method": "POST", "url": "/v1/chat/completions",
         "body": {"model": "gpt-4o-mini", "messages": [{"role": "user", "content": p}]}}
        for i, p in enumerate(prompts)
    ]

    response = await client.batches.create(input_data=batch_input)
    return await client.batches.retrieve(response.id)

---

3. 缓存策略

3.1 缓存键设计

import hashlib
import json

def cache_key(prompt: str, model: str, params: dict) -> str:
    """生成缓存键"""
    data = {
        "prompt": prompt,
        "model": model,
        "params": {k: v for k, v in params.items() if k != "messages"}
    }
    return hashlib.sha256(json.dumps(data, sort_keys=True).encode()).hexdigest()[:16]

3.2 Redis 缓存实现

import redis
import json
from datetime import timedelta

r = redis.Redis()

def cached_llm_call(prompt: str, model: str, ttl: int = 600) -> str:
    """带缓存的 LLM 调用"""
    key = f"llm:{cache_key(prompt, model, {})}"

    # 尝试读取缓存
    cached = r.get(key)
    if cached:
        return json.loads(cached)["result"]

    # 调用 LLM
    result = call_llm(prompt, model)

    # 写入缓存
    r.setex(key, ttl, json.dumps({"result": result, "created_at": time.time()}))

    return result

3.3 缓存失效策略

数据类型	TTL 建议	说明
静态知识（文档、FAQ）	7 天	内容不常变
动态数据（指标、日志）	5 分钟	时效性强
生成内容（报告、代码）	1 小时	中等鲜度
实时查询（股价、天气）	1 分钟	高时效性

---

4. 模型分级策略

4.1 分级决策树

用户问题
    │
    ├─ 简单事实查询？───────────→ GPT-4o-mini（便宜）
    │
    ├─ 代码生成/调试？───────────→ Claude 3.5 Sonnet（质量）
    │
    ├─ 长文档分析？──────────────→ Claude 3 Opus（长上下文）
    │
    └─ 复杂推理/创意？───────────→ GPT-4o（最强）

4.2 混合部署方案

MODEL_FARM = {
    "simple": {"model": "qwen2.5:7b", "endpoint": "http://local-ollama:11434"},
    "medium": {"model": "gpt-4o-mini", "endpoint": "https://api.openai.com"},
    "complex": {"model": "claude-3-5-sonnet", "endpoint": "https://api.anthropic.com"}
}

def smart_router(question: str) -> str:
    """智能路由到合适模型"""
    # 用轻量模型分类问题复杂度
    classifier = call_llm(f"判断问题复杂度（简单/中等/复杂）：{question}", "qwen2.5:7b")

    if "简单" in classifier:
        return MODEL_FARM["simple"]
    elif "中等" in classifier:
        return MODEL_FARM["medium"]
    else:
        return MODEL_FARM["complex"]

---

5. 流式输出优化

5.1 流式调用

async def streaming_chat(prompt: str):
    """流式输出，减少等待时间"""
    async with AsyncOpenAI() as client:
        stream = await client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
            stream=True
        )

        async for chunk in stream:
            if chunk.choices[0].delta.content:
                yield chunk.choices[0].delta.content

5.2 Early Stopping

def should_stop_early(response: str) -> bool:
    """判断是否可以提前终止"""
    # 检测答案是否完整
    endings = ['\n\n', '。', '。', '因此', '综上所述']
    return any(response.rstrip().endswith(e) for e in endings)

async def smart_stream(prompt: str):
    """智能流式输出"""
    collected = []
    async for chunk in streaming_chat(prompt):
        collected.append(chunk)
        if should_stop_early(''.join(collected)):
            break  # 提前终止，节省 Token
    return ''.join(collected)

---

6. 成本监控

6.1 Token 计数器

from prometheus_client import Counter, Histogram

TOKEN_COST = Counter(
    'llm_token_cost_total',
    'Token cost in dollars',
    ['model']
)

TOKEN_USAGE = Histogram(
    'llm_token_usage',
    'Token usage per request',
    ['model', 'direction']  # prompt, completion
)

def track_cost(model: str, prompt_tokens: int, completion_tokens: int):
    """追踪 Token 消耗和成本"""
    pricing = get_pricing(model)
    cost = (prompt_tokens * pricing['input'] + 
            completion_tokens * pricing['output']) / 1_000_000

    TOKEN_COST.labels(model=model).inc(cost)
    TOKEN_USAGE.labels(model=model, direction='prompt').observe(prompt_tokens)
    TOKEN_USAGE.labels(model=model, direction='completion').observe(completion_tokens)

6.2 预算告警

DAILY_BUDGET = 100.0  # 每日预算 $100
spent_today = 0

@app.on_event("shutdown")
def check_budget():
    global spent_today
    if spent_today > DAILY_BUDGET:
        send_alert(f"今日 LLM 成本 ${spent_today:.2f} 超出预算 ${DAILY_BUDGET}")

    if spent_today > DAILY_BUDGET * 0.8:
        send_warning(f"今日 LLM 成本已达 ${spent_today:.2f}，预计超支")

---

7. 成本优化案例

案例 1：缓存提升 40% 命中率

场景：FAQ 机器人，重复问题多
策略：相同问题缓存 24 小时
效果：命中率 42%，月省 $300

案例 2：模型分级降低 60% 成本

场景：客服系统，问题复杂度差异大
策略：简单问题用小模型，复杂问题用大模型
效果：平均成本降低 60%，用户满意度不变

案例 3：流式输出减少 30% Token

场景：代码生成助手
策略：智能提前终止 + 流式输出
效果：平均 Token 节省 30%，响应更快

---

8. 延伸阅读

• OpenAI Pricing
• Anthropic Pricing
• Cost Optimization Guide

---

FAQ

如何计算 LLM API 调用成本？ 成本 = (输入 Token + 输出 Token) × 单价。OpenAI GPT-4o：输入 $2.50/1M，输出 $10/1M。本地部署：电费 + 硬件折旧。建议建立成本追踪系统。

缓存策略如何节省成本？ 相同 Prompt 直接返回缓存结果，避免重复调用。缓存键用 Prompt 哈希，TTL 根据数据鲜度设置（10 分钟到 7 天不等）。命中率可达 30-60%。

如何选择性价比最高的模型？ 分级使用：简单查询用小型模型（7B 以下），复杂推理用大型模型（70B+）。混合部署：本地小模型 + 云端大模型。目标是满足质量要求下的最低成本。

流式输出如何降低成本？ 流式输出减少用户等待时间，可提前终止无效生成。配合 early stopping 机制，平均节省 20-30% Token 消耗。

---