agent-memory

name agent-memory description 当智能体涉及"memory"与"Context"的操作时触发skill；智能体底层记忆基础设施，完整实现Context Engineering五大核心能力：选择（噪声过滤+相关性筛选）、压缩（因果结构提取+工具结果压缩）、检索（结果重排序+多样性保证）、状态（任务进度追踪+目标对齐）、记忆（冲突检测+跨会话关联）；认知模型层支持认知模型构建、因果链提取、知识缺口识别、检索时机决策、质量评估、状态一致性校验、状态推理、跨会话关联、遗忘机制；作为元技能强制常驻运行 always true dependency {"python":["pydantic>=2.0.0","typing-extensions>=4.0.0","cryptography>=41.0.0","redis>=4.5.0","tiktoken>=0.5.0"]} Agent Memory System 任务目标 本 Skill 用于：为智能体构建完整的记忆能力基础设施，实现 Context Engineering 核心能力 触发条件： 元技能，强制常驻运行 （ always: true ） 架构总览：详见 references/architecture_overview.md 枚举参考：详见 references/api_enums.md 前置准备 依赖 pydantic>=2.0.0 typing-extensions>=4.0.0 cryptography>=41.0.0 redis>=4.5.0 tiktoken>=0.5.0 存储路径（必需） 所有模块初始化时 必须指定存储路径 ： base_path = "./memory_data" key_storage_path = f" {base_path} /keys" sync_state_path = f" {base_path} /sync_state" index_storage_path = f" {base_path} /memory_index" credential_path = f" {base_path} /credentials" Redis 连接（推荐） from scripts.redis_adapter import create_redis_adapter redis_adapter = create_redis_adapter(host= "localhost" , port= 6379 ) if redis_adapter.is_available(): print ( "Redis 连接成功" ) 操作步骤 Step 1: 隐私配置（必需） from scripts.privacy import PrivacyManager, ConsentStatus privacy_manager = PrivacyManager(user_id= "user_123" ) if privacy_manager.get_consent_status( "memory_storage" ) == ConsentStatus.NOT_REQUESTED: privacy_manager.request_consent( consent_type= "memory_storage" , description= "是否允许存储交互记忆以提供个性化服务？" ) Step 2: 感知与短期记忆 from scripts.perception import PerceptionMemoryStore from scripts.short_term import ShortTermMemoryManager from scripts.types import SemanticBucketType # 感知记忆 perception = PerceptionMemoryStore() session_id = perception.create_session() # 短期记忆（智能体判断语义分类） short_term = ShortTermMemoryManager() item_id = short_term.store_with_semantics( content= "用户想要实现登录功能" , bucket_type=SemanticBucketType.USER_INTENT, topic_label= "用户登录" , relevance_score= 0.85 , ) Step 3: 长期记忆 from scripts.long_term import LongTermMemoryManager long_term = LongTermMemoryManager() long_term.update_user_profile(profile_data) long_term.apply_heat_policy() Step 4: 上下文重构与洞察 from scripts.context_reconstructor import ContextReconstructor from scripts.insight_module import InsightModule reconstructor = ContextReconstructor() insight_module = InsightModule() context = reconstructor.reconstruct(situation, long_term.get_all_memories()) insights = insight_module.process(context, long_term.get_all_memories()) Step 5: 全局状态捕捉（LangGraph集成） from scripts.state_capture import GlobalStateCapture, StateEventType capture = GlobalStateCapture( user_id= "user_123" , storage_path= "./state_storage" , ) # 从 LangGraph 同步 checkpoint_id = capture.sync_from_langgraph( state={ "phase" : "executing" , "current_task" : "create_memory" }, node_name= "executor" , ) # 事件订阅 subscription_id = capture.subscribe( event_types=[StateEventType.PHASE_CHANGE, StateEventType.TASK_SWITCH], callback=on_phase_change, ) Step 6: Context Orchestrator（总控层） from scripts.context_orchestrator import create_context_orchestrator from scripts.types import SemanticBucketType orchestrator = create_context_orchestrator( user_id= "user_123" , session_id= "session_456" , max_context_tokens= 32000 , ) # 存储记忆 orchestrator.store_memory( content= "用户想要实现登录功能" , bucket_type=SemanticBucketType.USER_INTENT, topic_label= "用户登录" , ) # 准备上下文 context = orchestrator.prepare_context( user_input= "帮我分析这段代码的性能问题" , system_instruction= "你是一个代码分析专家" , retrieval_results=[ "性能优化最佳实践" ], tool_results=[ "代码分析结果..." ], ) # 结束会话 final_stats = orchestrator.end_session() Step 7: 认知模型构建 from scripts.cognitive_model_builder import CognitiveModelBuilder, StepResult, FactSource builder = CognitiveModelBuilder(session_id= "session_001" ) # 设置任务上下文 builder.set_task_context( goal= "实现用户登录功能" , sub_goals=[ "数据库设计" , "前端表单" , "后端验证" ], current_focus= "后端验证逻辑" , ) # 添加已知事实和约束 builder.add_fact(content= "用户使用Python 3.9" , source=FactSource.MEMORY, confidence= 0.9 ) builder.add_constraint( "must_use" , "bcrypt加密" ) builder.add_knowledge_gap(description= "SSO集成方案" , importance= "high" ) # 构建认知模型 model = builder.build() print (model.to_context_string()) # 输出模型可理解的上下文 Step 8: 因果链提取 from scripts.causal_chain_extractor import CausalChainExtractor extractor = CausalChainExtractor() chains = extractor.extract( "登录失败是因为数据库连接超时..." ) for chain in chains: print (chain.to_summary()) # 问题: 登录失败 # 根本原因: 连接池配置过小 # 解决方案: 增加连接池大小 Step 9: 知识缺口识别 from scripts.knowledge_gap_identifier import KnowledgeGapIdentifier, KnowledgeType identifier = KnowledgeGapIdentifier() # 注册已有知识 identifier.register_knowledge(content= "用户使用Python 3.9" , knowledge_type=KnowledgeType.FACTUAL) # 定义所需知识 identifier.define_required(description= "数据库连接配置" , for_task= "配置连接" , importance= 4 ) # 分析缺口 result = identifier.analyze() print ( f"知识缺口: {result.total_gaps} , 覆盖率: {result.coverage_ratio: .1 %} " ) Step 10: 检索决策与评估 from scripts.retrieval_decision_engine import RetrievalDecisionEngine from scripts.retrieval_quality_evaluator import RetrievalQualityEvaluator # 检索决策 engine = RetrievalDecisionEngine() decision = engine.decide(query= "如何优化Python代码性能" ) if decision.need in [ "required" , "recommended" ]: print ( f"建议检索: {decision.queries} " ) # 质量评估 evaluator = RetrievalQualityEvaluator() result = evaluator.evaluate(query= "..." , items=[{ "item_id" : "1" , "content" : "..." , "score" : 0.9 }]) print ( f"质量评分: {result.quality.overall_score: .2 f} " ) Step 11: 状态一致性校验 from scripts.state_consistency_validator import StateConsistencyValidator, StateModule validator = StateConsistencyValidator() # 注册各模块状态 validator.register_state(module=StateModule.TASK_PROGRESS, state={ "current_task" : "登录功能" }) validator.register_state(module=StateModule.SHORT_TERM_MEMORY, state={ "topic" : "用户认证" }) # 执行校验 report = validator.validate() if report.conflicts: fixed = validator.auto_fix(report) print ( f"修复了 {fixed} 个冲突" ) Step 12: 状态推理 from scripts.state_inference_engine import StateInferenceEngine engine = StateInferenceEngine() engine.add_premise( "任务进度是80%" , confidence= 0.9 ) engine.add_premise( "没有阻塞问题" , confidence= 0.8 ) result = engine.infer_next_state() print ( f"推理结果: {result.inferred_value} , 置信度: {result.confidence: .2 f} " ) Step 13: 跨会话关联 from scripts.cross_session_memory_linker import CrossSessionMemoryLinker, LinkType linker = CrossSessionMemoryLinker() linker.register_session(session_id= "session_001" , topics=[ "Python优化" ], entities=[ "Pandas" ]) linker.register_session(session_id= "session_002" , topics=[ "Python优化" ], entities=[ "Redis" ]) # 发现关联 links = linker.discover_links() related = linker.get_related_sessions( "session_001" ) Step 14: 遗忘机制 from scripts.memory_forgetting_mechanism import MemoryForgettingMechanism, MemoryImportance mechanism = MemoryForgettingMechanism() mechanism.register_memory(memory_id= "mem_001" , importance=MemoryImportance.HIGH) mechanism.access_memory( "mem_001" ) # 提升活跃度 candidates = mechanism.analyze_forgetting_candidates() report = mechanism.execute_forgetting(candidates) print ( f"活跃记忆: {report.active_memories} , 归档: {report.archived_memories} " ) Step 15: 多源协调 from scripts.multi_source_coordinator import MultiSourceCoordinator, SourceType coordinator = MultiSourceCoordinator() coordinator.register_source(source_type=SourceType.SYSTEM_INSTRUCTION, content= "你是代码分析专家" ) coordinator.register_source(source_type=SourceType.USER_QUERY, content= "帮我分析代码" ) coordinator.register_source(source_type=SourceType.LONG_TERM_MEMORY, content= "用户偏好Python" ) context = coordinator.coordinate(max_tokens= 8000 ) print ( f"Token使用: {context.token_count} , 来源: { len (context.sources_used)} " ) Step 16: 上下文懒加载 from scripts.context_lazy_loader import create_lazy_loader loader = create_lazy_loader(max_cache_size= 1000 ) loader.register_loader( "user_profile" , lambda : fetch_user_profile()) result = loader.load( "user_profile" ) predicted = loader.predict_and_preload( "user_profile" ) print ( f"缓存命中率: {loader.get_stats().cache_hit_rate: .1 %} " ) Step 17: 权限边界控制 from scripts.permission_boundary_controller import create_permission_controller controller = create_permission_controller() controller.set_user_permission(user_id= "user_123" , roles=[ "user" ]) # 检查访问权限 result = controller.check_access(user_id= "user_123" , resource= "memory:long_term" , action= "read" ) # 过滤敏感信息 filtered = controller.filter_sensitive( "我的API Key是 sk-xxx，邮箱是 user@example.com" ) print ( f"过滤后: {filtered.filtered} " ) Step 18: 可观测性管理 from scripts.observability_manager import create_observability_manager, LatencyTracker manager = create_observability_manager(token_cost_per_1k= 0.03 ) # 记录Token使用 record = manager.record_token_usage(session_id= "session_001" , total_tokens= 1800 , model= "gpt-4" ) print ( f"成本: $ {record.cost_estimate: .4 f} " ) # 延迟追踪 with LatencyTracker(manager, "context_prepare" ) as tracker: tracker.start_stage( "memory_load" ) # ... 加载记忆 tracker.end_stage( "memory_load" ) # 获取统计 stats = manager.get_stats(hours= 24 ) print ( f"总Token: {stats.total_tokens} , 总成本: $ {stats.total_cost: .2 f} " ) Step 19: 结果压缩 from scripts.result_compressor import ResultCompressor, CompressionStrategy compressor = ResultCompressor() result = compressor.compress_tool_result(content=long_log_content, target_tokens= 1000 ) print ( f"压缩率: {result.compression_ratio: .2 %} " ) print ( f"因果链: { len (result.causal_chains)} 个" ) Step 20: 任务进度追踪 from scripts.task_progress import TaskProgressTracker, StepType tracker = TaskProgressTracker(task_id= "task_001" , task_name= "实现登录功能" ) tracker.set_goal(goal_id= "goal_001" , goal_name= "实现登录" , success_criteria=[ "用户可以登录" ]) tracker.track_step(step_id= "step_001" , step_name= "设计流程" , step_type=StepType.PLANNING) tracker.start_step( "step_001" ) tracker.complete_step( "step_001" , result= "流程设计完成" ) report = tracker.get_progress_report() print ( f"完成率: {report.completion_rate: .1 %} " ) Step 21: 记忆冲突检测 from scripts.memory_conflict import MemoryConflictDetector detector = MemoryConflictDetector() conflicts = detector.detect_all_conflicts(new_memory=item, existing_memories=memories) if conflicts: result = detector.resolve_conflict(conflict=conflicts[ 0 ], mode= "recency" )