openClaw_agent_dm/skills/mem0-integration/openclaw_commands.py

#!/usr/bin/env python3
"""
mem0 OpenClaw Commands
处理 /memory 命令
"""

import os
import sys

# 设置环境变量
os.environ['OPENAI_API_BASE'] = 'https://dashscope.aliyuncs.com/compatible-mode/v1'
os.environ['OPENAI_API_KEY'] = 'sk-c1715ee0479841399fd359c574647648'

sys.path.insert(0, os.path.dirname(__file__))

from mem0_client import Mem0Client

# 初始化客户端
mem0 = Mem0Client()

def handle_memory_command(args: str, context: dict) -> str:
    """
    处理 /memory 命令
    
    Args:
        args: 命令参数
        context: 上下文信息 (user_id, agent_id 等)
    
    Returns:
        回复文本
    """
    parts = args.split(' ', 2)
    action = parts[0] if parts else 'help'
    
    user_id = context.get('user_id', 'default')
    agent_id = context.get('agent_id', 'main')
    
    if action == 'add':
        content = parts[1] if len(parts) > 1 else ''
        return handle_add(content, user_id, agent_id)
    elif action == 'search':
        query = parts[1] if len(parts) > 1 else ''
        return handle_search(query, user_id, agent_id)
    elif action == 'list':
        return handle_list(user_id, agent_id)
    elif action == 'delete':
        memory_id = parts[1] if len(parts) > 1 else ''
        return handle_delete(memory_id, user_id, agent_id)
    elif action == 'status':
        return handle_status()
    else:
        return get_help()

def handle_add(content: str, user_id: str, agent_id: str) -> str:
    """添加记忆"""
    if not content:
        return "❌ 请提供记忆内容\n\n用法：/memory add <内容>"
    
    result = mem0.add([{"role": "user", "content": content}], user_id=user_id)
    
    if result and 'error' in result:
        return f"❌ 添加失败：{result['error']}"
    
    return f"✅ 记忆已添加\n\n📝 内容：{content[:200]}..."

def handle_search(query: str, user_id: str, agent_id: str) -> str:
    """搜索记忆"""
    if not query:
        return "❌ 请提供搜索关键词\n\n用法：/memory search <关键词>"
    
    results = mem0.search(query, user_id=user_id, limit=5)
    
    if not results:
        return f"🔍 未找到与 \"{query}\" 相关的记忆"
    
    response = f"🔍 找到 {len(results)} 条相关记忆:\n\n"
    for i, mem in enumerate(results, 1):
        memory_text = mem.get('memory', 'N/A')
        response += f"{i}. {memory_text}\n"
        if mem.get('created_at'):
            response += f"   📅 {mem['created_at']}\n"
        response += "\n"
    
    return response

def handle_list(user_id: str, agent_id: str) -> str:
    """列出所有记忆"""
    results = mem0.get_all(user_id=user_id)
    
    if not results:
        return "📭 暂无记忆"
    
    response = f"📋 共 {len(results)} 条记忆:\n\n"
    for i, mem in enumerate(results[:20], 1):
        memory_text = mem.get('memory', 'N/A')
        truncated = memory_text[:100] + '...' if len(memory_text) > 100 else memory_text
        response += f"{i}. {truncated}\n"
        response += f"   🆔 {mem.get('id', 'N/A')}\n\n"
    
    if len(results) > 20:
        response += f"... 还有 {len(results) - 20} 条记忆\n"
    
    return response

def handle_delete(memory_id: str, user_id: str, agent_id: str) -> str:
    """删除记忆"""
    if not memory_id:
        return "❌ 请提供记忆 ID\n\n用法：/memory delete <ID>"
    
    success = mem0.delete(memory_id, user_id=user_id)
    
    if success:
        return f"✅ 记忆已删除：{memory_id}"
    else:
        return f"❌ 删除失败：记忆不存在或权限不足"

def handle_status() -> str:
    """查看状态"""
    status = mem0.get_status()
    
    response = "📊 mem0 状态:\n\n"
    response += f"本地记忆：{'✅' if status['initialized'] else '❌'}\n"
    response += f"Qdrant 地址：{status['qdrant']}\n"
    
    return response

def get_help() -> str:
    """获取帮助文本"""
    return """📖 mem0 记忆管理命令

用法：/memory <命令> [参数]

可用命令:
  add <内容>      - 添加记忆
  search <关键词>  - 搜索记忆
  list            - 列出所有记忆
  delete <ID>     - 删除记忆
  status          - 查看状态
  help            - 显示帮助

示例:
  /memory add 用户偏好使用 UTC 时区
  /memory search 时区
  /memory list
  /memory status"""

# 测试
if __name__ == '__main__':
    print("测试 /memory 命令...")
    print(handle_status())