缓存 | 硅基 AGI · 智能体学习与测评

LLM缓存策略详解

LLM缓存的价值 LLM推理是昂贵的——每次调用消耗GPU算力和API费用。但很多请求是重复或高度相似的。缓存可以让"已经算过的不再重算"，是投入产出比最高的优化手段。三层缓存架构请求 → L1: 精确缓存 → L2: 语义缓存 → L3: 前缀缓存 → LLM L1：精确缓存 import redis.asyncio as redis class ExactCache: """精确匹配缓存：相同输入返回相同输出""" def __init__(self, redis_url="redis://localhost:6379"): self.redis = redis.from_url(redis_url) def cache_key(self, model, messages, temperature, max_tokens): """生成缓存键""" content = json.dumps({ "model": model, "messages": messages, "temperature": temperature, "max_tokens": max_tokens, }, sort_keys=True, ensure_ascii=False) return hashlib.sha256(content.encode()).hexdigest() async def get(self, model, messages, **params): key = self.cache_key(model, messages, params.get("temperature", 0.7), params.get("max_tokens", 2048)) cached = await self.redis.get(key) return json.loads(cached) if cached else None async def set(self, model, messages, response, ttl=3600, **params): key = self.cache_key(model, messages, params.get("temperature", 0.7), params.get("max_tokens", 2048)) await self.redis.setex(key, ttl, json.dumps(response, ensure_ascii=False)) 适用场景 temperature=0的确定性输出常见问题FAQ 系统提示词相同的请求 L2：语义缓存 class SemanticCache: """语义缓存：相似查询命中缓存""" def __init__(self, vector_store, embed_model, threshold=0.95): self.store = vector_store self.embed_model = embed_model self.threshold = threshold async def get(self, query, model="default", **params): # 向量化查询 query_embedding = await self.embed_model.embed(query) # 搜索相似缓存 results = await self.store.search( query_embedding, filter={"model": model}, top_k=1 ) if results and results[0]["score"] >= self.threshold: cached = json.loads(results[0]["document"]) # 检查参数兼容性 if self.params_compatible(cached["params"], params): return cached["response"] return None async def set(self, query, response, model="default", ttl=3600, **params): embedding = await self.embed_model.embed(query) await self.store.add( id=str(uuid.uuid4()), embedding=embedding, document=json.dumps({ "query": query, "response": response, "model": model, "params": params, "timestamp": time.time(), }, ensure_ascii=False), ttl=ttl ) def params_compatible(self, cached_params, request_params): """检查缓存参数是否兼容当前请求""" # temperature差异较大则不兼容 if abs(cached_params.get("temperature", 0.7) - request_params.get("temperature", 0.7)) > 0.1: return False return True 语义缓存的阈值选择阈值命中率准确率适用场景 0.99 低极高严格场景（医疗、法律） 0.95 中高通用场景 0.90 高中 FAQ类场景 0.85 很高低不推荐（风险大） L3：前缀缓存（KV Cache共享） class PrefixCacheManager: """前缀缓存：共享相同前缀的KV Cache""" def __init__(self): self.prefix_cache = {} # prefix_hash -> kv_cache def compute_prefix_hash(self, messages): """计算消息前缀的哈希""" # 系统提示 + 历史对话通常构成共享前缀 prefix_text = json.dumps(messages[:-1]) # 除最后一条消息 return hashlib.sha256(prefix_text.encode()).hexdigest() async def get_kv_cache(self, messages): """获取前缀的KV Cache""" prefix_hash = self.compute_prefix_hash(messages) return self.prefix_cache.get(prefix_hash) async def store_kv_cache(self, messages, kv_cache): """存储前缀的KV Cache""" prefix_hash = self.compute_prefix_hash(messages) self.prefix_cache[prefix_hash] = kv_cache vLLM内置前缀缓存支持： ...

AI成本优化2026实战

AI成本的结构 AI系统的成本主要由以下几部分构成： API调用费：按token计费的LLM API费用（通常占60-70%）基础设施：GPU服务器/云服务费用（自部署场景）向量数据库：存储和检索费用数据标注：人工标注和评估成本工程开发：系统开发和维护的人力成本模型分层策略 class ModelTierRouter: """根据任务复杂度路由到不同级别的模型""" def __init__(self): self.tiers = { "simple": {"model": "qwen3-7b", "cost_per_1k": 0.0005}, "medium": {"model": "qwen3-32b", "cost_per_1k": 0.002}, "complex": {"model": "qwen3-72b", "cost_per_1k": 0.008}, } def route(self, query, context=None): # 简单规则路由 if len(query) < 50 and "?" in query: return self.tiers["simple"] # 基于意图分类 complexity = self.estimate_complexity(query, context) if complexity < 0.3: return self.tiers["simple"] elif complexity < 0.7: return self.tiers["medium"] else: return self.tiers["complex"] def estimate_complexity(self, query, context): """估计查询复杂度""" score = 0 # 长查询更复杂 score += min(len(query) / 1000, 0.3) # 需要推理的关键词 reasoning_words = ["分析", "比较", "为什么", "推导", "计算"] score += 0.2 * any(w in query for w in reasoning_words) # 多轮上下文更复杂 if context and len(context) > 5: score += 0.2 # 代码生成更复杂 if "代码" in query or "函数" in query: score += 0.3 return min(score, 1.0) 缓存策略 class SemanticCache: """语义缓存：相似查询命中缓存""" def __init__(self, vector_store, similarity_threshold=0.95): self.store = vector_store self.threshold = similarity_threshold async def get(self, query): """查询缓存""" query_embedding = await self.embed(query) results = await self.store.search( query_embedding, top_k=1 ) if results and results[0].score > self.threshold: # 缓存命中 return json.loads(results[0].document) return None async def set(self, query, response, ttl=3600): """写入缓存""" embedding = await self.embed(query) await self.store.add( id=generate_uuid(), embedding=embedding, document=json.dumps({ "query": query, "response": response, "timestamp": time.time(), }), ttl=ttl ) Token优化 class TokenOptimizer: """减少token消耗的各种策略""" def compress_history(self, messages): """压缩对话历史""" if len(messages) <= 4: return messages # 保留最近2轮 + 摘要 recent = messages[-4:] old = messages[:-4] summary = self.summarize(old) return [{"role": "system", "content": f"历史摘要：{summary}"}] + recent def optimize_prompt(self, prompt): """精简提示词""" # 移除冗余空格和换行 prompt = re.sub(r'\n{3,}', '\n\n', prompt) prompt = re.sub(r' {2,}', ' ', prompt) return prompt.strip() def truncate_context(self, documents, max_tokens=2000): """智能截断文档""" result = [] current_tokens = 0 for doc in documents: doc_tokens = len(doc) // 4 if current_tokens + doc_tokens > max_tokens: # 截断最后一个文档而非完全丢弃 remaining = max_tokens - current_tokens if remaining > 100: result.append(doc[:remaining * 4]) break result.append(doc) current_tokens += doc_tokens return result 批处理优化 class BatchProcessor: """合并多个请求降低API调用次数""" async def batch_generate(self, requests): """将多个独立请求合并为一个""" # 合并多个查询为一个prompt combined_prompt = "请依次回答以下问题：\n\n" for i, req in enumerate(requests): combined_prompt += f"问题{i+1}：{req['query']}\n" combined_prompt += "\n请按问题编号分别回答。" response = await self.llm.generate(combined_prompt) # 解析响应 answers = self.parse_batch_response(response, len(requests)) return answers 成本监控 class CostMonitor: def __init__(self): self.daily_costs = defaultdict(float) self.user_costs = defaultdict(float) self.model_costs = defaultdict(float) def record(self, user_id, model, input_tokens, output_tokens): pricing = { "qwen3-7b": {"input": 0.0003, "output": 0.0006}, "qwen3-32b": {"input": 0.001, "output": 0.002}, "qwen3-72b": {"input": 0.004, "output": 0.008}, } rates = pricing.get(model, {"input": 0.001, "output": 0.002}) cost = (input_tokens / 1000 * rates["input"] + output_tokens / 1000 * rates["output"]) today = datetime.now().strftime("%Y-%m-%d") self.daily_costs[today] += cost self.user_costs[user_id] += cost self.model_costs[model] += cost def alert_if_over_budget(self, daily_budget=100): today = datetime.now().strftime("%Y-%m-%d") if self.daily_costs[today] > daily_budget: self.send_alert( f"日成本超预算：${self.daily_costs[today]:.2f} / ${daily_budget}" ) 成本优化效果策略节省比例实现难度模型分层 30-50% 低语义缓存 20-40% 中 Token优化 10-20% 低批处理 15-25% 中本地部署替代 50-80% 高结语 AI成本优化是一个持续过程。模型分层让简单任务用小模型，语义缓存避免重复计算，Token优化减少浪费，批处理提升效率。组合使用这些策略，可以在不降低用户体验的前提下将AI成本降低50-70%。加入讨论这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

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

引言 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成本。 ...

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

引言一个生产级 Agent 的月度 LLM 账单可以从几百美元到数十万美元不等。2026年，随着 Agent 在企业中的大规模部署，成本优化已成为工程团队的核心 KPI。本文将从 Token 定价模型出发，系统化拆解 Agent 成本结构，并提供可落地的优化方案。一、Token 经济学基础 1.1 定价模型（2026年Q2市场价）模型输入 ($/1M tokens) 输出 ($/1M tokens) 缓存输入 ($/1M tokens) 上下文窗口 GPT-5 $5.00 $15.00 $2.50 256K GPT-5-mini $0.30 $1.50 $0.15 128K Claude Opus 4 $8.00 $24.00 $4.00 200K Claude Sonnet 4 $3.00 $15.00 $1.50 200K Gemini 2.5 Pro $2.50 $10.00 $1.25 2M DeepSeek V4 $0.15 $0.60 $0.07 128K Llama 4 405B $0.80 $2.40 $0.40 128K 关键洞察：输出 Token 的价格是输入的 3-5 倍。因此，减少输出 Token 比减少输入 Token 更具成本效益。 ...

LLM 缓存策略：语义缓存与多级缓存架构

1. 为什么 LLM 需要缓存 LLM 调用成本高、延迟高。以 GPT-4o 为例：延迟：单次请求 500ms - 3000ms 成本：输入 $2.5/1M tokens，输出 $10/1M tokens 并发限制：RPM（每分钟请求数）和 TPM（每分钟 tokens）受限缓存可以显著降低延迟和成本。但 LLM 缓存不同于传统缓存：用户问题"今天天气"和"天气怎么样"语义相近，应该命中同一缓存条目——这就是语义缓存的核心价值。 2. 语义缓存原理 2.1 相似度计算 from dataclasses import dataclass import numpy as np @dataclass class CacheEntry: query: str query_embedding: np.ndarray response: str model: str tokens_used: int timestamp: float hit_count: int = 0 ttl_seconds: int = 3600 class SemanticCache: """ 语义缓存：基于向量相似度的缓存命中核心思想：如果新查询与已缓存查询的语义距离 < 阈值，则复用缓存结果 """ def __init__(self, embedding_model, similarity_threshold: float = 0.95): self.embedding_model = embedding_model self.threshold = similarity_threshold self.cache: list[CacheEntry] = [] self.index = None # 可选：FAISS/Annoy 加速 async def get(self, query: str) -> tuple[bool, str | None]: query_embedding = await self.embedding_model.embed(query) # 线性搜索（生产环境用向量索引） for entry in self.cache: similarity = self._cosine_similarity(query_embedding, entry.query_embedding) if similarity >= self.threshold: entry.hit_count += 1 return True, entry.response return False, None async def set(self, query: str, response: str, model: str, tokens: int): embedding = await self.embedding_model.embed(query) entry = CacheEntry( query=query, query_embedding=embedding, response=response, model=model, tokens_used=tokens, timestamp=time.time() ) self.cache.append(entry) def _cosine_similarity(self, v1: np.ndarray, v2: np.ndarray) -> float: return float(np.dot(v1, v2) / (np.linalg.norm(v1) * np.linalg.norm(v2))) 2.2 阈值调优 import matplotlib.pyplot as plt class ThresholdOptimizer: """语义缓存阈值优化器""" def __init__(self, cache: SemanticCache, test_data: list[tuple[str, str, bool]]): """ test_data: [(query1, cached_query, should_hit), ...] """ self.cache = cache self.test_data = test_data def evaluate(self, threshold: float) -> dict: self.cache.threshold = threshold true_positives = 0 false_positives = 0 false_negatives = 0 true_negatives = 0 for query, cached, should_hit in self.test_data: # 手动计算相似度 e1 = await self.cache.embedding_model.embed(query) e2 = await self.cache.embedding_model.embed(cached) sim = self.cache._cosine_similarity(e1, e2) is_hit = sim >= threshold if should_hit and is_hit: true_positives += 1 elif not should_hit and is_hit: false_positives += 1 elif should_hit and not is_hit: false_negatives += 1 else: true_negatives += 1 precision = true_positives / (true_positives + false_positives) if (true_positives + false_positives) > 0 else 0 recall = true_positives / (true_positives + false_negatives) if (true_positives + false_negatives) > 0 else 0 f1 = 2 * precision * recall / (precision + recall) if (precision + recall) > 0 else 0 return {"threshold": threshold, "precision": precision, "recall": recall, "f1": f1} def find_optimal_threshold(self) -> float: thresholds = np.arange(0.80, 1.0, 0.01) results = [self.evaluate(t) for t in thresholds] best = max(results, key=lambda x: x["f1"]) return best["threshold"] 推荐阈值： ...

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

一个真实案例我们运营一个客服 Agent，日均处理 5000 次对话。优化前月成本约 $3,200，优化后降至 $680——降本 78%。以下是完整的优化路径。成本构成分析优化前月成本分布： ├── 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 第二层：语义缓存相似问题命中缓存，避免重复调用 LLM： ...