A-Mem扩展开发如何自定义记忆组织和检索算法的完整指南【免费下载链接】A-memThe code for NeurIPS 2025 paper A-Mem: Agentic Memory for LLM Agents项目地址: https://gitcode.com/gh_mirrors/ag/A-memA-MemAgentic Memory是一个革命性的LLM智能体记忆系统它通过动态记忆组织和智能检索算法为大型语言模型智能体提供了强大的记忆管理能力。本文将为您提供一份全面的A-Mem扩展开发指南帮助您深入了解如何自定义记忆组织和检索算法构建更智能的AI代理系统。 A-Mem核心架构解析A-Mem系统基于创新的智能体记忆架构将传统静态记忆系统转变为动态、可演化的知识网络。系统核心由三个主要组件构成记忆笔记MemoryNote- 基本记忆单元包含内容、关键词、标签和上下文元数据智能体记忆系统AgenticMemorySystem- 管理记忆的添加、检索和演化检索器Retriever- 负责相似度计算和记忆查找A-Mem系统框架展示了LLM智能体与记忆组件之间的动态交互 自定义记忆组织算法1. 修改记忆分析逻辑在memory_layer.py文件中MemoryNote类的analyze_content方法负责分析新记忆内容并提取元数据。您可以通过修改提示模板来自定义提取逻辑# 自定义分析提示模板 custom_analysis_prompt 分析以下内容并提取 1. 核心概念至少5个关键词 2. 情感倾向正面/负面/中性 3. 相关领域技术/商业/学术等 4. 紧急程度高/中/低 内容{content} 2. 调整记忆演化策略记忆演化是A-Mem的核心特性允许系统动态调整记忆之间的关系。在AgenticMemorySystem类的process_memory方法中您可以自定义演化决策逻辑# 自定义演化阈值和策略 class CustomMemorySystem(AgenticMemorySystem): def __init__(self, custom_evo_threshold50, **kwargs): super().__init__(**kwargs) self.evo_threshold custom_evo_threshold # 自定义演化提示模板 self.evolution_system_prompt 您是一个智能记忆演化代理... 自定义演化逻辑 3. 实现分层记忆结构A-Mem支持分层记忆组织。您可以通过扩展MemoryNote类来添加更多层级class HierarchicalMemoryNote(MemoryNote): def __init__(self, **kwargs): super().__init__(**kwargs) self.parent_id None # 父级记忆ID self.child_ids [] # 子级记忆ID self.hierarchy_level 0 # 层级深度 self.cluster_id None # 聚类标识 自定义检索算法1. 混合检索策略A-Mem默认使用基于嵌入的检索但您可以轻松实现混合检索策略。在memory_layer.py中SimpleEmbeddingRetriever类提供了基础的检索功能from memory_layer import SimpleEmbeddingRetriever class HybridRetriever(SimpleEmbeddingRetriever): def __init__(self, model_nameall-MiniLM-L6-v2, bm25_weight0.3): super().__init__(model_name) self.bm25_weight bm25_weight self.bm25_index None def add_documents(self, documents): super().add_documents(documents) # 构建BM25索引 tokenized_docs [simple_tokenize(doc) for doc in documents] self.bm25_index BM25Okapi(tokenized_docs) def search(self, query, k5): # 嵌入相似度得分 embedding_scores super().search(query, k) # BM25得分 tokenized_query simple_tokenize(query) bm25_scores self.bm25_index.get_scores(tokenized_query) # 混合得分 combined_scores [] for i, (emb_score, bm25_score) in enumerate(zip(embedding_scores, bm25_scores)): combined_score (1 - self.bm25_weight) * emb_score self.bm25_weight * bm25_score combined_scores.append((i, combined_score)) # 按得分排序并返回top-k combined_scores.sort(keylambda x: x[1], reverseTrue) return [idx for idx, _ in combined_scores[:k]]2. 基于时间的检索优化对于时间敏感的记忆您可以实现时间加权检索算法class TemporalAwareRetriever(SimpleEmbeddingRetriever): def __init__(self, model_nameall-MiniLM-L6-v2, recency_weight0.2): super().__init__(model_name) self.recency_weight recency_weight self.timestamps [] def add_documents(self, documents, timestampsNone): super().add_documents(documents) current_time datetime.now() if timestamps: self.timestamps timestamps else: self.timestamps [current_time] * len(documents) def calculate_recency_score(self, timestamp): 计算时间衰减分数 time_diff (datetime.now() - timestamp).total_seconds() # 指数衰减越新的记忆分数越高 return math.exp(-time_diff / (30 * 24 * 3600)) # 30天半衰期 def search(self, query, k5): semantic_scores super().search(query, k) # 结合时间衰减 combined_scores [] for idx in semantic_scores: recency_score self.calculate_recency_score(self.timestamps[idx]) # 获取语义相似度需要实际计算 semantic_score 1.0 # 这里需要实际计算 combined_score (1 - self.recency_weight) * semantic_score self.recency_weight * recency_score combined_scores.append((idx, combined_score)) combined_scores.sort(keylambda x: x[1], reverseTrue) return [idx for idx, _ in combined_scores[:k]]3. 上下文感知检索实现基于上下文的检索考虑记忆之间的关联性class ContextAwareRetriever(SimpleEmbeddingRetriever): def __init__(self, model_nameall-MiniLM-L6-v2, context_window3): super().__init__(model_name) self.context_window context_window self.memory_graph {} # 记忆关联图 def add_context_links(self, memory_id, related_ids): 添加记忆关联 if memory_id not in self.memory_graph: self.memory_graph[memory_id] set() self.memory_graph[memory_id].update(related_ids) def search_with_context(self, query, current_context_ids, k5): 考虑上下文的检索 # 基础语义检索 base_results super().search(query, k * 2) # 上下文增强 context_boosted [] for idx in base_results: score 1.0 # 基础相似度得分 # 检查是否在上下文关联中 if idx in self.memory_graph: context_overlap len(self.memory_graph[idx].intersection(current_context_ids)) if context_overlap 0: # 上下文关联增强 score * (1 0.1 * context_overlap) context_boosted.append((idx, score)) context_boosted.sort(keylambda x: x[1], reverseTrue) return [idx for idx, _ in context_boosted[:k]] 实现自定义记忆演化算法1. 智能记忆合并策略在memory_layer.py的process_memory方法基础上实现更智能的记忆合并def intelligent_memory_merge(self, new_note, similar_notes): 智能记忆合并算法 merged_content new_note.content for note in similar_notes: similarity self.calculate_similarity(new_note.content, note.content) if similarity 0.8: # 高相似度阈值 # 合并内容 merged_content self.merge_contents(merged_content, note.content) # 更新标签 new_note.tags list(set(new_note.tags note.tags)) # 建立双向链接 new_note.links.append(note.id) note.links.append(new_note.id) new_note.content merged_content return new_note2. 自适应演化阈值实现根据记忆系统状态自动调整的演化阈值class AdaptiveEvolutionSystem(AgenticMemorySystem): def __init__(self, **kwargs): super().__init__(**kwargs) self.evolution_history [] self.adaptive_threshold self.evo_threshold def calculate_evolution_need(self): 计算演化需求 memory_count len(self.memories) avg_retrieval sum(m.retrieval_count for m in self.memories.values()) / max(memory_count, 1) # 动态调整阈值 if memory_count 100 and avg_retrieval 5: # 记忆多但使用少需要更多演化 self.adaptive_threshold max(10, self.evo_threshold // 2) elif memory_count 50: # 记忆少减少演化频率 self.adaptive_threshold self.evo_threshold * 2 else: self.adaptive_threshold self.evo_threshold def process_memory(self, note): 重写处理逻辑加入自适应调整 self.calculate_evolution_need() # 调用父类逻辑 return super().process_memory(note) 性能优化技巧1. 记忆索引优化class OptimizedMemorySystem(AgenticMemorySystem): def __init__(self, **kwargs): super().__init__(**kwargs) self.content_index {} # 内容倒排索引 self.tag_index {} # 标签索引 self.keyword_index {} # 关键词索引 def add_note(self, content, **kwargs): note_id super().add_note(content, **kwargs) note self.memories[note_id] # 更新索引 self.update_indexes(note) return note_id def update_indexes(self, note): 更新所有索引 # 内容索引 for word in simple_tokenize(note.content): if word not in self.content_index: self.content_index[word] set() self.content_index[word].add(note.id) # 标签索引 for tag in note.tags: if tag not in self.tag_index: self.tag_index[tag] set() self.tag_index[tag].add(note.id) # 关键词索引 for keyword in note.keywords: if keyword not in self.keyword_index: self.keyword_index[keyword] set() self.keyword_index[keyword].add(note.id)2. 缓存策略实现from functools import lru_cache from datetime import datetime, timedelta class CachedRetrievalSystem(AgenticMemorySystem): def __init__(self, cache_ttl300, **kwargs): # 5分钟缓存 super().__init__(**kwargs) self.cache_ttl cache_ttl self.query_cache {} lru_cache(maxsize1000) def find_related_memories_cached(self, query: str, k: int 5): 带缓存的记忆查找 cache_key f{query}_{k} if cache_key in self.query_cache: cached_result, timestamp self.query_cache[cache_key] if datetime.now() - timestamp timedelta(secondsself.cache_ttl): return cached_result # 执行实际检索 result super().find_related_memories(query, k) # 更新缓存 self.query_cache[cache_key] (result, datetime.now()) return result 测试与验证1. 创建测试套件在test_advanced_robust.py的基础上添加自定义算法的测试def test_custom_retrieval(): 测试自定义检索算法 system CustomMemorySystem() # 添加测试记忆 test_memories [ Python编程基础教程, 机器学习算法原理, 深度学习神经网络, 自然语言处理技术, 计算机视觉应用 ] for memory in test_memories: system.add_note(memory) # 测试检索 query 人工智能学习 results system.find_related_memories(query, k3) print(f查询: {query}) print(f检索结果: {results}) # 验证结果 assert len(results) 3, 应返回3个相关记忆2. 性能基准测试import time from memory_layer import AgenticMemorySystem def benchmark_retrieval_performance(): 检索性能基准测试 system AgenticMemorySystem() # 批量添加记忆 start_time time.time() for i in range(1000): system.add_note(f测试记忆内容 {i}) add_time time.time() - start_time # 检索性能测试 start_time time.time() for _ in range(100): system.find_related_memories(测试查询, k5) retrieval_time time.time() - start_time print(f添加1000个记忆耗时: {add_time:.2f}秒) print(f100次检索平均耗时: {retrieval_time/100:.4f}秒) print(f总记忆数: {len(system.memories)}) 集成到现有系统1. 与现有LLM框架集成A-Mem可以轻松集成到现有的LLM应用中from memory_layer import AgenticMemorySystem from langchain.llms import OpenAI from langchain.chains import ConversationChain class AIChatbotWithMemory: def __init__(self, llm_modelgpt-4): self.llm OpenAI(modelllm_model) self.memory_system AgenticMemorySystem() self.conversation_history [] def chat(self, user_input): # 检索相关记忆 related_memories self.memory_system.find_related_memories(user_input, k3) # 构建上下文 context self.build_context(user_input, related_memories) # 生成回复 response self.llm.generate(context) # 保存对话到记忆 self.memory_system.add_note(f用户: {user_input}\n助手: {response}) # 更新对话历史 self.conversation_history.append({ user: user_input, assistant: response, timestamp: datetime.now() }) return response def build_context(self, query, memories): 构建包含记忆的上下文 memory_context \n.join([f相关记忆 {i1}: {mem} for i, mem in enumerate(memories)]) return f 对话历史最近3条: {self.get_recent_history()} 相关记忆: {memory_context} 当前查询: {query} 请基于以上信息生成有帮助的回复: 2. 多智能体协作系统class MultiAgentSystem: def __init__(self, num_agents3): self.agents [] self.shared_memory AgenticMemorySystem() # 初始化多个智能体 for i in range(num_agents): agent { id: fagent_{i}, specialty: [技术, 商业, 创意][i % 3], personal_memory: AgenticMemorySystem(), contribution_count: 0 } self.agents.append(agent) def collaborative_problem_solving(self, problem): 协作问题解决 solutions [] for agent in self.agents: # 从共享记忆中检索相关信息 relevant_memories self.shared_memory.find_related_memories(problem, k2) # 从个人记忆中检索 personal_memories agent[personal_memory].find_related_memories(problem, k2) # 生成解决方案 solution self.generate_solution(agent, problem, relevant_memories personal_memories) solutions.append(solution) # 将解决方案添加到共享记忆 self.shared_memory.add_note(f智能体{agent[id]}的解决方案: {solution}) agent[contribution_count] 1 # 整合最佳解决方案 best_solution self.consolidate_solutions(solutions) return best_solution 监控与调试1. 记忆系统状态监控class MemorySystemMonitor: def __init__(self, memory_system): self.memory_system memory_system self.metrics { total_memories: 0, avg_retrieval_count: 0, evolution_count: 0, memory_size_distribution: {}, retrieval_latency: [] } def collect_metrics(self): 收集系统指标 memories self.memory_system.memories self.metrics[total_memories] len(memories) if memories: retrieval_counts [m.retrieval_count for m in memories.values()] self.metrics[avg_retrieval_count] sum(retrieval_counts) / len(retrieval_counts) # 记忆大小分布 size_groups {small: 0, medium: 0, large: 0} for memory in memories.values(): size len(memory.content) if size 100: size_groups[small] 1 elif size 500: size_groups[medium] 1 else: size_groups[large] 1 self.metrics[memory_size_distribution] size_groups return self.metrics def generate_report(self): 生成监控报告 metrics self.collect_metrics() report f A-Mem系统监控报告 基本信息: - 总记忆数量: {metrics[total_memories]} - 平均检索次数: {metrics[avg_retrieval_count]:.2f} - 演化次数: {self.memory_system.evo_cnt} 记忆大小分布: - 小型记忆 (100字符): {metrics[memory_size_distribution].get(small, 0)} - 中型记忆 (100-500字符): {metrics[memory_size_distribution].get(medium, 0)} - 大型记忆 (500字符): {metrics[memory_size_distribution].get(large, 0)} 检索性能: - 最近检索延迟: {metrics[retrieval_latency][-5:] if metrics[retrieval_latency] else 无数据} 建议: {self.generate_recommendations(metrics)} return report2. 调试工具class MemorySystemDebugger: def __init__(self, memory_system): self.system memory_system def debug_memory_retrieval(self, query, k5): 调试记忆检索过程 print(f调试检索: {query}) print( * 50) # 显示所有记忆 print(f系统中共有 {len(self.system.memories)} 个记忆:) for idx, (mem_id, memory) in enumerate(self.system.memories.items()): print(f{idx1}. ID: {mem_id}) print(f 内容: {memory.content[:50]}...) print(f 标签: {memory.tags}) print(f 关键词: {memory.keywords}) print() # 执行检索 print(f执行检索 (k{k}):) results self.system.find_related_memories(query, k) print(f检索结果:) for i, result in enumerate(results): print(f{i1}. {result}) return results def analyze_memory_evolution(self): 分析记忆演化模式 evolution_patterns {} for memory in self.system.memories.values(): if memory.evolution_history: pattern { initial_tags: memory.evolution_history[0].get(tags, []), current_tags: memory.tags, evolution_count: len(memory.evolution_history), links_count: len(memory.links) } evolution_patterns[memory.id] pattern print(记忆演化分析:) print( * 50) for mem_id, pattern in evolution_patterns.items(): print(f记忆 {mem_id}:) print(f 初始标签: {pattern[initial_tags]}) print(f 当前标签: {pattern[current_tags]}) print(f 演化次数: {pattern[evolution_count]}) print(f 链接数量: {pattern[links_count]}) print() 最佳实践总结1. 设计原则模块化设计: 将记忆组织、检索和演化逻辑分离便于独立测试和优化可配置性: 通过参数化设计支持不同场景的灵活配置可扩展性: 预留接口方便添加新的检索算法或演化策略性能监控: 内置性能指标收集便于优化和调试2. 实现建议从小规模开始: 先实现核心功能再逐步添加高级特性充分测试: 为每个自定义算法编写单元测试性能优化: 关注内存使用和检索延迟适时引入缓存文档完善: 为自定义功能提供清晰的API文档和使用示例3. 部署考虑内存管理: 大规模记忆系统需要有效的内存管理策略持久化存储: 实现记忆的持久化存储和加载机制并发安全: 多线程环境下的线程安全设计监控告警: 系统健康状态监控和异常告警 结语A-Mem系统为LLM智能体提供了强大的记忆管理能力而其真正的价值在于其可扩展性和自定义能力。通过本文提供的完整指南您可以深入理解A-Mem的核心架构和工作原理掌握自定义记忆组织和检索算法的关键技术实现适应特定需求的智能记忆演化策略构建高性能、可扩展的记忆管理系统将A-Mem无缝集成到现有的AI应用中无论是构建对话系统、知识管理工具还是复杂的多智能体协作平台A-Mem的自定义扩展能力都能为您提供强大的技术支撑。开始您的A-Mem扩展开发之旅构建更智能、更强大的AI应用吧传统记忆系统上图与A-Mem智能体记忆系统下图的对比A-Mem系统实现了动态记忆操作和灵活的智能体-记忆交互【免费下载链接】A-memThe code for NeurIPS 2025 paper A-Mem: Agentic Memory for LLM Agents项目地址: https://gitcode.com/gh_mirrors/ag/A-mem创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考