Agent 记忆系统（四）：向量数据库、知识图谱与记忆检索全景指南

from qdrant_client import QdrantClient
from qdrant_client.models import (
    Distance, VectorParams, PointStruct,
    Filter, FieldCondition, MatchValue
)
from sentence_transformers import SentenceTransformer

# 初始化：连接本地 Qdrant 服务
client = QdrantClient(host="localhost", port=6333)
embedder = SentenceTransformer("BAAI/bge-m3")

# 创建集合（Collection）：存储 Agent 记忆
client.create_collection(
    collection_name="agent_memory",
    vectors_config=VectorParams(
        size=1024,           # bge-m3 的向量维度
        distance=Distance.COSINE  # 余弦相似度
    )
)

# 添加记忆：将经验片段存入向量数据库
memories = [
    {"text": "用户偏好简洁的代码示例，不喜欢冗长解释", "type": "user_preference", "session": "sess_001"},
    {"text": "Python 项目中用 pytest 做单元测试，覆盖率要求 80%+", "type": "best_practice", "domain": "python"},
    {"text": "上次部署失败是因为内存不足，建议增加 swap 空间", "type": "lesson_learned", "project": "web-app"},
]

points = []
for i, mem in enumerate(memories):
    vector = embedder.encode(mem["text"]).tolist()
    points.append(PointStruct(
        id=i,
        vector=vector,
        payload={"text": mem["text"], "type": mem["type"], **{k: v for k, v in mem.items() if k != "text"}}
    ))

client.upsert(collection_name="agent_memory", points=points)

# 检索记忆：根据当前场景找到相关记忆
query = "用户不喜欢太长的代码解释"
query_vector = embedder.encode(query).tolist()

# 混合检索：语义搜索 + 类型过滤
results = client.search(
    collection_name="agent_memory",
    query_vector=query_vector,
    query_filter=Filter(
        must=[FieldCondition(key="type", match=MatchValue(value="user_preference"))]
    ),
    limit=3
)

for hit in results:
    print(f"[{hit.score:.3f}] {hit.payload['text']}")

from qdrant_client import QdrantClient
from sentence_transformers import CrossEncoder, SentenceTransformer
import numpy as np

class AgentMemoryRetriever:
    """Agent 记忆检索器：向量搜索 + 交叉编码重排序"""
    
    def __init__(self, collection_name: str):
        # 检索模型：快速、轻量
        self.embedder = SentenceTransformer("BAAI/bge-m3")
        # 重排序模型：精准但较慢
        self.reranker = CrossEncoder("BAAI/bge-reranker-v2-m3")
        self.client = QdrantClient(host="localhost", port=6333)
        self.collection = collection_name
    
    def retrieve(self, query: str, top_k: int = 5, 
                 memory_type: str = None, limit_candidates: int = 50) -> list:
        """
        两阶段记忆检索管道：
        1. 向量搜索获取候选（快速、召回率高）
        2. 交叉编码重排序（精准、准确率高）
        """
        # 阶段 1：向量检索
        query_vector = self.embedder.encode(query).tolist()
        
        search_kwargs = {
            "collection_name": self.collection,
            "query_vector": query_vector,
            "limit": limit_candidates,
        }
        
        # 可选的元数据过滤
        if memory_type:
            from qdrant_client.models import Filter, FieldCondition, MatchValue
            search_kwargs["query_filter"] = Filter(
                must=[FieldCondition(key="type", match=MatchValue(value=memory_type))]
            )
        
        candidates = self.client.search(**search_kwargs)
        
        if not candidates:
            return []
        
        # 阶段 2：交叉编码重排序
        pairs = [(query, c.payload["text"]) for c in candidates]
        scores = self.reranker.predict(pairs)
        
        # 融合原始向量分数和重排序分数
        combined = []
        for i, candidate in enumerate(candidates):
            combined_score = 0.3 * candidate.score + 0.7 * float(scores[i])
            combined.append({
                "text": candidate.payload["text"],
                "score": combined_score,
                "metadata": {k: v for k, v in candidate.payload.items() if k != "text"}
            })
        
        # 按融合分数排序，返回 Top-K
        combined.sort(key=lambda x: x["score"], reverse=True)
        return combined[:top_k]

# 使用示例
retriever = AgentMemoryRetriever("agent_memory")
memories = retriever.retrieve(
    query="这个项目的部署注意事项",
    memory_type="lesson_learned",
    top_k=3
)
for mem in memories:
    print(f"[{mem['score']:.3f}] {mem['text']}")

import networkx as nx
from datetime import datetime
import json

class AgentKnowledgeGraph:
    """Agent 的知识图谱记忆系统"""
    
    def __init__(self):
        self.graph = nx.MultiDiGraph()  # 有向多重图（允许同类型多条边）
    
    def add_entity(self, entity_id: str, entity_type: str, 
                   attributes: dict = None):
        """添加实体到知识图谱"""
        self.graph.add_node(entity_id, 
                           type=entity_type,
                           created_at=datetime.now().isoformat(),
                           **(attributes or {}))
    
    def add_relation(self, source: str, relation: str, 
                     target: str, weight: float = 1.0,
                     attributes: dict = None):
        """在两个实体间添加关系"""
        self.graph.add_edge(source, target,
                          key=relation,
                          relation_type=relation,
                          weight=weight,
                          created_at=datetime.now().isoformat(),
                          **(attributes or {}))
    
    def query_entity(self, entity_id: str) -> dict:
        """查询实体的完整信息"""
        if not self.graph.has_node(entity_id):
            return None
        node_data = self.graph.nodes[entity_id]
        # 获取所有关系
        relations = []
        for _, target, data in self.graph.out_edges(entity_id, data=True):
            relations.append({
                "relation": data["relation_type"],
                "target": target,
                "weight": data.get("weight", 1.0)
            })
        return {"id": entity_id, **node_data, "relations": relations}
    
    def multi_hop_query(self, start_entity: str, 
                        relations: list[str], 
                        max_hops: int = 3) -> list:
        """
        多跳关系查询：从起点沿指定关系类型查找目标
        示例：查询"张三的同事的妻子"
          multi_hop_query("张三", ["colleague_of", "spouse_of"])
        """
        current_nodes = {start_entity}
        path_history = []
        
        for relation in relations:
            next_nodes = set()
            for node in current_nodes:
                for _, target, data in self.graph.out_edges(node, data=True):
                    if data["relation_type"] == relation:
                        next_nodes.add(target)
                        path_history.append({
                            "from": node,
                            "relation": relation,
                            "to": target
                        })
            current_nodes = next_nodes
            if not current_nodes:
                break
        
        return {
            "results": list(current_nodes),
            "paths": path_history,
            "hops": len([r for r in relations if any(
                p["relation"] == r for p in path_history
            )])
        }
    
    def get_context_for_agent(self, entity_id: str, 
                               depth: int = 2) -> str:
        """
        为 Agent 生成某个实体的上下文字典
        用于构建 prompt 中的记忆上下文
        """
        context_parts = []
        visited = set()
        
        def traverse(node_id: str, current_depth: int):
            if node_id in visited or current_depth > depth:
                return
            visited.add(node_id)
            
            node_data = self.graph.nodes[node_id]
            context_parts.append(
                f"[{node_data.get('type', 'unknown')}] {node_id}"
            )
            
            for _, target, data in self.graph.out_edges(node_id, data=True):
                rel = data["relation_type"]
                context_parts.append(f"  └─ {rel} → {target}")
                traverse(target, current_depth + 1)
        
        traverse(entity_id, 0)
        return "\n".join(context_parts)

# 使用示例：构建项目知识图谱
kg = AgentKnowledgeGraph()

# 添加实体
kg.add_entity("project-alpha", "project", {"status": "active", "lead": "张三"})
kg.add_entity("张三", "person", {"role": "tech_lead"})
kg.add_entity("李四", "person", {"role": "developer"})
kg.add_entity("React", "technology", {"version": "18.2"})
kg.add_entity("TypeScript", "technology", {"version": "5.0"})

# 添加关系
kg.add_relation("张三", "leads", "project-alpha")
kg.add_relation("李四", "works_on", "project-alpha")
kg.add_relation("project-alpha", "uses", "React")
kg.add_relation("project-alpha", "uses", "TypeScript")
kg.add_relation("张三", "colleague_of", "李四")

# 多跳查询：张三的同事参与的项目
result = kg.multi_hop_query("张三", ["colleague_of", "works_on"])
print(f"结果: {result['results']}")  # ['project-alpha']
print(f"路径: {result['paths']}")

import time
import math
from typing import Optional
from dataclasses import dataclass, field

@dataclass
class MemoryItem:
    """记忆项：包含内容、元数据和生命周期信息"""
    id: str
    text: str
    memory_type: str  # "experience", "preference", "fact", "lesson"
    importance: float = 1.0
    created_at: float = field(default_factory=time.time)
    last_accessed: float = field(default_factory=time.time)
    access_count: int = 0
    embedding: Optional[list] = None
    is_compressed: bool = False
    original_text: Optional[str] = None  # 压缩前的原文

class MemoryManager:
    """记忆管理器：负责压缩、遗忘和生命周期管理"""
    
    def __init__(self, max_memories: int = 10000, 
                 decay_rate: float = 0.01):
        self.memories: dict[str, MemoryItem] = {}
        self.max_memories = max_memories
        self.decay_rate = decay_rate  # 时间衰减速率 λ
    
    def add(self, memory: MemoryItem) -> str:
        """添加新记忆，必要时触发压缩"""
        self.memories[memory.id] = memory
        
        # 如果超过容量上限，执行遗忘
        if len(self.memories) > self.max_memories:
            self._forget_least_important()
        
        return memory.id
    
    def access(self, memory_id: str) -> Optional[MemoryItem]:
        """检索记忆，更新访问统计"""
        if memory_id not in self.memories:
            return None
        
        mem = self.memories[memory_id]
        mem.last_accessed = time.time()
        mem.access_count += 1
        return mem
    
    def compute_relevance_score(self, memory: MemoryItem, 
                                 current_time: float = None) -> float:
        """
        计算记忆的相关性分数：
        综合考虑时间衰减、使用频率和初始重要性
        """
        current_time = current_time or time.time()
        
        # 时间衰减：指数衰减
        age_days = (current_time - memory.created_at) / 86400
        time_factor = math.exp(-self.decay_rate * age_days)
        
        # 使用频率衰减：对数增长（用得越多越重要）
        usage_factor = math.log1p(memory.access_count) / 5.0
        usage_factor = min(usage_factor, 1.0)  # 上限 1.0
        
        # 综合分数
        score = memory.importance * (0.6 * time_factor + 0.4 * usage_factor)
        return score
    
    def _forget_least_important(self):
        """遗忘相关性分数最低的记忆"""
        if len(self.memories) <= self.max_memories:
            return
        
        # 计算所有记忆的相关性分数
        scored = []
        now = time.time()
        for mid, mem in self.memories.items():
            score = self.compute_relevance_score(mem, now)
            scored.append((score, mid))
        
        # 按分数排序，删除最低的
        scored.sort(key=lambda x: x[0])
        to_remove = len(self.memories) - self.max_memories
        
        for _, mid in scored[:to_remove]:
            del self.memories[mid]
    
    def compress_memory(self, memory_id: str, 
                        compressed_text: str) -> bool:
        """
        压缩指定记忆：保留原文，替换为压缩版本
        """
        if memory_id not in self.memories:
            return False
        
        mem = self.memories[memory_id]
        mem.original_text = mem.text
        mem.text = compressed_text
        mem.is_compressed = True
        # 压缩后重要性提升（因为信息密度更高）
        mem.importance = min(mem.importance * 1.2, 2.0)
        return True
    
    def get_memory_stats(self) -> dict:
        """获取记忆库统计信息"""
        total = len(self.memories)
        compressed = sum(1 for m in self.memories.values() if m.is_compressed)
        avg_access = (sum(m.access_count for m in self.memories.values()) / total
                     if total > 0 else 0)
        now = time.time()
        avg_relevance = (sum(self.compute_relevance_score(m, now) 
                            for m in self.memories.values()) / total
                        if total > 0 else 0)
        
        return {
            "total": total,
            "compressed": compressed,
            "compression_rate": compressed / total if total > 0 else 0,
            "avg_access_count": round(avg_access, 2),
            "avg_relevance_score": round(avg_relevance, 4),
            "capacity_usage": total / self.max_memories
        }

# 使用示例
manager = MemoryManager(max_memories=100, decay_rate=0.005)

# 添加记忆
for i in range(50):
    manager.add(MemoryItem(
        id=f"mem_{i}",
        text=f"这是第 {i} 条记忆内容",
        memory_type="experience" if i % 3 == 0 else "fact",
        importance=0.5 + (i % 10) * 0.1
    ))

# 模拟部分记忆被频繁访问
for _ in range(20):
    manager.access("mem_0")
for _ in range(5):
    manager.access("mem_1")

# 压缩一条记忆
manager.compress_memory("mem_2", "关键：第2条的要点总结")

# 查看统计
stats = manager.get_memory_stats()
print(f"记忆库状态: {stats}")

// Agent 记忆系统的核心接口定义

interface MemoryItem {
  id: string;
  text: string;
  type: "experience" | "preference" | "fact" | "lesson" | "procedure";
  importance: number;       // 0.0 - 2.0
  source: string;           // 信息来源
  sourceCredibility: number; // 0.0 - 1.0
  createdAt: number;
  lastAccessed: number;
  accessCount: number;
  vector?: number[];        // 向量嵌入
  isCompressed: boolean;
  version: number;
}

interface MemoryQuery {
  text: string;             // 查询文本
  type?: string;            // 记忆类型过滤
  timeRange?: [number, number]; // 时间范围过滤
  maxResults?: number;      // 最大返回数
  requireGraph?: boolean;   // 是否需要图谱多跳推理
}

interface MemoryRetrievalResult {
  item: MemoryItem;
  score: number;            // 相关性分数
  source: "vector" | "graph" | "hybrid";
  reasoning?: string;       // 检索原因（可解释性）
}

class AgentMemorySystem {
  private vectorStore: VectorStore;
  private knowledgeGraph: KnowledgeGraph;
  private cache: Map<string, MemoryItem>;
  private management: MemoryManager;

  constructor(config: MemoryConfig) {
    this.vectorStore = new QdrantStore(config.vector);
    this.knowledgeGraph = new Neo4jGraph(config.graph);
    this.cache = new Map();
    this.management = new MemoryManager(config.management);
  }

  /** 添加记忆到系统 */
  async addMemory(item: Omit<MemoryItem, "id" | "createdAt" | "lastAccessed" | "accessCount">): Promise<string> {
    const memory: MemoryItem = {
      ...item,
      id: crypto.randomUUID(),
      createdAt: Date.now(),
      lastAccessed: Date.now(),
      accessCount: 0,
      version: 1,
    };

    // 1. 冲突检测
    const conflicts = await this.detectConflicts(memory);
    if (conflicts.length > 0) {
      // 智能合并冲突记忆
      await this.mergeMemories(memory, conflicts);
    }

    // 2. 向量化并存储
    const vector = await this.embed(memory.text);
    memory.vector = vector;
    await this.vectorStore.upsert(memory.id, vector, memory);

    // 3. 提取实体关系，更新图谱
    const entities = await this.extractEntities(memory.text);
    await this.knowledgeGraph.addMemoryNode(memory.id, entities, memory);

    // 4. 缓存热记忆
    this.cache.set(memory.id, memory);

    return memory.id;
  }

  /** 检索记忆：路由到合适的检索引擎 */
  async retrieve(query: MemoryQuery): Promise<MemoryRetrievalResult[]> {
    const results: MemoryRetrievalResult[] = [];

    // 并行查询
    const [vectorResults, graphResults] = await Promise.all([
      this.vectorSearch(query),
      query.requireGraph ? this.graphSearch(query) : Promise.resolve([]),
    ]);

    // 结果融合和重排序
    const merged = this.mergeAndRerank(vectorResults, graphResults);

    return merged.slice(0, query.maxResults ?? 5);
  }

  /** 记忆维护：压缩、遗忘、一致性检查 */
  async maintenance(): Promise<MaintenanceReport> {
    const compressed = await this.management.compressStaleMemories();
    const forgotten = await this.management.forgetExpiredMemories();
    const conflicts = await this.management.runConsistencyCheck();

    return { compressed, forgotten, conflictsFound: conflicts.length };
  }

  private async detectConflicts(newMemory: MemoryItem): Promise<MemoryItem[]> {
    // 基于元数据过滤 + 向量搜索检测冲突
    const similar = await this.vectorStore.searchSimilar(
      newMemory.vector!,
      { type: newMemory.type },
      10
    );

    return similar.filter(item => 
      item.sourceCredibility < newMemory.sourceCredibility &&
      this.isConflicting(item.text, newMemory.text)
    );
  }

  private async vectorSearch(query: MemoryQuery): Promise<MemoryRetrievalResult[]> {
    const vector = await this.embed(query.text);
    const results = await this.vectorStore.searchSimilar(vector, {
      type: query.type,
      timeRange: query.timeRange,
    }, 20);

    return results.map(item => ({
      item,
      score: this.computeRelevanceScore(item),
      source: "vector" as const,
    }));
  }

  private computeRelevanceScore(item: MemoryItem): number {
    const ageDays = (Date.now() - item.createdAt) / 86400000;
    const timeFactor = Math.exp(-0.01 * ageDays);
    const usageFactor = Math.log1p(item.accessCount) / 5.0;
    return item.importance * (0.6 * timeFactor + 0.4 * Math.min(usageFactor, 1.0));
  }
}