企业官网建设流程全解析

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

Token成本是Agent系统规模化运行的核心瓶颈。一次普通的Agent调用可能涉及3-5次LLM请求（规划、工具调用、结果整合），每次请求数百到数千Token，成本随调用量线性增长。本文系统讲解三层优化策略：第一层，Redis缓存减少重复LLM调用；第二层，提示压缩（FrugalPrompt）降低输入Token；第三层，动态模型路由——用小模型判断任务复杂度，简单任务走轻量模型、复杂任务走推理增强模型。通过真实的成本对比数据，量化展示各优化手段的实际效果。

一、Agent系统的Token消耗特征

在分析优化策略之前，首先要理解Agent系统的Token消耗模式。与单轮问答不同，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 缓存的适用场景

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_responses=True) 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_keys=True) 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_ascii=False), ttl) return result return wrapper return decorator

2.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(minutes=minutes) 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 压缩策略

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测试

在真实场景中测试压缩对质量的影响：

最佳实践：20-30%压缩是最佳平衡点，在不显著影响质量的前提下节省成本。

四、第三层优化：动态模型路由

不同任务的复杂度差异极大。“今天天气怎么样？”和“分析这份财报中的异常数据”所需的模型能力完全不同。让简单任务走轻量模型，复杂任务走推理增强模型，是成本优化的核心策略。

4.1 模型选型矩阵

价格差距可达10倍。如果70%的请求可以被路由到轻量模型，整体成本可降低60%以上。

4.2 复杂度判断器

使用小模型作为前置分类器，判断任务的复杂度：

import json from openai import OpenAI complexity_classifier = OpenAI( api_key=os.getenv("DEEPSEEK_API_KEY"), base_url="https://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( model="deepseek-v3.2", # 轻量模型，成本极低 messages=[ {"role": "system", "content": "你是任务复杂度分析专家。"}, {"role": "user", "content": COMPLEXITY_PROMPT.format(task=task_content)} ], temperature=0.1, max_tokens=50, 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( model=model_name, messages=messages, **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 路由效果数据

动态路由以标准模型约60%的成本，获得了接近顶级模型90%的质量。

五、三层优化的综合效果

综合建议：

• 如果你的系统有大量重复问题（如企业FAQ），优先实现缓存
• 如果你处理长对话或长文档，优先实现提示压缩
• 如果你的系统面临大规模用户，优先实现动态模型路由

三者结合使用，Token成本可降低60%以上，而质量下降在可接受范围内。

思考与动手建议：

1. 在你的Agent系统中接入Redis缓存，对比开启缓存前后的响应时间和Token消耗。
2. 从你的历史对话中抽取10个真实查询，手动标注它们的复杂度级别，然后与复杂度判断器的输出对比，评估分类准确性。
3. 尝试将动态路由引入你的Agent系统，观察成本降低和用户反馈的变化。