Python构建AI原生Agent的四层记忆系统架构

1. Python构建AI原生Agent的记忆系统架构

在构建AI原生Agent时，记忆系统是最核心的组件之一。它决定了Agent能否有效积累经验、理解上下文并做出合理决策。我们的记忆系统采用四层架构设计，模拟人类记忆的不同层次：

1.1 四层记忆架构设计

python复制HelloAgents记忆系统
├── 基础设施层 (Infrastructure Layer)
│   ├── MemoryManager - 记忆管理器（统一调度和协调）
│   ├── MemoryItem - 记忆数据结构（标准化记忆项）
│   ├── MemoryConfig - 配置管理（系统参数设置）
│   └── BaseMemory - 记忆基类（通用接口定义）
├── 记忆类型层 (Memory Types Layer)
│   ├── WorkingMemory - 工作记忆（临时信息，TTL管理）
│   ├── EpisodicMemory - 情景记忆（具体事件，时间序列）
│   ├── SemanticMemory - 语义记忆（抽象知识，图谱关系）
│   └── PerceptualMemory - 感知记忆（多模态数据）
├── 存储后端层 (Storage Backend Layer)
│   ├── QdrantVectorStore - 向量存储（高性能语义检索）
│   ├── Neo4jGraphStore - 图存储（知识图谱管理）
│   └── SQLiteDocumentStore - 文档存储（结构化持久化）
└── 嵌入服务层 (Embedding Service Layer)
    ├── DashScopeEmbedding - 通义千问嵌入（云端API）
    ├── LocalTransformerEmbedding - 本地嵌入（离线部署）
    └── TFIDFEmbedding - TFIDF嵌入（轻量级兜底）

这种分层设计的关键优势在于：

1. 模块化：各层职责明确，可独立扩展
2. 灵活性：可根据需求组合不同记忆类型
3. 性能优化：不同类型记忆采用最适合的存储方案

1.2 记忆类型实现细节

1.2.1 工作记忆实现

工作记忆相当于Agent的"短期记忆"，主要用于存储当前对话的上下文信息。我们采用纯内存存储方案，配合TTL机制进行自动清理：

python复制class WorkingMemory:
    def __init__(self, config: MemoryConfig):
        self.max_capacity = config.working_memory_capacity or 50
        self.max_age_minutes = config.working_memory_ttl or 60
        self.memories = []

    def retrieve(self, query: str, limit: int = 5) -> List[MemoryItem]:
        """混合检索：TF-IDF向量化 + 关键词匹配"""
        self._expire_old_memories()
        
        # 计算综合分数
        scored_memories = []
        for memory in self.memories:
            vector_score = vector_scores.get(memory.id, 0.0)
            keyword_score = self._calculate_keyword_score(query, memory.content)
            
            # 混合评分公式
            base_relevance = vector_score * 0.7 + keyword_score * 0.3 if vector_score > 0 else keyword_score
            time_decay = self._calculate_time_decay(memory.timestamp)
            importance_weight = 0.8 + (memory.importance * 0.4)
            
            final_score = base_relevance * time_decay * importance_weight
            scored_memories.append((final_score, memory))

重要提示：工作记忆的评分算法是关键，我们采用(相似度 × 时间衰减) × (0.8 + 重要性 × 0.4)的公式，确保检索结果既相关又重要。

1.2.2 情景记忆实现

情景记忆用于存储具体事件和经历，采用SQLite+Qdrant的混合存储方案：

python复制class EpisodicMemory:
    def _calculate_episode_score(self, hit) -> float:
        """情景记忆评分算法"""
        vec_score = float(hit.get("score", 0.0))
        recency_score = self._calculate_recency(hit["metadata"]["timestamp"])
        importance = hit["metadata"].get("importance", 0.5)
        
        # 评分公式：(向量相似度 × 0.8 + 时间近因性 × 0.2) × 重要性权重
        base_relevance = vec_score * 0.8 + recency_score * 0.2
        importance_weight = 0.8 + (importance * 0.4)
        
        return base_relevance * importance_weight

1.2.3 语义记忆实现

语义记忆采用Neo4j图数据库+Qdrant向量数据库的混合架构：

python复制class SemanticMemory(BaseMemory):
    def _combine_and_rank_results(self, vector_results, graph_results, query, limit):
        """混合排序结果"""
        for memory_id, result in combined.items():
            vector_score = result["vector_score"]
            graph_score = result["graph_score"]
            importance = result.get("importance", 0.5)
            
            # 最终得分：相似度 * 重要性权重
            combined_score = (vector_score * 0.7 + graph_score * 0.3) * (0.8 + importance * 0.4)
            result["combined_score"] = combined_score

2. RAG系统设计与实现

2.1 RAG系统架构

python复制HelloAgents RAG系统
├── 文档处理层 (Document Processing Layer)
│   ├── DocumentProcessor - 文档处理器（多格式解析）
│   ├── Document - 文档对象（元数据管理）
│   └── Pipeline - RAG管道（端到端处理）
├── 嵌入表示层 (Embedding Layer)
│   └── 统一嵌入接口 - 复用记忆系统的嵌入服务
├── 向量存储层 (Vector Storage Layer)
│   └── QdrantVectorStore - 向量数据库（命名空间隔离）
└── 智能问答层 (Intelligent Q&A Layer)
    ├── 多策略检索 - 向量检索 + MQE + HyDE
    ├── 上下文构建 - 智能片段合并与截断
    └── LLM增强生成 - 基于上下文的准确问答

2.2 文档处理流程

2.2.1 文档加载与转换

我们使用MarkItDown将各种格式文档统一转换为Markdown：

python复制def _convert_to_markdown(path: str) -> str:
    """将任意格式文档转换为Markdown文本"""
    if not os.path.exists(path):
        return ""
    
    # 对PDF文件使用增强处理
    if path.lower().endswith('.pdf'):
        return _enhanced_pdf_processing(path)
    
    # 其他格式使用MarkItDown统一转换
    try:
        result = md_instance.convert(path)
        return result.text_content or ""
    except Exception:
        return _fallback_text_reader(path)

2.2.2 智能分块策略

基于Markdown标题结构的智能分块：

python复制def _split_paragraphs_with_headings(text: str) -> List[Dict]:
    """根据标题层次分割段落，保持语义完整性"""
    lines = text.splitlines()
    heading_stack = []
    paragraphs = []
    
    for ln in lines:
        if ln.strip().startswith("#"):
            # 处理标题行
            level = len(ln) - len(ln.lstrip('#'))
            title = ln.lstrip('#').strip()
            
            if level <= len(heading_stack):
                heading_stack = heading_stack[:level-1]
            heading_stack.append(title)

2.3 RAG工具实现

python复制class RAGTool(Tool):
    def __init__(
        self,
        knowledge_base_path: str = "./knowledge_base",
        qdrant_url: str = None,
        qdrant_api_key: str = None,
        collection_name: str = "rag_knowledge_base"
    ):
        # 初始化RAG管道
        self._pipelines = {}
        self.llm = HelloAgentsLLM()
        
        # 创建默认管道
        default_pipeline = create_rag_pipeline(
            qdrant_url=qdrant_url,
            qdrant_api_key=qdrant_api_key,
            collection_name=collection_name
        )
        self._pipelines["default"] = default_pipeline

3. 核心工具与接口设计

3.1 MemoryTool统一接口

python复制class MemoryTool(Tool):
    def execute(self, action: str, **kwargs) -> str:
        """执行记忆操作"""
        if action == "add":
            return self._add_memory(**kwargs)
        elif action == "search":
            return self._search_memory(**kwargs)
        elif action == "summary":
            return self._get_summary(**kwargs)

3.2 记忆添加实现

python复制def _add_memory(
    self,
    content: str = "",
    memory_type: str = "working",
    importance: float = 0.5,
    **metadata
) -> str:
    """添加记忆"""
    # 确保会话ID存在
    if self.current_session_id is None:
        self.current_session_id = f"session_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
    
    # 添加会话信息到元数据
    metadata.update({
        "session_id": self.current_session_id,
        "timestamp": datetime.now().isoformat()
    })
    
    return self.memory_manager.add_memory(
        content=content,
        memory_type=memory_type,
        importance=importance,
        metadata=metadata
    )

3.3 记忆检索实现

python复制def _search_memory(
    self,
    query: str,
    limit: int = 5,
    memory_types: List[str] = None
) -> str:
    """搜索记忆"""
    results = self.memory_manager.retrieve_memories(
        query=query,
        limit=limit,
        memory_types=memory_types
    )
    
    # 格式化结果
    formatted = []
    for i, memory in enumerate(results, 1):
        content_preview = memory.content[:80] + "..." if len(memory.content) > 80 else memory.content
        formatted.append(
            f"{i}. [{memory.memory_type}] {content_preview} (重要性: {memory.importance:.2f})"
        )
    
    return "\n".join(formatted)

4. 实战经验与优化建议

4.1 性能优化技巧

1. 批量操作：对记忆的添加和检索尽量使用批量接口
2. 缓存策略：对频繁访问的记忆内容实现本地缓存
3. 异步处理：耗时的记忆操作使用异步任务处理

4.2 常见问题排查

1. 记忆检索不准确：

   • 检查嵌入模型是否匹配
   • 验证评分算法参数是否合理
   • 确认记忆内容是否完整存储

2. 系统响应慢：

   • 检查向量数据库连接
   • 监控内存使用情况
   • 优化检索时的limit参数

3. 记忆丢失问题：

   • 验证存储后端连接
   • 检查TTL设置是否合理
   • 确认重要记忆的importance值是否足够高

4.3 扩展建议

1. 多模态支持：扩展感知记忆处理图像、音频等非文本数据
2. 记忆压缩：实现记忆的自动摘要和压缩
3. 记忆关联：建立记忆之间的关联关系网络
4. 记忆迁移：支持记忆在不同Agent间的迁移和共享

在实际项目中，我们发现记忆系统的性能瓶颈往往出现在向量检索环节。针对这个问题，我们采取了以下优化措施：

1. 对高频查询实现缓存
2. 使用更高效的向量索引算法
3. 对检索结果进行预过滤
4. 实现分级检索策略

另一个关键点是记忆的重要性评估。我们开发了自动评估算法，结合以下因素计算记忆的重要性：

1. 使用频率
2. 关联记忆数量
3. 用户显式标记
4. 时间衰减因子

这些实践经验使我们的记忆系统在实际应用中表现出色，能够支持复杂的Agent交互场景。