引言

Agent系统的运营成本主要由三部分构成：LLM Token费用（60-70%）、基础设施成本（20-25%）和工具/API调用费用（5-15%）。随着用户规模增长，如果不做系统性的成本优化，每月的运营成本可能达到数十万甚至数百万美元。

2026年，Agent成本优化（Agent FinOps）已成为独立的技术领域。本文将从实际工程角度，系统讲解如何在不损害用户体验的前提下，将Agent系统的运营成本降低50-70%。

成本构成分析

class CostBreakdown:
    """Agent系统成本分解"""
    
    TYPICAL_DISTRIBUTION = {
        "llm_inference": {
            "proportion": 0.65,
            "sub_items": {
                "input_tokens": 0.35,
                "output_tokens": 0.55,
                "embedding": 0.10,
            }
        },
        "infrastructure": {
            "proportion": 0.22,
            "sub_items": {
                "gpu_compute": 0.55,
                "cpu_compute": 0.20,
                "storage": 0.15,
                "network": 0.10,
            }
        },
        "external_apis": {
            "proportion": 0.10,
            "sub_items": {
                "search_api": 0.40,
                "tool_apis": 0.35,
                "data_sources": 0.25,
            }
        },
        "observability": {
            "proportion": 0.03,
            "sub_items": {
                "logging": 0.40,
                "tracing": 0.30,
                "monitoring": 0.30,
            }
        }
    }

Token优化

Prompt压缩

class PromptOptimizer:
    """Prompt压缩与优化"""
    
    async def optimize_prompt(self, messages: list) -> list:
        """优化对话历史，减少Token消耗"""
        
        # 策略1：摘要旧消息
        if len(messages) > 10:
            old_messages = messages[:-4]
            recent_messages = messages[-4:]
            
            summary = await self._summarize(old_messages)
            
            messages = [
                {"role": "system", "content": f"Previous context: {summary}"}
            ] + recent_messages
        
        # 策略2：移除冗余系统提示
        messages = self._deduplicate_system_messages(messages)
        
        # 策略3：压缩工具描述
        messages = self._compress_tool_definitions(messages)
        
        return messages
    
    async def _summarize(self, messages: list) -> str:
        """使用小模型生成摘要"""
        summary_prompt = "Summarize the following conversation in 200 words:"
        for msg in messages:
            summary_prompt += f"\n{msg['role']}: {msg['content'][:200]}"
        
        response = await self.llm.generate(
            model="gpt-4o-mini",  # 用小模型做摘要
            prompt=summary_prompt,
            max_tokens=300
        )
        return response
    
    def _compress_tool_definitions(self, messages: list) -> list:
        """压缩工具定义"""
        for msg in messages:
            if msg["role"] == "system" and "tools" in msg.get("content", ""):
                # 只保留当前轮次需要的工具
                needed_tools = self._get_relevant_tools(msg["content"])
                msg["content"] = self._rewrite_tool_section(needed_tools)
        return messages

响应长度控制

class ResponseLengthController:
    """响应长度控制"""
    
    LENGTH_BUDGETS = {
        "simple_qa": {"max_tokens": 200, "avg_tokens": 100},
        "explanation": {"max_tokens": 500, "avg_tokens": 300},
        "code_generation": {"max_tokens": 2000, "avg_tokens": 800},
        "analysis": {"max_tokens": 1500, "avg_tokens": 800},
        "creative": {"max_tokens": 1000, "avg_tokens": 500},
    }
    
    def get_token_budget(self, request: dict, route: dict) -> int:
        """获取Token预算"""
        task_type = route.get("task_type", "explanation")
        budget = self.LENGTH_BUDGETS.get(task_type, self.LENGTH_BUDGETS["explanation"])
        
        # 根据用户tier调整
        user_tier = request.get("user_tier", "free")
        if user_tier == "free":
            budget["max_tokens"] = int(budget["max_tokens"] * 0.7)
        
        return budget["max_tokens"]

缓存策略

多级缓存

class AgentCacheStack:
    """Agent多级缓存"""
    
    def __init__(self):
        self.layers = [
            self._exact_match_cache,    # L1: 精确匹配
            self._semantic_cache,        # L2: 语义缓存
            self._tool_result_cache,     # L3: 工具结果缓存
            self._embedding_cache,       # L4: 嵌入缓存
        ]
    
    async def get_or_compute(self, request: dict) -> dict:
        """多级缓存查询"""
        
        # L1: 精确匹配
        cache_key = self._generate_key(request)
        cached = await self._exact_match_cache.get(cache_key)
        if cached:
            self._record_cache_hit("L1_exact")
            return cached
        
        # L2: 语义相似查询
        similar = await self._semantic_cache.find_similar(
            query=request["input"],
            threshold=0.95
        )
        if similar:
            self._record_cache_hit("L2_semantic")
            return similar
        
        # L3: 检查工具结果缓存
        if request.get("tool_calls"):
            for tool_call in request["tool_calls"]:
                tool_result = await self._tool_result_cache.get(
                    tool_call["name"], 
                    tool_call["params"]
                )
                if tool_result:
                    tool_call["cached_result"] = tool_result
        
        # 缓存未命中，执行计算
        result = await self._execute(request)
        
        # 回填缓存
        await self._fill_caches(request, result)
        
        return result
    
    async def _fill_caches(self, request: dict, result: dict):
        """回填各级缓存"""
        # L1
        await self._exact_match_cache.set(
            self._generate_key(request),
            result,
            ttl=3600  # 1小时
        )
        
        # L2
        await self._semantic_cache.store(
            query=request["input"],
            response=result,
            embedding=await self._embed(request["input"]),
            ttl=86400  # 24小时
        )
        
        # L3
        if result.get("tool_results"):
            for tool_result in result["tool_results"]:
                await self._tool_result_cache.set(
                    tool_result["tool_name"],
                    tool_result["params"],
                    tool_result["output"],
                    ttl=1800  # 30分钟
                )

缓存命中率监控

class CacheMetrics:
    """缓存指标监控"""
    
    async def get_cache_report(self) -> dict:
        return {
            "l1_exact": {
                "hit_rate": await self._get_rate("cache_l1_hit", "cache_l1_miss"),
                "size": await self._get_size("exact_cache"),
                "eviction_rate": await self._get_rate("cache_l1_evict"),
            },
            "l2_semantic": {
                "hit_rate": await self._get_rate("cache_l2_hit", "cache_l2_miss"),
                "size": await self._get_size("semantic_cache"),
                "avg_similarity": await self._get_avg("cache_l2_similarity"),
            },
            "l3_tool": {
                "hit_rate": await self._get_rate("cache_l3_hit", "cache_l3_miss"),
                "savings_usd": await self._get_savings("tool_cache"),
            },
            "total_savings": {
                "tokens_saved": await self._get_total("tokens_saved"),
                "cost_saved_usd": await self._get_total("cost_saved"),
                "latency_saved_ms": await self._get_total("latency_saved"),
            }
        }

模型选择优化

class ModelCostOptimizer:
    """模型成本优化器"""
    
    MODEL_COSTS = {
        "gpt-4o": {"input": 2.50, "output": 10.00},  # per 1M tokens
        "gpt-4o-mini": {"input": 0.15, "output": 0.60},
        "claude-3.5-sonnet": {"input": 3.00, "output": 15.00},
        "claude-3-haiku": {"input": 0.25, "output": 1.25},
        "deepseek-v3": {"input": 0.14, "output": 0.28},
        "local-llama-70b": {"input": 0.05, "output": 0.05},  # 自部署
    }
    
    async def select_cost_optimal_model(
        self,
        request: dict,
        quality_threshold: float = 0.8
    ) -> str:
        """选择成本最优模型"""
        
        # 估算各模型成本和质量
        estimates = []
        for model, cost in self.MODEL_COSTS.items():
            estimated_tokens = self._estimate_tokens(request)
            total_cost = (
                estimated_tokens["input"] * cost["input"] +
                estimated_tokens["output"] * cost["output"]
            ) / 1_000_000
            
            quality = await self._estimate_quality(model, request)
            
            if quality >= quality_threshold:
                estimates.append({
                    "model": model,
                    "cost": total_cost,
                    "quality": quality,
                    "value_ratio": quality / total_cost if total_cost > 0 else float('inf')
                })
        
        # 选择性价比最高的
        return max(estimates, key=lambda e: e["value_ratio"])["model"]
    
    def _estimate_tokens(self, request: dict) -> dict:
        """估算Token消耗"""
        input_tokens = len(request["input"]) // 4  # 粗略估算
        output_tokens = min(input_tokens, 500)  # 响应通常比输入短
        return {"input": input_tokens, "output": output_tokens}

批处理优化

class BatchProcessor:
    """请求批处理——合并多个请求降低单次成本"""
    
    async def batch_llm_calls(
        self,
        requests: list,
        max_batch_size: int = 10,
        max_wait_ms: int = 100
    ) -> list:
        """批处理LLM调用"""
        batch = []
        results = {}
        
        for request in requests:
            batch.append(request)
            
            if len(batch) >= max_batch_size:
                batch_results = await self._process_batch(batch)
                results.update(batch_results)
                batch = []
        
        # 处理剩余
        if batch:
            batch_results = await self._process_batch(batch)
            results.update(batch_results)
        
        return [results[r["id"]] for r in requests]
    
    async def _process_batch(self, batch: list) -> dict:
        """处理批次"""
        # 合并多个请求为一个prompt
        combined_prompt = self._combine_prompts(batch)
        
        response = await self.llm.generate(
            model="gpt-4o-mini",
            prompt=combined_prompt,
            max_tokens=2000
        )
        
        # 解析并分配结果
        return self._parse_batch_response(response, batch)

基础设施优化

class InfrastructureOptimizer:
    """基础设施成本优化"""
    
    async def optimize_gpu_utilization(self) -> dict:
        """优化GPU利用率"""
        current_util = await self._get_avg_gpu_utilization()
        
        recommendations = []
        
        if current_util < 60:
            recommendations.append({
                "action": "reduce_gpu_instances",
                "current": self.gpu_count,
                "recommended": max(1, int(self.gpu_count * 0.7)),
                "estimated_savings": (self.gpu_count - max(1, int(self.gpu_count * 0.7))) * 2000  # $2000/GPU/month
            })
        
        # 建议使用Spot实例
        recommendations.append({
            "action": "use_spot_instances",
            "estimated_savings": self.gpu_count * 800,  # 节省40%
            "risk": "可能被中断，需要检查点机制"
        })
        
        # 建议模型量化
        recommendations.append({
            "action": "enable_model_quantization",
            "estimated_savings": self.gpu_count * 500,
            "quality_impact": "轻微下降（<2%）"
        })
        
        return recommendations
    
    async def optimize_storage(self) -> dict:
        """优化存储成本"""
        # 向量数据库分片优化
        vector_db_size = await self._get_vector_db_size()
        
        recommendations = []
        
        # 冷热数据分离
        recommendations.append({
            "action": "tiered_storage",
            "hot_tier_gb": vector_db_size * 0.2,  # 20%热数据
            "cold_tier_gb": vector_db_size * 0.8,
            "estimated_savings": vector_db_size * 0.8 * 0.08  # 冷存储便宜90%
        })
        
        return recommendations

成本看板

class CostDashboard:
    """成本看板"""
    
    async def generate_report(self, period: str = "monthly") -> dict:
        return {
            "period": period,
            "total_cost": await self._get_total_cost(period),
            "cost_by_category": {
                "llm": await self._get_category_cost("llm", period),
                "infrastructure": await self._get_category_cost("infra", period),
                "external_apis": await self._get_category_cost("apis", period),
            },
            "cost_per_request": await self._get_cost_per_request(period),
            "cost_per_tenant": await self._get_cost_per_tenant(period),
            "optimization_savings": {
                "cache_savings": await self._get_cache_savings(period),
                "model_downgrade_savings": await self._get_downgrade_savings(period),
                "batch_savings": await self._get_batch_savings(period),
            },
            "trend": await self._get_trend(period),
            "projected_next_month": await self._project_cost(),
        }

总结

Agent成本优化是一个系统工程，需要从Token、模型、缓存、批处理、基础设施多个层面协同发力。其中缓存策略是最立竿见影的优化手段——一个设计良好的多级缓存系统可以将LLM调用减少30-50%。模型选择优化通过让简单请求使用小模型，可以在不影响体验的前提下节省40-60%的Token成本。

核心原则：成本优化的前提是不损害用户体验。每次优化都应该用质量指标验证——如果质量评分下降超过5%，优化就是得不偿失的。Agent FinOps应该成为持续的工作，而非一次性的项目。

加入讨论

这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞，认知差异创造无限可能。