AI Agent Memory：从理论到实战，掌握长短期记忆的核心技术【1】

CFzjy

971人浏览 · 2026-03-19 23:16:01

CFzjy · 2026-03-19 23:16:01 发布

你是否遇到过这些问题？

AI Agent总是"健忘"，无法记住之前的对话？

长对话场景下，上下文窗口不够用？

不知道如何设计高效的记忆系统？

面试时被问到Memory机制，只能支支吾吾？

别担心！这篇文章带你深入理解AI Memory的核心原理、最新技术和实战应用，让你在面试中脱颖而出！

📚 目录

一、为什么AI需要Memory？
二、Memory的核心架构
三、短期记忆实现方案
四、长期记忆实现方案
五、最新记忆策略与论文
六、实战案例：构建智能记忆系统
七、面试加分项：深度见解
八、面试高频问题全解析

一、为什么AI需要Memory？

1.1 AI Agent的"健忘症"

传统LLM的三大记忆困境：

困境	具体表现	影响
上下文窗口限制	只能记住最近N个token	长对话信息丢失
无持久化能力	每次对话都是"新开始"	无法积累知识
检索效率低	线性扫描历史对话	响应速度慢

1.2 Memory的核心价值

1.3 人类记忆 vs AI记忆

核心类比：

人类记忆	AI记忆	实现技术
感官记忆	输入缓冲	Prompt窗口
短期记忆	会话记忆	Redis/内存存储
长期记忆	持久化记忆	向量数据库
工作记忆	检索增强	RAG技术

二、Memory的核心架构

2.1 分层记忆架构

2.2 记忆生命周期

2.3 核心组件设计

from abc import ABC, abstractmethod
from typing import Any, Optional, List, Dict
from datetime import datetime
from pydantic import BaseModel, Field

class MemoryItem(BaseModel):
    """记忆项基类"""
    id: str
    content: str
    metadata: Dict[str, Any] = Field(default_factory=dict)
    created_at: datetime = Field(default_factory=datetime.now)
    importance: float = Field(default=0.5, ge=0, le=1)
    access_count: int = Field(default=0)
    last_accessed: Optional[datetime] = None

class BaseMemory(ABC):
    """记忆基类"""
    
    @abstractmethod
    def add(self, item: MemoryItem) -> str:
        """添加记忆"""
        pass
    
    @abstractmethod
    def get(self, memory_id: str) -> Optional[MemoryItem]:
        """获取记忆"""
        pass
    
    @abstractmethod
    def search(self, query: str, top_k: int = 10) -> List[MemoryItem]:
        """搜索记忆"""
        pass
    
    @abstractmethod
    def forget(self, memory_id: str) -> bool:
        """遗忘记忆"""
        pass
    
    @abstractmethod
    def update(self, memory_id: str, **kwargs) -> bool:
        """更新记忆"""
        pass

三、短期记忆实现方案

3.1 滑动窗口策略

from collections import deque
from typing import List, Optional

class SlidingWindowMemory(BaseMemory):
    """
    滑动窗口记忆
    适用场景：简单对话、固定上下文长度
    
    优点：实现简单、内存占用固定
    缺点：无法保留重要信息、信息丢失不可控
    """
    
    def __init__(self, max_messages: int = 10, max_tokens: int = 4000):
        self.max_messages = max_messages
        self.max_tokens = max_tokens
        self.messages: deque = deque(maxlen=max_messages)
        self.total_tokens = 0
    
    def add(self, item: MemoryItem) -> str:
        """添加消息到窗口"""
        # 检查是否需要淘汰旧消息
        while (
            len(self.messages) >= self.max_messages or
            self.total_tokens + len(item.content) > self.max_tokens
        ):
            if self.messages:
                removed = self.messages.popleft()
                self.total_tokens -= len(removed.content)
            else:
                break
        
        self.messages.append(item)
        self.total_tokens += len(item.content)
        
        return item.id
    
    def get(self, memory_id: str) -> Optional[MemoryItem]:
        """获取指定消息"""
        for msg in self.messages:
            if msg.id == memory_id:
                msg.access_count += 1
                msg.last_accessed = datetime.now()
                return msg
        return None
    
    def search(self, query: str, top_k: int = 10) -> List[MemoryItem]:
        """搜索消息（简单关键词匹配）"""
        results = []
        for msg in self.messages:
            if query.lower() in msg.content.lower():
                results.append(msg)
        return results[:top_k]
    
    def forget(self, memory_id: str) -> bool:
        """删除消息"""
        for i, msg in enumerate(self.messages):
            if msg.id == memory_id:
                del self.messages[i]
                self.total_tokens -= len(msg.content)
                return True
        return False
    
    def update(self, memory_id: str, **kwargs) -> bool:
        """更新消息"""
        msg = self.get(memory_id)
        if msg:
            for key, value in kwargs.items():
                if hasattr(msg, key):
                    setattr(msg, key, value)
            return True
        return False
    
    def get_context(self) -> str:
        """获取当前上下文"""
        return "\n".join([msg.content for msg in self.messages])

3.2 摘要压缩策略

from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate

class SummaryMemory(BaseMemory):
    """
    摘要压缩记忆
    适用场景：长对话、需要保留历史信息
    
    优点：保留关键信息、节省Token
    缺点：摘要可能丢失细节、依赖LLM质量
    """
    
    def __init__(
        self,
        llm_model: str = "gpt-4",
        max_original_tokens: int = 2000,
        summary_ratio: float = 0.3
    ):
        self.llm = ChatOpenAI(model=llm_model, temperature=0.3)
        self.max_original_tokens = max_original_tokens
        self.summary_ratio = summary_ratio
        
        self.original_messages: List[MemoryItem] = []
        self.summaries: List[MemoryItem] = []
        
        self.summary_prompt = ChatPromptTemplate.from_messages([
            ("system", """你是一个专业的对话摘要助手。请将以下对话历史压缩为简洁的摘要。

要求：
1. 保留关键信息：用户意图、重要决策、任务状态
2. 保留用户偏好和个性化信息
3. 保留时间敏感信息
4. 省略寒暄、重复内容
5. 摘要长度约为原文的{ratio}%"""),
            ("human", "对话历史：\n{conversation}\n\n请生成摘要：")
        ])
    
    def add(self, item: MemoryItem) -> str:
        """添加消息"""
        self.original_messages.append(item)
        
        # 检查是否需要压缩
        if self._should_compress():
            self._compress()
        
        return item.id
    
    def _should_compress(self) -> bool:
        """判断是否需要压缩"""
        total_tokens = sum(len(msg.content) for msg in self.original_messages)
        return total_tokens > self.max_original_tokens
    
    def _compress(self):
        """压缩历史对话"""
        # 获取需要压缩的消息
        messages_to_compress = self.original_messages[:-5]  # 保留最近5条
        
        if not messages_to_compress:
            return
        
        # 构建对话文本
        conversation = "\n".join([
            f"[{msg.metadata.get('role', 'user')}]: {msg.content}"
            for msg in messages_to_compress
        ])
        
        # 调用LLM生成摘要
        chain = self.summary_prompt | self.llm
        summary = chain.invoke({
            "conversation": conversation,
            "ratio": int(self.summary_ratio * 100)
        })
        
        # 创建摘要记忆项
        summary_item = MemoryItem(
            id=f"summary_{datetime.now().timestamp()}",
            content=summary.content,
            metadata={
                "type": "summary",
                "original_count": len(messages_to_compress),
                "compression_ratio": self.summary_ratio
            },
            importance=0.8
        )
        
        # 更新记忆
        self.summaries.append(summary_item)
        self.original_messages = self.original_messages[-5:]
    
    def get_context(self) -> str:
        """获取当前上下文"""
        context_parts = []
        
        # 添加摘要
        if self.summaries:
            context_parts.append("【历史摘要】")
            for summary in self.summaries:
                context_parts.append(summary.content)
        
        # 添加最近对话
        if self.original_messages:
            context_parts.append("\n【最近对话】")
            for msg in self.original_messages:
                role = msg.metadata.get("role", "user")
                context_parts.append(f"[{role}]: {msg.content}")
        
        return "\n".join(context_parts)
    
    def search(self, query: str, top_k: int = 10) -> List[MemoryItem]:
        """搜索记忆"""
        results = []
        
        # 搜索摘要
        for summary in self.summaries:
            if query.lower() in summary.content.lower():
                results.append(summary)
        
        # 搜索原始消息
        for msg in self.original_messages:
            if query.lower() in msg.content.lower():
                results.append(msg)
        
        return results[:top_k]
    
    def get(self, memory_id: str) -> Optional[MemoryItem]:
        """获取指定记忆"""
        for msg in self.original_messages:
            if msg.id == memory_id:
                return msg
        for summary in self.summaries:
            if summary.id == memory_id:
                return summary
        return None
    
    def forget(self, memory_id: str) -> bool:
        """删除记忆"""
        for i, msg in enumerate(self.original_messages):
            if msg.id == memory_id:
                del self.original_messages[i]
                return True
        for i, summary in enumerate(self.summaries):
            if summary.id == memory_id:
                del self.summaries[i]
                return True
        return False
    
    def update(self, memory_id: str, **kwargs) -> bool:
        """更新记忆"""
        item = self.get(memory_id)
        if item:
            for key, value in kwargs.items():
                if hasattr(item, key):
                    setattr(item, key, value)
            return True
        return False

3.3 重要性加权策略

import numpy as np
from typing import List, Tuple
from datetime import datetime, timedelta

class ImportanceWeightedMemory(BaseMemory):
    """
    重要性加权记忆
    适用场景：需要区分信息重要性的场景
    
    核心思想：
    1. 根据内容重要性分配存储资源
    2. 重要信息长期保留，次要信息短期保留
    3. 动态调整重要性分数
    """
    
    def __init__(
        self,
        llm_model: str = "gpt-4",
        high_importance_threshold: float = 0.8,
        low_importance_threshold: float = 0.5,
        decay_rate: float = 0.1
    ):
        self.llm = ChatOpenAI(model=llm_model, temperature=0)
        
        self.high_threshold = high_importance_threshold
        self.low_threshold = low_importance_threshold
        self.decay_rate = decay_rate
        
        # 分层存储
        self.high_importance: List[MemoryItem] = []
        self.medium_importance: List[MemoryItem] = []
        self.low_importance: List[MemoryItem] = []
        
        # 重要性评估Prompt
        self.importance_prompt = ChatPromptTemplate.from_messages([
            ("system", """评估以下对话内容的重要性（0-1分）。

重要性评估标准：
1. 包含用户个人信息（姓名、偏好、背景）：0.9-1.0
2. 包含重要决策或承诺：0.8-0.9
3. 包含任务关键信息：0.7-0.8
4. 包含有用知识或经验：0.6-0.7
5. 普通对话内容：0.4-0.6
6. 寒暄、无关内容：0.0-0.4

请只返回一个0-1之间的数字，不要解释。"""),
            ("human", "内容：{content}")
        ])
    
    def _evaluate_importance(self, content: str) -> float:
        """评估内容重要性"""
        chain = self.importance_prompt | self.llm
        result = chain.invoke({"content": content})
        
        try:
            score = float(result.content.strip())
            return max(0.0, min(1.0, score))
        except:
            return 0.5
    
    def add(self, item: MemoryItem) -> str:
        """添加记忆"""
        # 评估重要性
        if item.importance == 0.5:  # 默认值，需要评估
            item.importance = self._evaluate_importance(item.content)
        
        # 分层存储
        if item.importance >= self.high_threshold:
            self.high_importance.append(item)
        elif item.importance >= self.low_threshold:
            self.medium_importance.append(item)
        else:
            self.low_importance.append(item)
        
        return item.id
    
    def _apply_decay(self):
        """应用时间衰减"""
        now = datetime.now()
        
        for storage in [self.medium_importance, self.low_importance]:
            to_remove = []
            for item in storage:
                # 计算时间衰减
                days_passed = (now - item.created_at).days
                decay_factor = np.exp(-self.decay_rate * days_passed)
                item.importance *= decay_factor
                
                # 检查是否需要移除
                if item.importance < 0.1:
                    to_remove.append(item)
            
            for item in to_remove:
                storage.remove(item)
    
    def get_context(self, max_tokens: int = 4000) -> str:
        """获取上下文（按重要性排序）"""
        self._apply_decay()
        
        context_parts = []
        current_tokens = 0
        
        # 按重要性排序
        all_items = (
            self.high_importance +
            self.medium_importance +
            self.low_importance
        )
        all_items.sort(key=lambda x: x.importance, reverse=True)
        
        # 选择记忆
        for item in all_items:
            item_tokens = len(item.content)
            if current_tokens + item_tokens > max_tokens:
                break
            
            context_parts.append(item.content)
            current_tokens += item_tokens
            
            # 更新访问信息
            item.access_count += 1
            item.last_accessed = datetime.now()
        
        return "\n".join(context_parts)
    
    def search(self, query: str, top_k: int = 10) -> List[MemoryItem]:
        """搜索记忆"""
        results = []
        
        for storage in [
            self.high_importance,
            self.medium_importance,
            self.low_importance
        ]:
            for item in storage:
                if query.lower() in item.content.lower():
                    results.append(item)
        
        # 按重要性排序
        results.sort(key=lambda x: x.importance, reverse=True)
        
        return results[:top_k]
    
    def get(self, memory_id: str) -> Optional[MemoryItem]:
        """获取指定记忆"""
        for storage in [
            self.high_importance,
            self.medium_importance,
            self.low_importance
        ]:
            for item in storage:
                if item.id == memory_id:
                    return item
        return None
    
    def forget(self, memory_id: str) -> bool:
        """删除记忆"""
        for storage in [
            self.high_importance,
            self.medium_importance,
            self.low_importance
        ]:
            for i, item in enumerate(storage):
                if item.id == memory_id:
                    del storage[i]
                    return True
        return False
    
    def update(self, memory_id: str, **kwargs) -> bool:
        """更新记忆"""
        item = self.get(memory_id)
        if item:
            for key, value in kwargs.items():
                if hasattr(item, key):
                    setattr(item, key, value)
            return True
        return False

四、长期记忆实现方案

4.1 向量数据库存储

from pymilvus import (
    connections,
    Collection,
    FieldSchema,
    CollectionSchema,
    DataType,
    utility
)
from sentence_transformers import SentenceTransformer
from typing import List, Optional
import uuid

class VectorMemory(BaseMemory):
    """
    向量数据库长期记忆
    适用场景：大规模记忆存储、语义检索
    
    优点：
    1. 支持大规模存储（百万级）
    2. 语义相似度检索
    3. 持久化存储
    
    缺点：
    1. 需要额外的向量数据库
    2. 检索延迟相对较高
    """
    
    def __init__(
        self,
        collection_name: str = "agent_memory",
        embedding_model: str = "BAAI/bge-large-zh",
        milvus_host: str = "localhost",
        milvus_port: int = 19530
    ):
        # 连接Milvus
        connections.connect(
            alias="default",
            host=milvus_host,
            port=milvus_port
        )
        
        self.collection_name = collection_name
        self.embedding_model = SentenceTransformer(embedding_model)
        self.embedding_dim = 1024  # bge-large-zh
        
        # 创建或获取Collection
        self._init_collection()
    
    def _init_collection(self):
        """初始化Collection"""
        if utility.has_collection(self.collection_name):
            self.collection = Collection(self.collection_name)
        else:
            # 定义字段
            fields = [
                FieldSchema(name="id", dtype=DataType.VARCHAR, max_length=64, is_primary=True),
                FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=self.embedding_dim),
                FieldSchema(name="content", dtype=DataType.VARCHAR, max_length=65535),
                FieldSchema(name="importance", dtype=DataType.FLOAT),
                FieldSchema(name="created_at", dtype=DataType.INT64),
                FieldSchema(name="access_count", dtype=DataType.INT64),
            ]
            
            # 创建Schema
            schema = CollectionSchema(
                fields=fields,
                description="Agent Memory Collection"
            )
            
            # 创建Collection
            self.collection = Collection(
                name=self.collection_name,
                schema=schema
            )
            
            # 创建索引
            index_params = {
                "metric_type": "COSINE",
                "index_type": "IVF_FLAT",
                "params": {"nlist": 1024}
            }
            self.collection.create_index(
                field_name="embedding",
                index_params=index_params
            )
        
        # 加载Collection
        self.collection.load()
    
    def _get_embedding(self, text: str) -> List[float]:
        """获取文本向量"""
        embedding = self.embedding_model.encode(text)
        return embedding.tolist()
    
    def add(self, item: MemoryItem) -> str:
        """添加记忆"""
        # 生成ID
        if not item.id:
            item.id = str(uuid.uuid4())
        
        # 获取向量
        embedding = self._get_embedding(item.content)
        
        # 插入数据
        self.collection.insert([
            [item.id],
            [embedding],
            [item.content],
            [item.importance],
            [int(item.created_at.timestamp())],
            [item.access_count]
        ])
        
        # 刷新
        self.collection.flush()
        
        return item.id
    
    def search(
        self,
        query: str,
        top_k: int = 10,
        filters: Optional[str] = None
    ) -> List[MemoryItem]:
        """
        语义检索
        
        Args:
            query: 查询文本
            top_k: 返回数量
            filters: 过滤条件（Milvus表达式）
        """
        # 获取查询向量
        query_embedding = self._get_embedding(query)
        
        # 搜索参数
        search_params = {
            "metric_type": "COSINE",
            "params": {"nprobe": 10}
        }
        
        # 执行搜索
        results = self.collection.search(
            data=[query_embedding],
            anns_field="embedding",
            param=search_params,
            limit=top_k,
            expr=filters,
            output_fields=["content", "importance", "created_at", "access_count"]
        )
        
        # 转换结果
        memory_items = []
        for hits in results:
            for hit in hits:
                item = MemoryItem(
                    id=hit.id,
                    content=hit.entity.get("content"),
                    importance=hit.entity.get("importance"),
                    created_at=datetime.fromtimestamp(hit.entity.get("created_at")),
                    access_count=hit.entity.get("access_count")
                )
                memory_items.append(item)
        
        return memory_items
    
    def get(self, memory_id: str) -> Optional[MemoryItem]:
        """获取指定记忆"""
        results = self.collection.query(
            expr=f'id == "{memory_id}"',
            output_fields=["content", "importance", "created_at", "access_count"]
        )
        
        if results:
            result = results[0]
            return MemoryItem(
                id=memory_id,
                content=result.get("content"),
                importance=result.get("importance"),
                created_at=datetime.fromtimestamp(result.get("created_at")),
                access_count=result.get("access_count")
            )
        
        return None
    
    def forget(self, memory_id: str) -> bool:
        """删除记忆"""
        try:
            self.collection.delete(f'id == "{memory_id}"')
            return True
        except:
            return False
    
    def update(self, memory_id: str, **kwargs) -> bool:
        """更新记忆"""
        # Milvus不支持直接更新，需要删除后重新插入
        item = self.get(memory_id)
        if not item:
            return False
        
        # 更新属性
        for key, value in kwargs.items():
            if hasattr(item, key):
                setattr(item, key, value)
        
        # 删除旧记录
        self.forget(memory_id)
        
        # 插入新记录
        self.add(item)
        
        return True
    
    def get_context(
        self,
        query: str,
        max_tokens: int = 4000,
        min_importance: float = 0.3
    ) -> str:
        """获取相关上下文"""
        # 搜索相关记忆
        results = self.search(
            query,
            top_k=50,
            filters=f"importance >= {min_importance}"
        )
        
        # 按重要性和相关性排序
        results.sort(
            key=lambda x: x.importance,
            reverse=True
        )
        
        # 选择记忆
        context_parts = []
        current_tokens = 0
        
        for item in results:
            item_tokens = len(item.content)
            if current_tokens + item_tokens > max_tokens:
                break
            
            context_parts.append(item.content)
            current_tokens += item_tokens
        
        return "\n".join(context_parts)

4.2 知识图谱记忆

from neo4j import GraphDatabase
from typing import List, Dict, Any

class KnowledgeGraphMemory(BaseMemory):
    """
    知识图谱记忆
    适用场景：结构化知识存储、关系推理
    
    优点：
    1. 支持复杂关系存储
    2. 支持图遍历和推理
    3. 知识结构化
    
    缺点：
    1. 需要Neo4j数据库
    2. 非结构化文本存储能力弱
    """
    
    def __init__(
        self,
        uri: str = "bolt://localhost:7687",
        user: str = "neo4j",
        password: str = "password"
    ):
        self.driver = GraphDatabase.driver(uri, auth=(user, password))
    
    def close(self):
        """关闭连接"""
        self.driver.close()
    
    def add_entity(
        self,
        entity_type: str,
        entity_id: str,
        properties: Dict[str, Any]
    ):
        """添加实体"""
        with self.driver.session() as session:
            query = f"""
            MERGE (e:{entity_type} {{id: $entity_id}})
            SET e += $properties
            SET e.created_at = datetime()
            SET e.access_count = 0
            """
            session.run(query, {
                "entity_id": entity_id,
                "properties": properties
            })
    
    def add_relation(
        self,
        from_type: str,
        from_id: str,
        relation: str,
        to_type: str,
        to_id: str,
        properties: Optional[Dict[str, Any]] = None
    ):
        """添加关系"""
        with self.driver.session() as session:
            query = f"""
            MATCH (from:{from_type} {{id: $from_id}})
            MATCH (to:{to_type} {{id: $to_id}})
            MERGE (from)-[r:{relation}]->(to)
            SET r += $properties
            SET r.created_at = datetime()
            """
            session.run(query, {
                "from_id": from_id,
                "to_id": to_id,
                "properties": properties or {}
            })
    
    def get_user_profile(self, user_id: str) -> Dict[str, Any]:
        """获取用户画像"""
        with self.driver.session() as session:
            query = """
            MATCH (u:User {id: $user_id})
            OPTIONAL MATCH (u)-[r]->(related)
            RETURN u, collect({relation: type(r), target: related.id, properties: properties(related)}) as relations
            """
            result = session.run(query, {"user_id": user_id})
            record = result.single()
            
            if record:
                return {
                    "user": dict(record["u"]),
                    "relations": record["relations"]
                }
            
            return {}
    
    def find_related_knowledge(
        self,
        entity_id: str,
        max_depth: int = 2
    ) -> List[Dict[str, Any]]:
        """查找相关知识（图遍历）"""
        with self.driver.session() as session:
            query = f"""
            MATCH path = (start {{id: $entity_id}})-[*1..{max_depth}]-(related)
            RETURN 
                related.id as id,
                labels(related)[0] as type,
                properties(related) as properties,
                length(path) as distance
            ORDER BY distance
            LIMIT 20
            """
            result = session.run(query, {"entity_id": entity_id})
            
            return [dict(record) for record in result]
    
    def add(self, item: MemoryItem) -> str:
        """添加记忆（提取实体和关系）"""
        # 这里需要NLP提取实体和关系
        # 简化实现：直接存储为Memory节点
        with self.driver.session() as session:
            query = """
            CREATE (m:Memory {
                id: $id,
                content: $content,
                importance: $importance,
                created_at: datetime(),
                access_count: 0
            })
            SET m += $metadata
            """
            session.run(query, {
                "id": item.id,
                "content": item.content,
                "importance": item.importance,
                "metadata": item.metadata
            })
        
        return item.id
    
    def search(self, query: str, top_k: int = 10) -> List[MemoryItem]:
        """搜索记忆"""
        with self.driver.session() as session:
            query_cypher = """
            CALL db.index.fulltext.queryNodes('memory_content', $query)
            YIELD node, score
            RETURN node.id as id, node.content as content, 
                   node.importance as importance, score
            ORDER BY score DESC, importance DESC
            LIMIT $top_k
            """
            result = session.run(query_cypher, {
                "query": query,
                "top_k": top_k
            })
            
            return [
                MemoryItem(
                    id=record["id"],
                    content=record["content"],
                    importance=record["importance"]
                )
                for record in result
            ]
    
    def get(self, memory_id: str) -> Optional[MemoryItem]:
        """获取指定记忆"""
        with self.driver.session() as session:
            query = """
            MATCH (m:Memory {id: $memory_id})
            RETURN m.id as id, m.content as content, 
                   m.importance as importance, m.created_at as created_at
            """
            result = session.run(query, {"memory_id": memory_id})
            record = result.single()
            
            if record:
                return MemoryItem(
                    id=record["id"],
                    content=record["content"],
                    importance=record["importance"],
                    created_at=record["created_at"]
                )
            
            return None
    
    def forget(self, memory_id: str) -> bool:
        """删除记忆"""
        with self.driver.session() as session:
            query = """
            MATCH (m:Memory {id: $memory_id})
            DETACH DELETE m
            """
            session.run(query, {"memory_id": memory_id})
            return True
    
    def update(self, memory_id: str, **kwargs) -> bool:
        """更新记忆"""
        with self.driver.session() as session:
            query = """
            MATCH (m:Memory {id: $memory_id})
            SET m += $properties
            """
            session.run(query, {
                "memory_id": memory_id,
                "properties": kwargs
            })
            return True

4.3 混合记忆架构

from typing import List, Optional, Dict
import redis
import json

class HybridMemory(BaseMemory):
    """
    混合记忆系统
    整合短期、长期、知识图谱记忆
    
    架构设计：
    1. 即时记忆：当前上下文（LLM窗口）
    2. 短期记忆：Redis（会话级）
    3. 长期记忆：Milvus（持久化）
    4. 知识记忆：Neo4j（结构化）
    """
    
    def __init__(
        self,
        redis_url: str = "redis://localhost:6379",
        milvus_host: str = "localhost",
        milvus_port: int = 19530,
        neo4j_uri: str = "bolt://localhost:7687",
        neo4j_user: str = "neo4j",
        neo4j_password: str = "password"
    ):
        # 初始化各存储层
        self.redis_client = redis.from_url(redis_url)
        self.vector_memory = VectorMemory(
            milvus_host=milvus_host,
            milvus_port=milvus_port
        )
        self.kg_memory = KnowledgeGraphMemory(
            uri=neo4j_uri,
            user=neo4j_user,
            password=neo4j_password
        )
        
        # 配置
        self.session_ttl = 3600  # 会话记忆1小时过期
        self.high_importance_threshold = 0.8
    
    def add(self, item: MemoryItem) -> str:
        """添加记忆（智能路由）"""
        memory_id = item.id or str(uuid.uuid4())
        item.id = memory_id
        
        # 1. 写入短期记忆（Redis）
        self._add_to_short_term(item)
        
        # 2. 根据重要性写入长期记忆
        if item.importance >= 0.5:
            self.vector_memory.add(item)
        
        # 3. 高重要性记忆提取实体写入知识图谱
        if item.importance >= self.high_importance_threshold:
            self._extract_and_store_entities(item)
        
        return memory_id
    
    def _add_to_short_term(self, item: MemoryItem):
        """添加到短期记忆"""
        key = f"memory:session:{item.id}"
        
        data = {
            "id": item.id,
            "content": item.content,
            "importance": item.importance,
            "metadata": item.metadata,
            "created_at": item.created_at.isoformat()
        }
        
        self.redis_client.setex(
            key,
            self.session_ttl,
            json.dumps(data, ensure_ascii=False)
        )
    
    def _extract_and_store_entities(self, item: MemoryItem):
        """提取实体并存储到知识图谱"""
        # 简化实现：实际需要NLP提取
        # 这里假设metadata中已包含实体信息
        entities = item.metadata.get("entities", [])
        
        for entity in entities:
            self.kg_memory.add_entity(
                entity_type=entity.get("type", "Entity"),
                entity_id=entity.get("id"),
                properties=entity.get("properties", {})
            )
    
    def search(
        self,
        query: str,
        top_k: int = 10,
        include_short_term: bool = True,
        include_long_term: bool = True,
        include_knowledge: bool = True
    ) -> List[MemoryItem]:
        """搜索记忆（并行检索）"""
        results = []
        
        # 1. 搜索短期记忆
        if include_short_term:
            short_term_results = self._search_short_term(query)
            results.extend(short_term_results)
        
        # 2. 搜索长期记忆
        if include_long_term:
            long_term_results = self.vector_memory.search(query, top_k=top_k)
            results.extend(long_term_results)
        
        # 3. 搜索知识图谱
        if include_knowledge:
            kg_results = self._search_knowledge_graph(query)
            results.extend(kg_results)
        
        # 去重并排序
        unique_results = {}
        for item in results:
            if item.id not in unique_results:
                unique_results[item.id] = item
        
        sorted_results = sorted(
            unique_results.values(),
            key=lambda x: x.importance,
            reverse=True
        )
        
        return sorted_results[:top_k]
    
    def _search_short_term(self, query: str) -> List[MemoryItem]:
        """搜索短期记忆"""
        results = []
        
        # 扫描所有会话记忆
        for key in self.redis_client.scan_iter("memory:session:*"):
            data = json.loads(self.redis_client.get(key))
            
            if query.lower() in data["content"].lower():
                item = MemoryItem(
                    id=data["id"],
                    content=data["content"],
                    importance=data["importance"],
                    metadata=data["metadata"],
                    created_at=datetime.fromisoformat(data["created_at"])
                )
                results.append(item)
        
        return results
    
    def _search_knowledge_graph(self, query: str) -> List[MemoryItem]:
        """搜索知识图谱"""
        # 简化实现
        return self.kg_memory.search(query, top_k=5)
    
    def get_context(
        self,
        query: str,
        max_tokens: int = 4000
    ) -> str:
        """获取上下文"""
        # 搜索相关记忆
        memories = self.search(query, top_k=20)
        
        # 构建上下文
        context_parts = []
        current_tokens = 0
        
        for memory in memories:
            tokens = len(memory.content)
            if current_tokens + tokens > max_tokens:
                break
            
            context_parts.append(memory.content)
            current_tokens += tokens
        
        return "\n".join(context_parts)
    
    def get(self, memory_id: str) -> Optional[MemoryItem]:
        """获取指定记忆"""
        # 先查短期记忆
        key = f"memory:session:{memory_id}"
        data = self.redis_client.get(key)
        
        if data:
            data = json.loads(data)
            return MemoryItem(
                id=data["id"],
                content=data["content"],
                importance=data["importance"],
                metadata=data["metadata"],
                created_at=datetime.fromisoformat(data["created_at"])
            )
        
        # 再查长期记忆
        return self.vector_memory.get(memory_id)
    
    def forget(self, memory_id: str) -> bool:
        """删除记忆"""
        # 删除短期记忆
        self.redis_client.delete(f"memory:session:{memory_id}")
        
        # 删除长期记忆
        self.vector_memory.forget(memory_id)
        
        # 删除知识图谱记忆
        self.kg_memory.forget(memory_id)
        
        return True
    
    def update(self, memory_id: str, **kwargs) -> bool:
        """更新记忆"""
        # 更新短期记忆
        key = f"memory:session:{memory_id}"
        data = self.redis_client.get(key)
        
        if data:
            data = json.loads(data)
            data.update(kwargs)
            self.redis_client.setex(
                key,
                self.session_ttl,
                json.dumps(data, ensure_ascii=False)
            )
        
        # 更新长期记忆
        self.vector_memory.update(memory_id, **kwargs)
        
        return True
    
    def consolidate(self):
        """
        记忆巩固
        将短期记忆中的重要信息转移到长期记忆
        """
        for key in self.redis_client.scan_iter("memory:session:*"):
            data = json.loads(self.redis_client.get(key))
            
            # 检查重要性
            if data["importance"] >= 0.5:
                # 检查是否已在长期记忆中
                if not self.vector_memory.get(data["id"]):
                    # 转移到长期记忆
                    item = MemoryItem(
                        id=data["id"],
                        content=data["content"],
                        importance=data["importance"],
                        metadata=data["metadata"],
                        created_at=datetime.fromisoformat(data["created_at"])
                    )
                    self.vector_memory.add(item)