大模型 API 成本优化：从月账单十万到三万的架构演进

一、Token 消耗与响应延迟的双重夹击：AI 创业公司的成本痛点

在 AI 创业公司的早期阶段，大模型 API 账单往往是最大的运营成本项之一。我们团队在产品上线后的第三个月，OpenAI 和 Anthropic 的合计月账单就突破了十万大关。同时，部分核心场景的平均响应时间超过 12 秒，用户流失率明显上升。

通过分析日志发现，成本爆炸主要来自三个方面：一是用户输入的超长文本没有合理截断；二是 Prompt 中重复的系统提示词占比超过 40%；三是多个 Agent 串行调用产生的累积消耗。在延迟方面，问题则出在网络重试策略和缺少请求优先级队列。

这些问题并非个例，几乎所有 AI 创业公司都会遇到。技术如果不服务于真实的成本控制，那只是美好的幻觉。我们需要一套可落地的成本优化方案，在不降低用户体验的前提下，将成本压下来。

二、Prompt 压缩与缓存架构设计：成本控制的核心机制

flowchart TD
    subgraph 用户请求层
        A[用户输入] --> B[输入截断器]
        B --> C[Prompt 模板]
    end
    
    subgraph 缓存与优化层
        C --> D[Prompt 压缩引擎]
        D --> E[语义缓存查询]
        E -->|缓存命中| F[直接返回结果]
        E -->|缓存未命中| G[Token 优化器]
    end
    
    subgraph 模型调用层
        G --> H[LLM API 调用]
        H --> I[结果写入缓存]
        I --> J[返回用户]
    end
    
    F --> J

2.1 系统提示词的分层设计与动态加载

我们将系统提示词拆分为基础层、领域层和任务层三个部分。基础层包含模型身份设定和安全约束，大约 200 Token；领域层根据用户使用的功能模块动态加载，例如数据分析模块会加载数据处理相关的指令；任务层则仅在特定场景下激活，如代码生成时加载最佳实践指南。

通过这种分层设计，大多数场景下的系统提示词从原来的 1500 Token 降低到 500 Token 以内，单次调用节省超过 65% 的基础 Token 消耗。

2.2 语义缓存的实现与失效策略

语义缓存是我们成本优化的核心武器。我们使用向量数据库对用户输入进行语义相似度匹配，当相似度超过 0.92 时直接返回缓存结果。但缓存策略不能一刀切，需要根据场景设置不同的 TTL：

• 工具调用类场景：TTL = 5 分钟（工具状态可能变化）
• 知识问答类场景：TTL = 24 小时
• 代码生成类场景：TTL = 7 天（语言规范相对稳定）

缓存命中率从上线初期的 12% 逐步提升到 47%，成为成本下降的最大贡献者。

2.3 Token 优化器的生产级实现

import tiktoken
from typing import List, Tuple, Optional
from dataclasses import dataclass

@dataclass
class PromptOptimizationResult:
    optimized_text: str
    token_savings: int
    original_tokens: int
    optimization_ratio: float

class TokenOptimizer:
    """
    Token 优化器，通过多种策略减少 Prompt 的 Token 消耗
    """
    
    def __init__(self, model_name: str = "gpt-4o"):
        self.encoder = tiktoken.encoding_for_model(model_name)
        self.max_context_window = {
            "gpt-4o": 128000,
            "gpt-3.5-turbo": 16384
        }.get(model_name, 16384)
    
    def count_tokens(self, text: str) -> int:
        """计算文本的 Token 数量"""
        return len(self.encoder.encode(text))
    
    def truncate_input(self, text: str, max_tokens: int = 4000) -> str:
        """智能截断输入文本，保留开头和结尾"""
        tokens = self.encoder.encode(text)
        if len(tokens) <= max_tokens:
            return text
        
        # 保留开头 30% 和结尾 70%
        keep_start = int(max_tokens * 0.3)
        keep_end = int(max_tokens * 0.7)
        
        start_part = self.encoder.decode(tokens[:keep_start])
        end_part = self.encoder.decode(tokens[-keep_end:])
        
        return f"{start_part}\n\n[...内容已截断...]\n\n{end_part}"
    
    def remove_redundant_whitespace(self, text: str) -> str:
        """移除冗余的空白字符"""
        lines = text.split("\n")
        cleaned_lines = []
        
        for line in lines:
            stripped = line.strip()
            if stripped:  # 只保留非空行
                cleaned_lines.append(line.rstrip())
        
        return "\n".join(cleaned_lines)
    
    def optimize_conversation_history(self, 
                                     messages: List[dict],
                                     max_tokens: int = 8000) -> List[dict]:
        """优化对话历史，移除早期低价值的消息"""
        total_tokens = sum(self.count_tokens(msg.get("content", "")) for msg in messages)
        
        if total_tokens <= max_tokens:
            return messages
        
        # 从最早的消息开始移除，直到满足 Token 限制
        optimized = messages.copy()
        
        while total_tokens > max_tokens and len(optimized) > 2:
            # 移除最早的用户-助手消息对
            if len(optimized) >= 2:
                removed_tokens = (
                    self.count_tokens(optimized[0].get("content", "")) +
                    self.count_tokens(optimized[1].get("content", ""))
                )
                optimized = optimized[2:]
                total_tokens -= removed_tokens
            else:
                break
        
        return optimized
    
    def full_optimization(self, text: str) -> PromptOptimizationResult:
        """执行完整的优化流程"""
        original_tokens = self.count_tokens(text)
        
        # 步骤 1: 移除冗余空白
        optimized = self.remove_redundant_whitespace(text)
        
        # 步骤 2: 智能截断（如果需要）
        optimized = self.truncate_input(optimized)
        
        final_tokens = self.count_tokens(optimized)
        savings = original_tokens - final_tokens
        ratio = savings / original_tokens if original_tokens > 0 else 0.0
        
        return PromptOptimizationResult(
            optimized_text=optimized,
            token_savings=savings,
            original_tokens=original_tokens,
            optimization_ratio=ratio
        )


# 使用示例
optimizer = TokenOptimizer()

# 优化单个 Prompt
long_prompt = """
这是一段非常长的系统提示词...
（省略数百行内容）
...这是提示词的结尾。
"""

result = optimizer.full_optimization(long_prompt)
print(f"原始 Token: {result.original_tokens}")
print(f"优化后 Token: {result.original_tokens - result.token_savings}")
print(f"节省比例: {result.optimization_ratio:.1%}")

三、批量请求与队列系统实现：高并发场景的成本与延迟控制

3.1 请求优先级队列设计

在高并发场景下，我们需要区分不同请求的优先级。我们将请求分为四个优先级等级：

• P0（关键路径）：用户直接交互的核心功能，如对话回复
• P1（重要功能）：数据分析、报告生成等需要快速响应的功能
• P2（后台任务）：定期数据同步、批量处理等
• P3（离线任务）：模型微调准备、日志分析等

import asyncio
import heapq
from enum import IntEnum
from typing import Callable, Any, Optional
from dataclasses import dataclass, field
from datetime import datetime, timedelta

class Priority(IntEnum):
    P0 = 0  # 关键路径，最高优先级
    P1 = 1  # 重要功能
    P2 = 2  # 后台任务
    P3 = 3  # 离线任务，最低优先级

@dataclass(order=True)
class QueuedRequest:
    priority: Priority
    enqueue_time: datetime
    request_id: str = field(compare=False)
    callback: Callable = field(compare=False)
    timeout: timedelta = field(default=timedelta(seconds=30), compare=False)

class PriorityRequestQueue:
    """
    优先级请求队列，支持批量处理和超时控制
    """
    
    def __init__(self, 
                 max_batch_size: int = 10,
                 batch_timeout: float = 2.0):
        self.queue: List[QueuedRequest] = []
        self.max_batch_size = max_batch_size
        self.batch_timeout = batch_timeout
        self.lock = asyncio.Lock()
        self.request_counter = 0
    
    async def enqueue(self, 
                      priority: Priority,
                      callback: Callable,
                      timeout: Optional[timedelta] = None) -> str:
        """将请求加入队列"""
        async with self.lock:
            self.request_counter += 1
            request_id = f"req_{self.request_counter}"
            
            queued = QueuedRequest(
                priority=priority,
                enqueue_time=datetime.now(),
                request_id=request_id,
                callback=callback,
                timeout=timeout or timedelta(seconds=30)
            )
            
            heapq.heappush(self.queue, queued)
            return request_id
    
    async def dequeue_batch(self) -> List[QueuedRequest]:
        """批量取出请求"""
        async with self.lock:
            batch = []
            
            while self.queue and len(batch) < self.max_batch_size:
                request = heapq.heappop(self.queue)
                
                # 检查是否已超时
                if datetime.now() - request.enqueue_time < request.timeout:
                    batch.append(request)
                # 超时的请求直接丢弃
            
            return batch
    
    async def process_queue(self, processor: Callable[[List[dict]], Any]):
        """处理队列中的请求"""
        while True:
            try:
                batch = await self.dequeue_batch()
                
                if batch:
                    # 构建批量请求
                    requests = []
                    for req in batch:
                        # 这里需要与 callback 约定好数据结构
                        requests.append({
                            "request_id": req.request_id,
                            "callback": req.callback
                        })
                    
                    # 批量处理
                    await processor(requests)
                
                # 等待下一批或超时
                await asyncio.sleep(self.batch_timeout if not batch else 0.1)
                
            except Exception as e:
                print(f"队列处理异常: {e}")
                await asyncio.sleep(1)


# 使用示例
async def llm_batch_processor(requests: List[dict]):
    """批量处理 LLM 请求"""
    # 实际实现中，这里会调用 LLM API 的批量接口
    print(f"批量处理 {len(requests)} 个请求")
    
    # 模拟处理结果
    for req in requests:
        await req["callback"]({"result": "处理完成"})


queue = PriorityRequestQueue(max_batch_size=5)

# 启动队列处理器
asyncio.create_task(queue.process_queue(llm_batch_processor))

3.2 模型降级与负载均衡策略

当 API 调用量突增或某个模型服务出现故障时，需要一套降级策略：

1. 模型降级：GPT-4o 不可用时自动降级到 GPT-3.5-turbo，同时在 UI 上提示用户
2. 负载均衡：在多个供应商（OpenAI、Anthropic、国内模型）之间分配流量
3. 速率限制：每个用户和全局都设置合理的 QPS 限制，防止滥用

四、成本与体验的权衡分析：优化方案的边界条件

任何优化方案都有其适用边界，成本优化也不例外。在实施过程中，我们需要明确以下权衡：

优化策略	成本节省	体验影响	适用场景	禁用场景
Prompt 截断	高	低（智能截断）	长文本理解	代码审查、法律文档分析
语义缓存	极高	低（相似结果复用）	知识库问答	实时数据查询
模型降级	中	中（质量下降）	非核心功能	精准内容生成
批量处理	中	中（延迟增加）	后台任务	实时交互

4.1 监控与告警体系的重要性

为了及时发现优化策略带来的问题，我们建立了完整的监控体系：

• 成本指标：每请求平均 Token 消耗、缓存命中率、各模型调用占比
• 体验指标：平均响应时间、P95 延迟、用户满意度评分
• 质量指标：输出评分（通过小模型自动评估）、错误率

通过对比优化前后的指标变化，我们可以量化每个优化策略的效果，并及时调整参数。例如，当发现输出评分下降超过 5% 时，我们会降低缓存相似度阈值或调整 Prompt 截断策略。

4.2 长期优化方向：自建模型与混合架构

从长期来看，完全依赖第三方 API 不是可持续的方案。我们正在探索混合架构：

1. 核心场景：继续使用高质量的 GPT-4o 或 Claude 3
2. 标准场景：使用微调后的开源模型（如 Llama 3）
3. 简单场景：使用训练的分类模型或规则引擎

这种混合架构预计可以在保证质量的前提下，将成本进一步降低 50% 以上。

五、总结

大模型 API 成本优化是一个持续迭代的过程。通过 Prompt 分层设计、语义缓存、Token 优化和批量处理等手段，我们成功将月账单从十万降低到三万，同时保持了用户体验。

在实施优化时，关键是建立完整的监控体系，量化每个策略的效果，并根据数据及时调整。同时，要明确优化方案的边界条件，避免为了省钱而牺牲核心功能的质量。

从长期来看，混合架构（大模型 + 微调模型 + 规则）是更可持续的成本控制方案，既保留了大模型的能力，又降低了运营成本。