Merge pull request #65 from agentscope-ai/dev_0105

Dev 0105

pyproject.toml

@@ -33,7 +33,7 @@ classifiers = [
 keywords = ["llm", "memory", "experience", "memoryscope", "ai", "mcp", "http"]
 
 dependencies = [
-    "flowllm[reme]>=0.2.0.7",
+    "flowllm[reme]>=0.2.0.9",
 ]
 
 [project.optional-dependencies]
@@ -48,7 +48,7 @@ dev = [
 ]
 
 token = [
-    "flowllm[token]>=0.2.0.5"
+    "flowllm[token]>=0.2.0.9"
 ]
 
 full = [

reme_ai/__init__.py

@@ -30,4 +30,4 @@ __all__ = [
     "vector_store",
 ]
 
-__version__ = "0.2.0.4"
+__version__ = "0.2.0.5"

reme_ai/core/__init__.py

@@ -0,0 +1,17 @@
+"""Core module for ReMe AI framework."""
+
+# pylint: disable=wrong-import-position
+# flake8: noqa: F401
+
+from . import config
+from . import context
+from . import embedding
+from . import enumeration
+from . import flow
+from . import llm
+from . import op
+from . import schema
+from . import service
+from . import token_counter
+from . import utils
+from . import vector_store

reme_ai/core/application.py

@@ -6,7 +6,7 @@ import os
 from .context import C
 from .flow import BaseFlow
 from .schema import ServiceConfig, Response
-from .utils import PydanticConfigParser, init_logger, execute_stream_task, run_coro_safely
+from .utils import PydanticConfigParser, init_logger, execute_stream_task, run_coro_safely, load_env
 
 
 class Application:
@@ -38,6 +38,15 @@ class Application:
         Initialize the Application with configuration settings.
 
         Args:
+            *args: Additional arguments passed to parser. Examples:
+                - "llm.default.model_name=qwen3-30b-a3b-thinking-2507"
+                - "llm.default.backend=openai_compatible"
+                - "llm.default.temperature=0.6"
+                - "embedding_model.default.model_name=text-embedding-v4"
+                - "embedding_model.default.backend=openai_compatible"
+                - "embedding_model.default.dimensions=1024"
+                - "vector_store.default.backend=memory"
+                - "vector_store.default.embedding_model=default"
             llm_api_key: API key for LLM service
             llm_api_base: Base URL for LLM service
             embedding_api_key: API key for embedding service
@@ -50,8 +59,11 @@ class Application:
             embedding_model: Embedding model configuration dictionary
             vector_store: Vector store configuration dictionary
             token_counter: Token counter configuration dictionary
-            **kwargs: Additional configuration arguments
+            **kwargs: Additional keyword arguments passed to parser. Same format as args but as kwargs. Examples:
+                - **{"llm.default.model_name": "qwen3-30b-a3b-thinking-2507"}
         """
+
+        load_env()
         self._update_env("REME_LLM_API_KEY", llm_api_key)
         self._update_env("REME_LLM_BASE_URL", llm_api_base)
         self._update_env("REME_EMBEDDING_API_KEY", embedding_api_key)
@@ -87,25 +99,25 @@ class Application:
         C.print_logo()
 
     @staticmethod
-    def _update_env(key: str, value: str | None) -> None:
+    def _update_env(key: str, value: str | None):
         """Update environment variable if value is provided."""
         if value:
             os.environ[key] = value
 
     @staticmethod
-    async def start() -> None:
+    async def start():
         """Initialize the service context and prepare external MCP servers."""
         C.initialize_service_context()
         await C.prepare_mcp_servers()
 
     @staticmethod
-    def start_sync() -> None:
+    def start_sync():
         """Synchronous version of start()."""
         C.initialize_service_context()
         run_coro_safely(C.prepare_mcp_servers())
 
     @staticmethod
-    async def stop(wait_thread_pool: bool = True, wait_ray: bool = True) -> None:
+    async def stop(wait_thread_pool: bool = True, wait_ray: bool = True):
         """
         Stop the application and cleanup resources.
 
@@ -118,7 +130,7 @@ class Application:
         C.shutdown_ray(wait=wait_ray)
 
     @staticmethod
-    def stop_sync(wait_thread_pool: bool = True, wait_ray: bool = True) -> None:
+    def stop_sync(wait_thread_pool: bool = True, wait_ray: bool = True):
         """Synchronous version of stop()."""
         C.close_sync()
         C.shutdown_thread_pool(wait=wait_thread_pool)
@@ -195,9 +207,9 @@ class Application:
         task = asyncio.create_task(flow.call(stream_queue=stream_queue, **kwargs))
 
         async for chunk in execute_stream_task(
-            queue=stream_queue,
+            stream_queue=stream_queue,
             task=task,
-            flow_name=name,
+            task_name=name,
             as_bytes=False,
         ):
             yield chunk

reme_ai/core/config/default.yaml

@@ -16,6 +16,10 @@ llm:
   default:
     backend: openai
     model_name: qwen3-30b-a3b-instruct-2507
+
+  qwen3_max_instruct:
+    backend: openai
+    model_name: qwen3-max
     temperature: 0.6
 
 embedding_model:
@@ -32,3 +36,8 @@ vector_store:
 token_counter:
   default:
     backend: base
+
+  hf:
+    backend: hf
+    model_name: Qwen/Qwen3-Coder-30B-A3B-Instruct
+    use_mirror: true

reme_ai/core/context/service_context.py

@@ -60,6 +60,9 @@ class ServiceContext(BaseContext):
         # MCP server mapping: maps server_name -> {tool_name: ToolCall}
         self.mcp_server_mapping: dict[str, dict] = {}
 
+        # Initialization flag: ensures initialize_service_context is called only once
+        self._initialized: bool = False
+
     def register(self, name: str, register_type: RegistryEnum):
         """Return a decorator to register a component within a specific registry category.
 
@@ -268,7 +271,12 @@ class ServiceContext(BaseContext):
         9. Service backend instance
 
         Note: This method should be called after service_config is set.
+        This method can only be called once. Subsequent calls will be ignored.
         """
+        if self._initialized:
+            logger.warning("initialize_service_context has already been called. Skipping re-initialization.")
+            return
+
         self.language = self.service_config.language
         self.thread_pool = ThreadPoolExecutor(max_workers=self.service_config.thread_pool_max_workers)
 
@@ -286,6 +294,9 @@ class ServiceContext(BaseContext):
         self._initialize_flow()
         self._initialize_service()
 
+        # Mark as initialized
+        self._initialized = True
+
     def _initialize_llm(self):
         """Initialize all configured LLM instances.
 

reme_ai/core/enumeration/__init__.py

@@ -3,6 +3,7 @@
 from .chunk_enum import ChunkEnum
 from .http_enum import HttpEnum
 from .json_schema_enum import JsonSchemaEnum
+from .memory_type import MemoryType
 from .registry_enum import RegistryEnum
 from .role import Role
 
@@ -10,6 +11,7 @@ __all__ = [
     "ChunkEnum",
     "HttpEnum",
     "JsonSchemaEnum",
+    "MemoryType",
     "RegistryEnum",
     "Role",
 ]

reme_ai/core/enumeration/memory_type.py

@@ -0,0 +1,25 @@
+"""Memory type enumeration for the three-layer memory architecture."""
+
+from enum import Enum
+
+
+class MemoryType(str, Enum):
+    """
+    Three-layer memory architecture for agent memory management.
+
+    Layer 1 - High-level Abstraction Memory:
+        - IDENTITY: Self-cognition (identity, personality, current state)
+        - PERSONAL: Person-specific memory (preferences and context about specific individuals)
+        - PROCEDURAL: Procedural memory (how-to knowledge, e.g., 4 steps to write financial reports)
+        - TOOL: Tool memory (tool usage patterns, success rates, token consumption, latency)
+
+    Layer 2 - Summary Memory (Compressed): Summarized digest of raw message history
+    Layer 3 - History Memory (Raw): Raw message history
+    """
+
+    IDENTITY = "identity"
+    PERSONAL = "personal"
+    PROCEDURAL = "procedural"
+    TOOL = "tool"
+    SUMMARY = "summary"
+    HISTORY = "history"

reme_ai/core/main.py

@@ -0,0 +1,48 @@
+"""ReMe application classes for simplified configuration and execution."""
+
+import sys
+
+from .application import Application
+from .config import ReMeConfigParser
+
+
+class ReMeApp(Application):
+    """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,
+        config_path: str | None = None,
+        enable_logo: bool = True,
+        **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=config_path,
+            enable_logo=enable_logo,
+            **kwargs,
+        )
+
+    async def async_execute(self, name: str, **kwargs) -> dict:
+        """Execute a flow asynchronously and return the result as a dictionary."""
+        return (await self.execute_flow(name=name, **kwargs)).model_dump()
+
+
+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()

reme_ai/core/op/base_op.py

@@ -40,6 +40,7 @@ class BaseOp:
         token_counter: str | BaseTokenCounter = "default",
         enable_cache: bool = False,
         cache_path: str = "cache/op",
+        cache_expire_hours: float | None = None,
         sub_ops: dict[str, "BaseOp"] | list["BaseOp"] | Optional["BaseOp"] = None,
         input_mapping: dict[str, str] | None = None,
         output_mapping: dict[str, str] | None = None,
@@ -62,6 +63,7 @@ class BaseOp:
 
         self.enable_cache = enable_cache
         self.cache_path = cache_path
+        self.cache_expire_hours = cache_expire_hours
         self.sub_ops: list[BaseOp] = []
         self.add_sub_ops(sub_ops)
 
@@ -156,7 +158,10 @@ class BaseOp:
             return None
 
         keys = list(output_properties.keys())
-        return self.context[keys[0]]
+        if len(keys) >= 1 and keys[0] in self.context:
+            return self.context[keys[0]]
+        else:
+            return None
 
     @output.setter
     def output(self, value: Any):

reme_ai/core/reme.py

@@ -1,6 +1,4 @@
-"""ReMe application classes for simplified configuration and execution."""
-
-import sys
+"""ReMe classes for simplified configuration and execution."""
 
 from .application import Application
 from .config import ReMeConfigParser
@@ -48,45 +46,3 @@ class ReMe(Application):
 
     async def retrieve(self):
         """Execute retrieve operations."""
-
-
-class ReMeApp(Application):
-    """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,
-        config_path: str | None = None,
-        enable_logo: bool = True,
-        **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=config_path,
-            enable_logo=enable_logo,
-            **kwargs,
-        )
-
-    async def async_execute(self, name: str, **kwargs) -> dict:
-        """Execute a flow asynchronously and return the result as a dictionary."""
-        return (await self.execute_flow(name=name, **kwargs)).model_dump()
-
-
-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()

reme_ai/core/schema/__init__.py

@@ -1,5 +1,6 @@
 """schema"""
 
+from .memory_node import MemoryNode
 from .message import ContentBlock, Message, Trajectory
 from .request import Request
 from .response import Response
@@ -19,6 +20,7 @@ from .tool_call import ToolAttr, ToolCall
 from .vector_node import VectorNode
 
 __all__ = [
+    "MemoryNode",
     "ContentBlock",
     "EmbeddingModelConfig",
     "FlowConfig",

reme_ai/core/schema/memory_node.py

@@ -0,0 +1,213 @@
+"""Memory schema module for the ReMe AI system.
+
+This module defines the MemoryNode class for storing and retrieving
+memories in the ReMe system.
+"""
+
+import hashlib
+import json
+from typing import Any
+
+from pydantic import BaseModel, Field, model_validator
+
+from ..enumeration import MemoryType
+from ..schema import VectorNode
+from ..utils import get_now_time
+
+# Length of the memory ID (first N characters of SHA-256 hash)
+MEMORY_ID_LENGTH: int = 16
+
+
+class MemoryNode(BaseModel):
+    """Memory node for storing memories in the ReMe system.
+
+    Attributes:
+        memory_id: Unique identifier, auto-generated from content hash.
+        memory_type: Type of memory (e.g., SUMMARY, PERSONAL).
+        memory_target: Target or topic this memory relates to.
+        when_to_use: Condition description for vector retrieval.
+        content: Actual memory content.
+        ref_memory_id: Reference to related raw history memory.
+        time_created: Creation timestamp.
+        time_modified: Last modification timestamp.
+        author: Author or source of this memory.
+        score: Relevance or importance score.
+        metadata: Additional metadata for extensibility.
+    """
+
+    memory_id: str = Field(default="", description="Unique memory identifier")
+    memory_type: MemoryType = Field(default=..., description="Type of memory")
+    memory_target: str = Field(default="", description="Target or topic of the memory")
+    when_to_use: str = Field(default="", description="Condition description for vector retrieval")
+    content: str = Field(default="", description="Actual memory content")
+    ref_memory_id: str = Field(default="", description="Reference to related raw history memory ID")
+
+    time_created: str = Field(default_factory=get_now_time, description="Creation timestamp")
+    time_modified: str = Field(default_factory=get_now_time, description="Last modification timestamp")
+    author: str = Field(default="", description="Author or source of the memory")
+    score: float = Field(default=0, description="Relevance or importance score")
+
+    metadata: dict[str, Any] = Field(default_factory=dict, description="Additional metadata")
+
+    def _update_modified_time(self) -> "MemoryNode":
+        """Update time_modified to current timestamp.
+
+        Returns:
+            Self: Returns self for method chaining.
+        """
+        self.time_modified = get_now_time()
+        return self
+
+    def _update_memory_id(self) -> "MemoryNode":
+        """Generate memory_id from SHA-256 hash of content.
+
+        Takes the first MEMORY_ID_LENGTH characters of the hash.
+
+        Returns:
+            Self: Returns self for method chaining.
+        """
+        if not self.content:
+            return self
+
+        hash_obj = hashlib.sha256(self.content.encode("utf-8"))
+        hex_dig = hash_obj.hexdigest()
+        self.memory_id = hex_dig[:MEMORY_ID_LENGTH]
+        return self
+
+    @model_validator(mode="after")
+    def _update_after_init(self) -> "MemoryNode":
+        """Post-initialization validator.
+
+        Auto-generates memory_id from content if not provided.
+
+        Returns:
+            Self: Returns self for method chaining.
+        """
+        if not self.memory_id:
+            self._update_memory_id()
+        return self
+
+    def __setattr__(self, name: str, value):
+        """Auto-update timestamps and memory_id when content or when_to_use changes.
+
+        Args:
+            name: Attribute name being set.
+            value: New value for the attribute.
+        """
+        should_update: bool = name in ("when_to_use", "content") and getattr(self, name, None) != value
+        super().__setattr__(name, value)
+        if should_update:
+            self._update_modified_time()
+            if name == "content":
+                self._update_memory_id()
+
+    def to_vector_node(self) -> VectorNode:
+        """Convert to VectorNode for vector storage.
+
+        When when_to_use is set, use it as vector content and store content in metadata.
+        When when_to_use is empty, use content as vector content directly.
+
+        Returns:
+            VectorNode: Vector node representation of this memory.
+        """
+        # Build base metadata (shared fields)
+        metadata: dict[str, Any] = {
+            "memory_type": self.memory_type.value,
+            "memory_target": self.memory_target,
+            "ref_memory_id": self.ref_memory_id,
+            "time_created": self.time_created,
+            "time_modified": self.time_modified,
+            "author": self.author,
+            "score": self.score,
+            **self.metadata,
+        }
+
+        if self.when_to_use:
+            # Use when_to_use for vector embedding, store content in metadata
+            vector_content = self.when_to_use
+            metadata["content"] = self.content
+        else:
+            # Use content directly for vector embedding
+            vector_content = self.content
+
+        return VectorNode(
+            vector_id=self.memory_id,
+            content=vector_content,
+            metadata=metadata,
+        )
+
+    def format_memory(self) -> str:
+        """Format memory as human-readable string.
+
+        Returns:
+            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}",
+        ]
+
+        if self.when_to_use:
+            parts.append(self.when_to_use)
+
+        if self.content:
+            parts.append(self.content)
+
+        if self.metadata:
+            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}")
+
+        return " ".join(parts)
+
+    @classmethod
+    def from_vector_node(cls, node: VectorNode) -> "MemoryNode":
+        """Reconstruct MemoryNode from VectorNode.
+
+        Reverses the to_vector_node conversion:
+        - If metadata contains 'content': node.content -> when_to_use, metadata['content'] -> content
+        - Otherwise: node.content -> content, when_to_use remains empty
+
+        Args:
+            node: VectorNode containing memory data.
+
+        Returns:
+            Self: Reconstructed MemoryNode instance.
+
+        Raises:
+            ValueError: If memory_type in metadata is invalid.
+        """
+        metadata = node.metadata.copy()
+        memory_type_str = metadata.pop("memory_type", None)
+
+        try:
+            memory_type: MemoryType = MemoryType(memory_type_str)
+        except ValueError as e:
+            raise ValueError(
+                f"Invalid memory_type '{memory_type_str}' in VectorNode metadata. "
+                f"Valid types are: {[t.value for t in MemoryType]}",
+            ) from e
+
+        # Restore when_to_use and content based on metadata structure
+        if "content" in metadata:
+            # Original had when_to_use set
+            when_to_use = node.content
+            content = metadata.pop("content", "")
+        else:
+            # Original had empty when_to_use
+            when_to_use = ""
+            content = node.content
+
+        return cls(
+            memory_id=node.vector_id,
+            memory_type=memory_type,
+            memory_target=metadata.pop("memory_target", ""),
+            when_to_use=when_to_use,
+            content=content,
+            ref_memory_id=metadata.pop("ref_memory_id", ""),
+            time_created=metadata.pop("time_created", ""),
+            time_modified=metadata.pop("time_modified", ""),
+            author=metadata.pop("author", ""),
+            score=metadata.pop("score", 0),
+            metadata=metadata,
+        )

reme_ai/core/service/http_service.py

@@ -51,15 +51,14 @@ class HttpService(BaseService):
         tool_call, request_model = self._prepare_route(flow)
 
         async def execute_stream_endpoint(request: request_model) -> StreamingResponse:
-            queue = asyncio.Queue()
-            # Start flow as a background task
-            task = asyncio.create_task(flow.call(stream_queue=queue, **request.model_dump(exclude_none=True)))
+            stream_queue = asyncio.Queue()
+            task = asyncio.create_task(flow.call(stream_queue=stream_queue, **request.model_dump(exclude_none=True)))
 
             async def generate_stream() -> AsyncGenerator[bytes, None]:
                 async for chunk in execute_stream_task(
-                    queue=queue,
+                    stream_queue=stream_queue,
                     task=task,
-                    flow_name=tool_call.name,
+                    task_name=tool_call.name,
                     as_bytes=True,
                 ):
                     yield chunk

reme_ai/core/tool/__init__.py

@@ -1,7 +0,0 @@
-"""tool"""
-
-from .mcp_tool import MCPTool
-
-__all__ = [
-    "MCPTool",
-]

reme_ai/core/utils/__init__.py

@@ -4,15 +4,16 @@ from .cache_handler import CacheHandler
 from .case_converter import snake_to_camel, camel_to_snake
 from .common_utils import run_coro_safely, execute_stream_task
 from .env_utils import load_env
+from .execute_tuils import exec_code, run_shell_command
 from .http_client import HttpClient
-from .llm_utils import extract_content, format_messages
+from .llm_utils import extract_content, format_messages, deduplicate_memories
 from .logger_utils import init_logger
 from .logo_utils import print_logo
 from .mcp_client import MCPClient
 from .pydantic_config_parser import PydanticConfigParser
 from .pydantic_utils import create_pydantic_model
 from .singleton import singleton
-from .timer import timer
+from .time import timer, get_now_time
 
 __all__ = [
     "CacheHandler",
@@ -21,9 +22,12 @@ __all__ = [
     "run_coro_safely",
     "execute_stream_task",
     "load_env",
+    "exec_code",
+    "run_shell_command",
     "HttpClient",
     "extract_content",
     "format_messages",
+    "deduplicate_memories",
     "init_logger",
     "print_logo",
     "MCPClient",
@@ -31,4 +35,5 @@ __all__ = [
     "create_pydantic_model",
     "singleton",
     "timer",
+    "get_now_time",
 ]

reme_ai/core/utils/common_utils.py

@@ -26,9 +26,9 @@ def run_coro_safely(coro: Coroutine[Any, Any, Any]) -> Any | asyncio.Task[Any]:
 
 
 async def execute_stream_task(
-    queue: asyncio.Queue,
+    stream_queue: asyncio.Queue,
     task: asyncio.Task,
-    flow_name: str | None = None,
+    task_name: str | None = None,
     as_bytes: bool = False,
 ) -> AsyncGenerator[str | bytes, None]:
     """
@@ -38,9 +38,9 @@ async def execute_stream_task(
     Properly manages errors and resource cleanup.
 
     Args:
-        queue: Queue to receive StreamChunk objects from
+        stream_queue: Queue to receive StreamChunk objects from
         task: Background task executing the flow
-        flow_name: Optional flow name for logging purposes
+        task_name: Optional flow name for logging purposes
         as_bytes: If True, yield bytes for HTTP responses; if False, yield strings
 
     Yields:
@@ -51,7 +51,7 @@ async def execute_stream_task(
     try:
         while True:
             # Wait for next chunk or check if task failed
-            get_chunk = asyncio.create_task(queue.get())
+            get_chunk = asyncio.create_task(stream_queue.get())
             done, _ = await asyncio.wait({get_chunk, task}, return_when=asyncio.FIRST_COMPLETED)
 
             if get_chunk in done:
@@ -69,7 +69,7 @@ async def execute_stream_task(
                 break
 
     except Exception as e:
-        log_msg = f"Stream error in {flow_name}: {e}" if flow_name else f"Stream error: {e}"
+        log_msg = f"Stream error in {task_name}: {e}" if task_name else f"Stream error: {e}"
         logger.exception(log_msg)
 
         err = StreamChunk(chunk_type=ChunkEnum.ERROR, chunk=str(e), done=True)

reme_ai/core/utils/execute_tuils.py

@@ -0,0 +1,60 @@
+"""Utility functions for executing code and shell commands.
+
+This module provides helper functions for running Python code and shell commands,
+with support for async execution and output capture.
+"""
+
+import asyncio
+import contextlib
+from io import StringIO
+
+
+async def run_shell_command(cmd: str, timeout: float | None = 30) -> tuple[str, str, int]:
+    """Execute a shell command asynchronously.
+
+    Args:
+        cmd: The shell command to execute.
+        timeout: Maximum time to wait for command completion in seconds. None for no timeout.
+
+    Returns:
+        A tuple containing (stdout, stderr, return_code) as strings and integer.
+    """
+    process = await asyncio.create_subprocess_shell(
+        cmd,
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.PIPE,
+    )
+
+    if timeout:
+        stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=timeout)
+    else:
+        stdout, stderr = await process.communicate()
+
+    return (
+        stdout.decode("utf-8", errors="ignore"),
+        stderr.decode("utf-8", errors="ignore"),
+        process.returncode,
+    )
+
+
+def exec_code(code: str) -> str:
+    """Execute Python code and capture the output.
+
+    Args:
+        code: The Python code string to execute.
+
+    Returns:
+        The captured stdout output, or the error message if execution fails.
+    """
+    try:
+        redirected_output = StringIO()
+        with contextlib.redirect_stdout(redirected_output):
+            exec(code)
+
+        return redirected_output.getvalue()
+
+    except Exception as e:
+        return str(e)
+
+    except BaseException as e:
+        return str(e)

reme_ai/core/utils/llm_utils.py

@@ -4,19 +4,27 @@ import json
 import re
 
 from ..enumeration import Role
-from ..schema import Message
+from ..schema import Message, MemoryNode
 
 
 def format_messages(messages: list[Message | dict], enable_system: bool = False) -> str:
     """Formats a list of messages into a single string, optionally filtering system roles."""
     formatted_lines = []
-    for message in messages:
+    for i, message in enumerate(messages):
         if isinstance(message, dict):
             message = Message(**message)
         if not enable_system and message.role is Role.SYSTEM:
             continue
 
-        formatted_lines.append(message.format_message())
+        formatted_lines.append(
+            message.format_message(
+                index=i,
+                add_time=True,
+                use_name=True,
+                add_reasoning=True,
+                add_tools=True,
+            ),
+        )
     return "\n".join(formatted_lines)
 
 
@@ -38,3 +46,12 @@ def extract_content(text: str, language_tag: str = "json", greedy: bool = False)
             result = None
 
     return result
+
+
+def deduplicate_memories(memories: list[MemoryNode]) -> list[MemoryNode]:
+    """Deduplicates a list of memories by memory ID."""
+    seen_memories: dict[str, MemoryNode] = {}
+    for memory in memories:
+        if memory.memory_id not in seen_memories:
+            seen_memories[memory.memory_id] = memory
+    return list(seen_memories.values())

reme_ai/core/utils/time.py

@@ -2,6 +2,7 @@
 Utility module for timing function execution with log metadata preservation.
 """
 
+import datetime
 import functools
 import inspect
 import time
@@ -13,6 +14,15 @@ from loguru import logger
 F = TypeVar("F", bound=Callable[..., Any])
 
 
+def get_now_time() -> str:
+    """Get current timestamp in YYYY-MM-DD HH:MM:SS format.
+
+    Returns:
+        str: Current timestamp string in format 'YYYY-MM-DD HH:MM:SS'.
+    """
+    return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+
+
 def timer(func: F) -> F:
     """
     Decorator that logs execution time and patches log records with original function metadata.

reme_ai/mem_agent/__init__.py

@@ -0,0 +1,9 @@
+"""Agent module providing chat operations."""
+
+from .simple_chat import SimpleChat
+from .stream_chat import StreamChat
+
+__all__ = [
+    "StreamChat",
+    "SimpleChat",
+]

reme_ai/mem_agent/simple_chat.py

@@ -0,0 +1,62 @@
+"""Simple chat agent for non-streaming conversations."""
+
+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
+
+
+@C.register_op()
+class SimpleChat(BaseOp):
+    """Simple chat agent that handles non-streaming conversations."""
+
+    def _build_tool_call(self) -> ToolCall:
+        return ToolCall(
+            **{
+                "description": "simple chat agent",
+                "parameters": {
+                    "type": "object",
+                    "properties": {
+                        "query": {
+                            "type": "string",
+                            "description": "query",
+                        },
+                        "messages": {
+                            "type": "array",
+                            "items": {
+                                "type": "object",
+                                "properties": {
+                                    "role": {
+                                        "type": "string",
+                                        "description": "role",
+                                    },
+                                    "content": {
+                                        "type": "string",
+                                        "description": "content",
+                                    },
+                                },
+                                "required": ["role", "content"],
+                            },
+                        },
+                    },
+                    "required": [],
+                },
+            },
+        )
+
+    async def execute(self):
+        if "query" in self.context:
+            messages = [
+                Message(role=Role.SYSTEM, content="You are a helpful assistant."),
+                Message(role=Role.USER, content=self.context.query),
+            ]
+        elif "messages" in self.context:
+            messages = [Message(**m) if isinstance(m, dict) else m for m in self.context.messages if m]
+        else:
+            raise ValueError("query or messages must be provided!")
+        logger.info(f"messages={messages}")
+        assistant_message = await self.llm.chat(messages=messages)
+        logger.info(f"assistant_message={assistant_message.simple_dump()}")
+        self.output = assistant_message.content

reme_ai/mem_agent/stream_chat.py

@@ -0,0 +1,64 @@
+"""Streaming chat agent for real-time conversation streaming."""
+
+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
+
+
+@C.register_op()
+class StreamChat(BaseOp):
+    """Streaming chat agent that handles real-time conversation streaming."""
+
+    def _build_tool_call(self) -> ToolCall:
+        return ToolCall(
+            **{
+                "description": "simple chat agent",
+                "parameters": {
+                    "type": "object",
+                    "properties": {
+                        "query": {
+                            "type": "string",
+                            "description": "query",
+                        },
+                        "messages": {
+                            "type": "array",
+                            "items": {
+                                "type": "object",
+                                "properties": {
+                                    "role": {
+                                        "type": "string",
+                                        "description": "role",
+                                    },
+                                    "content": {
+                                        "type": "string",
+                                        "description": "content",
+                                    },
+                                },
+                                "required": ["role", "content"],
+                            },
+                        },
+                    },
+                    "required": [],
+                },
+            },
+        )
+
+    async def execute(self):
+        """Execute streaming chat operation with query or messages."""
+        if "query" in self.context:
+            messages = [
+                Message(role=Role.SYSTEM, content="You are a helpful assistant."),
+                Message(role=Role.USER, content=self.context.query),
+            ]
+        elif "messages" in self.context:
+            messages = [Message(**m) if isinstance(m, dict) else m for m in self.context.messages if m]
+        else:
+            raise ValueError("query or messages must be provided!")
+        logger.info(f"messages={messages}")
+
+        async for stream_chunk in self.llm.stream_chat(messages):
+            if stream_chunk.chunk_type in [ChunkEnum.ANSWER, ChunkEnum.THINK, ChunkEnum.ERROR, ChunkEnum.TOOL]:
+                await self.context.add_stream_chunk(stream_chunk)

reme_ai/tool/__init__.py

@@ -0,0 +1,17 @@
+"""tool"""
+
+from . import execute
+from . import memory
+from . import search
+from .base_memory_tool import BaseMemoryTool
+from .mcp_tool import MCPTool
+from .think_tool import ThinkTool
+
+__all__ = [
+    "execute",
+    "memory",
+    "search",
+    "BaseMemoryTool",
+    "MCPTool",
+    "ThinkTool",
+]

reme_ai/tool/base_memory_tool.py

@@ -0,0 +1,102 @@
+"""Base class for memory tool"""
+
+from abc import ABCMeta
+from pathlib import Path
+
+from ..core.enumeration import MemoryType
+from ..core.op import BaseOp
+from ..core.schema import ToolCall, MemoryNode
+from ..core.utils import CacheHandler
+
+
+class BaseMemoryTool(BaseOp, metaclass=ABCMeta):
+    """Base class for memory tool"""
+
+    def __init__(
+        self,
+        enable_multiple: bool = True,
+        enable_thinking_params: bool = False,
+        meta_memory_path: str = "./meta_memory",
+        **kwargs,
+    ):
+        super().__init__(**kwargs)
+        self.enable_multiple: bool = enable_multiple
+        self.enable_thinking_params: bool = enable_thinking_params
+        self.meta_memory_path: str = meta_memory_path
+        self._meta_memory: CacheHandler | None = None
+
+    def _build_parameters(self) -> dict:
+        return {}
+
+    def _build_multiple_parameters(self) -> dict:
+        return {}
+
+    def _build_tool_call(self) -> ToolCall:
+        if self.enable_multiple:
+            parameters = self._build_multiple_parameters()
+        else:
+            parameters = self._build_parameters()
+
+        if self.enable_thinking_params and "thinking" not in parameters["properties"]:
+            parameters["properties"] = {
+                "thinking": {
+                    "type": "string",
+                    "description": "Your thinking and reasoning about how to fill in the parameters",
+                },
+                **parameters["properties"],
+            }
+            parameters["required"] = ["thinking", *parameters["required"]]
+
+        return ToolCall(
+            **{
+                "description": self.get_prompt("tool" + ("_multiple" if self.enable_multiple else "")),
+                "parameters": parameters,
+            },
+        )
+
+    @property
+    def meta_memory(self) -> CacheHandler:
+        """Get or create the meta memory cache handler."""
+        if self._meta_memory is None:
+            self._meta_memory = CacheHandler(Path(self.meta_memory_path) / self.vector_store.collection_name)
+        return self._meta_memory
+
+    @property
+    def memory_type(self) -> MemoryType:
+        """Get the memory type from context."""
+        return MemoryType(self.context.get("memory_type"))
+
+    @property
+    def memory_target(self) -> str:
+        """Get the memory target from context."""
+        return self.context.get("memory_target", "")
+
+    @property
+    def ref_memory_id(self) -> str:
+        """Get the reference memory ID from context."""
+        return self.context.get("ref_memory_id", "")
+
+    @property
+    def author(self) -> str:
+        """Get the author from context."""
+        return self.context.get("author", "")
+
+    def _build_memory_node(
+        self,
+        memory_content: str,
+        when_to_use: str = "",
+        metadata: dict | None = None,
+    ) -> MemoryNode:
+        """Build MemoryNode from content, when_to_use, and metadata.
+
+        This is a shared utility method for subclasses that need to create MemoryNode instances.
+        """
+        return MemoryNode(
+            memory_type=self.memory_type,
+            memory_target=self.memory_target,
+            when_to_use=when_to_use or "",
+            content=memory_content,
+            ref_memory_id=self.ref_memory_id,
+            author=self.author,
+            metadata=metadata or {},
+        )

reme_ai/tool/execute/__init__.py

@@ -0,0 +1,9 @@
+"""execute tool"""
+
+from .execute_code import ExecuteCode
+from .execute_shell import ExecuteShell
+
+__all__ = [
+    "ExecuteCode",
+    "ExecuteShell",
+]

reme_ai/tool/execute/execute_code.py

@@ -0,0 +1,43 @@
+"""Code execution tool for running Python code dynamically.
+
+This module provides an operation that can execute Python code strings
+and return the output or error messages.
+"""
+
+from ...core.context import C
+from ...core.op import BaseOp
+from ...core.schema import ToolCall
+
+from ...core.utils import exec_code
+
+
+@C.register_op()
+class ExecuteCode(BaseOp):
+    """Operation for executing Python code dynamically.
+
+    This operation takes Python code as input, executes it in a safe context,
+    and returns the output or any error messages that occur during execution.
+    """
+
+    def _build_tool_call(self) -> ToolCall:
+        return ToolCall(
+            **{
+                "description": self.get_prompt("tool"),
+                "parameters": {
+                    "type": "object",
+                    "properties": {
+                        "code": {
+                            "type": "string",
+                            "description": "code",
+                        },
+                    },
+                    "required": ["code"],
+                },
+            },
+        )
+
+    async def execute(self):
+        self.execute_sync()
+
+    def execute_sync(self):
+        self.output = exec_code(self.context.code)

reme_ai/tool/execute/execute_code.yaml

@@ -0,0 +1,5 @@
+tool: |
+  Execute python code can be used in scenarios such as analysis or calculation, and the final result can be printed using the `print` function.
+
+tool_zh: |
+  执行 Python 代码可用于分析或计算等场景，最终结果可以使用 print 函数输出。

reme_ai/tool/execute/execute_shell.py

@@ -0,0 +1,49 @@
+"""Shell command execution tool.
+
+This module provides an operation that can execute shell commands
+asynchronously and return the output, error, and exit code.
+"""
+
+from ...core.context import C
+from ...core.op import BaseOp
+from ...core.schema import ToolCall
+
+from ...core.utils import run_shell_command
+
+
+@C.register_op()
+class ExecuteShell(BaseOp):
+    """Operation for executing shell commands asynchronously.
+
+    This operation takes a shell command as input, executes it asynchronously,
+    and returns the stdout, stderr, and exit code in a formatted result.
+    """
+
+    def _build_tool_call(self) -> ToolCall:
+        return ToolCall(
+            **{
+                "description": self.get_prompt("tool"),
+                "parameters": {
+                    "type": "object",
+                    "properties": {
+                        "command": {
+                            "type": "string",
+                            "description": "command",
+                        },
+                    },
+                    "required": ["command"],
+                },
+            },
+        )
+
+    async def execute(self):
+        command: str = self.context.command
+        stdout, stderr, return_code = await run_shell_command(command)
+        result_parts = [
+            f"Command: {command}",
+            f"Output: {stdout if stdout else '(empty)'}",
+            f"Error: {stderr if stderr else '(none)'}",
+            f"Exit Code: {return_code if return_code is not None else '(none)'}",
+        ]
+
+        self.output = "\n".join(result_parts)

reme_ai/tool/execute/execute_shell.yaml

@@ -0,0 +1,7 @@
+tool: |
+  A tool capable of executing shell commands can use `pwd` to check the current location, `cd` to navigate to a new directory, `ls` to view the contents of a directory, and execute scripts.
+  Note that the starting directory is always the same each time the tool is invoked. If you need to perform multiple operations within a specific directory, you must include the full path in each command, for example: `cd aa/bb && bash xxx`.
+
+tool_zh: |
+  一个能够执行 Shell 命令的工具可以使用 pwd 查看当前所在位置，使用 cd 切换到新目录，使用 ls 查看目录内容，并可执行脚本。
+  请注意，每次调用该工具时，起始目录始终相同。如果你需要在某个特定目录中执行多个操作，必须在每条命令中包含完整路径，例如：cd aa/bb && bash xxx。

reme_ai/tool/mcp_tool.py

@@ -2,10 +2,10 @@
 
 from typing import List
 
-from ..context import C
-from ..op import BaseOp
-from ..schema import ToolCall
-from ..utils import MCPClient
+from ..core.context import C
+from ..core.op import BaseOp
+from ..core.schema import ToolCall
+from ..core.utils import MCPClient
 
 
 @C.register_op()

reme_ai/tool/memory/__init__.py

@@ -0,0 +1,27 @@
+"""Memory tool operations."""
+
+from .history.add_history_memory import AddHistoryMemory
+from .history.read_history_memory import ReadHistoryMemory
+from .identity.read_identity_memory import ReadIdentityMemory
+from .identity.update_identity_memory import UpdateIdentityMemory
+from .meta.add_meta_memory import AddMetaMemory
+from .meta.read_meta_memory import ReadMetaMemory
+from .vector.add_memory import AddMemory
+from .vector.add_summary_memory import AddSummaryMemory
+from .vector.delete_memory import DeleteMemory
+from .vector.update_memory import UpdateMemory
+from .vector.vector_retrieve_memory import VectorRetrieveMemory
+
+__all__ = [
+    "AddHistoryMemory",
+    "ReadHistoryMemory",
+    "ReadIdentityMemory",
+    "UpdateIdentityMemory",
+    "AddMetaMemory",
+    "ReadMetaMemory",
+    "AddMemory",
+    "AddSummaryMemory",
+    "DeleteMemory",
+    "UpdateMemory",
+    "VectorRetrieveMemory",
+]

reme_ai/tool/memory/history/add_history_memory.py

@@ -0,0 +1,111 @@
+"""Add history memory operation."""
+
+from loguru import logger
+
+from ...base_memory_tool import BaseMemoryTool
+from ....core.context import C
+from ....core.schema import MemoryNode
+
+
+@C.register_op()
+class AddHistoryMemory(BaseMemoryTool):
+    """Add history memory from conversation messages."""
+
+    def __init__(self, add_metadata: bool = True, **kwargs):
+        super().__init__(**kwargs)
+        self.add_metadata: bool = add_metadata
+
+    def _build_item_schema(self) -> tuple[dict, list[str]]:
+        properties = {
+            "messages": {
+                "type": "array",
+                "description": self.get_prompt("messages"),
+                "items": {"type": "object"},
+            },
+        }
+        required = ["messages"]
+
+        if self.add_metadata:
+            properties["metadata"] = {
+                "type": "object",
+                "description": self.get_prompt("metadata"),
+            }
+
+        return properties, required
+
+    def _build_parameters(self) -> dict:
+        properties, required = self._build_item_schema()
+        return {
+            "type": "object",
+            "properties": properties,
+            "required": required,
+        }
+
+    def _build_multiple_parameters(self) -> dict:
+        item_properties, required_fields = self._build_item_schema()
+        return {
+            "type": "object",
+            "properties": {
+                "histories": {
+                    "type": "array",
+                    "description": self.get_prompt("histories"),
+                    "items": {
+                        "type": "object",
+                        "properties": item_properties,
+                        "required": required_fields,
+                    },
+                },
+            },
+            "required": ["histories"],
+        }
+
+    def _format_messages(self, messages: list) -> str:
+        return "\n".join([f"{msg.get('role', 'unknown')}: {msg.get('content', '')}" for msg in messages])
+
+    def _extract_history_data(self, hist_dict: dict) -> tuple[list, dict]:
+        messages = hist_dict.get("messages", [])
+        metadata = hist_dict.get("metadata", {}) if self.add_metadata else {}
+        return messages, metadata
+
+    async def execute(self):
+        memory_nodes: list[MemoryNode] = []
+
+        if self.enable_multiple:
+            histories: list[dict] = self.context.get("histories", [])
+            if not histories:
+                self.output = "No histories provided for addition."
+                return
+
+            for hist in histories:
+                messages, metadata = self._extract_history_data(hist)
+                if not messages:
+                    logger.warning("Skipping history with empty messages")
+                    continue
+
+                memory_content = self._format_messages(messages)
+                memory_nodes.append(
+                    self._build_memory_node(memory_content, when_to_use="", metadata=metadata),
+                )
+        else:
+            messages, metadata = self._extract_history_data(self.context)
+            if not messages:
+                self.output = "No messages provided for addition."
+                return
+
+            memory_content = self._format_messages(messages)
+            memory_nodes.append(
+                self._build_memory_node(memory_content, when_to_use="", metadata=metadata),
+            )
+
+        if not memory_nodes:
+            self.output = "No valid histories provided for addition."
+            return
+
+        vector_nodes = [node.to_vector_node() for node in memory_nodes]
+        vector_ids: list[str] = [node.vector_id for node in vector_nodes]
+
+        await self.vector_store.delete(vector_ids=vector_ids)
+        await self.vector_store.insert(nodes=vector_nodes)
+
+        self.output = f"Successfully added {len(memory_nodes)} history memories to vector_store."
+        logger.info(self.output)

reme_ai/tool/memory/history/add_history_memory.yaml

@@ -0,0 +1,14 @@
+tool: |
+  Add history memory from conversation messages.
+
+tool_multiple: |
+  Add multiple history memories in a single operation.
+
+messages: |
+  List of message objects with 'role' and 'content' fields.
+
+metadata: |
+  Optional metadata (time, session_id, topic, etc.).
+
+histories: |
+  List of history objects, each with messages and optional metadata.

reme_ai/tool/memory/history/read_history_memory.py

@@ -0,0 +1,75 @@
+"""Read history memory operation."""
+
+from loguru import logger
+
+from ...base_memory_tool import BaseMemoryTool
+from ....core.context import C
+from ....core.schema import MemoryNode
+
+
+@C.register_op()
+class ReadHistoryMemory(BaseMemoryTool):
+    """Read history memories by IDs."""
+
+    def _build_parameters(self) -> dict:
+        return {
+            "type": "object",
+            "properties": {
+                "memory_id": {
+                    "type": "string",
+                    "description": self.get_prompt("memory_id"),
+                },
+            },
+            "required": ["memory_id"],
+        }
+
+    def _build_multiple_parameters(self) -> dict:
+        return {
+            "type": "object",
+            "properties": {
+                "memory_ids": {
+                    "type": "array",
+                    "description": self.get_prompt("memory_ids"),
+                    "items": {"type": "string"},
+                },
+            },
+            "required": ["memory_ids"],
+        }
+
+    async def execute(self):
+        if self.enable_multiple:
+            memory_ids: list[str] = self.context.get("memory_ids", [])
+        else:
+            memory_id = self.context.get("memory_id", "")
+            memory_ids: list[str] = [memory_id] if memory_id else []
+
+        memory_ids = [mid for mid in memory_ids if mid]
+
+        if not memory_ids:
+            self.output = "No valid history memory IDs provided for reading."
+            logger.warning(self.output)
+            return
+
+        nodes = await self.vector_store.search(
+            query="",
+            top_k=len(memory_ids),
+            filter_dict={"vector_id": memory_ids},
+        )
+
+        if not nodes:
+            self.output = "No history memories found with the provided IDs."
+            logger.warning(self.output)
+            return
+
+        memories: list[MemoryNode] = [MemoryNode.from_vector_node(n) for n in nodes]
+
+        output_lines = []
+        for memory in memories:
+            output_lines.append(f"Memory ID: {memory.vector_id}")
+            output_lines.append(f"Content:\n{memory.content}")
+            if memory.metadata:
+                output_lines.append(f"Metadata: {memory.metadata}")
+            output_lines.append("---")
+
+        self.output = "\n".join(output_lines)
+        logger.info(f"Successfully read {len(memories)} history memories.")

reme_ai/tool/memory/history/read_history_memory.yaml

@@ -0,0 +1,11 @@
+tool: |
+  Read history memory by ID.
+
+tool_multiple: |
+  Read multiple history memories by IDs.
+
+memory_id: |
+  Unique identifier of the history memory.
+
+memory_ids: |
+  List of unique identifiers of history memories.

reme_ai/tool/memory/identity/read_identity_memory.py

@@ -0,0 +1,33 @@
+"""Read identity memory operation."""
+
+from loguru import logger
+
+from ...base_memory_tool import BaseMemoryTool
+from ....core.context import C
+
+
+@C.register_op()
+class ReadIdentityMemory(BaseMemoryTool):
+    """Read identity memory for agent self-cognition."""
+
+    def __init__(self, **kwargs):
+        kwargs["enable_multiple"] = False
+        super().__init__(**kwargs)
+
+    def _build_parameters(self) -> dict:
+        return {
+            "type": "object",
+            "properties": {},
+            "required": [],
+        }
+
+    async def execute(self):
+        result = self.meta_memory.load("identity_memory")
+        identity_memory = result if result is not None else ""
+
+        if identity_memory:
+            self.output = f"Identity memory:\n{identity_memory}"
+            logger.info("Retrieved identity memory")
+        else:
+            self.output = "No identity memory found."
+            logger.info(self.output)

reme_ai/tool/memory/identity/read_identity_memory.yaml

@@ -0,0 +1,3 @@
+tool: |
+  Read the identity memory for the agent.
+  Retrieve self-cognition information such as identity, role, personality, or current state.

reme_ai/tool/memory/identity/update_identity_memory.py

@@ -0,0 +1,39 @@
+"""Update identity memory operation."""
+
+from loguru import logger
+
+from ...base_memory_tool import BaseMemoryTool
+from ....core.context import C
+
+
+@C.register_op()
+class UpdateIdentityMemory(BaseMemoryTool):
+    """Update identity memory for agent self-cognition."""
+
+    def __init__(self, **kwargs):
+        kwargs["enable_multiple"] = False
+        super().__init__(**kwargs)
+
+    def _build_parameters(self) -> dict:
+        return {
+            "type": "object",
+            "properties": {
+                "identity_memory": {
+                    "type": "string",
+                    "description": self.get_prompt("identity_memory"),
+                },
+            },
+            "required": ["identity_memory"],
+        }
+
+    async def execute(self):
+        identity_memory = self.context.get("identity_memory", "")
+
+        if not identity_memory:
+            self.output = "No valid identity memory provided for update."
+            logger.warning(self.output)
+            return
+
+        self.meta_memory.save("identity_memory", identity_memory)
+        self.output = "Successfully updated identity memory."
+        logger.info(self.output)

reme_ai/tool/memory/identity/update_identity_memory.yaml

@@ -0,0 +1,7 @@
+tool: |
+  Update the identity memory for the agent.
+  Store self-cognition information such as identity, role, personality, or current state.
+
+identity_memory: |
+  The identity memory content to store.
+  Should be a clear statement capturing the agent's self-cognition or current state.

reme_ai/tool/memory/meta/add_meta_memory.py

@@ -0,0 +1,121 @@
+"""Add meta memory operation for adding memory metadata."""
+
+import json
+
+from loguru import logger
+
+from ...base_memory_tool import BaseMemoryTool
+from ....core.context import C
+from ....core.enumeration import MemoryType
+
+
+@C.register_op()
+class AddMetaMemory(BaseMemoryTool):
+    """Add memory metadata (memory_type and memory_target) to meta storage.
+
+    Supports single/multiple addition modes via `enable_multiple` parameter.
+    """
+
+    def _build_item_schema(self) -> tuple[dict, list[str]]:
+        """Build shared schema properties and required fields for meta memory items.
+
+        Returns:
+            Tuple of (properties dict, required fields list).
+        """
+        properties = {
+            "memory_type": {
+                "type": "string",
+                "description": self.get_prompt("memory_type"),
+                "enum": [MemoryType.PERSONAL.value, MemoryType.PROCEDURAL.value],
+            },
+            "memory_target": {
+                "type": "string",
+                "description": self.get_prompt("memory_target"),
+            },
+        }
+        required = ["memory_type", "memory_target"]
+        return properties, required
+
+    def _build_parameters(self) -> dict:
+        """Build input schema for single meta memory addition."""
+        properties, required = self._build_item_schema()
+        return {
+            "type": "object",
+            "properties": properties,
+            "required": required,
+        }
+
+    def _build_multiple_parameters(self) -> dict:
+        """Build input schema for multiple meta memory addition."""
+        item_properties, required_fields = self._build_item_schema()
+        return {
+            "type": "object",
+            "properties": {
+                "meta_memories": {
+                    "type": "array",
+                    "description": self.get_prompt("meta_memories"),
+                    "items": {
+                        "type": "object",
+                        "properties": item_properties,
+                        "required": required_fields,
+                    },
+                },
+            },
+            "required": ["meta_memories"],
+        }
+
+    def _load_meta_memories(self) -> list[dict]:
+        """Load existing meta memories from cache."""
+        result = self.meta_memory.load("meta_memories")
+        return result if result is not None else []
+
+    def _save_meta_memories(self, memories: list[dict]) -> bool:
+        """Save meta memories to cache."""
+        return self.meta_memory.save("meta_memories", memories)
+
+    async def execute(self):
+        """Execute addition: load existing, merge with new, and save.
+
+        Duplicates (same memory_type and memory_target) are skipped.
+        """
+        existing_memories: list[dict] = self._load_meta_memories()
+        existing_set = {(m["memory_type"], m["memory_target"]) for m in existing_memories}
+
+        # Build new memories to add based on mode
+        new_memories: list[dict] = []
+        if self.enable_multiple:
+            meta_memories: list[dict] = self.context.get("meta_memories", [])
+            for mem in meta_memories:
+                memory_type = mem.get("memory_type", "")
+                memory_target = mem.get("memory_target", "")
+                if memory_type and (memory_type, memory_target) not in existing_set:
+                    new_memories.append(
+                        {
+                            "memory_type": memory_type,
+                            "memory_target": memory_target,
+                        },
+                    )
+                    existing_set.add((memory_type, memory_target))
+        else:
+            memory_type = self.context.get("memory_type", "")
+            memory_target = self.context.get("memory_target", "")
+            if memory_type and (memory_type, memory_target) not in existing_set:
+                new_memories.append(
+                    {
+                        "memory_type": memory_type,
+                        "memory_target": memory_target,
+                    },
+                )
+
+        if not new_memories:
+            self.output = "No new meta memories to add (all entries already exist or invalid)."
+            return
+
+        # Merge and save
+        all_memories = existing_memories + new_memories
+        self._save_meta_memories(all_memories)
+
+        # Format output
+        added_str = json.dumps(new_memories, ensure_ascii=False)
+        self.output = f"Successfully added {len(new_memories)} meta memory entries: {added_str}"
+        logger.info(self.output)

reme_ai/tool/memory/meta/add_meta_memory.yaml

@@ -0,0 +1,24 @@
+tool: |
+  Add a memory metadata entry to register a new memory type and target.
+  Use this tool to define what types of memories should be tracked, such as:
+  - Personal memories: "John", "Alice" (person-specific preferences and context)
+  - Procedural memories: "deployment_process", "code_review_steps" (how-to knowledge)
+
+tool_multiple: |
+  Add multiple memory metadata entries to register multiple memory types and targets at once.
+  Use this tool to define multiple memory tracking categories in a single operation.
+  Each entry specifies a memory_type and memory_target for organizing different memory domains.
+
+meta_memories: |
+  A list of memory metadata entries to add. Each entry contains memory_type and memory_target.
+
+memory_type: |
+  The type of memory to register. Valid values are: personal, procedural.
+  - personal: Person-specific memory storing preferences and context about specific individuals
+  - procedural: Procedural memory storing how-to knowledge and step-by-step processes
+
+memory_target: |
+  The target identifier for this memory category.
+  Examples:
+  - For personal memory: person's name (e.g., "John", "Alice")
+  - For procedural memory: process name (e.g., "deployment", "code_review")

reme_ai/tool/memory/meta/read_meta_memory.py

@@ -0,0 +1,110 @@
+"""Read meta memory operation for retrieving memory metadata."""
+
+from loguru import logger
+
+from ...base_memory_tool import BaseMemoryTool
+from ....core.context import C
+from ....core.enumeration import MemoryType
+
+
+@C.register_op()
+class ReadMetaMemory(BaseMemoryTool):
+    """Read memory metadata (memory_type and memory_target) from meta storage.
+
+    This operation reads stored memory metadata and optionally includes
+    TOOL and IDENTITY type memories.
+    """
+
+    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:
+        """Build input schema for reading meta memory.
+
+        No input parameters required for reading.
+        """
+        return {
+            "type": "object",
+            "properties": {},
+            "required": [],
+        }
+
+    def _load_meta_memories(self) -> list[dict[str, str]]:
+        """Load meta memories from cache and apply filters."""
+        result = self.meta_memory.load("meta_memories")
+        all_memories = result if result is not None else []
+
+        filtered_memories = []
+        for m in all_memories:
+            memory_type = MemoryType(m.get("memory_type"))
+            if memory_type in (MemoryType.PERSONAL, MemoryType.PROCEDURAL):
+                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(
+                {
+                    "memory_type": MemoryType.IDENTITY.value,
+                    "memory_target": "self",
+                },
+            )
+
+        return filtered_memories
+
+    def _format_memory_metadata(self, memories: list[dict[str, str]]) -> str:
+        """Format memory metadata into a readable string.
+
+        Args:
+            memories: List of memory metadata entries.
+
+        Returns:
+            str: Formatted memory metadata string.
+        """
+        if not memories:
+            return ""
+
+        lines = []
+        for memory in memories:
+            memory_type = memory["memory_type"]
+            memory_target = memory["memory_target"]
+            description = self.get_prompt(f"type_{memory_type}")
+            lines.append(f"- {memory_type}({memory_target}): {description}")
+
+        return "\n".join(lines)
+
+    async def execute(self):
+        """Execute the read meta memory operation.
+
+        Reads memory metadata from cache storage and formats output.
+        """
+        memories = self._load_meta_memories()
+
+        if memories:
+            formatted = self._format_memory_metadata(memories)
+            self.output = formatted
+            logger.info(f"Retrieved {len(memories)} meta memory entries")
+        else:
+            self.output = "No memory metadata found."
+            logger.info(self.output)

reme_ai/tool/memory/meta/read_meta_memory.yaml

@@ -0,0 +1,16 @@
+tool: |
+  Read the memory metadata registry to see what types of memories are being tracked.
+  Use this tool to retrieve all registered memory types and their targets.
+  This helps understand what memory categories are available for storing and retrieving information.
+
+type_identity: |
+  Self-cognition memory storing agent's identity, personality, and current state.
+
+type_personal: |
+  Person-specific memory storing preferences and context about specific individuals.
+
+type_procedural: |
+  Procedural memory storing how-to knowledge and step-by-step processes.
+
+type_tool: |
+  Tool memory storing tool usage patterns, success rates, token consumption, and latency.

reme_ai/tool/memory/vector/add_memory.py

@@ -0,0 +1,144 @@
+"""Add memory operation for vector store."""
+
+from loguru import logger
+
+from ...base_memory_tool import BaseMemoryTool
+from ....core.context import C
+from ....core.schema import MemoryNode
+
+
+@C.register_op()
+class AddMemory(BaseMemoryTool):
+    """Add memories to vector store with optional when_to_use and metadata.
+
+    Supports single/multiple addition modes via `enable_multiple` parameter.
+    """
+
+    def __init__(self, add_when_to_use: bool = False, add_metadata: bool = True, **kwargs):
+        """Initialize AddMemory.
+
+        Args:
+            add_when_to_use: Include when_to_use field for better retrieval.
+            add_metadata: Include metadata field for additional info.
+            **kwargs: Additional arguments for BaseMemoryTool.
+        """
+        super().__init__(**kwargs)
+        self.add_when_to_use: bool = add_when_to_use
+        self.add_metadata: bool = add_metadata
+
+    def _build_item_schema(self) -> tuple[dict, list[str]]:
+        """Build shared schema properties and required fields for memory items.
+
+        Returns:
+            Tuple of (properties dict, required fields list).
+        """
+        properties = {}
+        required = []
+
+        if self.add_when_to_use:
+            properties["when_to_use"] = {
+                "type": "string",
+                "description": self.get_prompt("when_to_use"),
+            }
+            required.append("when_to_use")
+
+        properties["memory_content"] = {
+            "type": "string",
+            "description": self.get_prompt("memory_content"),
+        }
+        required.append("memory_content")
+
+        if self.add_metadata:
+            properties["metadata"] = {
+                "type": "object",
+                "description": self.get_prompt("metadata"),
+            }
+
+        return properties, required
+
+    def _build_parameters(self) -> dict:
+        """Build input schema for single memory addition."""
+        properties, required = self._build_item_schema()
+        return {
+            "type": "object",
+            "properties": properties,
+            "required": required,
+        }
+
+    def _build_multiple_parameters(self) -> dict:
+        """Build input schema for multiple memory addition."""
+        item_properties, required_fields = self._build_item_schema()
+        return {
+            "type": "object",
+            "properties": {
+                "memories": {
+                    "type": "array",
+                    "description": self.get_prompt("memories"),
+                    "items": {
+                        "type": "object",
+                        "properties": item_properties,
+                        "required": required_fields,
+                    },
+                },
+            },
+            "required": ["memories"],
+        }
+
+    def _extract_memory_data(self, mem_dict: dict) -> tuple[str, str, dict]:
+        """Extract memory data from a dictionary with proper defaults.
+
+        Args:
+            mem_dict: Dictionary containing memory fields.
+
+        Returns:
+            Tuple of (memory_content, when_to_use, metadata).
+        """
+        memory_content = mem_dict.get("memory_content", "")
+        when_to_use = mem_dict.get("when_to_use", "") if self.add_when_to_use else ""
+        metadata = mem_dict.get("metadata", {}) if self.add_metadata else {}
+        return memory_content, when_to_use, metadata
+
+    async def execute(self):
+        """Execute addition: delete existing IDs (upsert), then insert new memories."""
+        memory_nodes: list[MemoryNode] = []
+
+        if self.enable_multiple:
+            memories: list[dict] = self.context.get("memories", [])
+            if not memories:
+                self.output = "No memories provided for addition."
+                return
+
+            for mem in memories:
+                memory_content, when_to_use, metadata = self._extract_memory_data(mem)
+                if not memory_content:
+                    logger.warning("Skipping memory with empty content")
+                    continue
+
+                memory_nodes.append(
+                    self._build_memory_node(memory_content, when_to_use, metadata),
+                )
+
+        else:
+            memory_content, when_to_use, metadata = self._extract_memory_data(self.context)
+            if not memory_content:
+                self.output = "No memory content provided for addition."
+                return
+
+            memory_nodes.append(
+                self._build_memory_node(memory_content, when_to_use, metadata),
+            )
+
+        if not memory_nodes:
+            self.output = "No valid memories provided for addition."
+            return
+
+        # Convert to VectorNodes and collect IDs
+        vector_nodes = [node.to_vector_node() for node in memory_nodes]
+        vector_ids: list[str] = [node.vector_id for node in vector_nodes]
+
+        # Delete existing IDs (upsert behavior), then insert
+        await self.vector_store.delete(vector_ids=vector_ids)
+        await self.vector_store.insert(nodes=vector_nodes)
+
+        self.output = f"Successfully added {len(memory_nodes)} memories to vector_store."
+        logger.info(self.output)

reme_ai/tool/memory/vector/add_memory.yaml

@@ -0,0 +1,37 @@
+tool: |
+  Add a memory to the vector store for future retrieval.
+  Use this tool to store important information that should be remembered, such as:
+  - Meta information: "I am very happy"
+  - Personal preferences: "John prefers dark mode", "Alice works in PST timezone"
+  - Procedural knowledge: "To deploy, run build then push", "Always validate input before processing"
+  - Tool usage tips: "search_tool works best with short queries", "Use cache tool for frequently accessed data"
+
+tool_multiple: |
+  Add multiple memories to the vector store for future retrieval.
+  Use this tool to store multiple pieces of important information in a single operation.
+  Each memory can include when_to_use conditions and metadata for better organization and retrieval.
+  Examples: storing multiple user preferences, multiple procedural steps, or multiple tool usage tips.
+
+when_to_use: |
+  Optional condition description for when to retrieve this memory.
+  This field is used for vector embedding to improve retrieval accuracy by providing contextual information.
+  Examples:
+  - "when user asks about authentication"
+  - "when deploying to production"
+  - "when using search_tool"
+  - "when handling error cases"
+
+memory_content: |
+  The content of the memory to store.
+  Should be a clear, concise statement that captures the information to remember.
+  Keep it focused on a single piece of information for better retrieval accuracy.
+
+metadata: |
+  Optional metadata for the memory, providing additional context. Can include:
+  - time: The timestamp or date associated with the memory (e.g., "2025-01-06 10:30:00")
+  - source: Where this information came from (e.g., "user_input", "documentation", "observation")
+  - tags: List of tags for categorization (e.g., ["authentication", "security"])
+  - Any other custom key-value pairs relevant to the memory
+
+memories: |
+  A list of memory objects to store.

reme_ai/tool/memory/vector/add_summary_memory.py

@@ -0,0 +1,67 @@
+"""Add summary memory operation for vector store."""
+
+from loguru import logger
+
+from .add_memory import AddMemory
+from ....core.context import C
+
+
+@C.register_op()
+class AddSummaryMemory(AddMemory):
+    """Add LLM-summarized memories to vector store.
+
+    Differences from AddMemory:
+    - Single memory mode only (enable_multiple=False)
+    - Uses 'summary_memory' parameter instead of 'memory_content'
+    - No when_to_use field (add_when_to_use=False)
+    """
+
+    def __init__(
+        self,
+        add_metadata: bool = True,
+        **kwargs,
+    ):
+        """Initialize AddSummaryMemory.
+
+        Args:
+            add_metadata: Include metadata field for additional info.
+            **kwargs: Additional arguments for AddMemory.
+        """
+        # Force single mode and disable when_to_use
+        kwargs["enable_multiple"] = False
+        kwargs["add_when_to_use"] = False
+        super().__init__(add_metadata=add_metadata, **kwargs)
+
+    def _build_parameters(self) -> dict:
+        """Build input schema for summary memory addition."""
+        properties = {
+            "summary_memory": {
+                "type": "string",
+                "description": self.get_prompt("summary_memory"),
+            },
+        }
+        required = ["summary_memory"]
+
+        if self.add_metadata:
+            properties["metadata"] = {
+                "type": "object",
+                "description": self.get_prompt("metadata"),
+            }
+
+        return {
+            "type": "object",
+            "properties": properties,
+            "required": required,
+        }
+
+    async def execute(self):
+        """Execute addition: map summary_memory to memory_content and call parent."""
+        # Map summary_memory to memory_content
+        summary_memory = self.context.get("summary_memory", "")
+        if not summary_memory:
+            self.output = "No summary memory content provided for addition."
+            logger.warning(self.output)
+            return
+
+        self.context["memory_content"] = summary_memory
+        await super().execute()

reme_ai/tool/memory/vector/add_summary_memory.yaml

@@ -0,0 +1,27 @@
+tool: |
+  Add a summary memory to the vector store for future retrieval.
+  Use this tool to store a summarized version of the provided context.
+  The LLM should first summarize the context, then call this tool with the summarized content.
+
+  This tool is specifically designed for storing summaries of conversations, events, or information
+  that has been condensed from a larger context. Examples:
+  - Summarizing a long conversation: "User discussed project requirements for a web app with authentication"
+  - Summarizing a decision: "Team decided to use PostgreSQL for the database after evaluating options"
+  - Summarizing an event: "Successfully deployed version 2.0 to production with new features"
+
+summary_memory: |
+  The summarized content to store as memory.
+  Should be a clear, concise summary that captures the key information from the context.
+  Keep it focused and informative - aim for 1-3 sentences that convey the essential points.
+  Examples:
+  - "User prefers Python for backend development and has experience with FastAPI framework"
+  - "Project deadline is January 15th, requires authentication, payment integration, and admin dashboard"
+  - "Bug in user registration was caused by missing email validation, fixed by adding regex check"
+
+metadata: |
+  Optional metadata for the memory, providing additional context. Can include:
+  - time: The timestamp or date associated with the memory (e.g., "2025-01-06 10:30:00")
+  - source: Where this information came from (e.g., "conversation", "meeting", "observation")
+  - tags: List of tags for categorization (e.g., ["project", "decision"])
+  - summary_type: Type of summary (e.g., "conversation", "decision", "event", "task")
+  - Any other custom key-value pairs relevant to the memory

reme_ai/tool/memory/vector/delete_memory.py

@@ -0,0 +1,60 @@
+"""Delete memory operation for vector store."""
+
+from loguru import logger
+
+from ...base_memory_tool import BaseMemoryTool
+from ....core.context import C
+
+
+@C.register_op()
+class DeleteMemory(BaseMemoryTool):
+    """Delete memories from vector store by IDs.
+
+    Supports single/multiple deletion modes via `enable_multiple` parameter.
+    """
+
+    def _build_parameters(self) -> dict:
+        """Build input schema for single memory deletion."""
+        return {
+            "type": "object",
+            "properties": {
+                "memory_id": {
+                    "type": "string",
+                    "description": self.get_prompt("memory_id"),
+                },
+            },
+            "required": ["memory_id"],
+        }
+
+    def _build_multiple_parameters(self) -> dict:
+        """Build input schema for multiple memory deletion."""
+        return {
+            "type": "object",
+            "properties": {
+                "memory_ids": {
+                    "type": "array",
+                    "description": self.get_prompt("memory_ids"),
+                    "items": {"type": "string"},
+                },
+            },
+            "required": ["memory_ids"],
+        }
+
+    async def execute(self):
+        """Execute deletion: remove memories from vector store by IDs."""
+        if self.enable_multiple:
+            memory_ids = self.context.get("memory_ids", [])
+        else:
+            single_id = self.context.get("memory_id", "")
+            memory_ids = [single_id] if single_id else []
+
+        # Filter out empty IDs
+        memory_ids = [mid for mid in memory_ids if mid]
+
+        if not memory_ids:
+            self.output = "No valid memory IDs provided for deletion."
+            return
+
+        await self.vector_store.delete(vector_ids=memory_ids)
+        self.output = f"Successfully deleted {len(memory_ids)} memories from vector_store."
+        logger.info(self.output)

reme_ai/tool/memory/vector/delete_memory.yaml

@@ -0,0 +1,25 @@
+tool: |
+  Delete a memory from the vector store using its unique ID.
+  Use this tool when:
+  - The user explicitly requests to remove or forget information
+  - A memory is identified as outdated, incorrect, or no longer relevant
+  - Information needs to be removed for privacy or compliance reasons
+  - Duplicate or conflicting memories need to be cleaned up
+  Memory ID can be obtained from previous memory retrieval results.
+
+tool_multiple: |
+  Delete multiple memories from the vector store using their unique IDs.
+  Use this tool for batch deletion when:
+  - The user explicitly requests to remove or forget multiple pieces of information
+  - Multiple memories are identified as outdated, incorrect, or no longer relevant
+  - Bulk cleanup of information is needed for privacy or compliance reasons
+  - Multiple duplicate or conflicting memories need to be removed
+  Memory IDs can be obtained from previous memory retrieval results.
+
+memory_id: |
+  The unique identifier (memory_id) of the memory to delete.
+  This ID is returned when memories are retrieved or added.
+
+memory_ids: |
+  A list of unique identifiers (memory_ids) of the memories to delete.
+  Each ID should be a valid memory_id obtained from previous operations.

reme_ai/tool/memory/vector/update_memory.py

@@ -0,0 +1,153 @@
+"""Update memory operation for vector store."""
+
+from loguru import logger
+
+from ...base_memory_tool import BaseMemoryTool
+from ....core.context import C
+from ....core.schema import MemoryNode
+
+
+@C.register_op()
+class UpdateMemory(BaseMemoryTool):
+    """Update memories by deleting old ones and inserting new ones.
+
+    Supports single/multiple update modes via `enable_multiple` parameter.
+    """
+
+    def __init__(
+        self,
+        add_when_to_use: bool = False,
+        add_metadata: bool = True,
+        **kwargs,
+    ):
+        """Initialize UpdateMemory.
+
+        Args:
+            add_when_to_use: Include when_to_use field for better retrieval.
+            add_metadata: Include metadata field for additional info.
+            **kwargs: Additional arguments for BaseMemoryTool.
+        """
+        super().__init__(**kwargs)
+        self.add_when_to_use: bool = add_when_to_use
+        self.add_metadata: bool = add_metadata
+
+    def _build_item_schema(self) -> tuple[dict, list[str]]:
+        """Build shared schema properties and required fields for memory items.
+
+        Returns:
+            Tuple of (properties dict, required fields list).
+        """
+        properties = {
+            "memory_id": {
+                "type": "string",
+                "description": self.get_prompt("memory_id"),
+            },
+        }
+        required = ["memory_id"]
+
+        if self.add_when_to_use:
+            properties["when_to_use"] = {
+                "type": "string",
+                "description": self.get_prompt("when_to_use"),
+            }
+
+        properties["memory_content"] = {
+            "type": "string",
+            "description": self.get_prompt("memory_content"),
+        }
+        required.append("memory_content")
+
+        if self.add_metadata:
+            properties["metadata"] = {
+                "type": "object",
+                "description": self.get_prompt("metadata"),
+            }
+
+        return properties, required
+
+    def _build_parameters(self) -> dict:
+        """Build input schema for single memory update."""
+        properties, required = self._build_item_schema()
+        return {
+            "type": "object",
+            "properties": properties,
+            "required": required,
+        }
+
+    def _build_multiple_parameters(self) -> dict:
+        """Build input schema for multiple memory update."""
+        item_properties, required_fields = self._build_item_schema()
+        return {
+            "type": "object",
+            "properties": {
+                "memories": {
+                    "type": "array",
+                    "description": self.get_prompt("memories"),
+                    "items": {
+                        "type": "object",
+                        "properties": item_properties,
+                        "required": required_fields,
+                    },
+                },
+            },
+            "required": ["memories"],
+        }
+
+    def _extract_memory_data(self, mem_dict: dict) -> tuple[str, str, str, dict]:
+        """Extract memory update data from a dictionary with proper defaults.
+
+        Args:
+            mem_dict: Dictionary containing memory fields.
+
+        Returns:
+            Tuple of (memory_id, memory_content, when_to_use, metadata).
+        """
+        memory_id = mem_dict.get("memory_id", "")
+        memory_content = mem_dict.get("memory_content", "")
+        when_to_use = mem_dict.get("when_to_use", "") if self.add_when_to_use else ""
+        metadata = mem_dict.get("metadata", {}) if self.add_metadata else {}
+        return memory_id, memory_content, when_to_use, metadata
+
+    async def execute(self):
+        """Execute update: delete old memories by ID, insert new ones with updated content."""
+        # Collect old IDs to delete and new nodes to insert
+        old_memory_ids: list[str] = []
+        new_memory_nodes: list[MemoryNode] = []
+
+        if self.enable_multiple:
+            memories: list[dict] = self.context.get("memories", [])
+            if not memories:
+                self.output = "No memories provided for update."
+                return
+
+            for mem in memories:
+                memory_id, memory_content, when_to_use, metadata = self._extract_memory_data(mem)
+                if not memory_id or not memory_content:
+                    logger.warning(f"Skipping memory with missing id or content: {mem}")
+                    continue
+                old_memory_ids.append(memory_id)
+                new_memory_nodes.append(self._build_memory_node(memory_content, when_to_use, metadata))
+
+        else:
+            memory_id, memory_content, when_to_use, metadata = self._extract_memory_data(self.context)
+            if not memory_id or not memory_content:
+                self.output = "No memory ID or content provided for update."
+                return
+            old_memory_ids.append(memory_id)
+            new_memory_nodes.append(self._build_memory_node(memory_content, when_to_use, metadata))
+
+        if not old_memory_ids or not new_memory_nodes:
+            self.output = "No valid memories provided for update."
+            return
+
+        # Convert to VectorNodes and collect IDs
+        vector_nodes = [node.to_vector_node() for node in new_memory_nodes]
+        new_vector_ids = [node.vector_id for node in vector_nodes]
+
+        # Delete old and duplicate new IDs (upsert behavior)
+        all_ids_to_delete = list(set(old_memory_ids + new_vector_ids))
+        await self.vector_store.delete(vector_ids=all_ids_to_delete)
+        await self.vector_store.insert(nodes=vector_nodes)
+
+        self.output = f"Update: deleted {len(old_memory_ids)} old memories, added {len(new_memory_nodes)} new memories."
+        logger.info(self.output)

reme_ai/tool/memory/vector/update_memory.yaml

@@ -0,0 +1,45 @@
+tool: |
+  Update a memory in the vector store by replacing the old memory with new content.
+  Use this tool when:
+  - The user wants to modify or correct existing information
+  - A memory needs to be updated with new details while keeping its relevance
+  - Information has changed and the old memory is no longer accurate
+  - You need to refine or improve the clarity of stored information
+  Memory ID can be obtained from previous memory retrieval results.
+
+tool_multiple: |
+  Update multiple memories in the vector store by replacing old memories with new content.
+  Use this tool for batch updates when:
+  - The user wants to modify or correct multiple pieces of existing information
+  - Multiple memories need to be updated with new details while keeping their relevance
+  - Information has changed across multiple memories
+  - You need to refine or improve multiple stored memories at once
+  Memory IDs can be obtained from previous memory retrieval results.
+
+memory_id: |
+  The unique identifier (memory_id) of the old memory to be replaced.
+  This ID is returned when memories are retrieved or added.
+
+when_to_use: |
+  Optional condition description for when to retrieve this memory.
+  This field is used for vector embedding to improve retrieval accuracy by providing contextual information.
+  Examples:
+  - "when user asks about authentication"
+  - "when deploying to production"
+  - "when using search_tool"
+  - "when handling error cases"
+
+memory_content: |
+  The new content of the memory to store.
+  Should be a clear, concise statement that captures the updated information to remember.
+  Keep it focused on a single piece of information for better retrieval accuracy.
+
+metadata: |
+  Optional metadata for the new memory, providing additional context. Can include:
+  - time: The timestamp or date associated with the memory (e.g., "2025-01-06 10:30:00")
+  - source: Where this information came from (e.g., "user_input", "documentation", "observation")
+  - tags: List of tags for categorization (e.g., ["authentication", "security"])
+  - Any other custom key-value pairs relevant to the memory
+
+memories: |
+  A list of memory update objects.

reme_ai/tool/memory/vector/vector_retrieve_memory.py

@@ -0,0 +1,212 @@
+"""Vector-based memory retrieval using semantic similarity search."""
+
+from loguru import logger
+
+from ...base_memory_tool import BaseMemoryTool
+from ....core.context import C
+from ....core.enumeration import MemoryType
+from ....core.schema import MemoryNode, VectorNode
+from ....core.utils import deduplicate_memories
+
+
+@C.register_op()
+class VectorRetrieveMemory(BaseMemoryTool):
+    """Retrieve memories using vector similarity search.
+
+    Supports single/multiple query modes via `enable_multiple` parameter.
+    When `add_memory_type_target` is False, memory_type/memory_target are from context.
+    """
+
+    def __init__(
+        self,
+        enable_summary_memory: bool = False,
+        add_memory_type_target: bool = False,
+        top_k: int = 10,
+        **kwargs,
+    ):
+        """Initialize VectorRetrieveMemory.
+
+        Args:
+            enable_summary_memory: Include summary memories in results.
+            add_memory_type_target: Include memory_type/memory_target in schema (else from context).
+            top_k: Max memories to retrieve per query.
+            **kwargs: Additional args for BaseMemoryTool.
+        """
+        super().__init__(**kwargs)
+        self.enable_summary_memory: bool = enable_summary_memory
+        self.add_memory_type_target: bool = add_memory_type_target
+        self.top_k: int = top_k
+
+    def _build_query_schema(self) -> tuple[dict, list[str]]:
+        """Build shared schema properties and required fields for query items.
+
+        Returns:
+            Tuple of (properties dict, required fields list).
+        """
+        properties = {}
+        required = []
+
+        if self.add_memory_type_target:
+            properties["memory_type"] = {
+                "type": "string",
+                "description": self.get_prompt("memory_type"),
+                "enum": [
+                    MemoryType.IDENTITY.value,
+                    MemoryType.PERSONAL.value,
+                    MemoryType.PROCEDURAL.value,
+                    MemoryType.TOOL.value,
+                ],
+            }
+            properties["memory_target"] = {
+                "type": "string",
+                "description": self.get_prompt("memory_target"),
+            }
+            required.extend(["memory_type", "memory_target"])
+
+        properties["query"] = {
+            "type": "string",
+            "description": self.get_prompt("query"),
+        }
+        required.append("query")
+
+        return properties, required
+
+    def _build_parameters(self) -> dict:
+        """Build input schema for single query mode.
+
+        Returns:
+            Schema with memory_type/memory_target/query (if add_memory_type_target) or query only.
+        """
+        properties, required = self._build_query_schema()
+        return {
+            "type": "object",
+            "properties": properties,
+            "required": required,
+        }
+
+    def _build_multiple_parameters(self) -> dict:
+        """Build input schema for multiple query mode.
+
+        Returns:
+            Schema with query_items array. Each item has memory_type/memory_target/query
+            (if add_memory_type_target) or query only.
+        """
+        item_properties, item_required = self._build_query_schema()
+        return {
+            "type": "object",
+            "properties": {
+                "query_items": {
+                    "type": "array",
+                    "description": self.get_prompt("query_items"),
+                    "items": {
+                        "type": "object",
+                        "properties": item_properties,
+                        "required": item_required,
+                    },
+                },
+            },
+            "required": ["query_items"],
+        }
+
+    async def _retrieve_by_query(
+        self,
+        memory_type: str,
+        memory_target: str,
+        query: str,
+    ) -> list[MemoryNode]:
+        """Retrieve memories by query using vector similarity search.
+
+        Args:
+            memory_type: Memory type to search.
+            memory_target: Memory target to search.
+            query: Query string for similarity search.
+
+        Returns:
+            List of matching memories.
+        """
+        memory_type_list = [MemoryType(memory_type)]
+        if self.enable_summary_memory:
+            memory_type_list.append(MemoryType.SUMMARY)
+
+        filter_dict = {
+            "memory_type": [mt.value for mt in memory_type_list],
+            "memory_target": [memory_target],
+        }
+
+        nodes: list[VectorNode] = await self.vector_store.search(
+            query=query,
+            top_k=self.top_k,
+            filter_dict=filter_dict,
+        )
+
+        memory_nodes: list[MemoryNode] = [MemoryNode.from_vector_node(n) for n in nodes]
+
+        # Filter TOOL memories: keep only if when_to_use matches query (tool name)
+        filtered_memory_nodes = [
+            m for m in memory_nodes if not (m.memory_type == MemoryType.TOOL and m.when_to_use != query)
+        ]
+
+        return filtered_memory_nodes
+
+    async def execute(self):
+        """Execute memory retrieval based on query texts.
+
+        Handles single/multiple query modes. When add_memory_type_target is False,
+        memory_type/memory_target are from context. Outputs formatted results or error message.
+        """
+        default_memory_type: str = self.context.get("memory_type", "")
+        default_memory_target: str = self.context.get("memory_target", "")
+
+        # Normalize to list of query items
+        if self.enable_multiple:
+            query_items: list[dict] = self.context.get("query_items", [])
+            if not query_items:
+                self.output = "No query items provided for retrieval."
+                return
+        else:
+            query = self.context.get("query", "")
+            if not query:
+                self.output = "No query provided for retrieval."
+                return
+
+            query_items = [
+                {
+                    "memory_type": default_memory_type,
+                    "memory_target": default_memory_target,
+                    "query": query,
+                },
+            ]
+
+        # Filter out items without query text
+        query_items = [item for item in query_items if item.get("query")]
+
+        if not query_items:
+            self.output = "No valid query texts provided for retrieval."
+            return
+
+        # Retrieve memories for all queries
+        memories: list[MemoryNode] = []
+        for item in query_items:
+            memory_type = item.get("memory_type") or default_memory_type
+            memory_target = item.get("memory_target") or default_memory_target
+
+            if not memory_type or not memory_target:
+                logger.warning(f"Skipping query with missing memory_type or memory_target: {item}")
+                continue
+
+            retrieved = await self._retrieve_by_query(
+                memory_type=memory_type,
+                memory_target=memory_target,
+                query=item["query"],
+            )
+            memories.extend(retrieved)
+
+        # Deduplicate and format output
+        memories = deduplicate_memories(memories)
+
+        if not memories:
+            self.output = "No memories found matching the query."
+        else:
+            self.output = "\n".join([m.format_memory() for m in memories])
+
+        logger.info(f"Retrieved {len(memories)} memories")

reme_ai/tool/memory/vector/vector_retrieve_memory.yaml

@@ -0,0 +1,31 @@
+tool: |
+  Retrieve memories from the memory store using vector similarity search.
+  Use this tool to find relevant memories based on semantic similarity to the query.
+  The search returns the most relevant memories ranked by similarity score.
+
+tool_multiple: |
+  Retrieve memories from the memory store using multiple queries with vector similarity search.
+  Use this tool to find relevant memories based on semantic similarity to multiple queries.
+  This is useful when you need to search for different types of information in a single operation.
+  The search returns the most relevant memories ranked by similarity score for each query.
+
+memory_type: |
+  The type of memory to search for. Must be one of:
+  - "identity": Information about the AI agent's identity, role, or characteristics
+  - "personal": Information about users, their preferences, or personal details
+  - "procedural": Step-by-step instructions, workflows, or how-to knowledge
+  - "tool": Tool usage tips, examples, and best practices
+
+memory_target: |
+  The target of the memory to search within.
+  - For "personal" memory: the person's name or identifier (e.g., "john", "alice")
+  - For "procedural" memory: the process or task name (e.g., "deployment", "authentication")
+  - For "tool" memory: the tool name (e.g., "search_tool", "calculator")
+  - For "identity" memory: typically "self" or the agent's identifier
+
+query: |
+  The query text for vector similarity search.
+  Use descriptive queries that capture the semantic meaning of what you're looking for.
+
+query_items: |
+  A list of query items for vector similarity search.

reme_ai/tool/search/__init__.py

@@ -0,0 +1,11 @@
+"""search tool"""
+
+from .dashscope_search import DashscopeSearch
+from .mock_search import MockSearch
+from .tavily_search import TavilySearch
+
+__all__ = [
+    "DashscopeSearch",
+    "MockSearch",
+    "TavilySearch",
+]

reme_ai/tool/search/dashscope_search.py

@@ -0,0 +1,111 @@
+"""Dashscope web search tool.
+
+This module provides an operation that uses Alibaba Cloud's Dashscope API
+to perform web searches with various search strategies.
+"""
+
+import os
+from typing import Literal
+
+from loguru import logger
+
+from ...core.context import C
+from ...core.op import BaseOp
+from ...core.schema import ToolCall
+
+
+@C.register_op()
+class DashscopeSearch(BaseOp):
+    """Operation for performing web searches using Dashscope API.
+
+    This operation uses Alibaba Cloud's Dashscope service to search the web
+    with support for different search strategies (turbo, max, agent) and
+    optional role-based prompting.
+    """
+
+    def __init__(
+        self,
+        model: str = "qwen-plus",  # qwen-flash
+        search_strategy: Literal["turbo", "max", "agent"] = "turbo",  # agent only for qwen3-max
+        enable_role_prompt: bool = True,
+        **kwargs,
+    ):
+
+        super().__init__(**kwargs)
+        self.model: str = model
+        self.search_strategy: Literal["turbo", "max", "agent"] = search_strategy
+        self.enable_role_prompt: bool = enable_role_prompt
+
+        # see ref: https://help.aliyun.com/zh/model-studio/web-search
+        self.api_key = os.getenv("DASHSCOPE_API_KEY", "")
+
+    def _build_tool_call(self) -> ToolCall:
+        return ToolCall(
+            **{
+                "description": self.get_prompt("tool"),
+                "parameters": {
+                    "type": "object",
+                    "properties": {
+                        "query": {
+                            "type": "string",
+                            "description": "query",
+                        },
+                    },
+                    "required": ["query"],
+                },
+            },
+        )
+
+    async def execute(self):
+        query: str = self.context.query
+        if self.enable_cache:
+            cached_result = self.cache.load(query)
+            if cached_result:
+                self.output = cached_result["response_content"]
+                return
+
+        if self.enable_role_prompt:
+            user_query = self.prompt_format("role_prompt", query=query)
+        else:
+            user_query = query
+        logger.info(f"user_query={user_query}")
+        messages: list = [{"role": "user", "content": user_query}]
+
+        import dashscope
+
+        response = await dashscope.AioGeneration.call(
+            api_key=self.api_key,
+            model=self.model,
+            messages=messages,
+            enable_search=True,
+            search_options={
+                "forced_search": True,
+                "enable_source": True,
+                "enable_citation": False,
+                "search_strategy": self.search_strategy,
+            },
+            result_format="message",
+        )
+
+        search_results = []
+        response_content = ""
+
+        if response.output:
+            if response.output.search_info:
+                search_results = response.output.search_info.get("search_results", [])
+
+            if response.output.choices and len(response.output.choices) > 0:
+                response_content = response.output.choices[0].message.content
+
+        final_result = {
+            "query": query,
+            "search_results": search_results,
+            "response_content": response_content,
+            "model": self.model,
+            "search_strategy": self.search_strategy,
+        }
+
+        if self.enable_cache:
+            self.cache.save(query, final_result, expire_hours=self.cache_expire_hours)
+
+        self.output = final_result["response_content"]

reme_ai/tool/search/dashscope_search.yaml

@@ -0,0 +1,20 @@
+tool: |
+  Use search keywords to retrieve relevant information from the internet.
+  If you have multiple keywords, please call this tool separately for each one.
+
+tool_zh: |
+  使用搜索关键词从互联网检索相关信息。如果您有多个关键词，请分别为每个关键词单独调用此工具。
+
+role_prompt: |
+  # user's question
+  {query}
+
+  # task
+  Extract the original content related to the user's query directly from the context, maintain accuracy, and avoid excessive processing.
+
+role_prompt_zh: |
+  # 用户问题
+  {query}
+
+  # task
+  直接从上下文中提取与用户问题相关的原始内容，保持准确性，避免过度处理。

reme_ai/tool/search/mock_search.py

@@ -0,0 +1,64 @@
+"""Mock search tool for testing purposes.
+
+This module provides a mock search operation that generates simulated
+search results using an LLM, useful for testing without making actual API calls.
+"""
+
+import json
+import random
+
+from loguru import logger
+
+from ...core.context import C
+from ...core.enumeration import Role
+from ...core.op import BaseOp
+from ...core.schema import ToolCall, Message
+from ...core.utils import extract_content
+
+
+@C.register_op()
+class MockSearch(BaseOp):
+    """Operation for generating mock search results.
+
+    This operation generates simulated search results using an LLM,
+    useful for testing and development without requiring actual search API access.
+    """
+
+    def _build_tool_call(self) -> ToolCall:
+        return ToolCall(
+            **{
+                "description": self.get_prompt("tool"),
+                "parameters": {
+                    "type": "object",
+                    "properties": {
+                        "query": {
+                            "type": "string",
+                            "description": "query",
+                        },
+                    },
+                    "required": ["query"],
+                },
+            },
+        )
+
+    async def execute(self):
+        query: str = self.context.query
+        num_results: int = random.randint(0, 5)
+        messages = [
+            Message(
+                role=Role.SYSTEM,
+                content="You are a helpful assistant that generates realistic search results in JSON format.",
+            ),
+            Message(
+                role=Role.USER,
+                content=self.prompt_format("mock_search_prompt", query=query, num_results=num_results),
+            ),
+        ]
+
+        logger.info(f"messages={messages}")
+
+        def callback_fn(message: Message):
+            return extract_content(message.content, "json")
+
+        search_results: str = await self.llm.chat(messages=messages, callback_fn=callback_fn)
+        self.output = json.dumps(search_results, ensure_ascii=False, indent=2)

reme_ai/tool/search/mock_search.yaml

@@ -0,0 +1,81 @@
+tool: |
+  Use search keywords to retrieve relevant information from the internet.
+  If you have multiple keywords, please call this tool separately for each one.
+
+tool_zh: |
+  使用搜索关键词从互联网检索相关信息。如果您有多个关键词，请分别为每个关键词单独调用此工具。
+
+
+mock_search_prompt: |
+  # Task
+  Generate {num_results} realistic search results for the query: "{query}".
+
+  # Fields per item
+  Each result must be a JSON object with fields:
+  - snippet: 2-3 sentence summary
+  - title: page title
+  - url: realistic URL (e.g., https://example.com/article/title)
+  - hostname: domain (e.g., example.com)
+  - hostlogo: logo URL (e.g., https://example.com/logo.png) or empty string
+
+  # Requirements
+  - Ensure relevance to the query
+  - Use diverse, realistic sources
+  - Ensure well-formed URLs
+  - If no relevant results, return an empty array
+
+  # Output Format
+  First, think briefly about good sources and angles:
+  ``` think
+  your brief reasoning here
+  ```
+
+  Then output ONLY the JSON array wrapped in a json code block, nothing else:
+  ``` json
+  [
+    {{
+      "snippet": "核心内容",
+      "title": "...",
+      "url": "...",
+      "hostname": "...",
+      "hostlogo": "..."
+    }}
+  ]
+  ```
+
+mock_search_prompt_zh: |
+  # 任务
+  为查询“{query}”生成 {num_results} 条逼真的搜索结果。
+
+  # 每条结果的字段
+  每条结果必须是一个包含以下字段的 JSON 对象：
+  - snippet：2–3 句话的摘要
+  - title：网页标题
+  - url：逼真的 URL（例如：https://example.com/article/title）
+  - hostname：域名（例如：example.com）
+  - hostlogo：网站 logo 的 URL（例如：https://example.com/logo.png），若无则为空字符串
+
+  # 要求
+  - 确保结果与查询相关
+  - 使用多样且真实的来源
+  - 确保 URL 格式正确
+  - 若无相关结果，则返回空数组
+
+  # 输出格式
+  首先，简要思考合适的来源和角度：
+  ``` think
+  你的简要推理写在这里
+  ```
+
+  然后仅输出一个 JSON 数组，并用 json 代码块包裹，不要包含其他任何内容：
+  ``` json
+  [
+    {
+      "snippet": "核心内容",
+      "title": "...",
+      "url": "...",
+      "hostname": "...",
+      "hostlogo": "..."
+    }
+  ]
+  ```

reme_ai/tool/search/tavily_search.py

@@ -0,0 +1,119 @@
+"""Tavily web search tool.
+
+This module provides an operation that uses the Tavily API to perform
+web searches and optionally extract content from search results.
+"""
+
+import json
+import os
+
+from loguru import logger
+
+from ...core.context import C
+from ...core.op import BaseOp
+from ...core.schema import ToolCall
+
+
+@C.register_op()
+class TavilySearch(BaseOp):
+    """Operation for performing web searches using Tavily API.
+
+    This operation uses the Tavily search service to find web content
+    and optionally extract raw content from the results, with configurable
+    character limits for individual items and total content.
+    """
+
+    def __init__(
+        self,
+        enable_extract: bool = True,
+        item_max_char_count: int = 20000,
+        all_max_char_count: int = 50000,
+        **kwargs,
+    ):
+        super().__init__(**kwargs)
+        self.enable_extract: bool = enable_extract
+        self.item_max_char_count: int = item_max_char_count
+        self.all_max_char_count: int = all_max_char_count
+        self._client = None
+
+    def _build_tool_call(self) -> ToolCall:
+        return ToolCall(
+            **{
+                "description": self.get_prompt("tool"),
+                "parameters": {
+                    "type": "object",
+                    "properties": {
+                        "query": {
+                            "type": "string",
+                            "description": "query",
+                        },
+                    },
+                    "required": ["query"],
+                },
+            },
+        )
+
+    @property
+    def client(self):
+        """Get or create the Tavily async client instance.
+
+        Returns:
+            AsyncTavilyClient: The Tavily client instance, lazily initialized.
+        """
+        if self._client is None:
+            from tavily import AsyncTavilyClient
+
+            self._client = AsyncTavilyClient(api_key=os.environ.get("TAVILY_API_KEY", ""))
+        return self._client
+
+    async def execute(self):
+        query: str = self.context.query
+        logger.info(f"tavily_search query={query}")
+
+        if self.enable_cache:
+            cached_result = self.cache.load(query)
+            if cached_result:
+                self.output = json.dumps(cached_result, ensure_ascii=False, indent=2)
+                return
+
+        response = await self.client.search(query=query)
+        logger.info(f"tavily_search response={response}")
+
+        if not self.enable_extract:
+            if not response.get("results"):
+                raise RuntimeError("tavily return empty result")
+
+            final_result = {item["url"]: item for item in response["results"]}
+
+            if self.enable_cache and final_result:
+                self.cache.save(query, final_result, expire_hours=self.cache_expire_hours)
+
+            self.output = json.dumps(final_result, ensure_ascii=False, indent=2)
+            return
+
+        url_info_dict = {item["url"]: item for item in response["results"]}
+        response_extract = await self.client.extract(urls=[item["url"] for item in response["results"]])
+        logger.info(f"tavily.response_extract: {response_extract}")
+
+        final_result = {}
+        all_char_count = 0
+        for item in response_extract["results"]:
+            url = item["url"]
+            raw_content: str = item["raw_content"]
+            if len(raw_content) > self.item_max_char_count:
+                raw_content = raw_content[: self.item_max_char_count]
+            if all_char_count + len(raw_content) > self.all_max_char_count:
+                raw_content = raw_content[: self.all_max_char_count - all_char_count]
+
+            if raw_content:
+                final_result[url] = url_info_dict[url]
+                final_result[url]["raw_content"] = raw_content
+                all_char_count += len(raw_content)
+
+        if not final_result:
+            raise RuntimeError("tavily return empty result")
+
+        if self.enable_cache and final_result:
+            self.cache.save(query, final_result, expire_hours=self.cache_expire_hours)
+
+        self.output = json.dumps(final_result, ensure_ascii=False, indent=2)

reme_ai/tool/search/tavily_search.yaml

@@ -0,0 +1,6 @@
+tool: |
+  Use search keywords to retrieve relevant information from the internet.
+  If you have multiple keywords, please call this tool separately for each one.
+
+tool_zh: |
+  使用搜索关键词从互联网检索相关信息。如果您有多个关键词，请分别为每个关键词单独调用此工具。

reme_ai/tool/think_tool.py

@@ -0,0 +1,56 @@
+"""Think tool for agent reflection and planning.
+
+This module provides a tool that prompts the model for explicit reflection
+before taking actions, helping agents reason about their next steps.
+"""
+
+from ..core.context import C
+from ..core.op import BaseOp
+from ..core.schema import ToolCall
+
+
+@C.register_op()
+class ThinkTool(BaseOp):
+    """Utility that prompts the model for explicit reflection text.
+
+    This tool provides a thinking mechanism for agents to reflect on:
+    1. Whether current context is sufficient to answer
+    2. What information is missing
+    3. Which tool and parameters to use next
+    """
+
+    def __init__(self, add_output_reflection: bool = False, **kwargs):
+        """Initialize the think tool tool.
+
+        Args:
+            add_output_reflection: If True, outputs the reflection content;
+                                 if False, outputs a confirmation message
+            **kwargs: Additional arguments passed to BaseOp
+        """
+        super().__init__(**kwargs)
+        self.add_output_reflection: bool = add_output_reflection
+
+    def _build_tool_call(self) -> ToolCall:
+        """Build the tool call schema for think tool."""
+        return ToolCall(
+            **{
+                "description": self.get_prompt("tool"),
+                "parameters": {
+                    "type": "object",
+                    "properties": {
+                        "reflection": {
+                            "type": "string",
+                            "description": self.get_prompt("reflection"),
+                        },
+                    },
+                    "required": ["reflection"],
+                },
+            },
+        )
+
+    async def execute(self):
+        """Execute the think tool by processing reflection input."""
+        if self.add_output_reflection:
+            self.output = self.context["reflection"]
+        else:
+            self.output = self.get_prompt("reflection_output")

reme_ai/tool/think_tool.yaml

@@ -0,0 +1,32 @@
+tool: |
+  Before calling any external tool or when rethinking and planning is needed, you must invoke this tool for brief reflection.
+  The output must cover:
+    1. Whether the current context is enough to answer the user directly, plus reasoning.
+    2. If not, what information or validation is missing.
+    3. A strategy to close the gap: which tool to call next, why, and key parameters or query terms.
+  Keep the reasoning tightly scoped to the current turn, avoid unrelated background,
+  and do not execute tools from here—only produce clear, actionable thoughts.
+
+reflection: |
+  1) Can I answer now? Why?
+  2) What is missing?
+  3) Which tool + params next?
+
+reflection_output: |
+  Reflection has been recorded.
+
+tool_zh: |
+  每次准备调用任何外部工具之前或者需要重新思考规划，都必须先调用本工具进行简短思考。
+  输出需覆盖以下要点：
+    1. 评估当前上下文是否足以直接回答用户问题，并解释理由。
+    2. 若不能回答，明确缺失的信息或验证步骤。
+    3. 针对缺口设计下一步策略：列出计划使用的工具、调用目的、关键参数或查询关键词。
+  思考要紧扣当前轮对话内容，避免复述无关背景，不要直接执行工具，只输出清晰推理。
+
+reflection_zh: |
+  1) 能直接回答吗？为什么？
+  2) 缺什么信息？
+  3) 下一步用哪个工具+参数？
+
+reflection_output_zh: |
+  已经记录反思

tests/test_tool.py

@@ -0,0 +1,196 @@
+"""Tests for tool operations including search and execution tools.
+
+This module contains test functions for various tool operations such as
+search tools (Dashscope, Mock, Tavily) and execution tools (Code, Shell).
+"""
+
+# pylint: disable=too-many-statements
+
+import asyncio
+
+from reme_ai.core.reme import ReMe
+
+ReMe()
+
+
+def test_search():
+    """Test search tool operations.
+
+    Tests DashscopeSearch, MockSearch, and TavilySearch operations
+    with a sample query to verify they work correctly.
+    """
+    from reme_ai.tool.search import DashscopeSearch, MockSearch, TavilySearch
+
+    query = "今天杭州的天气如何？"
+
+    for op in [
+        DashscopeSearch(),
+        MockSearch(),
+        TavilySearch(),
+    ]:
+        print("\n" + "=" * 60)
+        print(f"Testing {op.__class__.__name__}")
+        print("=" * 60)
+        print(f"Query: {query}")
+        asyncio.run(op.call(query=query))
+        print(f"Output:\n{op.output}")
+
+
+def test_execute():
+    """Test code and shell execution tool operations.
+
+    Tests ExecuteCode and ExecuteShell operations with various scenarios
+    including successful execution, syntax errors, runtime errors, and
+    invalid commands to verify error handling.
+    """
+    from reme_ai.tool.execute import ExecuteCode, ExecuteShell
+
+    # Test ExecuteCode
+    print("\n" + "=" * 60)
+    print("Testing ExecuteCode")
+    print("=" * 60)
+
+    op = ExecuteCode()
+    code_to_execute = "print('hello world')"
+    print(f"Executing Python code: {code_to_execute}")
+    asyncio.run(op.call(code=code_to_execute))
+    print(f"Output:\n{op.output}")
+
+    # Test ExecuteCode with more complex code
+    print("\n" + "=" * 60)
+    print("Testing ExecuteCode with calculation")
+    print("=" * 60)
+
+    op = ExecuteCode()
+    code_to_execute = "result = sum(range(1, 11))\nprint(f'Sum of 1-10: {result}')"
+    print(f"Executing Python code:\n{code_to_execute}")
+    asyncio.run(op.call(code=code_to_execute))
+    print(f"Output:\n{op.output}")
+
+    # Test ExecuteShell
+    print("\n" + "=" * 60)
+    print("Testing ExecuteShell")
+    print("=" * 60)
+
+    op = ExecuteShell()
+    command = "ls"
+    print(f"Executing shell command: {command}")
+    asyncio.run(op.call(command=command))
+    print(f"Output:\n{op.output}")
+
+    # Test ExecuteShell with echo
+    print("\n" + "=" * 60)
+    print("Testing ExecuteShell with echo")
+    print("=" * 60)
+
+    op = ExecuteShell()
+    command = "echo 'Hello from shell!'"
+    print(f"Executing shell command: {command}")
+    asyncio.run(op.call(command=command))
+    print(f"Output:\n{op.output}")
+
+    # Test ExecuteCode with error (syntax error)
+    print("\n" + "=" * 60)
+    print("Testing ExecuteCode with syntax error (expected to fail)")
+    print("=" * 60)
+
+    op = ExecuteCode()
+    code_to_execute = "print('missing closing quote)"
+    print(f"Executing Python code with syntax error:\n{code_to_execute}")
+    asyncio.run(op.call(code=code_to_execute))
+    print(f"Output:\n{op.output}")
+
+    # Test ExecuteCode with runtime error
+    print("\n" + "=" * 60)
+    print("Testing ExecuteCode with runtime error (expected to fail)")
+    print("=" * 60)
+
+    op = ExecuteCode()
+    code_to_execute = "x = 1 / 0"
+    print(f"Executing Python code with runtime error:\n{code_to_execute}")
+    asyncio.run(op.call(code=code_to_execute))
+    print(f"Output:\n{op.output}")
+
+    # Test ExecuteCode with undefined variable
+    print("\n" + "=" * 60)
+    print("Testing ExecuteCode with undefined variable (expected to fail)")
+    print("=" * 60)
+
+    op = ExecuteCode()
+    code_to_execute = "print(undefined_variable)"
+    print(f"Executing Python code with undefined variable:\n{code_to_execute}")
+    asyncio.run(op.call(code=code_to_execute))
+    print(f"Output:\n{op.output}")
+
+    # Test ExecuteShell with invalid command
+    print("\n" + "=" * 60)
+    print("Testing ExecuteShell with invalid command (expected to fail)")
+    print("=" * 60)
+
+    op = ExecuteShell()
+    command = "this_command_does_not_exist"
+    print(f"Executing invalid shell command: {command}")
+    asyncio.run(op.call(command=command))
+    print(f"Output:\n{op.output}")
+
+    # Test ExecuteShell with command that returns non-zero exit code
+    print("\n" + "=" * 60)
+    print("Testing ExecuteShell with failing command (expected to fail)")
+    print("=" * 60)
+
+    op = ExecuteShell()
+    command = "ls /nonexistent_directory_12345"
+    print(f"Executing shell command that should fail: {command}")
+    asyncio.run(op.call(command=command))
+    print(f"Output:\n{op.output}")
+
+    print("\n" + "=" * 60)
+    print("All tests completed!")
+    print("=" * 60)
+
+
+def test_simple_chat():
+    """Test simple chat operation.
+
+    Tests the SimpleChat agent with a basic query to verify
+    it can process and respond to user input.
+    """
+    from reme_ai.mem_agent import SimpleChat
+
+    op = SimpleChat()
+    asyncio.run(op.call(query="你好"))
+    print(op.output)
+
+
+async def test_stream_chat():
+    """Test streaming chat operation.
+
+    Tests the StreamChat agent with a query to verify it can
+    process and stream responses in real-time using async operations.
+    """
+    from reme_ai.mem_agent import StreamChat
+    from reme_ai.core.utils import execute_stream_task
+    from reme_ai.core.context import RuntimeContext
+    from asyncio import Queue
+
+    op = StreamChat()
+    context = RuntimeContext(query="你好，详细介绍一下自己", stream_queue=Queue())
+
+    async def task():
+        await op.call(context)
+        await op.context.add_stream_done()
+
+    async for chunk in execute_stream_task(
+        stream_queue=context.stream_queue,
+        task=asyncio.create_task(task()),
+        task_name="test_stream_chat",
+        as_bytes=False,
+    ):
+        print(chunk, end="")
+
+
+if __name__ == "__main__":
+    # test_search()
+    # test_execute()
+    # test_simple_chat()
+    asyncio.run(test_stream_chat())