scriptoriumadmin/ReMe
1
0
Fork 0
mirror of https://github.com/agentscope-ai/ReMe.git synced 2026-08-28 05:25:04 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 11

refactor(agent): restructure memory agents and base react implementation

Browse source
This commit is contained in:
jinli.yl 2026-01-26 16:30:58 +08:00
parent 035703029f
commit d3fa645832
21 changed files with 592 additions and 884 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
reme/__init__.py
View file

@ -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"

31
reme/agent/memory/base_memory_agent.py
View file

@ -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"""

2
reme/agent/memory/default/__init__.py
View file

@ -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

58
reme/agent/memory/default/personal_retriever.py
View file

@ -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 "<MEMORY_FOUND>" in self.output:
self.success = True
elif "<MEMORY_NOT_FOUND>" in self.output:
self.success = False
self.meta_info = self.context.user_profile + "\n" + self.meta_info

52
reme/agent/memory/default/personal_retriever.yaml
View file

@ -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: `<MEMORY_FOUND>` if found, `<MEMORY_NOT_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.

132
reme/agent/memory/default/personal_summarizer.py
View file

@ -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,
}

57
reme/agent/memory/default/personal_summarizer.yaml
View file

@ -1,26 +1,45 @@
system_prompt_phase1: |
You are a memory agent managing **{memory_type}** memories about **{memory_target}**.
## Latest Conversation:
Message format: `round<index> [<timestamp>] <role/name>: <content>` (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<index> [<timestamp>] <role/name>: <content>` (timestamp: YYYY-MM-DD HH:MM:SS).
{context}
Format: `round<index> [<timestamp>] <role/name>: <content>` (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=<id> conversation_time=<timestamp> <content>`.
{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.

132
reme/agent/memory/default/reme_retriever.py
View file

@ -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,
}

27
reme/agent/memory/default/reme_retriever.yaml
View file

@ -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 <memory_type>(<memory_target>).
Format: "- <memory_type>(<memory_target>): <description>"
{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.

48
reme/agent/memory/default/reme_summarizer.py
View file

@ -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,
}

34
reme/core/op/base_react.py
View file

@ -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,

247
reme/reme.py Normal file
View file

@ -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()

90
reme/reme_app.py
View file

@ -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()

12
reme/tool/memory/__init__.py
View file

@ -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__:

5
reme/tool/memory/hands_off/hands_off.py
View file

@ -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,
}

5
reme/tool/memory/user_profile/read_user_profile.py
View file

@ -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:

3
reme/tool/memory/vector/retrieve_memory.py
View file

@ -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):

2
reme/tool/memory/vector/retrieve_recent_memory.py
View file

@ -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):

502
reme_ai/reme.py
View file

@ -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

21
test/test_reme.py → tests/test_reme.py
View file

@ -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)

12
tests/test_tool.py
View file

@ -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())