Agent 成本优化实战：Token 经济学深度分析

单次成本 = Σ(每轮 LLM 调用成本) + Σ(工具调用成本)

每轮 LLM 成本 = (输入Token × 输入单价) + (输出Token × 输出单价)
              + (缓存命中Token × 缓存单价)

工具调用成本 = API调用费 + 计算资源费

class AgentCostModel:
    """Agent 成本建模"""
    
    PRICING = {
        "gpt-5": {"input": 5.0, "output": 15.0, "cached": 2.5},
        "gpt-5-mini": {"input": 0.3, "output": 1.5, "cached": 0.15},
    }
    
    def estimate(
        self,
        model: str,
        system_prompt_tokens: int,
        conversation_tokens: int,
        avg_output_tokens: int,
        iterations: int,
        cache_hit_ratio: float = 0.0
    ) -> CostEstimate:
        pricing = self.PRICING[model]
        
        per_iteration_cost = 0
        input_tokens = system_prompt_tokens + conversation_tokens
        
        # 缓存部分
        cached_tokens = int(input_tokens * cache_hit_ratio)
        non_cached_tokens = input_tokens - cached_tokens
        
        per_iteration_cost = (
            (non_cached_tokens / 1_000_000) * pricing["input"] +
            (cached_tokens / 1_000_000) * pricing["cached"] +
            (avg_output_tokens / 1_000_000) * pricing["output"]
        )
        
        total_cost = per_iteration_cost * iterations
        
        return CostEstimate(
            per_iteration=per_iteration_cost,
            total=total_cost,
            input_tokens_per_iter=input_tokens,
            output_tokens_per_iter=avg_output_tokens,
            cache_savings=per_iteration_cost * cache_hit_ratio * 0.5
        )

# 示例：GPT-5 Agent，5轮迭代
model = AgentCostModel()
estimate = model.estimate(
    model="gpt-5",
    system_prompt_tokens=2000,
    conversation_tokens=3000,
    avg_output_tokens=500,
    iterations=5,
    cache_hit_ratio=0.3
)
# 单次交互成本：~$0.15
# 月度成本（10万次/月）：~$15,000

class ModelRouter:
    """智能模型路由器"""
    
    def __init__(self):
        self.complexity_classifier = self._load_classifier()
    
    def route(self, request: str, context: dict) -> str:
        complexity = self._assess_complexity(request, context)
        
        if complexity == "simple":
            return "gpt-5-mini"     # 30% 成本
        elif complexity == "moderate":
            return "claude-sonnet-4"  # 60% 成本
        else:
            return "gpt-5"           # 100% 成本
    
    def _assess_complexity(self, request: str, context: dict) -> str:
        # 规则 + 分类器混合判断
        if len(request) < 50 and "?" in request:
            return "simple"
        
        if any(kw in request.lower() for kw in ["分析", "设计", "对比", "架构"]):
            return "moderate"
        
        if context.get("tool_count", 0) > 5:
            return "complex"
        
        if context.get("conversation_turns", 0) > 10:
            return "complex"
        
        return self.complexity_classifier.predict(request)

import hashlib
import numpy as np

class SemanticCache:
    """语义缓存：基于向量相似度命中"""
    
    def __init__(self, redis_client, embedding_model):
        self.redis = redis_client
        self.embedder = embedding_model
        self.similarity_threshold = 0.95
    
    async def get(self, query: str, context_hash: str) -> str | None:
        # 1. 精确匹配
        exact_key = f"cache:exact:{context_hash}:{hashlib.md5(query.encode()).hexdigest()}"
        cached = await self.redis.get(exact_key)
        if cached:
            return cached
        
        # 2. 语义匹配
        query_embedding = await self._embed(query)
        
        # 在同一 context 下搜索相似查询
        similar = await self._vector_search(query_embedding, context_hash)
        if similar and similar.score > self.similarity_threshold:
            return similar.response
        
        return None
    
    async def set(self, query: str, response: str, context_hash: str, ttl: int = 3600):
        embedding = await self._embed(query)
        
        # 存储精确缓存
        exact_key = f"cache:exact:{context_hash}:{hashlib.md5(query.encode()).hexdigest()}"
        await self.redis.setex(exact_key, ttl, response)
        
        # 存储向量索引
        await self._vector_store(query, embedding, response, context_hash, ttl)

class PromptCompressor:
    """Prompt 压缩器"""
    
    async def compress(self, messages: list[dict]) -> list[dict]:
        # 1. 对话历史滑动窗口
        if len(messages) > 20:
            # 保留最近 10 轮 + 摘要
            old_messages = messages[:-10]
            summary = await self._summarize(old_messages)
            messages = [
                {"role": "system", "content": f"Previous conversation summary: {summary}"}
            ] + messages[-10:]
        
        # 2. 移除冗余
        messages = [m for m in messages if m["content"].strip()]
        
        # 3. 合并连续相同角色
        merged = []
        for msg in messages:
            if merged and merged[-1]["role"] == msg["role"]:
                merged[-1]["content"] += "\n" + msg["content"]
            else:
                merged.append(msg.copy())
        
        # 4. 截断过长内容
        for msg in merged:
            if len(msg["content"]) > 5000:
                msg["content"] = msg["content"][:2500] + "\n[...truncated...]\n" + msg["content"][-2500:]
        
        return merged
    
    async def _summarize(self, messages: list[dict]) -> str:
        """使用便宜模型生成摘要"""
        conversation = "\n".join(f"{m['role']}: {m['content'][:200]}" for m in messages)
        response = await cheap_llm.invoke(
            f"Summarize in 200 words: {conversation}"
        )
        return response.content

class OutputOptimizer:
    """输出优化器"""
    
    # 强制结构化输出减少 Token
    OUTPUT_FORMATS = {
        "classification": "Respond with only the label. No explanation.",
        "extraction": "Respond with JSON only. No markdown formatting.",
        "summary": f"Respond in max 3 sentences (under {100} tokens).",
        "code": "Respond with code only. No explanation unless asked.",
    }
    
    def optimize_prompt(self, base_prompt: str, task_type: str) -> str:
        constraint = self.OUTPUT_FORMATS.get(task_type, "")
        return f"{base_prompt}\n\n{constraint}"
    
    def estimate_savings(self, task_type: str) -> float:
        """估算不同任务的 Token 节省"""
        savings = {
            "classification": 0.90,  # 从 200 tokens → 5 tokens
            "extraction": 0.60,
            "summary": 0.50,
            "code": 0.40,
        }
        return savings.get(task_type, 0.0)

class BatchProcessor:
    """批处理优化"""
    
    async def batch_inference(
        self,
        requests: list[str],
        model: str = "gpt-5"
    ) -> list[str]:
        """使用 OpenAI Batch API（50% 折扣）"""
        
        # 创建批处理文件
        batch_requests = [
            {
                "custom_id": f"req-{i}",
                "method": "POST",
                "url": "/v1/chat/completions",
                "body": {
                    "model": model,
                    "messages": [{"role": "user", "content": req}],
                    "max_tokens": 500,
                }
            }
            for i, req in enumerate(requests)
        ]
        
        # 上传批处理文件
        batch_file = await self.client.files.create(
            file=json.dumps(batch_requests).encode(),
            purpose="batch"
        )
        
        # 创建批处理任务
        batch = await self.client.batches.create(
            input_file_id=batch_file.id,
            endpoint="/v1/chat/completions",
            completion_window="24h"
        )
        
        # 等待完成（异步）
        results = await self._wait_for_batch(batch.id)
        return results

class SelfHostedRouter:
    """自托管模型路由"""
    
    def __init__(self):
        self.local_models = {
            "llama-4-70b": {
                "endpoint": "http://gpu-cluster:8000/v1",
                "cost_per_1m_tokens": 0.05,  # 仅电费+折旧
                "quality_score": 0.85,       # 相对 GPT-5
                "latency_p95_ms": 500,
            },
            "qwen-3-72b": {
                "endpoint": "http://gpu-cluster:8001/v1",
                "cost_per_1m_tokens": 0.05,
                "quality_score": 0.82,
                "latency_p95_ms": 450,
            }
        }
    
    def should_route_local(self, request: str, budget_priority: float) -> bool:
        """决策：是否路由到本地模型"""
        if budget_priority < 0.5:
            return False  # 质量优先
        
        complexity = self._assess_complexity(request)
        if complexity == "simple":
            return True  # 简单任务用本地
        
        return False

class CostDashboard:
    """Agent 成本仪表盘数据源"""
    
    METRICS = {
        # 实时指标
        "cost_per_request": "P50/P95/P99 单次请求成本",
        "token_efficiency": "有效输出Token / 总Token",
        "cache_hit_rate": "缓存命中率",
        "model_distribution": "各模型调用占比",
        
        # 聚合指标
        "daily_cost": "日总成本",
        "cost_per_user": "每用户日均成本",
        "cost_per_task": "每任务类型平均成本",
        
        # 告警指标
        "cost_spike": "成本突增检测",
        "budget_burn_rate": "预算消耗速度",
    }
    
    def generate_report(self, period: str = "daily") -> dict:
        return {
            "total_cost": self._query_sum("agent.cost.total", period),
            "by_model": self._query_group("agent.cost.total", "model", period),
            "by_agent": self._query_group("agent.cost.total", "agent", period),
            "by_user_tier": self._query_group("agent.cost.total", "tier", period),
            "trend": self._query_trend("agent.cost.total", period),
            "projected_monthly": self._project_monthly(),
            "optimization_savings": self._calc_savings(),
        }