test_memory_retrieval.py•6.28 kB
#!/usr/bin/env python3
"""
MemOS记忆检索测试脚本
用于验证飞书MCP文档的检索精准度问题
"""
import sys
from pathlib import Path
# 添加src路径
sys.path.insert(0, str(Path(__file__).parent / "src"))
from usage_examples import load_env_file
from advanced_examples import AdvancedMemOS
from enhanced_simple_memos import EnhancedSimpleMemOS
def test_memory_retrieval():
"""测试记忆检索功能"""
print("🧪 MemOS记忆检索测试")
print("=" * 60)
# 加载环境变量
load_env_file()
# 初始化MemOS
memos_data_path = "./feishu_memos_data"
memos = AdvancedMemOS(memos_data_path)
print(f"📁 数据路径: {memos_data_path}")
# 测试查询列表
test_queries = [
"飞书MCP工具调用失败",
"飞书MCP OAuth授权问题",
"飞书MCP权限配置",
"Tool not found 错误",
"飞书MCP端口问题",
"飞书MCP删除功能",
"飞书MCP工具加载问题",
"飞书MCP应用配置"
]
print(f"🔍 开始测试 {len(test_queries)} 个查询...")
print()
for i, query in enumerate(test_queries, 1):
print(f"📝 测试 {i}: {query}")
print("-" * 40)
try:
# 查询记忆
results = memos.search_memories(query, limit=5)
if results:
print(f"✅ 找到 {len(results)} 条记忆")
for j, memory in enumerate(results, 1):
# 截取内容前100个字符
content = memory["content"]
content_preview = content[:100] + "..." if len(content) > 100 else content
print(f" {j}. 相关度: {memory['score']:.3f}")
print(f" 内容: {content_preview}")
print(f" 标签: {memory.get('tags', [])}")
print()
else:
print("❌ 未找到相关记忆")
except Exception as e:
print(f"❌ 查询失败: {e}")
print()
# 统计信息
try:
all_memories = memos.get_all_memories()
print(f"📊 数据库统计:")
print(f" 总记忆数: {len(all_memories)}")
# 分析标签分布
tag_counts = {}
for memory in all_memories:
tags = memory.get('metadata', {}).get('tags', [])
for tag in tags:
tag_counts[tag] = tag_counts.get(tag, 0) + 1
print(f" 标签种类: {len(tag_counts)}")
print(f" 热门标签:")
for tag, count in sorted(tag_counts.items(), key=lambda x: x[1], reverse=True)[:10]:
print(f" - {tag}: {count}")
except Exception as e:
print(f"❌ 统计失败: {e}")
def test_expert_fixes():
"""测试专家修复方案的效果"""
print("🔧 测试专家修复方案效果")
print("=" * 60)
# 加载环境变量
load_env_file()
# 初始化增强版MemOS
memos_data_path = "./expert_fix_test_data"
memos = EnhancedSimpleMemOS(memos_data_path)
print(f"📁 数据路径: {memos_data_path}")
print(f"📊 集合名称: {memos.collection_name}")
# 测试记忆添加
test_content = "专家修复测试：这是一条测试记忆，用于验证存储和检索功能是否正常工作。"
print(f"\n📝 添加测试记忆: {test_content}")
memory_id = memos.add_memory(test_content, tags=["专家修复", "测试"])
if memory_id:
print(f"✅ 记忆添加成功，ID: {memory_id}")
else:
print("❌ 记忆添加失败")
return False
# 测试记忆检索
print(f"\n🔍 搜索测试记忆...")
results = memos.search_memories("专家修复测试", limit=3)
if results:
print(f"✅ 搜索成功，找到 {len(results)} 条记忆:")
for i, result in enumerate(results, 1):
print(f" {i}. ID: {result['id']}, 分数: {result['score']:.3f}")
print(f" 内容: {result['content'][:50]}...")
return True
else:
print("❌ 搜索失败，未找到任何记忆")
return False
if __name__ == "__main__":
test_memory_retrieval()
print("\n" + "="*60)
test_expert_fixes()
def test_specific_feishu_content():
"""测试特定飞书内容的检索"""
print("\n🎯 飞书MCP特定内容测试")
print("=" * 60)
# 加载环境变量
load_env_file()
# 初始化MemOS
memos = AdvancedMemOS("./feishu_memos_data")
# 测试特定的飞书MCP关键词
feishu_keywords = [
"飞书",
"MCP",
"OAuth",
"工具调用",
"权限配置",
"端口3310",
"lark-openapi-mcp",
"VSCode Augment"
]
print("🔍 测试飞书MCP关键词检索...")
for keyword in feishu_keywords:
print(f"\n🔎 关键词: {keyword}")
try:
results = memos.search_memories(keyword, limit=3)
if results:
print(f" ✅ 找到 {len(results)} 条记忆")
for i, memory in enumerate(results, 1):
content = memory["content"]
score = memory["score"]
# 检查是否真的包含飞书MCP相关内容
is_feishu_related = any(term in content.lower() for term in ['飞书', 'feishu', 'mcp', 'lark'])
status = "✅" if is_feishu_related else "❌"
print(f" {i}. {status} 相关度: {score:.3f} | 飞书相关: {is_feishu_related}")
else:
print(" ❌ 未找到记忆")
except Exception as e:
print(f" ❌ 查询失败: {e}")
def main():
"""主函数"""
print("🚀 开始MemOS记忆检索测试")
print("用于诊断飞书MCP文档检索精准度问题")
print()
# 基础检索测试
test_memory_retrieval()
# 飞书特定内容测试
test_specific_feishu_content()
print("\n📋 测试完成!")
print("请将测试结果提供给专家进行分析。")
if __name__ == "__main__":
main()