diff --git a/pyproject.toml b/pyproject.toml index abb263da..cae94451 100644 --- a/pyproject.toml +++ b/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 = [ diff --git a/reme_ai/__init__.py b/reme_ai/__init__.py index bd3a7687..71af9c42 100644 --- a/reme_ai/__init__.py +++ b/reme_ai/__init__.py @@ -30,4 +30,4 @@ __all__ = [ "vector_store", ] -__version__ = "0.2.0.4" +__version__ = "0.2.0.5" diff --git a/reme_ai/core/__init__.py b/reme_ai/core/__init__.py index e69de29b..8eab5792 100644 --- a/reme_ai/core/__init__.py +++ b/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 diff --git a/reme_ai/core/application.py b/reme_ai/core/application.py index e15f4fcf..c0561705 100644 --- a/reme_ai/core/application.py +++ b/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 diff --git a/reme_ai/core/config/default.yaml b/reme_ai/core/config/default.yaml index cd04da1d..8209216d 100644 --- a/reme_ai/core/config/default.yaml +++ b/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 diff --git a/reme_ai/core/context/service_context.py b/reme_ai/core/context/service_context.py index c5bcde0c..e22147e9 100644 --- a/reme_ai/core/context/service_context.py +++ b/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. diff --git a/reme_ai/core/enumeration/__init__.py b/reme_ai/core/enumeration/__init__.py index 3323bdc9..7202a949 100644 --- a/reme_ai/core/enumeration/__init__.py +++ b/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", ] diff --git a/reme_ai/core/enumeration/memory_type.py b/reme_ai/core/enumeration/memory_type.py new file mode 100644 index 00000000..22d35481 --- /dev/null +++ b/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" diff --git a/reme_ai/core/main.py b/reme_ai/core/main.py new file mode 100644 index 00000000..81811c1f --- /dev/null +++ b/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() diff --git a/reme_ai/core/op/base_op.py b/reme_ai/core/op/base_op.py index 7eeb5fbd..4c82b64c 100644 --- a/reme_ai/core/op/base_op.py +++ b/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): diff --git a/reme_ai/core/reme.py b/reme_ai/core/reme.py index d1240cb6..552a526b 100644 --- a/reme_ai/core/reme.py +++ b/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() diff --git a/reme_ai/core/schema/__init__.py b/reme_ai/core/schema/__init__.py index f55f054e..b7b73719 100644 --- a/reme_ai/core/schema/__init__.py +++ b/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", diff --git a/reme_ai/core/schema/memory_node.py b/reme_ai/core/schema/memory_node.py new file mode 100644 index 00000000..b8b6c92a --- /dev/null +++ b/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, + ) diff --git a/reme_ai/core/service/http_service.py b/reme_ai/core/service/http_service.py index f63d7333..8ef5d4ce 100644 --- a/reme_ai/core/service/http_service.py +++ b/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 diff --git a/reme_ai/core/tool/__init__.py b/reme_ai/core/tool/__init__.py deleted file mode 100644 index 02dbd468..00000000 --- a/reme_ai/core/tool/__init__.py +++ /dev/null @@ -1,7 +0,0 @@ -"""tool""" - -from .mcp_tool import MCPTool - -__all__ = [ - "MCPTool", -] diff --git a/reme_ai/core/utils/__init__.py b/reme_ai/core/utils/__init__.py index 87e01299..242f7be9 100644 --- a/reme_ai/core/utils/__init__.py +++ b/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", ] diff --git a/reme_ai/core/utils/common_utils.py b/reme_ai/core/utils/common_utils.py index d370064d..3f171f64 100644 --- a/reme_ai/core/utils/common_utils.py +++ b/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) diff --git a/reme_ai/core/utils/execute_tuils.py b/reme_ai/core/utils/execute_tuils.py new file mode 100644 index 00000000..a4d680ab --- /dev/null +++ b/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) diff --git a/reme_ai/core/utils/llm_utils.py b/reme_ai/core/utils/llm_utils.py index ae0e763f..2362a278 100644 --- a/reme_ai/core/utils/llm_utils.py +++ b/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()) diff --git a/reme_ai/core/utils/timer.py b/reme_ai/core/utils/time.py similarity index 89% rename from reme_ai/core/utils/timer.py rename to reme_ai/core/utils/time.py index f03225fa..541feab7 100644 --- a/reme_ai/core/utils/timer.py +++ b/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. diff --git a/reme_ai/mem_agent/__init__.py b/reme_ai/mem_agent/__init__.py new file mode 100644 index 00000000..5ecdd48b --- /dev/null +++ b/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", +] diff --git a/reme_ai/mem_agent/simple_chat.py b/reme_ai/mem_agent/simple_chat.py new file mode 100644 index 00000000..0ba33c97 --- /dev/null +++ b/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 diff --git a/reme_ai/mem_agent/stream_chat.py b/reme_ai/mem_agent/stream_chat.py new file mode 100644 index 00000000..95ee6bb5 --- /dev/null +++ b/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) diff --git a/reme_ai/tool/__init__.py b/reme_ai/tool/__init__.py new file mode 100644 index 00000000..d7d25f85 --- /dev/null +++ b/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", +] diff --git a/reme_ai/tool/base_memory_tool.py b/reme_ai/tool/base_memory_tool.py new file mode 100644 index 00000000..4e57cd1e --- /dev/null +++ b/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 {}, + ) diff --git a/reme_ai/tool/execute/__init__.py b/reme_ai/tool/execute/__init__.py new file mode 100644 index 00000000..f78a73ac --- /dev/null +++ b/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", +] diff --git a/reme_ai/tool/execute/execute_code.py b/reme_ai/tool/execute/execute_code.py new file mode 100644 index 00000000..ea259487 --- /dev/null +++ b/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) diff --git a/reme_ai/tool/execute/execute_code.yaml b/reme_ai/tool/execute/execute_code.yaml new file mode 100644 index 00000000..fdaef055 --- /dev/null +++ b/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 函数输出。 \ No newline at end of file diff --git a/reme_ai/tool/execute/execute_shell.py b/reme_ai/tool/execute/execute_shell.py new file mode 100644 index 00000000..6e244ddb --- /dev/null +++ b/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) diff --git a/reme_ai/tool/execute/execute_shell.yaml b/reme_ai/tool/execute/execute_shell.yaml new file mode 100644 index 00000000..6261bd8b --- /dev/null +++ b/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。 \ No newline at end of file diff --git a/reme_ai/core/tool/mcp_tool.py b/reme_ai/tool/mcp_tool.py similarity index 95% rename from reme_ai/core/tool/mcp_tool.py rename to reme_ai/tool/mcp_tool.py index 802e4138..f06275b0 100644 --- a/reme_ai/core/tool/mcp_tool.py +++ b/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() diff --git a/reme_ai/tool/memory/__init__.py b/reme_ai/tool/memory/__init__.py new file mode 100644 index 00000000..b6dfb163 --- /dev/null +++ b/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", +] diff --git a/reme_ai/tool/memory/history/__init__.py b/reme_ai/tool/memory/history/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/reme_ai/tool/memory/history/add_history_memory.py b/reme_ai/tool/memory/history/add_history_memory.py new file mode 100644 index 00000000..14a00953 --- /dev/null +++ b/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) diff --git a/reme_ai/tool/memory/history/add_history_memory.yaml b/reme_ai/tool/memory/history/add_history_memory.yaml new file mode 100644 index 00000000..18fd9163 --- /dev/null +++ b/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. diff --git a/reme_ai/tool/memory/history/read_history_memory.py b/reme_ai/tool/memory/history/read_history_memory.py new file mode 100644 index 00000000..ac478401 --- /dev/null +++ b/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.") diff --git a/reme_ai/tool/memory/history/read_history_memory.yaml b/reme_ai/tool/memory/history/read_history_memory.yaml new file mode 100644 index 00000000..f50878a7 --- /dev/null +++ b/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. diff --git a/reme_ai/tool/memory/identity/__init__.py b/reme_ai/tool/memory/identity/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/reme_ai/tool/memory/identity/read_identity_memory.py b/reme_ai/tool/memory/identity/read_identity_memory.py new file mode 100644 index 00000000..ecb5b438 --- /dev/null +++ b/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) diff --git a/reme_ai/tool/memory/identity/read_identity_memory.yaml b/reme_ai/tool/memory/identity/read_identity_memory.yaml new file mode 100644 index 00000000..6e84e4bf --- /dev/null +++ b/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. diff --git a/reme_ai/tool/memory/identity/update_identity_memory.py b/reme_ai/tool/memory/identity/update_identity_memory.py new file mode 100644 index 00000000..ba0e2de7 --- /dev/null +++ b/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) diff --git a/reme_ai/tool/memory/identity/update_identity_memory.yaml b/reme_ai/tool/memory/identity/update_identity_memory.yaml new file mode 100644 index 00000000..48033bfe --- /dev/null +++ b/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. diff --git a/reme_ai/tool/memory/meta/__init__.py b/reme_ai/tool/memory/meta/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/reme_ai/tool/memory/meta/add_meta_memory.py b/reme_ai/tool/memory/meta/add_meta_memory.py new file mode 100644 index 00000000..994fdf9e --- /dev/null +++ b/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) diff --git a/reme_ai/tool/memory/meta/add_meta_memory.yaml b/reme_ai/tool/memory/meta/add_meta_memory.yaml new file mode 100644 index 00000000..ce4645a4 --- /dev/null +++ b/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") diff --git a/reme_ai/tool/memory/meta/read_meta_memory.py b/reme_ai/tool/memory/meta/read_meta_memory.py new file mode 100644 index 00000000..ae31df78 --- /dev/null +++ b/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) diff --git a/reme_ai/tool/memory/meta/read_meta_memory.yaml b/reme_ai/tool/memory/meta/read_meta_memory.yaml new file mode 100644 index 00000000..dff25278 --- /dev/null +++ b/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. diff --git a/reme_ai/tool/memory/vector/__init__.py b/reme_ai/tool/memory/vector/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/reme_ai/tool/memory/vector/add_memory.py b/reme_ai/tool/memory/vector/add_memory.py new file mode 100644 index 00000000..5fc16f2c --- /dev/null +++ b/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) diff --git a/reme_ai/tool/memory/vector/add_memory.yaml b/reme_ai/tool/memory/vector/add_memory.yaml new file mode 100644 index 00000000..d484f349 --- /dev/null +++ b/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. diff --git a/reme_ai/tool/memory/vector/add_summary_memory.py b/reme_ai/tool/memory/vector/add_summary_memory.py new file mode 100644 index 00000000..684300d7 --- /dev/null +++ b/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() diff --git a/reme_ai/tool/memory/vector/add_summary_memory.yaml b/reme_ai/tool/memory/vector/add_summary_memory.yaml new file mode 100644 index 00000000..e9e469c4 --- /dev/null +++ b/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 diff --git a/reme_ai/tool/memory/vector/delete_memory.py b/reme_ai/tool/memory/vector/delete_memory.py new file mode 100644 index 00000000..4fb066a5 --- /dev/null +++ b/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) diff --git a/reme_ai/tool/memory/vector/delete_memory.yaml b/reme_ai/tool/memory/vector/delete_memory.yaml new file mode 100644 index 00000000..fe5934dd --- /dev/null +++ b/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. diff --git a/reme_ai/tool/memory/vector/update_memory.py b/reme_ai/tool/memory/vector/update_memory.py new file mode 100644 index 00000000..8d3fc7cb --- /dev/null +++ b/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) diff --git a/reme_ai/tool/memory/vector/update_memory.yaml b/reme_ai/tool/memory/vector/update_memory.yaml new file mode 100644 index 00000000..16e14c57 --- /dev/null +++ b/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. diff --git a/reme_ai/tool/memory/vector/vector_retrieve_memory.py b/reme_ai/tool/memory/vector/vector_retrieve_memory.py new file mode 100644 index 00000000..9e0cc6c7 --- /dev/null +++ b/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") diff --git a/reme_ai/tool/memory/vector/vector_retrieve_memory.yaml b/reme_ai/tool/memory/vector/vector_retrieve_memory.yaml new file mode 100644 index 00000000..c30a6efe --- /dev/null +++ b/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. diff --git a/reme_ai/tool/search/__init__.py b/reme_ai/tool/search/__init__.py new file mode 100644 index 00000000..5a73dc2e --- /dev/null +++ b/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", +] diff --git a/reme_ai/tool/search/dashscope_search.py b/reme_ai/tool/search/dashscope_search.py new file mode 100644 index 00000000..19bd8104 --- /dev/null +++ b/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"] diff --git a/reme_ai/tool/search/dashscope_search.yaml b/reme_ai/tool/search/dashscope_search.yaml new file mode 100644 index 00000000..0dcef347 --- /dev/null +++ b/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 + 直接从上下文中提取与用户问题相关的原始内容，保持准确性，避免过度处理。 \ No newline at end of file diff --git a/reme_ai/tool/search/mock_search.py b/reme_ai/tool/search/mock_search.py new file mode 100644 index 00000000..187463dc --- /dev/null +++ b/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) diff --git a/reme_ai/tool/search/mock_search.yaml b/reme_ai/tool/search/mock_search.yaml new file mode 100644 index 00000000..c952ff20 --- /dev/null +++ b/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": "..." + } + ] + ``` \ No newline at end of file diff --git a/reme_ai/tool/search/tavily_search.py b/reme_ai/tool/search/tavily_search.py new file mode 100644 index 00000000..5c194bdc --- /dev/null +++ b/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) diff --git a/reme_ai/tool/search/tavily_search.yaml b/reme_ai/tool/search/tavily_search.yaml new file mode 100644 index 00000000..2499cedb --- /dev/null +++ b/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: | + 使用搜索关键词从互联网检索相关信息。如果您有多个关键词，请分别为每个关键词单独调用此工具。 \ No newline at end of file diff --git a/reme_ai/tool/think_tool.py b/reme_ai/tool/think_tool.py new file mode 100644 index 00000000..2e83b3ad --- /dev/null +++ b/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") diff --git a/reme_ai/tool/think_tool.yaml b/reme_ai/tool/think_tool.yaml new file mode 100644 index 00000000..75903557 --- /dev/null +++ b/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: | + 已经记录反思 \ No newline at end of file diff --git a/tests/test_tool.py b/tests/test_tool.py new file mode 100644 index 00000000..7f6c7afb --- /dev/null +++ b/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())