AI API成本控制实战指南：我们如何把月账单从$5000降到$800

import time import logging logger = logging.getLogger("api_cost_tracker") class CostTracker: def __init__(self): self.calls = [] def track_call(self, model, prompt_tokens, completion_tokens, cost_per_1k): prompt_cost = (prompt_tokens / 1000) * cost_per_1k["prompt"] completion_cost = (completion_tokens / 1000) * cost_per_1k["completion"] total_cost = prompt_cost + completion_cost record = { "model": model, "prompt_tokens": prompt_tokens, "completion_tokens": completion_tokens, "total_cost": round(total_cost, 6), "timestamp": time.time() } self.calls.append(record) logger.info(f"API Call: {model} | " f"Tokens: {prompt_tokens + completion_tokens} | " f"Cost: ${total_cost:.4f}") return record def get_daily_summary(self): from collections import defaultdict daily = defaultdict(lambda: {"cost": 0, "calls": 0}) for call in self.calls: day = time.strftime("%Y-%m-%d", time.localtime(call["timestamp"])) daily[day]["cost"] += call["total_cost"] daily[day]["calls"] += 1 return dict(daily)

import hashlib import json from datetime import datetime, timedelta class SemanticCache: def __init__(self, similarity_threshold=0.92, ttl_hours=24): self.cache_store = {} # 生产环境建议用Redis self.threshold = similarity_threshold self.ttl = timedelta(hours=ttl_hours) self.hits = 0 self.misses = 0 def _get_embedding(self, text): """调用轻量级Embedding模型获取向量""" # 使用 text-embedding-3-small，成本极低 # $0.02 / 1M tokens response = embed_client.embeddings.create( model="text-embedding-3-small", input=text ) return response.data[0].embedding def _cosine_similarity(self, vec_a, vec_b): dot = sum(a * b for a, b in zip(vec_a, vec_b)) norm_a = sum(a ** 2 for a in vec_a) ** 0.5 norm_b = sum(b ** 2 for b in vec_b) ** 0.5 return dot / (norm_a * norm_b) if norm_a and norm_b else 0 def get(self, query): """查询缓存，返回命中结果或None""" query_vec = self._get_embedding(query) now = datetime.now() for cache_key, cache_entry in self.cache_store.items(): if now - cache_entry["time"] > self.ttl: continue sim = self._cosine_similarity(query_vec, cache_entry["vector"]) if sim >= self.threshold: self.hits += 1 cache_entry["hit_count"] += 1 return { "answer": cache_entry["answer"], "similarity": round(sim, 4), "cached": True } self.misses += 1 return None def set(self, query, answer): """将问答对存入缓存""" vec = self._get_embedding(query) key = hashlib.md5(query.encode()).hexdigest() self.cache_store[key] = { "vector": vec, "answer": answer, "time": datetime.now(), "hit_count": 0 } @property def hit_rate(self): total = self.hits + self.misses return self.hits / total if total > 0 else 0

class TaskRouter: """根据任务特征自动选择最优模型""" # 任务类型到模型的映射 TASK_MODEL_MAP = { "classification": "gpt-4o-mini", # 分类任务 "extraction": "gpt-4o-mini", # 信息提取 "simple_qa": "gpt-4o-mini", # 简单问答 "summarization": "gpt-4o", # 摘要生成 "translation": "gpt-4o", # 翻译 "content_writing": "gpt-4o", # 内容写作 "code_generation": "o3", # 代码生成 "complex_reasoning": "o3", # 复杂推理 "creative_writing": "o3", # 创意写作 } def classify_task(self, user_input, context=None): """ 使用规则 + 轻量分类器判断任务类型 返回推荐的模型名称 """ # 规则1：根据关键词快速匹配 keyword_rules = { "classification": ["分类", "归类", "判断是", "属于哪"], "extraction": ["提取", "找出", "列出所有", "抓取"], "simple_qa": ["是什么", "什么是", "怎么查", "FAQ"], "summarization": ["总结", "摘要", "概括", "归纳"], "translation": ["翻译", "translate", "译成"], "code_generation": ["写代码", "编程", "实现", "开发"], "complex_reasoning": ["分析", "推理", "为什么", "原因"], "creative_writing": ["创作", "写一篇", "故事", "诗歌"], } for task_type, keywords in keyword_rules.items(): for kw in keywords: if kw in user_input: return self.TASK_MODEL_MAP[task_type] # 规则2：根据输入长度判断复杂度 if len(user_input) < 50: return "gpt-4o-mini" # 短输入大概率是简单任务 # 规则3：默认使用标准模型 return "gpt-4o" def get_model(self, user_input, context=None): task_type = self.classify_task(user_input, context) model = self.TASK_MODEL_MAP.get(task_type, "gpt-4o") return model, task_type # 使用示例 router = TaskRouter() model, task = router.get_model("请把这段话翻译成英文") # 返回: ("gpt-4o", "translation") model, task = router.get_model("这个客户反馈属于哪个类别？") # 返回: ("gpt-4o-mini", "classification")

class ContextCompressor: def __init__(self, max_context_rounds=5, summary_max_tokens=200): self.max_rounds = max_context_rounds self.summary_max_tokens = summary_max_tokens def compress_context(self, conversation_history): """ 压缩对话历史： - 保留最近N轮完整对话 - 将更早的对话压缩为摘要 """ if len(conversation_history) <= self.max_rounds: return conversation_history # 需要压缩的早期对话 old_messages = conversation_history[:-self.max_rounds] # 保留的近期对话 recent_messages = conversation_history[-self.max_rounds:] # 用轻量模型生成摘要 summary_prompt = ( "请将以下对话历史压缩为一段简洁的摘要，" f"不超过{self.summary_max_tokens}字。" "保留关键信息：用户诉求、已解决的问题、待处理事项。\n\n" + "\n".join( f"{msg['role']}: {msg['content']}" for msg in old_messages ) ) summary = self._call_mini_model(summary_prompt) # 构建压缩后的上下文 compressed = [ {"role": "system", "content": f"[之前的对话摘要] {summary}"} ] + recent_messages # 计算节省的Token数 original_tokens = sum( len(msg["content"]) for msg in conversation_history ) compressed_tokens = len(summary) + sum( len(msg["content"]) for msg in recent_messages ) saved_pct = (1 - compressed_tokens / original_tokens) * 100 return compressed, { "original_tokens": original_tokens, "compressed_tokens": compressed_tokens, "saved_percent": round(saved_pct, 1) } def _call_mini_model(self, prompt): """调用轻量模型生成摘要""" # 实际实现中调用 gpt-4o-mini response = client.chat.completions.create( model="gpt-4o-mini", messages=[{"role": "user", "content": prompt}], max_tokens=self.summary_max_tokens ) return response.choices[0].message.content

# 优化前：不限制输出长度 response = client.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": "总结这篇文章的要点"}] ) # 平均输出: 450 tokens, 成本: $0.0045 # 优化后：根据需求设置合理的max_tokens response = client.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": "总结这篇文章的要点，不超过100字"}], max_tokens=150 # 留一点余量 ) # 平均输出: 120 tokens, 成本: $0.0012 # 节省: 73%

import json # 要求模型输出JSON格式，天然简洁 response = client.chat.completions.create( model="gpt-4o-mini", messages=[ {"role": "system", "content": ( "你是一个客服助手。请以JSON格式回答用户问题。" "格式: {\"answer\": \"简短回答\", " "\"category\": \"问题分类\", " "\"needs_human\": true/false}" )}, {"role": "user", "content": "我的订单还没收到，已经5天了"} ], max_tokens=200 ) result = json.loads(response.choices[0].message.content) # 输出示例: # { # "answer": "您的订单预计3个工作日内送达，如超时可申请补偿。", # "category": "物流问题", # "needs_human": false # } # Token消耗: 约80 tokens，比自由文本输出节省60%+

# 批量处理示例：将1000条分类任务打包 import json batch_requests = [] for i, feedback in enumerate(feedback_list): batch_requests.append({ "custom_id": f"req-{i}", "method": "POST", "url": "/v1/chat/completions", "body": { "model": "gpt-4o-mini", "messages": [ {"role": "system", "content": "将用户反馈分类为：产品质量/物流/服务/其他"}, {"role": "user", "content": feedback} ], "max_tokens": 20 } }) # 提交批量任务（价格减半） batch_file = write_jsonl(batch_requests) batch = client.batches.create( input_file_id=batch_file.id, endpoint="/v1/chat/completions", completion_window="24h" ) # 1000条分类任务成本：从$0.15降至$0.075