diff --git a/reme_cli/__init__.py b/reme_cli/__init__.py index 5d602c30..f721a453 100644 --- a/reme_cli/__init__.py +++ b/reme_cli/__init__.py @@ -3,4 +3,7 @@ from reme_cli.application import Application from reme_cli.component import BaseComponent -__all__ = ["BaseComponent", "Application"] +__all__ = [ + "BaseComponent", + "Application", +] diff --git a/reme_cli/component/__init__.py b/reme_cli/component/__init__.py index eda848b1..19dbfdc9 100644 --- a/reme_cli/component/__init__.py +++ b/reme_cli/component/__init__.py @@ -1,19 +1,35 @@ +"""Components""" from .application_context import ApplicationContext from .base_component import BaseComponent -from .client import BaseClient, HttpClient, ReMeClient, get_client +from .base_step import BaseStep from .component_registry import ComponentRegistry, R from .prompt_handler import PromptHandler from .runtime_context import RuntimeContext +from . import as_llm +from . import as_llm_formatter +from . import client +from . import embedding +from . import file_store +from . import file_watcher +from . import job +from . import service + __all__ = [ "ApplicationContext", "BaseComponent", - "BaseClient", + "BaseStep", "ComponentRegistry", - "HttpClient", "R", - "ReMeClient", "PromptHandler", "RuntimeContext", - "get_client", + # base components + "as_llm", + "as_llm_formatter", + "client", + "embedding", + "file_store", + "file_watcher", + "job", + "service", ] diff --git a/reme_cli/component/as_llm_formatter/reme_openai_chat_formatter.py b/reme_cli/component/as_llm_formatter/reme_openai_chat_formatter.py index 635cd52f..9a2d34ef 100644 --- a/reme_cli/component/as_llm_formatter/reme_openai_chat_formatter.py +++ b/reme_cli/component/as_llm_formatter/reme_openai_chat_formatter.py @@ -10,10 +10,6 @@ from agentscope.formatter._openai_formatter import ( ) from agentscope.message import Msg, TextBlock, ImageBlock, URLSource -from ...utils import get_logger - -logger = get_logger() - def _format_openai_video_block(video_block: dict) -> dict[str, Any]: """Format a video block for OpenAI API. @@ -135,7 +131,8 @@ class ReMeOpenAIChatFormatter(OpenAIChatFormatter): content_blocks.append(_format_openai_video_block(block)) else: - logger.warning("Unsupported block type %s, skipped.", typ) + ... + # logger.warning("Unsupported block type %s, skipped.", typ) msg_openai = { "role": msg.role, diff --git a/reme_cli/component/base_component.py b/reme_cli/component/base_component.py index 863ea7e2..5d2d4ebd 100644 --- a/reme_cli/component/base_component.py +++ b/reme_cli/component/base_component.py @@ -163,6 +163,9 @@ class BaseComponent(ABC): """ return self._is_started + async def __call__(self, **kwargs): + """Call the component instance as a function.""" + async def __aenter__(self) -> "BaseComponent": """Enter the async context manager by starting the component. diff --git a/reme_cli/component/client/__init__.py b/reme_cli/component/client/__init__.py index 33eb075d..d92cda29 100644 --- a/reme_cli/component/client/__init__.py +++ b/reme_cli/component/client/__init__.py @@ -1,15 +1,9 @@ -"""Client implementations for ReMe services.""" +"""Client""" from .base_client import BaseClient from .http_client import HttpClient -from .reme_client import ReMeClient, get_client, REME_PORT_ENV, DEFAULT_PORT, DEFAULT_HOST __all__ = [ "BaseClient", "HttpClient", - "ReMeClient", - "get_client", - "REME_PORT_ENV", - "DEFAULT_PORT", - "DEFAULT_HOST", ] \ No newline at end of file diff --git a/reme_cli/component/client/base_client.py b/reme_cli/component/client/base_client.py index f6be1ed0..6f16002e 100644 --- a/reme_cli/component/client/base_client.py +++ b/reme_cli/component/client/base_client.py @@ -1,40 +1,25 @@ """Abstract base class for client implementations.""" -from abc import ABC, abstractmethod -from typing import Any +from abc import abstractmethod -from ...schema import Response +from ..base_component import BaseComponent +from ...enumeration import ComponentEnum -class BaseClient(ABC): - """Abstract base class for clients that communicate with ReMe services. +class BaseClient(BaseComponent): + """Abstract base class for clients that communicate with ReMe services.""" + component_type = ComponentEnum.CLIENT - Clients provide a unified interface for invoking jobs/actions on a - ReMe application, whether locally or remotely via HTTP, MCP, etc. - """ + def __init__(self, **kwargs) -> None: + super().__init__(**kwargs) + self.client = None + + async def _start(self, app_context=None) -> None: + """Initialize the client.""" + + async def _close(self) -> None: + """Close the client.""" @abstractmethod - async def invoke(self, action: str, **config: Any) -> Response: - """Invoke an action (job) with the given configuration. - - Args: - action: The name of the action/job endpoint to invoke. - **config: Configuration parameters passed as POST body. - - Returns: - Response from the invoked action. - """ - pass - - @abstractmethod - async def close(self) -> None: - """Close the client and release resources.""" - pass - - async def __aenter__(self) -> "BaseClient": - """Async context manager entry.""" - return self - - async def __aexit__(self, exc_type, exc_val, exc_tb) -> None: - """Async context manager exit.""" - await self.close() \ No newline at end of file + async def __call__(self, action: str, **kwargs) -> dict: + """Invoke an action with the given configuration.""" diff --git a/reme_cli/component/client/http_client.py b/reme_cli/component/client/http_client.py index 2dfe17cb..e68fe75e 100644 --- a/reme_cli/component/client/http_client.py +++ b/reme_cli/component/client/http_client.py @@ -1,85 +1,61 @@ -"""HTTP client implementation using httpx.""" +"""HTTP client for ReMe services.""" +import json +import os from typing import Any import httpx from .base_client import BaseClient -from ...schema import Response +from ..component_registry import R +from ...constants import REME_SERVICE_INFO, REME_DEFAULT_HOST, REME_DEFAULT_PORT +@R.register("http") class HttpClient(BaseClient): - """HTTP client for communicating with ReMe HTTP service. + """HTTP client for ReMe service.""" - Provides a simple interface to invoke jobs/actions via HTTP POST requests. - """ + def __init__( + self, + action: str, + host: str | None = None, + port: int | None = None, + timeout: float = 30.0, + **kwargs + ): + super().__init__(**kwargs) - def __init__(self, base_url: str = "http://127.0.0.1:8000", timeout: float = 30.0, **kwargs): - """Initialize the HTTP client. + if host and port: + pass + elif service_info := os.environ.get(REME_SERVICE_INFO): + try: + data = json.loads(service_info) + host = data.get("host", host) + port = data.get("port", port) + except Exception: + pass + else: + host = REME_DEFAULT_HOST + port = REME_DEFAULT_PORT - Args: - base_url: Base URL of the ReMe HTTP service. - timeout: Request timeout in seconds. - **kwargs: Additional arguments passed to httpx.AsyncClient. - """ - self.base_url = base_url.rstrip("/") + self.action = action + self.base_url = f"http://{host}:{port}" self.timeout = timeout - self._client: httpx.AsyncClient | None = None - self._kwargs = kwargs - @property - def client(self) -> httpx.AsyncClient: - """Get or create the underlying httpx client.""" - if self._client is None: - self._client = httpx.AsyncClient( + async def _start(self, app_context=None) -> None: + """Initialize the HTTP client.""" + if self.client is None: + self.client = httpx.AsyncClient( base_url=self.base_url, timeout=self.timeout, - **self._kwargs ) - return self._client - async def invoke(self, action: str, **config: Any) -> Response: - """Invoke an action via HTTP POST request. - - Args: - action: The action/job endpoint name (appended to base_url). - **config: Configuration parameters sent as JSON body. - - Returns: - Response from the service. - """ - response = await self.client.post(f"/{action}", json=config) - response.raise_for_status() - return Response.model_validate(response.json()) - - async def invoke_raw(self, action: str, **config: Any) -> dict: - """Invoke an action and return raw dict response. - - Args: - action: The action/job endpoint name. - **config: Configuration parameters sent as JSON body. - - Returns: - Raw response dict. - """ - response = await self.client.post(f"/{action}", json=config) + async def __call__(self, **_kwargs) -> dict: + response = await self.client.post(f"/{self.action}", json=self.kwargs) response.raise_for_status() return response.json() - async def health_check(self) -> bool: - """Check if the service is healthy. - - Returns: - True if healthy, False otherwise. - """ - try: - response = await self.client.get("/health") - return response.status_code == 200 - except Exception: - return False - - async def close(self) -> None: - """Close the underlying HTTP client.""" - if self._client is not None: - await self._client.aclose() - self._client = None \ No newline at end of file + async def _close(self) -> None: + if self.client is not None: + await self.client.aclose() + self.client = None diff --git a/reme_cli/component/client/reme_client.py b/reme_cli/component/client/reme_client.py deleted file mode 100644 index cad040cb..00000000 --- a/reme_cli/component/client/reme_client.py +++ /dev/null @@ -1,102 +0,0 @@ -"""ReMe client for easy interaction with ReMe HTTP service.""" - -import os -from typing import Any - -from .http_client import HttpClient -from .base_client import BaseClient -from ...schema import Response - -# Import the environment variable name from http_service -REME_PORT_ENV = "REME_PORT" -DEFAULT_PORT = 8000 -DEFAULT_HOST = "127.0.0.1" - - -class ReMeClient(BaseClient): - """High-level client for ReMe HTTP service. - - Automatically reads port from environment variable REME_PORT if set. - Provides a simple interface to call actions with config parameters. - """ - - def __init__( - self, - host: str = DEFAULT_HOST, - port: int | None = None, - timeout: float = 30.0, - **kwargs - ): - """Initialize the ReMe client. - - Args: - host: Host address of the ReMe service. - port: Port number. If None, reads from REME_PORT env var, - or falls back to 8000. - timeout: Request timeout in seconds. - **kwargs: Additional arguments passed to HttpClient. - """ - if port is None: - port_str = os.environ.get(REME_PORT_ENV) - port = int(port_str) if port_str else DEFAULT_PORT - - self._http_client = HttpClient( - base_url=f"http://{host}:{port}", - timeout=timeout, - **kwargs - ) - - @property - def http_client(self) -> HttpClient: - """Get the underlying HTTP client.""" - return self._http_client - - async def invoke(self, action: str, **config: Any) -> Response: - """Invoke an action with the given configuration. - - Args: - action: The action/job endpoint name. - **config: Configuration parameters passed as POST body. - - Returns: - Response from the invoked action. - """ - return await self._http_client.invoke(action, **config) - - async def call(self, action: str, **config: Any) -> Response: - """Alias for invoke method for more natural usage. - - Args: - action: The action/job endpoint name. - **config: Configuration parameters passed as POST body. - - Returns: - Response from the invoked action. - """ - return await self.invoke(action, **config) - - async def health_check(self) -> bool: - """Check if the ReMe service is healthy. - - Returns: - True if healthy, False otherwise. - """ - return await self._http_client.health_check() - - async def close(self) -> None: - """Close the underlying HTTP client.""" - await self._http_client.close() - - -def get_client(host: str = DEFAULT_HOST, port: int | None = None, **kwargs) -> ReMeClient: - """Factory function to create a ReMeClient. - - Args: - host: Host address of the ReMe service. - port: Port number. If None, reads from REME_PORT env var. - **kwargs: Additional arguments passed to ReMeClient. - - Returns: - A configured ReMeClient instance. - """ - return ReMeClient(host=host, port=port, **kwargs) \ No newline at end of file diff --git a/reme_cli/component/component_registry.py b/reme_cli/component/component_registry.py index 1383706b..00212193 100644 --- a/reme_cli/component/component_registry.py +++ b/reme_cli/component/component_registry.py @@ -13,7 +13,6 @@ from .base_component import BaseComponent from ..enumeration import ComponentEnum from ..utils import get_logger -logger = get_logger() T = TypeVar("T", bound=BaseComponent) @@ -22,6 +21,7 @@ class ComponentRegistry: def __init__(self) -> None: self._registry: dict[ComponentEnum, dict[str, type[BaseComponent]]] = {} + self.logger = get_logger() def _do_register(self, cls: type[T], name: str) -> type[T]: if not hasattr(cls, 'component_type'): @@ -31,7 +31,7 @@ class ComponentRegistry: component_type = cls.component_type if name in self._registry[component_type]: - logger.warning(f"Component '{name}' already registered for {component_type}, overwriting") + self.logger.warning(f"Component '{name}' already registered for {component_type}, overwriting") self._registry[component_type][name] = cls return cls diff --git a/reme_cli/component/embedding/base_embedding_model.py b/reme_cli/component/embedding/base_embedding_model.py index 1eec581b..a6bc4402 100644 --- a/reme_cli/component/embedding/base_embedding_model.py +++ b/reme_cli/component/embedding/base_embedding_model.py @@ -9,6 +9,7 @@ from collections import OrderedDict from pathlib import Path from ..base_component import BaseComponent +from ...enumeration import ComponentEnum from ...schema import BaseNode @@ -21,6 +22,7 @@ class BaseEmbeddingModel(BaseComponent): - Retry logic with exponential backoff - Batch embedding support """ + component_type = ComponentEnum.EMBEDDING_MODEL def __init__( self, diff --git a/reme_cli/component/file_store/base_file_store.py b/reme_cli/component/file_store/base_file_store.py index b4bbad57..a7b3f617 100644 --- a/reme_cli/component/file_store/base_file_store.py +++ b/reme_cli/component/file_store/base_file_store.py @@ -16,6 +16,7 @@ class BaseFileStore(BaseComponent): Provides embedding resolution, validation, and safe embedding retrieval with automatic fallback on failure. """ + component_type = ComponentEnum.FILE_STORE def __init__( self, diff --git a/reme_cli/component/file_store/local_file_store.py b/reme_cli/component/file_store/local_file_store.py index 480a56bf..2d2fba1b 100644 --- a/reme_cli/component/file_store/local_file_store.py +++ b/reme_cli/component/file_store/local_file_store.py @@ -6,10 +6,12 @@ from pathlib import Path import numpy as np from .base_file_store import BaseFileStore +from ..component_registry import R from ...schema import FileChunk, FileMetadata from ...utils import batch_cosine_similarity +@R.register("local") class LocalFileStore(BaseFileStore): """In-memory file storage with JSONL disk persistence. diff --git a/reme_cli/component/file_watcher/md_file_watcher.py b/reme_cli/component/file_watcher/md_file_watcher.py index 0eb11f6e..43b01221 100644 --- a/reme_cli/component/file_watcher/md_file_watcher.py +++ b/reme_cli/component/file_watcher/md_file_watcher.py @@ -6,10 +6,12 @@ from pathlib import Path from watchfiles import Change from .base_file_watcher import BaseFileWatcher +from ..component_registry import R from ...schema import FileMetadata from ...utils import hash_text, chunk_markdown +@R.register("md") class MdFileWatcher(BaseFileWatcher): """Markdown file watcher that syncs .md files to memory store.""" diff --git a/reme_cli/component/job/base_job.py b/reme_cli/component/job/base_job.py index 08c56e09..af6aff9f 100644 --- a/reme_cli/component/job/base_job.py +++ b/reme_cli/component/job/base_job.py @@ -7,6 +7,7 @@ from ...enumeration import ComponentEnum from ...schema import Response, ComponentConfig +@R.register("base") class BaseJob(BaseComponent): """Base job that executes a sequence of steps. @@ -14,6 +15,7 @@ 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, diff --git a/reme_cli/component/job/stream_job.py b/reme_cli/component/job/stream_job.py index ab4aaa5b..ae362310 100644 --- a/reme_cli/component/job/stream_job.py +++ b/reme_cli/component/job/stream_job.py @@ -3,10 +3,12 @@ import asyncio from .base_job import BaseJob +from ..component_registry import R from ..runtime_context import RuntimeContext from ...enumeration import ChunkEnum +@R.register("stream") class StreamJob(BaseJob): """Job that streams execution results in real-time. diff --git a/reme_cli/component/service/base_service.py b/reme_cli/component/service/base_service.py index caae3a04..ff669547 100644 --- a/reme_cli/component/service/base_service.py +++ b/reme_cli/component/service/base_service.py @@ -4,6 +4,7 @@ from abc import abstractmethod from ..base_component import BaseComponent from ..job.base_job import BaseJob +from ...enumeration import ComponentEnum class BaseService(BaseComponent): @@ -12,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 diff --git a/reme_cli/component/service/http_service.py b/reme_cli/component/service/http_service.py index 0c804675..ebffd595 100644 --- a/reme_cli/component/service/http_service.py +++ b/reme_cli/component/service/http_service.py @@ -1,10 +1,10 @@ """HTTP service implementation using FastAPI and uvicorn.""" import asyncio +import json import os from collections.abc import AsyncGenerator from contextlib import asynccontextmanager -from typing import TYPE_CHECKING import uvicorn from fastapi import FastAPI @@ -12,25 +12,23 @@ from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import StreamingResponse from .base_service import BaseService +from ..component_registry import R from ..job import BaseJob, StreamJob +from ...constants import REME_DEFAULT_HOST, REME_DEFAULT_PORT, REME_SERVICE_INFO from ...schema import Request, Response from ...utils import execute_stream_task -# Environment variable name for ReMe port -REME_PORT_ENV = "REME_PORT" - -if TYPE_CHECKING: - from ...application import Application - +@R.register("http") class HttpService(BaseService): """HTTP service that exposes jobs as REST endpoints. 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 = "0.0.0.0", port: int = 8000, **kwargs): + def __init__(self, host: str = REME_DEFAULT_HOST, port: int = REME_DEFAULT_PORT, **kwargs): """Initialize the HTTP service. Args: @@ -59,7 +57,7 @@ class HttpService(BaseService): )(execute_endpoint) def _add_stream_job(self, job: StreamJob) -> None: - """Register a stream job as a SSE endpoint. + """Register a stream job as an SSE endpoint. Args: job: The stream job to register. @@ -78,6 +76,7 @@ class HttpService(BaseService): task_name=job.name, output_format="bytes", ): + assert isinstance(chunk, bytes) yield chunk return StreamingResponse(generate_stream(), media_type="text/event-stream") @@ -107,6 +106,12 @@ class HttpService(BaseService): @asynccontextmanager async def lifespan(_: FastAPI): await app.start() + 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 await app.close() @@ -119,7 +124,7 @@ class HttpService(BaseService): allow_methods=["*"], allow_headers=["*"], ) - self.service.get("/health")(lambda: {"status": "healthy"}) + self.service.post("/health")(lambda: {"status": "healthy"}) def start_service(self, app: "Application") -> None: """Start the HTTP server. @@ -127,6 +132,4 @@ class HttpService(BaseService): Args: app: The application instance. """ - os.environ[REME_PORT_ENV] = str(self.port) - self.logger.info(f"Setting {REME_PORT_ENV}={self.port}") uvicorn.run(self.service, host=self.host, port=self.port, **self.kwargs) diff --git a/reme_cli/constants.py b/reme_cli/constants.py new file mode 100644 index 00000000..fce66e64 --- /dev/null +++ b/reme_cli/constants.py @@ -0,0 +1,7 @@ +"""Constants""" + +REME_SERVICE_INFO = "REME_SERVICE_INFO" + +REME_DEFAULT_HOST = "127.0.0.1" + +REME_DEFAULT_PORT = 2333 diff --git a/reme_cli/enumeration/component_enum.py b/reme_cli/enumeration/component_enum.py index 49ee5b4a..bce70d7b 100644 --- a/reme_cli/enumeration/component_enum.py +++ b/reme_cli/enumeration/component_enum.py @@ -28,6 +28,8 @@ class ComponentEnum(str, Enum): SERVICE = "service" + CLIENT = "client" + STEP = "step" JOB = "job" diff --git a/reme_cli/reme.py b/reme_cli/reme.py index e624392d..b0647437 100644 --- a/reme_cli/reme.py +++ b/reme_cli/reme.py @@ -1,15 +1,59 @@ +import asyncio import sys +from agentscope.formatter import FormatterBase +from agentscope.message import Msg +from agentscope.model import ChatModelBase +from agentscope.token import TokenCounterBase +from agentscope.tool import Toolkit, ToolResponse + from .application import Application +from .component import R from .config import parse_args +from .enumeration import ComponentEnum class ReMe(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, + ) -> str: + ... -class ReMeClient: - ... + async def memory_search(self, query: str, max_results: int = 5, min_score: float = 0.1) -> ToolResponse: + ... + + 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, + ) -> str: + ... + + 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, + ) -> str: + ... def main(): @@ -19,8 +63,10 @@ def main(): reme.run_app() else: - ... - """Main entry point for running ReMe from command line.""" + backend: str = config.pop("backend", "http") + client_cls = R.get(ComponentEnum.CLIENT, backend) + client = client_cls(action=action, **config) + asyncio.run(client()) if __name__ == "__main__": diff --git a/reme_cli/utils/common_utils.py b/reme_cli/utils/common_utils.py index 5b138553..d2319f2b 100644 --- a/reme_cli/utils/common_utils.py +++ b/reme_cli/utils/common_utils.py @@ -6,13 +6,27 @@ stream processing for task execution. import asyncio import hashlib -from typing import AsyncGenerator, Literal +from collections.abc import AsyncGenerator, Coroutine +from typing import Any, Literal from .logger_utils import get_logger from ..enumeration import ChunkEnum from ..schema import StreamChunk -logger = get_logger() + +def run_coro_safely(coro: Coroutine[Any, Any, Any]) -> Any | asyncio.Task[Any]: + """Run a coroutine in the current event loop or a new one if none exists.""" + try: + # Attempt to retrieve the event loop associated with the current thread + loop = asyncio.get_running_loop() + + except RuntimeError: + # Start a new event loop to run the coroutine to completion + return asyncio.run(coro) + + else: + # Schedule the coroutine as a background task in the active loop + return loop.create_task(coro) def hash_text(text: str, encoding: str = "utf-8") -> str: @@ -68,6 +82,7 @@ async def execute_stream_task( Raises: Exception: Re-raises any exception from the background task. """ + logger = get_logger() try: while True: # Wait for next chunk or check if task failed