Redis

引言 LLM推理是昂贵的——每次调用消耗Token、产生延迟、花费金钱。在Agent系统中，大量请求其实是重复的或高度相似的。缓存是解决这个问题的最有效手段。 一个设计良好的缓存架构可以将LLM调用减少60-80%，将响应延迟降低一个数量级，将运行成本压缩到原来的几分之一。2026年，缓存已经成为Agent系统的标配组件。 一、Agent缓存的多层模型 1.1 L1：响应缓存 缓存完整请求-响应对。当完全相同的请求再次出现时，直接返回缓存结果。 Request: "总结这篇文章" + article_content (hash: a1b2c3) Cache Hit → Return cached summary Cache Miss → Call LLM → Cache result → Return 适用场景：用户重复提问、模板化任务。 注意事项：必须对请求进行标准化处理——“总结这篇"和"帮我总结这篇"应该命中同一缓存。使用请求的语义哈希而非原始字符串作为缓存键。 1.2 L2：前缀缓存 LLM推理中，请求的前缀部分如果与之前请求相同，可以复用已计算的KV Cache。这在多轮对话中特别有效。 对话第1轮: [System Prompt + User Msg 1] → 生成回复1 对话第2轮: [System Prompt + User Msg 1 + Reply 1 + User Msg 2] ↑ 前缀相同，可复用KV Cache 2026年主流推理框架（vLLM、SGLang）都已支持前缀缓存，命中率通常在70%以上。 1.3 L3：语义缓存 即使请求不完全相同，只要语义相似就命中缓存。通过embedding计算请求的向量表示，与缓存中的请求向量比较相似度。 def semantic_cache_lookup(query, cache, threshold=0.95): query_embedding = embed(query) for cached_query, cached_response, cached_embedding in cache: similarity = cosine_similarity(query_embedding, cached_embedding) if similarity > threshold: return cached_response return None 适用场景：用户用不同措辞问同一个问题。 ...

引言 Agent的状态管理是系统设计中最容易被忽视却又最关键的环节。一个Agent在执行任务时，可能经历"理解意图→检索记忆→调用工具→评估结果→生成回复"等多个阶段，每个阶段都有不同的状态和转移条件。状态管理不当会导致上下文丢失、重复执行、死循环等严重问题。 2026年，随着Agent系统复杂度的指数级增长，系统化的状态管理架构已成为生产部署的必备条件。 Agent状态的三个层次 第一层：会话状态（Session State） 会话状态是最基础的状态层，管理单次用户交互的上下文： from dataclasses import dataclass, field from datetime import datetime from enum import Enum class SessionStatus(Enum): ACTIVE = "active" PAUSED = "paused" COMPLETED = "completed" FAILED = "failed" TIMEOUT = "timeout" @dataclass class SessionState: """会话状态——管理单次交互的完整生命周期""" session_id: str user_id: str status: SessionStatus created_at: datetime updated_at: datetime message_history: list = field(default_factory=list) active_tools: list = field(default_factory=list) pending_actions: list = field(default_factory=list) context_window: dict = field(default_factory=dict) metadata: dict = field(default_factory=dict) def add_message(self, role: str, content: str): self.message_history.append({ "role": role, "content": content, "timestamp": datetime.now().isoformat() }) self.updated_at = datetime.now() def is_expired(self, ttl_seconds: int = 3600) -> bool: elapsed = (datetime.now() - self.updated_at).total_seconds() return elapsed > ttl_seconds 第二层：工作流状态（Workflow State） 工作流状态管理Agent执行复杂多步骤任务时的进度： from enum import Enum class WorkflowStepStatus(Enum): PENDING = "pending" RUNNING = "running" SUCCESS = "success" FAILED = "failed" SKIPPED = "skipped" RETRYING = "retrying" @dataclass class WorkflowStep: step_id: str step_name: str status: WorkflowStepStatus dependencies: list # 前置步骤ID inputs: dict outputs: dict retry_count: int = 0 max_retries: int = 3 started_at: datetime = None completed_at: datetime = None @dataclass class WorkflowState: """工作流状态——管理多步骤任务的执行进度""" workflow_id: str session_id: str steps: dict # step_id -> WorkflowStep current_step: str context: dict # 跨步骤共享的上下文 def get_ready_steps(self) -> list: """获取可执行的步骤（依赖已完成）""" ready = [] for step_id, step in self.steps.items(): if step.status != WorkflowStepStatus.PENDING: continue deps_satisfied = all( self.steps[dep].status == WorkflowStepStatus.SUCCESS for dep in step.dependencies ) if deps_satisfied: ready.append(step) return ready def is_complete(self) -> bool: return all( s.status in [WorkflowStepStatus.SUCCESS, WorkflowStepStatus.SKIPPED] for s in self.steps.values() ) 第三层：持久状态（Persistent State） 持久状态跨越会话边界，包括用户偏好、长期记忆和已学习的模式： ...

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"] 推荐阈值： ...