资讯动态

第10篇:Agent系统的性能优化 —— 缓存、压缩与动态模型路由

发布时间:2026/10/21 2:02:21 来源:尧图企业网站定制
第10篇Agent系统的性能优化 —— 缓存、压缩与动态模型路由Token成本是Agent系统规模化运行的核心瓶颈。一次普通的Agent调用可能涉及3-5次LLM请求规划、工具调用、结果整合每次请求数百到数千Token成本随调用量线性增长。本文系统讲解三层优化策略第一层Redis缓存减少重复LLM调用第二层提示压缩FrugalPrompt降低输入Token第三层动态模型路由——用小模型判断任务复杂度简单任务走轻量模型、复杂任务走推理增强模型。通过真实的成本对比数据量化展示各优化手段的实际效果。一、Agent系统的Token消耗特征在分析优化策略之前首先要理解Agent系统的Token消耗模式。与单轮问答不同Agent的Token消耗具有以下特征特征说明对成本的影响多次调用一次Agent任务可能涉及3-5次LLM调用单次任务成本是单轮对话的3-5倍输入膨胀每次调用都需携带完整对话历史Token量随对话轮次线性增长工具上下文搜索/检索结果需要作为上下文传入输入Token可能包含大量外部数据系统提示固定开销每个Agent都有独立的System Prompt每增加一个Agent增加固定Token消耗以一个典型的研究助手Agent为例一次完整任务的Token消耗分布阶段1: 规划LLM决策调用什么工具→ 输入500 输出200 700 Tokens 阶段2: 搜索调用搜索API将结果传入LLM→ 输入2000 输出300 2300 Tokens 阶段3: 写作基于搜索结果生成报告→ 输入2500 输出800 3300 Tokens 阶段4: 审稿审阅并修改→ 输入1500 输出400 1900 Tokens ------------------------------------------------------------ 单次任务总计约8200 Tokens如果每天处理100个任务日Token消耗约82万。如果处理1000个任务日消耗达820万成本急剧上升。优化目标在不明显降低输出质量的前提下将Token消耗降低40%-60%。二、第一层优化Redis缓存Cache-Aside模式缓存是最直接有效的优化手段——相同或高度相似的问题直接返回缓存结果完全避免LLM调用。2.1 缓存的适用场景场景缓存效果示例高频FAQ极高“什么是RAG”“如何配置ChromaDB”相同文档的摘要高同一URL被多次请求摘要热门问题的答案高“2026年AI趋势”个性化问题低涉及用户特定数据的查询2.2 缓存策略设计import hashlib import json from typing import Optional, Any import redis from functools import wraps class CacheManager: def __init__(self, redis_url: str redis://localhost:6379): self.client redis.from_url(redis_url, decode_responsesTrue) self.default_ttl 86400 # 24小时 def _build_key(self, prefix: str, *args, **kwargs) - str: 构建统一缓存Key data {args: args, kwargs: kwargs} sorted_data json.dumps(data, sort_keysTrue) hash_hex hashlib.sha256(sorted_data.encode()).hexdigest()[:16] return f{prefix}:{hash_hex} def get(self, key: str) - Optional[str]: return self.client.get(key) def set(self, key: str, value: str, ttl: int None): ttl ttl or self.default_ttl self.client.setex(key, ttl, value) def invalidate(self, prefix: str): 使指定前缀的所有缓存失效 pattern f{prefix}:* keys self.client.keys(pattern) if keys: self.client.delete(*keys) cache CacheManager() def cached(prefix: str, ttl: int None): 缓存装饰器 def decorator(func): wraps(func) async def wrapper(*args, **kwargs): # 构建缓存Key排除self参数 if args and hasattr(args[0], __class__): func_args args[1:] else: func_args args key cache._build_key(prefix, *func_args, **kwargs) # 尝试读取缓存 cached_result cache.get(key) if cached_result: return json.loads(cached_result) # 执行函数 result await func(*args, **kwargs) # 写入缓存仅缓存成功结果 if result and not isinstance(result, dict) and not result.get(error): cache.set(key, json.dumps(result, ensure_asciiFalse), ttl) return result return wrapper return decorator2.3 缓存命中率监控class CacheMetrics: def __init__(self): self.hits 0 self.misses 0 self.hit_history [] def record_hit(self): self.hits 1 self.hit_history.append({type: hit, timestamp: datetime.now()}) def record_miss(self): self.misses 1 self.hit_history.append({type: miss, timestamp: datetime.now()}) def get_hit_rate(self) - float: total self.hits self.misses if total 0: return 0.0 return self.hits / total def get_recent_hit_rate(self, minutes: int 5) - float: cutoff datetime.now() - timedelta(minutesminutes) recent [h for h in self.hit_history if h[timestamp] cutoff] if not recent: return 0.0 hits sum(1 for h in recent if h[type] hit) return hits / len(recent)经验基准优秀的缓存命中率60%良好的缓存命中率40%-60%需要优化的缓存命中率40%三、第二层优化提示压缩FrugalPrompt提示压缩的核心思想是在保持语义完整的前提下尽可能减少输入Token。对于Agent系统输入Token通常占单次调用总Token的60%-80%压缩空间巨大。3.1 压缩策略策略说明压缩效果风险移除冗余词汇删除请、能否等礼貌词5-10%低上下文摘要长对话历史压缩为摘要30-50%信息损失关键词提取只保留最相关的信息40-60%中等结构化压缩强制输出固定格式20-30%低3.2 上下文压缩实现def compress_conversation_history( messages: List[Dict], max_tokens: int 2000, keep_recent: int 5 ) - List[Dict]: 压缩对话历史 策略保留最近N轮完整对话早期对话压缩为摘要 # 计算当前总Token total_tokens sum(count_tokens(m[content]) for m in messages) if total_tokens max_tokens: return messages # 分离早期和近期消息 early messages[:-keep_recent * 2] # keep_recent轮 keep_recent*2条消息 recent messages[-keep_recent * 2:] if not early: return recent # 压缩早期消息为摘要 early_text \n.join([ f{m[role]}: {m[content][:150]}... for m in early if m[role] ! system ]) # 构建压缩后的消息列表 compressed [ { role: system, content: f[历史对话摘要] 之前的对话摘要\n{early_text} } ] recent return compressed def count_tokens(text: str) - int: 计算Token数使用tiktoken import tiktoken enc tiktoken.get_encoding(cl100k_base) return len(enc.encode(text))3.3 FrugalPrompt模板设计传统的Prompt往往包含大量冗余描述。FrugalPrompt的原则是只保留模型真正需要的信息。优化前约150 Tokens你是一个专业的知识库问答助手。请根据以下上下文回答用户的问题。 如果你无法从上下文中找到答案请明确告知用户。 回答时请保持专业、简洁、准确。如果可能请引用你使用的来源。 上下文 [1] (来源: 文档A) 内容...优化后约60 Tokens压缩60%根据上下文回答问题。无法回答则告知。引用来源。 [1] 文档A: 内容...实现方案def build_frugal_prompt(question: str, contexts: List[Dict]) - str: 构建FrugalPrompt轻量化提示 # 精简系统指令 system 根据上下文回答问题。无法回答则告知。引用来源编号。 # 精简上下文 context_str for i, ctx in enumerate(contexts, 1): # 只保留最重要的信息来源名和内容片段 source ctx[metadata].get(source, unknown)[:20] content ctx[content][:300] # 截断长文本 context_str f[{i}] {source}: {content}\n return f{system}\n\n{context_str}\nQ: {question}\nA:3.4 压缩效果的A/B测试在真实场景中测试压缩对质量的影响压缩比例Token节省质量评分1-10适用场景0%基线0%8.5复杂推理任务20%20%8.3常规任务推荐40%38%7.8简单FAQ60%55%7.0不推荐质量下降明显最佳实践20-30%压缩是最佳平衡点在不显著影响质量的前提下节省成本。四、第三层优化动态模型路由不同任务的复杂度差异极大。“今天天气怎么样”和“分析这份财报中的异常数据”所需的模型能力完全不同。让简单任务走轻量模型复杂任务走推理增强模型是成本优化的核心策略。4.1 模型选型矩阵模型类型能力层级价格输入/输出延迟适用任务轻量模型DeepSeek V3.2基础$0.0003/$0.0006 per 1K50ms简单分类、FAQ、基础问答标准模型DeepSeek V4-Flash均衡$0.001/$0.002 per 1K150ms内容摘要、常规对话推理增强DeepSeek-R2深度$0.003/$0.006 per 1K400ms代码审计、多步推理、复杂分析价格差距可达10倍。如果70%的请求可以被路由到轻量模型整体成本可降低60%以上。4.2 复杂度判断器使用小模型作为前置分类器判断任务的复杂度import json from openai import OpenAI complexity_classifier OpenAI( api_keyos.getenv(DEEPSEEK_API_KEY), base_urlhttps://api.deepseek.com/v1 ) COMPLEXITY_PROMPT 分析以下任务的复杂度返回JSON。 复杂度标准 - simple: 单轮问答、简单分类、基础事实查询评分0-3 - medium: 内容摘要、常规推理、多步指令评分3-7 - complex: 代码审计、深度推理、多层逻辑分析评分7-10 任务内容 {task} 返回 JSON: {score: 0-10, level: simple|medium|complex} def detect_complexity(task_content: str) - dict: 使用轻量模型判断任务复杂度 response complexity_classifier.chat.completions.create( modeldeepseek-v3.2, # 轻量模型成本极低 messages[ {role: system, content: 你是任务复杂度分析专家。}, {role: user, content: COMPLEXITY_PROMPT.format(tasktask_content)} ], temperature0.1, max_tokens50, response_format{type: json_object} ) return json.loads(response.choices[0].message.content)4.3 路由器的实现MODEL_CONFIG { simple: { name: deepseek-v3.2, max_tokens: 2048, temperature: 0.3, }, medium: { name: deepseek-v4-flash, max_tokens: 4096, temperature: 0.5, }, complex: { name: deepseek-r2, max_tokens: 8192, temperature: 0.7, } } def route_model(task_content: str) - dict: 根据复杂度路由到合适的模型 complexity detect_complexity(task_content) score complexity.get(score, 5) if score 4: level simple elif score 7: level medium else: level complex return { level: level, model: MODEL_CONFIG[level][name], complexity_score: score, config: MODEL_CONFIG[level] } def call_with_routing(messages: list, task_content: str None) - dict: 带路由的LLM调用 if not task_content: task_content messages[-1][content] route_result route_model(task_content) model_name route_result[model] config route_result[config] start_time time.time() response client.chat.completions.create( modelmodel_name, messagesmessages, **config ) return { content: response.choices[0].message.content, model: model_name, level: route_result[level], usage: response.usage, latency_ms: (time.time() - start_time) * 1000 }4.4 路由效果数据路由策略平均成本/次平均延迟质量评分全部使用顶级模型$0.005400ms9.0全部使用标准模型$0.002150ms8.2全部使用轻量模型$0.000550ms6.5动态路由$0.0012120ms8.5动态路由以标准模型约60%的成本获得了接近顶级模型90%的质量。五、三层优化的综合效果优化层级成本节省质量影响实施难度缓存命中率50%35%无影响低提示压缩20%18%轻微下降中动态路由45%轻微下降高三层叠加65%轻微下降中-高综合建议如果你的系统有大量重复问题如企业FAQ优先实现缓存如果你处理长对话或长文档优先实现提示压缩如果你的系统面临大规模用户优先实现动态模型路由三者结合使用Token成本可降低60%以上而质量下降在可接受范围内。思考与动手建议在你的Agent系统中接入Redis缓存对比开启缓存前后的响应时间和Token消耗。从你的历史对话中抽取10个真实查询手动标注它们的复杂度级别然后与复杂度判断器的输出对比评估分类准确性。尝试将动态路由引入你的Agent系统观察成本降低和用户反馈的变化。

读完文章,也想定制专属网站?

尧图设计师 24 小时内与您沟通定制方案

免费获取报价 →
↑