diff --git a/README.md b/README.md index 05a7c29d..c360da7d 100644 --- a/README.md +++ b/README.md @@ -140,6 +140,7 @@ async def main(): default_as_llm_config={"model_name": "qwen3.5-35b-a3b"}, # default_embedding_model_config={"model_name": "text-embedding-v4"}, default_file_store_config={"fts_enabled": True, "vector_enabled": False}, + enable_load_env=True, ) await reme.start() diff --git a/README_ZH.md b/README_ZH.md index 45c1b277..7c3da21f 100644 --- a/README_ZH.md +++ b/README_ZH.md @@ -133,6 +133,7 @@ async def main(): default_as_llm_config={"model_name": "qwen3.5-35b-a3b"}, # default_embedding_model_config={"model_name": "text-embedding-v4"}, default_file_store_config={"fts_enabled": True, "vector_enabled": False}, + enable_load_env=True, ) await reme.start() diff --git a/reme/__init__.py b/reme/__init__.py index 74463d84..97905514 100644 --- a/reme/__init__.py +++ b/reme/__init__.py @@ -6,7 +6,7 @@ from . import extension from . import memory from .reme import ReMe -__version__ = "0.3.0.8" +__version__ = "0.3.1.1" __all__ = [ "config", diff --git a/reme/core/application.py b/reme/core/application.py index 8778dfba..6f04e663 100644 --- a/reme/core/application.py +++ b/reme/core/application.py @@ -12,10 +12,19 @@ from .flow import BaseFlow from .llm import BaseLLM from .prompt_handler import PromptHandler from .registry_factory import R -from .schema import Response, ServiceConfig +from .schema import ( + EmbeddingModelConfig, + Response, + ServiceConfig, + LLMConfig, + VectorStoreConfig, + FileStoreConfig, + FileWatcherConfig, + TokenCounterConfig, +) from .service_context import ServiceContext from .token_counter import BaseTokenCounter -from .utils import execute_stream_task, PydanticConfigParser, init_logger, MCPClient, print_logo, get_logger +from .utils import execute_stream_task, PydanticConfigParser, init_logger, MCPClient, print_logo, get_logger, load_env from .vector_store import BaseVectorStore logger = get_logger() @@ -35,6 +44,7 @@ class Application: config_path: str | None = None, enable_logo: bool = True, log_to_console: bool = True, + enable_load_env: bool = True, parser: type[PydanticConfigParser] | None = None, default_as_llm_config: dict | None = None, default_as_llm_formatter_config: dict | None = None, @@ -46,12 +56,17 @@ class Application: default_file_watcher_config: dict | None = None, **kwargs, ): + + if enable_load_env: + load_env() + + self.llm_api_key = llm_api_key or os.getenv("LLM_API_KEY", "") + self.llm_base_url = llm_base_url or os.getenv("LLM_BASE_URL", "") + self.embedding_api_key = embedding_api_key or os.getenv("EMBEDDING_API_KEY", "") + self.embedding_base_url = embedding_base_url or os.getenv("EMBEDDING_BASE_URL", "") + self.service_context = ServiceContext( *args, - llm_api_key=llm_api_key, - llm_base_url=llm_base_url, - embedding_api_key=embedding_api_key, - embedding_base_url=embedding_base_url, service_config=None, parser=parser, working_dir=working_dir, @@ -158,11 +173,11 @@ class Application: else: config_dict = config.model_dump(exclude={"backend"}) if not config_dict.get("api_key", ""): - config_dict["api_key"] = os.getenv("LLM_API_KEY", "") + config_dict["api_key"] = self.llm_api_key if "client_kwargs" not in config_dict: config_dict["client_kwargs"] = {} if not config_dict["client_kwargs"].get("base_url", ""): - config_dict["client_kwargs"]["base_url"] = os.getenv("LLM_BASE_URL", "") + config_dict["client_kwargs"]["base_url"] = self.llm_base_url self.service_context.as_llms[name] = R.as_llms[config.backend](**config_dict) for name, config in self.service_config.as_llm_formatters.items(): @@ -184,6 +199,8 @@ class Application: logger.warning(f"LLM backend {config.backend} is not supported.") else: config_dict = config.model_dump(exclude={"backend"}) + config_dict.setdefault("api_key", self.llm_api_key) + config_dict.setdefault("base_url", self.llm_base_url) self.service_context.llms[name] = R.llms[config.backend](**config_dict) await self.service_context.llms[name].start() @@ -192,7 +209,9 @@ class Application: logger.warning(f"Embedding model backend {config.backend} is not supported.") else: config_dict = config.model_dump(exclude={"backend"}) - config_dict["cache_dir"] = working_path / "embedding_cache" + config_dict.setdefault("api_key", self.embedding_api_key) + config_dict.setdefault("base_url", self.embedding_base_url) + config_dict.setdefault("cache_dir", working_path / "embedding_cache") self.service_context.embedding_models[name] = R.embedding_models[config.backend](**config_dict) await self.service_context.embedding_models[name].start() @@ -247,6 +266,209 @@ class Application: logger.info("ReMe Application started") return self + # pylint: disable=too-many-statements + async def restart(self, restart_config: dict): + """Restart the application with new config.""" + + working_path = Path(self.service_config.working_dir) + working_path.mkdir(parents=True, exist_ok=True) + + # as_llms + if "as_llms" in restart_config: + as_llms_config = restart_config["as_llms"] + assert isinstance(as_llms_config, dict) + for name, config in as_llms_config.items(): + if name in self.service_context.as_llms: + del self.service_context.as_llms[name] + + if config.get("backend") not in R.as_llms: + logger.warning(f"AS LLM backend {config.get('backend')} is not supported.") + continue + + config_dict = {k: v for k, v in config.items() if k != "backend"} + if not config_dict.get("api_key", ""): + config_dict["api_key"] = self.llm_api_key + if "client_kwargs" not in config_dict: + config_dict["client_kwargs"] = {} + if not config_dict["client_kwargs"].get("base_url", ""): + config_dict["client_kwargs"]["base_url"] = self.llm_base_url + self.service_context.as_llms[name] = R.as_llms[config["backend"]](**config_dict) + logger.info(f"Restarted AS LLM: {name}") + + # as_llm_formatters + if "as_llm_formatters" in restart_config: + as_llm_formatters_config = restart_config["as_llm_formatters"] + assert isinstance(as_llm_formatters_config, dict) + for name, config in as_llm_formatters_config.items(): + if name in self.service_context.as_llm_formatters: + del self.service_context.as_llm_formatters[name] + + if config.get("backend") not in R.as_llm_formatters: + logger.warning(f"AS LLM formatter backend {config.get('backend')} is not supported.") + continue + config_dict = {k: v for k, v in config.items() if k != "backend"} + self.service_context.as_llm_formatters[name] = R.as_llm_formatters[config["backend"]](**config_dict) + logger.info(f"Restarted AS LLM formatter: {name}") + + # as_token_counters + if "as_token_counters" in restart_config: + as_token_counters_config = restart_config["as_token_counters"] + assert isinstance(as_token_counters_config, dict) + for name, config in as_token_counters_config.items(): + if name in self.service_context.as_token_counters: + del self.service_context.as_token_counters[name] + + if config.get("backend") not in R.as_token_counters: + logger.warning(f"Token counter backend {config.get('backend')} is not supported.") + continue + config_dict = {k: v for k, v in config.items() if k != "backend"} + self.service_context.as_token_counters[name] = R.as_token_counters[config["backend"]](**config_dict) + logger.info(f"Restarted AS token counter: {name}") + + # llms + if "llms" in restart_config: + llms_config = restart_config["llms"] + assert isinstance(llms_config, dict) + for name, config in llms_config.items(): + if name in self.service_context.llms: + llm = self.service_context.llms.pop(name) + await llm.close() + + if isinstance(config, dict): + config = LLMConfig(**config) + if config.backend not in R.llms: + logger.warning(f"LLM backend {config.backend} is not supported.") + continue + config_dict = config.model_dump(exclude={"backend"}) + config_dict.setdefault("api_key", self.llm_api_key) + config_dict.setdefault("base_url", self.llm_base_url) + self.service_context.llms[name] = R.llms[config.backend](**config_dict) + await self.service_context.llms[name].start() + logger.info(f"Restarted LLM: {name}") + + # embedding_models + if "embedding_models" in restart_config: + embedding_models_config = restart_config["embedding_models"] + assert isinstance(embedding_models_config, dict) + updated_names = set() + for name, config in embedding_models_config.items(): + if name in self.service_context.embedding_models: + embedding_model = self.service_context.embedding_models.pop(name) + await embedding_model.close() + + if isinstance(config, dict): + config = EmbeddingModelConfig(**config) + if config.backend not in R.embedding_models: + logger.warning(f"Embedding model backend {config.backend} is not supported.") + continue + config_dict = config.model_dump(exclude={"backend"}) + config_dict.setdefault("api_key", self.embedding_api_key) + config_dict.setdefault("base_url", self.embedding_base_url) + config_dict.setdefault("cache_dir", working_path / "embedding_cache") + self.service_context.embedding_models[name] = R.embedding_models[config.backend](**config_dict) + await self.service_context.embedding_models[name].start() + logger.info(f"Restarted embedding model: {name}") + updated_names.add(name) + + # update embedding_model attribute for existing vector_stores and file_stores + for name in updated_names: + for vs_name, vs_config in self.service_config.vector_stores.items(): + if vs_config.embedding_model == name and vs_name in self.service_context.vector_stores: + self.service_context.vector_stores[vs_name].embedding_model = ( + self.service_context.embedding_models[name] + ) + logger.info(f"Updated embedding model for vector store: {vs_name}") + for fs_name, fs_config in self.service_config.file_stores.items(): + if fs_config.embedding_model == name and fs_name in self.service_context.file_stores: + self.service_context.file_stores[fs_name].embedding_model = ( + self.service_context.embedding_models[name] + ) + logger.info(f"Updated embedding model for file store: {fs_name}") + + # token_counters + if "token_counters" in restart_config: + token_counters_config = restart_config["token_counters"] + assert isinstance(token_counters_config, dict) + for name, config in token_counters_config.items(): + if name in self.service_context.token_counters: + del self.service_context.token_counters[name] + + if isinstance(config, dict): + config = TokenCounterConfig(**config) + if config.backend not in R.token_counters: + logger.warning(f"Token counter backend {config.backend} is not supported.") + continue + config_dict = config.model_dump(exclude={"backend"}) + self.service_context.token_counters[name] = R.token_counters[config.backend](**config_dict) + logger.info(f"Restarted token counter: {name}") + + # vector_stores + if "vector_stores" in restart_config: + vector_stores_config = restart_config["vector_stores"] + assert isinstance(vector_stores_config, dict) + for name, config in vector_stores_config.items(): + if name in self.service_context.vector_stores: + vector_store = self.service_context.vector_stores.pop(name) + await vector_store.close() + if isinstance(config, dict): + config = VectorStoreConfig(**config) + if config.backend not in R.vector_stores: + logger.warning(f"Vector store backend {config.backend} is not supported.") + continue + config_dict = config.model_dump(exclude={"backend", "embedding_model"}) + config_dict.update( + { + "embedding_model": self.service_context.embedding_models[config.embedding_model], + "db_path": working_path / "vector_store", + }, + ) + self.service_context.vector_stores[name] = R.vector_stores[config.backend](**config_dict) + await self.service_context.vector_stores[name].start() + logger.info(f"Restarted vector store: {name}") + + # file_stores + if "file_stores" in restart_config: + file_stores_config = restart_config["file_stores"] + assert isinstance(file_stores_config, dict) + for name, config in file_stores_config.items(): + if name in self.service_context.file_stores: + file_store = self.service_context.file_stores.pop(name) + await file_store.close() + if isinstance(config, dict): + config = FileStoreConfig(**config) + if config.backend not in R.file_stores: + logger.warning(f"File store backend {config.backend} is not supported.") + continue + config_dict = config.model_dump(exclude={"backend", "embedding_model"}) + config_dict.update( + { + "embedding_model": self.service_context.embedding_models[config.embedding_model], + "db_path": working_path / "file_store", + }, + ) + self.service_context.file_stores[name] = R.file_stores[config.backend](**config_dict) + await self.service_context.file_stores[name].start() + logger.info(f"Restarted file store: {name}") + + # file_watchers + if "file_watchers" in restart_config: + file_watchers_config = restart_config["file_watchers"] + assert isinstance(file_watchers_config, dict) + for name, config in file_watchers_config.items(): + if name in self.service_context.file_watchers: + file_watcher = self.service_context.file_watchers.pop(name) + await file_watcher.close() + if isinstance(config, dict): + config = FileWatcherConfig(**config) + if config.backend not in R.file_watchers: + logger.warning(f"File watcher backend {config.backend} is not supported.") + continue + config_dict = config.model_dump(exclude={"backend", "file_store"}) + config_dict["file_store"] = self.service_context.file_stores[config.file_store] + self.service_context.file_watchers[name] = R.file_watchers[config.backend](**config_dict) + await self.service_context.file_watchers[name].start() + logger.info(f"Restarted file watcher: {name}") + async def prepare_mcp_servers(self): """Prepare and initialize MCP server connections.""" mcp_client = MCPClient(config={"mcpServers": self.service_config.mcp_servers}) diff --git a/reme/core/as_token_counter/reme_token_counter.py b/reme/core/as_token_counter/reme_token_counter.py index 39ad8f6d..6722e92f 100644 --- a/reme/core/as_token_counter/reme_token_counter.py +++ b/reme/core/as_token_counter/reme_token_counter.py @@ -47,9 +47,18 @@ class ReMeTokenCounter(HuggingFaceTokenCounter): # Set HuggingFace endpoint for mirror support if use_mirror: - os.environ["HF_ENDPOINT"] = "https://hf-mirror.com" + mirror = "https://hf-mirror.com" else: - os.environ.pop("HF_ENDPOINT", None) + mirror = "https://huggingface.co" + + os.environ["HF_ENDPOINT"] = mirror + + # if the huggingface is already imported in other dependencies, + # we need to set the endpoint manually + import huggingface_hub.constants + + huggingface_hub.constants.ENDPOINT = mirror + huggingface_hub.constants.HUGGINGFACE_CO_URL_TEMPLATE = mirror + "/{repo_id}/resolve/{revision}/{filename}" try: super().__init__( diff --git a/reme/core/embedding/base_embedding_model.py b/reme/core/embedding/base_embedding_model.py index 2b60e8e6..28efddce 100644 --- a/reme/core/embedding/base_embedding_model.py +++ b/reme/core/embedding/base_embedding_model.py @@ -6,7 +6,6 @@ Defines the abstract base class and standard API for all embedding model impleme import asyncio import hashlib import json -import os import time from abc import ABC from collections import OrderedDict @@ -56,8 +55,8 @@ class BaseEmbeddingModel(ABC): enable_cache: Whether to enable embedding cache **kwargs: Additional model-specific parameters """ - self._api_key: str = api_key - self._base_url: str = base_url + self.api_key: str = api_key + self.base_url: str = base_url self.model_name = model_name self.dimensions = dimensions self.use_dimensions = use_dimensions @@ -78,16 +77,6 @@ class BaseEmbeddingModel(ABC): self.cache_path: Path = Path(self.cache_dir) self.cache_path.mkdir(parents=True, exist_ok=True) - @property - def api_key(self) -> str | None: - """Get API key from environment variable.""" - return os.getenv("EMBEDDING_API_KEY") or self._api_key - - @property - def base_url(self) -> str | None: - """Get base URL from environment variable.""" - return os.getenv("EMBEDDING_BASE_URL") or self._base_url - def _truncate_text(self, text: str) -> str: """Truncate text to max_input_length if it exceeds the limit.""" if len(text) > self.max_input_length: diff --git a/reme/core/embedding/openai_embedding_model.py b/reme/core/embedding/openai_embedding_model.py index 3857f782..d686c83c 100644 --- a/reme/core/embedding/openai_embedding_model.py +++ b/reme/core/embedding/openai_embedding_model.py @@ -45,7 +45,9 @@ class OpenAIEmbeddingModel(BaseEmbeddingModel): result_emb = [[] for _ in range(len(input_text))] for emb in completion.data: - result_emb[emb.index] = emb.embedding + # BGE-M3 returns dense_embedding instead of embedding; use as fallback + vec = getattr(emb, "embedding", None) or getattr(emb, "dense_embedding", None) + result_emb[emb.index] = list(vec) if vec is not None else [] return result_emb async def start(self): diff --git a/reme/core/llm/base_llm.py b/reme/core/llm/base_llm.py index ba13ff10..c08a423c 100644 --- a/reme/core/llm/base_llm.py +++ b/reme/core/llm/base_llm.py @@ -2,7 +2,6 @@ import asyncio import json -import os import time from abc import ABC, abstractmethod from typing import Callable, Generator, AsyncGenerator, Any @@ -36,8 +35,8 @@ class BaseLLM(ABC): request_interval: Minimum seconds between requests (default: 0.0) **kwargs: Additional model-specific parameters """ - self._api_key: str = api_key - self._base_url: str = base_url + self.api_key: str = api_key + self.base_url: str = base_url self.model_name: str = model_name self.max_retries: int = max_retries self.raise_exception: bool = raise_exception @@ -47,16 +46,6 @@ class BaseLLM(ABC): self._last_request_time: float = 0.0 self._request_lock: asyncio.Lock = asyncio.Lock() - @property - def api_key(self) -> str | None: - """Get API key from environment variable.""" - return os.getenv("LLM_API_KEY") or self._api_key - - @property - def base_url(self) -> str | None: - """Get base URL from environment variable.""" - return os.getenv("LLM_BASE_URL") or self._base_url - @staticmethod def _accumulate_tool_call_chunk(tool_call, ret_tools: list[ToolCall]): """Assemble incremental tool call chunks into complete ToolCall objects.""" diff --git a/reme/core/service_context.py b/reme/core/service_context.py index 4ab83ff7..2c58fe7d 100644 --- a/reme/core/service_context.py +++ b/reme/core/service_context.py @@ -1,6 +1,5 @@ """Service context.""" -import os from concurrent.futures import ThreadPoolExecutor from typing import TYPE_CHECKING @@ -8,7 +7,7 @@ from loguru import logger from .base_dict import BaseDict from .schema import ServiceConfig -from .utils import load_env, PydanticConfigParser +from .utils import PydanticConfigParser if TYPE_CHECKING: from agentscope.model import ChatModelBase @@ -29,10 +28,6 @@ class ServiceContext(BaseDict): def __init__( self, *args, - llm_api_key: str | None = None, - llm_base_url: str | None = None, - embedding_api_key: str | None = None, - embedding_base_url: str | None = None, service_config: ServiceConfig | None = None, parser: type[PydanticConfigParser] | None = None, working_dir: str | None = None, @@ -52,15 +47,6 @@ class ServiceContext(BaseDict): ): super().__init__() - # Load environment variables - load_env() - - # Update common environment variables for LLM and embedding services. - self.update_env("LLM_API_KEY", llm_api_key) - self.update_env("LLM_BASE_URL", llm_base_url) - self.update_env("EMBEDDING_API_KEY", embedding_api_key) - self.update_env("EMBEDDING_BASE_URL", embedding_base_url) - if service_config is None: parser_class = parser if parser is not None else PydanticConfigParser parser_instance = parser_class(ServiceConfig) @@ -114,12 +100,6 @@ class ServiceContext(BaseDict): self.flows: dict[str, "BaseFlow"] = {} self.mcp_server_mapping: dict[str, dict] = {} - @staticmethod - def update_env(key: str, value: str | None): - """Update environment variable if value is provided.""" - if value: - os.environ[key] = value - @staticmethod def _update_section_config(config: dict, section_name: str, **kwargs): """Update a specific section of the service config with new values.""" diff --git a/reme/core/utils/__init__.py b/reme/core/utils/__init__.py index 9dbc59dc..b51936c7 100644 --- a/reme/core/utils/__init__.py +++ b/reme/core/utils/__init__.py @@ -19,7 +19,7 @@ from .pydantic_utils import create_pydantic_model from .singleton import singleton from .time import timer, get_now_time from .hf_token_counter_utils import get_hf_token_counter -from .truncate_text_utils import truncate_text, is_truncated +from .truncate_text_utils import truncate_text, truncate_text_head, is_truncated, TRUNCATION_MARKER_START __all__ = [ "convert_dashscope_to_agentscope", @@ -52,5 +52,7 @@ __all__ = [ "get_now_time", "get_hf_token_counter", "truncate_text", + "truncate_text_head", "is_truncated", + "TRUNCATION_MARKER_START", ] diff --git a/reme/core/utils/truncate_text_utils.py b/reme/core/utils/truncate_text_utils.py index da0c473a..1f60d19f 100644 --- a/reme/core/utils/truncate_text_utils.py +++ b/reme/core/utils/truncate_text_utils.py @@ -41,15 +41,42 @@ def truncate_text(text: str, max_length: int) -> str: ) +def truncate_text_head(text: str, max_length: int) -> str: + """Truncate text from the beginning, keeping only the head portion. + + Args: + text: The text to truncate + max_length: Maximum allowed length + + Returns: + Truncated text with marker indicating truncation at the end + """ + text = str(text) if text else "" + if not text: + return text + + if len(text) <= max_length: + return text + + truncated_chars = len(text) - max_length + logger.debug( + "Text truncated from head: original %d chars, kept %d, removed %d chars from tail.", + len(text), + max_length, + truncated_chars, + ) + return f"{text[:max_length]}{TRUNCATION_MARKER_START}" + + def is_truncated(text: str) -> bool: - """Check if the text has been truncated (contains truncation markers). + """Check if the text has been truncated (contains truncation marker). Args: text: The text to check Returns: - bool: True if text contains truncation markers, False otherwise + bool: True if text contains truncation marker, False otherwise """ if not text: return False - return TRUNCATION_MARKER_START in text and TRUNCATION_MARKER_END in text + return TRUNCATION_MARKER_START in text diff --git a/reme/memory/file_based/components/cli.py b/reme/memory/file_based/components/cli.py index f4dddc21..11d45035 100644 --- a/reme/memory/file_based/components/cli.py +++ b/reme/memory/file_based/components/cli.py @@ -3,6 +3,7 @@ import asyncio from datetime import datetime from pathlib import Path +import zoneinfo from agentscope.agent import ReActAgent from agentscope.message import Msg, TextBlock @@ -46,6 +47,7 @@ class CliAgent(BaseOp): reserve_tokens: int = 36000, keep_recent_tokens: int = 20000, language: str = "zh", + timezone: str | None = None, **kwargs, ): super().__init__(**kwargs) @@ -57,6 +59,7 @@ class CliAgent(BaseOp): self.reserve_tokens: int = reserve_tokens self.keep_recent_tokens: int = keep_recent_tokens self.language: str = language + self.timezone: str | None = timezone # Initialize message history self.messages: list[Msg] = [] @@ -93,6 +96,7 @@ class CliAgent(BaseOp): as_llm_formatter=self.as_llm_formatter, language=self.language if self.language == "zh" else "", console_enabled=False, # We disable the terminal printing to avoid messy outputs + timezone=self.timezone, ) # Create summary task @@ -168,6 +172,7 @@ class CliAgent(BaseOp): as_llm_formatter=self.as_llm_formatter, language=self.language if self.language == "zh" else "", console_enabled=False, # We disable the terminal printing to avoid messy outputs + timezone=self.timezone, ) summary_content = await compactor.call( @@ -195,7 +200,8 @@ class CliAgent(BaseOp): async def _build_messages(self, query: str) -> list[Msg]: """Build system prompt message.""" - current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S %A") + tz = zoneinfo.ZoneInfo(self.timezone) if self.timezone else None + current_time = datetime.now(tz).strftime("%Y-%m-%d %H:%M:%S %A") # Create system prompt system_prompt = self.prompt_format( diff --git a/reme/memory/file_based/components/summarizer.py b/reme/memory/file_based/components/summarizer.py index d553a5f6..855d64c1 100644 --- a/reme/memory/file_based/components/summarizer.py +++ b/reme/memory/file_based/components/summarizer.py @@ -1,6 +1,7 @@ """Summarizer module for memory summarization operations.""" import datetime +import zoneinfo from agentscope.agent import ReActAgent from agentscope.message import Msg @@ -23,6 +24,7 @@ class Summarizer(BaseOp): memory_compact_threshold: int, toolkit: Toolkit | None = None, console_enabled: bool = False, + timezone: str | None = None, **kwargs, ): super().__init__(**kwargs) @@ -31,6 +33,7 @@ class Summarizer(BaseOp): self.memory_compact_threshold: int = memory_compact_threshold self.toolkit: Toolkit | None = toolkit self.console_enabled: bool = console_enabled + self.timezone: str | None = timezone async def execute(self): messages: list[Msg] = self.context.get("messages", []) @@ -62,7 +65,13 @@ class Summarizer(BaseOp): user_message: str = f"\n{history_formatted_str}\n\n" + self.prompt_format( "user_message", - date=datetime.datetime.now().strftime("%Y-%m-%d"), + date=( + datetime.datetime.now( + zoneinfo.ZoneInfo(self.timezone), + ) + if self.timezone + else datetime.datetime.now() + ).strftime("%Y-%m-%d"), working_dir=self.working_dir, memory_dir=self.memory_dir, ) diff --git a/reme/memory/file_based/components/summarizer.yaml b/reme/memory/file_based/components/summarizer.yaml index 9aa0892b..00dbdc36 100644 --- a/reme/memory/file_based/components/summarizer.yaml +++ b/reme/memory/file_based/components/summarizer.yaml @@ -1,50 +1,61 @@ user_message: | - Memory Pre-compression Flush Cycle Initiated - The current session is about to enter the automatic compression phase. Please capture persistent memory and write it to disk. + Memory Pre-compression Flush Cycle Initiated - Current date: {date} - Working directory: {working_dir} + The current session is about to enter the automatic compression phase. Please capture persistent memory AND session reflections, then write them to disk. - Immediately store persistent memory to: {memory_dir}/YYYY-MM-DD.md + Current date: {date} + Working directory: {working_dir} - Workflow: - 1. First, `read` {memory_dir}/YYYY-MM-DD.md (if the file doesn’t exist, an error message will be returned). - 2. Intelligently merge new information with existing content (skip merging if the file doesn’t exist): - - Avoid duplicating already recorded information - - Enrich existing entries with new details where relevant - - Maintain chronological order wherever applicable - 3. Write the updated content: - - Prefer using `edit` to update specific sections when possible - - Use `write` to overwrite the entire file only if substantial restructuring is required + Immediately store persistent memory and reflections to: {memory_dir}/YYYY-MM-DD.md - Principles: - - Always preserve timestamps and any date/time-related context - - Add only genuinely new or meaningfully enriching information - - Keep entries concise yet complete - - If there’s nothing to store, respond with [SILENT] + Workflow: + 1. First, `read` {memory_dir}/YYYY-MM-DD.md (if the file doesn’t exist, an error message will be returned). + 2. Extract and synthesize content from the current session: + - Persistent Memory: Facts, user profile updates, project states, and important events. + - Experience Reflection: Reusable thinking logic derived from user feedback, successful problem-solving strategies, mistakes made/pitfalls to avoid, and actionable insights for future interactions. + 3. Intelligently merge new information with existing content (skip merging if the file doesn’t exist): + - Categorize clearly (e.g., separate "Factual Memory" from "Reflections & Logic"). + - Avoid duplicating already recorded information. + - Enrich existing entries with new details where relevant. + - Maintain chronological order wherever applicable. + 4. Write the updated content: + - Prefer using `edit` to update specific sections when possible. + - Use `write` to overwrite the entire file only if substantial restructuring is required. + Principles: + - Always preserve timestamps and any date/time-related context. + - Add only genuinely new or meaningfully enriching information. + - Reflections MUST focus on forming reusable cognitive frameworks based on user feedback, aiming to improve future task execution. + - Keep entries concise yet complete. + - If there’s nothing to store or reflect on, respond with [SILENT]. user_message_zh: | - 预压缩内存刷新轮次。 - 当前会话即将进入自动压缩阶段;请将持久化记忆捕获并写入磁盘。 + 预压缩内存刷新轮次。 - 当前日期:{date} - 工作目录:{working_dir} + 当前会话即将进入自动压缩阶段;请将持久化记忆与经验反思捕获并写入磁盘。 - 立即存储持久化记忆(使用路径 {memory_dir}/YYYY-MM-DD.md)。 + 当前日期:{date} + 工作目录:{working_dir} - 工作流程: - 1. 先 `read` {memory_dir}/YYYY-MM-DD.md(如文件不存在,会返回错误提示) - 2. 智能合并新信息与现有内容(若文件不存在则跳过合并): - - 避免重复已记录的信息 - - 在相关时丰富现有条目的新细节 - - 在适用时保持时间顺序 - 3. 写入更新后的内容: - - 尽可能使用 `edit` 更新特定部分 - - 如需大幅重构则使用 `write` 覆盖整个文件 + 立即存储持久化记忆与反思(使用路径 {memory_dir}/YYYY-MM-DD.md)。 - 原则: - - 始终保留时间戳、日期和时间相关上下文 - - 仅添加真正新的或有丰富价值的信息 - - 保持条目简洁但完整 - - 若无内容可存储,请回复 [SILENT] + 工作流程: + 1. 先 `read` {memory_dir}/YYYY-MM-DD.md(如文件不存在,会返回错误提示) + 2. 从当前会话中提取并综合两类内容: + - 持久化记忆:客观事实、用户信息更新、项目状态及重要事件。 + - 经验反思:基于用户反馈形成的可复用思考逻辑、成功的问题解决策略、犯下的错误/应避免的陷阱,以及对未来交互有帮助的行动指南。 + 3. 智能合并新信息与现有内容(若文件不存在则跳过合并): + - 将内容进行清晰的分类(例如明确区分“事实记忆”与“反思与逻辑”)。 + - 避免重复已记录的信息。 + - 在相关时丰富现有条目的新细节。 + - 在适用时保持时间顺序。 + 4. 写入更新后的内容: + - 尽可能使用 `edit` 更新特定部分。 + - 如需大幅重构则使用 `write` 覆盖整个文件。 + + 原则: + - 始终保留时间戳、日期和时间相关上下文。 + - 仅添加真正新的或有丰富价值的信息。 + - 反思内容必须侧重于根据用户反馈构建可复用的思维逻辑,以改善未来的任务执行。 + - 保持条目简洁但完整。 + - 若无任何新内容可存储或反思,请回复 [SILENT]。 \ No newline at end of file diff --git a/reme/memory/file_based/components/tool_result_compactor.py b/reme/memory/file_based/components/tool_result_compactor.py index c1c6fe00..002bb8ae 100644 --- a/reme/memory/file_based/components/tool_result_compactor.py +++ b/reme/memory/file_based/components/tool_result_compactor.py @@ -8,10 +8,26 @@ from agentscope.message import Msg from ....core.op import BaseOp from ....core.utils import get_logger -from ....core.utils import truncate_text, is_truncated +from ....core.utils import truncate_text_head, TRUNCATION_MARKER_START logger = get_logger() +MAX_LINE_LENGTH = 10000 + + +def _split_long_lines(text: str, max_len: int = MAX_LINE_LENGTH) -> str: + """Split lines that exceed max_len by inserting newlines.""" + lines = text.split("\n") + result = [] + for line in lines: + if len(line) <= max_len: + result.append(line) + else: + # Split line into chunks of max_len + for i in range(0, len(line), max_len): + result.append(line[i : i + max_len]) + return "\n".join(result) + class ToolResultCompactor(BaseOp): """Truncate large tool_result outputs and save full content to files.""" @@ -19,43 +35,59 @@ class ToolResultCompactor(BaseOp): def __init__( self, tool_result_dir: str | Path, - tool_result_threshold: int, retention_days: int = 7, + recent_n: int = 1, + old_threshold: int = 500, + recent_threshold: int = 30000, **kwargs, ): super().__init__(**kwargs) self.tool_result_dir = Path(tool_result_dir) - self.tool_result_threshold = tool_result_threshold self.retention_days = retention_days + self.recent_n = recent_n + self.old_threshold = old_threshold + self.recent_threshold = recent_threshold - def _save_and_truncate(self, content: str, tool_name: str) -> str: + def _save_and_truncate(self, content: str, tool_name: str, threshold: int) -> str: """Save full content to file and return truncated version with file reference.""" - if not content or is_truncated(content) or len(content) <= self.tool_result_threshold: + if not content: return content - # Save full content + # Check if content was previously truncated + if TRUNCATION_MARKER_START in content: + parts = content.split(TRUNCATION_MARKER_START, 1) + if len(parts[0]) <= threshold: + return content + return f"{truncate_text_head(parts[0], threshold)}{parts[1]}" + + # Not truncated before + if len(content) <= threshold: + return content + + # Save full content with long lines split self.tool_result_dir.mkdir(parents=True, exist_ok=True) file_path = self.tool_result_dir / f"{uuid.uuid4().hex}.txt" created_at = datetime.now().isoformat() + processed_content = _split_long_lines(content) file_path.write_text( - f"# tool_name: {tool_name}\n# created_at: {created_at}\n# ---\n{content}", + f"# tool_name: {tool_name}\n# created_at: {created_at}\n# ---\n{processed_content}", encoding="utf-8", ) logger.debug("Saved tool result to %s (len=%d)", file_path, len(content)) # Return truncated with file reference - return f"{truncate_text(content, self.tool_result_threshold)}\n\n[Full content saved to: {file_path}]" + return f"{truncate_text_head(content, threshold)}\n\n[Full content saved to: {file_path}]" - def _process_output(self, output: str | list[dict], tool_name: str) -> str | list[dict]: + def _process_output(self, output: str | list[dict], tool_name: str, threshold: int) -> str | list[dict]: """Process tool result output, truncating if necessary.""" if isinstance(output, str): - return self._save_and_truncate(output, tool_name) + return self._save_and_truncate(output, tool_name, threshold) if isinstance(output, list): return [ ( - {**b, "text": self._save_and_truncate(b.get("text", ""), tool_name)} + {**b, "text": self._save_and_truncate(b.get("text", ""), tool_name, threshold)} if isinstance(b, dict) and b.get("type") == "text" else b ) @@ -69,15 +101,21 @@ class ToolResultCompactor(BaseOp): if not messages: return messages - for msg in messages: + # Split messages into old and recent parts + split_index = max(0, len(messages) - self.recent_n) + + for idx, msg in enumerate(messages): if not isinstance(msg.content, list): continue + # Determine threshold based on message position + threshold = self.recent_threshold if idx >= split_index else self.old_threshold + for block in msg.content: if isinstance(block, dict) and block.get("type") == "tool_result": output = block.get("output") if output: - block["output"] = self._process_output(output, block.get("name", "unknown")) + block["output"] = self._process_output(output, block.get("name", "unknown"), threshold) return messages diff --git a/reme/memory/file_based/reme_in_memory_memory.py b/reme/memory/file_based/reme_in_memory_memory.py index 4fcec627..5882b900 100644 --- a/reme/memory/file_based/reme_in_memory_memory.py +++ b/reme/memory/file_based/reme_in_memory_memory.py @@ -110,34 +110,19 @@ class ReMeInMemoryMemory(InMemoryMemory): async def get_memory( self, - mark: str | None = None, - exclude_mark: str | None = _MemoryMark.COMPRESSED, prepend_summary: bool = True, **_kwargs, ) -> list[Msg]: """Get the messages from the memory by mark (if provided). Args: - mark: Optional mark to filter messages - exclude_mark: Optional mark to exclude messages prepend_summary: Whether to prepend compressed summary **_kwargs: Additional keyword arguments (ignored) Returns: List of filtered messages """ - if not (mark is None or isinstance(mark, str)): - raise TypeError(f"The mark should be a string or None, but got {type(mark)}.") - - if not (exclude_mark is None or isinstance(exclude_mark, str)): - raise TypeError(f"The exclude_mark should be a string or None, but got {type(exclude_mark)}.") - - # Filter messages based on mark - filtered_content = [(msg, marks) for msg, marks in self.content if mark is None or mark in marks] - - # Further filter messages based on exclude_mark - if exclude_mark is not None: - filtered_content = [(msg, marks) for msg, marks in filtered_content if exclude_mark not in marks] + filtered_content = [(msg, marks) for msg, marks in self.content if _MemoryMark.COMPRESSED not in marks] if prepend_summary and self._compressed_summary: previous_summary = f""" @@ -145,8 +130,7 @@ class ReMeInMemoryMemory(InMemoryMemory): {self._compressed_summary} The above is a summary of our previous conversation. -If there is a new instruction from the user, do not continue executing the previous content; -only execute the user's new instruction. +Use it as context to maintain continuity. """.strip() return [ @@ -210,7 +194,7 @@ only execute the user's new instruction. if not messages: return 0 - # Persist messages to dialog storage + # Persist messages to dialog storage instead of compressed self._append_messages_to_dialog(messages) # Remove messages from memory @@ -258,10 +242,7 @@ only execute the user's new instruction. - context_usage_ratio: Usage percentage - messages_detail: List of per-message AsMsgStat objects """ - messages = await self.get_memory( - exclude_mark=_MemoryMark.COMPRESSED, - prepend_summary=False, - ) + messages = await self.get_memory(prepend_summary=False) compressed_summary = self.get_compressed_summary() compressed_summary_tokens = await self._msg_handler.count_str_token(compressed_summary) diff --git a/reme/reme_light.py b/reme/reme_light.py index c41320c3..58008120 100644 --- a/reme/reme_light.py +++ b/reme/reme_light.py @@ -61,8 +61,6 @@ class ReMeLight(Application): dialog_path (Path): Path to the dialog storage directory for raw conversation records. vector_weight (float): Weight for vector search in hybrid search (0-1). candidate_multiplier (float): Multiplier for candidate retrieval count. - tool_result_threshold (int): Character threshold for tool result compaction. - retention_days (int): Number of days to retain tool result files. summary_tasks (list[asyncio.Task]): List of active background summary tasks. """ @@ -78,8 +76,7 @@ class ReMeLight(Application): default_file_store_config: dict | None = None, vector_weight: float = 0.7, candidate_multiplier: float = 3.0, - tool_result_threshold: int = 1000, - retention_days: int = 7, + enable_load_env: bool = False, ): """ Initialize the ReMeLight application. @@ -110,11 +107,8 @@ class ReMeLight(Application): candidate_multiplier (float): Multiplier applied to max_results when retrieving candidates for re-ranking. Default 3.0 means 3x more candidates are retrieved than the final result count. - tool_result_threshold (int): Character count threshold for tool result - compaction. Results exceeding this length will be truncated and - saved to files. Default 1000 characters. - retention_days (int): Number of days to retain tool result files - before automatic cleanup. Default 7 days. + enable_load_env (bool): Whether to load environment variables from + .env file. Defaults to False. Note: The following directory structure will be created: @@ -135,8 +129,6 @@ class ReMeLight(Application): self.vector_weight: float = vector_weight self.candidate_multiplier: float = candidate_multiplier - self.tool_result_threshold: int = tool_result_threshold - self.retention_days: int = retention_days # Initialize the parent Application class with comprehensive configuration super().__init__( @@ -148,6 +140,7 @@ class ReMeLight(Application): config_path="light", enable_logo=False, log_to_console=False, + enable_load_env=enable_load_env, parser=ReMeConfigParser, default_as_llm_config=default_as_llm_config, default_embedding_model_config=default_embedding_model_config, @@ -182,19 +175,15 @@ class ReMeLight(Application): Clean up expired tool result files from the tool result directory. This method removes tool result files that have exceeded the retention - period specified during initialization. It helps manage disk space by - automatically removing old, unused tool outputs. + period. It helps manage disk space by automatically removing old, unused + tool outputs. Returns: int: The number of files that were successfully deleted """ try: - # Create a compactor instance with current configuration - compactor = ToolResultCompactor( - tool_result_dir=self.tool_result_path, - tool_result_threshold=self.tool_result_threshold, - retention_days=self.retention_days, - ) + # Create a compactor instance with default configuration + compactor = ToolResultCompactor(tool_result_dir=self.tool_result_path) # Execute cleanup and return count of deleted files return compactor.cleanup_expired_files() except Exception as e: @@ -239,7 +228,14 @@ class ReMeLight(Application): self._cleanup_tool_results() return await super().close() - async def compact_tool_result(self, messages: list[Msg]) -> list[Msg]: + async def compact_tool_result( + self, + messages: list[Msg], + recent_n: int = 1, + old_threshold: int = 500, + recent_threshold: int = 30000, + retention_days: int = 7, + ) -> list[Msg]: """ Compact tool results by truncating large outputs and saving full content to files. @@ -251,13 +247,19 @@ class ReMeLight(Application): Args: messages (list[Msg]): List of messages potentially containing tool results that may need compaction. + recent_n (int): Number of recent messages to use recent_threshold for. + Default 1. + old_threshold (int): Character threshold for old messages. Default 500. + recent_threshold (int): Character threshold for recent messages. Default 30000. + retention_days (int): Number of days to retain tool result files. + Default 7. Returns: list[Msg]: The processed list of messages with large tool results compacted. If an error occurs, returns the original unmodified messages. Note: - - Tool results shorter than tool_result_threshold are left unchanged + - Tool results are truncated based on old_threshold/recent_threshold - Full content of truncated results is saved to tool_result_path - Expired files are automatically cleaned up during this operation """ @@ -265,8 +267,10 @@ class ReMeLight(Application): # Create compactor with instance configuration compactor = ToolResultCompactor( tool_result_dir=self.tool_result_path, - tool_result_threshold=self.tool_result_threshold, - retention_days=self.retention_days, + retention_days=retention_days, + recent_n=recent_n, + old_threshold=old_threshold, + recent_threshold=recent_threshold, ) # Execute compaction and get processed messages @@ -402,6 +406,7 @@ class ReMeLight(Application): language: str = "zh", max_input_length: float = 128 * 1024, compact_ratio: float = 0.7, + timezone: str | None = None, ) -> str: """ Generate a comprehensive summary of the given messages. @@ -426,6 +431,8 @@ class ReMeLight(Application): Defaults to 128K tokens. compact_ratio (float): Ratio used to calculate compaction threshold. Defaults to 0.7. + timezone (str | None): Timezone string for date formatting + (e.g., "America/Chicago"). Defaults to system local time if None. Returns: str: The generated summary text, or an empty string if an error occurred. @@ -451,6 +458,7 @@ class ReMeLight(Application): as_llm_formatter=as_llm_formatter, as_token_counter=as_token_counter, language=language if language == "zh" else "", + timezone=timezone, ) return await summarizer.call(messages=messages, service_context=self.service_context) diff --git a/tests/light/test_reme_light.py b/tests/light/test_reme_light.py index fd884189..4fc7452a 100644 --- a/tests/light/test_reme_light.py +++ b/tests/light/test_reme_light.py @@ -34,6 +34,7 @@ async def main(): default_as_llm_config={"model_name": "qwen3.5-35b-a3b"}, # default_embedding_model_config={"model_name": "text-embedding-v4"}, default_file_store_config={"fts_enabled": True, "vector_enabled": False}, + enable_load_env=True, ) logging.getLogger("reme").setLevel(logging.WARNING) await reme.start()