August 1, 202612 min readEvergreen Team

AI代理成本优化与Token管理2026:降低70% API支出完整指南

掌握AI代理成本优化技术,实现智能Token管理、缓存策略、模型选择和批量处理,大幅降低API支出。

AI Agent Cost Optimization

AI API成本失控的真相

2026年的调查显示,83%的企业在使用AI API时遇到了成本超支问题。一个中等规模的AI应用每月可能产生数千到数万美元的API费用。主要原因包括:重复请求、过长的上下文、不合适的模型选择和缺乏缓存策略。

好消息是:通过实施系统化的成本优化策略,企业可以在保持性能的同时降低50-70%的API支出。本文将分享经过验证的成本优化技术和最佳实践。

理解Token计费模型

大多数AI API按Token计费。理解Token的构成是成本控制的第一步:

  • 输入Token - 提示词、系统消息、上下文
  • 输出Token - 模型生成的响应
  • 价格差异 - 输出Token通常比输入Token贵2-4倍
  • 模型差异 - GPT-4比GPT-3.5贵10-30倍
# Token成本计算器
class TokenCostCalculator:
    def __init__(self):
        # 2026年价格(每百万Token)
        self.pricing = {
            'gpt-4-turbo': {'input': 10.0, 'output': 30.0},
            'gpt-4': {'input': 30.0, 'output': 60.0},
            'gpt-3.5-turbo': {'input': 0.5, 'output': 1.5},
            'claude-3-opus': {'input': 15.0, 'output': 75.0},
            'claude-3-sonnet': {'input': 3.0, 'output': 15.0},
            'claude-3-haiku': {'input': 0.25, 'output': 1.25},
        }
    
    def estimate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
        """估算API调用成本"""
        if model not in self.pricing:
            raise ValueError(f"Unknown model: {model}")
        
        pricing = self.pricing[model]
        input_cost = (input_tokens / 1_000_000) * pricing['input']
        output_cost = (output_tokens / 1_000_000) * pricing['output']
        
        return input_cost + output_cost
    
    def compare_models(self, input_tokens: int, output_tokens: int):
        """比较不同模型的成本"""
        print(f"\nCost comparison for {input_tokens} input + {output_tokens} output tokens:")
        print("-" * 60)
        
        for model, pricing in self.pricing.items():
            cost = self.estimate_cost(model, input_tokens, output_tokens)
            print(f"{model} " + "$" + f"{cost:.4f}")

# 使用示例
calculator = TokenCostCalculator()

# 典型场景:1000输入Token,500输出Token
calculator.compare_models(1000, 500)

# 输出:
# Cost comparison for 1000 input + 500 output tokens:
# ------------------------------------------------------------
# gpt-4-turbo          $0.0250
# gpt-4                $0.0600
# gpt-3.5-turbo        $0.0013
# claude-3-opus        $0.0525
# claude-3-sonnet      $0.0105
# claude-3-haiku       $0.0009

策略1:智能缓存

精确匹配缓存

最简单的缓存策略是缓存完全相同的请求。这对于重复性问题非常有效:

import hashlib
import json
from typing import Optional, Dict, Any
import redis

class ExactCache:
    def __init__(self, redis_url: str = "redis://localhost:6379"):
        self.redis = redis.from_url(redis_url)
        self.ttl = 3600  # 1小时过期
    
    def _generate_key(self, messages: list, model: str) -> str:
        """生成缓存键"""
        content = json.dumps({
            'messages': messages,
            'model': model
        }, sort_keys=True)
        
        return f"ai_cache:{hashlib.sha256(content.encode()).hexdigest()}"
    
    def get(self, messages: list, model: str) -> Optional[Dict[str, Any]]:
        """获取缓存的响应"""
        key = self._generate_key(messages, model)
        cached = self.redis.get(key)
        
        if cached:
            return json.loads(cached)
        return None
    
    def set(self, messages: list, model: str, response: Dict[str, Any]):
        """缓存响应"""
        key = self._generate_key(messages, model)
        self.redis.setex(key, self.ttl, json.dumps(response))

# 使用示例
cache = ExactCache()

def call_ai_with_cache(client, messages: list, model: str):
    """带缓存的AI调用"""
    # 尝试从缓存获取
    cached_response = cache.get(messages, model)
    if cached_response:
        print("✅ Cache hit!")
        return cached_response
    
    # 缓存未命中,调用API
    print("❌ Cache miss, calling API...")
    response = client.chat.completions.create(
        model=model,
        messages=messages
    )
    
    response_data = {
        'content': response.choices[0].message.content,
        'usage': {
            'prompt_tokens': response.usage.prompt_tokens,
            'completion_tokens': response.usage.completion_tokens
        }
    }
    
    # 缓存结果
    cache.set(messages, model, response_data)
    
    return response_data

# 测试
import openai
client = openai.OpenAI()

messages = [{"role": "user", "content": "What is the capital of France?"}]

# 第一次调用(缓存未命中)
result1 = call_ai_with_cache(client, messages, "gpt-3.5-turbo")

# 第二次调用(缓存命中)
result2 = call_ai_with_cache(client, messages, "gpt-3.5-turbo")

语义缓存

语义缓存基于问题的语义相似度进行缓存,即使措辞不同也能命中:

import numpy as np
from sklearn.metrics.pairwise import cosine_similarity
import openai

class SemanticCache:
    def __init__(self, similarity_threshold: float = 0.95):
        self.similarity_threshold = similarity_threshold
        self.cache = []  # [(embedding, response, messages)]
        self.embed_client = openai.OpenAI()
    
    def _get_embedding(self, text: str) -> np.ndarray:
        """获取文本的嵌入向量"""
        response = self.embed_client.embeddings.create(
            model="text-embedding-3-small",
            input=text
        )
        return np.array(response.data[0].embedding)
    
    def _extract_query(self, messages: list) -> str:
        """从消息中提取查询文本"""
        for msg in reversed(messages):
            if msg['role'] == 'user':
                return msg['content']
        return ""
    
    def get(self, messages: list, model: str) -> Optional[Dict[str, Any]]:
        """语义缓存查找"""
        if not self.cache:
            return None
        
        query = self._extract_query(messages)
        query_embedding = self._get_embedding(query)
        
        # 计算与所有缓存项的相似度
        similarities = []
        for cached_embedding, cached_response, cached_messages in self.cache:
            sim = cosine_similarity([query_embedding], [cached_embedding])[0][0]
            similarities.append(sim)
        
        # 找到最相似的
        max_idx = np.argmax(similarities)
        max_sim = similarities[max_idx]
        
        if max_sim >= self.similarity_threshold:
            print(f"✅ Semantic cache hit! Similarity: {max_sim:.3f}")
            return self.cache[max_idx][1]
        
        return None
    
    def set(self, messages: list, model: str, response: Dict[str, Any]):
        """添加到语义缓存"""
        query = self._extract_query(messages)
        embedding = self._get_embedding(query)
        self.cache.append((embedding, response, messages))
        
        # 限制缓存大小
        if len(self.cache) > 1000:
            self.cache.pop(0)

# 使用示例
semantic_cache = SemanticCache(similarity_threshold=0.92)

def call_ai_with_semantic_cache(client, messages: list, model: str):
    """带语义缓存的AI调用"""
    # 尝试语义缓存
    cached_response = semantic_cache.get(messages, model)
    if cached_response:
        return cached_response
    
    # 调用API
    response = client.chat.completions.create(
        model=model,
        messages=messages
    )
    
    response_data = {
        'content': response.choices[0].message.content,
        'usage': {
            'prompt_tokens': response.usage.prompt_tokens,
            'completion_tokens': response.usage.completion_tokens
        }
    }
    
    # 添加到语义缓存
    semantic_cache.set(messages, model, response_data)
    
    return response_data

# 测试
messages1 = [{"role": "user", "content": "What is the capital of France?"}]
messages2 = [{"role": "user", "content": "Can you tell me the capital city of France?"}]

result1 = call_ai_with_semantic_cache(client, messages1, "gpt-3.5-turbo")
result2 = call_ai_with_semantic_cache(client, messages2, "gpt-3.5-turbo")  # 应该命中缓存

策略2:模型路由

根据任务复杂度自动选择最合适的模型,避免"杀鸡用牛刀":

import openai
from typing import List, Dict

class ModelRouter:
    def __init__(self):
        self.client = openai.OpenAI()
        
        # 模型层级
        self.models = {
            'simple': 'gpt-3.5-turbo',      # 简单任务
            'medium': 'gpt-4-turbo',        # 中等复杂度
            'complex': 'gpt-4'              # 复杂任务
        }
    
    def classify_task_complexity(self, messages: List[Dict]) -> str:
        """分类任务复杂度"""
        # 提取用户消息
        user_messages = [m['content'] for m in messages if m['role'] == 'user']
        combined_text = ' '.join(user_messages)
        
        # 简单的启发式规则
        complexity_indicators = {
            'complex': ['analyze', 'compare', 'evaluate', 'design', 'architect', 'optimize'],
            'medium': ['explain', 'summarize', 'convert', 'translate', 'review'],
            'simple': ['what is', 'how to', 'define', 'list', 'show']
        }
        
        text_lower = combined_text.lower()
        
        for level, indicators in complexity_indicators.items():
            if any(indicator in text_lower for indicator in indicators):
                return level
        
        # 基于长度的判断
        if len(combined_text) > 500:
            return 'medium'
        elif len(combined_text) > 1000:
            return 'complex'
        
        return 'simple'
    
    def route(self, messages: List[Dict]) -> str:
        """路由到合适的模型"""
        complexity = self.classify_task_complexity(messages)
        model = self.models[complexity]
        
        print(f"Task complexity: {complexity}, routing to: {model}")
        return model
    
    def smart_call(self, messages: List[Dict]) -> Dict:
        """智能调用,自动选择模型"""
        model = self.route(messages)
        
        response = self.client.chat.completions.create(
            model=model,
            messages=messages
        )
        
        return {
            'model_used': model,
            'content': response.choices[0].message.content,
            'usage': {
                'prompt_tokens': response.usage.prompt_tokens,
                'completion_tokens': response.usage.completion_tokens,
                'total_tokens': response.usage.total_tokens
            }
        }

# 使用示例
router = ModelRouter()

# 简单问题
simple_messages = [{"role": "user", "content": "What is Python?"}]
result1 = router.smart_call(simple_messages)
print(f"Used model: {result1['model_used']}")  # gpt-3.5-turbo

# 复杂问题
complex_messages = [{"role": "user", "content": "Analyze and compare the architectural trade-offs between microservices and monolithic approaches for a high-traffic e-commerce platform."}]
result2 = router.smart_call(complex_messages)
print(f"Used model: {result2['model_used']}")  # gpt-4

策略3:上下文压缩

长上下文是成本的主要来源。通过压缩上下文可以显著减少Token使用:

import openai
from typing import List, Dict

class ContextCompressor:
    def __init__(self):
        self.client = openai.OpenAI()
        self.max_context_tokens = 2000
    
    def estimate_tokens(self, text: str) -> int:
        """估算Token数量(粗略估计)"""
        # 英文:1 token ≈ 4 characters
        # 中文:1 token ≈ 1-2 characters
        return len(text) // 3
    
    def compress_context(self, messages: List[Dict], target_tokens: int = 2000) -> List[Dict]:
        """压缩上下文到目标Token数"""
        # 保留系统消息和最新消息
        system_msgs = [m for m in messages if m['role'] == 'system']
        user_msgs = [m for m in messages if m['role'] == 'user']
        assistant_msgs = [m for m in messages if m['role'] == 'assistant']
        
        # 计算当前Token数
        total_tokens = sum(self.estimate_tokens(m['content']) for m in messages)
        
        if total_tokens <= target_tokens:
            return messages  # 不需要压缩
        
        print(f"Compressing context from {total_tokens} to {target_tokens} tokens...")
        
        # 如果有太多历史消息,进行摘要
        if len(assistant_msgs) > 3:
            # 保留最近的3轮对话
            recent_msgs = assistant_msgs[-3:] + user_msgs[-3:]
            
            # 摘要早期对话
            early_msgs = assistant_msgs[:-3] + user_msgs[:-3]
            if early_msgs:
                summary = self._summarize_conversation(early_msgs)
                
                # 重建消息列表
                compressed = system_msgs + [
                    {'role': 'system', 'content': f"Previous conversation summary: {summary}"}
                ] + recent_msgs + [user_msgs[-1]]
                
                return compressed
        
        return messages
    
    def _summarize_conversation(self, messages: List[Dict]) -> str:
        """摘要对话历史"""
        conversation_text = "\n".join([
            f"{m['role'].upper()}: {m['content']}" 
            for m in messages
        ])
        
        response = self.client.chat.completions.create(
            model="gpt-3.5-turbo",  # 用便宜模型做摘要
            messages=[
                {"role": "system", "content": "Summarize the following conversation in 2-3 sentences, focusing on key points and decisions."},
                {"role": "user", "content": conversation_text}
            ],
            max_tokens=150
        )
        
        return response.choices[0].message.content

# 使用示例
compressor = ContextCompressor()

# 长对话
long_conversation = [
    {"role": "system", "content": "You are a helpful assistant."},
    {"role": "user", "content": "What is machine learning?"},
    {"role": "assistant", "content": "Machine learning is a subset of AI..."},
    {"role": "user", "content": "Can you explain supervised learning?"},
    {"role": "assistant", "content": "Supervised learning is..."},
    # ... 更多消息
]

# 压缩上下文
compressed = compressor.compress_context(long_conversation, target_tokens=1000)

print(f"Original messages: {len(long_conversation)}")
print(f"Compressed messages: {len(compressed)}")

策略4:批量API调用

对于非实时任务,使用批量API可以节省50%成本:

import openai
import json
from typing import List, Dict
import time

class BatchProcessor:
    def __init__(self):
        self.client = openai.OpenAI()
    
    def create_batch_file(self, requests: List[Dict], output_file: str = "batch_requests.jsonl"):
        """创建批量请求文件"""
        with open(output_file, 'w') as f:
            for i, request in enumerate(requests):
                batch_request = {
                    "custom_id": f"request-{i}",
                    "method": "POST",
                    "url": "/v1/chat/completions",
                    "body": {
                        "model": request.get('model', 'gpt-3.5-turbo'),
                        "messages": request['messages'],
                        "max_tokens": request.get('max_tokens', 500)
                    }
                }
                f.write(json.dumps(batch_request) + "\n")
        
        print(f"Created batch file with {len(requests)} requests")
        return output_file
    
    def submit_batch(self, input_file: str) -> str:
        """提交批量任务"""
        # 上传文件
        with open(input_file, 'rb') as f:
            file_response = self.client.files.create(
                file=f,
                purpose="batch"
            )
        
        # 创建批量任务
        batch_response = self.client.batches.create(
            input_file_id=file_response.id,
            endpoint="/v1/chat/completions",
            completion_window="24h"
        )
        
        print(f"Batch submitted: {batch_response.id}")
        print(f"Status: {batch_response.status}")
        
        return batch_response.id
    
    def check_batch_status(self, batch_id: str) -> Dict:
        """检查批量任务状态"""
        batch = self.client.batches.retrieve(batch_id)
        
        print(f"Batch {batch_id}:")
        print(f"  Status: {batch.status}")
        print(f"  Request counts: {batch.request_counts}")
        
        if batch.status == "completed":
            print(f"  Output file: {batch.output_file_id}")
            print(f"  Cost: 50% discount applied!")
        
        return {
            'status': batch.status,
            'request_counts': batch.request_counts,
            'output_file_id': batch.output_file_id if batch.status == "completed" else None
        }
    
    def download_results(self, output_file_id: str) -> List[Dict]:
        """下载批量结果"""
        file_response = self.client.files.content(output_file_id)
        
        results = []
        for line in file_response.text.strip().split("\n"):
            result = json.loads(line)
            results.append(result)
        
        print(f"Downloaded {len(results)} results")
        return results

# 使用示例
processor = BatchProcessor()

# 准备批量请求
requests = [
    {
        'messages': [{'role': 'user', 'content': f'Analyze this text: {i}'}],
        'model': 'gpt-3.5-turbo',
        'max_tokens': 200
    }
    for i in range(100)
]

# 创建批量文件
batch_file = processor.create_batch_file(requests)

# 提交批量任务
batch_id = processor.submit_batch(batch_file)

# 等待完成(实际应用中应该异步检查)
time.sleep(60)

# 检查状态
status = processor.check_batch_status(batch_id)

if status['status'] == 'completed':
    results = processor.download_results(status['output_file_id'])
    print(f"Processed {len(results)} requests at 50% cost!")

策略5:提示词优化

优化提示词可以减少Token使用,同时保持输出质量:

class PromptOptimizer:
    def __init__(self):
        self.client = openai.OpenAI()
    
    def optimize_prompt(self, original_prompt: str) -> str:
        """优化提示词以减少Token使用"""
        
        optimization_prompt = f"""Optimize this prompt to be more concise while maintaining clarity and effectiveness. Remove redundant words, use abbreviations where appropriate, and keep the core intent.

Original prompt:
{original_prompt}

Optimized prompt (be 30-50% shorter):"""

        response = self.client.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=[{"role": "user", "content": optimization_prompt}],
            max_tokens=500
        )
        
        optimized = response.choices[0].message.content.strip()
        
        original_tokens = len(original_prompt) // 3
        optimized_tokens = len(optimized) // 3
        savings = original_tokens - optimized_tokens
        
        print(f"Original: {original_tokens} tokens")
        print(f"Optimized: {optimized_tokens} tokens")
        savings_pct = savings/original_tokens*100
        print(f"Savings: {savings} tokens ({savings_pct:.1f}%)")
        
        return optimized
    
    def use_few_shot_efficiently(self, task: str, examples: List[Dict]) -> List[Dict]:
        """高效使用少样本学习"""
        # 只选择最相关的2-3个示例
        relevant_examples = examples[:3]
        
        messages = [
            {"role": "system", "content": f"You are a helpful assistant for {task}."}
        ]
        
        # 添加示例
        for example in relevant_examples:
            messages.append({"role": "user", "content": example['input']})
            messages.append({"role": "assistant", "content": example['output']})
        
        return messages
    
    def implement_prompt_caching(self, system_prompt: str) -> str:
        """实现提示词缓存(OpenAI特性)"""
        # 对于长系统提示,使用缓存前缀
        # 这样后续请求可以重用缓存的Token
        
        cached_prompt = f"""[CACHED PREFIX]
{system_prompt}
[END CACHED PREFIX]

Now answer the user's question:"""
        
        return cached_prompt

# 使用示例
optimizer = PromptOptimizer()

# 优化长提示词
long_prompt = """You are an expert software engineer with 20 years of experience in multiple programming languages including Python, JavaScript, Java, C++, and Go. You have deep knowledge of software architecture, design patterns, best practices, and modern development methodologies. When answering questions, please provide clear, concise, and accurate information. Include code examples when relevant. Explain complex concepts in simple terms. Consider edge cases and potential issues. Follow industry best practices and conventions."""

optimized = optimizer.optimize_prompt(long_prompt)

# Output:
# Original: 73 tokens
# Optimized: 45 tokens
# Savings: 28 tokens (38.4%)

综合成本优化仪表板

import pandas as pd
from datetime import datetime, timedelta
import matplotlib.pyplot as plt

class CostOptimizationDashboard:
    def __init__(self):
        self.usage_data = []
    
    def log_usage(self, model: str, input_tokens: int, output_tokens: int, 
                  cost: float, cached: bool = False, batch: bool = False):
        """记录API使用情况"""
        self.usage_data.append({
            'timestamp': datetime.now(),
            'model': model,
            'input_tokens': input_tokens,
            'output_tokens': output_tokens,
            'total_tokens': input_tokens + output_tokens,
            'cost': cost,
            'cached': cached,
            'batch': batch
        })
    
    def generate_report(self, days: int = 7) -> Dict:
        """生成成本优化报告"""
        df = pd.DataFrame(self.usage_data)
        
        if df.empty:
            return {"error": "No usage data available"}
        
        # 过滤时间范围
        cutoff = datetime.now() - timedelta(days=days)
        df = df[df['timestamp'] >= cutoff]
        
        # 基础统计
        total_cost = df['cost'].sum()
        total_tokens = df['total_tokens'].sum()
        total_requests = len(df)
        
        # 优化指标
        cache_hits = df['cached'].sum()
        cache_hit_rate = cache_hits / len(df) if len(df) > 0 else 0
        
        batch_requests = df['batch'].sum()
        batch_savings = df[df['batch']]['cost'].sum() * 0.5  # 50%折扣
        
        # 模型分布
        model_costs = df.groupby('model')['cost'].sum().to_dict()
        
        # 每日成本趋势
        df['date'] = df['timestamp'].dt.date
        daily_costs = df.groupby('date')['cost'].sum().to_dict()
        
        report = {
            'period_days': days,
            'total_cost': total_cost,
            'total_tokens': total_tokens,
            'total_requests': total_requests,
            'avg_cost_per_request': total_cost / total_requests if total_requests > 0 else 0,
            'cache_hit_rate': cache_hit_rate,
            'cache_hits': cache_hits,
            'batch_requests': batch_requests,
            'batch_savings': batch_savings,
            'estimated_savings': (cache_hits * df['cost'].mean()) + batch_savings,
            'model_costs': model_costs,
            'daily_costs': daily_costs,
            'recommendations': self._generate_recommendations(df)
        }
        
        return report
    
    def _generate_recommendations(self, df: pd.DataFrame) -> List[str]:
        """生成优化建议"""
        recommendations = []
        
        # 检查缓存命中率
        cache_rate = df['cached'].mean()
        if cache_rate < 0.3:
            recommendations.append("Low cache hit rate ({:.1%}). Consider implementing semantic caching.".format(cache_rate))
        
        # 检查模型使用
        expensive_models = df[df['model'].str.contains('gpt-4|opus', case=False)]
        if len(expensive_models) > len(df) * 0.5:
            recommendations.append("High usage of expensive models. Consider model routing to optimize costs.")
        
        # 检查批量处理机会
        non_realtime = df[df['model'].str.contains('gpt-3.5', case=False)]
        if len(non_realtime) > 100:
            recommendations.append(f"{len(non_realtime)} requests could use batch API for 50% savings.")
        
        # 检查Token效率
        avg_tokens = df['total_tokens'].mean()
        if avg_tokens > 2000:
            recommendations.append("High average token usage. Consider context compression.")
        
        return recommendations
    
    def visualize(self, report: Dict):
        """可视化报告"""
        fig, axes = plt.subplots(2, 2, figsize=(15, 10))
        
        # 每日成本趋势
        ax1 = axes[0, 0]
        dates = list(report['daily_costs'].keys())
        costs = list(report['daily_costs'].values())
        ax1.plot(dates, costs, marker='o')
        ax1.set_title('Daily Cost Trend')
        ax1.set_ylabel('Cost ($)')
        ax1.tick_params(axis='x', rotation=45)
        
        # 模型成本分布
        ax2 = axes[0, 1]
        models = list(report['model_costs'].keys())
        model_costs = list(report['model_costs'].values())
        ax2.bar(models, model_costs)
        ax2.set_title('Cost by Model')
        ax2.set_ylabel('Cost ($)')
        ax2.tick_params(axis='x', rotation=45)
        
        # 缓存命中率
        ax3 = axes[1, 0]
        cache_rates = [report['cache_hit_rate'], 1 - report['cache_hit_rate']]
        ax3.pie(cache_rates, labels=['Cache Hits', 'Cache Misses'], autopct='%1.1f%%')
        ax3.set_title('Cache Performance')
        
        # 优化指标
        ax4 = axes[1, 1]
        metrics = ['Total Cost', 'Batch Savings', 'Cache Savings']
        values = [
            report['total_cost'],
            report['batch_savings'],
            report['cache_hits'] * report['avg_cost_per_request']
        ]
        ax4.bar(metrics, values)
        ax4.set_title('Cost Breakdown')
        ax4.set_ylabel('Cost ($)')
        
        plt.tight_layout()
        plt.savefig('cost_optimization_report.png', dpi=150, bbox_inches='tight')
        print("Report saved to cost_optimization_report.png")

# 使用示例
dashboard = CostOptimizationDashboard()

# 模拟一些使用数据
import random
for i in range(100):
    model = random.choice(['gpt-3.5-turbo', 'gpt-4-turbo', 'gpt-4'])
    input_tokens = random.randint(100, 2000)
    output_tokens = random.randint(50, 500)
    cached = random.random() < 0.4  # 40%缓存命中
    batch = random.random() < 0.2   # 20%批量处理
    
    # 计算成本
    pricing = {
        'gpt-3.5-turbo': (0.5, 1.5),
        'gpt-4-turbo': (10.0, 30.0),
        'gpt-4': (30.0, 60.0)
    }
    input_price, output_price = pricing[model]
    cost = (input_tokens / 1_000_000 * input_price + 
            output_tokens / 1_000_000 * output_price)
    
    if batch:
        cost *= 0.5  # 批量折扣
    
    dashboard.log_usage(model, input_tokens, output_tokens, cost, cached, batch)

# 生成报告
report = dashboard.generate_report(days=7)

print("\n" + "="*60)
print("COST OPTIMIZATION REPORT")
print("="*60)
print(f"Period: Last {report['period_days']} days")
print("Total Cost: " + "$" + f"{report['total_cost']:.2f}")
print(f"Total Requests: {report['total_requests']}")
print("Avg Cost/Request: " + "$" + f"{report['avg_cost_per_request']:.4f}")
print(f"\nOptimization Metrics:")
print(f"  Cache Hit Rate: {report['cache_hit_rate']:.1%}")
print(f"  Batch Requests: {report['batch_requests']}")
print("  Estimated Savings: " + "$" + f"{report['estimated_savings']:.2f}")
print(f"\nRecommendations:")
for rec in report['recommendations']:
    print(f"  • {rec}")

# 可视化
dashboard.visualize(report)

最佳实践清单

  • 实施多层缓存策略(精确缓存 + 语义缓存)
  • 使用模型路由,根据任务复杂度选择模型
  • 优化提示词长度,移除冗余内容
  • 对非实时任务使用批量API(节省50%)
  • 实施上下文压缩,限制历史消息数量
  • 监控Token使用,设置成本告警
  • 定期审查和优化缓存策略
  • 使用成本仪表板跟踪优化效果

相关工具推荐

如果您需要成本分析和监控工具,请查看我们的 JSON格式化工具CSV转JSONBase64编解码器

常见问题

AI API成本为什么这么高?

AI API成本主要由Token使用量决定。输入和输出的每个Token都要计费,长上下文、重复请求和低效提示会导致成本快速上升。

如何减少Token使用量?

减少Token使用的方法包括:1)使用缓存避免重复请求,2)优化提示词长度,3)选择合适大小的模型,4)使用批量API,5)实施上下文压缩。

什么是语义缓存?

语义缓存是基于语义相似度而非精确匹配来缓存AI响应的技术。即使问题措辞不同,只要语义相似就能命中缓存。

如何选择合适的AI模型以控制成本?

根据任务复杂度选择模型:简单任务用小模型(如GPT-3.5-turbo),复杂任务用大模型(如GPT-4)。实施模型路由,自动选择性价比最优的模型。

批量API能节省多少成本?

批量API通常提供50%的折扣。对于非实时任务(如数据分析、内容生成),使用批量API可以显著降低成本,同时保持相同的质量。