Agent 项目成本优化实战：从 Token 到基础设施的全面降本

优化前月成本分布：
├── LLM API 调用     72%  ($2,304)
│   ├── 主模型 (GPT-4o)    58%
│   ├── 嵌入模型           8%
│   └── 审核模型           6%
├── 向量数据库       12%  ($384)
├── 服务器/带宽       8%  ($256)
└── 监控/日志         8%  ($256)

class ModelRouter:
    def __init__(self):
        self.routes = {
            "simple_qa": {
                "model": "gpt-4o-mini",
                "criteria": lambda q: len(q) < 50 and not q.requires_tools
            },
            "standard": {
                "model": "claude-3.5-sonnet",
                "criteria": lambda q: q.complexity < 5
            },
            "complex": {
                "model": "gpt-4o",
                "criteria": lambda q: True  # fallback
            }
        }
    
    def route(self, query):
        for level, config in self.routes.items():
            if config["criteria"](query):
                return config["model"]
        return "gpt-4o"

# 效果：62% 的请求被路由到 mini 模型
# 月节省：$1,180

from sklearn.metrics.pairwise import cosine_similarity

class SemanticCache:
    def __init__(self, threshold=0.95):
        self.cache = {}  # {embedding: (query, response, timestamp)}
        self.threshold = threshold
    
    def get(self, query):
        query_emb = embed(query)
        
        for emb, (cached_q, cached_r, ts) in self.cache.items():
            sim = cosine_similarity([query_emb], [emb])[0][0]
            if sim > self.threshold:
                if self.is_fresh(ts):
                    return cached_r
        
        return None
    
    def set(self, query, response, ttl=3600):
        emb = embed(query)
        self.cache[emb] = (query, response, time.time())

# 效果：23% 的请求命中缓存
# 月节省：$420

class ContextCompressor:
    def compress(self, messages, max_tokens=2000):
        total = count_tokens(messages)
        if total <= max_tokens:
            return messages
        
        # 保留最近3轮对话
        recent = messages[-6:]
        old = messages[:-6]
        
        # 用小模型做摘要
        summary = small_llm.summarize(old)
        
        return [
            {"role": "system", "content": f"之前的对话摘要：{summary}"}
        ] + recent

# 效果：平均 Token 消耗降低 45%
# 月节省：$380

import asyncio

class BatchProcessor:
    async def process_batch(self, queries):
        # OpenAI Batch API：50% 折扣
        batch = create_batch_request(queries)
        result = await openai.batch.create(batch)
        return result
    
    async def embed_batch(self, texts):
        # 批量嵌入，减少 API 调用次数
        return await openai.embeddings.create(
            input=texts,  # 一次最多2048个
            model="text-embedding-3-small"
        )

# 效果：嵌入成本降低 50%
# 月节省：$160

# 优化前（185 tokens）
BAD_PROMPT = """你是一个专业的客服代表。你的工作是回答用户的问题。
你需要礼貌、专业、准确。如果用户的问题超出了你的知识范围，请告诉用户
你会转接给人工客服。在回答之前，请先思考用户问题的核心需求是什么，
然后给出针对性的回答。回答时请使用中文，格式清晰，分段合理。"""

# 优化后（42 tokens）
GOOD_PROMPT = """客服规范：
1. 中文回答
2. 不确定时转人工
3. 先理解需求再回答"""

# 效果：System Prompt 每次调用都发送，优化后节省 143 tokens/次
# 按日均5000次计算，月节省：$85

# Nginx 层缓存静态资源和常见查询
proxy_cache_path /var/cache/nginx levels=1:2 keys_zone=api_cache:10m;

location /api/qa {
    proxy_cache api_cache;
    proxy_cache_valid 200 1h;
    proxy_cache_key "$request_method$request_uri$request_body";
    proxy_pass http://backend;
}

# 向量数据库优化：从 Pinecone 迁移到自建 Qdrant
# Pinecone: $70/month (1M vectors)
# Qdrant 自建: $10/month (2核2G服务器)
# 月节省：$60

class CostMonitor:
    def __init__(self):
        self.daily_budget = 25  # $25/day
    
    def check(self):
        today_cost = self.get_today_cost()
        
        if today_cost > self.daily_budget * 0.8:
            # 接近预算，自动降级
            self.downgrade_model()
            alert(f"日成本 ${today_cost} 接近预算")
        
        if today_cost > self.daily_budget:
            # 超预算，紧急措施
            self.enable_aggressive_cache()
            alert(f"日成本 ${today_cost} 超预算！", severity="critical")