#!/usr/bin/env python # -*- coding: UTF-8 -*- """ @time: 2026/01/08 @file: powermem.py @desc: PowerMem memory provider for xiaozhi-esp32-server PowerMem is an open-source agent memory component from OceanBase GitHub: https://github.com/oceanbase/powermem Website: https://www.powermem.ai/ @Author: wayyoungboy """ import asyncio import json import traceback from typing import Optional, Dict, Any from ..base import MemoryProviderBase, logger TAG = __name__ class MemoryProvider(MemoryProviderBase): """ PowerMem memory provider implementation. PowerMem is an open-source agent memory component that provides efficient memory management for AI agents. Supports multiple storage backends (sqlite, oceanbase, postgres), LLM providers (qwen, openai, etc.) and embedding providers. Config options: - enable_user_profile: bool - Enable UserMemory for user profiling (requires OceanBase) - database_provider: str - Storage backend (sqlite, oceanbase, postgres) - llm_provider: str - LLM provider (qwen, openai, etc.) - embedding_provider: str - Embedding provider (qwen, openai, etc.) """ def __init__(self, config: Dict[str, Any], summary_memory: Optional[str] = None): super().__init__(config) self.use_powermem = False self.memory_client = None self.enable_user_profile = False self.last_profile_content = "" # Cache for user profile from UserMemory try: self.enable_user_profile = str(config.get("enable_user_profile", False)).lower() == 'true' # Get configuration parameters database_provider = config.get("database_provider", "sqlite") llm_provider = config.get("llm_provider", "qwen") embedding_provider = config.get("embedding_provider", "qwen") # Build powermem configuration dict # PowerMem supports two config styles: # 1. powermem style: database, llm, embedding # 2. mem0 style: vector_store, llm, embedder powermem_config = {} # Configure vector store / database if "vector_store" in config: powermem_config["vector_store"] = config["vector_store"] elif "database" in config: powermem_config["database"] = config["database"] else: powermem_config["vector_store"] = { "provider": database_provider, "config": {} } # Configure LLM if "llm" in config: powermem_config["llm"] = config["llm"] else: llm_config = {} if config.get("llm_api_key"): llm_config["api_key"] = config["llm_api_key"] if config.get("llm_model"): llm_config["model"] = config["llm_model"] # Handle base_url based on provider type # - qwen provider uses dashscope_base_url # - openai provider uses openai_base_url if llm_provider == "qwen": base_url = config.get("dashscope_base_url") or config.get("llm_base_url") if base_url: llm_config["dashscope_base_url"] = base_url else: base_url = config.get("openai_base_url") or config.get("llm_base_url") if base_url: llm_config["openai_base_url"] = base_url powermem_config["llm"] = { "provider": llm_provider, "config": llm_config } # Configure embedder if "embedder" in config: powermem_config["embedder"] = config["embedder"] else: embedder_config = {} if config.get("embedding_api_key"): embedder_config["api_key"] = config["embedding_api_key"] if config.get("embedding_model"): embedder_config["model"] = config["embedding_model"] # Handle base_url based on provider type # - qwen provider uses dashscope_base_url # - openai provider uses openai_base_url # Priority: embedding_xxx_base_url > embedding_base_url > xxx_base_url if embedding_provider == "qwen": base_url = config.get("embedding_dashscope_base_url") or config.get("embedding_base_url") if base_url: embedder_config["dashscope_base_url"] = base_url else: base_url = config.get("embedding_openai_base_url") or config.get("embedding_base_url") if base_url: embedder_config["openai_base_url"] = base_url powermem_config["embedder"] = { "provider": embedding_provider, "config": embedder_config } # Initialize memory client based on mode if self.enable_user_profile: from powermem import UserMemory self.memory_client = UserMemory(config=powermem_config) memory_mode = "UserMemory (用户画像模式)" else: from powermem import AsyncMemory self.memory_client = AsyncMemory(config=powermem_config) memory_mode = "AsyncMemory (普通记忆模式)" self.use_powermem = True logger.bind(tag=TAG).info( f"PowerMem initialized successfully: mode={memory_mode}, " f"database={powermem_config['vector_store']['provider']}, llm={powermem_config['llm']['provider']}, embedding={powermem_config['embedder']['provider']}" ) except ImportError as e: logger.bind(tag=TAG).error( f"PowerMem not installed. Please install with: pip install powermem. Error: {e}" ) self.use_powermem = False except Exception as e: logger.bind(tag=TAG).error(f"Failed to initialize PowerMem: {str(e)}") logger.bind(tag=TAG).debug(f"Detailed error: {traceback.format_exc()}") self.use_powermem = False async def save_memory(self, msgs, session_id=None): """ Save conversation messages to PowerMem. Args: msgs: List of message objects with 'role' and 'content' attributes session_id: Session identifier (optional, for compatibility) Returns: Result from PowerMem API or None if failed """ try: if self.use_powermem and self.memory_client is not None and len(msgs) >= 2: # Format the content as a message list for PowerMem messages = [] for message in msgs: if message.role == "system": continue content = message.content # Extract content from JSON format if present (for ASR with emotion/language tags) # Same logic as in query_memory method try: if content and content.strip().startswith("{") and content.strip().endswith("}"): data = json.loads(content) if "content" in data: content = data["content"] except (json.JSONDecodeError, KeyError, TypeError): # If parsing fails, use original content pass messages.append({"role": message.role, "content": content}) # Add memory using PowerMem SDK if self.enable_user_profile: # UserMemory uses sync add result = await asyncio.to_thread( self.memory_client.add, messages=messages, user_id=self.role_id ) else: # AsyncMemory uses async add result = await self.memory_client.add( messages=messages, user_id=self.role_id ) logger.bind(tag=TAG).debug(f"Save memory result: {result}") # Cache user profile if UserMemory mode and profile was extracted if self.enable_user_profile and result: if result.get('profile_extracted'): self.last_profile_content = result.get('profile_content', '') logger.bind(tag=TAG).debug(f"User profile extracted: {self.last_profile_content}") else: if not self.use_powermem or self.memory_client is None: logger.bind(tag=TAG).warning("PowerMem is not available, skipping save_memory") elif len(msgs) < 2: logger.bind(tag=TAG).debug("Not enough messages to save (need at least 2)") except Exception as e: logger.bind(tag=TAG).error(f"Error saving memory: {str(e)}") logger.bind(tag=TAG).debug(f"Detailed error: {traceback.format_exc()}") return None async def query_memory(self, query: str) -> str: """ Query memories from PowerMem based on similarity search. Args: query: The search query string (may be JSON format with metadata) Returns: Formatted string of relevant memories or empty string if none found """ if not self.use_powermem or self.memory_client is None: logger.bind(tag=TAG).warning("PowerMem is not available, skipping query_memory") return "" try: if not getattr(self, "role_id", None): logger.bind(tag=TAG).debug("No role_id set, returning empty memory") return "" # Extract content from JSON format if present (for ASR with emotion/language tags) search_query = query try: if query.strip().startswith("{") and query.strip().endswith("}"): data = json.loads(query) if "content" in data: search_query = data["content"] except (json.JSONDecodeError, KeyError): # If parsing fails, use original query pass result_parts = [] # If user profile mode is enabled, include user profile in results if self.enable_user_profile: profile = await self.get_user_profile() if profile: result_parts.append(f"【用户画像】\n{profile}") # Search memories using PowerMem SDK if self.enable_user_profile: # UserMemory uses sync search results = await asyncio.to_thread( self.memory_client.search, query=search_query, user_id=self.role_id, limit=30 ) else: # AsyncMemory uses async search results = await self.memory_client.search( query=search_query, user_id=self.role_id, limit=30 ) if results and "results" in results: # Format each memory entry with its update time memories = [] for entry in results.get("results", []): # Get timestamp from updated_at or created_at timestamp = "" if "updated_at" in entry and entry["updated_at"]: timestamp = str(entry["updated_at"]) elif "created_at" in entry and entry["created_at"]: timestamp = str(entry["created_at"]) if timestamp: try: # Parse and reformat the timestamp (remove milliseconds if present) if "." in timestamp: dt = timestamp.split(".")[0] else: dt = timestamp formatted_time = dt.replace("T", " ") except Exception: formatted_time = timestamp else: formatted_time = "" memory = entry.get("memory", "") or entry.get("content", "") if memory: if formatted_time: # Store tuple of (timestamp, formatted_string) for sorting memories.append((timestamp, f"[{formatted_time}] {memory}")) else: memories.append(("", memory)) # Sort by timestamp in descending order (newest first) memories.sort(key=lambda x: x[0], reverse=True) # Extract only the formatted strings if memories: memories_str = "\n".join(f"- {memory[1]}" for memory in memories) result_parts.append(f"【相关记忆】\n{memories_str}") final_result = "\n\n".join(result_parts) logger.bind(tag=TAG).debug(f"Query results: {final_result}") return final_result except Exception as e: logger.bind(tag=TAG).error(f"Error querying memory: {str(e)}") logger.bind(tag=TAG).debug(f"Detailed error: {traceback.format_exc()}") return "" async def get_user_profile(self) -> str: """ Get user profile from PowerMem (only available in UserMemory mode). Uses a cache-first strategy: 1. Check if last_profile_content is already cached 2. If cached, return immediately (performance optimization) 3. If cache is empty, fetch from PowerMem SDK 4. Store fetched profile in cache for next query Returns: Formatted user profile string or empty string if not available """ if not self.use_powermem or self.memory_client is None: return "" if not self.enable_user_profile: logger.bind(tag=TAG).debug("User profile mode is not enabled") return "" # Return cached profile content if available (fast path) if self.last_profile_content: logger.bind(tag=TAG).debug("Returning cached user profile") return self.last_profile_content # Cache is empty, fetch from PowerMem SDK logger.bind(tag=TAG).info("Cache miss, fetching user profile from PowerMem SDK") try: # Call UserMemory.profile() to get profile data profile_data = await asyncio.to_thread( self.memory_client.profile, self.role_id ) if not profile_data: logger.bind(tag=TAG).warning("PowerMem SDK returned empty profile data") return "" # Try to use profile_content first profile_content = profile_data.get("profile_content") if profile_content: # Update cache with fetched profile_content self.last_profile_content = profile_content logger.bind(tag=TAG).info(f"Successfully fetched and cached user profile from profile_content (length: {len(self.last_profile_content)})") return self.last_profile_content # If profile_content is empty, fallback to topics topics = profile_data.get("topics") if topics: import json # Serialize topics dict to JSON string for structured profile self.last_profile_content = json.dumps(topics, ensure_ascii=False, indent=2) logger.bind(tag=TAG).info(f"Successfully fetched and cached user profile from topics (length: {len(self.last_profile_content)})") return self.last_profile_content # Both profile_content and topics are empty logger.bind(tag=TAG).warning("PowerMem SDK returned profile with empty profile_content and topics") return "" except Exception as e: logger.bind(tag=TAG).error(f"Failed to fetch user profile from SDK: {str(e)}") logger.bind(tag=TAG).debug(f"Detailed error: {traceback.format_exc()}") return ""