refactor(core): restructure module imports and enhance embedding functionality

reme_ai/core/config/default.yaml

@@ -32,6 +32,7 @@ vector_store:
   default:
     backend: local
     embedding_model: default
+    collection_name: reme
 
 token_counter:
   default:

reme_ai/core/embedding/base_embedding_model.py

@@ -36,17 +36,17 @@ class BaseEmbeddingModel(ABC):
         self.raise_exception = raise_exception
         self.kwargs = kwargs
 
-    async def _get_embeddings(self, input_text: list[str]) -> list[list[float]]:
+    async def _get_embeddings(self, input_text: list[str], **kwargs) -> list[list[float]]:
         """Internal async implementation for calling the embedding API with batch input."""
 
-    def _get_embeddings_sync(self, input_text: list[str]) -> list[list[float]]:
+    def _get_embeddings_sync(self, input_text: list[str], **kwargs) -> list[list[float]]:
         """Internal synchronous implementation for calling the embedding API with batch input."""
 
-    async def get_embedding(self, input_text: str) -> list[float]:
+    async def get_embedding(self, input_text: str, **kwargs) -> list[float]:
         """Async get embedding for a single text with exponential backoff retries."""
         for i in range(self.max_retries):
             try:
-                result = await self._get_embeddings([input_text])
+                result = await self._get_embeddings([input_text], **kwargs)
                 return result[0]
             except Exception as e:
                 logger.error(f"Model {self.model_name} failed: {e}")
@@ -57,7 +57,7 @@ class BaseEmbeddingModel(ABC):
                 await asyncio.sleep(i + 1)
         return []
 
-    async def get_embeddings(self, input_text: list[str]) -> list[list[float]]:
+    async def get_embeddings(self, input_text: list[str], **kwargs) -> list[list[float]]:
         """Async get embeddings with automatic batching and exponential backoff retries."""
         # Split into batches and process sequentially to respect rate limits
         results = []
@@ -66,7 +66,7 @@ class BaseEmbeddingModel(ABC):
             # Process each batch with retry logic
             for retry in range(self.max_retries):
                 try:
-                    batch_res = await self._get_embeddings(batch)
+                    batch_res = await self._get_embeddings(batch, **kwargs)
                     if batch_res:
                         results.extend(batch_res)
                     break
@@ -79,11 +79,11 @@ class BaseEmbeddingModel(ABC):
                         await asyncio.sleep(retry + 1)
         return results
 
-    def get_embedding_sync(self, input_text: str) -> list[float]:
+    def get_embedding_sync(self, input_text: str, **kwargs) -> list[float]:
         """Synchronous get embedding for a single text with retry logic."""
         for i in range(self.max_retries):
             try:
-                result = self._get_embeddings_sync([input_text])
+                result = self._get_embeddings_sync([input_text], **kwargs)
                 return result[0]
             except Exception as exc:
                 logger.error(f"Model {self.model_name} failed: {exc}")
@@ -94,7 +94,7 @@ class BaseEmbeddingModel(ABC):
                 time.sleep(i + 1)
         return []
 
-    def get_embeddings_sync(self, input_text: list[str]) -> list[list[float]]:
+    def get_embeddings_sync(self, input_text: list[str], **kwargs) -> list[list[float]]:
         """Synchronous get embeddings with automatic batching and retry logic."""
         results = []
         for i in range(0, len(input_text), self.max_batch_size):
@@ -102,7 +102,7 @@ class BaseEmbeddingModel(ABC):
             # Process each batch with retry logic
             for retry in range(self.max_retries):
                 try:
-                    batch_res = self._get_embeddings_sync(batch)
+                    batch_res = self._get_embeddings_sync(batch, **kwargs)
                     if batch_res:
                         results.extend(batch_res)
                     break
@@ -115,15 +115,15 @@ class BaseEmbeddingModel(ABC):
                         time.sleep(retry + 1)
         return results
 
-    async def get_node_embedding(self, node: VectorNode) -> VectorNode:
+    async def get_node_embedding(self, node: VectorNode, **kwargs) -> VectorNode:
         """Async generate and populate vector field for a single VectorNode object."""
-        node.vector = await self.get_embedding(node.content)
+        node.vector = await self.get_embedding(node.content, **kwargs)
         return node
 
-    async def get_node_embeddings(self, nodes: list[VectorNode]) -> list[VectorNode]:
+    async def get_node_embeddings(self, nodes: list[VectorNode], **kwargs) -> list[VectorNode]:
         """Async generate and populate vector fields for a batch of VectorNode objects."""
         contents = [node.content for node in nodes]
-        embeddings: list[list[float]] = await self.get_embeddings(contents)
+        embeddings: list[list[float]] = await self.get_embeddings(contents, **kwargs)
 
         if len(embeddings) == len(nodes):
             for node, vec in zip(nodes, embeddings):
@@ -132,15 +132,15 @@ class BaseEmbeddingModel(ABC):
             logger.warning(f"Mismatch: got {len(embeddings)} vectors for {len(nodes)} nodes")
         return nodes
 
-    def get_node_embedding_sync(self, node: VectorNode) -> VectorNode:
+    def get_node_embedding_sync(self, node: VectorNode, **kwargs) -> VectorNode:
         """Synchronously generate and populate vector field for a single VectorNode object."""
-        node.vector = self.get_embedding_sync(node.content)
+        node.vector = self.get_embedding_sync(node.content, **kwargs)
         return node
 
-    def get_node_embeddings_sync(self, nodes: list[VectorNode]) -> list[VectorNode]:
+    def get_node_embeddings_sync(self, nodes: list[VectorNode], **kwargs) -> list[VectorNode]:
         """Synchronously generate and populate vector fields for a batch of VectorNode objects."""
         contents = [node.content for node in nodes]
-        embeddings: list[list[float]] = self.get_embeddings_sync(contents)
+        embeddings: list[list[float]] = self.get_embeddings_sync(contents, **kwargs)
 
         if len(embeddings) == len(nodes):
             for node, vec in zip(nodes, embeddings):

reme_ai/core/embedding/openai_embedding_model.py

@@ -33,13 +33,15 @@ class OpenAIEmbeddingModel(BaseEmbeddingModel):
         """Create and return an internal AsyncOpenAI client instance."""
         return AsyncOpenAI(api_key=self.api_key, base_url=self.base_url)
 
-    async def _get_embeddings(self, input_text: list[str]) -> list[list[float]]:
+    async def _get_embeddings(self, input_text: list[str], **kwargs) -> list[list[float]]:
         """Fetch embeddings from the API for a batch of strings."""
         completion = await self._client.embeddings.create(
             model=self.model_name,
             input=input_text,
             dimensions=self.dimensions,
             encoding_format=self.encoding_format,
+            **self.kwargs,
+            **kwargs,
         )
 
         result_emb = [[] for _ in range(len(input_text))]

reme_ai/core/embedding/openai_embedding_model_sync.py

@@ -14,13 +14,15 @@ class OpenAIEmbeddingModelSync(OpenAIEmbeddingModel):
         """Create and return an internal synchronous OpenAI client instance."""
         return OpenAI(api_key=self.api_key, base_url=self.base_url)
 
-    def _get_embeddings_sync(self, input_text: list[str]) -> list[list[float]]:
+    def _get_embeddings_sync(self, input_text: list[str], **kwargs) -> list[list[float]]:
         """Fetch embeddings synchronously from the API for a batch of strings."""
         completion = self._client.embeddings.create(
             model=self.model_name,
             input=input_text,
             dimensions=self.dimensions,
             encoding_format=self.encoding_format,
+            **self.kwargs,
+            **kwargs,
         )
 
         result_emb = [[] for _ in range(len(input_text))]

reme_ai/core/flow/__init__.py

@@ -3,9 +3,11 @@
 from .base_flow import BaseFlow
 from .cmd_flow import CmdFlow
 from .expression_flow import ExpressionFlow
+from .simple_flow import SimpleFlow
 
 __all__ = [
     "BaseFlow",
     "CmdFlow",
     "ExpressionFlow",
+    "SimpleFlow",
 ]

reme_ai/core/flow/base_flow.py

@@ -25,6 +25,7 @@ class BaseFlow(ABC):
     def __init__(
         self,
         name: str = "",
+        flow_op: BaseOp | None = None,
         stream: bool = False,
         raise_exception: bool = True,
         enable_cache: bool = False,
@@ -43,7 +44,7 @@ class BaseFlow(ABC):
         self.cache_expire_hours: float = cache_expire_hours
         self.flow_params: dict = kwargs
 
-        self._flow_op: BaseOp | None = None
+        self._flow_op: BaseOp | None = flow_op
         self._cache: CacheHandler | None = None
         self._flow_printed: bool = False
         self._tool_call: ToolCall | None = None
@@ -129,6 +130,12 @@ class BaseFlow(ABC):
             self._flow_op = self._build_flow()
         return self._flow_op
 
+    @flow_op.setter
+    def flow_op(self, op: BaseOp):
+        """Set the root operation of the flow."""
+        self._flow_op = op
+        self._flow_printed = False
+
     @property
     def async_mode(self) -> bool:
         """Check if the current flow operation tree is asynchronous."""

reme_ai/core/flow/simple_flow.py

@@ -0,0 +1,17 @@
+"""Simple flow implementation that directly uses a predefined flow operation."""
+
+from .base_flow import BaseFlow
+from ..op import BaseOp
+from ..schema import ToolCall
+
+
+class SimpleFlow(BaseFlow):
+    """Simple flow that directly uses a predefined flow operation."""
+
+    def _build_flow(self) -> BaseOp:
+        assert self._flow_op is not None
+        return self._flow_op.copy()
+
+    def _build_tool_call(self) -> ToolCall:
+        assert self._flow_op is not None
+        return self._flow_op.tool_call

reme_ai/core/llm/base_llm.py

@@ -24,53 +24,6 @@ class BaseLLM(ABC):
         self.raise_exception: bool = raise_exception
         self.kwargs: dict = kwargs
 
-    @staticmethod
-    def _process_stream_chunk(
-        stream_chunk: StreamChunk,
-        state: dict,
-        enable_stream_print: bool = False,
-    ) -> None:
-        """Update the aggregation state by processing an individual stream chunk."""
-        if stream_chunk.chunk_type is ChunkEnum.USAGE:
-            if enable_stream_print:
-                print(f"\n<usage>{json.dumps(stream_chunk.chunk, ensure_ascii=False, indent=2)}</usage>", flush=True)
-
-        elif stream_chunk.chunk_type is ChunkEnum.THINK:
-            if enable_stream_print:
-                if not state["enter_think"]:
-                    state["enter_think"] = True
-                    print("<think>\n", end="", flush=True)
-                print(stream_chunk.chunk, end="", flush=True)
-            state["reasoning_content"] += stream_chunk.chunk
-
-        elif stream_chunk.chunk_type is ChunkEnum.ANSWER:
-            if enable_stream_print:
-                if not state["enter_answer"]:
-                    state["enter_answer"] = True
-                    if state["enter_think"]:
-                        print("\n</think>", flush=True)
-                print(stream_chunk.chunk, end="", flush=True)
-            state["answer_content"] += stream_chunk.chunk
-
-        elif stream_chunk.chunk_type is ChunkEnum.TOOL:
-            if enable_stream_print:
-                print(f"\n<tool>{json.dumps(stream_chunk.chunk, ensure_ascii=False, indent=2)}</tool>", flush=True)
-            state["tool_calls"].append(stream_chunk.chunk)
-
-        elif stream_chunk.chunk_type is ChunkEnum.ERROR:
-            if enable_stream_print:
-                print(f"\n<error>{stream_chunk.chunk}</error>", flush=True)
-
-    @staticmethod
-    def _create_message_from_state(state: dict) -> Message:
-        """Construct a Message object from the accumulated aggregation state."""
-        return Message(
-            role=Role.ASSISTANT,
-            reasoning_content=state["reasoning_content"],
-            content=state["answer_content"],
-            tool_calls=state["tool_calls"],
-        )
-
     @staticmethod
     def _accumulate_tool_call_chunk(tool_call, ret_tools: list[ToolCall]):
         """Assemble incremental tool call fragments into complete ToolCall objects."""
@@ -137,14 +90,15 @@ class BaseLLM(ABC):
         """Internal synchronous generator for streaming raw response chunks."""
         raise NotImplementedError
 
-    async def _stream_with_retry(
+    async def stream_chat(
         self,
-        operation_name: str,
         messages: list[Message],
-        tools: list[ToolCall] | None,
-        stream_kwargs: dict,
+        tools: list[ToolCall] | None = None,
+        **kwargs,
     ) -> AsyncGenerator[StreamChunk, None]:
-        """Execute the async streaming operation with retry logic and error recovery."""
+        """Public async interface for streaming chat completions with retries."""
+        stream_kwargs = self._build_stream_kwargs(messages, tools, **kwargs)
+
         for i in range(self.max_retries):
             try:
                 async for chunk in self._stream_chat(messages=messages, tools=tools, stream_kwargs=stream_kwargs):
@@ -152,7 +106,7 @@ class BaseLLM(ABC):
                 return
 
             except Exception as e:
-                logger.exception(f"{operation_name} with model={self.model_name} encounter error with e={e.args}")
+                logger.exception(f"stream chat with model={self.model_name} encounter error with e={e.args}")
 
                 if i == self.max_retries - 1:
                     if self.raise_exception:
@@ -163,21 +117,22 @@ class BaseLLM(ABC):
                 yield StreamChunk(chunk_type=ChunkEnum.ERROR, chunk=str(e))
                 await asyncio.sleep(i + 1)
 
-    def _stream_with_retry_sync(
+    def stream_chat_sync(
         self,
-        operation_name: str,
         messages: list[Message],
-        tools: list[ToolCall] | None,
-        stream_kwargs: dict,
+        tools: list[ToolCall] | None = None,
+        **kwargs,
     ) -> Generator[StreamChunk, None, None]:
-        """Execute the synchronous streaming operation with retry logic and error recovery."""
+        """Public synchronous interface for streaming chat completions with retries."""
+        stream_kwargs = self._build_stream_kwargs(messages, tools, **kwargs)
+
         for i in range(self.max_retries):
             try:
                 yield from self._stream_chat_sync(messages=messages, tools=tools, stream_kwargs=stream_kwargs)
                 return
 
             except Exception as e:
-                logger.exception(f"{operation_name} with model={self.model_name} encounter error with e={e.args}")
+                logger.exception(f"stream chat sync with model={self.model_name} encounter error with e={e.args}")
 
                 if i == self.max_retries - 1:
                     if self.raise_exception:
@@ -188,27 +143,6 @@ class BaseLLM(ABC):
                 yield StreamChunk(chunk_type=ChunkEnum.ERROR, chunk=str(e))
                 time.sleep(i + 1)
 
-    async def stream_chat(
-        self,
-        messages: list[Message],
-        tools: list[ToolCall] | None = None,
-        **kwargs,
-    ) -> AsyncGenerator[StreamChunk, None]:
-        """Public async interface for streaming chat completions with retries."""
-        stream_kwargs = self._build_stream_kwargs(messages, tools, **kwargs)
-        async for chunk in self._stream_with_retry("stream chat", messages, tools, stream_kwargs):
-            yield chunk
-
-    def stream_chat_sync(
-        self,
-        messages: list[Message],
-        tools: list[ToolCall] | None = None,
-        **kwargs,
-    ) -> Generator[StreamChunk, None, None]:
-        """Public synchronous interface for streaming chat completions with retries."""
-        stream_kwargs = self._build_stream_kwargs(messages, tools, **kwargs)
-        yield from self._stream_with_retry_sync("stream chat sync", messages, tools, stream_kwargs)
-
     async def _chat(
         self,
         messages: list[Message],
@@ -227,9 +161,46 @@ class BaseLLM(ABC):
 
         stream_kwargs = self._build_stream_kwargs(messages, tools, **kwargs)
         async for stream_chunk in self._stream_chat(messages=messages, tools=tools, stream_kwargs=stream_kwargs):
-            self._process_stream_chunk(stream_chunk, state, enable_stream_print)
+            # Process stream chunk
+            if stream_chunk.chunk_type is ChunkEnum.USAGE:
+                if enable_stream_print:
+                    print(
+                        f"\n<usage>{json.dumps(stream_chunk.chunk, ensure_ascii=False, indent=2)}</usage>",
+                        flush=True,
+                    )
 
-        return self._create_message_from_state(state)
+            elif stream_chunk.chunk_type is ChunkEnum.THINK:
+                if enable_stream_print:
+                    if not state["enter_think"]:
+                        state["enter_think"] = True
+                        print("<think>\n", end="", flush=True)
+                    print(stream_chunk.chunk, end="", flush=True)
+                state["reasoning_content"] += stream_chunk.chunk
+
+            elif stream_chunk.chunk_type is ChunkEnum.ANSWER:
+                if enable_stream_print:
+                    if not state["enter_answer"]:
+                        state["enter_answer"] = True
+                        if state["enter_think"]:
+                            print("\n</think>", flush=True)
+                    print(stream_chunk.chunk, end="", flush=True)
+                state["answer_content"] += stream_chunk.chunk
+
+            elif stream_chunk.chunk_type is ChunkEnum.TOOL:
+                if enable_stream_print:
+                    print(f"\n<tool>{json.dumps(stream_chunk.chunk, ensure_ascii=False, indent=2)}</tool>", flush=True)
+                state["tool_calls"].append(stream_chunk.chunk)
+
+            elif stream_chunk.chunk_type is ChunkEnum.ERROR:
+                if enable_stream_print:
+                    print(f"\n<error>{stream_chunk.chunk}</error>", flush=True)
+
+        return Message(
+            role=Role.ASSISTANT,
+            reasoning_content=state["reasoning_content"],
+            content=state["answer_content"],
+            tool_calls=state["tool_calls"],
+        )
 
     def _chat_sync(
         self,
@@ -249,57 +220,46 @@ class BaseLLM(ABC):
 
         stream_kwargs = self._build_stream_kwargs(messages, tools, **kwargs)
         for stream_chunk in self._stream_chat_sync(messages=messages, tools=tools, stream_kwargs=stream_kwargs):
-            self._process_stream_chunk(stream_chunk, state, enable_stream_print)
+            # Process stream chunk
+            if stream_chunk.chunk_type is ChunkEnum.USAGE:
+                if enable_stream_print:
+                    print(
+                        f"\n<usage>{json.dumps(stream_chunk.chunk, ensure_ascii=False, indent=2)}</usage>",
+                        flush=True,
+                    )
 
-        return self._create_message_from_state(state)
+            elif stream_chunk.chunk_type is ChunkEnum.THINK:
+                if enable_stream_print:
+                    if not state["enter_think"]:
+                        state["enter_think"] = True
+                        print("<think>\n", end="", flush=True)
+                    print(stream_chunk.chunk, end="", flush=True)
+                state["reasoning_content"] += stream_chunk.chunk
 
-    async def _execute_with_retry(
-        self,
-        operation_name: str,
-        operation_fn: Callable[[], Any],
-        callback_fn: Callable[[Message], Any] | None = None,
-        default_value: Any = None,
-    ) -> Message | Any:
-        """Execute a generic async operation with error handling and retry logic."""
-        for i in range(self.max_retries):
-            try:
-                result = await operation_fn()
-                return callback_fn(result) if callback_fn else result
+            elif stream_chunk.chunk_type is ChunkEnum.ANSWER:
+                if enable_stream_print:
+                    if not state["enter_answer"]:
+                        state["enter_answer"] = True
+                        if state["enter_think"]:
+                            print("\n</think>", flush=True)
+                    print(stream_chunk.chunk, end="", flush=True)
+                state["answer_content"] += stream_chunk.chunk
 
-            except Exception as e:
-                logger.exception(f"{operation_name} with model={self.model_name} encounter error with e={e.args}")
+            elif stream_chunk.chunk_type is ChunkEnum.TOOL:
+                if enable_stream_print:
+                    print(f"\n<tool>{json.dumps(stream_chunk.chunk, ensure_ascii=False, indent=2)}</tool>", flush=True)
+                state["tool_calls"].append(stream_chunk.chunk)
 
-                if i == self.max_retries - 1:
-                    if self.raise_exception:
-                        raise e
-                    return default_value
+            elif stream_chunk.chunk_type is ChunkEnum.ERROR:
+                if enable_stream_print:
+                    print(f"\n<error>{stream_chunk.chunk}</error>", flush=True)
 
-                await asyncio.sleep(1 + i)
-        return default_value
-
-    def _execute_with_retry_sync(
-        self,
-        operation_name: str,
-        operation_fn: Callable[[], Message],
-        callback_fn: Callable[[Message], Any] | None = None,
-        default_value: Any = None,
-    ) -> Message | Any:
-        """Execute a generic synchronous operation with error handling and retry logic."""
-        for i in range(self.max_retries):
-            try:
-                result = operation_fn()
-                return callback_fn(result) if callback_fn else result
-
-            except Exception as e:
-                logger.exception(f"{operation_name} with model={self.model_name} encounter error with e={e.args}")
-
-                if i == self.max_retries - 1:
-                    if self.raise_exception:
-                        raise e
-                    return default_value
-
-                time.sleep(1 + i)
-        return default_value
+        return Message(
+            role=Role.ASSISTANT,
+            reasoning_content=state["reasoning_content"],
+            content=state["answer_content"],
+            tool_calls=state["tool_calls"],
+        )
 
     async def chat(
         self,
@@ -311,17 +271,26 @@ class BaseLLM(ABC):
         **kwargs,
     ) -> Message | Any:
         """Perform an async chat completion with integrated retries and error handling."""
-        return await self._execute_with_retry(
-            operation_name="chat",
-            operation_fn=lambda: self._chat(
-                messages=messages,
-                tools=tools,
-                enable_stream_print=enable_stream_print,
-                **kwargs,
-            ),
-            callback_fn=callback_fn,
-            default_value=default_value,
-        )
+        for i in range(self.max_retries):
+            try:
+                result = await self._chat(
+                    messages=messages,
+                    tools=tools,
+                    enable_stream_print=enable_stream_print,
+                    **kwargs,
+                )
+                return callback_fn(result) if callback_fn else result
+
+            except Exception as e:
+                logger.exception(f"chat with model={self.model_name} encounter error with e={e.args}")
+
+                if i == self.max_retries - 1:
+                    if self.raise_exception:
+                        raise e
+                    return default_value
+
+                await asyncio.sleep(1 + i)
+        return default_value
 
     def chat_sync(
         self,
@@ -333,17 +302,26 @@ class BaseLLM(ABC):
         **kwargs,
     ) -> Message | Any:
         """Perform a synchronous chat completion with integrated retries and error handling."""
-        return self._execute_with_retry_sync(
-            operation_name="chat sync",
-            operation_fn=lambda: self._chat_sync(
-                messages=messages,
-                tools=tools,
-                enable_stream_print=enable_stream_print,
-                **kwargs,
-            ),
-            callback_fn=callback_fn,
-            default_value=default_value,
-        )
+        for i in range(self.max_retries):
+            try:
+                result = self._chat_sync(
+                    messages=messages,
+                    tools=tools,
+                    enable_stream_print=enable_stream_print,
+                    **kwargs,
+                )
+                return callback_fn(result) if callback_fn else result
+
+            except Exception as e:
+                logger.exception(f"chat sync with model={self.model_name} encounter error with e={e.args}")
+
+                if i == self.max_retries - 1:
+                    if self.raise_exception:
+                        raise e
+                    return default_value
+
+                time.sleep(1 + i)
+        return default_value
 
     async def close(self):
         """Release any asynchronous resources or connections held by the client."""

reme_ai/core/op/base_op.py

@@ -377,5 +377,6 @@ class BaseOp:
         copy_op = self.__class__(*self._init_args, **{**self._init_kwargs, **kwargs})
         if self.sub_ops:
             copy_op.sub_ops.clear()
-            copy_op.add_sub_ops(self.sub_ops)
+            for op in self.sub_ops:
+                copy_op.add_sub_op(op.copy())
         return copy_op

reme_ai/core/reme.py

@@ -1,48 +0,0 @@
-"""ReMe classes for simplified configuration and execution."""
-
-from .application import Application
-from .config import ReMeConfigParser
-from .context import C
-
-
-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()
-
-    async def summary(self):
-        """Execute summary operations."""
-
-    async def retrieve(self):
-        """Execute retrieve operations."""

reme_ai/core/schema/memory_node.py

@@ -153,7 +153,7 @@ class MemoryNode(BaseModel):
             str: Formatted string with when_to_use, content, and ref_memory_id.
         """
         parts: list[str] = [
-            f"memory_id={self.memory_id}" f"modified_time={self.time_modified}",
+            f"memory_id={self.memory_id} modified_time={self.time_modified}",
         ]
 
         if self.when_to_use:
@@ -166,7 +166,7 @@ class MemoryNode(BaseModel):
             parts.append(f"metadata={json.dumps(self.metadata, ensure_ascii=False)}")
 
         if self.ref_memory_id:
-            parts.append(f"history_memory.ref_memory_id={self.ref_memory_id}")
+            parts.append(f"ref_memory_id={self.ref_memory_id}")
 
         return " ".join(parts)
 

reme_ai/core/schema/message.py

@@ -83,7 +83,8 @@ class Message(BaseModel):
         add_reasoning: bool = True,
         add_time_created: bool = False,
         add_metadata: bool = False,
-    ) -> dict:
+        enable_json_dump: bool = False,
+    ) -> dict | str:
         """Transforms the message into a simplified dictionary for standard APIs."""
         result = {}
         if add_name and self.name:
@@ -107,7 +108,10 @@ class Message(BaseModel):
         if add_metadata:
             result["metadata"] = self.metadata
 
-        return result
+        if enable_json_dump:
+            return json.dumps(result, ensure_ascii=False)
+        else:
+            return result
 
     def format_message(
         self,

reme_ai/core/schema/service_config.py

@@ -74,7 +74,7 @@ class VectorStoreConfig(BaseModel):
     model_config = ConfigDict(extra="allow")
 
     backend: str = Field(default="local")
-    collection_name: str = Field(default="remy")
+    collection_name: str = Field(default="reme")
     embedding_model: str = Field(default="default")
 
 

reme_ai/core/service/cmd_service.py

@@ -26,9 +26,7 @@ class CmdService(BaseService):
         super().run()
 
         if self._cmd_flow.async_mode:
-            response = run_coro_safely(
-                self._cmd_flow.call(**C.service_config.cmd.model_extra),
-            )
+            response = run_coro_safely(self._cmd_flow.call(**C.service_config.cmd.model_extra))
         else:
             response = self._cmd_flow.call_sync(**C.service_config.cmd.model_extra)
 

reme_ai/mem_agent/__init__.py

@@ -1,15 +1,13 @@
 """memory agent"""
 
+from . import chat
 from . import retriever
 from . import summarizer
 from .base_memory_agent import BaseMemoryAgent
-from .simple_chat import SimpleChat
-from .stream_chat import StreamChat
 
 __all__ = [
+    "chat",
     "retriever",
     "summarizer",
     "BaseMemoryAgent",
-    "StreamChat",
-    "SimpleChat",
 ]

reme_ai/mem_agent/base_memory_agent.py

@@ -20,18 +20,16 @@ class BaseMemoryAgent(BaseOp, metaclass=ABCMeta):
         self,
         tools: list[BaseMemoryTool],
         add_think_tool: bool = False,  # only for instruct model
-        force_tool_language: bool = True,
         tool_call_interval: float = 0,
         max_steps: int = 20,
         **kwargs,
     ):
-        super().__init__(**kwargs)
-        self.tools: list[BaseMemoryTool] = tools or []
+        tools = tools or []
         if add_think_tool:
-            self.tools.append(ThinkTool())
-        if force_tool_language and self.language:
-            for tool in self.tools:
-                tool.language = self.language
+            tools.append(ThinkTool())
+        kwargs["sub_ops"] = tools
+        super().__init__(**kwargs)
+        self.sub_ops: list[BaseMemoryTool] = [t for t in self.sub_ops if isinstance(t, BaseMemoryTool)]
         self.tool_call_interval: float = tool_call_interval
         self.max_steps: int = max_steps
 
@@ -72,6 +70,15 @@ class BaseMemoryAgent(BaseOp, metaclass=ABCMeta):
             },
         )
 
+    @property
+    def tools(self):
+        """Returns the list of memory tools available to this agent."""
+        return self.sub_ops
+
+    @tools.setter
+    def tools(self, tools: list[BaseMemoryTool]):
+        self.sub_ops = tools
+
     def get_messages(self) -> list[Message]:
         """Extracts and returns messages from the context query or messages."""
         if self.context.get("query"):
@@ -93,7 +100,7 @@ class BaseMemoryAgent(BaseOp, metaclass=ABCMeta):
             **kwargs,
         )
         messages.append(assistant_message)
-        logger.info(f"step{step + 1}.assistant={assistant_message.model_dump_json()}")
+        logger.info(f"step{step + 1}.assistant={assistant_message.simple_dump(enable_json_dump=True)}")
         should_act = bool(assistant_message.tool_calls)
         return assistant_message, should_act
 
@@ -110,7 +117,7 @@ class BaseMemoryAgent(BaseOp, metaclass=ABCMeta):
                 logger.warning(f"unknown tool_call.name={tool_call.name}")
                 continue
 
-            logger.info(f"step{step + 1}.{j} submit tool_calls={tool_call.name} argument={tool_call.argument_dict}")
+            logger.info(f"step{step + 1}.{j} submit tool_calls={tool_call.name} argument={tool_call.arguments}")
             tool_copy: BaseMemoryTool = tool_dict[tool_call.name].copy()
             tool_copy.tool_call.id = tool_call.id
             tool_list.append(tool_copy)
@@ -150,7 +157,7 @@ class BaseMemoryAgent(BaseOp, metaclass=ABCMeta):
     async def execute(self):
         messages = await self.build_messages()
         for i, message in enumerate(messages):
-            logger.info(f"step0.{i} {message.role} {message.name or ''} {message.simple_dump()}")
+            logger.info(f"step0.{i} {message.role} {message.name or ''} {message.simple_dump(enable_json_dump=True)}")
 
         self.messages, self.success = await self.react(messages)
         if self.success and self.messages:
@@ -163,6 +170,11 @@ class BaseMemoryAgent(BaseOp, metaclass=ABCMeta):
         """Returns the target memory identifier from context."""
         return self.context.get("memory_target", "")
 
+    @property
+    def description(self) -> str:
+        """Returns the description of the messages."""
+        return self.context.get("description", "")
+
     @property
     def ref_memory_id(self) -> str:
         """Returns the reference memory ID from context."""

reme_ai/mem_agent/chat/__init__.py

@@ -0,0 +1,11 @@
+"""chat agent"""
+
+from .remy_agent import ReMyAgent
+from .simple_chat import SimpleChat
+from .stream_chat import StreamChat
+
+__all__ = [
+    "ReMyAgent",
+    "StreamChat",
+    "SimpleChat",
+]

reme_ai/mem_agent/chat/simple_chat.py

@@ -2,10 +2,10 @@
 
 from loguru import logger
 
-from ..core.context import C
-from ..core.enumeration import Role
-from ..core.op import BaseOp
-from ..core.schema import Message, ToolCall
+from ...core.context import C
+from ...core.enumeration import Role
+from ...core.op import BaseOp
+from ...core.schema import Message, ToolCall
 
 
 @C.register_op()

reme_ai/mem_agent/chat/stream_chat.py

@@ -2,10 +2,10 @@
 
 from loguru import logger
 
-from ..core.context import C
-from ..core.enumeration import Role, ChunkEnum
-from ..core.op import BaseOp
-from ..core.schema import Message, ToolCall
+from ...core.context import C
+from ...core.enumeration import Role, ChunkEnum
+from ...core.op import BaseOp
+from ...core.schema import Message, ToolCall
 
 
 @C.register_op()

reme_ai/mem_agent/retriever/__init__.py

@@ -1,9 +1,7 @@
 """memory retriever"""
 
 from .reme_retriever import ReMeRetriever
-from .remy_agent import ReMyAgent
 
 __all__ = [
     "ReMeRetriever",
-    "ReMyAgent",
 ]

reme_ai/mem_agent/retriever/reme_retriever.py

@@ -13,25 +13,32 @@ from ...core.utils import get_now_time, format_messages
 class ReMeRetriever(BaseMemoryAgent):
     """Memory agent that retrieves and builds messages with meta memory context."""
 
-    def __init__(self, enable_tool_memory: bool = True, **kwargs):
-        """Initialize retriever with tool memory option."""
+    def __init__(self, meta_memories: list[dict] = None, **kwargs):
         super().__init__(**kwargs)
-        self.enable_tool_memory = enable_tool_memory
+        self.meta_memories: list[dict] = meta_memories
 
-    async def _read_meta_memories(self) -> str:
+    @staticmethod
+    async def _read_meta_memories() -> str:
         """Read and return meta memories as string."""
         from ...mem_tool import ReadMetaMemory
 
-        op = ReadMetaMemory(enable_tool_memory=self.enable_tool_memory, enable_identity_memory=False)
+        op = ReadMetaMemory(enable_identity_memory=False)
         await op.call()
         return str(op.output)
 
     async def build_messages(self) -> List[Message]:
         """Build messages with system prompt and user message."""
+        from ...mem_tool import ReadMetaMemory
+
+        if self.meta_memories:
+            meta_memory_info = ReadMetaMemory().format_memory_metadata(self.meta_memories)
+        else:
+            meta_memory_info = await self._read_meta_memories()
+
         system_prompt = self.prompt_format(
             prompt_name="system_prompt",
             now_time=get_now_time(),
-            meta_memory_info=await self._read_meta_memories(),
+            meta_memory_info=meta_memory_info,
             context=format_messages(self.get_messages()),
         )
 

reme_ai/mem_agent/summarizer/personal_summarizer.py

@@ -18,7 +18,7 @@ class PersonalSummarizer(BaseMemoryAgent):
         system_prompt = self.prompt_format(
             prompt_name="system_prompt",
             now_time=get_now_time(),
-            context=format_messages(self.get_messages()),
+            context=self.description + "\n" + format_messages(self.get_messages()),
             memory_type=self.memory_type.value,
             memory_target=self.memory_target,
         )
@@ -37,5 +37,6 @@ class PersonalSummarizer(BaseMemoryAgent):
             memory_target=self.memory_target,
             memory_type=self.memory_type.value,
             author=self.author,
+            ref_memory_id=self.ref_memory_id,
             **kwargs,
         )

reme_ai/mem_agent/summarizer/personal_summarizer.yaml

@@ -6,7 +6,7 @@ tool: |
   or conflicts with existing memories, and perform add, update, or delete operations as needed.
 
 system_prompt: |
-  You are a professional memory agent specializing in the domain of **{memory_target}**. Your task is to update the main agent's {memory_type} memory regarding {memory_target} based on the context.
+  You are a professional memory agent. Your task is to update the main agent's {memory_type} memory regarding {memory_target} based on the context.
 
   ## Context:
   {context}

reme_ai/mem_agent/summarizer/procedural_summarizer.yaml

@@ -7,7 +7,7 @@ tool: |
   and successful strategies from successes to improve future performance.
 
 system_prompt: |
-  You are a professional memory Agent specializing in the domain of **{memory_target}**. Your task is to update the main agent's {memory_type} memory regarding {memory_target} based on the context.
+  You are a professional memory Agent specializing. Your task is to update the main agent's {memory_type} memory regarding {memory_target} based on the context.
 
   ## Context:
   {context}

reme_ai/mem_tool/__init__.py

@@ -1,6 +1,7 @@
 """Memory tool operations."""
 
 from .base_memory_tool import BaseMemoryTool
+from .hands_off_tool import HandsOffTool
 from .history.add_history_memory import AddHistoryMemory
 from .history.read_history_memory import ReadHistoryMemory
 from .identity.read_identity_memory import ReadIdentityMemory
@@ -16,6 +17,7 @@ from .vector.vector_retrieve_memory import VectorRetrieveMemory
 
 __all__ = [
     "BaseMemoryTool",
+    "HandsOffTool",
     "AddHistoryMemory",
     "ReadHistoryMemory",
     "ReadIdentityMemory",

reme_ai/mem_tool/hands_off_tool.py

@@ -17,17 +17,17 @@ if TYPE_CHECKING:
 class HandsOffTool(BaseMemoryTool):
     """Distribute memory tasks to appropriate agents based on memory_type."""
 
-    def __init__(self, memory_agents: list["BaseMemoryAgent"], force_agent_language: bool = True, **kwargs):
+    def __init__(self, memory_agents: list["BaseMemoryAgent"], **kwargs):
+        kwargs["sub_ops"] = memory_agents or []
         super().__init__(**kwargs)
-        self.memory_agent_dict: dict[MemoryType, "BaseMemoryAgent"] = {}
-        if memory_agents:
-            for agent in memory_agents:
-                if agent.memory_type is None:
-                    continue
+        from ..mem_agent import BaseMemoryAgent
 
-                self.memory_agent_dict[agent.memory_type] = agent
-                if force_agent_language and self.language:
-                    agent.language = self.language
+        self.sub_ops: list[BaseMemoryAgent] = [a for a in self.sub_ops if isinstance(a, BaseMemoryAgent)]
+
+    @property
+    def memory_agent_dict(self) -> dict[MemoryType, "BaseMemoryAgent"]:
+        """Returns a dictionary mapping memory types to their corresponding agents."""
+        return {a.memory_type: a for a in self.sub_ops}
 
     def _build_item_schema(self) -> tuple[dict, list[str]]:
         """Build shared schema properties and required fields for memory tasks."""
@@ -117,11 +117,13 @@ class HandsOffTool(BaseMemoryTool):
                 continue
 
             agent_copy = self.memory_agent_dict[memory_type].copy()
-            agent_list.append({
-                "agent": agent_copy,
-                "memory_type": memory_type,
-                "memory_target": memory_target,
-            })
+            agent_list.append(
+                {
+                    "agent": agent_copy,
+                    "memory_type": memory_type,
+                    "memory_target": memory_target,
+                },
+            )
 
             logger.info(f"Task {i}: Submitting {memory_type.value} agent for target={memory_target}")
             self.submit_async_task(
@@ -138,12 +140,14 @@ class HandsOffTool(BaseMemoryTool):
         results = []
         for i, (agent, memory_type, memory_target) in enumerate(agent_list):
             result_str = str(agent.output)
-            results.append({
-                "memory_type": memory_type.value,
-                "memory_target": memory_target,
-                "result": result_str[:200] + ("..." if len(result_str) > 200 else ""),
-            })
+            results.append(
+                {
+                    "memory_type": memory_type.value,
+                    "memory_target": memory_target,
+                    "result": result_str[:200] + ("..." if len(result_str) > 200 else ""),
+                },
+            )
             logger.info(f"Task {i}: Completed {memory_type.value} agent for target={memory_target}")
 
         results_str = json.dumps(results, ensure_ascii=False, indent=2)
-        self.set_output(f"Successfully executed {len(results)} memory tasks:\n{results_str}")
+        self.output = f"Successfully executed {len(results)} memory tasks:\n{results_str}"

reme_ai/mem_tool/meta/read_meta_memory.py

@@ -17,20 +17,17 @@ class ReadMetaMemory(BaseMemoryTool):
 
     def __init__(
         self,
-        enable_tool_memory: bool = False,
         enable_identity_memory: bool = False,
         **kwargs,
     ):
         """Initialize ReadMetaMemory.
 
         Args:
-            enable_tool_memory: Include TOOL type meta memory. Defaults to False.
             enable_identity_memory: Include IDENTITY type meta memory. Defaults to False.
             **kwargs: Additional arguments for BaseMemoryTool.
         """
         kwargs["enable_multiple"] = False
         super().__init__(**kwargs)
-        self.enable_tool_memory = enable_tool_memory
         self.enable_identity_memory = enable_identity_memory
 
     def _build_parameters(self) -> dict:
@@ -54,14 +51,6 @@ class ReadMetaMemory(BaseMemoryTool):
             if m.get("memory_type") in [MemoryType.PERSONAL.value, MemoryType.PROCEDURAL.value]:
                 filtered_memories.append(m)
 
-        if self.enable_tool_memory:
-            filtered_memories.append(
-                {
-                    "memory_type": MemoryType.TOOL.value,
-                    "memory_target": "tool_guidelines",
-                },
-            )
-
         if self.enable_identity_memory:
             filtered_memories.append(
                 {
@@ -72,7 +61,7 @@ class ReadMetaMemory(BaseMemoryTool):
 
         return filtered_memories
 
-    def _format_memory_metadata(self, memories: list[dict[str, str]]) -> str:
+    def format_memory_metadata(self, memories: list[dict[str, str]]) -> str:
         """Format memory metadata into a readable string.
 
         Args:
@@ -101,7 +90,7 @@ class ReadMetaMemory(BaseMemoryTool):
         memories = self._load_meta_memories()
 
         if memories:
-            self.output = self._format_memory_metadata(memories)
+            self.output = self.format_memory_metadata(memories)
             logger.info(f"Retrieved {len(memories)} meta memory entries")
         else:
             self.output = "No memory metadata found."

reme_ai/reme.py

@@ -0,0 +1,168 @@
+"""ReMe classes for simplified configuration and execution."""
+
+from typing import Literal
+
+from .core.application import Application
+from .core.config import ReMeConfigParser
+from .core.context import C
+from .core.enumeration import Role
+from .core.vector_store import BaseVectorStore
+from .mem_agent.summarizer import (
+    ReMeSummarizer,
+    # ToolSummarizer,
+    PersonalSummarizer,
+    ProceduralSummarizer,
+    # IdentitySummarizer,
+)
+from .mem_agent.retriever import ReMeRetriever
+
+# from .mem_agent.chat import ReMyAgent
+from .mem_tool import (
+    HandsOffTool,
+    ReadHistoryMemory,
+    # ReadIdentityMemory,
+    # UpdateIdentityMemory,
+    AddMetaMemory,
+    AddMemory,
+    AddSummaryMemory,
+    DeleteMemory,
+    UpdateMemory,
+    VectorRetrieveMemory,
+)
+from .core.schema import Message
+
+
+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,
+        enable_identity_memory: bool = True,
+        enable_tool_memory: bool = True,
+        force_tool_language: bool = True,
+        add_think_tool: bool = False,
+        **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.enable_identity_memory = enable_identity_memory
+        self.enable_tool_memory = enable_tool_memory
+        self.force_tool_language = force_tool_language
+        self.add_think_tool = add_think_tool
+
+        self._personal_summarizer = PersonalSummarizer(
+            tools=[VectorRetrieveMemory(), AddMemory(), DeleteMemory(), UpdateMemory()],
+        )
+        self._procedural_summarizer = ProceduralSummarizer(
+            tools=[VectorRetrieveMemory(), AddMemory(), DeleteMemory(), UpdateMemory()],
+        )
+        hands_off_tool = HandsOffTool(memory_agents=[self._personal_summarizer, self._procedural_summarizer])
+        self._reme_summarizer = ReMeSummarizer(
+            tools=[AddMetaMemory(), AddSummaryMemory(), hands_off_tool],
+            enable_identity_memory=self.enable_identity_memory,
+            enable_tool_memory=self.enable_tool_memory,
+            force_tool_language=self.force_tool_language,
+            add_think_tool=self.add_think_tool,
+        )
+        self._reme_retriever = ReMeRetriever(
+            tools=[VectorRetrieveMemory(add_memory_type_target=True), ReadHistoryMemory()],
+        )
+
+        self.vector_store: BaseVectorStore = C.get_vector_store("default")
+
+    @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 = "",
+        memory_mode: Literal["personal", "procedural", "auto"] = "personal",
+        **kwargs,
+    ):
+        """Summarizes messages and stores them as memory based on the specified memory mode."""
+        messages = self._prepare_messages(messages, user_id, assistant_id)
+
+        if memory_mode == "personal":
+            return await self._personal_summarizer.call(
+                messages=messages,
+                description=description,
+                memory_target=user_id,
+                **kwargs,
+            )
+        elif memory_mode == "procedural":
+            return await self._procedural_summarizer.call(
+                messages=messages,
+                description=description,
+                memory_target=user_id,
+                **kwargs,
+            )
+        else:
+            return await self._reme_summarizer.call(
+                messages=messages,
+                description=description,
+                memory_target=user_id,
+                **kwargs,
+            )
+
+    async def retrieve(
+        self,
+        query: str = "",
+        messages: list[dict] | None = None,
+        description: str = "",
+        user_id: str = "",
+        assistant_id: str = "",
+        memory_mode: Literal["personal", "procedural", "auto"] = "personal",
+        **kwargs,
+    ):
+        """Retrieves relevant memories based on the query and specified memory mode."""
+        messages = self._prepare_messages(messages, user_id, assistant_id)
+
+        if memory_mode == "personal":
+            self._reme_retriever.meta_memories = [{"memory_type": "personal", "memory_target": user_id}]
+            return await self._reme_retriever.call(query=query, messages=messages, description=description, **kwargs)
+
+        elif memory_mode == "procedural":
+            self._reme_retriever.meta_memories = [{"memory_type": "procedural", "memory_target": user_id}]
+            return await self._reme_retriever.call(query=query, messages=messages, description=description, **kwargs)
+
+        else:
+            return await self._reme_retriever.call(query=query, messages=messages, description=description, **kwargs)

tests/test_reme.py

@@ -0,0 +1,111 @@
+"""Test module for ReMe memory system functionality."""
+
+import asyncio
+
+from reme_ai.core.schema import VectorNode, MemoryNode
+from reme_ai.reme import ReMe
+
+reme = ReMe(
+    vector_store={"collection_name": "reme"},
+)
+
+
+async def test_reme():
+    """Tests ReMe memory system with personal information storage and retrieval."""
+    # 构建一段包含个人信息的对话
+    await reme.vector_store.delete_collection("reme")
+
+    messages = [
+        {
+            "role": "user",
+            "content": "你好，我是张伟，今年28岁，是一名软件工程师。",
+        },
+        {
+            "role": "assistant",
+            "content": "你好张伟！很高兴认识你。作为一名软件工程师，你主要从事什么方向的开发工作呢？",
+        },
+        {
+            "role": "user",
+            "content": "我主要做后端开发，擅长Python和Go语言。最近在研究AI Agent相关的技术。",
+        },
+        {
+            "role": "assistant",
+            "content": "很棒！Python和Go都是非常实用的语言。AI Agent是当前很热门的方向，你在这方面有什么具体的研究重点吗？",
+        },
+        {
+            "role": "user",
+            "content": "我特别关注记忆系统的设计，希望能让AI Agent具有长期记忆能力。我的工作地点在北京，平时喜欢看技术博客和参加技术分享会。",
+        },
+        {
+            "role": "assistant",
+            "content": "记忆系统确实是AI Agent的核心能力之一。北京有很多优秀的技术社区和活动，相信你能找到很多志同道合的朋友。",
+        },
+        {
+            "role": "user",
+            "content": "是的，我每周末都会去参加一些技术沙龙。对了，我的邮箱是zhangwei@example.com，如果有好的技术资料可以发给我。",
+        },
+        {
+            "role": "assistant",
+            "content": "好的，我记下了。保持学习的热情很重要，祝你在AI Agent领域的研究越来越深入！",
+        },
+    ]
+
+    print("=" * 60)
+    print("步骤1: 开始总结对话并生成记忆")
+    print("=" * 60)
+
+    # 对对话进行总结，生成记忆
+    await reme.summary(
+        messages=messages,
+        user_id="zhangwei",
+        description="用户自我介绍和技术兴趣分享",
+        ref_memory_id="ref_123",
+    )
+
+    print("\n✓ 记忆总结完成")
+
+    print("\n" + "=" * 60)
+    print("步骤2: 查看已存储的记忆节点")
+    print("=" * 60)
+
+    # 列出所有存储的记忆节点
+    nodes: list[VectorNode] = await reme.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()}")
+
+    print("\n" + "=" * 60)
+    print("步骤3: 测试记忆检索 - 验证个人信息")
+    print("=" * 60)
+
+    # 测试问题1: 检索用户姓名
+    query1 = "用户叫什么名字？"
+    print(f"\n问题1: {query1}")
+    result1 = await reme.retrieve(query=query1, user_id="zhangwei")
+    print(f"检索结果:\n{result1}")
+
+    # 测试问题2: 检索技术背景
+    query2 = "用户擅长什么编程语言和技术方向？"
+    print(f"\n问题2: {query2}")
+    result2 = await reme.retrieve(query=query2, user_id="zhangwei")
+    print(f"检索结果:\n{result2}")
+
+    # 测试问题3: 检索个人信息
+    query3 = "用户的工作地点和联系方式是什么？"
+    print(f"\n问题3: {query3}")
+    result3 = await reme.retrieve(query=query3, user_id="zhangwei")
+    print(f"检索结果:\n{result3}")
+
+    # 测试问题4: 检索兴趣爱好
+    query4 = "用户平时有什么爱好或活动？"
+    print(f"\n问题4: {query4}")
+    result4 = await reme.retrieve(query=query4, user_id="zhangwei")
+    print(f"检索结果:\n{result4}")
+
+    print("\n" + "=" * 60)
+    print("测试完成！")
+    print("=" * 60)
+
+
+if __name__ == "__main__":
+    asyncio.run(test_reme())