AI API 成本优化：我如何把月账单从 8000 降到 1200
上个月老板看到 API 账单的时候，脸色不太好看。
"这个月怎么花了 8000 多？"
说实话，我也没想到会这么贵。项目用了 GPT-4 做代码审查，每天处理几百个 PR，token 消耗确实有点猛。
经过一个月的优化，现在月账单稳定在 1200 左右。这篇文章分享我踩过的坑和总结的经验。
成本分析：钱都花在哪了
首先得搞清楚钱是怎么花的。我写了一个脚本分析 API 调用日志：
PYTHON
import json
from collections import defaultdict
from datetime import datetime
def analyze_api_costs(log_file):
"""分析 API 调用成本"""
costs = defaultdict(float)
# 各模型价格 (每 1K token)
PRICES = {
"gpt-4": {"input": 0.03, "output": 0.06},
"gpt-4-turbo": {"input": 0.01, "output": 0.03},
"gpt-3.5-turbo": {"input": 0.0005, "output": 0.0015},
}
with open(log_file) as f:
for line in f:
record = json.loads(line)
model = record["model"]
input_tokens = record["prompt_tokens"]
output_tokens = record["completion_tokens"]
price = PRICES.get(model, PRICES["gpt-3.5-turbo"])
cost = (input_tokens * price["input"] +
output_tokens * price["output"]) / 1000
costs[model] += cost
return dict(costs)
# 运行分析
costs = analyze_api_costs("api_logs.json")
for model, cost in sorted(costs.items(), key=lambda x: -x[1]):
print(f"{model}: ${cost:.2f}")结果发现：
- GPT-4 占了 70% 的成本
- 很多请求其实不需要 GPT-4
- 重复请求没有缓存
优化策略 1：模型降级
不是所有任务都需要最贵的模型。我建立了一个路由策略：
PYTHON
class ModelRouter:
"""根据任务复杂度选择合适的模型"""
def __init__(self):
self.client = OpenAI()
def route(self, task_type: str, prompt: str) -> str:
"""路由到合适的模型"""
# 简单任务用 GPT-3.5
if task_type in ["format", "translate", "summarize"]:
return self.call_gpt35(prompt)
# 代码相关用 GPT-4-turbo (性价比更高)
if task_type in ["code_review", "refactor"]:
return self.call_gpt4_turbo(prompt)
# 复杂推理才用 GPT-4
if task_type in ["architecture", "debug_complex"]:
return self.call_gpt4(prompt)
return self.call_gpt35(prompt)
def call_gpt35(self, prompt: str) -> str:
response = self.client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": prompt}],
temperature=0.3
)
return response.choices[0].message.content
def call_gpt4_turbo(self, prompt: str) -> str:
response = self.client.chat.completions.create(
model="gpt-4-turbo-preview",
messages=[{"role": "user", "content": prompt}],
temperature=0.3
)
return response.choices[0].message.content这一招直接省了 40% 的成本。
优化策略 2：智能缓存
很多请求是重复的，比如同一个文件的代码审查。我加了一个语义缓存：
PYTHON
import hashlib
import json
from pathlib import Path
class SemanticCache:
"""基于内容哈希的语义缓存"""
def __init__(self, cache_dir: str = ".cache"):
self.cache_dir = Path(cache_dir)
self.cache_dir.mkdir(exist_ok=True)
def _get_key(self, prompt: str, model: str) -> str:
"""生成缓存键"""
content = f"{model}:{prompt}"
return hashlib.sha256(content.encode()).hexdigest()[:16]
def get(self, prompt: str, model: str) -> str | None:
"""获取缓存"""
key = self._get_key(prompt, model)
cache_file = self.cache_dir / f"{key}.json"
if cache_file.exists():
data = json.loads(cache_file.read_text())
# 缓存 24 小时
if time.time() - data["timestamp"] < 86400:
return data["response"]
return None
def set(self, prompt: str, model: str, response: str):
"""设置缓存"""
key = self._get_key(prompt, model)
cache_file = self.cache_dir / f"{key}.json"
data = {
"prompt": prompt,
"model": model,
"response": response,
"timestamp": time.time()
}
cache_file.write_text(json.dumps(data))
# 使用示例
cache = SemanticCache()
def smart_call(prompt: str, model: str) -> str:
# 先查缓存
cached = cache.get(prompt, model)
if cached:
print(f"[Cache Hit] {model}")
return cached
# 调用 API
response = call_api(prompt, model)
# 写入缓存
cache.set(prompt, model, response)
return response缓存命中率达到了 35%，又省了一笔。
优化策略 3：Prompt 压缩
Prompt 越长，token 消耗越大。我写了一个 prompt 压缩器：
PYTHON
class PromptCompressor:
"""压缩 Prompt，减少 token 消耗"""
def __init__(self):
# 常见冗余词
self.fillers = [
"please", "kindly", "I would like you to",
"could you", "would you mind", "if possible"
]
def compress(self, prompt: str) -> str:
"""压缩 prompt"""
result = prompt
# 移除冗余词
for filler in self.fillers:
result = result.replace(filler, "")
# 压缩空白
result = " ".join(result.split())
# 移除重复指令
lines = result.split("\n")
seen = set()
unique_lines = []
for line in lines:
normalized = line.strip().lower()
if normalized and normalized not in seen:
seen.add(normalized)
unique_lines.append(line)
return "\n".join(unique_lines)
# 对比效果
compressor = PromptCompressor()
original = """
Please kindly review the following code. I would like you to
check for any potential bugs, security issues, and performance
problems. Could you also suggest improvements if possible?
Please kindly review the following code...
"""
compressed = compressor.compress(original)
print(f"Original: {len(original)} chars")
print(f"Compressed: {len(compressed)} chars")
# 输出: Original: 234 chars, Compressed: 142 chars优化策略 4：批量处理
单个请求的 overhead 很高。我把多个小请求合并成批量请求：
PYTHON
class BatchProcessor:
"""批量处理请求，减少 API 调用次数"""
def __init__(self, batch_size: int = 10):
self.batch_size = batch_size
self.queue = []
def add(self, task: dict):
"""添加任务到队列"""
self.queue.append(task)
if len(self.queue) >= self.batch_size:
return self.flush()
return None
def flush(self):
"""处理队列中的所有任务"""
if not self.queue:
return []
# 构建批量 prompt
batch_prompt = "请依次处理以下任务，每个任务单独输出结果：\n\n"
for i, task in enumerate(self.queue, 1):
batch_prompt += f"任务 {i}: {task['prompt']}\n\n"
# 单次 API 调用
response = call_api(batch_prompt, "gpt-4-turbo")
# 解析结果
results = self._parse_batch_response(response)
self.queue = []
return results
def _parse_batch_response(self, response: str) -> list:
"""解析批量响应"""
# 按任务分割结果
results = []
current = []
for line in response.split("\n"):
if line.startswith("任务") and ":" in line:
if current:
results.append("\n".join(current))
current = [line]
else:
current.append(line)
if current:
results.append("\n".join(current))
return results优化效果总结
| 策略 | 节省比例 | 实施难度 |
|------|----------|----------|
| 模型降级 | 40% | 低 |
| 智能缓存 | 25% | 中 |
| Prompt 压缩 | 15% | 低 |
| 批量处理 | 10% | 中 |
综合下来，月账单从 8000 降到了 1200，节省了 85%。
一些额外建议
1. 设置预算告警 - 在 OpenAI 后台设置用量上限，超了自动停止
2. 监控 token 使用 - 每天看一次 dashboard，发现异常及时处理
3. 考虑替代方案 - 对于简单任务，本地模型可能更划算
4. 谈判企业价格 - 用量大的话可以联系 OpenAI 谈折扣
API 成本优化是个持续的过程，希望这些经验对你有帮助。
读者评论 3