scriptoriumadmin/ReMe
1
0
Fork 0
mirror of https://github.com/agentscope-ai/ReMe.git synced 2026-09-12 23:01:15 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

feat(core): refactor application architecture with new component base classes

Browse source 
- Add BaseClient, BaseFileStore, BaseFileWatcher, BaseJob, BaseService, and BaseStep classes
- Move component initialization logic from ApplicationContext to Application class
- Add logo printing and logging initialization in Application startup
- Create client module with base client implementation
- Add file store base class with embedding resolution and validation
- Implement file watcher base class with watchfiles integration
- Add job base class for sequential step execution orchestration
- Create service base class for job exposure mechanisms
- Refactor BaseStep with LLM workflow execution capabilities
- Add case converter utility for naming convention transformations
- Update import structure and module organization
- Add proper type hints and docstrings across all components
- Implement component registry integration for dynamic loading
- Add error handling for missing backend configurations
This commit is contained in:
jinli.yl 2026-04-14 16:43:36 +08:00
parent b9ce64de7f
commit 5fc57e7f4e
33 changed files with 442 additions and 287 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
pyproject.toml
View file

@ -127,4 +127,9 @@ python_functions = ["test_*"]
# Exclude script-style tests that require manual execution
addopts = "--ignore=tests/test_embedding.py --ignore=tests/test_embedding_cache.py --ignore=tests/test_embedding_sync.py --ignore=tests/test_file_store.py"
[tool.pylint.messages_control]
disable = [
"unused-argument",
]
# python -m build && twine upload dist/*

75
reme_cli/application.py
View file

@ -1,9 +1,13 @@
"""Application module for managing the main application lifecycle."""
import asyncio
from pathlib import Path
from typing import AsyncGenerator
from enumeration import ComponentEnum
from .component import BaseComponent, ApplicationContext
from .schema import Response, StreamChunk
from .utils import execute_stream_task
from .utils import execute_stream_task, print_logo, get_logger
class Application(BaseComponent):
@ -13,8 +17,65 @@ class Application(BaseComponent):
super().__init__()
self.context = ApplicationContext(**kwargs)
working_path = Path(self.config.working_dir).absolute()
working_path.mkdir(parents=True, exist_ok=True)
memory_path = working_path / "memory"
memory_path.mkdir(parents=True, exist_ok=True)
if self.config.enable_logo:
print_logo(self.config)
logger = get_logger(
log_to_console=self.config.log_to_console,
log_to_file=self.config.log_to_file,
force_init=True,
)
logger.info(f"Initializing {self.config.app_name} Application")
from .component import R
# Initialize the service
service_config = self.config.service
if not service_config.backend:
raise ValueError("Service configuration is missing the required 'backend' field")
service_cls = R.get(ComponentEnum.SERVICE, service_config.backend)
if not service_cls:
raise ValueError(
f"Service references an unregistered backend '{service_config.backend}' "
f"of type '{ComponentEnum.SERVICE}'",
)
self.context.service = service_cls(**service_config.model_dump(exclude={"backend"}))
# Initialize all components grouped by type and name
for component_type, component_configs in self.config.components.items():
self.context.components[component_type] = {}
for name, config in component_configs.items():
if not config.backend:
raise ValueError(f"Component '{name}' is missing the required 'backend' field")
backend_cls = R.get(component_type, config.backend)
if not backend_cls:
raise ValueError(
f"Component '{name}' references an unregistered backend '{config.backend}' "
f"of type '{component_type}'",
)
self.context.components[component_type][name] = backend_cls(**config.model_dump(exclude={"backend"}))
# Initialize all jobs
for job_config in self.config.jobs:
if not job_config.backend:
raise ValueError(f"Job '{job_config.name}' is missing the required 'backend' field")
job_cls = R.get(ComponentEnum.JOB, job_config.backend)
if not job_cls:
raise ValueError(
f"Job '{job_config.name}' references an unregistered backend '{job_config.backend}' "
f"of type '{ComponentEnum.JOB}'",
)
self.context.jobs[job_config.name] = job_cls(**job_config.model_dump(exclude={"backend"}))
@property
def config(self):
"""Get application configuration."""
return self.context.app_config
async def _start(self, app_context=None) -> None:
@ -48,22 +109,24 @@ class Application(BaseComponent):
self.logger.exception(f"Failed to close component {component.__class__.__name__}: {e}")
async def run_job(self, name: str, **kwargs) -> Response:
"""Execute a registered job by name."""
if name not in self.context.jobs:
raise KeyError(f"Job '{name}' not found")
job = self.context.jobs[name]
return await job(app_context=self.context, **kwargs)
async def run_stream_job(self, name: str, **kwargs) -> AsyncGenerator[StreamChunk, None]:
"""Execute a streaming job and yield chunks."""
if name not in self.context.jobs:
raise KeyError(f"Job '{name}' not found")
job = self.context.jobs[name]
stream_queue = asyncio.Queue()
task = asyncio.create_task(job(stream_queue=stream_queue, app_context=self.context, **kwargs))
async for chunk in execute_stream_task(
stream_queue=stream_queue,
task=task,
task_name=name,
output_format="chunk",
stream_queue=stream_queue,
task=task,
task_name=name,
output_format="chunk",
):
assert isinstance(chunk, StreamChunk)
yield chunk
@ -71,4 +134,4 @@ class Application(BaseComponent):
def run_app(self):
"""Run the application as a service."""
service = self.context.service
service.run_app(app=self)
service.run_app(app=self)

1
reme_cli/component/__init__.py
View file

@ -1,4 +1,5 @@
"""Components"""
from .application_context import ApplicationContext
from .base_component import BaseComponent
from .base_step import BaseStep

40
reme_cli/component/application_context.py
View file

@ -25,47 +25,9 @@ class ApplicationContext:
"""
self.app_config: ApplicationConfig = ApplicationConfig(**kwargs)
from .component_registry import R
from .base_component import BaseComponent
from .job.base_job import BaseJob
# Initialize the service
service_config = self.app_config.service
if not service_config.backend:
raise ValueError("Service configuration is missing the required 'backend' field")
service_cls = R.get(ComponentEnum.SERVICE, service_config.backend)
if not service_cls:
raise ValueError(
f"Service references an unregistered backend '{service_config.backend}' "
f"of type '{ComponentEnum.SERVICE}'"
)
self.service = service_cls(**service_config.model_dump(exclude={"backend"}))
# Initialize all components grouped by type and name
self.service = None
self.components: dict[ComponentEnum, dict[str, BaseComponent]] = {}
for component_type, component_configs in self.app_config.components.items():
self.components[component_type] = {}
for name, config in component_configs.items():
if not config.backend:
raise ValueError(f"Component '{name}' is missing the required 'backend' field")
backend_cls = R.get(component_type, config.backend)
if not backend_cls:
raise ValueError(
f"Component '{name}' references an unregistered backend '{config.backend}' "
f"of type '{component_type}'"
)
self.components[component_type][name] = backend_cls(**config.model_dump(exclude={"backend"}))
# Initialize all jobs
self.jobs: dict[str, BaseJob] = {}
for job_config in self.app_config.jobs:
if not job_config.backend:
raise ValueError(f"Job '{job_config.name}' is missing the required 'backend' field")
job_cls = R.get(ComponentEnum.JOB, job_config.backend)
if not job_cls:
raise ValueError(
f"Job '{job_config.name}' references an unregistered backend '{job_config.backend}' "
f"of type '{ComponentEnum.JOB}'"
)
self.jobs[job_config.name] = job_cls(**job_config.model_dump(exclude={"backend"}))

63
reme_cli/component/as_llm_formatter/reme_openai_chat_formatter.py
View file

@ -35,8 +35,8 @@ class ReMeOpenAIChatFormatter(OpenAIChatFormatter):
"""Extends OpenAIChatFormatter with tool result image promotion and reasoning content support."""
async def _format(
self,
msgs: list[Msg],
self,
msgs: list[Msg],
) -> list[dict[str, Any]]:
"""Format messages into OpenAI API format.
@ -69,40 +69,46 @@ class ReMeOpenAIChatFormatter(OpenAIChatFormatter):
reasoning_content_blocks.append({**block})
elif typ == "tool_use":
tool_calls.append({
"id": block.get("id"),
"type": "function",
"function": {
"name": block.get("name"),
"arguments": json.dumps(block.get("input", {}), ensure_ascii=False),
tool_calls.append(
{
"id": block.get("id"),
"type": "function",
"function": {
"name": block.get("name"),
"arguments": json.dumps(block.get("input", {}), ensure_ascii=False),
},
},
})
)
elif typ == "tool_result":
textual_output, multimodal_data = self.convert_tool_result_to_string(block["output"])
messages.append({
"role": "tool",
"tool_call_id": block.get("id"),
"content": textual_output,
"name": block.get("name"),
})
messages.append(
{
"role": "tool",
"tool_call_id": block.get("id"),
"content": textual_output,
"name": block.get("name"),
},
)
# Promote tool result images into a follow-up user message
promoted_blocks = []
for url, multimodal_block in multimodal_data:
if multimodal_block["type"] == "image" and self.promote_tool_result_images:
promoted_blocks.extend([
TextBlock(type="text", text=f"\n- The image from '{url}': "),
ImageBlock(type="image", source=URLSource(type="url", url=url)),
])
promoted_blocks.extend(
[
TextBlock(type="text", text=f"\n- The image from '{url}': "),
ImageBlock(type="image", source=URLSource(type="url", url=url)),
],
)
if promoted_blocks:
promoted_blocks = [
TextBlock(
type="text",
text="<system-info>The following are the image contents from the tool "
f"result of '{block['name']}':",
f"result of '{block['name']}':",
),
*promoted_blocks,
TextBlock(type="text", text="</system-info>"),
@ -119,10 +125,12 @@ class ReMeOpenAIChatFormatter(OpenAIChatFormatter):
# Skip assistant audio output
if msg.role == "assistant":
continue
content_blocks.append({
"type": "input_audio",
"input_audio": _to_openai_audio_data(block["source"]),
})
content_blocks.append(
{
"type": "input_audio",
"input_audio": _to_openai_audio_data(block["source"]),
},
)
elif typ == "video":
# Skip assistant video output
@ -131,8 +139,7 @@ class ReMeOpenAIChatFormatter(OpenAIChatFormatter):
content_blocks.append(_format_openai_video_block(block))
else:
...
# logger.warning("Unsupported block type %s, skipped.", typ)
pass # Unsupported block type, skip
msg_openai = {
"role": msg.role,
@ -144,9 +151,7 @@ class ReMeOpenAIChatFormatter(OpenAIChatFormatter):
msg_openai["tool_calls"] = tool_calls
if reasoning_content_blocks:
reasoning_msg = "\n".join(
r.get("thinking", "") for r in reasoning_content_blocks
)
reasoning_msg = "\n".join(r.get("thinking", "") for r in reasoning_content_blocks)
if reasoning_msg:
msg_openai["reasoning_content"] = reasoning_msg

9
reme_cli/component/base_component.py
View file

@ -35,6 +35,7 @@ class BaseComponent(ABC):
_is_started: Internal flag indicating whether the component has been
started and not yet closed.
"""
from .application_context import ApplicationContext
component_type = ComponentEnum.BASE
@ -181,10 +182,10 @@ class BaseComponent(ABC):
return self
async def __aexit__(
self,
exc_type: type[BaseException] | None,
exc_val: BaseException | None,
exc_tb,
self,
exc_type: type[BaseException] | None,
exc_val: BaseException | None,
exc_tb,
) -> bool:
"""Exit the async context manager by closing the component.

20
reme_cli/component/base_step.py
View file

@ -29,13 +29,13 @@ class BaseStep(BaseComponent):
return instance
def __init__(
self,
name: str = "",
language: str = "",
prompt_dict: dict[str, str] | None = None,
input_mapping: dict[str, str] | None = None,
output_mapping: dict[str, str] | None = None,
**kwargs,
self,
name: str = "",
language: str = "",
prompt_dict: dict[str, str] | None = None,
input_mapping: dict[str, str] | None = None,
output_mapping: dict[str, str] | None = None,
**kwargs,
):
"""Initialize step configurations."""
super().__init__(**kwargs)
@ -73,15 +73,18 @@ class BaseStep(BaseComponent):
@property
def application_context(self) -> ApplicationContext:
"""Get the application context from runtime context."""
assert self.context is not None, "Runtime context not set."
return self.context.application_context
@property
def app_config(self) -> ApplicationConfig:
"""Get the application configuration."""
return self.application_context.app_config
@property
def as_llm(self) -> BaseAsLLM:
"""Get the AsLLM instance by name."""
name: str = self.kwargs.get("as_llm", "default")
llms = self.application_context.components[ComponentEnum.AS_LLM]
if name not in llms:
@ -93,6 +96,7 @@ class BaseStep(BaseComponent):
@property
def as_llm_formatter(self) -> BaseAsLLMFormatter:
"""Get the AsLLMFormatter instance by name."""
name: str = self.kwargs.get("as_llm_formatter", "default")
formatters = self.application_context.components[ComponentEnum.AS_LLM_FORMATTER]
if name not in formatters:
@ -104,6 +108,7 @@ class BaseStep(BaseComponent):
@property
def file_store(self) -> BaseFileStore:
"""Get the FileStore instance by name."""
name: str = self.kwargs.get("file_store", "default")
stores = self.application_context.components[ComponentEnum.FILE_STORE]
if name not in stores:
@ -115,6 +120,7 @@ class BaseStep(BaseComponent):
@property
def embedding(self) -> BaseEmbeddingModel:
"""Get the EmbeddingModel instance by name."""
name: str = self.kwargs.get("embedding", "default")
models = self.application_context.components[ComponentEnum.EMBEDDING_MODEL]
if name not in models:

2
reme_cli/component/client/__init__.py
View file

@ -6,4 +6,4 @@ from .http_client import HttpClient
__all__ = [
"BaseClient",
"HttpClient",
]
]

1
reme_cli/component/client/base_client.py
View file

@ -8,6 +8,7 @@ from ...enumeration import ComponentEnum
class BaseClient(BaseComponent):
"""Abstract base class for clients that communicate with ReMe services."""
component_type = ComponentEnum.CLIENT
def __init__(self, **kwargs) -> None:

13
reme_cli/component/client/http_client.py
View file

@ -2,7 +2,6 @@
import json
import os
from typing import Any
import httpx
@ -16,12 +15,12 @@ class HttpClient(BaseClient):
"""HTTP client for ReMe service."""
def __init__(
self,
action: str,
host: str | None = None,
port: int | None = None,
timeout: float = 30.0,
**kwargs
self,
action: str,
host: str | None = None,
port: int | None = None,
timeout: float = 30.0,
**kwargs,
):
super().__init__(**kwargs)

12
reme_cli/component/component_registry.py
View file

@ -24,7 +24,8 @@ class ComponentRegistry:
self.logger = get_logger()
def _do_register(self, cls: type[T], name: str) -> type[T]:
if not hasattr(cls, 'component_type'):
"""Register a component class with the given name."""
if not hasattr(cls, "component_type"):
raise TypeError(f"{cls.__name__} must have 'component_type' attribute")
if not name:
raise ValueError("Component name cannot be empty")
@ -37,8 +38,11 @@ class ComponentRegistry:
return cls
def register(
self, cls_or_name: type[T] | str, name: str | None = None
self,
cls_or_name: type[T] | str,
name: str | None = None,
) -> Callable[[type[T]], type[T]] | type[T]:
"""Register a component class. Supports direct and decorator modes."""
# Direct registration: R.register(MyClass, "name")
if isinstance(cls_or_name, type):
return self._do_register(cast(type[T], cls_or_name), name or cls_or_name.__name__)
@ -52,18 +56,22 @@ class ComponentRegistry:
return decorator
def get(self, component_type: ComponentEnum, name: str) -> type[BaseComponent] | None:
"""Get a registered component class by type and name."""
return self._registry.get(component_type, {}).get(name)
def get_all(self, component_type: ComponentEnum) -> dict[str, type[BaseComponent]]:
"""Get all registered components of a given type."""
return dict(self._registry.get(component_type, {}))
def unregister(self, component_type: ComponentEnum, name: str) -> bool:
"""Remove a component from the registry. Returns True if found."""
if name in self._registry.get(component_type, {}):
del self._registry[component_type][name]
return True
return False
def clear(self) -> None:
"""Clear all registered components."""
self._registry.clear()

48
reme_cli/component/embedding/base_embedding_model.py
View file

@ -22,24 +22,25 @@ class BaseEmbeddingModel(BaseComponent):
- Retry logic with exponential backoff
- Batch embedding support
"""
component_type = ComponentEnum.EMBEDDING_MODEL
def __init__(
self,
api_key: str | None = None,
base_url: str | None = None,
model_name: str = "",
dimensions: int = 1024,
use_dimensions: bool = False,
max_batch_size: int = 10,
max_retries: int = 3,
raise_exception: bool = True,
max_input_length: int = 8192,
cache_dir: str | Path = ".reme",
max_cache_size: int = 2000,
enable_cache: bool = True,
encoding: str = "utf-8",
**kwargs,
self,
api_key: str | None = None,
base_url: str | None = None,
model_name: str = "",
dimensions: int = 1024,
use_dimensions: bool = False,
max_batch_size: int = 10,
max_retries: int = 3,
raise_exception: bool = True,
max_input_length: int = 8192,
cache_dir: str | Path = ".reme",
max_cache_size: int = 2000,
enable_cache: bool = True,
encoding: str = "utf-8",
**kwargs,
):
"""Initialize embedding model configuration.
@ -90,12 +91,12 @@ class BaseEmbeddingModel(BaseComponent):
if actual_len < self.dimensions:
self.logger.warning(
f"[ACTUAL_EMB_LENGTH] Embedding {actual_len} < expected {self.dimensions}, padding with zeros"
f"[ACTUAL_EMB_LENGTH] Embedding {actual_len} < expected {self.dimensions}, padding with zeros",
)
return embedding + [0.0] * (self.dimensions - actual_len)
self.logger.warning(
f"[ACTUAL_EMB_LENGTH] Embedding {actual_len} > expected {self.dimensions}, truncating"
f"[ACTUAL_EMB_LENGTH] Embedding {actual_len} > expected {self.dimensions}, truncating",
)
return embedding[: self.dimensions]
@ -145,7 +146,7 @@ class BaseEmbeddingModel(BaseComponent):
if len(embedding) != self.dimensions:
self.logger.warning(
f"Cache dimension mismatch for {cache_key}: "
f"expected {self.dimensions}, got {len(embedding)}"
f"expected {self.dimensions}, got {len(embedding)}",
)
continue
if len(self._embedding_cache) >= self.max_cache_size:
@ -191,7 +192,7 @@ class BaseEmbeddingModel(BaseComponent):
embeddings = self._embedding_cache[cache_key]
if len(embeddings) != self.dimensions:
self.logger.warning(f"Cached embedding dimension mismatch, removing entry")
self.logger.warning("Cached embedding dimension mismatch, removing entry")
del self._embedding_cache[cache_key]
self._cache_misses += 1
return None
@ -255,7 +256,8 @@ class BaseEmbeddingModel(BaseComponent):
self._put_to_cache(truncated_text, embedding)
return embedding
self.logger.warning(
f"Model {self.model_name} returned {len(result) if result else 0} results, expected 1")
f"Model {self.model_name} returned {len(result) if result else 0} results, expected 1",
)
if retry == self.max_retries - 1:
if self.raise_exception:
raise RuntimeError("Embedding API returned empty result")
@ -286,8 +288,8 @@ class BaseEmbeddingModel(BaseComponent):
if texts_to_compute:
uncached_texts = [text for _, text in texts_to_compute]
for i in range(0, len(uncached_texts), self.max_batch_size):
batch_texts = uncached_texts[i: i + self.max_batch_size]
batch_indices = [idx for idx, _ in texts_to_compute[i: i + self.max_batch_size]]
batch_texts = uncached_texts[i : i + self.max_batch_size]
batch_indices = [idx for idx, _ in texts_to_compute[i : i + self.max_batch_size]]
for retry in range(self.max_retries):
try:
@ -300,7 +302,7 @@ class BaseEmbeddingModel(BaseComponent):
break
self.logger.warning(
f"Batch returned {len(batch_embeddings) if batch_embeddings else 0} "
f"results for {len(batch_texts)} inputs"
f"results for {len(batch_texts)} inputs",
)
if retry == self.max_retries - 1:
if self.raise_exception:

26
reme_cli/component/file_store/base_file_store.py
View file

@ -16,15 +16,16 @@ class BaseFileStore(BaseComponent):
Provides embedding resolution, validation, and safe embedding retrieval
with automatic fallback on failure.
"""
component_type = ComponentEnum.FILE_STORE
def __init__(
self,
store_name: str,
db_path: str | Path,
embedding_model: str = "default",
fts_enabled: bool = True,
**kwargs,
self,
store_name: str,
db_path: str | Path,
embedding_model: str = "default",
fts_enabled: bool = True,
**kwargs,
):
super().__init__(**kwargs)
self._embedding_model_name: str = embedding_model
@ -37,7 +38,8 @@ class BaseFileStore(BaseComponent):
if not re.match(r"^[a-zA-Z0-9_]+$", store_name):
raise ValueError(
f"Invalid store name '{store_name}'. Only alphanumeric characters and underscores are allowed.")
f"Invalid store name '{store_name}'. Only alphanumeric characters and underscores are allowed.",
)
if not self.vector_enabled and not self.fts_enabled:
raise ValueError("At least one of embedding_model or fts_enabled must be set.")
@ -153,11 +155,11 @@ class BaseFileStore(BaseComponent):
@abstractmethod
async def hybrid_search(
self,
query: str,
limit: int,
vector_weight: float = 0.7,
candidate_multiplier: float = 3.0,
self,
query: str,
limit: int,
vector_weight: float = 0.7,
candidate_multiplier: float = 3.0,
) -> list[FileChunk]:
"""Perform hybrid search combining vector and keyword results.

73
reme_cli/component/file_store/local_file_store.py
View file

@ -48,7 +48,7 @@ class LocalFileStore(BaseFileStore):
"""Persist chunks to JSONL file with atomic write."""
lines = [json.dumps(c.model_dump(mode="json"), ensure_ascii=False) for c in self._chunks.values()]
content = "\n".join(lines)
temp_path = self._chunks_file.with_suffix('.tmp')
temp_path = self._chunks_file.with_suffix(".tmp")
try:
temp_path.write_text(content, encoding=self._encoding)
temp_path.replace(self._chunks_file)
@ -72,10 +72,11 @@ class LocalFileStore(BaseFileStore):
async def _save_metadata(self) -> None:
"""Persist file metadata to JSON file with atomic write."""
raw = {path: meta.model_dump(exclude={"content", "metadata"}, mode="json")
for path, meta in self._files.items()}
raw = {
path: meta.model_dump(exclude={"content", "metadata"}, mode="json") for path, meta in self._files.items()
}
content = json.dumps(raw, indent=2, ensure_ascii=False)
temp_path = self._metadata_file.with_suffix('.tmp')
temp_path = self._metadata_file.with_suffix(".tmp")
try:
temp_path.write_text(content, encoding=self._encoding)
temp_path.replace(self._metadata_file)
@ -141,9 +142,7 @@ class LocalFileStore(BaseFileStore):
for cid in chunk_ids:
self._chunks.pop(cid, None)
if path in self._files:
self._files[path].chunk_count = sum(
1 for chunk in self._chunks.values() if chunk.path == path
)
self._files[path].chunk_count = sum(1 for chunk in self._chunks.values() if chunk.path == path)
async def upsert_chunks(self, chunks: list[FileChunk]) -> None:
"""Insert or update specific chunks without affecting others."""
@ -211,16 +210,18 @@ class LocalFileStore(BaseFileStore):
results = []
for chunk, sim in zip(candidates, similarities):
results.append(FileChunk(
id=chunk.id,
path=chunk.path,
start_line=chunk.start_line,
end_line=chunk.end_line,
hash=chunk.hash,
text=chunk.text,
embedding=chunk.embedding,
scores={"vector": float(sim), "score": float(sim)},
))
results.append(
FileChunk(
id=chunk.id,
path=chunk.path,
start_line=chunk.start_line,
end_line=chunk.end_line,
hash=chunk.hash,
text=chunk.text,
embedding=chunk.embedding,
scores={"vector": float(sim), "score": float(sim)},
),
)
results.sort(key=lambda r: r.score, reverse=True)
return results[:limit]
@ -249,25 +250,27 @@ class LocalFileStore(BaseFileStore):
phrase_bonus = 0.2 if n_words > 1 and query_lower in text_lower else 0.0
score = min(1.0, base_score + phrase_bonus)
results.append(FileChunk(
id=chunk.id,
path=chunk.path,
start_line=chunk.start_line,
end_line=chunk.end_line,
hash=chunk.hash,
text=chunk.text,
scores={"keyword": score, "score": score},
))
results.append(
FileChunk(
id=chunk.id,
path=chunk.path,
start_line=chunk.start_line,
end_line=chunk.end_line,
hash=chunk.hash,
text=chunk.text,
scores={"keyword": score, "score": score},
),
)
results.sort(key=lambda r: r.score, reverse=True)
return results[:limit]
async def hybrid_search(
self,
query: str,
limit: int,
vector_weight: float = 0.7,
candidate_multiplier: float = 3.0,
self,
query: str,
limit: int,
vector_weight: float = 0.7,
candidate_multiplier: float = 3.0,
) -> list[FileChunk]:
"""Hybrid search combining vector and keyword results."""
assert 0.0 <= vector_weight <= 1.0
@ -299,10 +302,10 @@ class LocalFileStore(BaseFileStore):
@staticmethod
def _merge_hybrid_results(
vector: list[FileChunk],
keyword: list[FileChunk],
vector_weight: float,
text_weight: float,
vector: list[FileChunk],
keyword: list[FileChunk],
vector_weight: float,
text_weight: float,
) -> list[FileChunk]:
"""Merge vector and keyword results with weighted scoring."""
merged: dict[str, FileChunk] = {}

34
reme_cli/component/file_watcher/base_file_watcher.py
View file

@ -24,17 +24,17 @@ class BaseFileWatcher(BaseComponent):
component_type = ComponentEnum.FILE_WATCHER
def __init__(
self,
watch_paths: list[str] | str,
suffix_filters: list[str] | None = None,
recursive: bool = False,
debounce: int = 2000,
chunk_tokens: int = 400,
chunk_overlap: int = 80,
file_store: str = "default",
rebuild_index_on_start: bool = True,
poll_delay_ms: int = 2000,
**kwargs,
self,
watch_paths: list[str] | str,
suffix_filters: list[str] | None = None,
recursive: bool = False,
debounce: int = 2000,
chunk_tokens: int = 400,
chunk_overlap: int = 80,
file_store: str = "default",
rebuild_index_on_start: bool = True,
poll_delay_ms: int = 2000,
**kwargs,
):
"""Initialize file watcher configuration.
@ -179,12 +179,12 @@ class BaseFileWatcher(BaseComponent):
try:
self.logger.info(f"Starting watch on: {valid_paths}")
async for changes in awatch(
*valid_paths,
watch_filter=self.watch_filter,
recursive=self.recursive,
debounce=self.debounce,
poll_delay_ms=self.poll_delay_ms,
stop_event=self._stop_event,
*valid_paths,
watch_filter=self.watch_filter,
recursive=self.recursive,
debounce=self.debounce,
poll_delay_ms=self.poll_delay_ms,
stop_event=self._stop_event,
):
if self._stop_event.is_set():
break

14
reme_cli/component/file_watcher/md_file_watcher.py
View file

@ -35,13 +35,13 @@ class MdFileWatcher(BaseFileWatcher):
if change_type in [Change.added, Change.modified]:
file_meta = await self._build_file_metadata(path)
chunks = (
chunk_markdown(
file_meta.content,
file_meta.path,
self.chunk_tokens,
self.chunk_overlap,
)
or []
chunk_markdown(
file_meta.content,
file_meta.path,
self.chunk_tokens,
self.chunk_overlap,
)
or []
)
if chunks:
chunks = await self.file_store.get_chunk_embeddings(chunks)

18
reme_cli/component/job/base_job.py
View file

@ -15,15 +15,16 @@ class BaseJob(BaseComponent):
through each step. Steps are configured via ComponentConfig and instantiated
lazily when the job starts.
"""
component_type = ComponentEnum.JOB
def __init__(
self,
name: str = "",
description: str = "",
parameters: dict | None = None,
steps: list[ComponentConfig] | None = None,
**kwargs
self,
name: str = "",
description: str = "",
parameters: dict | None = None,
steps: list[ComponentConfig] | None = None,
**kwargs,
):
"""Initialize the job.
@ -59,7 +60,10 @@ class BaseJob(BaseComponent):
if not backend_cls:
raise ValueError(f"{step_config.backend} is not registered.")
step = backend_cls(**step_config.model_dump(exclude={"backend"}))
step = backend_cls(
language=app_context.app_config.language,
**step_config.model_dump(exclude={"backend"}),
)
self.steps.append(step)
async def _close(self) -> None:

9
reme_cli/component/prompt_handler.py
View file

@ -17,8 +17,11 @@ class PromptHandler:
self.data: dict[str, str] = {k: v for k, v in kwargs.items() if isinstance(v, str)}
self.language: str = language.strip()
def load_prompt_by_file(self, prompt_file_path: str | Path | None = None,
overwrite: bool = True) -> "PromptHandler":
def load_prompt_by_file(
self,
prompt_file_path: str | Path | None = None,
overwrite: bool = True,
) -> "PromptHandler":
"""Load prompts from a YAML or JSON file."""
if prompt_file_path is None:
return self
@ -92,7 +95,7 @@ class PromptHandler:
for flag, enabled in flags.items():
prefix = f"[{flag}]"
while remaining.startswith(prefix):
remaining = remaining[len(prefix):]
remaining = remaining[len(prefix) :]
if enabled:
should_include = True
if should_include or not any(line.startswith(f"[{f}]") for f in flags):

6
reme_cli/component/runtime_context.py
View file

@ -74,9 +74,9 @@ class RuntimeContext:
return self
def validate_required_keys(
self,
required_keys: dict[str, bool],
context_name: str = "context",
self,
required_keys: dict[str, bool],
context_name: str = "context",
) -> "RuntimeContext":
"""Ensure all required keys are present in the context.

2
reme_cli/component/service/base_service.py
View file

@ -13,6 +13,7 @@ class BaseService(BaseComponent):
Services provide different ways to invoke jobs (HTTP, CLI, MCP, etc.).
Subclasses must implement add_job to register jobs with the service.
"""
component_type = ComponentEnum.SERVICE
from ...application import Application
@ -53,6 +54,7 @@ class BaseService(BaseComponent):
"""Start the service."""
def add_jobs(self, app: "Application") -> None:
"""Register all jobs from the application context."""
for name, job in app.context.jobs.values():
try:
self.add_job(job)

25
reme_cli/component/service/http_service.py
View file

@ -26,6 +26,7 @@ class HttpService(BaseService):
Regular jobs return JSON responses, while StreamJobs return
server-sent events (SSE) for real-time streaming.
"""
from ...application import Application
def __init__(self, host: str = REME_DEFAULT_HOST, port: int = REME_DEFAULT_PORT, **kwargs):
@ -66,15 +67,15 @@ class HttpService(BaseService):
async def execute_stream_endpoint(request: Request) -> StreamingResponse:
stream_queue = asyncio.Queue()
task = asyncio.create_task(
job(stream_queue=stream_queue, **request.model_dump(exclude_none=True))
job(stream_queue=stream_queue, **request.model_dump(exclude_none=True)),
)
async def generate_stream() -> AsyncGenerator[bytes, None]:
async for chunk in execute_stream_task(
stream_queue=stream_queue,
task=task,
task_name=job.name,
output_format="bytes",
stream_queue=stream_queue,
task=task,
task_name=job.name,
output_format="bytes",
):
assert isinstance(chunk, bytes)
yield chunk
@ -96,7 +97,7 @@ class HttpService(BaseService):
else:
self._add_job(job)
def build_service(self, app: "Application") -> None:
def build_service(self, app: Application) -> None:
"""Build the FastAPI application with CORS middleware.
Args:
@ -106,10 +107,12 @@ class HttpService(BaseService):
@asynccontextmanager
async def lifespan(_: FastAPI):
await app.start()
service_info = json.dumps({
"host": self.host,
"port": self.port,
})
service_info = json.dumps(
{
"host": self.host,
"port": self.port,
},
)
os.environ[REME_SERVICE_INFO] = service_info
self.logger.info(f"ReMe Service started: {REME_SERVICE_INFO}={service_info}")
yield
@ -126,7 +129,7 @@ class HttpService(BaseService):
)
self.service.post("/health")(lambda: {"status": "healthy"})
def start_service(self, app: "Application") -> None:
def start_service(self, app: Application) -> None:
"""Start the HTTP server.
Args:

1
reme_cli/config/__init__.py
View file

@ -1,4 +1,5 @@
"""Config"""
from .config_parser import parse_args
__all__ = [

34
reme_cli/config/config_parser.py
View file

@ -48,36 +48,24 @@ def _convert_value(value_str: str) -> Any:
Use JSON format (e.g., '"yes"', '"no"') to preserve these as strings.
"""
s = value_str.strip()
# Null
if s.lower() in ("none", "null"):
return None
# Boolean: only accept strict true/false to avoid surprising conversions
lower = s.lower()
# Handle special values (null, bool)
if lower in ("none", "null"):
return None
if lower == "true":
return True
if lower == "false":
return False
# Int
try:
return int(s)
except ValueError:
pass
# Float
try:
return float(s)
except ValueError:
pass
# JSON (list, dict, quoted strings)
try:
return json.loads(s)
except json.JSONDecodeError:
pass
# Try numeric and JSON conversions
for converter in (int, float, json.loads):
try:
return converter(s)
except (ValueError, json.JSONDecodeError):
continue
# Fallback to string
return s

62
reme_cli/reme.py
View file

@ -1,3 +1,5 @@
"""ReMe CLI application entry point."""
import asyncio
import sys
@ -14,49 +16,51 @@ from .enumeration import ComponentEnum
class ReMe(Application):
"""ReMe memory management application."""
async def summary_memory(
self,
messages: list[Msg],
as_llm: str | ChatModelBase = "default",
as_llm_formatter: str | FormatterBase = "default",
as_token_counter: str | TokenCounterBase = "default",
toolkit: Toolkit | None = None,
language: str = "zh",
max_input_length: float = 128 * 1024,
compact_ratio: float = 0.7,
timezone: str | None = None,
add_thinking_block: bool = True,
self,
messages: list[Msg],
as_llm: str | ChatModelBase = "default",
as_llm_formatter: str | FormatterBase = "default",
as_token_counter: str | TokenCounterBase = "default",
toolkit: Toolkit | None = None,
language: str = "zh",
max_input_length: float = 128 * 1024,
compact_ratio: float = 0.7,
timezone: str | None = None,
add_thinking_block: bool = True,
) -> str:
...
"""Summarize and compact memory messages."""
async def memory_search(self, query: str, max_results: int = 5, min_score: float = 0.1) -> ToolResponse:
...
"""Search memory for relevant entries."""
async def dream(
self,
as_llm: str | ChatModelBase = "default",
as_llm_formatter: str | FormatterBase = "default",
as_token_counter: str | TokenCounterBase = "default",
toolkit: Toolkit | None = None,
language: str = "zh",
timezone: str | None = None,
self,
as_llm: str | ChatModelBase = "default",
as_llm_formatter: str | FormatterBase = "default",
as_token_counter: str | TokenCounterBase = "default",
toolkit: Toolkit | None = None,
language: str = "zh",
timezone: str | None = None,
) -> str:
...
"""Process and consolidate memories in background."""
async def proactive(
self,
as_llm: str | ChatModelBase = "default",
as_llm_formatter: str | FormatterBase = "default",
as_token_counter: str | TokenCounterBase = "default",
toolkit: Toolkit | None = None,
language: str = "zh",
timezone: str | None = None,
self,
as_llm: str | ChatModelBase = "default",
as_llm_formatter: str | FormatterBase = "default",
as_token_counter: str | TokenCounterBase = "default",
toolkit: Toolkit | None = None,
language: str = "zh",
timezone: str | None = None,
) -> str:
...
"""Generate proactive memory insights."""
def main():
"""Entry point for ReMe CLI."""
action, config = parse_args(sys.argv[1:])
if action == "app":
reme = ReMe(**config)

3
reme_cli/schema/application_config.py
View file

@ -75,5 +75,6 @@ class ApplicationConfig(BaseModel):
service: ComponentConfig = Field(default_factory=ComponentConfig, description="Service endpoint config")
jobs: list[JobConfig] = Field(default_factory=list, description="Job definitions")
components: dict[ComponentEnum, dict[str, ComponentConfig]] = Field(
default_factory=dict, description="Component registry by type"
default_factory=dict,
description="Component registry by type",
)

1
reme_cli/schema/response.py
View file

@ -20,6 +20,7 @@ class Response(BaseModel):
success: Whether the operation completed successfully.
metadata: Additional context and diagnostic information.
"""
model_config = ConfigDict(extra="allow")
answer: str | Any = Field(default="", description="Response content or result data")

2
reme_cli/utils/__init__.py
View file

@ -4,6 +4,7 @@ from .case_converter import camel_to_snake, snake_to_camel
from .chunking_utils import chunk_markdown
from .common_utils import hash_text, execute_stream_task
from .logger_utils import get_logger
from .logo_utils import print_logo
from .similarity_utils import cosine_similarity, batch_cosine_similarity
from .singleton import singleton
@ -14,6 +15,7 @@ __all__ = [
"hash_text",
"execute_stream_task",
"get_logger",
"print_logo",
"cosine_similarity",
"batch_cosine_similarity",
"singleton",

6
reme_cli/utils/case_converter.py
View file

@ -43,8 +43,4 @@ def snake_to_camel(content: str) -> str:
Returns:
The converted PascalCase string with acronyms preserved.
"""
return "".join(
_ACRONYM_MAP.get(part.lower(), part.capitalize())
for part in content.split("_")
if part
)
return "".join(_ACRONYM_MAP.get(part.lower(), part.capitalize()) for part in content.split("_") if part)

10
reme_cli/utils/chunking_utils.py
View file

@ -9,10 +9,10 @@ from ..schema import FileChunk
def chunk_markdown(
text: str,
path: str,
chunk_tokens: int,
overlap: int,
text: str,
path: str,
chunk_tokens: int,
overlap: int,
) -> list[FileChunk]:
"""Split Markdown text into chunks with configurable size and overlap.
@ -125,7 +125,7 @@ def chunk_markdown(
else:
# If line is too long, split by maximum character count
for start in range(0, len(line), max_chars):
segments.append(line[start: start + max_chars])
segments.append(line[start : start + max_chars])
for segment in segments:
line_size = len(segment) + 1 # +1 for newline

11
reme_cli/utils/common_utils.py
View file

@ -47,10 +47,10 @@ def hash_text(text: str, encoding: str = "utf-8") -> str:
async def execute_stream_task(
stream_queue: asyncio.Queue,
task: asyncio.Task,
task_name: str | None = None,
output_format: Literal["str", "bytes", "chunk"] = "str",
stream_queue: asyncio.Queue,
task: asyncio.Task,
task_name: str | None = None,
output_format: Literal["str", "bytes", "chunk"] = "str",
) -> AsyncGenerator[str | bytes | StreamChunk, None]:
"""Core stream flow execution logic.
@ -88,7 +88,8 @@ async def execute_stream_task(
# Wait for next chunk or check if task failed
get_chunk = asyncio.create_task(stream_queue.get())
done, _pending = await asyncio.wait(
{get_chunk, task}, return_when=asyncio.FIRST_COMPLETED
{get_chunk, task},
return_when=asyncio.FIRST_COMPLETED,
)
# Priority 1: Check if main task finished (may have exception)

10
reme_cli/utils/logger_utils.py
View file

@ -14,11 +14,11 @@ _initialized = False
def get_logger(
log_dir: str = "logs",
level: str = "INFO",
log_to_console: bool = True,
log_to_file: bool = True,
force_init: bool = False,
log_dir: str = "logs",
level: str = "INFO",
log_to_console: bool = True,
log_to_file: bool = True,
force_init: bool = False,
):
"""Get a configured logger instance.

91
reme_cli/utils/logo_utils.py Normal file
View file

@ -0,0 +1,91 @@
"""Terminal branding and configuration display utilities."""
import importlib.metadata
from typing import TYPE_CHECKING
from rich.console import Console, Group
from rich.panel import Panel
from rich.table import Table
from rich.text import Text
if TYPE_CHECKING:
from ..schema import ApplicationConfig
def get_version(package_name: str) -> str:
"""Return the installed version of a package or 'unknown'."""
try:
return importlib.metadata.version(package_name)
except importlib.metadata.PackageNotFoundError:
return ""
def print_logo(app_config: "ApplicationConfig"):
"""Print a stylized ASCII logo and service metadata to the console."""
ascii_art = [
r" ██████╗ ███████╗ ███╗ ███╗ ███████╗ ",
r" ██╔══██╗ ██╔════╝ ████╗ ████║ ██╔════╝ ",
r" ██████╔╝ █████╗ ██╔████╔██║ █████╗ ",
r" ██╔══██╗ ██╔══╝ ██║╚██╔╝██║ ██╔══╝ ",
r" ██║ ██║ ███████╗ ██║ ╚═╝ ██║ ███████╗ ",
r" ╚═╝ ╚═╝ ╚══════╝ ╚═╝ ╚═╝ ╚══════╝ ",
]
start_color = (85, 239, 196)
end_color = (162, 155, 254)
logo_text = Text()
for line in ascii_art:
line_len = max(1, len(line) - 1)
for i, char in enumerate(line):
# Calculate gradient shift per character
ratio = i / line_len
rgb = tuple(int(s + (e - s) * ratio) for s, e in zip(start_color, end_color))
logo_text.append(char, style=f"bold rgb({rgb[0]},{rgb[1]},{rgb[2]})")
logo_text.append("\n")
# Layout configuration info
info_table = Table.grid(padding=(0, 1))
info_table.add_column(style="bold", justify="center")
info_table.add_column(style="bold cyan", justify="left")
info_table.add_column(style="white", justify="left")
# Get service config (ComponentConfig with extra="allow")
service = app_config.service
backend = service.backend
# Add core service info
info_table.add_row("📦", "Backend:", backend)
match backend:
case "http":
host = service.model_extra.get("host", "localhost") if service.model_extra else "localhost"
port = service.model_extra.get("port", 8000) if service.model_extra else 8000
info_table.add_row("🔗", "URL:", f"http://{host}:{port}")
info_table.add_row("📚", "FastAPI:", Text(get_version("fastapi"), style="dim"))
case "mcp":
transport = service.model_extra.get("transport", "stdio") if service.model_extra else "stdio"
info_table.add_row("🚌", "Transport:", transport)
if transport != "stdio":
host = service.model_extra.get("host", "localhost") if service.model_extra else "localhost"
port = service.model_extra.get("port", 8000) if service.model_extra else 8000
url = f"http://{host}:{port}"
if transport == "sse":
url += "/sse"
info_table.add_row("🔗", "URL:", url)
info_table.add_row("📚", "FastMCP:", Text(get_version("fastmcp"), style="dim"))
info_table.add_row("🚀", "ReMe:", Text(get_version("reme-ai"), style="dim"))
# Render layout within a panel
panel = Panel(
Group(logo_text, info_table),
title=app_config.app_name,
title_align="left",
border_style="dim",
padding=(1, 4),
expand=False,
)
# use justify="center" to adjust position
Console().print(Group("\n", panel, "\n"))

2
reme_cli/utils/similarity_utils.py
View file

@ -74,7 +74,7 @@ def batch_cosine_similarity(nd_array1: np.ndarray, nd_array2: np.ndarray) -> np.
"""
if nd_array1.shape[1] != nd_array2.shape[1]:
raise ValueError(
f"Embedding dimensions must match: {nd_array1.shape[1]} != {nd_array2.shape[1]}"
f"Embedding dimensions must match: {nd_array1.shape[1]} != {nd_array2.shape[1]}",
)
# Compute dot products: (batch_size1, emb_size) @ (emb_size, batch_size2)