feat(memory-graph): Step 1 - 集成记忆工具到插件系统

完成记忆系统工具的插件化集成： 1. 创建记忆工具适配器 (memory_plugin_tools.py) - CreateMemoryTool: 创建新记忆 - LinkMemoriesTool: 关联两条记忆 - SearchMemoriesTool: 搜索相关记忆 - 适配 BaseTool 接口，支持 LLM 调用 2. 创建全局 MemoryManager 单例 (manager_singleton.py) - initialize_memory_manager(): 初始化全局实例 - get_memory_manager(): 获取单例实例 - shutdown_memory_manager(): 关闭管理器 - 线程安全的单例模式 3. 创建记忆系统插件 (plugins/memory_graph_plugin/) - MemoryGraphPlugin: 插件主类 - 自动注册3个记忆工具到系统 - on_plugin_loaded(): 初始化 MemoryManager - on_unload(): 清理资源 4. 修复类型问题 - ToolParamType.OBJECT STRING (JSON格式) - ToolParamType.NUMBER FLOAT - attributes 参数支持 JSON 字符串解析 - 修复 min_importance None 比较错误 5. 添加集成测试 (test_plugin_integration.py) - 测试工具导入和定义 - 测试 MemoryManager 初始化 - 测试工具执行（创建、搜索记忆） - 测试单例模式测试结果: 所有测试通过 LLM 现在可以通过工具调用主动管理记忆！
2025-11-05 18:42:27 +08:00
parent 64b8636e9e
commit fc71aad817
6 changed files with 543 additions and 1 deletions
--- a/tests/memory_graph/test_plugin_integration.py
+++ b/tests/memory_graph/test_plugin_integration.py
@@ -0,0 +1,126 @@
+"""
+测试记忆系统插件集成
+
+验证：
+1. 插件能否正常加载
+2. 工具能否被识别为 LLM 可用工具
+3. 工具能否正常执行
+"""
+
+import asyncio
+import sys
+from pathlib import Path
+
+# 添加项目根目录到路径
+sys.path.insert(0, str(Path(__file__).parent.parent.parent))
+
+
+async def test_plugin_integration():
+    """测试插件集成"""
+    print("=" * 60)
+    print("测试记忆系统插件集成")
+    print("=" * 60)
+    print()
+
+    # 1. 测试导入插件工具
+    print("[1] 测试导入插件工具...")
+    try:
+        from src.memory_graph.plugin_tools.memory_plugin_tools import (
+            CreateMemoryTool,
+            LinkMemoriesTool,
+            SearchMemoriesTool,
+        )
+
+        print(f"  ✅ CreateMemoryTool: {CreateMemoryTool.name}")
+        print(f"  ✅ LinkMemoriesTool: {LinkMemoriesTool.name}")
+        print(f"  ✅ SearchMemoriesTool: {SearchMemoriesTool.name}")
+    except Exception as e:
+        print(f"  ❌ 导入失败: {e}")
+        return False
+
+    # 2. 测试工具定义
+    print("\n[2] 测试工具定义...")
+    try:
+        create_def = CreateMemoryTool.get_tool_definition()
+        link_def = LinkMemoriesTool.get_tool_definition()
+        search_def = SearchMemoriesTool.get_tool_definition()
+
+        print(f"  ✅ create_memory: {len(create_def['parameters'])} 个参数")
+        print(f"  ✅ link_memories: {len(link_def['parameters'])} 个参数")
+        print(f"  ✅ search_memories: {len(search_def['parameters'])} 个参数")
+    except Exception as e:
+        print(f"  ❌ 获取工具定义失败: {e}")
+        return False
+
+    # 3. 测试初始化 MemoryManager
+    print("\n[3] 测试初始化 MemoryManager...")
+    try:
+        from src.memory_graph.manager_singleton import (
+            get_memory_manager,
+            initialize_memory_manager,
+        )
+
+        # 初始化
+        manager = await initialize_memory_manager(data_dir="data/test_plugin_integration")
+        print(f"  ✅ MemoryManager 初始化成功")
+
+        # 获取单例
+        manager2 = get_memory_manager()
+        assert manager is manager2, "单例模式失败"
+        print(f"  ✅ 单例模式正常")
+
+    except Exception as e:
+        print(f"  ❌ 初始化失败: {e}")
+        import traceback
+
+        traceback.print_exc()
+        return False
+
+    # 4. 测试工具执行
+    print("\n[4] 测试工具执行...")
+    try:
+        # 创建记忆
+        create_tool = CreateMemoryTool()
+        result = await create_tool.execute(
+            {
+                "subject": "我",
+                "memory_type": "事件",
+                "topic": "测试记忆系统插件",
+                "attributes": {"时间": "今天"},
+                "importance": 0.8,
+            }
+        )
+        print(f"  ✅ create_memory: {result['content']}")
+
+        # 搜索记忆
+        search_tool = SearchMemoriesTool()
+        result = await search_tool.execute({"query": "测试", "top_k": 5})
+        print(f"  ✅ search_memories: 找到记忆")
+
+    except Exception as e:
+        print(f"  ❌ 工具执行失败: {e}")
+        import traceback
+
+        traceback.print_exc()
+        return False
+
+    # 5. 测试关闭
+    print("\n[5] 测试关闭...")
+    try:
+        from src.memory_graph.manager_singleton import shutdown_memory_manager
+
+        await shutdown_memory_manager()
+        print(f"  ✅ MemoryManager 关闭成功")
+    except Exception as e:
+        print(f"  ❌ 关闭失败: {e}")
+        return False
+
+    print("\n" + "=" * 60)
+    print("[SUCCESS] 所有测试通过！")
+    print("=" * 60)
+    return True
+
+
+if __name__ == "__main__":
+    result = asyncio.run(test_plugin_integration())
+    sys.exit(0 if result else 1)