Agent推理性能优化：提示词压缩、缓存与并发推理
深入探讨Agent系统的推理性能优化策略，包括提示词压缩技术、智能缓存机制和并发推理管理
概述与动机
在大语言模型驱动的Agent系统中，推理性能是决定系统响应速度和成本的关键因素。随着Agent应用复杂度的增加，单个Agent往往需要进行多次模型推理调用，而多Agent协作场景下的推理调用次数更是呈指数级增长。如何在保证输出质量的前提下，显著提升推理性能并降低成本，成为每个Agent开发者必须面对的核心挑战。
推理性能优化的需求主要来自三个方面：响应速度、成本控制和用户体验。在实时对话、代码生成、数据分析等交互式场景中，用户期望获得秒级甚至毫秒级的响应速度。同时，API调用成本随着Token使用量线性增长，优化Token使用能够显著降低运营成本。此外，在高并发场景下，系统的吞吐量和稳定性也依赖于高效的推理性能。
本文将深入探讨Agent推理性能优化的三大核心技术：提示词压缩、智能缓存和并发推理。我们将从理论基础出发，逐步深入到具体实现细节，提供完整的Python代码示例和架构设计思路，帮助读者构建高性能、低成本的Agent系统。
核心概念与架构设计
推理性能优化框架
Agent推理性能优化是一个系统工程，需要在多个层面进行协同优化。下图展示了完整的推理性能优化框架：
Rendering diagram...
提示词压缩技术
提示词压缩是指在不显著影响输出质量的前提下，减少输入给模型的Token数量。这包括三个层面的压缩：
结构化压缩：通过分析提示词的结构，识别和消除冗余部分。例如，去除重复的指令、合并相似的描述、简化不必要的格式说明。结构化压缩通常不会改变语义，但可以减少10-30%的Token使用。
语义压缩：使用更简洁的表达方式传达相同的语义。这包括使用更准确的词汇、采用更高效的句式结构、避免冗余的描述。语义压缩需要谨慎处理，以避免丢失关键信息。
层次化压缩：根据信息的重要性进行分级压缩，保留核心信息，压缩次要信息。对于多轮对话场景，可以保留最近几轮的完整对话，而将更早的对话压缩为摘要。
智能缓存机制
缓存是提升推理性能最有效的手段之一。智能缓存通过存储和重用之前的推理结果，避免重复计算，显著提升响应速度和降低成本。
语义缓存：基于语义相似度的缓存策略。即使输入不完全相同，如果语义相似度超过阈值，就可以返回缓存结果。这需要结合向量数据库和语义相似度算法。
结果缓存：基于精确匹配的缓存策略。对于完全相同的输入，直接返回缓存结果。这种策略适用于确定性任务和重复性查询。
混合策略：结合语义缓存和结果缓存的优点，在不同场景下选择最适合的缓存策略。例如，对于代码生成任务可以使用精确匹配缓存，而对于对话任务可以使用语义缓存。
并发推理管理
并发推理通过并行处理多个推理请求，提升系统的整体吞吐量和资源利用率。
批量请求：将多个相似的推理请求合并为一个批量请求，减少网络开销和API调用次数。批量推理特别适用于需要处理大量相似任务的场景。
异步推理：使用异步编程模型处理推理请求，避免阻塞主线程，提升系统的并发处理能力。异步推理适合需要同时处理多个独立任务的场景。
模型选择：根据任务复杂度和性能要求，选择合适的模型。简单任务可以使用较小的模型，而复杂任务使用较大的模型。这可以在保证质量的同时优化成本和性能。
关键技术实现
智能缓存系统实现
下面是一个完整的智能缓存系统实现，包含语义缓存、结果缓存和缓存失效策略：
import hashlib
import json
import time
from typing import Dict, List, Optional, Any, Callable
from dataclasses import dataclass, field
from collections import OrderedDict
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity
from sentence_transformers import SentenceTransformer
@dataclass
class CacheEntry:
"""缓存条目"""
key: str
value: str
timestamp: float
access_count: int = 0
embedding: Optional[np.ndarray] = None
class IntelligentCache:
"""智能缓存系统"""
def __init__(
self,
max_size: int = 1000,
semantic_threshold: float = 0.85,
ttl_seconds: int = 3600,
embedding_model: str = 'all-MiniLM-L6-v2'
):
"""
初始化智能缓存系统
Args:
max_size: 最大缓存条目数
semantic_threshold: 语义相似度阈值
ttl_seconds: 缓存过期时间(秒)
embedding_model: 嵌入模型名称
"""
self.max_size = max_size
self.semantic_threshold = semantic_threshold
self.ttl_seconds = ttl_seconds
self.embedding_model = SentenceTransformer(embedding_model)
# 结果缓存(精确匹配)
self.result_cache: OrderedDict[str, CacheEntry] = OrderedDict()
# 语义缓存(向量检索)
self.semantic_cache: List[CacheEntry] = []
# 统计信息
self.stats = {
'result_hits': 0,
'result_misses': 0,
'semantic_hits': 0,
'semantic_misses': 0,
'total_requests': 0
}
def _generate_key(self, prompt: str, model: str, **kwargs) -> str:
"""生成缓存键"""
cache_data = {
'prompt': prompt,
'model': model,
'kwargs': kwargs
}
data_str = json.dumps(cache_data, sort_keys=True)
return hashlib.sha256(data_str.encode()).hexdigest()
def _get_embedding(self, text: str) -> np.ndarray:
"""获取文本嵌入"""
return self.embedding_model.encode(text, convert_to_numpy=True)
def _is_expired(self, entry: CacheEntry) -> bool:
"""检查缓存是否过期"""
return time.time() - entry.timestamp > self.ttl_seconds
def _cleanup_expired(self):
"""清理过期缓存"""
# 清理结果缓存
expired_keys = [
key for key, entry in self.result_cache.items()
if self._is_expired(entry)
]
for key in expired_keys:
del self.result_cache[key]
# 清理语义缓存
self.semantic_cache = [
entry for entry in self.semantic_cache
if not self._is_expired(entry)
]
def _evict_lru(self):
"""LRU缓存淘汰"""
if len(self.result_cache) >= self.max_size:
# 淘汰最少使用的条目
lru_key = min(
self.result_cache.keys(),
key=lambda k: self.result_cache[k].access_count
)
del self.result_cache[lru_key]
def get(self, prompt: str, model: str, use_semantic: bool = True, **kwargs) -> Optional[str]:
"""
获取缓存结果
Args:
prompt: 输入提示词
model: 模型名称
use_semantic: 是否使用语义缓存
**kwargs: 其他参数
Returns:
缓存结果或None
"""
self.stats['total_requests'] += 1
self._cleanup_expired()
# 1. 尝试结果缓存(精确匹配)
key = self._generate_key(prompt, model, **kwargs)
if key in self.result_cache:
entry = self.result_cache[key]
if not self._is_expired(entry):
entry.access_count += 1
self.result_cache.move_to_end(key)
self.stats['result_hits'] += 1
return entry.value
self.stats['result_misses'] += 1
# 2. 尝试语义缓存
if use_semantic and self.semantic_cache:
prompt_embedding = self._get_embedding(prompt)
# 寻找语义相似的缓存
for entry in self.semantic_cache:
if entry.embedding is not None:
similarity = cosine_similarity(
[prompt_embedding],
[entry.embedding]
)[0][0]
if similarity >= self.semantic_threshold:
if not self._is_expired(entry):
entry.access_count += 1
self.stats['semantic_hits'] += 1
return entry.value
self.stats['semantic_misses'] += 1
return None
def set(self, prompt: str, response: str, model: str, **kwargs):
"""
设置缓存
Args:
prompt: 输入提示词
response: 模型响应
model: 模型名称
**kwargs: 其他参数
"""
# 设置结果缓存
key = self._generate_key(prompt, model, **kwargs)
self._evict_lru()
cache_entry = CacheEntry(
key=key,
value=response,
timestamp=time.time(),
embedding=self._get_embedding(prompt)
)
self.result_cache[key] = cache_entry
# 添加到语义缓存
self.semantic_cache.append(cache_entry)
# 限制语义缓存大小
if len(self.semantic_cache) > self.max_size:
self.semantic_cache = self.semantic_cache[-self.max_size:]
def clear(self):
"""清空所有缓存"""
self.result_cache.clear()
self.semantic_cache.clear()
def get_stats(self) -> Dict[str, Any]:
"""获取缓存统计信息"""
total_cache_hits = self.stats['result_hits'] + self.stats['semantic_hits']
hit_rate = total_cache_hits / self.stats['total_requests'] if self.stats['total_requests'] > 0 else 0
return {
**self.stats,
'hit_rate': hit_rate,
'result_cache_size': len(self.result_cache),
'semantic_cache_size': len(self.semantic_cache),
'total_cache_size': len(self.result_cache) + len(self.semantic_cache)
}
# 使用示例
def demonstrate_intelligent_cache():
"""演示智能缓存系统的使用"""
print("=== 智能缓存系统演示 ===\n")
# 初始化缓存系统
cache = IntelligentCache(
max_size=100,
semantic_threshold=0.85,
ttl_seconds=3600
)
# 模拟推理函数
def mock_llm_inference(prompt: str, model: str = "gpt-4") -> str:
"""模拟LLM推理"""
return f"Response to: {prompt}"
# 测试用例
test_cases = [
("什么是人工智能？", "gpt-4"),
("AI是什么意思？", "gpt-4"), # 语义相似
("Python如何处理异常？", "gpt-4"),
("什么是人工智能？", "gpt-4"), # 精确匹配
("AI的定义是什么？", "gpt-4"), # 语义相似
("Python异常处理方法", "gpt-4"), # 语义相似
]
print("测试缓存功能：")
for i, (prompt, model) in enumerate(test_cases, 1):
# 尝试从缓存获取
cached_result = cache.get(prompt, model)
if cached_result:
print(f"{i}. 缓存命中: '{prompt}'")
print(f" 缓存结果: {cached_result}")
else:
print(f"{i}. 缓存未命中，调用模型: '{prompt}'")
# 调用模型获取结果
result = mock_llm_inference(prompt, model)
# 存入缓存
cache.set(prompt, result, model)
print(f" 模型结果: {result}")
# 显示缓存统计
print("\n缓存统计信息：")
stats = cache.get_stats()
for key, value in stats.items():
if isinstance(value, float):
print(f"{key}: {value:.4f}")
else:
print(f"{key}: {value}")
if __name__ == "__main__":
demonstrate_intelligent_cache()
并发推理管理器实现
下面是一个完整的并发推理管理器实现，支持批量推理、异步推理和智能模型选择：
import asyncio
import time
from typing import List, Dict, Any, Optional, Callable, Awaitable
from dataclasses import dataclass
from enum import Enum
from concurrent.futures import ThreadPoolExecutor
import aiohttp
import numpy as np
class ModelComplexity(Enum):
"""模型复杂度级别"""
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
@dataclass
class ModelConfig:
"""模型配置"""
name: str
complexity: ModelComplexity
cost_per_1k_tokens: float
avg_latency_ms: int
quality_score: float # 0-1
max_tokens: int
# 预定义模型配置
MODEL_CONFIGS = {
"gpt-3.5-turbo": ModelConfig(
name="gpt-3.5-turbo",
complexity=ModelComplexity.LOW,
cost_per_1k_tokens=0.002,
avg_latency_ms=500,
quality_score=0.8,
max_tokens=4096
),
"gpt-4": ModelConfig(
name="gpt-4",
complexity=ModelComplexity.HIGH,
cost_per_1k_tokens=0.03,
avg_latency_ms=1500,
quality_score=0.95,
max_tokens=8192
),
"claude-3-sonnet": ModelConfig(
name="claude-3-sonnet",
complexity=ModelComplexity.MEDIUM,
cost_per_1k_tokens=0.015,
avg_latency_ms=1000,
quality_score=0.9,
max_tokens=6144
)
}
@dataclass
class InferenceRequest:
"""推理请求"""
prompt: str
model: Optional[str] = None
max_tokens: int = 1000
temperature: float = 0.7
metadata: Dict[str, Any] = None
callback: Optional[Callable[[str], None]] = None
@dataclass
class InferenceResult:
"""推理结果"""
request: InferenceRequest
response: str
model: str
tokens_used: int
latency_ms: int
cost: float
timestamp: float
from_cache: bool = False
class ConcurrentInferenceManager:
"""并发推理管理器"""
def __init__(
self,
default_model: str = "gpt-3.5-turbo",
max_concurrent_requests: int = 10,
enable_batching: bool = True,
batch_size: int = 5,
batch_timeout_ms: int = 100
):
"""
初始化并发推理管理器
Args:
default_model: 默认模型名称
max_concurrent_requests: 最大并发请求数
enable_batching: 是否启用批量推理
batch_size: 批量大小
batch_timeout_ms: 批量超时时间(毫秒)
"""
self.default_model = default_model
self.max_concurrent_requests = max_concurrent_requests
self.enable_batching = enable_batching
self.batch_size = batch_size
self.batch_timeout_ms = batch_timeout_ms
# 并发控制
self.semaphore = asyncio.Semaphore(max_concurrent_requests)
self.executor = ThreadPoolExecutor(max_workers=max_concurrent_requests)
# 批量处理队列
self.batch_queue: List[InferenceRequest] = []
self.batch_event = asyncio.Event()
# 统计信息
self.stats = {
'total_requests': 0,
'successful_requests': 0,
'failed_requests': 0,
'total_tokens': 0,
'total_cost': 0.0,
'total_latency_ms': 0,
'batched_requests': 0,
'model_usage': {}
}
def _estimate_complexity(self, prompt: str) -> ModelComplexity:
"""估计任务复杂度"""
# 简单启发式方法：根据提示词长度和复杂关键词估计
complexity_keywords = {
'high': ['设计', '架构', '复杂', '高级', '优化', '算法'],
'medium': ['实现', '开发', '测试', '部署', '分析']
}
prompt_lower = prompt.lower()
# 检查关键词
for level, keywords in complexity_keywords.items():
if any(keyword in prompt_lower for keyword in keywords):
return ModelComplexity(level)
# 默认基于长度
if len(prompt) > 1000:
return ModelComplexity.HIGH
elif len(prompt) > 500:
return ModelComplexity.MEDIUM
else:
return ModelComplexity.LOW
def _select_model(self, prompt: str, required_model: Optional[str] = None) -> str:
"""
智能模型选择
Args:
prompt: 输入提示词
required_model: 用户指定的模型
Returns:
选择的模型名称
"""
if required_model and required_model in MODEL_CONFIGS:
return required_model
# 根据复杂度选择模型
complexity = self._estimate_complexity(prompt)
if complexity == ModelComplexity.HIGH:
return "gpt-4"
elif complexity == ModelComplexity.MEDIUM:
return "claude-3-sonnet"
else:
return "gpt-3.5-turbo"
async def _call_llm_api(self, request: InferenceRequest, model: str) -> InferenceResult:
"""
调用LLM API
Args:
request: 推理请求
model: 模型名称
Returns:
推理结果
"""
# 这里模拟API调用，实际使用时替换为真实的API调用
await asyncio.sleep(MODEL_CONFIGS[model].avg_latency_ms / 1000.0)
# 模拟响应
response = f"Simulated response for: {request.prompt[:50]}..."
# 计算Token和成本
estimated_input_tokens = len(request.prompt) // 4
estimated_output_tokens = len(response) // 4
total_tokens = estimated_input_tokens + estimated_output_tokens
cost = (total_tokens / 1000) * MODEL_CONFIGS[model].cost_per_1k_tokens
return InferenceResult(
request=request,
response=response,
model=model,
tokens_used=total_tokens,
latency_ms=MODEL_CONFIGS[model].avg_latency_ms,
cost=cost,
timestamp=time.time()
)
async def _process_single_request(self, request: InferenceRequest) -> InferenceResult:
"""处理单个推理请求"""
async with self.semaphore:
start_time = time.time()
try:
# 智能模型选择
model = self._select_model(request.prompt, request.model)
# 调用API
result = await self._call_llm_api(request, model)
# 更新统计信息
self.stats['successful_requests'] += 1
self.stats['total_tokens'] += result.tokens_used
self.stats['total_cost'] += result.cost
self.stats['total_latency_ms'] += result.latency_ms
# 更新模型使用统计
if model not in self.stats['model_usage']:
self.stats['model_usage'][model] = 0
self.stats['model_usage'][model] += 1
return result
except Exception as e:
self.stats['failed_requests'] += 1
raise e
async def _process_batch(self, requests: List[InferenceRequest]) -> List[InferenceResult]:
"""处理批量推理请求"""
results = []
# 并行处理批量请求
tasks = [self._process_single_request(req) for req in requests]
results = await asyncio.gather(*tasks, return_exceptions=True)
# 处理异常
processed_results = []
for result in results:
if isinstance(result, Exception):
print(f"批量处理中的异常: {result}")
else:
processed_results.append(result)
self.stats['batched_requests'] += 1
return processed_results
async def inference(self, request: InferenceRequest) -> InferenceResult:
"""
执行推理
Args:
request: 推理请求
Returns:
推理结果
"""
self.stats['total_requests'] += 1
if self.enable_batching:
# 添加到批量队列
self.batch_queue.append(request)
# 如果队列达到批量大小，触发批量处理
if len(self.batch_queue) >= self.batch_size:
self.batch_event.set()
# 等待批量处理完成
await self.batch_event.wait()
self.batch_event.clear()
# 返回结果（简化处理，实际需要更复杂的结果匹配）
# 这里为演示目的，返回单个请求的结果
return await self._process_single_request(request)
else:
# 等待批量超时
try:
await asyncio.wait_for(
self.batch_event.wait(),
timeout=self.batch_timeout_ms / 1000.0
)
except asyncio.TimeoutError:
# 超时后处理当前请求
return await self._process_single_request(request)
else:
# 直接处理单个请求
return await self._process_single_request(request)
async def batch_inference(self, requests: List[InferenceRequest]) -> List[InferenceResult]:
"""
批量推理
Args:
requests: 推理请求列表
Returns:
推理结果列表
"""
self.stats['total_requests'] += len(requests)
return await self._process_batch(requests)
def get_stats(self) -> Dict[str, Any]:
"""获取统计信息"""
avg_latency = (
self.stats['total_latency_ms'] / self.stats['successful_requests']
if self.stats['successful_requests'] > 0 else 0
)
return {
**self.stats,
'avg_latency_ms': avg_latency,
'success_rate': (
self.stats['successful_requests'] / self.stats['total_requests']
if self.stats['total_requests'] > 0 else 0
)
}
# 使用示例
async def demonstrate_concurrent_inference():
"""演示并发推理管理器的使用"""
print("=== 并发推理管理器演示 ===\n")
# 初始化推理管理器
manager = ConcurrentInferenceManager(
default_model="gpt-3.5-turbo",
max_concurrent_requests=5,
enable_batching=True,
batch_size=3
)
# 创建测试请求
test_prompts = [
"写一个Python函数来计算斐波那契数列",
"解释什么是机器学习",
"设计一个微服务架构",
"如何优化SQL查询性能",
"实现一个简单的Web服务器",
"分析用户行为数据",
"设计一个推荐系统算法",
"处理大数据的最佳实践"
]
requests = [
InferenceRequest(prompt=prompt)
for prompt in test_prompts
]
print(f"开始处理 {len(requests)} 个推理请求...\n")
start_time = time.time()
# 并发推理
tasks = [manager.inference(req) for req in requests]
results = await asyncio.gather(*tasks)
end_time = time.time()
# 显示结果
print("推理结果：")
for i, result in enumerate(results, 1):
print(f"\n{i}. 模型: {result.model}")
print(f" 提示词: {result.request.prompt[:50]}...")
print(f" 响应: {result.response[:50]}...")
print(f" Token使用: {result.tokens_used}")
print(f" 延迟: {result.latency_ms}ms")
print(f" 成本: ${result.cost:.4f}")
# 显示统计信息
print(f"\n总处理时间: {end_time - start_time:.2f}秒")
print("\n推理统计信息：")
stats = manager.get_stats()
for key, value in stats.items():
if isinstance(value, float):
print(f"{key}: {value:.4f}")
elif isinstance(value, dict):
print(f"{key}:")
for model, count in value.items():
print(f" {model}: {count}")
else:
print(f"{key}: {value}")
if __name__ == "__main__":
asyncio.run(demonstrate_concurrent_inference())
提示词压缩工具实现
下面是一个智能提示词压缩工具的实现，包含结构化压缩、语义压缩和层次化压缩：
import re
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass
import tiktoken
@dataclass
class CompressionStats:
"""压缩统计信息"""
original_length: int
compressed_length: int
compression_ratio: float
compression_time_ms: int
techniques_used: List[str]
class PromptCompressor:
"""提示词压缩工具"""
def __init__(self, target_language: str = "zh-CN"):
"""
初始化提示词压缩器
Args:
target_language: 目标语言代码
"""
self.target_language = target_language
self.encoding = tiktoken.encoding_for_model("gpt-4")
# 常见的冗余模式
self.redundant_patterns = [
(r'请[你]?帮我?[来]?分析[一-下]?', '分析'),
(r'希望[你]?能够?[来]?提供', '提供'),
(r'请[你]?详细地?(描述|说明|解释)', r'\1'),
(r'(需要|要求|希望)[你]?能够?', ''),
(r'我需要[你]?帮我?', ''),
(r'能否[你]?请?', '请'),
(r'麻烦[你]?请?', '请'),
]
# 结构化指令模板
self.instruction_templates = {
'code_generation': [
'编写代码',
'实现功能',
'开发程序'
],
'code_explanation': [
'解释代码',
'说明原理',
'描述功能'
],
'problem_solving': [
'解决问题',
'分析问题',
'处理情况'
]
}
def count_tokens(self, text: str) -> int:
"""计算Token数量"""
return len(self.encoding.encode(text))
def _structural_compression(self, prompt: str) -> Tuple[str, List[str]]:
"""
结构化压缩
Args:
prompt: 原始提示词
Returns:
压缩后的提示词和使用的压缩技术
"""
compressed = prompt
techniques_used = []
# 1. 移除多余空白
original = compressed
compressed = re.sub(r'\s+', ' ', compressed).strip()
if compressed != original:
techniques_used.append('whitespace_normalization')
# 2. 应用冗余模式替换
for pattern, replacement in self.redundant_patterns:
original = compressed
compressed = re.sub(pattern, replacement, compressed)
if compressed != original:
techniques_used.append('pattern_replacement')
# 3. 移除重复短语
original = compressed
sentences = re.split(r'[。！？]', compressed)
unique_sentences = []
seen = set()
for sentence in sentences:
if sentence.strip():
normalized = sentence.strip().lower()
if normalized not in seen:
seen.add(normalized)
unique_sentences.append(sentence.strip())
compressed = '。'.join(unique_sentences) + '。'
if len(compressed) < len(original):
techniques_used.append('duplicate_removal')
return compressed, techniques_used
def _semantic_compression(self, prompt: str) -> Tuple[str, List[str]]:
"""
语义压缩
Args:
prompt: 原始提示词
Returns:
压缩后的提示词和使用的压缩技术
"""
compressed = prompt
techniques_used = []
# 1. 使用更简洁的表达
replacements = {
'非常详细地': '详细',
'极其重要': '重要',
'非常关键': '关键',
'需要进行': '需',
'实现一个': '实现',
'创建一个': '创建',
'生成一个': '生成',
'编写一个': '编写',
'设计一个': '设计',
}
original = compressed
for old, new in replacements.items():
compressed = compressed.replace(old, new)
if compressed != original:
techniques_used.append('concise_expression')
# 2. 简化句式结构
original = compressed
# 将"请A，然后B"简化为"A，B"
compressed = re.sub(r'请(.+?)，然后', r'\1，', compressed)
if compressed != original:
techniques_used.append('sentence_simplification')
# 3. 移除礼貌用语
original = compressed
courtesy_words = ['请', '麻烦', '劳驾', '谢谢', '感谢']
for word in courtesy_words:
# 只在特定上下文中移除
compressed = re.sub(rf'^{word}[，。]?', '', compressed)
if compressed != original:
techniques_used.append('courtesy_removal')
return compressed, techniques_used
def _hierarchical_compression(self, prompt: str, max_tokens: int) -> Tuple[str, List[str]]:
"""
层次化压缩
Args:
prompt: 原始提示词
max_tokens: 目标最大Token数
Returns:
压缩后的提示词和使用的压缩技术
"""
compressed = prompt
techniques_used = []
# 如果Token数量已满足要求，直接返回
if self.count_tokens(compressed) <= max_tokens:
return compressed, techniques_used
# 1. 保留核心指令，压缩说明性文字
original = compressed
# 识别核心指令（通常是第一句话或包含动词的句子）
sentences = re.split(r'[。！？]', compressed)
if sentences:
# 保留第一个完整句子作为核心指令
core_instruction = sentences[0].strip()
# 将剩余部分压缩为摘要
if len(sentences) > 1:
remaining = sentences[1:]
# 简单的摘要策略：保留每个句子的主语和谓语
summary_parts = []
for sentence in remaining:
if sentence.strip():
# 提取关键词
words = sentence.strip().split()
if words:
summary_parts.append(words[0]) # 保留第一个词
if summary_parts:
compressed = f"{core_instruction}。{','.join(summary_parts[:3])}。"
techniques_used.append('hierarchical_summarization')
# 2. 如果仍然超过限制，进行截断
if self.count_tokens(compressed) > max_tokens:
original = compressed
# 逐步截断直到满足要求
while self.count_tokens(compressed) > max_tokens and len(compressed) > 50:
compressed = compressed[:-10] # 每次删除10个字符
if len(compressed) < len(original):
techniques_used.append('truncation')
return compressed, techniques_used
def compress(
self,
prompt: str,
target_tokens: Optional[int] = None,
enable_structural: bool = True,
enable_semantic: bool = True,
enable_hierarchical: bool = True
) -> Tuple[str, CompressionStats]:
"""
执行压缩
Args:
prompt: 原始提示词
target_tokens: 目标Token数
enable_structural: 是否启用结构化压缩
enable_semantic: 是否启用语义压缩
enable_hierarchical: 是否启用层次化压缩
Returns:
压缩后的提示词和压缩统计
"""
import time
start_time = time.time()
original_length = self.count_tokens(prompt)
compressed = prompt
all_techniques_used = []
# 1. 结构化压缩
if enable_structural:
compressed, techniques = self._structural_compression(compressed)
all_techniques_used.extend(techniques)
# 2. 语义压缩
if enable_semantic:
compressed, techniques = self._semantic_compression(compressed)
all_techniques_used.extend(techniques)
# 3. 层次化压缩
if enable_hierarchical and target_tokens:
compressed, techniques = self._hierarchical_compression(
compressed, target_tokens
)
all_techniques_used.extend(techniques)
compressed_length = self.count_tokens(compressed)
compression_ratio = (original_length - compressed_length) / original_length
compression_time = int((time.time() - start_time) * 1000)
stats = CompressionStats(
original_length=original_length,
compressed_length=compressed_length,
compression_ratio=compression_ratio,
compression_time_ms=compression_time,
techniques_used=all_techniques_used
)
return compressed, stats
def batch_compress(
self,
prompts: List[str],
target_tokens: Optional[int] = None
) -> List[Tuple[str, CompressionStats]]:
"""
批量压缩
Args:
prompts: 提示词列表
target_tokens: 目标Token数
Returns:
压缩结果列表
"""
return [
self.compress(prompt, target_tokens)
for prompt in prompts
]
# 使用示例
def demonstrate_prompt_compression():
"""演示提示词压缩工具的使用"""
print("=== 提示词压缩工具演示 ===\n")
compressor = PromptCompressor()
# 测试用例
test_prompts = [
"请你帮我详细地分析一下Python中的异常处理机制，并且希望你能提供一些实际的代码示例来演示不同类型的异常处理方法。",
"我需要你帮我实现一个功能，请编写一个Python函数，这个函数需要能够处理用户输入的数据，并进行相应的验证和错误处理。",
"希望你能详细地解释什么是机器学习，包括其基本概念、主要算法类型以及在实际应用中的注意事项。",
"麻烦你请帮我设计一个微服务架构，这个架构需要考虑到系统的可扩展性、高可用性以及容错能力。"
]
print("单次压缩示例：")
print("=" * 80)
for i, prompt in enumerate(test_prompts, 1):
print(f"\n示例 {i}:")
print(f"原始提示词 ({compressor.count_tokens(prompt)} tokens):")
print(f"{prompt}\n")
# 执行压缩
compressed, stats = compressor.compress(
prompt,
target_tokens=100, # 目标压缩到100 tokens
enable_structural=True,
enable_semantic=True,
enable_hierarchical=True
)
print(f"压缩后提示词 ({compressor.count_tokens(compressed)} tokens):")
print(f"{compressed}\n")
print("压缩统计:")
print(f" 原始长度: {stats.original_length} tokens")
print(f" 压缩后长度: {stats.compressed_length} tokens")
print(f" 压缩率: {stats.compression_ratio:.2%}")
print(f" 压缩时间: {stats.compression_time_ms}ms")
print(f" 使用的技术: {', '.join(stats.techniques_used)}")
print("\n" + "=" * 80)
print("批量压缩示例：")
# 批量压缩
batch_results = compressor.batch_compress(
test_prompts[:2], # 只压缩前两个示例
target_tokens=80
)
print(f"\n批量压缩 {len(batch_results)} 个提示词:")
total_original = 0
total_compressed = 0
for i, (compressed, stats) in enumerate(batch_results, 1):
total_original += stats.original_length
total_compressed += stats.compressed_length
print(f"{i}. {stats.original_length} -> {stats.compressed_length} tokens "
f"({stats.compression_ratio:.2%})")
print(f"\n总计: {total_original} -> {total_compressed} tokens "
f"({(total_original-total_compressed)/total_original:.2%})")
if __name__ == "__main__":
demonstrate_prompt_compression()
最佳实践与常见陷阱
缓存系统最佳实践
合理设置缓存策略：不同类型的任务需要不同的缓存策略。对于代码生成、数据查询等确定性任务，使用精确匹配缓存效果最佳；对于对话、创意写作等开放性任务，语义缓存更适合。要避免过度缓存，以免返回过时的结果。
智能失效机制：缓存失效是保证数据一致性的关键。实现基于时间的TTL(Time To Live)机制，定期清理过期缓存。对于依赖外部数据的场景，可以实现基于事件驱动失效，当数据更新时主动失效相关缓存。
缓存预热：在系统启动时预加载高频访问的请求和响应，减少用户首次访问的延迟。缓存预热要结合实际业务场景，优先加载核心功能的缓存数据。
监控缓存效果：建立完善的缓存监控体系，跟踪缓存命中率、响应时间、存储占用等指标。定期分析缓存模式，优化缓存策略和参数配置。
并发推理最佳实践
合理设置并发度：根据系统资源和API限制设置合适的并发度。过高的并发度可能导致资源耗尽或API限流，过低的并发度无法充分利用系统性能。建议通过基准测试确定最优并发度。
批量请求优化：批量请求可以显著提升吞吐量，但要注意批量大小和超时设置的平衡。过大的批量会增加内存占用和失败重试成本，建议通过测试确定最优批量大小。
错误处理和重试：并发环境中错误率会更高，需要实现完善的错误处理和重试机制。使用指数退避算法避免雪崩效应，设置合理的重试次数和超时时间。
资源监控和限流：实时监控系统资源使用情况，实现动态限流机制。当系统负载过高时，自动降低并发度或延迟部分请求，保证系统稳定性。
提示词压缩最佳实践
保留核心语义：压缩时要注意保留提示词的核心语义和关键信息。过度压缩可能导致模型理解偏差，影响输出质量。建议在压缩后进行质量验证，确保压缩后的提示词仍能传达原始意图。
分级压缩策略：根据不同场景使用不同的压缩级别。对于重要任务使用保守压缩，确保质量；对于批量处理可以使用激进压缩，提升效率。
压缩效果评估：建立压缩效果评估机制，定期测试压缩后的提示词质量。可以对比压缩前后的输出质量，确保压缩不会显著影响结果。
结合缓存优化：将提示词压缩与缓存策略结合使用。对压缩后的提示词建立缓存索引，进一步提升系统性能。
常见陷阱及解决方案
缓存穿透：恶意或错误请求绕过缓存直接访问后端，导致系统压力过大。解决方案包括：布隆过滤器预检测、对空结果缓存、请求参数验证等。
缓存雪崩：大量缓存同时失效，导致后端服务压力激增。解决方案包括：设置不同的TTL、缓存预热、多级缓存架构等。
并发死锁：不当的并发控制可能导致死锁或活锁。解决方案包括：避免循环等待、设置超时、使用并发安全的数据结构等。
压缩质量下降：过度压缩导致模型理解偏差。解决方案包括：建立质量监控、设置压缩阈值、人工审核关键任务的压缩结果。
性能优化考虑
性能基准测试
建立完善的性能基准测试体系，定期评估优化效果。关键指标包括：
- 响应时间：P50、P95、P99延迟指标
- 吞吐量：每秒处理请求数(RPS)
- 资源利用率：CPU、内存、网络使用率
- 成本效益：单位请求的成本
import time
import statistics
from typing import List, Dict, Any
import matplotlib.pyplot as plt
class PerformanceBenchmark:
"""性能基准测试工具"""
def __init__(self):
self.results = []
self.metrics = {}
def run_benchmark(
self,
test_function: callable,
test_cases: List[Dict[str, Any]],
iterations: int = 10
) -> Dict[str, Any]:
"""
运行基准测试
Args:
test_function: 测试函数
test_cases: 测试用例列表
iterations: 迭代次数
Returns:
测试结果
"""
results = []
for test_case in test_cases:
case_results = []
for _ in range(iterations):
start_time = time.time()
try:
result = test_function(**test_case)
end_time = time.time()
case_results.append({
'latency_ms': (end_time - start_time) * 1000,
'success': True,
'result': result
})
except Exception as e:
end_time = time.time()
case_results.append({
'latency_ms': (end_time - start_time) * 1000,
'success': False,
'error': str(e)
})
# 计算统计数据
latencies = [
r['latency_ms'] for r in case_results if r['success']
]
success_rate = sum(
1 for r in case_results if r['success']
) / len(case_results)
case_stats = {
'test_case': test_case,
'avg_latency_ms': statistics.mean(latencies) if latencies else 0,
'p50_latency_ms': statistics.median(latencies) if latencies else 0,
'p95_latency_ms': statistics.quantiles(latencies, n=20)[18] if len(latencies) > 20 else 0,
'p99_latency_ms': statistics.quantiles(latencies, n=100)[98] if len(latencies) > 100 else 0,
'success_rate': success_rate,
'total_requests': len(case_results),
'successful_requests': len(latencies)
}
results.append(case_stats)
self.results = results
return self._calculate_summary_stats(results)
def _calculate_summary_stats(self, results: List[Dict[str, Any]]) -> Dict[str, Any]:
"""计算汇总统计"""
total_requests = sum(r['total_requests'] for r in results)
total_successful = sum(r['successful_requests'] for r in results)
all_latencies = []
for r in results:
# 重新计算所有成功的延迟
for _ in range(r['successful_requests']):
all_latencies.append(r['avg_latency_ms'])
return {
'total_requests': total_requests,
'total_successful': total_successful,
'overall_success_rate': total_successful / total_requests if total_requests > 0 else 0,
'avg_latency_ms': statistics.mean(all_latencies) if all_latencies else 0,
'p50_latency_ms': statistics.median(all_latencies) if all_latencies else 0,
'throughput_rps': total_successful / (sum(r['avg_latency_ms'] for r in results) / 1000) if results else 0,
'case_details': results
}
def compare_optimizations(
self,
baseline_results: Dict[str, Any],
optimized_results: Dict[str, Any]
) -> Dict[str, Any]:
"""
比较优化前后的性能
Args:
baseline_results: 基线测试结果
optimized_results: 优化后测试结果
Returns:
比较结果
"""
comparison = {
'latency_improvement': 0,
'throughput_improvement': 0,
'cost_saving': 0
}
if baseline_results['avg_latency_ms'] > 0:
comparison['latency_improvement'] = (
(baseline_results['avg_latency_ms'] - optimized_results['avg_latency_ms']) /
baseline_results['avg_latency_ms']
) * 100
if baseline_results['throughput_rps'] > 0:
comparison['throughput_improvement'] = (
(optimized_results['throughput_rps'] - baseline_results['throughput_rps']) /
baseline_results['throughput_rps']
) * 100
return comparison
监控和调优策略
实时监控：建立实时监控系统，跟踪关键性能指标。使用Prometheus + Grafana等工具构建可视化监控面板，及时发现性能问题。
自动调优：基于监控数据实现自动调优机制。当检测到性能下降时，自动调整参数或切换优化策略。
A/B测试：对于重大的优化变更，进行A/B测试验证效果。将流量分为对照组和实验组，比较优化前后的性能差异。
性能预算：为不同的服务设置性能预算，确保优化工作有明确的目标和约束。性能预算包括响应时间、资源使用、成本等多个维度。
成本优化策略
Token使用优化：通过提示词压缩和智能缓存减少Token使用量。建立Token使用监控系统，识别高成本的任务和用户。
模型选择优化：根据任务复杂度选择合适的模型，避免过度使用高成本模型。实现动态模型选择，在保证质量的前提下降低成本。
批量处理优化：将相似的任务批量处理，减少API调用次数。批量处理要平衡性能和成本，找到最优的批量大小。
资源调度优化：实现智能资源调度，在低峰期处理批量任务，在高峰期优先处理实时任务。利用云服务的弹性伸缩能力，降低资源成本。
参考资源
官方文档
- OpenAI API Documentation: https://platform.openai.com/docs/api-reference - OpenAI API官方文档，包含详细的性能优化建议和最佳实践
- LangChain Documentation: https://python.langchain.com/docs/ - LangChain框架文档，提供了缓存、批处理等优化功能的实现指南
- Anthropic API Documentation: https://docs.anthropic.com/ - Anthropic Claude API文档，包含模型选择和请求优化的建议
技术论文和文章
- "Semantic Caching for Language Models": 语义缓存的学术论文，讨论了基于向量相似度的缓存策略和算法
- "Prompt Engineering Guide": 提示词工程指南，详细介绍了提示词优化和压缩的技术
- "Scaling Laws for Neural Language Models": 神经语言模型的缩放定律，为模型选择提供了理论基础
开源工具和库
- LangChain: https://github.com/langchain-ai/langchain - 功能强大的LLM应用开发框架，内置缓存和批处理功能
- Semantic Cache: https://github.com/retrieve/semantic-cache - 专门的语义缓存库，支持多种向量数据库
- Prompt Optimizer: https://github.com/microsoft/promptoptimizer - 微软开源的提示词优化工具
- Redis AI: https://github.com/RedisAI/RedisAI - Redis的AI扩展，可以作为高性能缓存和模型推理的统一平台
实战案例
- ChatGPT Optimization: OpenAI关于ChatGPT性能优化的技术博客，分享了实际的生产环境优化经验
- Claude Performance: Anthropic关于Claude模型性能优化的最佳实践
- Enterprise LLM Deployment: 企业级LLM部署的白皮书，涵盖了缓存、批处理、模型选择等多个优化维度
通过本文的学习，读者应该能够掌握Agent推理性能优化的核心技术和实现方法，能够在实际项目中应用这些优化策略，构建高性能、低成本的Agent系统。下一篇文章将深入探讨上下文管理和Token成本优化的技术细节。