作者详细复盘了从 GPT-4o 随意调用到构建分层路由系统的过程,用数据库查询优化思维管理 LLM 调用,85% 请求迁移至 $0.01/M 模型。
我直接坦白——第一次看到我们的 AI 基础设施账单时,差点把咖啡喷出来。团队一直在用 GPT-4o 处理各种任务,其中一个不过是花哨的 FAQ 机器人,主要用十种不同方式回答"退款周期是多久"这种问题。一个月就要几千美元,实际上就是加了多层包装的 string.contains()。
这就是我从"这账单是什么鬼"到跑起一套分层路由系统的故事,85% 的请求用 $0.01/M 的模型处理。说实话,这些都不是我发明的——我只是从数据库查询优化器的 playbook 里偷了点子,用到了 LLM 调用上。如果你配过读副本或 CDN,已经懂了 80% 要讲的东西。
让我带你过一遍这七招,把我们从疯狂烧钱带到能在规划评审会上理直气壮汇报的预算。
我跟大多数团队交流时,发现他们选默认模型的的方式跟选默认文本编辑器一样——早早选一次,之后再也不回头。然后有人把它接进了十二个服务,成本随用量线性增长,等有人注意到时,你已经在用 GPT-4o 的价格来做客户评价摘要了。
本质上,你买的是能力梯度。不同任务需要不同的能力底线。情感分类器不需要博士水平;它只需要基本的模式识别。代码重构智能体则可以说更需要。
以下是我审计完实际工作负载后建的矩阵:
看这些数字。"昂贵选择"那一列,如果你把它们送到没有任何限速器的公开端点,应该让你浑身不舒服。
这是我保存在 models.yaml 里、启动时加载的路由表:
# models.py
from openai import OpenAI
client = OpenAI(
base_url="https://global-apis.com/v1",
api_key=os.environ["GLOBAL_API_KEY"],
)
MODEL_MAP = {
"chat_simple": "deepseek-v4-flash", # $0.25/M
"code": "deepseek-coder", # $0.25/M
"classify": "Qwen/Qwen3-8B", # $0.01/M
"summarize": "Qwen/Qwen3-32B", # $0.28/M
"translate": "Qwen-MT-Turbo", # $0.30/M
"reasoning": "deepseek-reasoner", # $2.50/M — last resort
}
def pick_model(user_input: str) -> str:
bucket = classify_complexity(user_input)
return MODEL_MAP[bucket]
resp = client.chat.completions.create(
model=pick_model(user_input),
messages=[{"role": "user", "content": user_input}],
)
那个 classify_complexity 函数简单得令人发指——基本上就是关键词/正则检查——但它节省的钱比所有其他优化加起来还多。私以为,这个单一改动值得本文其余所有内容。
一旦接受了不是每个请求都需要前沿模型,下一步就很明显了:建一个瀑布流。先试便宜的,快速失败,必要时升级。
这基本上就是编译器优化的工作方式——先走便宜的 passes,只有在便宜 pass 无法证明正确性时才走贵的。同一个思路,不同的领域。
def cascading_generate(prompt: str, budget_usd: float = 0.50) -> str:
"""
Try the cheapest model that can plausibly handle the request.
Escalate only when quality is insufficient.
"""
# Tier 1 — ultra-budget. Handles ~80% of traffic.
tier1 = call_model("Qwen/Qwen3-8B", prompt) # $0.01/M
if quality_score(tier1) >= 0.8:
return tier1
# Tier 2 — standard. Handles ~15% of traffic.
tier2 = call_model("deepseek-v4-flash", prompt) # $0.25/M
if quality_score(tier2) >= 0.9:
return tier2
# Tier 3 — premium. The remaining ~5%.
return call_model("deepseek-reasoner", prompt) # $2.50/M
quality_score 函数是秘密武器。对我们来说,它是一个小型分类器模型,看响应长度、拒绝短语的存在,以及对预期答案分布的轻量 embedding 距离检查。花了我们一个周末接起来,每个月在我们客户支持工作负载上省了约 $390——从每月 $420 降到 $28,方法是把 85% 的查询路由到 Qwen3-8B。
这里的教训不是"总是用便宜的模型"。教训是"默认用便宜的,证明你需要贵的才升级"。
我无法告诉你有多少次加入一个团队后发现他们的应用每天向 GPT 问了 4000 次完全相同的问题。"退货政策是什么?"根本不需要打到 LLM,如果你的退货政策在数据库里且从不变化的话。
LLM 缓存有三种口味,按复杂度递增排序:
精确匹配缓存——对 (model, messages) 元组做 hash,在 N 秒内存储响应。解决 50-80% 的常见查询(FAQ、文档、静态内容)。
语义缓存——将查询 embedding,在向量存储里找最近邻,如果余弦距离低于阈值就返回缓存响应。处理同义改写。
推测缓存——提前预计算可能出现的响应(想象一下:cron job 生成预期问题的答案)。
这是精确匹配版本,也是我建议你们从它开始的原因:
import hashlib
import json
import time
_cache: dict[str, dict] = {}
def cached_chat(model: str, messages: list, ttl_seconds: int = 3600):
key = hashlib.md5(
json.dumps({"model": model, "messages": messages}, sort_keys=True).encode()
).hexdigest()
hit = _cache.get(key)
if hit and (time.time() - hit["ts"]) < ttl_seconds:
metrics.counter("llm.cache.hit").inc()
return hit["response"]
response = client.chat.completions.create(
model=model, messages=messages
)
_cache[key] = {"response": response, "ts": time.time()}
metrics.counter("llm.cache.miss").inc()
return response
加 TTL 是因为模型行为会随时间漂移。加 LRU 淘汰是因为缓存不是免费的。加可观测性是因为你需要知道命中率,才能开始和财务掰扯账单。
Token 成本是很隐蔽的。人们看输出价格,忽略输入价格,然后上线一个 4000 token 的系统 prompt,然后奇怪为什么月账单看起来像电话号码。
压缩 prompt 不华丽,但它能回本。大致思路:用便宜模型先总结长上下文,再传给贵的。
def compress_prompt(text: str, target_ratio: float = 0.5) -> str:
if len(text) < 500:
return text # Don't bother with short prompts
target_chars = int(len(text) * target_ratio)
summary = call_model(
"Qwen/Qwen3-8B", # $0.01/M — almost free
f"Summarize the following in under {target_chars} characters, "
f"preserving all factual details:\n\n{text}",
)
return summary
让我给你看实际说服 PM 批准这项工作的具体数字:
一个 2000 token 的系统 prompt 压缩到 400 token,在 DeepSeek V4 Flash 上每次请求节省 $0.024。按每天 10,000 请求算,那就是 $240/天 → $87,600/年。仅仅来自一步压缩。出了这笔账之后,我甚至不需要掏出电子表格就能说服任何人。
诀窍是用一个比目标模型便宜很多的模型来压缩。用 GPT-4o 压缩 GPT-4o 的 prompt 是净亏损。用 $0.01/M 的 Qwen3-8B 压缩给 $0.25/M 的 DeepSeek V4 Flash 用?这才是在正确地玩优化游戏。
每次 LLM 调用都有开销——网络往返、JSON 解析、prompt 重新 tokenize。如果你连续发 N 个相似请求,可以合并成一个调用,用结构化 prompt 和单一响应。
这是直接从数据库 playbook 里拿的。如果你曾经写过 WHERE id IN (...) 而不是 N 个独立查询,你已经懂了。
# ❌ Before: N calls, N round-trips, N × input token cost
def classify_legacy(items: list[str]) -> list[str]:
results = []
for item in items:
r = client.chat.completions.create(
model="deepseek-v4-flash",
messages=[{
"role": "user",
"content": f"Classify sentiment as positive/negative: {item}"
}],
)
results.append(r.choices[0].message.content)
return results
# ✅ After: 1 call, 1 round-trip, batched prompt
def classify_batched(items: list[str]) -> list[str]:
numbered = "\n".join(f"{i}. {it}" for i, it in enumerate(items))
r = client.chat.completions.create(
model="deepseek-v4-flash",
messages=[{
"role": "user",
"content": (
f"Classify each line as POSITIVE or NEGATIVE. "
f"Return one label per line, in order:\n{numbered}"
),
}],
)
return r.choices[0].message.content.strip().splitlines()
大批量会丢失一些可靠性(模型可能漂移或跳过往后约 50-100 项的条目),所以要分块处理。私以为,分类任务的最佳批次大小是 20-30,生成任务是 10-15。
单纯批量处理预期能节省 10-20%,主要来自摊薄的输入 token 和减少的网络往返。更大的收益来自更大的批次,但你要用响应延迟来交换。
大多数团队忘了输出 token 比输入 token 贵——他们也没意识到可以在生成中途切断。如果你在摘要一篇文章,模型决定写结尾段,你可以停掉它。
我经常用的两个模式:
# 1. Streaming + early termination on a sentinel token
def summarize_with_stop(text: str) -> str:
stream = client.chat.completions.create(
model="Qwen/Qwen3-32B",
messages=[{"role": "user", "content": f"Summarize: {text}"}],
stream=True,
stop=["\n\n---", "In summary,", "Conclusion:"],
)
chunks = []
for chunk in stream:
if chunk.choices[0].finish_reason:
break
chunks.append(chunk.choices[0].delta.content or "")
return "".join(chunks)
# 2. Hard max_tokens cap, tuned per task
client.chat.completions.create(
model="Qwen3-8B",
messages=[...],
max_tokens=64, # classification doesn't need paragraphs
)
听起来trivial,但把 max_tokens 上限到真实最小值(而不是默认的 256 或 512), routinely 能从短表单任务的输出支出里削掉 15-25%。这相当于 SQL 里的 LIMIT——无聊,但不可或缺。
这是前期成本最高、长期回报也最高的策略。如果你有一个高频窄域任务——意图分类、实体提取、支持路由——在你的标注数据上微调一个小型开源模型,会让你的每次请求成本大幅下降。
以下是我见过的基本经济学演算:
预训练 Qwen3-8B 以 $0.01/M 的价格能正确处理约 75% 的意图分类。
在约 5k 标注样本上微调的 Qwen3-8B 能正确处理约 94%。
突然之间,你完全不需要把那些查询升级到 GPT-4o 了。
微调成本是一次性打击,可能是几百美元的 GPU 时间和一周清理标注数据的时间。盈亏平衡点在生产流量运行 1-3 个月左右,具体取决于量。
我不会在这里丢一个完整的训练脚本,因为 (a) 这超出了 API 成本的范围,(b) 它本身就是一个项目,但 tl;dr 是:从级联系统收集失败案例,标注,微调,重新部署。这就是循环。
还有两个值得简单提一下的技巧,因为它们和以上所有策略是叠加关系:
前缀缓存。大多数 LLM 提供商(以及 Global API 底层基础设施)会缓存长公共前缀的 KV 状态。如果你的系统 prompt 是稳定的,在各调用间保持完全一致——不要随机化空白或时间戳。Anthropic 和 OpenAI 都会对重复前缀大幅打折;Global API 的设置也很配合这个。
推测执行。当延迟比边际成本更重要时,你可以立即启动便宜模型,同时并行启动贵模型,返回先出结果的那个。听起来浪费,但对于有严格延迟预算的用户面向 UI,通常是净胜,因为用户感知速度转化留存,留存转化收入,这样 LLM 账单就变成了一个舍入误差。
以下是我现在在生产环境中的 services/llm.py,八九不离十:
import os
import hashlib
import json
import time
from openai import OpenAI
client = OpenAI(
base_url="https://global-apis.com/v1",
api_key=os.environ["GLOBAL_API_KEY"],
)
TIER_CONFIG = {
"ultra": {"model": "Qwen/Qwen3-8B", "cost_per_m": 0.01},
"std": {"model": "deepseek-v4-flash", "cost_per_m": 0.25},
"heavy": {"model": "Qwen3-32B", "cost_per_m": 0.28},
"premium":{"model": "deepseek-reasoner", "cost_per_m": 2.50},
}