关于AI成本优化的几点记录

AI成本 云成本 成本优化
作者: 醉月思 发布日期: 2024-08-11 分类: 技术实践

增长本身没问题，但有几个迹象不太对：

  1. 成本曲线不是平滑的，有些天莫名其妙高 30-50%
  2. 同样功能的 API 调用，成本差异明显
  3. GPU 实例经常显示"使用中"但实际上没人在跑任务
  4. 有些 API 调用返回错误，但依然被计费

这时候才意识到：我们对成本结构基本是盲的。

背景：账单是怎么变大的

事情是这样的。我们团队在做 AI Agent 开发，主要用这几个服务：

  • OpenAI API（用于 LLM 推理）
  • Anthropic Claude API（用于复杂任务）
  • AWS SageMaker（部署模型推理服务）
  • 阿里云 GPU 实例（训练和推理混用）

初期成本还好，但随着用户量上来，账单开始"稳步增长"。增长本身没问题，但有几个迹象不太对：

  1. 成本曲线不是平滑的，有些天莫名其妙高 30-50%
  2. 同样功能的 API 调用，成本差异明显
  3. GPU 实例经常显示"使用中"但实际上没人在跑任务
  4. 有些 API 调用返回错误，但依然被计费

这时候才意识到：我们对成本结构基本是盲的。

需求：把账单变成可读的信息

要解决这个问题，先得搞清楚成本构成。我列了几件必须搞清楚的事：

  • 每个服务的计费模型到底是什么？
  • 哪些操作是最主要的成本来源？
  • 有没有明显浪费或不合理的使用？
  • 如何建立日常监控，而不是每次等账单来了才发现问题？

需求很简单，但实现起来有不少细节要处理。

计费模型梳理

先从计费模型开始。不同服务的计费方式差别很大，这里只记几个关键的。

OpenAI API

OpenAI 的计费主要看两个维度：token 数量和模型定价。

  • 输入 token：通常比输出 token 便宜
  • 输出 token：生成内容的部分
  • 不同模型定价差异巨大

以 GPT-4o 为例（2026年7月价格）：

  • 输入：$2.50 / 1M tokens
  • 输出：$10.00 / 1M tokens

但问题是，这些价格会变。而且不同使用场景下，token 实际消耗可能有很大差异。

Anthropic Claude

Claude 的计费模型类似，但有几个区别：

  • 部分模型有更详细的缓存计费（Prompt Caching）
  • 有些操作按请求次数收费，不只按 token
  • 不同 region 价格不同

云 GPU 实例

GPU 实例是按时间收费的，规则更复杂：

  • 按小时计费，不足一小时也按一小时算
  • 不同 GPU 型号价格差异巨大
  • Spot 实例便宜但可能被回收
  • 预留实例长期划算但需要预测用量

这里有个常见的坑：很多人以为 GPU 实例按分钟计费，但实际上大部分云服务是按小时截断计费的。

# AWS EC2 实例计费示例
# 比如实例价格 $2.5/小时
# 运行 1.2 小时 = 计费 2 小时 = $5.0
# 运行 4.1 小时 = 计费 5 小时 = $12.5

这个问题在短期内看不出，长期积累就很可观。

实现监控体系

搞清楚计费模型后，第一步是建立监控。不然优化了也不知道效果。

数据收集

我写了一个简单的数据收集脚本，定期从各个服务拉取使用数据：

import requests
import json
from datetime import datetime, timedelta

def collect_openai_usage(api_key, start_date, end_date):
    """收集 OpenAI API 使用情况"""
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json"
    }
    
    # 获取使用记录
    response = requests.get(
        "https://api.openai.com/v1/usage",
        headers=headers,
        params={"start_date": start_date, "end_date": end_date}
    )
    
    if response.status_code == 200:
        usage_data = response.json()
        # 按模型和操作类型聚合
        usage_summary = {}
        for item in usage_data.get("data", []):
            model = item.get("model", "unknown")
            operation = item.get("operation_type", "unknown")
            cost = item.get("n_generated_tokens", 0) * item.get("cost_per_1k_tokens", 0) / 1000
            
            key = f"{model}_{operation}"
            usage_summary[key] = usage_summary.get(key, 0) + cost
        
        return usage_summary
    else:
        print(f"OpenAI API 请求失败: {response.status_code}")
        return {}

def collect_aws_billing(region, start_date, end_date):
    """收集 AWS 账单数据"""
    import boto3
    
    client = boto3.client('ce', region_name=region)
    
    response = client.get_cost_and_usage(
        TimePeriod={
            'Start': start_date,
            'End': end_date
        },
        Granularity='DAILY',
        Metrics=['UnblendedCost'],
        GroupBy=[
            {'Type': 'DIMENSION', 'Key': 'SERVICE'},
            {'Type': 'DIMENSION', 'Key': 'USAGE_TYPE'}
        ]
    )
    
    return response['ResultsByTime']

这些数据需要定期收集，我设置的是每天收集一次，存储到本地 SQLite 数据库。

可视化看板

有了数据后，我用 Python 的 matplotlib 画了一些趋势图和对比图：

import matplotlib.pyplot as plt
import matplotlib.font_manager as fm
from datetime import datetime

# 设置中文字体
plt.rcParams['font.sans-serif'] = ['SimHei', 'DejaVu Sans']
plt.rcParams['axes.unicode_minus'] = False

def plot_daily_costs(dates, costs, output_path):
    """绘制每日成本趋势图"""
    plt.figure(figsize=(12, 6))
    plt.plot(dates, costs, marker='o', linewidth=2, markersize=4)
    plt.title('AI 服务每日成本趋势', fontsize=14, fontweight='bold')
    plt.xlabel('日期', fontsize=12)
    plt.ylabel('成本 (美元)', fontsize=12)
    plt.grid(True, alpha=0.3)
    plt.xticks(rotation=45)
    plt.tight_layout()
    plt.savefig(output_path, dpi=300, bbox_inches='tight')
    plt.close()

这个看板帮我快速发现异常波动，比如某天成本突然上升时可以立即定位问题。

具体优化措施

有了监控数据后，就可以开始实际优化了。这里记录几个有效的措施。

API 层优化

第一个发现是 API 调用效率问题。我们有很多重复的请求，缓存能显著降低成本。

graph LR A[用户请求] --> B{缓存检查} B -->|命中| C[返回缓存结果] B -->|未命中| D[调用 API] D --> E[更新缓存] E --> F[返回结果] style B fill:#f9f,stroke:#333,stroke-width:2px style C fill:#9f9,stroke:#333,stroke-width:2px style D fill:#f99,stroke:#333,stroke-width:2px

我实现了一个简单的缓存层，主要规则：

  • 相同 prompt 的请求缓存 30 分钟
  • 对参数化任务，使用参数哈希作为缓存键
  • 对实时性要求高的任务，缩短缓存时间或直接跳过

代码示例：

import hashlib
import json
from functools import wraps
from datetime import datetime, timedelta
import pickle
import os

def cache_api_calls(cache_dir="api_cache", cache_minutes=30):
    """API 调用缓存装饰器"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # 生成缓存键
            cache_key = hashlib.md5(
                json.dumps({"args": args, "kwargs": kwargs}, sort_keys=True).encode()
            ).hexdigest()
            
            cache_file = os.path.join(cache_dir, f"{cache_key}.pkl")
            
            # 检查缓存是否存在且未过期
            if os.path.exists(cache_file):
                with open(cache_file, 'rb') as f:
                    cached_data = pickle.load(f)
                    if datetime.now() - cached_data['timestamp'] < timedelta(minutes=cache_minutes):
                        return cached_data['result']
            
            # 执行原函数
            result = func(*args, **kwargs)
            
            # 保存到缓存
            os.makedirs(cache_dir, exist_ok=True)
            with open(cache_file, 'wb') as f:
                pickle.dump({
                    'result': result,
                    'timestamp': datetime.now()
                }, f)
            
            return result
        return wrapper
    return decorator

@cache_api_calls(cache_minutes=30)
def call_openai_api(prompt, model="gpt-4o"):
    """调用 OpenAI API 并缓存结果"""
    # 实际 API 调用代码
    pass

这个简单的缓存层帮我们节省了约 25% 的 API 调用成本。

模型选择优化

另一个发现是模型使用不够精细。不是所有任务都需要用最贵的模型。

我做了一个简单的成本-性能对比测试：

def compare_models_cost_performance(prompt, models=["gpt-4o-mini", "gpt-4o", "gpt-4"]):
    """比较不同模型的成本和性能"""
    results = []
    
    for model in models:
        start_time = time.time()
        
        # 调用 API
        response = openai.ChatCompletion.create(
            model=model,
            messages=[{"role": "user", "content": prompt}]
        )
        
        # 计算成本
        input_tokens = response.usage.prompt_tokens
        output_tokens = response.usage.completion_tokens
        cost = calculate_cost(input_tokens, output_tokens, model)
        
        results.append({
            'model': model,
            'time': time.time() - start_time,
            'input_tokens': input_tokens,
            'output_tokens': output_tokens,
            'cost': cost,
            'quality': evaluate_quality(response)  # 需要自定义评估函数
        })
    
    return results

基于测试结果，我们制定了使用规则：

  • 简单问答和文档分析：用 gpt-4o-mini
  • 需要推理能力的任务：用 gpt-4o
  • 复杂代码生成和多步推理：用 gpt-4 或 Claude Opus

这样既保证了质量，又控制了成本。

GPU 资源调度

GPU 实例的优化重点是减少空闲时间。

import time
import boto3

class GPUManager:
    """GPU 实例管理器"""
    
    def __init__(self, instance_type, region):
        self.instance_type = instance_type
        self.region = region
        self.ec2 = boto3.client('ec2', region_name=region)
        self.instance_id = None
        self.last_activity = None
    
    def start_instance(self):
        """启动 GPU 实例"""
        if self.instance_id is None:
            # 启动新实例
            response = self.ec2.run_instances(
                ImageId='ami-12345678',  # 你的 AMI ID
                InstanceType=self.instance_type,
                MinCount=1,
                MaxCount=1
            )
            self.instance_id = response['Instances'][0]['InstanceId']
            self.wait_until_ready()
        self.last_activity = time.time()
    
    def stop_if_idle(self, idle_minutes=30):
        """如果实例空闲超过指定时间则停止"""
        if self.instance_id and self.last_activity:
            idle_time = (time.time() - self.last_activity) / 60  # 转换为分钟
            if idle_time >= idle_minutes:
                print(f"实例空闲 {idle_time:.1f} 分钟，正在停止...")
                self.ec2.stop_instances(InstanceIds=[self.instance_id])
                self.instance_id = None
    
    def wait_until_ready(self):
        """等待实例就绪"""
        while True:
            response = self.ec2.describe_instance_status(
                InstanceIds=[self.instance_id]
            )
            status = response['InstanceStatuses'][0]['InstanceStatus']['Status']
            if status == 'ok':
                break
            time.sleep(10)

这个管理器帮我们减少了很多不必要的 GPU 空闲时间。

踩坑记录

优化过程中也踩了不少坑，这里记录几个典型的。

缓存击穿问题

早期缓存实现有个问题：当大量请求同时到达时，缓存失效会导致同时调用 API，造成"缓存击穿"。

解决方案是加锁和预热：

from threading import Lock
import time

class CacheLock:
    def __init__(self):
        self.locks = {}
        self.global_lock = Lock()
    
    def get_lock(self, key):
        """获取指定键的锁"""
        with self.global_lock:
            if key not in self.locks:
                self.locks[key] = Lock()
            return self.locks[key]

cache_locks = CacheLock()

def cache_with_lock(cache_dir="api_cache", cache_minutes=30):
    """带锁的缓存装饰器"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            cache_key = hashlib.md5(
                json.dumps({"args": args, "kwargs": kwargs}, sort_keys=True).encode()
            ).hexdigest()
            
            lock = cache_locks.get_lock(cache_key)
            
            with lock:
                # 检查缓存
                cache_file = os.path.join(cache_dir, f"{cache_key}.pkl")
                
                if os.path.exists(cache_file):
                    with open(cache_file, 'rb') as f:
                        cached_data = pickle.load(f)
                        if datetime.now() - cached_data['timestamp'] < timedelta(minutes=cache_minutes):
                            return cached_data['result']
                
                # 执行原函数
                result = func(*args, **kwargs)
                
                # 保存缓存
                os.makedirs(cache_dir, exist_ok=True)
                with open(cache_file, 'wb') as f:
                    pickle.dump({
                        'result': result,
                        'timestamp': datetime.now()
                    }, f)
                
                return result
        return wrapper
    return decorator

Spot 实例被回收

一开始大量使用 Spot 实例降成本，但被频繁回收导致任务中断。

后来我们采用了混合策略：

  • 非关键任务用 Spot 实例
  • 关键任务用按需实例
  • 实现了实例回收检测和自动重新调度
def is_spot_instance_termination_warning():
    """检查是否收到 Spot 实例回收警告"""
    import urllib.request
    try:
        response = urllib.request.urlopen('http://169.254.169.254/latest/meta-data/spot/termination-time')
        return True
    except urllib.error.HTTPError as e:
        if e.code == 404:
            return False
        raise

def graceful_shutdown():
    """优雅关闭，保存状态"""
    print("收到 Spot 实例回收警告，开始优雅关闭...")
    # 保存当前任务状态
    # 上传中间结果
    # 通知监控系统

成本预测偏差

早期成本预测基于简单线性外推，实际效果很差。

后来改用更复杂的预测模型：

from sklearn.ensemble import RandomForestRegressor
from sklearn.preprocessing import StandardScaler
import numpy as np

class CostPredictor:
    """成本预测器"""
    
    def __init__(self):
        self.model = RandomForestRegressor(n_estimators=100)
        self.scaler = StandardScaler()
    
    def prepare_features(self, usage_data):
        """准备特征数据"""
        features = []
        for day_data in usage_data:
            # 提取当日特征
            daily_tokens = day_data.get('total_tokens', 0)
            active_users = day_data.get('active_users', 0)
            task_complexity = day_data.get('avg_task_complexity', 0)
            weekend = day_data.get('is_weekend', 0)
            
            # 添加移动平均特征
            features.append([
                daily_tokens, active_users, task_complexity, weekend,
                # 添加更多特征...
            ])
        
        return np.array(features)
    
    def train(self, historical_data):
        """训练预测模型"""
        X = self.prepare_features(historical_data[:-1])
        y = np.array([d['cost'] for d in historical_data[1:]])
        
        X_scaled = self.scaler.fit_transform(X)
        self.model.fit(X_scaled, y)
    
    def predict_next_day(self, current_data):
        """预测下一天成本"""
        features = self.prepare_features([current_data])
        features_scaled = self.scaler.transform(features)
        predicted_cost = self.model.predict(features_scaled)[0]
        return predicted_cost

这个预测器帮我们做更好的成本规划。

优化结果

经过几个月的优化，效果还是比较明显的：

优化措施成本降低实施难度备注
API 缓存~25%低效果立竿见影
模型选择优化~15%中需要测试不同场景
GPU 资源调度~20%高需要重构现有代码
Spot 实例混合~10%中有任务中断风险
成本监控~5%低主要避免意外增长

总体来说，成本降低了约 40%，而且现在有了清晰的监控体系，不会出现账单意外了。

几个实用建议

总结一下这次折腾的几个实用建议：

  1. 先把账单看懂：不同服务的计费模型差异很大，搞清楚规则再优化
  2. 建立监控体系：不知道成本分布就没法优化，监控是第一步
  3. 从低成本优化开始：缓存、模型选择这些措施成本低、效果好
  4. 不要过度优化：节省的成本不要超过优化本身的工作量
  5. 保持记录：记录每次优化的效果，方便后续调整策略

收尾

成本优化这件事，本质上是在质量和成本之间找平衡。

最理想的状态不是成本最低，而是在可接受的成本范围内提供足够的服务质量。这次优化也让我意识到：很多时候问题不在于"太贵"，而在于"不够透明"。账单不会撒谎，但如果不让账单说话，就永远不知道钱花在什么地方。

现在每次看账单，至少能说清楚每一块钱花在什么地方，大概就够了。

版权声明: 本文首发于 指尖魔法屋-关于AI成本优化的几点记录（https://blog.thinkmoon.cn/post/267-ai-cost-optimization-billing-savings-practice/） 转载或引用必须申明原指尖魔法屋来源及源地址！

相关阅读

根据标签、分类和主题为你挑选的相近文章

Harness Engineering 的介绍与编程实践
这篇文章从 AI 编程的真实工程场景出发，梳理 Harness Engineering 如何通过上下文、工具、沙盒、验证和人工闸门，让 Agent 从“会写代码”走向“能稳定交付”。
📁 技术实践
📅 2026-04-23