Agent 记忆系统架构实战：从向量数据库到知识图谱的混合记忆方案设计与实现

记忆系统需要处理三类数据：短期对话（Session）、长期事实（Facts）、关系网络（Relations）。每类数据对应不同的存储策略：

from pydantic import BaseModel, Field
from typing import List, Optional
from enum import Enum

class MemoryType(str, Enum):
    SHORT_TERM = "short_term"      # 会话级记忆（向量存储）
    LONG_TERM = "long_term"        # 持久化事实（向量 + 图谱）
    RELATION = "relation"          # 实体关系（仅图谱）

class Memory(BaseModel):
    id: str
    type: MemoryType
    content: str
    embedding: Optional[List[float]] = None
    entities: List[str] = Field(default_factory=list)  # 提取的实体
    timestamp: float
    ttl_hours: Optional[int] = None  # None = 永久

使用 Qdrant 作为向量存储，支持 HNSW 索引与 Payload 过滤：

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

# 初始化客户端
client = QdrantClient(host="localhost", port=6333)

# 创建 Collection（1536 维 = text-embedding-3-large）
COLLECTION_NAME = "agent_memory"
client.create_collection(
    collection_name=COLLECTION_NAME,
    vectors_config=VectorParams(size=1536, distance=Distance.COSINE),
)

# 创建 Payload 索引（用于按类型过滤）
client.create_payload_index(
    collection_name=COLLECTION_NAME,
    field_name="memory_type",
    field_schema="keyword"
)

写入记忆时需同时存储向量与 Payload 元数据：

from openai import OpenAI

openai_client = OpenAI()

def embed(text: str) -> List[float]:
    """生成 1536 维 Embedding"""
    response = openai_client.embeddings.create(
        model="text-embedding-3-large",
        input=text
    )
    return response.data[0].embedding

def store_memory(memory: Memory):
    # 生成向量
    if memory.embedding is None:
        memory.embedding = embed(memory.content)

    # 写入 Qdrant
    client.upsert(
        collection_name=COLLECTION_NAME,
        points=[
            PointStruct(
                id=hash(memory.id) % (2**63),
                vector=memory.embedding,
                payload=memory.model_dump()
            )
        ]
    )

使用 Neo4j 存储实体及其关系。核心设计：

• 节点（Node）：Entity（带 name、type 属性）
• 关系（Relationship）：RELATED_TO、PART_OF、CAUSES、USES 等

from neo4j import GraphDatabase

class GraphMemory:
    def __init__(self, uri: str, user: str, password: str):
        self.driver = GraphDatabase.driver(uri, auth=(user, password))

    def close(self):
        self.driver.close()

    def add_entity(self, name: str, type: str):
        """添加实体节点"""
        with self.driver.session() as session:
            session.run(
                "MERGE (e:Entity {name: $name}) SET e.type = $type",
                name=name, type=type
            )

    def add_relation(self, from_entity: str, to_entity: str, rel_type: str):
        """添加实体关系"""
        with self.driver.session() as session:
            session.run(
                f"""
                MATCH (a:Entity {{name: $from}})
                MATCH (b:Entity {{name: $to}})
                MERGE (a)-[:{rel_type}]->(b)
                """,
                from=from_entity, to=to_entity
            )

从对话中提取实体与关系的简单实现（实际项目可用 LLM 提取）：

import re

def extract_entities(text: str) -> List[str]:
    """简单规则提取：大写字母开头的词、技术术语"""
    # 简化示例：提取引号内的内容
    return re.findall(r'"([^"]+)"', text)

def build_graph_from_memory(memory: Memory, graph: GraphMemory):
    """将记忆中的实体写入图谱"""
    entities = extract_entities(memory.content)

    # 添加实体
    for entity in entities:
        graph.add_entity(entity, "concept")

    # 简单关系：如果两个实体在同一段记忆中出现，建立弱相关关系
    for i, e1 in enumerate(entities):
        for e2 in entities[i+1:]:
            graph.add_relation(e1, e2, "RELATED_TO")

混合路由器的核心逻辑：根据查询类型决定检索策略。

from typing import Literal

class QueryType(str, Enum):
    FACTUAL = "factual"         # 事实型：图谱优先
    SEMANTIC = "semantic"       # 语义型：向量优先
    RELATIONAL = "relational"   # 关系型：仅图谱

def classify_query(query: str) -> QueryType:
    """简单规则分类（生产环境用 LLM 分类器）"""
    query_lower = query.lower()

    # 关系型关键词
    if any(w in query_lower for w in ["关系", "连接", "属于", "依赖"]):
        return QueryType.RELATIONAL

    # 事实型关键词
    if any(w in query_lower for w in ["什么是", "定义", "原理", "架构"]):
        return QueryType.FACTUAL

    return QueryType.SEMANTIC

class HybridRouter:
    def __init__(self, vector_client: QdrantClient, graph: GraphMemory):
        self.vector_client = vector_client
        self.graph = graph

    def search(self, query: str, top_k: int = 5) -> List[dict]:
        query_type = classify_query(query)

        if query_type == QueryType.RELATIONAL:
            return self.graph_search(query)
        elif query_type == QueryType.FACTUAL:
            # 双路检索：向量 + 图谱
            vector_results = self.vector_search(query, top_k // 2)
            graph_results = self.graph_search(query)
            return self.merge_results(vector_results, graph_results)
        else:
            return self.vector_search(query, top_k)

向量检索实现：

def vector_search(self, query: str, top_k: int) -> List[dict]:
    query_embedding = embed(query)

    results = self.vector_client.search(
        collection_name=COLLECTION_NAME,
        query_vector=query_embedding,
        limit=top_k
    )

    return [
        {"source": "vector", "score": r.score, "content": r.payload["content"]}
        for r in results
    ]

图谱检索实现（遍历相关实体）：

def graph_search(self, query: str) -> List[dict]:
    """从图谱中检索相关实体与关系"""
    # 简单实现：提取查询中的实体，查找其关联
    entities = extract_entities(query)

    if not entities:
        return []

    with self.graph.driver.session() as session:
        cypher = """
        MATCH (e:Entity)-[r]-(related:Entity)
        WHERE e.name IN $entities
        RETURN e.name, type(r) as rel_type, related.name as related_name
        LIMIT 20
        """
        result = session.run(cypher, entities=entities)

        return [
            {
                "source": "graph",
                "content": f"{record['e.name']} -[{record['rel_type']}]-> {record['related_name']}"
            }
            for record in result
        ]

短期记忆会随时间膨胀，需要定期压缩并转移到长期存储：

from openai import OpenAI

class MemoryCompressor:
    def __init__(self):
        self.llm = OpenAI()

    def compress_session(self, memories: List[Memory]) -> Memory:
        """压缩多段会话记忆为摘要"""
        content = "\n".join(m.content for m in memories)

        response = self.llm.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": "将以下对话压缩为 3-5 个关键事实，每句一行。"},
                {"role": "user", "content: content}
            ]
        )

        return Memory(
            id=f"summary_{memories[0].id}",
            type=MemoryType.LONG_TERM,
            content=response.choices[0].message.content,
            timestamp=time.time(),
            ttl_hours=None  # 永久保存
        )

    def cleanup_expired(self):
        """清理过期记忆"""
        cutoff = time.time() - (24 * 3600)  # 24 小时前

        expired = client.scroll(
            collection_name=COLLECTION_NAME,
            scroll_filter=Filter(
                must=[FieldCondition(key="ttl_hours", match=MatchValue(value=24))]
            )
        )

        for point in expired:
            if point.payload["timestamp"] < cutoff:
                client.delete(
                    collection_name=COLLECTION_NAME,
                    points_selector=[point.id]
                )

将混合记忆系统集成为 LangChain 的 BaseMemory 接口：

from langchain.memory import BaseMemory
from typing import Dict, List

class HybridMemory(BaseMemory):
    router: HybridRouter
    compressor: MemoryCompressor
    session_memories: List[Memory] = []

    class Config:
        arbitrary_types_allowed = True

    def save_context(self, inputs: Dict, outputs: Dict):
        """保存对话上下文"""
        user_input = inputs.get("input", "")
        agent_output = outputs.get("output", "")

        # 写入短期记忆
        self.session_memories.extend([
            Memory(
                id=f"session_{len(self.session_memories)}",
                type=MemoryType.SHORT_TERM,
                content=user_input,
                timestamp=time.time(),
                ttl_hours=24
            ),
            Memory(
                id=f"session_{len(self.session_memories)+1}",
                type=MemoryType.SHORT_TERM,
                content=agent_output,
                timestamp=time.time(),
                ttl_hours=24
            )
        ])

        # 每 10 轮对话压缩一次
        if len(self.session_memories) >= 20:
            compressed = self.compressor.compress_session(self.session_memories)
            store_memory(compressed)
            self.session_memories = []

    def load_memory_variables(self, inputs: Dict) -> Dict:
        """加载相关记忆"""
        query = inputs.get("input", "")
        results = self.router.search(query, top_k=5)

        return {
            "history": "\n".join(r["content"] for r in results)
        }

使用示例：

from langchain.agents import AgentExecutor, create_openai_tools_agent

# 初始化混合记忆
hybrid_memory = HybridMemory(
    router=HybridRouter(client, graph),
    compressor=MemoryCompressor()
)

# 创建 Agent
agent = create_openai_tools_agent(llm, tools, prompt)
executor = AgentExecutor(
    agent=agent,
    tools=tools,
    memory=hybrid_memory,  # 注入记忆
    verbose=True
)

# 运行
response = executor.invoke({"input": "我之前部署的 MCP 服务器在哪里？"})
print(response["output"])