阶段 3：状态管理 🔥 推荐

一句话总结：没有记忆的 Agent 每次都是从零开始——状态管理让 Agent 能记住上下文。

📊 学习进度
状态：⬜ 未开始
预计时长：2-3 小时
已完成：0/3 个模块
在整体流程中的位置：AI Agent 开发·第 3 阶段

📍 本章定位
服务方案：方案 1（辅助 40%）/ 方案 3（重要 60%）
学习方式：🔥 推荐
在流程中的作用：让 Agent 具备上下文记忆能力
核心知识点：短期记忆、长期记忆、上下文窗口管理
预计时长：2-3 小时
完成后能做什么：能为 Agent 设计记忆系统

1. 传统模式：痛点与瓶颈

1.1 无状态 AI 应用的局限性

传统 AI 应用是"无状态"的——每次调用都是独立的，不记得之前的对话。用户每次都要重新描述背景信息，这在需要连续对话或长期跟踪的场景下效率极低。

无状态 AI 的典型问题：

1.2 量化痛点数据

痛点维度	无状态 AI	有状态 Agent	改善幅度
用户重复输入率	65%	8%	-88%
多轮对话连贯性	2.1 轮断裂	50+ 轮连贯	+2300%
个性化推荐准确率	30%	78%	+160%
长期任务完成率	15%	72%	+380%
用户满意度	3.2/5	4.5/5	+41%

数据来源：LangChain 2025 年用户研究、Anthropic 2025 年基准测试

1.3 OPC 场景下的核心矛盾

OPC 运营者需要 Agent 记住客户偏好、项目上下文、历史决策。一个没有记忆的 Agent，每次都要重新解释背景，这在客户服务、项目管理、内容创作等场景下是不可接受的。

2. OPC 模式：重新定义

2.1 核心理念

Agent 记忆系统 = 短期记忆 + 长期记忆 + 工作记忆。就像人类的大脑，Agent 需要不同类型的记忆来处理不同时间尺度的信息。

记忆系统架构：

2.2 人机分工矩阵

任务	人类角色	AI 角色	协作方式
记忆架构设计	决定记住什么、忘掉什么	建议最佳实践	人决策，AI 辅助
记忆策略	设计保留和淘汰策略	实现策略逻辑	人定义规则，AI 编码
向量数据库选型	选择存储方案	配置和优化	人决策，AI 执行
隐私保护	定义数据边界	实现脱敏逻辑	人审核，AI 实现

2.3 效率对比

指标	无状态模式	有状态 Agent	提效倍数
用户重复输入	65%	8%	8x
多轮对话连贯性	2.1 轮	50+ 轮	24x
个性化准确率	30%	78%	2.6x
长期任务完成率	15%	72%	4.8x

3. 实操案例

3.1 场景描述

场景：为 OPC 的客户服务 Agent 搭建记忆系统，使其能够：

记住当前对话上下文（短期记忆）
记住客户历史偏好和问题（长期记忆）
跟踪当前处理的任务状态（工作记忆）

技术栈：Claude API + Python + ChromaDB（向量数据库）

3.2 执行过程

3.2.1 短期记忆：对话历史管理

短期记忆存储当前会话的对话历史，核心挑战是上下文窗口限制。

上下文窗口管理策略：

策略	原理	适用场景	Token 节省
滑动窗口	保留最近 N 轮对话	简单客服	40-60%
摘要压缩	定期压缩历史对话	长对话	60-80%
检索增强	只检索相关历史	大量历史	70-90%
混合策略	滑动窗口 + 摘要	通用场景	50-70%

Python 实现：滑动窗口 + 摘要压缩：

python

from typing import List, Dict
import anthropic

class ConversationMemory:
    """对话记忆管理器 - 滑动窗口 + 摘要压缩"""
    
    def __init__(self, max_turns: int = 10, summary_threshold: int = 20):
        self.messages: List[Dict] = []
        self.summary: str = ""
        self.max_turns = max_turns
        self.summary_threshold = summary_threshold
        self.client = anthropic.Anthropic()
    
    def add_message(self, role: str, content: str):
        """添加消息到记忆"""
        self.messages.append({"role": role, "content": content})
        
        # 超过阈值时触发压缩
        if len(self.messages) > self.summary_threshold:
            self._compress_history()
    
    def get_context(self) -> List[Dict]:
        """获取当前上下文（含摘要）"""
        context = []
        
        # 如果有历史摘要，添加为系统消息
        if self.summary:
            context.append({
                "role": "user",
                "content": f"[历史对话摘要]\n{self.summary}"
            })
        
        # 添加最近 N 轮对话
        recent = self.messages[-self.max_turns * 2:]  # 每轮包含 user + assistant
        context.extend(recent)
        
        return context
    
    def _compress_history(self):
        """压缩历史对话为摘要"""
        # 提取需要压缩的旧消息
        old_messages = self.messages[:-self.max_turns * 2]
        
        # 调用 LLM 生成摘要
        summary_prompt = f"""请将以下对话历史压缩为简洁的摘要，保留关键信息：
        
之前的摘要：{self.summary or '无'}

需要压缩的对话：
{self._format_messages(old_messages)}

要求：
1. 保留用户的关键需求和偏好
2. 保留重要的决策和结论
3. 丢弃重复和无关信息
4. 控制在 200 字以内"""

        response = self.client.messages.create(
            model="claude-haiku-4-20250414",  # 用小模型做摘要，节省成本
            max_tokens=300,
            messages=[{"role": "user", "content": summary_prompt}]
        )
        
        self.summary = response.content[0].text
        # 保留最近的消息
        self.messages = self.messages[-self.max_turns * 2:]
    
    def _format_messages(self, messages: List[Dict]) -> str:
        """格式化消息列表"""
        return "\n".join([f"{m['role']}: {m['content']}" for m in messages])

# 使用示例
memory = ConversationMemory(max_turns=10, summary_threshold=20)
memory.add_message("user", "我想了解一下你们的 API 定价")
memory.add_message("assistant", "我们的 API 定价分为三个层级...")
memory.add_message("user", "我主要是做量化交易的，需要高频调用")
memory.add_message("assistant", "针对高频调用场景，推荐使用...")

# 获取压缩后的上下文
context = memory.get_context()

TypeScript 实现：

typescript

interface Message {
  role: "user" | "assistant";
  content: string;
}

class ConversationMemory {
  private messages: Message[] = [];
  private summary: string = "";
  private maxTurns: number;
  private summaryThreshold: number;

  constructor(maxTurns = 10, summaryThreshold = 20) {
    this.maxTurns = maxTurns;
    this.summaryThreshold = summaryThreshold;
  }

  addMessage(role: "user" | "assistant", content: string): void {
    this.messages.push({ role, content });
    if (this.messages.length > this.summaryThreshold * 2) {
      this.compressHistory();
    }
  }

  getContext(): Message[] {
    const context: Message[] = [];
    if (this.summary) {
      context.push({
        role: "user",
        content: `[历史对话摘要]\n${this.summary}`,
      });
    }
    const recent = this.messages.slice(-this.maxTurns * 2);
    context.push(...recent);
    return context;
  }

  private async compressHistory(): Promise<void> {
    const oldMessages = this.messages.slice(0, -this.maxTurns * 2);
    // 调用 LLM 生成摘要...
    this.messages = this.messages.slice(-this.maxTurns * 2);
  }
}

3.2.2 长期记忆：向量数据库存储

长期记忆使用向量数据库存储跨会话的信息，支持语义检索。

向量数据库选型：

数据库	类型	适用场景	特点
ChromaDB	嵌入式	原型/小规模	零配置，Python 原生
Pinecone	云服务	生产环境	全托管，高性能
Qdrant	自托管	中大规模	开源，Rust 实现
Weaviate	自托管	复杂查询	GraphQL 接口

Python 实现：ChromaDB 长期记忆：

python

import chromadb
from chromadb.utils import embedding_functions
from typing import List, Dict

class LongTermMemory:
    """长期记忆管理器 - 基于向量数据库"""
    
    def __init__(self, collection_name: str = "agent_memory"):
        # 使用 ChromaDB 作为向量数据库
        self.client = chromadb.PersistentClient(path="./memory_db")
        self.embedding_fn = embedding_functions.OpenAIEmbeddingFunction(
            model_name="text-embedding-3-small"
        )
        self.collection = self.client.get_or_create_collection(
            name=collection_name,
            embedding_function=self.embedding_fn
        )
    
    def store(self, key: str, content: str, metadata: Dict = None):
        """存储记忆"""
        self.collection.upsert(
            ids=[key],
            documents=[content],
            metadatas=[metadata or {}]
        )
    
    def retrieve(self, query: str, n_results: int = 5) -> List[Dict]:
        """检索相关记忆"""
        results = self.collection.query(
            query_texts=[query],
            n_results=n_results
        )
        
        memories = []
        for i, doc in enumerate(results["documents"][0]):
            memories.append({
                "content": doc,
                "metadata": results["metadatas"][0][i],
                "distance": results["distances"][0][i]
            })
        
        return memories
    
    def forget(self, key: str):
        """删除记忆"""
        self.collection.delete(ids=[key])

# 使用示例
ltm = LongTermMemory()

# 存储客户偏好
ltm.store(
    key="client_001_preference",
    content="客户偏好简洁的 API 响应格式，不需要额外的元数据",
    metadata={"client_id": "001", "category": "preference"}
)

# 检索相关记忆
memories = ltm.retrieve("这个客户喜欢什么样的 API 格式？")
for m in memories:
    print(f"记忆: {m['content']}, 相关度: {1 - m['distance']:.2f}")

3.2.3 工作记忆：任务状态跟踪

工作记忆存储当前任务的中间状态，任务完成后可选择持久化或丢弃。

python

from dataclasses import dataclass, field
from typing import Any, Dict
from datetime import datetime

@dataclass
class TaskState:
    """任务状态"""
    task_id: str
    status: str = "pending"  # pending | in_progress | completed | failed
    current_step: int = 0
    total_steps: int = 0
    context: Dict[str, Any] = field(default_factory=dict)
    created_at: datetime = field(default_factory=datetime.now)
    updated_at: datetime = field(default_factory=datetime.now)

class WorkingMemory:
    """工作记忆管理器 - 任务状态跟踪"""
    
    def __init__(self):
        self.tasks: Dict[str, TaskState] = {}
    
    def create_task(self, task_id: str, total_steps: int) -> TaskState:
        """创建新任务"""
        state = TaskState(task_id=task_id, total_steps=total_steps)
        self.tasks[task_id] = state
        return state
    
    def update_task(self, task_id: str, **kwargs) -> TaskState:
        """更新任务状态"""
        state = self.tasks[task_id]
        for key, value in kwargs.items():
            setattr(state, key, value)
        state.updated_at = datetime.now()
        return state
    
    def get_task(self, task_id: str) -> TaskState:
        """获取任务状态"""
        return self.tasks.get(task_id)
    
    def get_active_tasks(self) -> List[TaskState]:
        """获取所有活跃任务"""
        return [t for t in self.tasks.values() if t.status in ("pending", "in_progress")]

# 使用示例
wm = WorkingMemory()

# 创建任务
task = wm.create_task("competitor_analysis_001", total_steps=5)

# 更新任务进度
wm.update_task("competitor_analysis_001", 
    status="in_progress",
    current_step=2,
    context={"competitors": ["OpenAI", "Anthropic"], "completed": ["OpenAI"]}
)

# 获取任务状态
task = wm.get_task("competitor_analysis_001")
print(f"任务 {task.task_id}: {task.current_step}/{task.total_steps}")

3.2.4 记忆整合与优先级检索

记忆系统不是简单的"存储+检索"，还需要考虑记忆的整合（将碎片信息合并为连贯知识）和优先级（重要记忆优先检索）。

记忆整合机制：

python

from datetime import datetime, timedelta
from typing import List, Dict

class MemoryConsolidator:
    """记忆整合器 - 将碎片记忆合并为连贯知识"""

    def __init__(self, long_term_memory: LongTermMemory):
        self.ltm = long_term_memory
        self.client = anthropic.Anthropic()

    async def consolidate(self, category: str, time_window: timedelta = timedelta(days=7)):
        """整合指定类别和时间窗口内的碎片记忆"""
        # 1. 检索该类别的所有记忆
        memories = self.ltm.retrieve(category, n_results=50)

        # 2. 按时间排序
        memories.sort(key=lambda m: m.get("metadata", {}).get("timestamp", ""))

        # 3. 过滤时间窗口内的记忆
        cutoff = datetime.now() - time_window
        recent = [
            m for m in memories
            if datetime.fromisoformat(m["metadata"].get("timestamp", "2000-01-01")) > cutoff
        ]

        if len(recent) < 3:
            return  # 记忆太少，无需整合

        # 4. 调用 LLM 整合记忆
        prompt = f"""请将以下碎片记忆整合为一条连贯的知识摘要：

{chr(10).join([f"- {m['content']}" for m in recent])}

要求：
1. 保留关键事实和决策
2. 去除重复信息
3. 添加时间线
4. 控制在 300 字以内"""

        response = self.client.messages.create(
            model="claude-haiku-4-20250414",
            max_tokens=400,
            messages=[{"role": "user", "content": prompt}]
        )

        consolidated = response.content[0].text

        # 5. 存储整合后的记忆
        self.ltm.store(
            key=f"consolidated_{category}_{datetime.now().strftime('%Y%m%d')}",
            content=consolidated,
            metadata={
                "category": category,
                "type": "consolidated",
                "source_count": len(recent),
                "timestamp": datetime.now().isoformat()
            }
        )

        # 6. 标记原始记忆为已整合（降低检索优先级）
        for m in recent:
            m["metadata"]["consolidated"] = True

优先级检索：

python

class PriorityRetriever:
    """优先级检索器 - 根据重要度和时效性排序"""

    def __init__(self, long_term_memory: LongTermMemory):
        self.ltm = long_term_memory

    def retrieve_with_priority(
        self,
        query: str,
        n_results: int = 5,
        recency_weight: float = 0.3,
        importance_weight: float = 0.4,
        relevance_weight: float = 0.3
    ) -> List[Dict]:
        """带优先级的检索"""
        # 1. 向量检索（获取更多候选）
        candidates = self.ltm.retrieve(query, n_results=n_results * 3)

        # 2. 计算综合得分
        scored = []
        for mem in candidates:
            # 相关性得分（向量距离）
            relevance = 1 - mem.get("distance", 0.5)

            # 时效性得分
            timestamp = mem.get("metadata", {}).get("timestamp", "")
            if timestamp:
                age_days = (datetime.now() - datetime.fromisoformat(timestamp)).days
                recency = max(0, 1 - age_days / 30)  # 30 天内线性衰减
            else:
                recency = 0.5

            # 重要性得分
            importance = mem.get("metadata", {}).get("importance", 0.5)

            # 综合得分
            score = (
                relevance_weight * relevance +
                recency_weight * recency +
                importance_weight * importance
            )
            scored.append((score, mem))

        # 3. 排序并返回
        scored.sort(key=lambda x: x[0], reverse=True)
        return [mem for _, mem in scored[:n_results]]

3.2.5 统一记忆系统

将三种记忆整合为统一的 Agent 记忆系统：

python

class AgentMemorySystem:
    """Agent 统一记忆系统"""

    def __init__(self):
        self.short_term = ConversationMemory(max_turns=10)
        self.long_term = LongTermMemory(collection_name="agent_memory")
        self.working = WorkingMemory()
        self.consolidator = MemoryConsolidator(self.long_term)
        self.retriever = PriorityRetriever(self.long_term)

    def process_message(self, role: str, content: str) -> List[Dict]:
        """处理新消息，返回完整上下文"""
        # 1. 添加到短期记忆
        self.short_term.add_message(role, content)

        # 2. 优先级检索相关长期记忆
        relevant_memories = self.retriever.retrieve_with_priority(
            content, n_results=3
        )

        # 3. 构建完整上下文
        context = self.short_term.get_context()

        # 将相关长期记忆注入上下文
        if relevant_memories:
            memory_text = "\n".join([m["content"] for m in relevant_memories])
            context.insert(0, {
                "role": "user",
                "content": f"[相关历史记忆]\n{memory_text}"
            })

        return context

    def save_important_info(self, key: str, content: str, category: str, importance: float = 0.5):
        """保存重要信息到长期记忆"""
        self.long_term.store(
            key=key,
            content=content,
            metadata={
                "category": category,
                "importance": importance,
                "timestamp": datetime.now().isoformat()
            }
        )

    async def periodic_consolidation(self):
        """定期整合记忆（建议每天执行一次）"""
        categories = ["client_preference", "market_analysis", "trading_decision"]
        for category in categories:
            await self.consolidator.consolidate(category)

3.3 前后对比

维度	无记忆 Agent	有记忆 Agent	改善
用户重复输入	每次都要重复背景	自动加载历史	-88%
对话连贯性	2 轮后断裂	50+ 轮连贯	+2400%
个性化服务	无法个性化	基于历史偏好	+160%
Token 消耗	每次重新描述	只传增量	-60%
客户满意度	3.2/5	4.5/5	+41%

4. 趋势预判（未来 1-3 年）

4.1 技术演进方向

技术方向	当前状态	1 年后	3 年后
向量数据库性能	可用	高效	极致
多模态记忆	文本为主	图文音视频	全模态
记忆压缩算法	摘要为主	智能压缩	自适应压缩
隐私保护	基础脱敏	差分隐私	联邦记忆

4.2 角色变化趋势

角色	当前	1 年后	3 年后
记忆系统架构师	几乎不存在	新兴岗位	标准配置
向量数据库工程师	小众	增长	基础能力
AI 隐私审计	缺失	初步建立	强制要求

4.3 OPC 需要提前准备的能力

向量数据库使用：掌握 ChromaDB/Pinecone/Qdrant 基本操作
Embedding 模型理解：文本向量化原理和最佳实践
记忆策略设计：决定记住什么、忘掉什么、如何压缩
隐私合规意识：GDPR/CCPA 对 AI 记忆系统的要求

5. 核心洞察

🔑 关键洞察

Agent 的记忆系统不是"把所有对话都存下来"，而是"智能地决定记住什么、忘掉什么"。好的记忆系统应该像人类大脑一样——重要的事情记一辈子，不重要的事情很快忘记。向量数据库的语义检索能力，让 Agent 能够"想起"与当前问题相关的历史记忆。

⚠️ 成本警告

向量数据库的存储和检索成本不可忽视。一个每天处理 1000 次对话的 Agent，如果每次都存储完整对话，一个月的向量数据库成本可能超过 $50。务必设计合理的记忆淘汰策略。

6. 参考与延伸

[1] LangChain. "Memory Concepts" — 记忆管理文档（2025）

[2] ChromaDB. "Documentation" — 向量数据库文档（2025）

[3] Pinecone. "Vector Database" — 云向量数据库（2025）

[4] OpenAI. "Embeddings Guide" — 文本向量化指南（2025）

[5] LangGraph. "Persistence" — Agent 状态持久化（2025）

[6] Lewis et al. "Retrieval-Augmented Generation" — RAG 论文（2020）

[7] MemGPT. "Towards LLMs as Operating Systems" — 记忆管理论文（2023）

[8] Qdrant. "Hybrid Search" — 混合检索文档（2025）

[9] MTEB Leaderboard. "Embedding Benchmark" — Embedding 模型基准测试（2025）

[10] Milvus. "Documentation" — 大规模向量数据库（2025）

[11] Anthropic. "Prompt Caching" — Prompt 缓存降低成本（2025）

[12] Weaviate. "Vector Quantization" — 向量量化技术（2025）

常见问题

问题	原因	解决方案
上下文溢出	对话太长	滑动窗口 + 摘要压缩
记忆混乱	信息太多	重要度排序 + 分类存储
成本高	Token/存储消耗多	压缩 + 检索 + 淘汰策略
检索不准	Embedding 质量差	更换 Embedding 模型
隐私泄露	记忆未脱敏	存储前脱敏处理

记忆系统性能优化

向量数据库性能对比

不同向量数据库在性能、成本和易用性上有显著差异。选择合适的向量数据库是记忆系统性能优化的关键。

数据库	写入速度	查询延迟	最大数据量	价格	推荐场景
ChromaDB	1000 docs/s	10ms	100万条	免费	原型/小规模
Pinecone	5000 docs/s	5ms	10亿条	$70/月起	生产环境
Qdrant	3000 docs/s	8ms	1亿条	免费/云服务	中大规模
Weaviate	2000 docs/s	12ms	5000万条	免费/云服务	复杂查询
Milvus	8000 docs/s	3ms	10亿条	免费	超大规模

数据来源：各数据库官方基准测试，2025 年

Embedding 模型选择

Embedding 模型的质量直接影响记忆检索的准确性。以下是主流 Embedding 模型对比：

模型	维度	质量(MTEB)	价格	推荐场景
text-embedding-3-small	1536	62.3	$0.02/1M tokens	通用场景
text-embedding-3-large	3072	64.6	$0.13/1M tokens	高精度需求
voyage-3	1024	65.2	$0.06/1M tokens	代码/技术文档
Cohere embed-v3	1024	64.1	$0.1/1M tokens	多语言场景

数据来源：MTEB Leaderboard，2025 年

记忆淘汰策略

长期运行的 Agent 会产生大量记忆数据。合理的淘汰策略是控制成本和保持检索效率的关键。

四种淘汰策略对比：

策略	原理	优点	缺点	适用场景
LRU	最近最少使用	实现简单	可能丢弃重要旧记忆	通用场景
LFU	最不常用	保留高频使用记忆	新记忆容易被淘汰	稳定场景
重要度	按重要度评分	保留关键信息	需要评估机制	专业场景
混合	LRU + 重要度	平衡多维度	实现复杂	生产环境

混合淘汰策略实现：

python

from datetime import datetime, timedelta
from typing import List, Dict, Tuple

class MemoryEvictionPolicy:
    """记忆淘汰策略 - 混合 LRU + 重要度"""

    def __init__(
        self,
        max_memories: int = 10000,
        recency_weight: float = 0.4,
        importance_weight: float = 0.4,
        frequency_weight: float = 0.2
    ):
        self.max_memories = max_memories
        self.recency_weight = recency_weight
        self.importance_weight = importance_weight
        self.frequency_weight = frequency_weight

    def calculate_score(self, memory: dict) -> float:
        """计算记忆的综合得分"""
        # 时效性得分（越新越高）
        timestamp = memory.get("metadata", {}).get("timestamp", "")
        if timestamp:
            age_days = (datetime.now() - datetime.fromisoformat(timestamp)).days
            recency = max(0, 1 - age_days / 30)  # 30 天内线性衰减
        else:
            recency = 0.5

        # 重要性得分
        importance = memory.get("metadata", {}).get("importance", 0.5)

        # 使用频率得分
        access_count = memory.get("metadata", {}).get("access_count", 0)
        frequency = min(1.0, access_count / 10)  # 10 次访问为满分

        # 综合得分
        score = (
            self.recency_weight * recency +
            self.importance_weight * importance +
            self.frequency_weight * frequency
        )
        return score

    def select_for_eviction(self, memories: List[dict], count: int) -> List[dict]:
        """选择需要淘汰的记忆"""
        # 计算每个记忆的得分
        scored = [(self.calculate_score(m), m) for m in memories]

        # 按得分升序排列（得分最低的优先淘汰）
        scored.sort(key=lambda x: x[0])

        # 返回需要淘汰的记忆
        return [m for _, m in scored[:count]]

    def should_evict(self, current_count: int) -> bool:
        """判断是否需要淘汰"""
        return current_count > self.max_memories

记忆压缩技术

当记忆数据量增长时，压缩技术可以显著降低存储成本和检索延迟。

三种压缩技术对比：

技术	原理	压缩率	质量损失	实现复杂度
摘要压缩	LLM 生成摘要	70-80%	中	低
去重合并	合并相似记忆	30-50%	低	中
向量量化	降低向量维度	50-75%	低	高

摘要压缩实现：

python

class MemoryCompressor:
    """记忆压缩器 - 使用 LLM 生成摘要"""

    def __init__(self, client, model: str = "claude-haiku-4-20250414"):
        self.client = client
        self.model = model

    async def compress(self, memories: List[dict], max_length: int = 200) -> str:
        """将多条记忆压缩为摘要"""
        # 格式化记忆
        memory_text = "\n".join([
            f"- {m['content']}" for m in memories
        ])

        # 调用 LLM 生成摘要
        response = await self.client.messages.create(
            model=self.model,
            max_tokens=max_length * 2,
            messages=[{
                "role": "user",
                "content": f"请将以下记忆压缩为简洁摘要（不超过 {max_length} 字）：\n\n{memory_text}"
            }]
        )

        return response.content[0].text.strip()

    async def deduplicate(self, memories: List[dict], similarity_threshold: float = 0.9) -> List[dict]:
        """去重 - 合并相似记忆"""
        unique = []
        seen_embeddings = []

        for memory in memories:
            # 检查是否与已有记忆相似
            is_duplicate = False
            for seen in seen_embeddings:
                similarity = self._cosine_similarity(memory["embedding"], seen)
                if similarity > similarity_threshold:
                    is_duplicate = True
                    break

            if not is_duplicate:
                unique.append(memory)
                seen_embeddings.append(memory["embedding"])

        return unique

    def _cosine_similarity(self, a: List[float], b: List[float]) -> float:
        """计算余弦相似度"""
        dot_product = sum(x * y for x, y in zip(a, b))
        norm_a = sum(x ** 2 for x in a) ** 0.5
        norm_b = sum(x ** 2 for x in b) ** 0.5
        return dot_product / (norm_a * norm_b) if norm_a * norm_b > 0 else 0

实操案例：客户服务 Agent 记忆系统优化

场景：一个每天处理 500+ 次客户对话的 Agent，记忆系统需要优化以控制成本和保持性能。

优化前问题：

向量数据库存储 50 万条记忆，查询延迟 200ms
月度存储成本 $120
检索准确率 72%

优化方案：

优化措施	实施内容	效果
记忆淘汰	淘汰 90 天未访问的记忆	存储量减少 60%
摘要压缩	每周压缩旧记忆	存储量再减 30%
索引优化	使用 HNSW 索引	查询延迟降至 15ms
Embedding 升级	从 ada-002 升级到 text-embedding-3-large	检索准确率提升至 89%

优化后效果：

向量数据库存储 12 万条记忆（减少 76%）
查询延迟 15ms（减少 92%）
月度存储成本 $35（减少 71%）
检索准确率 89%（提升 24%）

下一步

完成状态管理后，进入阶段 4：多 Agent 协作 — 学习如何让多个 Agent 分工协作，构建更强大的系统。

阶段 3：状态管理 🔥 推荐 ​

1. 传统模式：痛点与瓶颈 ​

1.1 无状态 AI 应用的局限性 ​

1.2 量化痛点数据 ​

1.3 OPC 场景下的核心矛盾 ​

2. OPC 模式：重新定义 ​

2.1 核心理念 ​

2.2 人机分工矩阵 ​

2.3 效率对比 ​

3. 实操案例 ​

3.1 场景描述 ​

3.2 执行过程 ​

3.2.1 短期记忆：对话历史管理 ​

3.2.2 长期记忆：向量数据库存储 ​

3.2.3 工作记忆：任务状态跟踪 ​

3.2.4 记忆整合与优先级检索 ​

3.2.5 统一记忆系统 ​

3.3 前后对比 ​

4. 趋势预判（未来 1-3 年） ​

4.1 技术演进方向 ​

4.2 角色变化趋势 ​

4.3 OPC 需要提前准备的能力 ​

5. 核心洞察 ​

6. 参考与延伸 ​

常见问题 ​

记忆系统性能优化 ​

向量数据库性能对比 ​

Embedding 模型选择 ​

记忆淘汰策略 ​

记忆压缩技术 ​

实操案例：客户服务 Agent 记忆系统优化 ​

下一步 ​

阶段 3：状态管理 🔥 推荐

1. 传统模式：痛点与瓶颈

1.1 无状态 AI 应用的局限性

1.2 量化痛点数据

1.3 OPC 场景下的核心矛盾

2. OPC 模式：重新定义

2.1 核心理念

2.2 人机分工矩阵

2.3 效率对比

3. 实操案例

3.1 场景描述

3.2 执行过程

3.2.1 短期记忆：对话历史管理

3.2.2 长期记忆：向量数据库存储

3.2.3 工作记忆：任务状态跟踪

3.2.4 记忆整合与优先级检索

3.2.5 统一记忆系统

3.3 前后对比

4. 趋势预判（未来 1-3 年）

4.1 技术演进方向

4.2 角色变化趋势

4.3 OPC 需要提前准备的能力

5. 核心洞察

6. 参考与延伸

常见问题

记忆系统性能优化

向量数据库性能对比

Embedding 模型选择

记忆淘汰策略

记忆压缩技术

实操案例：客户服务 Agent 记忆系统优化

下一步