150 lines
5.5 KiB
Python
150 lines
5.5 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
详细查看和测试Vertex AI Memory Bank功能
|
||
"""
|
||
|
||
import sys
|
||
import os
|
||
import asyncio
|
||
import json
|
||
from datetime import datetime
|
||
sys.path.append('src')
|
||
|
||
from jixia.memory.factory import get_memory_backend
|
||
from config.doppler_config import get_google_genai_config
|
||
|
||
async def test_memory_bank_functionality():
|
||
print("🧠 详细测试Memory Bank功能")
|
||
print("=" * 60)
|
||
|
||
# 获取配置
|
||
config = get_google_genai_config()
|
||
project_id = config.get('project_id')
|
||
location = config.get('location', 'us-central1')
|
||
|
||
print(f"📊 项目ID: {project_id}")
|
||
print(f"📍 位置: {location}")
|
||
print(f"🕐 测试时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
||
print()
|
||
|
||
try:
|
||
# 获取Memory Bank后端
|
||
memory_backend = get_memory_backend()
|
||
print(f"✅ Memory Bank后端: {type(memory_backend).__name__}")
|
||
print()
|
||
|
||
# 选择一个智能体进行详细测试
|
||
test_agent = "lvdongbin"
|
||
print(f"🧙♂️ 测试智能体: {test_agent} (吕洞宾)")
|
||
print("-" * 40)
|
||
|
||
# 1. 创建/获取Memory Bank
|
||
print("1️⃣ 创建Memory Bank...")
|
||
memory_bank_id = await memory_backend.create_memory_bank(
|
||
agent_name=test_agent,
|
||
display_name=f"测试Memory Bank - {test_agent}"
|
||
)
|
||
print(f" ✅ Memory Bank ID: {memory_bank_id}")
|
||
print()
|
||
|
||
# 2. 添加不同类型的记忆
|
||
print("2️⃣ 添加测试记忆...")
|
||
|
||
# 添加对话记忆
|
||
conversation_memory = await memory_backend.add_memory(
|
||
agent_name=test_agent,
|
||
content="在关于AI伦理的辩论中，我强调了技术发展应该以人为本，不能忽视道德考量。",
|
||
memory_type="conversation",
|
||
debate_topic="AI伦理与技术发展",
|
||
metadata={"opponent": "铁拐李", "stance": "支持伦理优先"}
|
||
)
|
||
print(f" 📝 对话记忆: {conversation_memory}")
|
||
|
||
# 添加偏好记忆
|
||
preference_memory = await memory_backend.add_memory(
|
||
agent_name=test_agent,
|
||
content="我偏好使用古典哲学的智慧来论证现代问题，特别是道家思想。",
|
||
memory_type="preference",
|
||
metadata={"philosophy": "道家", "style": "古典智慧"}
|
||
)
|
||
print(f" ⚙️ 偏好记忆: {preference_memory}")
|
||
|
||
# 添加知识记忆
|
||
knowledge_memory = await memory_backend.add_memory(
|
||
agent_name=test_agent,
|
||
content="区块链技术的核心是去中心化和不可篡改性，这与道家'无为而治'的理念有相通之处。",
|
||
memory_type="knowledge",
|
||
debate_topic="区块链技术应用",
|
||
metadata={"domain": "技术", "connection": "哲学"}
|
||
)
|
||
print(f" 📚 知识记忆: {knowledge_memory}")
|
||
|
||
# 添加策略记忆
|
||
strategy_memory = await memory_backend.add_memory(
|
||
agent_name=test_agent,
|
||
content="在辩论中，当对手使用激进论点时，我会用温和的反问来引导思考，而不是直接对抗。",
|
||
memory_type="strategy",
|
||
metadata={"tactic": "温和引导", "effectiveness": "高"}
|
||
)
|
||
print(f" 🎯 策略记忆: {strategy_memory}")
|
||
print()
|
||
|
||
# 3. 测试记忆搜索
|
||
print("3️⃣ 测试记忆搜索...")
|
||
|
||
# 搜索关于AI的记忆
|
||
ai_memories = await memory_backend.search_memories(
|
||
agent_name=test_agent,
|
||
query="AI 人工智能 伦理",
|
||
limit=5
|
||
)
|
||
print(f" 🔍 搜索'AI 人工智能 伦理': 找到 {len(ai_memories)} 条记忆")
|
||
for i, memory in enumerate(ai_memories, 1):
|
||
print(f" {i}. {memory.get('content', '')[:50]}...")
|
||
print()
|
||
|
||
# 搜索策略相关记忆
|
||
strategy_memories = await memory_backend.search_memories(
|
||
agent_name=test_agent,
|
||
query="辩论 策略",
|
||
memory_type="strategy",
|
||
limit=3
|
||
)
|
||
print(f" 🎯 搜索策略记忆: 找到 {len(strategy_memories)} 条记忆")
|
||
for i, memory in enumerate(strategy_memories, 1):
|
||
print(f" {i}. {memory.get('content', '')[:50]}...")
|
||
print()
|
||
|
||
# 4. 获取智能体上下文
|
||
print("4️⃣ 获取智能体上下文...")
|
||
context = await memory_backend.get_agent_context(
|
||
agent_name=test_agent,
|
||
debate_topic="AI伦理与技术发展"
|
||
)
|
||
print(f" 📋 上下文长度: {len(context)} 字符")
|
||
print(f" 📋 上下文预览: {context[:200]}...")
|
||
print()
|
||
|
||
# 5. 显示所有记忆类型的统计
|
||
print("5️⃣ 记忆统计...")
|
||
memory_types = ["conversation", "preference", "knowledge", "strategy"]
|
||
for mem_type in memory_types:
|
||
memories = await memory_backend.search_memories(
|
||
agent_name=test_agent,
|
||
query="",
|
||
memory_type=mem_type,
|
||
limit=100
|
||
)
|
||
print(f" 📊 {mem_type}: {len(memories)} 条记忆")
|
||
|
||
print()
|
||
print("🎉 Memory Bank功能测试完成!")
|
||
print("=" * 60)
|
||
|
||
except Exception as e:
|
||
print(f"❌ 测试失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(test_memory_bank_functionality()) |