diff --git a/reme_ai/mem_agent/wk/__init__.py b/reme_ai/mem_agent/wk/__init__.py new file mode 100644 index 00000000..90e61363 --- /dev/null +++ b/reme_ai/mem_agent/wk/__init__.py @@ -0,0 +1,9 @@ +from .personal_summarizer_wk import PersonalSummarizerWk +from .reme_retriever_wk import ReMeRetrieverV2 +from .reme_summarizer_wk import ReMeSummarizerWk + +__all__ = [ + "PersonalSummarizerWk", + "ReMeRetrieverV2", + "ReMeSummarizerWk", +] diff --git a/reme_ai/mem_agent/wk/personal_summarizer_wk.py b/reme_ai/mem_agent/wk/personal_summarizer_wk.py new file mode 100644 index 00000000..e974f99a --- /dev/null +++ b/reme_ai/mem_agent/wk/personal_summarizer_wk.py @@ -0,0 +1,69 @@ +from ..base_memory_agent import BaseMemoryAgent +from ...core.enumeration import Role, MemoryType +from ...core.schema import Message, ToolCall +from ...core.utils import format_messages + + +class PersonalSummarizerWk(BaseMemoryAgent): + memory_type: MemoryType = MemoryType.PERSONAL + + def _build_tool_call(self) -> ToolCall: + return ToolCall( + **{ + "description": self.get_prompt("tool"), + "parameters": { + "type": "object", + "properties": { + "messages": { + "type": "array", + "items": { + "type": "object", + "properties": { + "role": { + "type": "string", + "description": "role", + }, + "content": { + "type": "string", + "description": "content", + }, + }, + "required": ["role", "content"], + }, + }, + }, + "required": ["messages"], + }, + }, + ) + + async def build_messages(self) -> list[Message]: + """Construct messages with context, memory_target, and memory_type information.""" + system_prompt = self.prompt_format( + prompt_name="system_prompt", + context=self.description + "\n" + format_messages(self.get_messages()), + memory_type=self.memory_type.value, + memory_target=self.memory_target, + ) + + messages = [ + Message(role=Role.SYSTEM, content=system_prompt), + Message(role=Role.USER, content=self.get_prompt("user_message")), + ] + return messages + + async def _reasoning_step(self, messages: list[Message], step: int, **kwargs) -> tuple[Message, bool]: + return await super()._reasoning_step(messages, step, **kwargs) + + async def _acting_step(self, assistant_message: Message, step: int, **kwargs) -> list[Message]: + """Execute tool calls with memory_target, memory_type, and author context.""" + messages: list[Message] = await super()._acting_step( + assistant_message, + step, + memory_type=self.memory_type.value, + memory_target=self.memory_target, + ref_memory_id=self.ref_memory_id, + author=self.author, + **kwargs, + ) + return messages diff --git a/reme_ai/mem_agent/wk/personal_summarizer_wk.yaml b/reme_ai/mem_agent/wk/personal_summarizer_wk.yaml new file mode 100644 index 00000000..9f3bbb43 --- /dev/null +++ b/reme_ai/mem_agent/wk/personal_summarizer_wk.yaml @@ -0,0 +1,40 @@ +tool: | + Extract and store personal memories from conversation context using a three-step workflow. + Use this tool to analyze dialogues and extract important personal information about users, + such as preferences, habits, personal background, relationships, and significant facts. + +system_prompt: | + You are a professional memory agent managing **{memory_type}** memories about **{memory_target}** for the main agent. + + ## Latest Conversation: + The context below contains the most recent conversation. Each message is formatted as: `round [] : ` where timestamp is `YYYY-MM-DD HH:MM:SS`. + {context} + + **CRITICAL**: Extract information ONLY from what is explicitly stated. DO NOT infer, assume, or fabricate any information. + + ## Your Tasks + + ### Step 1: Generate Memory Drafts + Use `AddMemoryDrafts` to extract key facts from the latest conversation. + - Extract important information: preferences, habits, currentstatus, personal details, key facts, decisions, or conclusions. + - Use clear, concise phrasing based strictly on explicit statements. + - Record the timestamp of the source message for each memory including the year, month, and day. + + ### Step 2: Retrieve Similar and Recent Memories + Use `RetrieveRecentAndSimilarMemories` to query historical memories for each draft. + - Search for semantically similar memories and recent memories. + - This ensures Step 3 avoids duplicates and properly updates existing memories. + + ### Step 3: Update Memories + Use `UpdateMemories` to update the memory store by combining drafts with historical memories. + - **Delete conflicts**: Remove old memories that contradict the new drafts (keep most recent/accurate). + - **Add new**: Add drafts that represent completely new information. + - **Skip duplicates**: Do not add drafts that duplicate existing memories. + - **Preserve others**: Keep unrelated historical memories unchanged. + - Write concise memories using minimum words needed. Ensure no information loss. + +user_message: | + Please analyze the context and update the memory store following the three-step workflow: + 1. First use `AddMemoryDrafts` to generate initial memory drafts + 2. Then use `RetrieveRecentAndSimilarMemories` to find related existing memories + 3. Finally use `UpdateMemories` to remove outdated memories and add new consolidated memories diff --git a/reme_ai/mem_agent/wk/reme_retriever_wk.py b/reme_ai/mem_agent/wk/reme_retriever_wk.py new file mode 100644 index 00000000..c98de937 --- /dev/null +++ b/reme_ai/mem_agent/wk/reme_retriever_wk.py @@ -0,0 +1,44 @@ +"""ReMe retriever v2 that autonomously retrieves memories from multiple angles.""" + +from typing import List + +from ..base_memory_agent import BaseMemoryAgent +from ...core.enumeration import Role +from ...core.schema import Message +from ...core.utils import format_messages + + +class ReMeRetrieverV2(BaseMemoryAgent): + + def __init__(self, meta_memories: list[dict] | None = None, **kwargs): + super().__init__(**kwargs) + self.meta_memories: list[dict] = meta_memories or [] + + async def _read_meta_memories(self) -> str: + """Fetch all meta-memory entries that define specialized memory agents.""" + from ...mem_tool import ReadMetaMemory + + op = ReadMetaMemory(enable_identity_memory=False) + return op.format_memory_metadata(self.meta_memories) + + async def build_messages(self) -> List[Message]: + """Build messages with system prompt and user message.""" + if self.context.get("query"): + context = self.context.query + elif self.context.get("messages"): + context = format_messages(self.context.messages) + else: + raise ValueError("input must have either `query` or `messages`") + + system_prompt = self.prompt_format( + prompt_name="system_prompt", + meta_memory_info=await self._read_meta_memories(), + context=context, + ) + + messages = [ + Message(role=Role.SYSTEM, content=system_prompt), + Message(role=Role.USER, content=self.get_prompt("user_message")), + ] + + return messages diff --git a/reme_ai/mem_agent/wk/reme_retriever_wk.yaml b/reme_ai/mem_agent/wk/reme_retriever_wk.yaml new file mode 100644 index 00000000..281796d6 --- /dev/null +++ b/reme_ai/mem_agent/wk/reme_retriever_wk.yaml @@ -0,0 +1,125 @@ +tool: | + Autonomously retrieve relevant memories from multiple angles to answer user questions. + This retriever will: + - Try multiple vector search strategies (direct, metadata-filtered, partial) + - Attempt at least 3 different retrieval approaches before giving up + - Fall back to reading original conversation history if vector search is insufficient + - Clearly state "I don't know" if information cannot be found after exhaustive searching + - NEVER hallucinate or fabricate information not present in retrieved memories + Use this when you need comprehensive memory retrieval with persistent searching. + +system_prompt: | + You are an autonomous memory retrieval agent. Your task is to persistently search for relevant memories from multiple angles to answer the user's question. + + ## Available Meta Memories + Format: "- (): " + {meta_memory_info} + + ## User Context + {context} + + ## Your Retrieval Strategy + + You MUST use the `retrieve_memories` tool to search for relevant information. This is a MANDATORY step - do not skip it. + + 1. **Multi-Angle Vector Retrieval** (REQUIRED - at least 3 attempts): + You must try AT LEAST 3 different retrieval approaches using `retrieve_memories`: + + a) **Direct Vector Search**: Use the user's question directly or with minimal reformulation + - Query the most relevant memory_type and memory_target + - Use straightforward query phrasing + + b) **Alternative Phrasing**: Reformulate the query from a different angle + - Use synonyms or different expressions + - Break down complex questions into simpler components + - Try more specific or more general queries + + c) **Metadata-Filtered Search**: Add metadata filters to narrow down results + - **Time-based filtering**: Use year/month/day metadata fields to filter by time periods + * Example: {{"year": 2024}} for memories from 2024 + * Example: {{"year": 2024, "month": 5}} for memories from May 2024 + * Example: {{"year": 2024, "month": 5, "day": 15}} for memories from a specific date + - Combine vector search with metadata constraints + - Try partial metadata filtering if full filtering yields nothing (e.g., only year, or year+month) + + d) **Cross-Memory-Type Search**: If applicable, search across different memory types + - Try different memory_type and memory_target combinations + - Some information might be stored in unexpected memory categories + + e) **Keyword Extraction**: Extract key entities/concepts and search for them + - Identify important names, places, concepts + - Search for each key element separately + + 2. **Evaluate Retrieval Results** (After each attempt): + - Review what memories were returned + - Assess if they contain sufficient information to answer the question + - If insufficient, identify what's missing and adjust your next query accordingly + - Track which retrieval strategies you've already tried + + 3. **Persist Through Failures**: + - DO NOT give up after 1-2 failed attempts + - If a retrieval returns no results or irrelevant results, try a different approach + - Consider that the information might be phrased differently than expected + - Be creative with query reformulation + + 4. **Fallback to History Reading** (Only after 3+ vector retrieval attempts): + - If after at least 3 different vector retrieval attempts you still lack sufficient information: + * If any retrieved memories contain `ref_memory_id`, use `read_history` to read the original conversation + * Use `read_history` with the `ref_memory_id` to get complete context + * This can reveal details that weren't captured in the memory summaries + + 5. **Answer the Question**: + - Once you have sufficient information, provide a direct answer based ONLY on retrieved memories + - DO NOT fabricate, guess, or infer information not present in the memories + - **CRITICAL**: If after 3+ retrieval attempts you still cannot find relevant information: + * Simply state: "I don't know. After searching from multiple angles, I could not find relevant information to answer this question." + * DO NOT make up answers or hallucinate information + * DO NOT provide speculative or guessed responses + * It is better to say "I don't know" than to provide incorrect information + + ## Important Guidelines + + - **Be Persistent**: Always try at least 3 different retrieval strategies before concluding no information exists + - **Be Creative**: If one query approach fails, think of alternative ways to phrase or decompose the question + - **Use Tools**: You MUST use `retrieve_memories` for vector search. Use `read_history` if you have `ref_memory_id` and need more details + - **No Hallucination**: NEVER fabricate, guess, or hallucinate information. Only answer based on what you actually retrieved from memories + - **Admit When You Don't Know**: If after 3+ attempts you cannot find relevant information, clearly say "I don't know" rather than making up an answer + - **Track Your Attempts**: Keep count of how many different retrieval strategies you've tried + - **Metadata Awareness**: Utilize metadata filters when they might help narrow down results + * Memories store time information in metadata as year/month/day fields + * Use time-based filters when the question involves specific time periods or dates + * Try progressive filtering: start with year, then add month, then day if needed + + ## Example Retrieval Flow + + **Example 1: Simple Query** + Attempt 1: Direct query "user's favorite food" + → Result: No relevant memories found + + Attempt 2: Reformulated query "what does user like to eat" + → Result: Some memories about meals, but not specific preferences + + Attempt 3: Keyword search "food preferences" with metadata filter + → Result: Found relevant memory with ref_memory_id + + Attempt 4: Use read_history with ref_memory_id to get full context + → Result: Found detailed conversation about favorite foods + + Answer: [Provide answer based on retrieved information] + + **Example 2: Time-based Query** + Question: "What did the user do last summer?" + + Attempt 1: Direct query "user activities summer" with metadata {{"year": 2025, "month": [6, 7, 8]}} + → Result: Found some vacation memories + + Attempt 2: Broader query "user summer vacation travel" with metadata {{"year": 2025}} + → Result: Found additional travel-related memories + + Attempt 3: Use read_history for memories with ref_memory_id to get detailed context + → Result: Complete picture of summer activities + + Answer: [Provide answer based on retrieved information] + +user_message: | + Please retrieve relevant memories and answer the question. Remember to try multiple retrieval approaches before giving up. diff --git a/reme_ai/mem_agent/wk/reme_summarizer_wk.py b/reme_ai/mem_agent/wk/reme_summarizer_wk.py new file mode 100644 index 00000000..02a7dbf3 --- /dev/null +++ b/reme_ai/mem_agent/wk/reme_summarizer_wk.py @@ -0,0 +1,88 @@ +from loguru import logger + +from ..base_memory_agent import BaseMemoryAgent +from ...core.enumeration import Role, MemoryType +from ...core.schema import Message, MemoryNode, ToolCall +from ...core.utils import format_messages + + +class ReMeSummarizerWk(BaseMemoryAgent): + + def __init__(self, meta_memories: list[dict] | None = None, **kwargs): + """Initialize with meta memories list.""" + super().__init__(**kwargs) + self.meta_memories: list[dict] = meta_memories or [] + + def _build_tool_call(self) -> ToolCall: + return ToolCall( + **{ + "description": self.get_prompt("tool"), + "parameters": { + "type": "object", + "properties": { + "messages": { + "type": "array", + "items": { + "type": "object", + "properties": { + "role": { + "type": "string", + "description": "role", + }, + "content": { + "type": "string", + "description": "content", + }, + }, + "required": ["role", "content"], + }, + }, + }, + "required": ["messages"], + }, + }, + ) + + async def _read_meta_memories(self) -> str: + from ...mem_tool import ReadMetaMemory + + return ReadMetaMemory().format_memory_metadata(self.meta_memories) + + async def build_messages(self) -> list[Message]: + """Construct initial messages with context and meta-memory information.""" + messages = [Message(**m) if isinstance(m, dict) else m for m in self.context.messages] + self.context["messages_formated"] = self.description + "\n" + format_messages(messages) + self.context["ref_memory_id"] = MemoryNode( + memory_type=MemoryType.HISTORY, + content=self.context["messages_formated"], + ).memory_id + + meta_memory_info = await self._read_meta_memories() + logger.info(f"meta_memory_info={meta_memory_info}") + + system_prompt = self.prompt_format( + prompt_name="system_prompt", + meta_memory_info=meta_memory_info, + context=self.context["messages_formated"], + ) + + user_message = self.get_prompt("user_message") + messages = [ + Message(role=Role.SYSTEM, content=system_prompt), + Message(role=Role.USER, content=user_message), + ] + + return messages + + async def _acting_step(self, assistant_message: Message, step: int, **kwargs) -> list[Message]: + """Execute tool calls with ref_memory_id and author context.""" + return await super()._acting_step( + assistant_message, + step, + messages=self.context.get("messages", []), + description=self.context.get("description"), + ref_memory_id=self.context["ref_memory_id"], + messages_formated=self.context["messages_formated"], + author=self.author, + **kwargs, + ) diff --git a/reme_ai/mem_agent/wk/reme_summarizer_wk.yaml b/reme_ai/mem_agent/wk/reme_summarizer_wk.yaml new file mode 100644 index 00000000..30792a08 --- /dev/null +++ b/reme_ai/mem_agent/wk/reme_summarizer_wk.yaml @@ -0,0 +1,25 @@ +tool: | + Orchestrate the complete memory summarization for the agent. + +system_prompt: | + You are a Memory Agent responsible for performing necessary updates and summaries of the main Agent's memories based on the **context**. + + # Context + {context} + + ## Main Agent's Meta Memory + Each line of meta memory indicates the existence of a specialized Memory Agent dedicated to deep summarization and updating of memories within a specific dimension (memory_type + memory_target). + Format: "- (): " + {meta_memory_info} + + ## Your Task + Use `summary_and_hands_off` tool to: + 1. Create a concise summary in `summary_content` that captures key points, decisions, or important facts from the context. + 2. Identify which memory dimensions need updates and specify them in `memory_tasks` (each with `memory_type` and `memory_target`). + - The `memory_type` and `memory_target` must exactly match existing entries in the "Main Agent's Meta Memory" listed above. + - Multiple tasks can be specified to enable parallel processing by specialized agents. + + Note: If the context contains no memorable information (e.g., simple greetings), output ``. + +user_message: | + Please perform your task based on the context. diff --git a/reme_ai/mem_tool/wk/__init__.py b/reme_ai/mem_tool/wk/__init__.py new file mode 100644 index 00000000..9df51cc6 --- /dev/null +++ b/reme_ai/mem_tool/wk/__init__.py @@ -0,0 +1,15 @@ +from .add_memory import AddMemory +from .delete_memory import DeleteMemory +from .read_history import ReadHistory +from .summary_and_hands_off import SummaryAndHandsOff +from .update_memory import UpdateMemory +from .vector_retrieve_memory import VectorRetrieveMemory + +__all__ = [ + "AddMemory", + "DeleteMemory", + "ReadHistory", + "SummaryAndHandsOff", + "UpdateMemory", + "VectorRetrieveMemory", +] diff --git a/reme_ai/mem_tool/wk/add_memory.py b/reme_ai/mem_tool/wk/add_memory.py new file mode 100644 index 00000000..61cb154c --- /dev/null +++ b/reme_ai/mem_tool/wk/add_memory.py @@ -0,0 +1,79 @@ +from loguru import logger + +from ..base_memory_tool import BaseMemoryTool +from ...core.schema import MemoryNode + + +class AddMemory(BaseMemoryTool): + + def __init__(self, **kwargs): + kwargs['enable_multiple'] = True + super().__init__(**kwargs) + + def _build_item_schema(self) -> tuple[dict, list[str]]: + properties = { + "memory_content": { + "type": "string", + "description": self.get_prompt("memory_content"), + }, + "metadata": { + "type": "object", + "description": "metadata for the memory.", + } + } + required = ["memory_content"] + return properties, required + + def _build_multiple_parameters(self) -> dict: + item_properties, required_fields = self._build_item_schema() + return { + "type": "object", + "properties": { + "memories": { + "type": "array", + "description": self.get_prompt("memories"), + "items": { + "type": "object", + "properties": item_properties, + "required": required_fields, + }, + }, + }, + "required": ["memories"], + } + + def _extract_memory_data(self, mem_dict: dict) -> tuple[str, dict]: + memory_content = mem_dict.get("memory_content", "") + raw_metadata = mem_dict.get("metadata", {}) + metadata = {key: str(value).strip() for key, value in raw_metadata.items() if value} + return memory_content, metadata + + async def execute(self): + memory_nodes: list[MemoryNode] = [] + + memories: list[dict] = self.context.get("memories", []) + if not memories: + self.output = "No memories provided for addition." + return + + for mem in memories: + memory_content, metadata = self._extract_memory_data(mem) + if not memory_content: + logger.warning("Skipping memory with empty content") + continue + + memory_nodes.append(self._build_memory_node(memory_content, metadata=metadata)) + + if not memory_nodes: + self.output = "No valid memories provided for addition." + return + + vector_nodes = [node.to_vector_node() for node in memory_nodes] + vector_ids: list[str] = [node.vector_id for node in vector_nodes] + + await self.vector_store.delete(vector_ids=vector_ids) + await self.vector_store.insert(nodes=vector_nodes) + self.memory_nodes = memory_nodes + + self.output = f"Successfully added {len(memory_nodes)} memories to vector_store." + logger.info(self.output) diff --git a/reme_ai/mem_tool/wk/add_memory.yaml b/reme_ai/mem_tool/wk/add_memory.yaml new file mode 100644 index 00000000..1c89d161 --- /dev/null +++ b/reme_ai/mem_tool/wk/add_memory.yaml @@ -0,0 +1,8 @@ +tool_multiple: | + Add multiple memories to the vector store for future retrieval. + +memory_content: | + The content of the memory to store. + +memories: | + A list of memory objects to store. diff --git a/reme_ai/mem_tool/wk/delete_memory.py b/reme_ai/mem_tool/wk/delete_memory.py new file mode 100644 index 00000000..73b7ba0d --- /dev/null +++ b/reme_ai/mem_tool/wk/delete_memory.py @@ -0,0 +1,35 @@ +from loguru import logger + +from ..base_memory_tool import BaseMemoryTool + + +class DeleteMemory(BaseMemoryTool): + + def __init__(self, **kwargs): + kwargs['enable_multiple'] = True + super().__init__(**kwargs) + + def _build_multiple_parameters(self) -> dict: + return { + "type": "object", + "properties": { + "memory_ids": { + "type": "array", + "description": self.get_prompt("memory_ids"), + "items": {"type": "string"}, + }, + }, + "required": ["memory_ids"], + } + + async def execute(self): + memory_ids = [m for m in self.context.get("memory_ids", []) if m] + + if not memory_ids: + self.output = "No valid memory IDs provided for deletion." + return + + await self.vector_store.delete(vector_ids=memory_ids) + self.memory_nodes = memory_ids + self.output = f"Successfully deleted {len(memory_ids)} memories from vector_store." + logger.info(self.output) diff --git a/reme_ai/mem_tool/wk/delete_memory.yaml b/reme_ai/mem_tool/wk/delete_memory.yaml new file mode 100644 index 00000000..cbc1de55 --- /dev/null +++ b/reme_ai/mem_tool/wk/delete_memory.yaml @@ -0,0 +1,5 @@ +tool_multiple: | + Delete multiple memories from the vector store using their unique IDs. + +memory_ids: | + A list of unique identifiers (memory_ids) of the memories to delete. diff --git a/reme_ai/mem_tool/wk/read_history.py b/reme_ai/mem_tool/wk/read_history.py new file mode 100644 index 00000000..945c02e2 --- /dev/null +++ b/reme_ai/mem_tool/wk/read_history.py @@ -0,0 +1,41 @@ +from loguru import logger + +from ..base_memory_tool import BaseMemoryTool +from ...core.schema import MemoryNode + + +class ReadHistory(BaseMemoryTool): + def __init__(self, **kwargs): + kwargs["enable_multiple"] = False + super().__init__(**kwargs) + + def _build_parameters(self) -> dict: + return { + "type": "object", + "properties": { + "ref_memory_id": { + "type": "string", + "description": self.get_prompt("ref_memory_id"), + }, + }, + "required": ["ref_memory_id"], + } + + async def execute(self): + ref_memory_id = self.context.get("ref_memory_id", "") + + if not ref_memory_id: + self.output = "No valid reference memory ID provided." + logger.warning(self.output) + return + + nodes = await self.vector_store.get(vector_ids=[ref_memory_id]) + + if not nodes: + self.output = f"No history memory found with ID: {ref_memory_id}" + logger.warning(self.output) + return + + memory = MemoryNode.from_vector_node(nodes[0]) + self.output = memory.content + logger.info(f"Successfully read history memory: {ref_memory_id}") diff --git a/reme_ai/mem_tool/wk/read_history.yaml b/reme_ai/mem_tool/wk/read_history.yaml new file mode 100644 index 00000000..0b806ae8 --- /dev/null +++ b/reme_ai/mem_tool/wk/read_history.yaml @@ -0,0 +1,5 @@ +tool: | + Read original history dialogue by memory ID. + +ref_memory_id: | + Memory ID to query the original history. diff --git a/reme_ai/mem_tool/wk/summary_and_hands_off.py b/reme_ai/mem_tool/wk/summary_and_hands_off.py new file mode 100644 index 00000000..9a6a76cd --- /dev/null +++ b/reme_ai/mem_tool/wk/summary_and_hands_off.py @@ -0,0 +1,143 @@ +import json +from typing import TYPE_CHECKING + +from loguru import logger + +from ..base_memory_tool import BaseMemoryTool +from ...core.enumeration import MemoryType +from ...core.schema import MemoryNode, Message + +if TYPE_CHECKING: + from ...mem_agent import BaseMemoryAgent + + +class SummaryAndHandsOff(BaseMemoryTool): + def __init__(self, memory_agents: list["BaseMemoryAgent"], **kwargs): + kwargs["enable_multiple"] = True + kwargs["sub_ops"] = memory_agents or [] + super().__init__(**kwargs) + from ...mem_agent import BaseMemoryAgent + + self.sub_ops: list[BaseMemoryAgent] = [a for a in self.sub_ops if isinstance(a, BaseMemoryAgent)] + self.messages: list[Message] = [] + + @property + def memory_agent_dict(self) -> dict[MemoryType, "BaseMemoryAgent"]: + return {a.memory_type: a for a in self.sub_ops} + + def _build_item_schema(self) -> tuple[dict, list[str]]: + properties = { + "memory_type": { + "type": "string", + "description": self.get_prompt("memory_type"), + "enum": [k.value for k in self.memory_agent_dict], + }, + "memory_target": { + "type": "string", + "description": self.get_prompt("memory_target"), + }, + } + required = ["memory_type", "memory_target"] + return properties, required + + def _build_multiple_parameters(self) -> dict: + item_properties, required_fields = self._build_item_schema() + return { + "type": "object", + "properties": { + "summary_content": { + "type": "string", + "description": self.get_prompt("summary_content"), + }, + "memory_tasks": { + "type": "array", + "description": self.get_prompt("memory_tasks"), + "items": { + "type": "object", + "properties": item_properties, + "required": required_fields, + }, + }, + }, + "required": ["summary_content", "memory_tasks"], + } + + @staticmethod + def _parse_memory_type_target(task: dict): + return { + "memory_type": MemoryType(task.get("memory_type", "")), + "memory_target": task.get("memory_target", ""), + } + + def _collect_tasks(self) -> list[dict]: + tasks = [] + for task in self.context.get("memory_tasks", []): + tasks.append(self._parse_memory_type_target(task)) + return tasks + + async def execute(self): + summary_content = self.context.get("summary_content", "") + assert summary_content, "No summary content provided." + + summary_node = MemoryNode( + memory_type=MemoryType.HISTORY, + memory_target="", + when_to_use=summary_content, + content=self.messages_formated, + ref_memory_id="", + author=self.author, + metadata={}, + ) + logger.info(f"Adding summary node: {summary_node.model_dump_json(indent=2, exclude_none=True)}") + self.memory_nodes.append(summary_node) + vector_node = summary_node.to_vector_node() + await self.vector_store.delete(vector_ids=[vector_node.vector_id]) + await self.vector_store.insert([vector_node]) + + tasks = self._collect_tasks() + if not tasks: + self.output = "No valid memory tasks to execute." + return + + agent_list = [] + for i, task in enumerate(tasks): + memory_type: MemoryType = task["memory_type"] + memory_target: str = task["memory_target"] + + if memory_type not in self.memory_agent_dict: + logger.warning(f"No agent found for memory_type={memory_type}") + continue + + agent = self.memory_agent_dict[memory_type].copy() + agent_list.append([agent, memory_type, memory_target]) + + logger.info(f"Task {i}: Submitting {memory_type.value} agent for target={memory_target}") + self.submit_async_task( + agent.call, + query=self.context.get("query", ""), + messages=self.context.get("messages", []), + memory_type=memory_type, + memory_target=memory_target, + description=self.context.get("description"), + ref_memory_id=self.context.get("ref_memory_id", ""), + ) + + await self.join_async_tasks() + + results = [] + for i, (agent, memory_type, memory_target) in enumerate(agent_list): + result_str = str(agent.output) + if agent.memory_nodes: + self.memory_nodes.extend(agent.memory_nodes) + if agent.messages: + self.messages.extend(agent.messages) + + results.append({ + "memory_type": memory_type.value, + "memory_target": memory_target, + "result": result_str[:100] + ("..." if len(result_str) > 100 else ""), + }) + logger.info(f"Task {i}: Completed {memory_type.value} agent for target={memory_target}") + + results_str = json.dumps(results, ensure_ascii=False, indent=2) + self.output = f"Successfully executed summary and {len(results)} hands-off task(s):\n{results_str}" diff --git a/reme_ai/mem_tool/wk/summary_and_hands_off.yaml b/reme_ai/mem_tool/wk/summary_and_hands_off.yaml new file mode 100644 index 00000000..d2a42735 --- /dev/null +++ b/reme_ai/mem_tool/wk/summary_and_hands_off.yaml @@ -0,0 +1,14 @@ +tool_multiple: | + Summarize and distribute memory tasks to appropriate agents. + +summary_content: | + The summarized content to store. + +memory_type: | + The type of memory to process. + +memory_target: | + The target entity for this memory. + +memory_tasks: | + A list of memory tasks to distribute. diff --git a/reme_ai/mem_tool/wk/update_memory.py b/reme_ai/mem_tool/wk/update_memory.py new file mode 100644 index 00000000..c419261f --- /dev/null +++ b/reme_ai/mem_tool/wk/update_memory.py @@ -0,0 +1,86 @@ +from loguru import logger + +from ..base_memory_tool import BaseMemoryTool +from ...core.schema import MemoryNode + + +class UpdateMemory(BaseMemoryTool): + + def __init__(self, **kwargs): + kwargs['enable_multiple'] = True + super().__init__(**kwargs) + + def _build_item_schema(self) -> tuple[dict, list[str]]: + properties = { + "memory_id": { + "type": "string", + "description": self.get_prompt("memory_id"), + }, + "memory_content": { + "type": "string", + "description": self.get_prompt("memory_content"), + }, + "metadata": { + "type": "object", + "description": "metadata for the memory.", + } + } + required = ["memory_id", "memory_content", "metadata"] + return properties, required + + def _build_multiple_parameters(self) -> dict: + item_properties, required_fields = self._build_item_schema() + return { + "type": "object", + "properties": { + "memories": { + "type": "array", + "description": self.get_prompt("memories"), + "items": { + "type": "object", + "properties": item_properties, + "required": required_fields, + }, + }, + }, + "required": ["memories"], + } + + def _extract_memory_data(self, mem_dict: dict) -> tuple[str, str, dict]: + memory_id = mem_dict.get("memory_id", "") + memory_content = mem_dict.get("memory_content", "") + raw_metadata = mem_dict.get("metadata", {}) + metadata = {key: str(value).strip() for key, value in raw_metadata.items() if value} + return memory_id, memory_content, metadata + + async def execute(self): + old_memory_ids: list[str] = [] + new_memory_nodes: list[MemoryNode] = [] + + memories: list[dict] = self.context.get("memories", []) + if not memories: + self.output = "No memories provided for update." + return + + for mem in memories: + memory_id, memory_content, metadata = self._extract_memory_data(mem) + if not memory_id or not memory_content: + logger.warning(f"Skipping memory with missing id or content: {mem}") + continue + old_memory_ids.append(memory_id) + new_memory_nodes.append(self._build_memory_node(memory_content, metadata=metadata)) + + if not old_memory_ids or not new_memory_nodes: + self.output = "No valid memories provided for update." + return + + vector_nodes = [node.to_vector_node() for node in new_memory_nodes] + new_vector_ids = [node.vector_id for node in vector_nodes] + + all_ids_to_delete = list(set(old_memory_ids + new_vector_ids)) + await self.vector_store.delete(vector_ids=all_ids_to_delete) + await self.vector_store.insert(nodes=vector_nodes) + self.memory_nodes = new_memory_nodes + + self.output = f"Successfully updated {len(new_memory_nodes)} memories in vector_store." + logger.info(self.output) diff --git a/reme_ai/mem_tool/wk/update_memory.yaml b/reme_ai/mem_tool/wk/update_memory.yaml new file mode 100644 index 00000000..efb253dc --- /dev/null +++ b/reme_ai/mem_tool/wk/update_memory.yaml @@ -0,0 +1,11 @@ +tool_multiple: | + Update multiple memories in the vector store by replacing old memories with new content. + +memory_id: | + The unique identifier of the memory to be updated. + +memory_content: | + The new content of the memory to store. + +memories: | + A list of memory update objects. diff --git a/reme_ai/mem_tool/wk/vector_retrieve_memory.py b/reme_ai/mem_tool/wk/vector_retrieve_memory.py new file mode 100644 index 00000000..0cfc367e --- /dev/null +++ b/reme_ai/mem_tool/wk/vector_retrieve_memory.py @@ -0,0 +1,165 @@ +from loguru import logger + +from ..base_memory_tool import BaseMemoryTool +from ...core.enumeration import MemoryType +from ...core.schema import MemoryNode, VectorNode +from ...core.utils import deduplicate_memories + + +class VectorRetrieveMemory(BaseMemoryTool): + + def __init__( + self, + add_memory_type_target: bool = False, + top_k: int = 20, + **kwargs, + ): + super().__init__(**kwargs) + self.add_memory_type_target: bool = add_memory_type_target + self.top_k: int = top_k + + def _build_query_schema(self) -> tuple[dict, list[str]]: + properties = {} + required = [] + + if self.add_memory_type_target: + properties["memory_type"] = { + "type": "string", + "description": self.get_prompt("memory_type"), + } + properties["memory_target"] = { + "type": "string", + "description": self.get_prompt("memory_target"), + } + required.extend(["memory_type", "memory_target"]) + + properties["query"] = { + "type": "string", + "description": self.get_prompt("query"), + } + required.append("query") + + properties["metadata"] = { + "type": "object", + "description": self.get_prompt("metadata"), + } + + return properties, required + + def _build_parameters(self) -> dict: + properties, required = self._build_query_schema() + return { + "type": "object", + "properties": properties, + "required": required, + } + + def _build_multiple_parameters(self) -> dict: + item_properties, item_required = self._build_query_schema() + return { + "type": "object", + "properties": { + "query_items": { + "type": "array", + "description": self.get_prompt("query_items"), + "items": { + "type": "object", + "properties": item_properties, + "required": item_required, + }, + }, + }, + "required": ["query_items"], + } + + async def _retrieve_by_query( + self, + memory_type: str, + memory_target: str, + query: str, + metadata: dict | None = None, + ) -> list[MemoryNode]: + filter_dict = { + "memory_type": [memory_type], + "memory_target": [memory_target], + } + + if metadata: + for key, value in metadata.items(): + if value: + value = str(value).strip() + filter_dict[key] = [value] if not isinstance(value, list) else value + + nodes: list[VectorNode] = await self.vector_store.search(query=query, limit=self.top_k, filters=filter_dict) + + memory_nodes: list[MemoryNode] = [MemoryNode.from_vector_node(n) for n in nodes] + + filtered_memory_nodes = [ + m for m in memory_nodes if not (m.memory_type == MemoryType.TOOL and m.when_to_use != query) + ] + + return filtered_memory_nodes + + async def execute(self): + default_memory_type: str = self.context.get("memory_type", "") + default_memory_target: str = self.context.get("memory_target", "") + + if self.enable_multiple: + query_items: list[dict] = self.context.get("query_items", []) + if not query_items: + self.output = "No query items provided for retrieval." + return + else: + query = self.context.get("query", "") + if not query: + self.output = "No query provided for retrieval." + return + + query_items = [ + { + "memory_type": default_memory_type, + "memory_target": default_memory_target, + "query": query, + }, + ] + + query_items = [item for item in query_items if item.get("query")] + + if not query_items: + self.output = "No valid query texts provided for retrieval." + return + + memory_nodes: list[MemoryNode] = [] + for item in query_items: + memory_type = item.get("memory_type") or default_memory_type + memory_target = item.get("memory_target") or default_memory_target + metadata = item.get("metadata", {}) + + if not memory_type or not memory_target: + logger.warning(f"Skipping query with missing memory_type or memory_target: {item}") + continue + + retrieved = await self._retrieve_by_query( + memory_type=memory_type, + memory_target=memory_target, + query=item["query"], + metadata=metadata, + ) + memory_nodes.extend(retrieved) + + memory_nodes = deduplicate_memories(memory_nodes) + + retrieved_memory_ids = {node.memory_id for node in self.retrieved_nodes if node.memory_id} + + new_memory_nodes = [node for node in memory_nodes if node.memory_id not in retrieved_memory_ids] + + self.retrieved_nodes.extend(new_memory_nodes) + + self.memory_nodes = new_memory_nodes + + if not new_memory_nodes: + self.output = "No new memory_nodes found matching the query (duplicates removed)." + else: + self.output = "\n".join([m.format_memory() for m in new_memory_nodes]) + + logger.info(f"Retrieved {len(memory_nodes)} memory_nodes, {len(new_memory_nodes)} new after deduplication") diff --git a/reme_ai/mem_tool/wk/vector_retrieve_memory.yaml b/reme_ai/mem_tool/wk/vector_retrieve_memory.yaml new file mode 100644 index 00000000..b09e7371 --- /dev/null +++ b/reme_ai/mem_tool/wk/vector_retrieve_memory.yaml @@ -0,0 +1,20 @@ +tool: | + Retrieve memories using vector similarity search. + +tool_multiple: | + Retrieve memories using multiple queries with vector similarity search. + +memory_type: | + The type of memory to search for. + +memory_target: | + The target of the memory to search within. + +query: | + The query text for vector similarity search. + +query_items: | + A list of query items for vector similarity search. + +metadata: | + Optional metadata filters for narrowing search results.