78 lines
2.0 KiB
Python
78 lines
2.0 KiB
Python
# openclaw_integration.py - OpenClaw 集成接口
|
|
# 提供单例模式的记忆系统访问
|
|
|
|
from vector_memory import VectorMemorySystem
|
|
from memory_tier_manager import MemoryTierManager
|
|
import os
|
|
|
|
# 初始化（单例模式）
|
|
_memory_system = None
|
|
_tier_manager = None
|
|
|
|
|
|
def get_memory_system():
|
|
"""获取记忆系统单例"""
|
|
global _memory_system
|
|
|
|
if _memory_system is None:
|
|
api_key = os.getenv("SILICONFLOW_API_KEY")
|
|
if not api_key:
|
|
raise ValueError("请设置 SILICONFLOW_API_KEY 环境变量")
|
|
|
|
_memory_system = VectorMemorySystem(
|
|
persist_dir="./data/memory",
|
|
api_key=api_key
|
|
)
|
|
|
|
return _memory_system
|
|
|
|
|
|
def get_tier_manager():
|
|
"""获取分层管理器单例"""
|
|
global _tier_manager
|
|
|
|
if _tier_manager is None:
|
|
vm = get_memory_system()
|
|
_tier_manager = MemoryTierManager(vm)
|
|
|
|
return _tier_manager
|
|
|
|
|
|
def search_memory(query: str, top_k: int = 5):
|
|
"""搜索记忆 - 供 OpenClaw 调用"""
|
|
vm = get_memory_system()
|
|
return vm.search(query, top_k)
|
|
|
|
|
|
def add_memory(content: str, importance: int = 3, tags: list = None):
|
|
"""添加记忆 - 供 OpenClaw 调用"""
|
|
mtm = get_tier_manager()
|
|
return mtm.add_with_tier(content, importance, tags)
|
|
|
|
|
|
def get_all_memories(limit: int = 50):
|
|
"""获取所有记忆"""
|
|
mtm = get_tier_manager()
|
|
return mtm.get_recent_memories(limit=limit)
|
|
|
|
|
|
def get_core_memories():
|
|
"""获取核心记忆"""
|
|
mtm = get_tier_manager()
|
|
return mtm.get_core_memories()
|
|
|
|
|
|
# 使用示例
|
|
if __name__ == "__main__":
|
|
# 添加记忆
|
|
add_memory(
|
|
content="2026-03-21: 部署了向量记忆系统，采用硅基流动 BGE-M3 + Chroma + SQLite 架构",
|
|
importance=4,
|
|
tags=["向量记忆", "系统部署", "硅基流动"]
|
|
)
|
|
|
|
# 搜索记忆
|
|
results = search_memory("记忆系统")
|
|
for r in results:
|
|
print(f"- {r['content'][:50]}... (相似度: {1-r['distance']:.2%})")
|