AI API 成本优化实战

API成本为什么容易失控

DeepSeek API的价格确实很便宜——输入约¥1/百万token，输出约¥2/百万token。一个典型的对话可能只消耗几百到几千token，费用不到一分钱。但成本问题往往在规模化后才暴露：日均10万次对话×每次2000 token = 每天2亿token ≈ ¥200-400/天 ≈ ¥6000-12000/月。如果再加上RAG检索（每次检索消耗额外token）、多轮对话（上下文越来越长）、以及使用deepseek-reasoner（推理token额外计费），月度账单很容易突破数万元。

更隐蔽的是"隐性浪费"——很多token消耗是没有产生价值的：重复的系统提示词（每次请求都发送相同的system prompt）、过长的对话历史（用户在第15轮还在传第1轮的完整内容）、无效的重试（API超时后重新发送完整请求）。识别并消除这些浪费，是实现成本优化的第一步。

策略一：提示词压缩

系统提示词是最大的隐性浪费来源。如果你的system prompt有500个汉字（约750 token），日均1万次请求，仅system prompt每天就消耗750万token ≈ ¥7.5。优化策略：精简系统提示词（能用50字说清楚就不要用500字）、对固定系统提示词使用缓存（不要每次动态生成）、将不常用的规则移到"按需加载"（只在检测到相关场景时才附加特定规则）。

from openai import OpenAI
import hashlib, json, time

client = OpenAI(api_key="your-deepseek-api-key", base_url="https://api.deepseek.com")

class CostOptimizer:
    def __init__(self):
        self.cache = {}  # 语义缓存
        self.stats = {"total_tokens":0, "cached_hits":0, "saved_tokens":0}

    def semantic_cache_key(self, messages):
        """生成语义缓存键"""
        last_user_msg = next((m["content"] for m in reversed(messages) if m["role"]=="user"), "")
        return hashlib.md5(last_user_msg.encode()).hexdigest()

    def call_with_cache(self, messages, model="deepseek-chat", ttl=3600, **kwargs):
        """带语义缓存的API调用"""
        cache_key = self.semantic_cache_key(messages)
        if cache_key in self.cache:
            entry = self.cache[cache_key]
            if time.time() - entry["timestamp"] < ttl:
                self.stats["cached_hits"] += 1
                self.stats["saved_tokens"] += entry["tokens"]
                print(f"缓存命中! 节省约 {entry['tokens']} tokens")
                return entry["response"]

        response = client.chat.completions.create(
            model=model, messages=messages, **kwargs
        )
        result = response.choices[0].message.content
        tokens = response.usage.total_tokens
        self.stats["total_tokens"] += tokens
        self.cache[cache_key] = {"response":result, "tokens":tokens, "timestamp":time.time()}
        return result

    def compress_history(self, messages, max_turns=8):
        """压缩对话历史"""
        if len(messages) <= max_turns * 2 + 2:
            return messages
        # 保留system prompt + 最近N轮 + 早期摘要
        system = [m for m in messages if m["role"]=="system"]
        recent = messages[-(max_turns*2):]
        old = messages[1:-(max_turns*2)]
        if old:
            old_text = "\n".join(f"{m['role']}: {m['content'][:100]}" for m in old)
            summary = client.chat.completions.create(
                model="deepseek-chat",
                messages=[{"role":"user","content":f"压缩为50字摘要：{old_text}"}],
                max_tokens=100
            ).choices[0].message.content
            system.append({"role":"system","content":f"[对话摘要] {summary}"})
        return system + recent

    def smart_route(self, messages):
        """智能模型路由：简单问题用V3，复杂问题用R1"""
        last_msg = messages[-1]["content"] if messages else ""
        simple_signals = ["你好","谢谢","再见","是的","对的","好的"]
        complex_signals = ["分析","推理","证明","优化","比较","设计"]
        if any(s in last_msg for s in simple_signals) and not any(s in last_msg for s in complex_signals):
            return "deepseek-chat"  # 简单问题用V3
        if any(s in last_msg for s in complex_signals):
            return "deepseek-reasoner"  # 复杂问题用R1
        return "deepseek-chat"  # 默认V3

    def batch_requests(self, requests_list):
        """批量处理：将多个独立请求合并为一次API调用"""
        combined = "请分别回答以下问题，用数字序号分隔：\n\n"
        for i, req in enumerate(requests_list):
            combined += f"问题{i+1}：{req}\n"
        response = client.chat.completions.create(
            model="deepseek-chat",
            messages=[{"role":"user","content":combined}]
        )
        return response.choices[0].message.content

optimizer = CostOptimizer()
msgs = [{"role":"system","content":"你是AI助手"},{"role":"user","content":"什么是RAG？"}]
result = optimizer.call_with_cache(msgs)
compressed = optimizer.compress_history(msgs)
model = optimizer.smart_route(msgs)
print(f"缓存统计: {json.dumps(optimizer.stats, ensure_ascii=False)}")

策略二：多级缓存体系

缓存是降低API成本最有效的手段。构建三级缓存：精确匹配缓存（相同输入→相同输出，适合固定问答如FAQ）、语义相似缓存（相似输入→复用输出，使用embedding判断相似度，阈值0.95以上命中）、前缀缓存（DeepSeek API原生支持，相同前缀的请求自动复用KV Cache）。合理使用缓存可以减少30%-60%的API调用量。

策略三：模型分级路由

不是所有请求都需要最强的模型。构建一个路由器：简单问候和FAQ→使用最小的模型或缓存直接回复；中等复杂度问题→使用deepseek-chat（V3）；高复杂度推理→使用deepseek-reasoner（R1）。分级路由可以将R1的使用量控制在总量的10%-20%，大幅降低成本。

策略四：批处理与异步化

将多个独立请求合并为一次API调用（batch processing），可以减少HTTP往返开销，有时还能享受批处理折扣。对于非实时场景（如夜间批量生成报告），使用异步任务队列（Celery/RQ）将请求攒批，在业务低峰期集中处理。批处理可以将单次请求的成本降低20%-40%。

成本优化效果追踪

建立成本Dashboard，实时追踪：每日/每周/每月API费用、Top消耗功能/用户、缓存命中率、模型使用分布（V3 vs R1比例）、平均每次对话token消耗。没有数据驱动的成本优化就像盲人摸象——你可能在某个环节省了10%，却在另一个环节多花了50%。

Token消耗的精细化管理

除了之前提到的优化策略，还有一些容易被忽视的token消耗来源需要管理：提示词中的"空白字符"——多余的换行、缩进和空格虽然看似无害，但在大规模调用中累积起来可以占总token的3-5%。建议在发送请求前对消息进行"瘦身"处理——合并连续换行、移除行首尾空白。不必要的字段——在Function Calling中，模型会返回function_call的arguments，如果你的函数定义中有大量不必要参数的描述，会显著增加token消耗。建议精简函数定义，只保留必要参数和简短描述。RAG检索的冗余上下文——检索返回的top-k结果中，往往只有前3个真正有用，后面的可能是噪音。建议先用reranker重新排序后再截断，只保留真正相关的上下文。成本归属与内部结算：在多团队共享API Key的场景中，需要能够将成本归属到具体团队或项目，以便进行成本管控和内部结算。实现方式：在每次API调用的metadata中标记team_id和project_id，通过日志分析按团队/项目聚合成本。对于成本特别高的团队或项目，可以设置独立的API Key和预算上限。大型企业通常采用"中心采购、内部结算"的模式——AI平台团队统一采购DeepSeek API额度，各业务团队按实际使用量进行内部结算。

降本增效的综合路线图

如果你的AI API月账单超过¥10000，建议按照以下优先级逐步实施成本优化：第一步（立即见效，1周内完成）：实施精确匹配缓存+压缩系统提示词+设置max_tokens限制，通常可减少20-30%成本。第二步（中等投入，1个月内完成）：实施语义缓存+对话历史压缩+模型分级路由，可额外减少20-30%成本。第三步（长期持续）：建立成本监控Dashboard+定期成本回顾+批量处理优化+定期清理不活跃的缓存条目。综合实施后，月度API成本通常可以降低50%-70%。

最后需要强调的是，成本优化是一个持续的过程而非一次性的项目。建议建立月度成本回顾机制——每月初分析上月的费用数据，识别新的浪费来源，评估已有优化措施的效果，制定当月的优化目标。将成本优化纳入团队的OKR，让成本意识成为团队的共同文化。当每个人都意识到"每多传100个无意义的token就是在浪费团队预算"时，成本自然会得到有效控制。