diff --git a/reme/__init__.py b/reme/__init__.py index 32f34911..cd5d4e6c 100644 --- a/reme/__init__.py +++ b/reme/__init__.py @@ -5,7 +5,7 @@ from . import config from . import core from . import tool from . import workflow -from .reme_app import ReMeApp +from .reme import ReMe __all__ = [ "agent", @@ -13,7 +13,7 @@ __all__ = [ "core", "tool", "workflow", - "ReMeApp", + "ReMe", ] __version__ = "0.3.0.0a1" diff --git a/reme/agent/memory/base_memory_agent.py b/reme/agent/memory/base_memory_agent.py index e3715ce6..72c90634 100644 --- a/reme/agent/memory/base_memory_agent.py +++ b/reme/agent/memory/base_memory_agent.py @@ -1,6 +1,9 @@ """Base memory agent for handling memory operations with tool-based reasoning.""" from abc import ABCMeta +from typing import Literal + +from loguru import logger from ...core.enumeration import MemoryType from ...core.op import BaseReact @@ -8,8 +11,36 @@ from ...core.schema import MemoryNode class BaseMemoryAgent(BaseReact, metaclass=ABCMeta): + """Base class for memory agents that handle memory operations with tool-based reasoning.""" + memory_type: MemoryType | None = None + @staticmethod + async def read_meta_memories(meta_memories: list[dict]) -> str: + """Read and format meta memory information from the provided metadata list.""" + from ...tool.memory import ReadMetaMemory + + meta_memory_info = ReadMetaMemory().format_memory_metadata(meta_memories) + logger.info(f"meta_memory_info={meta_memory_info}") + return meta_memory_info + + async def read_user_profile(self, show_id: Literal["profile", "history"] = "profile") -> str: + """Read current user profile.""" + from ...tool.memory import ReadUserProfile + + read_tool = ReadUserProfile(show_id=show_id) + await read_tool.call(memory_target=self.memory_target) + return str(read_tool.response.answer) + + @staticmethod + async def read_history_node() -> MemoryNode: + """Read and return the current history node from the context.""" + from ...tool.memory import AddHistory + + add_history_tool = AddHistory() + await add_history_tool.call() + return add_history_tool.context.history_node + @property def memory_target(self) -> str: """memory_target""" diff --git a/reme/agent/memory/default/__init__.py b/reme/agent/memory/default/__init__.py index fa06423d..516a81c1 100644 --- a/reme/agent/memory/default/__init__.py +++ b/reme/agent/memory/default/__init__.py @@ -1,3 +1,5 @@ +"""Default memory agents for personal and ReMe memory operations.""" + from .personal_retriever import PersonalRetriever from .personal_summarizer import PersonalSummarizer from .reme_retriever import ReMeRetriever diff --git a/reme/agent/memory/default/personal_retriever.py b/reme/agent/memory/default/personal_retriever.py index 9d68ea22..1bafa317 100644 --- a/reme/agent/memory/default/personal_retriever.py +++ b/reme/agent/memory/default/personal_retriever.py @@ -1,5 +1,8 @@ +"""Personal memory retriever agent for retrieving personal memories through vector search.""" + from ..base_memory_agent import BaseMemoryAgent from ....core.enumeration import Role, MemoryType +from ....core.op import BaseTool from ....core.schema import Message from ....core.utils import format_messages @@ -10,53 +13,44 @@ class PersonalRetriever(BaseMemoryAgent): memory_type: MemoryType = MemoryType.PERSONAL async def build_messages(self) -> list[Message]: - from ....tool.memory.vector import ReadUserProfile - - # Get context from query or messages - context = ( - self.context.query - if self.context.get("query") - else format_messages(self.context.messages) if self.context.get("messages") else None - ) - if not context: + if self.context.get("query"): + context = self.context.query + elif self.context.get("messages"): + context = self.description + "\n" + format_messages(self.context.messages) + else: raise ValueError("input must have either `query` or `messages`") - # Read user profile with history IDs - read_profile_tool = ReadUserProfile(show_ids="history") - await read_profile_tool.call(memory_type=self.memory_type.value, memory_target=self.memory_target) - self.context.user_profile = user_profile = read_profile_tool.output - return [ Message( - role=Role.USER, + role=Role.SYSTEM, content=self.prompt_format( - prompt_name="user_message", + prompt_name="system_prompt", memory_type=self.memory_type.value, memory_target=self.memory_target, - user_profile=user_profile, - context=context, + user_profile=await self.read_user_profile(show_id="history"), + context=context.strip(), ), ), + Message( + role=Role.USER, + content=self.get_prompt("user_message"), + ), ] - async def _acting_step(self, assistant_message: Message, step: int, **kwargs) -> list[Message]: + async def _acting_step( + self, + assistant_message: Message, + tools: list[BaseTool], + step: int, + stage: str = "", + **kwargs, + ) -> tuple[list[BaseTool], list[Message]]: + """Execute tool calls with memory context.""" return await super()._acting_step( assistant_message, + tools, step, memory_type=self.memory_type.value, memory_target=self.memory_target, **kwargs, ) - - async def execute(self): - """Execute retriever and check for memory found markers.""" - await super().execute() - - # Check output markers - if self.output: - if "" in self.output: - self.success = True - elif "" in self.output: - self.success = False - - self.meta_info = self.context.user_profile + "\n" + self.meta_info diff --git a/reme/agent/memory/default/personal_retriever.yaml b/reme/agent/memory/default/personal_retriever.yaml index 11fb9c9e..7b81e4cc 100644 --- a/reme/agent/memory/default/personal_retriever.yaml +++ b/reme/agent/memory/default/personal_retriever.yaml @@ -1,16 +1,48 @@ -tool: | - Retrieve personal memories via vector search and history reading to answer user questions. +system_prompt: | + You are a memory agent managing **{memory_type}** memories about **{memory_target}**. -user_message: | - You manage **{memory_type}** memories about **{memory_target}**. - - ## Profile + ## User Profile {user_profile} ## Question {context} - ## Task - 1. Vector search (`retrieve_memory`): Try 3-5 queries with different phrasings, entities, keywords, and time ranges [start, end] in YYYYMMDD - 2. Read context (`read_history`): Use history_id from results to get full conversations - 3. Respond: `` if found, `` if not found after thorough search + ## Retrieval Strategy + + **Tool 1: Vector Search (`retrieve_memory`)** + - Purpose: Search for relevant memories using semantic similarity + - Try at least 3-5 different queries before moving to next tool: + * Direct question reformulation + * Different phrasings and perspectives + * Entity-focused queries (names, places, events) + * Various keyword combinations + - Time range filtering (optional): + * Format: single date '20200101' or range '20200101,20200102' + * Example: '20200101,20200102' for 20200101 <= time <= 20200102 + * Single-sided: '0,20200102' (before date) or '20200101,99999999' (after date) + - If no results: retry with different time ranges or remove time constraints + + **Tool 2: Read History (`read_history`) - ONLY AFTER Tool 1** + - Purpose: Read full original conversation context + - Use this ONLY after completing multiple retrieve_memory attempts + - Extract history_id from retrieved memory results + - Prioritize most relevant or recent history entries + - Read multiple histories if needed for complete understanding + + ## Response Requirements + - Answer ONLY based on retrieved memories and user profile - NO hallucination or inference + - Always cite the source: reference specific memories with their timestamps + - If information conflicts, present all versions with their respective times + - Try multiple search angles before concluding no information exists + + ## Output Format + When answering, structure your response as follows: + + - [timestamp][Relevant memory content from search results] + - [timestamp][Relevant user profile information] + + If no relevant information found after thorough search (5+ queries), state: + "No relevant information found after thorough search using multiple query strategies." + +user_message: | + Answer the question following the retrieval strategy and response requirements above. \ No newline at end of file diff --git a/reme/agent/memory/default/personal_summarizer.py b/reme/agent/memory/default/personal_summarizer.py index e69403bd..a317b713 100644 --- a/reme/agent/memory/default/personal_summarizer.py +++ b/reme/agent/memory/default/personal_summarizer.py @@ -1,49 +1,67 @@ +"""Personal memory summarizer agent for two-phase personal memory processing.""" + from loguru import logger from ..base_memory_agent import BaseMemoryAgent from ....core.enumeration import Role, MemoryType +from ....core.op import BaseTool from ....core.schema import Message class PersonalSummarizer(BaseMemoryAgent): - """Extract and update personal memories in two phases: add summaries then update profile.""" + """Two-phase personal memory processor: retrieve/add memories then update profile.""" memory_type: MemoryType = MemoryType.PERSONAL - async def build_messages_phase1(self) -> list[Message]: - """Phase 1: AddSummaryMemory""" - history_node = self.context.history_node + async def _build_phase1_messages(self) -> list[Message]: + """Build messages for phase 1: retrieve and add memory.""" return [ Message( - role=Role.USER, + role=Role.SYSTEM, content=self.prompt_format( - prompt_name="user_message_phase1", - context=history_node.content, + prompt_name="system_prompt_phase1", + context=self.context.history_node.content, memory_type=self.memory_type.value, memory_target=self.memory_target, ), ), - ] - - async def build_messages_phase2(self, user_profile: str) -> list[Message]: - """Phase 2: UpdateUserProfile""" - history_node = self.context.history_node - return [ Message( role=Role.USER, - content=self.prompt_format( - prompt_name="user_message_phase2", - context=history_node.content, - memory_type=self.memory_type.value, - memory_target=self.memory_target, - user_profile=user_profile, - ), + content=self.get_prompt("user_message_phase1"), ), ] - async def _acting_step(self, assistant_message: Message, step: int, stage: str = "", **kwargs) -> list[Message]: + async def _build_phase2_messages(self) -> list[Message]: + """Build messages for phase 2: update user profile.""" + return [ + Message( + role=Role.SYSTEM, + content=self.prompt_format( + prompt_name="system_prompt_phase2", + context=self.context.history_node.content, + memory_type=self.memory_type.value, + memory_target=self.memory_target, + user_profile=await self.read_user_profile(show_id="profile"), + ), + ), + Message( + role=Role.USER, + content=self.get_prompt("user_message_phase2"), + ), + ] + + async def _acting_step( + self, + assistant_message: Message, + tools: list[BaseTool], + step: int, + stage: str = "", + **kwargs, + ) -> tuple[list[BaseTool], list[Message]]: + """Execute tool calls with memory context.""" return await super()._acting_step( assistant_message, + tools, step, stage=stage, memory_type=self.memory_type.value, @@ -54,61 +72,27 @@ class PersonalSummarizer(BaseMemoryAgent): ) async def execute(self): - """Execute two phases: AddSummaryMemory -> UpdateUserProfile""" - from ....tool.memory.vector import ReadUserProfile + """Execute two-phase memory processing: retrieve/add -> update profile.""" + tools = self.tools + for i, tool in enumerate(tools): + logger.info(f"[{self.__class__.__name__}] tool_call[{i}]={tool.tool_call.name}") - # Log tools - for i, tool in enumerate(self.tools): - logger.info(f"[{self.__class__.__name__}] step0.{i} tool_call={tool.tool_call.name}") - - # Phase 1: AddSummaryMemory - logger.info(f"[{self.__class__.__name__}-S1] Phase 1: AddSummaryMemory") - original_tools = self.tools.copy() - self.tools = [t for t in self.tools if t.tool_call.name == "add_summary_memory"] - - messages_phase1 = await self.build_messages_phase1() + messages_phase1 = await self._build_phase1_messages() for i, message in enumerate(messages_phase1): - logger.info( - f"[{self.__class__.__name__}-S1] phase1.step0.{i} {message.role} {message.simple_dump(enable_json_dump=True)}" - ) + role = message.name or message.role + logger.info(f"[{self.__class__.__name__}-S1] role={role} {message.simple_dump(as_dict=False)}") + tools_phase1, messages_phase1, success_phase1 = await self.react(messages_phase1, tools[:-1], stage="S1") - messages_phase1, success_phase1 = await self.react(messages_phase1, stage="S1") - if not success_phase1: - logger.warning(f"[{self.__class__.__name__}-S1] Phase 1 incomplete") - - # Phase 2: UpdateUserProfile - logger.info(f"[{self.__class__.__name__}-S2] Phase 2: UpdateUserProfile") - self.tools = original_tools - read_profile_tool = next((t for t in self.tools if t.tool_call.name == "read_user_profile"), None) - - user_profile = "" - if read_profile_tool: - logger.info(f"[{self.__class__.__name__}-S2] Loading user profile") - await read_profile_tool.call( - memory_type=self.memory_type.value, - memory_target=self.memory_target, - show_ids="profile", - ) - user_profile = str(read_profile_tool.output) - else: - logger.warning(f"[{self.__class__.__name__}-S2] ReadUserProfile tool not found") - - self.tools = [t for t in self.tools if t.tool_call.name == "update_user_profile"] - - messages_phase2 = await self.build_messages_phase2(user_profile) + messages_phase2 = await self._build_phase2_messages() for i, message in enumerate(messages_phase2): - logger.info( - f"[{self.__class__.__name__}-S2] phase2.step0.{i} {message.role} {message.simple_dump(enable_json_dump=True)}" - ) + role = message.name or message.role + logger.info(f"[{self.__class__.__name__}-S2] role={role} {message.simple_dump(as_dict=False)}") + tools_phase2, messages_phase2, success_phase2 = await self.react(messages_phase2, tools[-1:], stage="S2") - messages_phase2, success_phase2 = await self.react(messages_phase2, stage="S2") - - # Restore tools and set output - self.tools = original_tools - self.messages = messages_phase1 + messages_phase2 - self.success = success_phase1 and success_phase2 - self.output = ( - messages_phase2[-1].content - if self.success and messages_phase2 - else "Memory processing completed with issues." - ) + return { + "answer": (messages_phase1[-1].content if success_phase1 else "") + + (messages_phase2[-1].content if success_phase2 else ""), + "success": success_phase1 and success_phase2, + "messages": messages_phase1 + messages_phase2, + "tools": tools_phase1 + tools_phase2, + } diff --git a/reme/agent/memory/default/personal_summarizer.yaml b/reme/agent/memory/default/personal_summarizer.yaml index ea9b5d71..77da7843 100644 --- a/reme/agent/memory/default/personal_summarizer.yaml +++ b/reme/agent/memory/default/personal_summarizer.yaml @@ -1,26 +1,45 @@ +system_prompt_phase1: | + You are a memory agent managing **{memory_type}** memories about **{memory_target}**. + + ## Latest Conversation: + Message format: `round [] : ` (timestamp: YYYY-MM-DD HH:MM:SS). + {context} + + ## Task: Retrieve Similar Memories and Add New Memories + **CRITICAL**: Extract ONLY explicitly stated information. DO NOT infer, assume, or fabricate. + + ### Step 1: Retrieve Similar Memories + Use `retrieve_memory` to search for existing similar memories about **{memory_target}**. + - Use appropriate queries to find relevant existing memories + - Check if new information already exists in the memory store + + ### Step 2: Add New Memories + Use `add_memory` to add new memories: + - Extract and summarize important information about **{memory_target}** + - Set `conversation_time` (format: 2020-01-01 00:00:00; use 0000-00-00 00:00:00 if unavailable) + - If the information is completely identical to existing memory, skip adding + user_message_phase1: | - You manage **{memory_type}** memories about **{memory_target}**. + First retrieve similar memories, then extract and add new personal memories from the conversation. - ## Conversation +system_prompt_phase2: | + You are a memory agent managing **{memory_type}** memories about **{memory_target}**. + + ## Latest Conversation: + Message format: `round [] : ` (timestamp: YYYY-MM-DD HH:MM:SS). {context} - Format: `round [] : ` (timestamp: YYYY-MM-DD HH:MM:SS). - - ## Task - Extract memories with `add_memory`. Set `conversation_time` (YYYY-MM-DD HH:MM:SS, use 0000-00-00 00:00:00 if unavailable). Extract ONLY explicit information, no inference. - -user_message_phase2: | - You manage **{memory_type}** memories about **{memory_target}**. - - ## Conversation - {context} - - ## Profile + ## Current User Profile: + UserProfile format: `profile_id= conversation_time= `. {user_profile} - ## Task - Update profile with `UpdateUserProfile`: - - `profile_ids_to_delete`: Remove conflicting/redundant entries - - `profiles_to_add`: Add new entries with `conversation_time` and `profile_content` (complete, self-contained, mutually exclusive) + ## Task: Update Profile with `UpdateUserProfile` + **CRITICAL**: Extract ONLY explicitly stated information. DO NOT infer, assume, or fabricate. - Extract ONLY explicit information, no inference. + Synchronize profile/memories with new information from the conversation, including **{memory_target}**' current status: + - `profile_ids_to_delete`: Remove outdated, conflicting, or redundant entries. + - `profiles_to_add`: Add new profiles/memories with `conversation_time`, e.g. `YYYY-MM-DD HH:MM:SS`, {memory_target} did something. + - Maintain profiles that are concise, mutually exclusive, and collectively comprehensive with no information loss. + +user_message_phase2: | + Update user profile using `UpdateUserProfile` based on the conversation and current profile. \ No newline at end of file diff --git a/reme/agent/memory/default/reme_retriever.py b/reme/agent/memory/default/reme_retriever.py index 1508b84a..4b4c3e22 100644 --- a/reme/agent/memory/default/reme_retriever.py +++ b/reme/agent/memory/default/reme_retriever.py @@ -1,7 +1,8 @@ -from loguru import logger +"""ReMe retriever agent that orchestrates multiple memory agents to retrieve information.""" from ..base_memory_agent import BaseMemoryAgent from ....core.enumeration import Role +from ....core.op import BaseTool from ....core.schema import Message from ....core.utils import format_messages @@ -12,31 +13,22 @@ class ReMeRetriever(BaseMemoryAgent): def __init__(self, meta_memories: list[dict] | None = None, **kwargs): super().__init__(**kwargs) self.meta_memories: list[dict] = meta_memories or [] - self.meta_info_dict: dict[str, str] = {} - - async def _read_meta_memories(self) -> str: - from ....tool.memory import ReadMetaMemory - - meta_memory_info = ReadMetaMemory().format_memory_metadata(self.meta_memories) - logger.info(f"meta_memory_info={meta_memory_info}") - return meta_memory_info async def build_messages(self) -> list[Message]: - user_query = ( - self.context.query - if self.context.get("query") - else format_messages(self.context.messages) if self.context.get("messages") else None - ) - if not user_query: - raise ValueError("Input must have either `query` or `messages`") + if self.context.get("query"): + context = self.context.query + elif self.context.get("messages"): + context = self.description + "\n" + format_messages(self.context.messages) + else: + raise ValueError("input must have either `query` or `messages`") return [ Message( role=Role.SYSTEM, content=self.prompt_format( prompt_name="system_prompt", - meta_memory_info=await self._read_meta_memories(), - user_query=user_query, + meta_memory_info=await self.read_meta_memories(self.meta_memories), + context=context.strip(), ), ), Message( @@ -45,72 +37,46 @@ class ReMeRetriever(BaseMemoryAgent): ), ] - async def _acting_step(self, assistant_message: Message, step: int, **kwargs) -> list[Message]: - import asyncio - from ....tool.memory import HandsOff - - if not assistant_message.tool_calls: - return [] - - tool_list: list = [] - tool_result_messages: list[Message] = [] - tool_dict = {t.tool_call.name: t for t in self.tools} - stage_prefix = "" - - # Add context parameters - kwargs["query"] = self.context.get("query", "") - kwargs["messages"] = self.context.get("messages", []) - - for j, tool_call in enumerate(assistant_message.tool_calls): - if tool_call.name not in tool_dict: - logger.warning(f"[{self.__class__.__name__}{stage_prefix}] unknown tool_call.name={tool_call.name}") - continue - - logger.info( - f"[{self.__class__.__name__}{stage_prefix}] step{step + 1}.{j} submit tool_calls={tool_call.name} argument={tool_call.arguments}" - ) - tool_copy = tool_dict[tool_call.name].copy() - tool_copy.tool_call.id = tool_call.id - tool_list.append(tool_copy) - kwargs.update(tool_call.argument_dict) - self.submit_async_task(tool_copy.call, retrieved_nodes=self.retrieved_nodes, **kwargs) - if self.tool_call_interval > 0: - await asyncio.sleep(self.tool_call_interval) - - await self.join_async_tasks() - - for j, op in enumerate(tool_list): - if op.memory_nodes: - self.memory_nodes.extend(op.memory_nodes) - - if hasattr(op, "messages") and op.messages: - self.tool_messages.extend(op.messages) - - # Collect meta_info_dict from HandsOff - if isinstance(op, HandsOff) and hasattr(op, "meta_info_dict"): - self.meta_info_dict.update(op.meta_info_dict) - logger.info(f"Collected meta_info_dict from HandsOff: {len(op.meta_info_dict)} entries") - - tool_result = str(op.output) - tool_message = Message( - role=Role.TOOL, - content=tool_result, - tool_call_id=op.tool_call.id, - ) - tool_result_messages.append(tool_message) - self.meta_info += tool_result + "\n" - logger.info( - f"[{self.__class__.__name__}{stage_prefix}] step{step + 1}.{j} join tool_result={tool_result[:2000]}...\n\n" - ) - - return tool_result_messages + async def _acting_step( + self, + assistant_message: Message, + tools: list[BaseTool], + step: int, + stage: str = "", + **kwargs, + ) -> tuple[list[BaseTool], list[Message]]: + return await super()._acting_step( + assistant_message, + tools, + step, + description=self.description, + messages=self.messages, + query=self.query, + history_node=self.history_node, + author=self.author, + **kwargs, + ) async def execute(self): - """Execute and assemble meta_info_dict into output.""" await super().execute() - # Assemble meta_info_dict - if self.meta_info_dict: - output_parts = [f"## {key}\n{value}" for key, value in self.meta_info_dict.items()] - self.output = "\n\n".join(output_parts) - logger.info(f"Assembled output from meta_info_dict with {len(self.meta_info_dict)} entries") + tools: list[BaseTool] = self.response.metadata["tools"] + hands_off_tool = tools[0] + agents: list[BaseMemoryAgent] = hands_off_tool.response.metadata["agents"] + + answer = "" + success = True + messages = [] + tools = [] + for agent in agents: + answer += "\n" + agent.response.answer + success = success and agent.response.metadata["success"] + messages += agent.response.metadata["messages"] + tools += agent.response.metadata["tools"] + + return { + "answer": answer.strip(), + "success": True, + "messages": self.messages, + "tools": tools, + } diff --git a/reme/agent/memory/default/reme_retriever.yaml b/reme/agent/memory/default/reme_retriever.yaml index f6b13250..849f562e 100644 --- a/reme/agent/memory/default/reme_retriever.yaml +++ b/reme/agent/memory/default/reme_retriever.yaml @@ -1,20 +1,23 @@ -tool: | - Retrieve information from specialized memory agents. - system_prompt: | - You orchestrate memory agents to answer user queries. + You are a Memory Orchestrator responsible for routing memory retrieval tasks to specialized agents based on the user query. - # Query - {user_query} + # User Query + {context} - ## Agents + ## Available Memory Agents + Each line indicates a specialized Memory Agent dedicated to storing and retrieving memories within a specific dimension (). Format: "- (): " {meta_memory_info} - ## Task - 1. Use `hands_off` to query agents (memory_type and memory_target must exactly match existing agents) - 2. Answer based on results - 3. If insufficient: "nothing found after thorough search." + ## Your Task + Use the `hands_off` tool to retrieve information from specialized agents: + 1. Analyze the user query and identify which memory dimensions are relevant + 2. Specify `memory_type` and `memory_target` for each retrieval task + - The `memory_type` and `memory_target` must **exactly match** existing entries in the "Available Memory Agents" listed above + - Do NOT query agents that don't exist above + 3. Multiple tasks can be specified to enable parallel retrieval from specialized agents + + Note: If the retrieved information is insufficient to answer the query, respond: "nothing found after thorough search." user_message: | - Retrieve information from agents and answer based on results. + Please analyze the user query and retrieve relevant information from the appropriate existing agents. \ No newline at end of file diff --git a/reme/agent/memory/default/reme_summarizer.py b/reme/agent/memory/default/reme_summarizer.py index 405766b2..e802f908 100644 --- a/reme/agent/memory/default/reme_summarizer.py +++ b/reme/agent/memory/default/reme_summarizer.py @@ -1,4 +1,4 @@ -from loguru import logger +"""ReMe summarizer agent that orchestrates multiple memory agents to summarize information.""" from ..base_memory_agent import BaseMemoryAgent from ....core.enumeration import Role @@ -7,31 +7,21 @@ from ....core.schema import Message class ReMeSummarizer(BaseMemoryAgent): + """Orchestrates multiple memory agents to summarize and store information across different memory types.""" 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: - from ....tool.memory import ReadMetaMemory - - meta_memory_info = ReadMetaMemory().format_memory_metadata(self.meta_memories) - logger.info(f"meta_memory_info={meta_memory_info}") - return meta_memory_info - async def build_messages(self) -> list[Message]: - from ....tool.memory import AddHistory - - add_history_tool = AddHistory() - await add_history_tool.call() - self.context.history_node = add_history_tool.context.add_history + self.context.history_node = await self.read_history_node() messages = [ Message( role=Role.SYSTEM, content=self.prompt_format( prompt_name="system_prompt", - meta_memory_info=await self._read_meta_memories(), + meta_memory_info=await self.read_meta_memories(self.meta_memories), context=self.context.history_node.content, ), ), @@ -46,16 +36,42 @@ class ReMeSummarizer(BaseMemoryAgent): async def _acting_step( self, assistant_message: Message, + tools: list[BaseTool], step: int, stage: str = "", **kwargs, ) -> tuple[list[BaseTool], list[Message]]: return await super()._acting_step( assistant_message, + tools, step, description=self.description, - messages=self.context.messages, - history_node=self.context.history_node, + messages=self.messages, + history_node=self.history_node, author=self.author, **kwargs, ) + + async def execute(self): + await super().execute() + + tools: list[BaseTool] = self.response.metadata["tools"] + hands_off_tool = tools[0] + agents: list[BaseMemoryAgent] = hands_off_tool.response.metadata["agents"] + + answer = "" + success = True + messages = [] + tools = [] + for agent in agents: + answer += "\n" + agent.response.answer + success = success and agent.response.metadata["success"] + messages += agent.response.metadata["messages"] + tools += agent.response.metadata["tools"] + + return { + "answer": answer.strip(), + "success": True, + "messages": self.messages, + "tools": tools, + } diff --git a/reme/core/op/base_react.py b/reme/core/op/base_react.py index 4f3816b8..5b24e30d 100644 --- a/reme/core/op/base_react.py +++ b/reme/core/op/base_react.py @@ -18,7 +18,7 @@ class BaseReact(BaseOp): def __init__( self, - tools: list[BaseTool], + tools: list["BaseTool"], tool_call_interval: float = 0, max_steps: int = 10, **kwargs, @@ -27,12 +27,14 @@ class BaseReact(BaseOp): kwargs["sub_ops"] = tools or [] super().__init__(**kwargs) # Filter only BaseTool instances from sub_ops + from . import BaseTool + self.sub_ops: list[BaseTool] = [t for t in self.sub_ops if isinstance(t, BaseTool)] self.tool_call_interval: float = tool_call_interval self.max_steps: int = max_steps @property - def tools(self) -> list[BaseTool]: + def tools(self) -> list["BaseTool"]: """Return available tools for the agent.""" return self.sub_ops @@ -49,18 +51,21 @@ class BaseReact(BaseOp): async def _reasoning_step( self, messages: list[Message], + tools: list["BaseTool"], step: int, stage: str = "", **kwargs, ) -> tuple[Message, bool]: """Execute one reasoning step where LLM decides whether to use tools.""" # Get tool definitions for LLM - tool_calls = [t.tool_call for t in self.tools] + tool_calls = [t.tool_call for t in tools] + # Generate assistant response with potential tool calls assistant_message: Message = await self.llm.chat(messages=messages, tools=tool_calls, **kwargs) messages.append(assistant_message) assistant_content: str = assistant_message.simple_dump(as_dict=False) logger.info(f"[{self.__class__.__name__} {stage or ''} step{step + 1}] assistant={assistant_content}") + # Determine if tools should be called should_act = bool(assistant_message.tool_calls) return assistant_message, should_act @@ -68,19 +73,20 @@ class BaseReact(BaseOp): async def _acting_step( self, assistant_message: Message, + tools: list["BaseTool"], step: int, stage: str = "", **kwargs, - ) -> tuple[list[BaseTool], list[Message]]: + ) -> tuple[list["BaseTool"], list[Message]]: """Execute tool calls requested by the assistant and collect results.""" - tool_list: list[BaseTool] = [] + tool_list: list["BaseTool"] = [] tool_messages: list[Message] = [] if not assistant_message.tool_calls: return tool_list, tool_messages # Create tool name to tool instance mapping - tool_dict = {t.tool_call.name: t for t in self.tools} + tool_dict = {t.tool_call.name: t for t in tools} for j, tool_call in enumerate(assistant_message.tool_calls): prefix: str = f"[{self.__class__.__name__} {stage or ''} step{step + 1}.{j}]" if tool_call.name not in tool_dict: @@ -116,13 +122,13 @@ class BaseReact(BaseOp): logger.info(f"{prefix} join tool={tool.name} result={tool.response.answer}") return tool_list, tool_messages - async def react(self, messages: list[Message], stage: str = ""): + async def react(self, messages: list[Message], tools: list["BaseTool"], stage: str = ""): """Run ReAct loop alternating between reasoning and acting until completion.""" success: bool = False - tools: list[BaseTool] = [] + used_tools: list[BaseTool] = [] for step in range(self.max_steps): # Reasoning: LLM decides next action - assistant_message, should_act = await self._reasoning_step(messages, step=step, stage=stage) + assistant_message, should_act = await self._reasoning_step(messages, tools, step=step, stage=stage) if not should_act: # No tools requested, task complete @@ -130,17 +136,17 @@ class BaseReact(BaseOp): break # Acting: execute tools and collect results - t_tools, tool_messages = await self._acting_step(assistant_message, step=step, stage=stage) - tools.extend(t_tools) + t_tools, tool_messages = await self._acting_step(assistant_message, tools, step=step, stage=stage) + used_tools.extend(t_tools) messages.extend(tool_messages) - return tools, messages, success + return used_tools, messages, success async def execute(self): """Execute the ReAct agent and return final results.""" # Log available tools for i, tool in enumerate(self.tools): - logger.info(f"[{self.__class__.__name__}] tool_call={tool.tool_call.simple_input_dump(as_dict=False)}") + logger.info(f"[{self.__class__.__name__}] tool_call[{i}]={tool.tool_call.simple_input_dump(as_dict=False)}") # Build and log initial messages messages = await self.build_messages() @@ -149,7 +155,7 @@ class BaseReact(BaseOp): logger.info(f"[{self.__class__.__name__}] role={role} {message.simple_dump(as_dict=False)}") # Run ReAct loop - t_tools, messages, success = await self.react(messages) + t_tools, messages, success = await self.react(messages, self.tools) return { "answer": messages[-1].content if success else "", "success": success, diff --git a/reme/reme.py b/reme/reme.py new file mode 100644 index 00000000..d304b351 --- /dev/null +++ b/reme/reme.py @@ -0,0 +1,247 @@ +"""ReMe application classes for simplified configuration and execution.""" + +import asyncio +import sys + +from .agent.memory.default import ReMeSummarizer, PersonalSummarizer, PersonalRetriever, ReMeRetriever +from .config import ReMeConfigParser +from .core.context import ServiceContext +from .core.embedding import BaseEmbeddingModel +from .core.flow import BaseFlow +from .core.llm import BaseLLM +from .core.schema import Response, Message +from .core.token_counter import BaseTokenCounter +from .core.utils import execute_stream_task +from .core.vector_store import BaseVectorStore +from .tool.memory import UpdateUserProfile, RetrieveMemory, AddMemory, HandsOff, ReadHistory + + +class ReMe: + """ReMe application with config file support and flow execution methods.""" + + def __init__( + self, + *args, + llm_api_key: str | None = None, + llm_api_base: str | None = None, + embedding_api_key: str | None = None, + embedding_api_base: str | None = None, + enable_logo: bool = True, + llm: dict | None = None, + embedding_model: dict | None = None, + vector_store: dict | None = None, + token_counter: dict | None = None, + **kwargs, + ): + self.service_context = ServiceContext( + *args, + llm_api_key=llm_api_key, + llm_api_base=llm_api_base, + embedding_api_key=embedding_api_key, + embedding_api_base=embedding_api_base, + service_config=None, + parser=ReMeConfigParser, + config_path=None, + enable_logo=enable_logo, + llm=llm, + embedding_model=embedding_model, + vector_store=vector_store, + token_counter=token_counter, + **kwargs, + ) + + async def __aenter__(self): + """Async context manager entry.""" + return self + + def __enter__(self): + """Context manager entry.""" + return self + + async def close(self): + """Close the application.""" + return await self.service_context.close() + + def close_sync(self): + """Close the application synchronously.""" + self.service_context.close_sync() + + async def __aexit__(self, exc_type=None, exc_val=None, exc_tb=None): + """Async context manager exit.""" + await self.close() + return False + + def __exit__(self, exc_type=None, exc_val=None, exc_tb=None): + """Context manager exit.""" + self.close_sync() + return False + + @property + def default_llm(self) -> BaseLLM: + """Return the default LLM instance from the service context.""" + return self.service_context.llms["default"] + + @property + def default_embedding_model(self) -> BaseEmbeddingModel: + """Return the default embedding model instance from the service context.""" + return self.service_context.embedding_models["default"] + + @property + def default_vector_store(self) -> BaseVectorStore: + """Return the default vector store instance from the service context.""" + return self.service_context.vector_stores["default"] + + @property + def default_token_counter(self) -> BaseTokenCounter: + """Return the default token counter instance from the service context.""" + return self.service_context.token_counters["default"] + + async def summary( + self, + messages: list[Message | dict], + description: str = "", + user_name: str | list[str] = "", + enable_thinking_params: bool = False, + meta_memories: list[dict] = None, + version: str = "default", + **kwargs, + ): + """Summarize messages and store them in memory for the specified user(s).""" + if user_name: + for message in messages and isinstance(user_name, str): + if isinstance(message, dict) and not message.get("name"): + message["name"] = user_name + elif isinstance(message, Message) and not message.name: + message.name = user_name + + if isinstance(user_name, str): + user_name = [user_name] + + if not meta_memories: + meta_memories = [ + { + "memory_type": "personal", + "memory_target": name, + } + for name in user_name + ] + + if version == "default": + reme_summarizer = ReMeSummarizer( + meta_memories=meta_memories, + tools=[ + HandsOff( + memory_agents=[ + PersonalSummarizer( + tools=[ + RetrieveMemory(enable_thinking_params=enable_thinking_params), + AddMemory(enable_thinking_params=enable_thinking_params), + UpdateUserProfile(enable_thinking_params=enable_thinking_params), + ], + ), + ], + ), + ], + ) + + else: + raise NotImplementedError + + return await reme_summarizer.call(messages=messages, description=description, **kwargs) + + else: + raise NotImplementedError + + async def retrieve( + self, + query: str = "", + top_k: int = 20, + description: str = "", + messages: list[dict] | None = None, + user_name: str | list[str] = "", + enable_thinking_params: bool = False, + meta_memories: list[dict] = None, + version: str = "default", + **kwargs, + ): + """Retrieve relevant memories for the specified user(s) based on query or messages.""" + if user_name: + if messages: + for message in messages and isinstance(user_name, str): + if isinstance(message, dict) and not message.get("name"): + message["name"] = user_name + elif isinstance(message, Message) and not message.name: + message.name = user_name + + if isinstance(user_name, str): + user_name = [user_name] + + if not meta_memories: + meta_memories = [ + { + "memory_type": "personal", + "memory_target": name, + } + for name in user_name + ] + + if version == "default": + + reme_retriever = ReMeRetriever( + meta_memories=meta_memories, + tools=[ + HandsOff( + memory_agents=[ + PersonalRetriever( + tools=[ + RetrieveMemory(enable_thinking_params=enable_thinking_params, top_k=top_k), + ReadHistory(enable_thinking_params=enable_thinking_params), + ], + ), + ], + ), + ], + ) + + else: + raise NotImplementedError + + return await reme_retriever.call(query=query, messages=messages, description=description, **kwargs) + + else: + raise NotImplementedError + + async def execute_flow(self, name: str, **kwargs) -> Response: + """Execute a flow with the given name and parameters.""" + assert name in self.service_context.flows, f"Flow {name} not found" + flow: BaseFlow = self.service_context.flows[name] + return await flow.call(**kwargs) + + async def execute_stream_flow(self, name: str, **kwargs): + """Execute a stream flow with the given name and parameters.""" + assert name in self.service_context.flows, f"Flow {name} not found" + flow: BaseFlow = self.service_context.flows[name] + assert flow.stream is True, "non-stream flow is not supported in execute_stream_flow!" + stream_queue = asyncio.Queue() + task = asyncio.create_task(flow.call(stream_queue=stream_queue, **kwargs)) + async for chunk in execute_stream_task( + stream_queue=stream_queue, + task=task, + task_name=name, + as_bytes=False, + ): + yield chunk + + def run_service(self): + """Run the configured service (HTTP, MCP, or CMD).""" + self.service_context.service.run() + + +def main(): + """Main entry point for running ReMe application from command line.""" + with ReMe(*sys.argv[1:]) as app: + app.run_service() + + +if __name__ == "__main__": + main() diff --git a/reme/reme_app.py b/reme/reme_app.py deleted file mode 100644 index e41aa8a6..00000000 --- a/reme/reme_app.py +++ /dev/null @@ -1,90 +0,0 @@ -"""ReMe application classes for simplified configuration and execution.""" - -import asyncio -import sys - -from .config import ReMeConfigParser -from .core.context import ServiceContext -from .core.flow import BaseFlow -from .core.schema import Response -from .core.utils import execute_stream_task - - -class ReMeApp: - """ReMe application with config file support and flow execution methods.""" - - def __init__( - self, - *args, - llm_api_key: str | None = None, - llm_api_base: str | None = None, - embedding_api_key: str | None = None, - embedding_api_base: str | None = None, - enable_logo: bool = True, - **kwargs, - ): - self.service_context = ServiceContext( - *args, - llm_api_key=llm_api_key, - llm_api_base=llm_api_base, - embedding_api_key=embedding_api_key, - embedding_api_base=embedding_api_base, - service_config=None, - parser=ReMeConfigParser, - config_path=None, - enable_logo=enable_logo, - **kwargs, - ) - - async def __aenter__(self): - """Async context manager entry.""" - return self - - def __enter__(self): - """Context manager entry.""" - return self - - async def __aexit__(self, exc_type=None, exc_val=None, exc_tb=None): - """Async context manager exit.""" - await self.service_context.close() - return False - - def __exit__(self, exc_type=None, exc_val=None, exc_tb=None): - """Context manager exit.""" - self.service_context.close_sync() - return False - - async def execute_flow(self, name: str, **kwargs) -> Response: - """Execute a flow with the given name and parameters.""" - assert name in self.service_context.flows, f"Flow {name} not found" - flow: BaseFlow = self.service_context.flows[name] - return await flow.call(**kwargs) - - async def execute_stream_flow(self, name: str, **kwargs): - """Execute a stream flow with the given name and parameters.""" - assert name in self.service_context.flows, f"Flow {name} not found" - flow: BaseFlow = self.service_context.flows[name] - assert flow.stream is True, "non-stream flow is not supported in execute_stream_flow!" - stream_queue = asyncio.Queue() - task = asyncio.create_task(flow.call(stream_queue=stream_queue, **kwargs)) - async for chunk in execute_stream_task( - stream_queue=stream_queue, - task=task, - task_name=name, - as_bytes=False, - ): - yield chunk - - def run_service(self): - """Run the configured service (HTTP, MCP, or CMD).""" - self.service_context.service.run() - - -def main(): - """Main entry point for running ReMe application from command line.""" - with ReMeApp(*sys.argv[1:]) as app: - app.run_service() - - -if __name__ == "__main__": - main() diff --git a/reme/tool/memory/__init__.py b/reme/tool/memory/__init__.py index c72f4d32..9c8bbd11 100644 --- a/reme/tool/memory/__init__.py +++ b/reme/tool/memory/__init__.py @@ -11,10 +11,10 @@ from .meta.read_meta_memory import ReadMetaMemory from .user_profile.read_user_profile import ReadUserProfile from .user_profile.update_user_profile import UpdateUserProfile from .vector.add_memory import AddMemory -from .vector.update_memory import UpdateMemory -from .vector.retrieve_memory import VectorRetrieveMemory -from .vector.retrieve_recent_memory import VectorRetrieveRecentMemory from .vector.delete_memory import DeleteMemory +from .vector.retrieve_memory import RetrieveMemory +from .vector.retrieve_recent_memory import RetrieveRecentMemory +from .vector.update_memory import UpdateMemory from ...core import R __all__ = [ @@ -29,10 +29,10 @@ __all__ = [ "ReadUserProfile", "UpdateUserProfile", "AddMemory", - "UpdateMemory", - "VectorRetrieveMemory", - "VectorRetrieveRecentMemory", "DeleteMemory", + "RetrieveMemory", + "RetrieveRecentMemory", + "UpdateMemory", ] for name in __all__: diff --git a/reme/tool/memory/hands_off/hands_off.py b/reme/tool/memory/hands_off/hands_off.py index 9ddc6eb2..c2a53872 100644 --- a/reme/tool/memory/hands_off/hands_off.py +++ b/reme/tool/memory/hands_off/hands_off.py @@ -102,4 +102,7 @@ class HandsOff(BaseMemoryTool): results.append(f"{memory_type.value}({memory_target}): {agent.response.answer}") logger.info(f"Completed {len(results)} task(s)") - return "\n".join(results) + return { + "answer": "\n".join(results), + "agents": agent_list, + } diff --git a/reme/tool/memory/user_profile/read_user_profile.py b/reme/tool/memory/user_profile/read_user_profile.py index dfbccd63..05346037 100644 --- a/reme/tool/memory/user_profile/read_user_profile.py +++ b/reme/tool/memory/user_profile/read_user_profile.py @@ -5,8 +5,8 @@ from typing import Literal from loguru import logger from ..base_memory_tool import BaseMemoryTool -from ....core.schema import ToolCall -from ....core.schema.memory_node import MemoryNode +from ....core.enumeration import MemoryType +from ....core.schema import ToolCall, MemoryNode class ReadUserProfile(BaseMemoryTool): @@ -30,6 +30,7 @@ class ReadUserProfile(BaseMemoryTool): ) async def execute(self): + self.context.memory_type = MemoryType.PERSONAL cached_data = self.local_memory.load(self.memory_cache_key, auto_clean=False) if not cached_data: diff --git a/reme/tool/memory/vector/retrieve_memory.py b/reme/tool/memory/vector/retrieve_memory.py index 94c8f697..de20a1df 100644 --- a/reme/tool/memory/vector/retrieve_memory.py +++ b/reme/tool/memory/vector/retrieve_memory.py @@ -3,12 +3,11 @@ from loguru import logger from ..base_memory_tool import BaseMemoryTool -from ....core.enumeration import MemoryType from ....core.schema import ToolCall, MemoryNode, VectorNode from ....core.utils import deduplicate_memories -class VectorRetrieveMemory(BaseMemoryTool): +class RetrieveMemory(BaseMemoryTool): """Tool to retrieve memories from vector store using similarity search""" def __init__(self, top_k: int = 20, **kwargs): diff --git a/reme/tool/memory/vector/retrieve_recent_memory.py b/reme/tool/memory/vector/retrieve_recent_memory.py index 796f04dd..391fc72c 100644 --- a/reme/tool/memory/vector/retrieve_recent_memory.py +++ b/reme/tool/memory/vector/retrieve_recent_memory.py @@ -7,7 +7,7 @@ from ....core.schema import ToolCall, MemoryNode, VectorNode from ....core.utils import deduplicate_memories -class VectorRetrieveRecentMemory(BaseMemoryTool): +class RetrieveRecentMemory(BaseMemoryTool): """Tool to retrieve most recent memories sorted by conversation time""" def __init__(self, top_k: int = 20, **kwargs): diff --git a/reme_ai/reme.py b/reme_ai/reme.py deleted file mode 100644 index 6ec678c9..00000000 --- a/reme_ai/reme.py +++ /dev/null @@ -1,502 +0,0 @@ -"""ReMe classes for simplified configuration and execution.""" - -from .core_old.application import Application -from .core_old.config import ReMeConfigParser -from .core_old.context import C -from .core_old.embedding import BaseEmbeddingModel -from .core_old.enumeration import Role -from .core_old.llm import BaseLLM -from .core_old.schema import Message -from .core_old.utils import singleton -from .core_old.vector_store import BaseVectorStore -from .mem_agent.retriever import ReMeRetriever -from .mem_agent.retriever_v2 import ReMeRetrieverV2 -from .mem_agent.summarizer import ReMeSummarizer, PersonalSummarizer -from .mem_agent.summarizer_v2 import ReMeSummarizerV2, PersonalSummarizerV2 -from .mem_agent.v3 import ( - PersonalSummarizerV3, - ReMeRetrieverV3, - ReMeSummarizerV3, -) -from .mem_agent.v4 import ( - PersonalSummarizerV4, - PersonalRetrieverV4, - ReMeRetrieverV4, - ReMeSummarizerV4, -) -from .mem_tool import ( - HandsOffTool, - ReadHistoryMemory, - AddMemory, - AddSummaryMemory, - DeleteMemory, - UpdateMemory, - VectorRetrieveMemory, -) -from .mem_tool.v2 import ( - AddMemoryDrafts, - RetrieveMemories, - RetrieveRecentAndSimilarMemories, - SummaryAndHandsOff, - UpdateMemories, -) -from .mem_tool.v3 import ( - AddMemory as AddMemoryV3, - ReadHistory as ReadHistoryV3, - ReadUserProfile, - RetrieveMemory, - SummaryAndHandsOff as SummaryAndHandsOffV3, - UpdateUserProfile, -) -from .mem_tool.v4 import ( - AddSummaryMemory as AddSummaryMemoryV4, - HandsOff as HandsOffV4, - ReadHistory as ReadHistoryV4, - ReadUserProfile as ReadUserProfileV4, - RetrieveMemory as RetrieveMemoryV4, - UpdateUserProfile as UpdateUserProfileV4, -) - - -class ReMe(Application): - """Simplified ReMe application that auto-initializes the service context.""" - - def __init__( - self, - *args, - llm_api_key: str | None = None, - llm_api_base: str | None = None, - embedding_api_key: str | None = None, - embedding_api_base: str | None = None, - enable_logo: bool = True, - llm: dict | None = None, - embedding_model: dict | None = None, - vector_store: dict | None = None, - token_counter: dict | None = None, - **kwargs, - ): - super().__init__( - *args, - llm_api_key=llm_api_key, - llm_api_base=llm_api_base, - embedding_api_key=embedding_api_key, - embedding_api_base=embedding_api_base, - service_config=None, - parser=ReMeConfigParser, - config_path=None, - enable_logo=enable_logo, - llm=llm, - embedding_model=embedding_model, - vector_store=vector_store, - token_counter=token_counter, - **kwargs, - ) - - C.initialize_service_context() - - self.llm: BaseLLM = C.get_llm("default") - self.vector_store: BaseVectorStore = C.get_vector_store("default") - self.embedding_model: BaseEmbeddingModel = C.get_embedding_model("default") - - @staticmethod - def get_llm(name: str) -> BaseLLM: - return C.get_llm(name) - - @staticmethod - def _prepare_messages(messages: list[dict | Message], user_id: str, assistant_id: str): - if not messages: - return [] - - messages = [Message(**m) if isinstance(m, dict) else m for m in messages] - for message in messages: - if message.role is Role.USER and user_id: - message.name = user_id - elif message.role is Role.ASSISTANT and assistant_id: - message.name = assistant_id - return messages - - async def summary( - self, - messages: list[dict], - description: str = "", - user_id: str = "", - assistant_id: str = "", - **kwargs, - ): - """Summarizes messages and stores them as memory based on the specified memory mode.""" - - if user_id: - # halumem: user_id -> message.name - # locomo: add description - metadata_summary = { - "year": "The `year` information associated with the memory(Optional)", - "month": "The `month` information associated with the memory(Optional)", - "day": "The `day` information associated with the memory(Optional)", - # "hour": "The `hour` information associated with the memory(Optional)", - # "year": "The year when the memory content occurred(Optional)", - # "month": "The month when the memory content occurred(Optional)", - # "day": "The day when the memory content occurred(Optional)", - # "hour": "The hour when the memory content occurred(Optional)", - } - meta_memories = [ - { - "memory_type": "personal", - "memory_target": user_id, - }, - ] - - messages = self._prepare_messages(messages, user_id, assistant_id) - - personal_summarizer = PersonalSummarizer( - tools=[ - VectorRetrieveMemory( - add_memory_type_target=False, - metadata_desc=None, - top_k=15, - ), - AddMemory(add_when_to_use=False, metadata_desc=metadata_summary), - DeleteMemory(), - UpdateMemory(add_when_to_use=False, metadata_desc=metadata_summary), - ], - ) - - reme_summarizer = ReMeSummarizer( - meta_memories=meta_memories, - enable_identity_memory=False, - tools=[ - # AddMetaMemory(), - AddSummaryMemory(metadata_desc=metadata_summary), - HandsOffTool(memory_agents=[personal_summarizer]), - ], - ) - - try: - await reme_summarizer.call(messages=messages, description=description, **kwargs) - return reme_summarizer.memory_nodes - except Exception as e: - print(f"Warning: reme_summarizer.call failed: {e}") - return [] - - else: - raise NotImplementedError - - async def retrieve( - self, - query: str = "", - messages: list[dict] | None = None, - description: str = "", - user_id: str = "", - assistant_id: str = "", - top_k: int = 20, - **kwargs, - ): - """Retrieves relevant memories based on the query and specified memory mode.""" - - if user_id: - messages = self._prepare_messages(messages, user_id, assistant_id) - - metadata_retrieve = { - "year": "The year to filter memories(Optional)", - "month": "The month to filter memories(Optional)", - "day": "The day to filter memories(Optional)", - # "hour": "The hour to filter memories(Optional)", - } - meta_memories = [ - { - "memory_type": "personal", - "memory_target": user_id, - }, - ] - - reme_retriever = ReMeRetriever( - meta_memories=meta_memories, - tools=[ - VectorRetrieveMemory( - add_memory_type_target=True, - metadata_desc=metadata_retrieve, - top_k=top_k, - ), - ReadHistoryMemory(), - ], - ) - - try: - await reme_retriever.call(query=query, messages=messages, description=description, **kwargs) - return reme_retriever.output - except Exception as e: - print(f"Warning: reme_retriever.call failed: {e}") - return "error, not retrieved" - - else: - raise NotImplementedError - - async def summary_v2( - self, - messages: list[dict], - description: str = "", - user_id: str = "", - assistant_id: str = "", - **kwargs, - ): - """Summarizes messages using V2 workflow with simplified tools.""" - - if user_id: - metadata_desc = { - "year": "The year when the message content occurred.", - "month": "The month when the message content occurred.", - "day": "The day when the message content occurred.", - } - meta_memories = [ - { - "memory_type": "personal", - "memory_target": user_id, - }, - ] - - messages = self._prepare_messages(messages, user_id, assistant_id) - - personal_summarizer_v2 = PersonalSummarizerV2( - tools=[ - AddMemoryDrafts(enable_thinking_params=True, metadata_desc=metadata_desc), - RetrieveRecentAndSimilarMemories( - enable_thinking_params=True, - metadata_desc=None, - recent_top_k=20, - similar_top_k=20, - ), - UpdateMemories(enable_thinking_params=True, metadata_desc=metadata_desc), - ], - ) - - reme_summarizer_v2 = ReMeSummarizerV2( - meta_memories=meta_memories, - tools=[ - SummaryAndHandsOff( - enable_thinking_params=True, - metadata_desc=metadata_desc, - memory_agents=[personal_summarizer_v2], - ), - ], - ) - - # try: - await reme_summarizer_v2.call(messages=messages, description=description, **kwargs) - return reme_summarizer_v2.memory_nodes, reme_summarizer_v2.messages, reme_summarizer_v2.success - # except Exception as e: - # print(f"Warning: reme_summarizer_v2.call failed: {e}") - # return [], [], False - - else: - raise NotImplementedError - - async def retrieve_v2( - self, - query: str = "", - messages: list[dict] | None = None, - description: str = "", - user_id: str = "", - assistant_id: str = "", - top_k: int = 20, - **kwargs, - ): - """Retrieves relevant memories using V2 workflow with autonomous retrieval.""" - - if user_id: - messages = self._prepare_messages(messages, user_id, assistant_id) - - metadata_retrieve = { - "year": "The year to filter memories.", - "month": "The month to filter memories.", - "day": "The day to filter memories.", - } - meta_memories = [ - { - "memory_type": "personal", - "memory_target": user_id, - }, - ] - - reme_retriever_v2 = ReMeRetrieverV2( - meta_memories=meta_memories, - tools=[ - RetrieveMemories( - enable_thinking_params=True, - metadata_desc=metadata_retrieve, - top_k=top_k, - ), - # ReadHistory(enable_thinking_params=True), - ], - ) - - # try: - await reme_retriever_v2.call(query=query, messages=messages, description=description, **kwargs) - return reme_retriever_v2.output, reme_retriever_v2.messages, reme_retriever_v2.success - # except Exception as e: - # print(f"Warning: reme_retriever_v2.call failed: {e}") - # return "error, not retrieved", [], False - - else: - raise NotImplementedError - - async def summary_v3( - self, - messages: list[dict], - description: str = "", - user_id: str = "", - assistant_id: str = "", - **kwargs, - ): - """Summarizes messages using V3 workflow with user profile management.""" - - if user_id: - meta_memories = [ - { - "memory_type": "personal", - "memory_target": user_id, - }, - ] - messages = self._prepare_messages(messages, user_id, assistant_id) - - personal_summarizer_v3 = PersonalSummarizerV3( - tools=[ - AddMemoryV3(enable_thinking_params=True), - ReadUserProfile(enable_thinking_params=True, add_memory_type_target=False), - UpdateUserProfile(enable_thinking_params=True), - ], - ) - - reme_summarizer_v3 = ReMeSummarizerV3( - meta_memories=meta_memories, - tools=[SummaryAndHandsOffV3(memory_agents=[personal_summarizer_v3])], - ) - - # try: - await reme_summarizer_v3.call(messages=messages, description=description, **kwargs) - return reme_summarizer_v3.memory_nodes, reme_summarizer_v3.messages, reme_summarizer_v3.success - # except Exception as e: - # print(f"Warning: reme_summarizer_v3.call failed: {e}") - # return [], [], False - - else: - raise NotImplementedError - - async def retrieve_v3( - self, - query: str = "", - messages: list[dict] | None = None, - description: str = "", - user_id: str = "", - assistant_id: str = "", - top_k: int = 20, - **kwargs, - ): - """Retrieves relevant memories using V3 workflow with user profile support.""" - - if user_id: - messages = self._prepare_messages(messages, user_id, assistant_id) - - meta_memories = [ - { - "memory_type": "personal", - "memory_target": user_id, - }, - ] - - reme_retriever_v3 = ReMeRetrieverV3( - meta_memories=meta_memories, - tools=[ - ReadUserProfile(enable_thinking_params=True, add_memory_type_target=True), - RetrieveMemory(enable_thinking_params=True, top_k=top_k), - ReadHistoryV3(enable_thinking_params=True), - ], - ) - - # try: - await reme_retriever_v3.call(query=query, messages=messages, description=description, **kwargs) - return reme_retriever_v3.output, reme_retriever_v3.messages, reme_retriever_v3.success - # except Exception as e: - # print(f"Warning: reme_retriever_v3.call failed: {e}") - # return "error, not retrieved", [], False - - else: - raise NotImplementedError - - async def summary_v4( - self, - messages: list[dict], - description: str = "", - user_id: str = "", - assistant_id: str = "", - enable_thinking_params: bool = False, - **kwargs, - ): - """Summarizes messages using V4 workflow with simplified memory management.""" - - if user_id: - meta_memories = [ - { - "memory_type": "personal", - "memory_target": user_id, - }, - ] - messages = self._prepare_messages(messages, user_id, assistant_id) - - personal_summarizer_v4 = PersonalSummarizerV4( - tools=[ - AddSummaryMemoryV4(enable_thinking_params=enable_thinking_params), - ReadUserProfileV4(enable_thinking_params=enable_thinking_params), - UpdateUserProfileV4(enable_thinking_params=enable_thinking_params), - ], - ) - - reme_summarizer_v4 = ReMeSummarizerV4( - meta_memories=meta_memories, - tools=[HandsOffV4(memory_agents=[personal_summarizer_v4])], - ) - - await reme_summarizer_v4.call(messages=messages, description=description, **kwargs) - return reme_summarizer_v4.memory_nodes, reme_summarizer_v4.tool_messages, reme_summarizer_v4.success - - else: - raise NotImplementedError - - async def retrieve_v4( - self, - query: str = "", - messages: list[dict] | None = None, - description: str = "", - user_id: str = "", - assistant_id: str = "", - top_k: int = 20, - enable_thinking_params: bool = False, - **kwargs, - ): - """Retrieves relevant memories using V4 workflow with enhanced retrieval.""" - - if user_id: - messages = self._prepare_messages(messages, user_id, assistant_id) - - meta_memories = [ - { - "memory_type": "personal", - "memory_target": user_id, - }, - ] - - personal_retriever_v4 = PersonalRetrieverV4( - tools=[ - RetrieveMemoryV4(enable_thinking_params=enable_thinking_params, top_k=top_k), - ReadHistoryV4(enable_thinking_params=enable_thinking_params), - ], - ) - - reme_retriever_v4 = ReMeRetrieverV4( - meta_memories=meta_memories, - tools=[HandsOffV4(memory_agents=[personal_retriever_v4])], - ) - - await reme_retriever_v4.call(query=query, messages=messages, description=description, **kwargs) - return reme_retriever_v4.output, reme_retriever_v4.tool_messages, reme_retriever_v4.success - - else: - raise NotImplementedError diff --git a/test/test_reme.py b/tests/test_reme.py similarity index 87% rename from test/test_reme.py rename to tests/test_reme.py index 3279da2a..f5929122 100644 --- a/test/test_reme.py +++ b/tests/test_reme.py @@ -2,9 +2,8 @@ import asyncio -from reme.reme import ReMe - from reme.core.schema import VectorNode, MemoryNode +from reme.reme import ReMe reme = ReMe( vector_store={"collection_name": "reme"}, @@ -14,7 +13,7 @@ reme = ReMe( async def test_reme(): """Tests ReMe memory system with personal information storage and retrieval.""" # 构建一段包含个人信息的对话 - await reme.vector_store.delete_all() + await reme.default_vector_store.delete_all() messages = [ { @@ -56,12 +55,10 @@ async def test_reme(): print("=" * 60) # 对对话进行总结，生成记忆 - # await reme.summary( - await reme.summary_v2( + await reme.summary( messages=messages, - user_id="zhangwei", + user_name="zhangwei", description="用户自我介绍和技术兴趣分享", - ref_memory_id="ref_123", ) print("\n✓ 记忆总结完成") @@ -71,7 +68,7 @@ async def test_reme(): print("=" * 60) # 列出所有存储的记忆节点 - nodes: list[VectorNode] = await reme.vector_store.list() + nodes: list[VectorNode] = await reme.default_vector_store.list() for i, node in enumerate(nodes, 1): memory_node = MemoryNode.from_vector_node(node) print(f"{i} {memory_node.memory_type} {memory_node.memory_target} {memory_node.format_memory()}") @@ -83,25 +80,25 @@ async def test_reme(): # 测试问题1: 检索用户姓名 query1 = "用户叫什么名字？" print(f"\n问题1: {query1}") - result1 = await reme.retrieve_v2(query=query1, user_id="zhangwei") + result1 = await reme.retrieve(query=query1, user_name="zhangwei") print(f"检索结果:\n{result1}") # 测试问题2: 检索技术背景 query2 = "用户擅长什么编程语言和技术方向？" print(f"\n问题2: {query2}") - result2 = await reme.retrieve_v2(query=query2, user_id="zhangwei") + result2 = await reme.retrieve(query=query2, user_name="zhangwei") print(f"检索结果:\n{result2}") # 测试问题3: 检索个人信息 query3 = "用户的工作地点和联系方式是什么？" print(f"\n问题3: {query3}") - result3 = await reme.retrieve_v2(query=query3, user_id="zhangwei") + result3 = await reme.retrieve(query=query3, user_name="zhangwei") print(f"检索结果:\n{result3}") # 测试问题4: 检索兴趣爱好 query4 = "用户平时有什么爱好或活动？" print(f"\n问题4: {query4}") - result4 = await reme.retrieve_v2(query=query4, user_id="zhangwei") + result4 = await reme.retrieve(query=query4, user_name="zhangwei") print(f"检索结果:\n{result4}") print("\n" + "=" * 60) diff --git a/tests/test_tool.py b/tests/test_tool.py index 8d1fb232..9e9ccaff 100644 --- a/tests/test_tool.py +++ b/tests/test_tool.py @@ -8,9 +8,9 @@ search tools (Dashscope, Mock, Tavily) and execution tools (Code, Shell). import asyncio -from reme.reme_app import ReMeApp +from reme import ReMe -app = ReMeApp() +app = ReMe() def test_search(): @@ -21,7 +21,7 @@ def test_search(): """ from reme.tool.search import DashscopeSearch, MockSearch, TavilySearch - query = "今天杭州的天气如何？" + query = "美股DFDV是做什么的？" for op in [ DashscopeSearch(), @@ -190,7 +190,7 @@ async def test_stream_chat(): if __name__ == "__main__": - # test_search() + test_search() # test_execute() - test_simple_chat() - asyncio.run(test_stream_chat()) + # test_simple_chat() + # asyncio.run(test_stream_chat())