From 054d480ee97c0cc7e3f846f475f6d2cebd24cd41 Mon Sep 17 00:00:00 2001 From: "jinli.yl" Date: Sun, 17 May 2026 01:29:53 +0800 Subject: [PATCH] up --- reme4/__init__.py | 4 +- reme4/components/client/http_client.py | 4 +- reme4/components/service/http_service.py | 1 - reme4/config/default.yaml | 20 ++- reme4/reme.py | 8 +- reme4/steps/__init__.py | 5 +- reme4/steps/common/__init__.py | 12 ++ reme4/steps/{ => common}/demo.py | 4 +- reme4/steps/common/health_check.py | 150 +++++++++++++++++++++++ reme4/steps/common/version.py | 18 +++ reme4/utils/__init__.py | 5 + reme4/utils/service_utils.py | 109 ++++++++++++++++ 12 files changed, 326 insertions(+), 14 deletions(-) rename reme4/steps/{ => common}/demo.py (96%) create mode 100644 reme4/steps/common/health_check.py create mode 100644 reme4/steps/common/version.py create mode 100644 reme4/utils/service_utils.py diff --git a/reme4/__init__.py b/reme4/__init__.py index 8dba02ec..e4db3de0 100644 --- a/reme4/__init__.py +++ b/reme4/__init__.py @@ -1,5 +1,7 @@ """ReMe CLI package.""" +__version__ = "0.4.0.0" + from . import config from . import constants from . import enumeration @@ -22,5 +24,3 @@ __all__ = [ "steps", "utils", ] - -__version__ = "0.4.0.0" diff --git a/reme4/components/client/http_client.py b/reme4/components/client/http_client.py index cbcc7ec9..96958c2c 100644 --- a/reme4/components/client/http_client.py +++ b/reme4/components/client/http_client.py @@ -54,9 +54,7 @@ class HttpClient(BaseClient): response = await self.client.post(f"/{self.action}", json=self.kwargs) response.raise_for_status() result = response.json() - if "answer" in result: - return str(result["answer"]) - return str(result) + return json.dumps(result, indent=2, ensure_ascii=False) async def _close(self) -> None: """Close the HTTP client.""" diff --git a/reme4/components/service/http_service.py b/reme4/components/service/http_service.py index 64806114..a1092947 100644 --- a/reme4/components/service/http_service.py +++ b/reme4/components/service/http_service.py @@ -82,7 +82,6 @@ class HttpService(BaseService): allow_methods=["*"], allow_headers=["*"], ) - self.service.post("/health")(lambda: {"status": "healthy"}) def start_service(self, app: "Application") -> None: # uvicorn 0.41 still imports websockets.legacy / WebSocketServerProtocol diff --git a/reme4/config/default.yaml b/reme4/config/default.yaml index 1187f491..b054b620 100644 --- a/reme4/config/default.yaml +++ b/reme4/config/default.yaml @@ -4,7 +4,7 @@ service: jobs: - backend: base - name: demo_job + name: demo description: "demo job description" parameters: type: object @@ -22,6 +22,24 @@ jobs: - backend: demo_echo_step1 - backend: demo_echo_step2 + - backend: base + name: version + description: "return reme4 package version" + parameters: + type: object + properties: {} + steps: + - backend: version_step + + - backend: base + name: health_check + description: "return a concise health-check snapshot of reme4 components" + parameters: + type: object + properties: {} + steps: + - backend: health_check_step + components: # 1. tokenizer — no dependencies tokenizer: diff --git a/reme4/reme.py b/reme4/reme.py index a1dbf1df..98562d18 100644 --- a/reme4/reme.py +++ b/reme4/reme.py @@ -7,7 +7,7 @@ from .application import Application from .components import R from .config import parse_args, resolve_app_config from .enumeration import ComponentEnum -from .utils import load_env +from .utils import cli_find_reme, load_env, precheck_start class ReMe(Application): @@ -19,7 +19,7 @@ async def call_server(action: str, **kwargs): backend: str = kwargs.pop("backend", "http") client_cls = R.get(ComponentEnum.CLIENT, backend) async with client_cls(action=action, **kwargs) as client: - await client() + print(await client()) def main(): @@ -28,7 +28,11 @@ def main(): if action == "start": load_env() kwargs = resolve_app_config(**kwargs) + if not precheck_start(kwargs.get("service")): + return ReMe(**kwargs).run_app() + elif action == "find_reme": + cli_find_reme() else: asyncio.run(call_server(action, **kwargs)) diff --git a/reme4/steps/__init__.py b/reme4/steps/__init__.py index c8db6ea6..fa25273c 100644 --- a/reme4/steps/__init__.py +++ b/reme4/steps/__init__.py @@ -1,10 +1,9 @@ """steps""" +from . import common from .base_step import BaseStep -from .demo import DemoEchoStep1, DemoEchoStep2 __all__ = [ + "common", "BaseStep", - "DemoEchoStep1", - "DemoEchoStep2", ] diff --git a/reme4/steps/common/__init__.py b/reme4/steps/common/__init__.py index e69de29b..dc57b38c 100644 --- a/reme4/steps/common/__init__.py +++ b/reme4/steps/common/__init__.py @@ -0,0 +1,12 @@ +"""Common steps.""" + +from .demo import DemoEchoStep1, DemoEchoStep2 +from .health_check import HealthCheckStep +from .version import VersionStep + +__all__ = [ + "DemoEchoStep1", + "DemoEchoStep2", + "HealthCheckStep", + "VersionStep", +] diff --git a/reme4/steps/demo.py b/reme4/steps/common/demo.py similarity index 96% rename from reme4/steps/demo.py rename to reme4/steps/common/demo.py index 9ad6e881..e7c8df1a 100644 --- a/reme4/steps/demo.py +++ b/reme4/steps/common/demo.py @@ -1,7 +1,7 @@ """Demo steps for smoke-testing the application stack.""" -from .base_step import BaseStep -from ..components import R +from ..base_step import BaseStep +from ...components import R @R.register("demo_echo_step1") diff --git a/reme4/steps/common/health_check.py b/reme4/steps/common/health_check.py new file mode 100644 index 00000000..43a40573 --- /dev/null +++ b/reme4/steps/common/health_check.py @@ -0,0 +1,150 @@ +"""Return a concise health check snapshot of ReMe runtime components.""" + +import sys +from collections.abc import Mapping + +import numpy as np + +from ..base_step import BaseStep +from ... import __version__ +from ...components import R +from ...enumeration import ComponentEnum + + +def _deep_size(obj, _seen: set | None = None) -> int: + """Recursive sizeof; uses ndarray.nbytes for numpy and walks containers / __dict__.""" + if _seen is None: + _seen = set() + obj_id = id(obj) + if obj_id in _seen: + return 0 + _seen.add(obj_id) + + if isinstance(obj, np.ndarray): + return int(obj.nbytes) + sys.getsizeof(obj) + + size = sys.getsizeof(obj) + if isinstance(obj, (str, bytes, bytearray, int, float, bool, type(None))): + return size + if isinstance(obj, Mapping): + size += sum(_deep_size(k, _seen) + _deep_size(v, _seen) for k, v in obj.items()) + elif isinstance(obj, (list, tuple, set, frozenset)): + size += sum(_deep_size(item, _seen) for item in obj) + elif hasattr(obj, "__dict__"): + size += _deep_size(vars(obj), _seen) + elif hasattr(obj, "__slots__"): + for slot in obj.__slots__: + if hasattr(obj, slot): + size += _deep_size(getattr(obj, slot), _seen) + return size + + +def _mb_str(*objs) -> str: + """Return summed deep size of objs formatted as 'X.XX MB'.""" + seen: set = set() + total = sum(_deep_size(o, seen) for o in objs) + return f"{total / (1024 * 1024):.2f} MB" + + +def _embedding_status(comp) -> dict: + return { + "is_started": comp.is_started, + "is_healthy": getattr(comp, "is_healthy", None), + "model_name": getattr(comp, "model_name", None), + "dimensions": getattr(comp, "dimensions", None), + "cache_size": len(getattr(comp, "_embedding_cache", {}) or {}), + "memory": _mb_str(getattr(comp, "_embedding_cache", {}) or {}), + } + + +def _file_graph_status(comp) -> dict: + nodes = getattr(comp, "_nodes", {}) or {} + inverse = getattr(comp, "_inverse", {}) or {} + pending = getattr(comp, "_pending", {}) or {} + return { + "is_started": comp.is_started, + "n_nodes": len(nodes), + "n_edges": sum(len(s) for s in inverse.values()), + "memory": _mb_str(nodes, inverse, pending), + } + + +def _file_store_status(comp) -> dict: + chunks = getattr(comp, "file_chunks", {}) or {} + return { + "is_started": comp.is_started, + "n_chunks": len(chunks), + "n_chunks_with_embedding": sum(1 for c in chunks.values() if getattr(c, "embedding", None) is not None), + "memory": _mb_str(chunks), + } + + +def _file_watcher_status(comp) -> dict: + bg = getattr(comp, "_background_task", None) + return { + "is_started": comp.is_started, + "background_running": bool(bg and not bg.done()), + "watch_paths": [str(p) for p in (getattr(comp, "watch_paths", []) or [])], + } + + +def _keyword_index_status(comp) -> dict: + return { + "is_started": comp.is_started, + "n_docs": getattr(comp, "n_docs", None), + "vocab_size": len(getattr(comp, "vocab", {}) or {}), + "memory": _mb_str( + getattr(comp, "vocab", {}) or {}, + getattr(comp, "inverted_index", {}) or {}, + getattr(comp, "doc_meta", {}) or {}, + getattr(comp, "_idf_cache", {}) or {}, + ), + } + + +_HANDLERS = { + ComponentEnum.EMBEDDING_MODEL: _embedding_status, + ComponentEnum.FILE_GRAPH: _file_graph_status, + ComponentEnum.FILE_STORE: _file_store_status, + ComponentEnum.FILE_WATCHER: _file_watcher_status, + ComponentEnum.KEYWORD_INDEX: _keyword_index_status, +} + + +def _is_status_healthy(ctype: ComponentEnum, status: dict) -> bool: + """Per-component health rule. Unstarted = unhealthy; type-specific extras checked.""" + if not status.get("is_started"): + return False + if ctype is ComponentEnum.EMBEDDING_MODEL and status.get("is_healthy") is False: + return False + if ctype is ComponentEnum.FILE_WATCHER and not status.get("background_running"): + return False + return True + + +@R.register("health_check_step") +class HealthCheckStep(BaseStep): + """Collect a concise health-check snapshot of the relevant components.""" + + async def execute(self): + assert self.context is not None + + components: dict = {} + healthy = True + if self.app_context is not None: + for ctype, handler in _HANDLERS.items(): + comp_map = self.app_context.components.get(ctype, {}) + bucket = {} + for name, comp in comp_map.items(): + s = handler(comp) + bucket[name] = s + if not _is_status_healthy(ctype, s): + healthy = False + components[ctype.value] = bucket + + health = {"version": __version__, "healthy": healthy, "components": components} + self.logger.info(f"[{self.name}] health collected: {health}") + + self.context.response.answer = f"ReMe v{__version__} - {'healthy' if healthy else 'unhealthy'}" + self.context.response.metadata["health"] = health + return self.context.response diff --git a/reme4/steps/common/version.py b/reme4/steps/common/version.py new file mode 100644 index 00000000..477802e6 --- /dev/null +++ b/reme4/steps/common/version.py @@ -0,0 +1,18 @@ +"""Return the package version.""" + +from ..base_step import BaseStep +from ... import __version__ +from ...components import R + + +@R.register("version_step") +class VersionStep(BaseStep): + """Emit reme4.__version__ as the response answer.""" + + async def execute(self): + assert self.context is not None + self.logger.info(f"[{self.name}] version={__version__}") + + self.context.response.answer = __version__ + self.context.response.metadata["version"] = __version__ + return self.context.response diff --git a/reme4/utils/__init__.py b/reme4/utils/__init__.py index c9a0e460..c9e2d275 100644 --- a/reme4/utils/__init__.py +++ b/reme4/utils/__init__.py @@ -4,6 +4,7 @@ from .common_utils import hash_text, execute_stream_task from .env_utils import load_env from .logger_utils import get_logger from .logo_utils import print_logo +from .service_utils import find_reme, locate_reme, precheck_start, cli_find_reme from .similarity_utils import cosine_similarity, batch_cosine_similarity __all__ = [ @@ -12,6 +13,10 @@ __all__ = [ "load_env", "get_logger", "print_logo", + "find_reme", + "locate_reme", + "precheck_start", + "cli_find_reme", "cosine_similarity", "batch_cosine_similarity", ] diff --git a/reme4/utils/service_utils.py b/reme4/utils/service_utils.py new file mode 100644 index 00000000..f7468893 --- /dev/null +++ b/reme4/utils/service_utils.py @@ -0,0 +1,109 @@ +"""Service discovery utilities.""" + +import asyncio +import socket +import subprocess +import sys + +from ..components.client.http_client import HttpClient +from ..constants import REME_DEFAULT_HOST, REME_DEFAULT_PORT + + +async def find_reme(host: str, port: int) -> str: + """Probe host:port. Returns 'reme', 'occupied', or 'free'.""" + try: + async with HttpClient(action="health_check", host=host, port=port, timeout=2.0) as c: + await c() + return "reme" + except Exception: + pass + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + try: + s.bind((host, port)) + return "free" + except OSError: + return "occupied" + + +def _run(cmd: list[str]) -> str: + """Run cmd; return stdout, or empty on failure.""" + try: + return subprocess.check_output(cmd, stderr=subprocess.DEVNULL, text=True) + except (subprocess.CalledProcessError, FileNotFoundError): + return "" + + +def _pid_on_port(port: int) -> int | None: + """Return PID listening on TCP port, or None.""" + out = _run(["lsof", "-nP", f"-iTCP:{port}", "-sTCP:LISTEN", "-t"]).strip() + return int(out.splitlines()[0]) if out else None + + +def _scan_reme_procs() -> list[tuple[int, str, int]]: + """Find 'reme ... start' processes. Returns [(pid, host, port), ...].""" + procs = [] + for line in _run(["pgrep", "-af", "reme.* start"]).splitlines(): + parts = line.split() + if not parts: + continue + try: + pid = int(parts[0]) + except ValueError: + continue + host, port = REME_DEFAULT_HOST, REME_DEFAULT_PORT + for t in parts[1:]: + if t.startswith("service.host="): + host = t.split("=", 1)[1] + elif t.startswith("service.port="): + try: + port = int(t.split("=", 1)[1]) + except ValueError: + pass + procs.append((pid, host, port)) + return procs + + +async def locate_reme() -> tuple[str, int, int | None] | None: + """Locate a running reme. Returns (host, port, pid) or None.""" + # 1. Try default host:port + if await find_reme(REME_DEFAULT_HOST, REME_DEFAULT_PORT) == "reme": + return REME_DEFAULT_HOST, REME_DEFAULT_PORT, _pid_on_port(REME_DEFAULT_PORT) + # 2. Scan reme processes and probe each + for pid, host, port in _scan_reme_procs(): + if await find_reme(host, port) == "reme": + return host, port, pid + return None + + +def precheck_start(svc_config: dict | None) -> bool: + """Pre-flight check before `start`. Returns True if caller should proceed. + + Prints a message and returns False if reme is already running. + Exits with code 1 if the port is occupied by a non-reme process. + """ + host = (svc_config or {}).get("host") or REME_DEFAULT_HOST + port = (svc_config or {}).get("port") or REME_DEFAULT_PORT + status = asyncio.run(find_reme(host, port)) + if status == "reme": + print(f"reme already running at {host}:{port}") + return False + if status == "occupied": + print( + f"port {port} is occupied by another process. " + f"Start with a different port: reme4 start service.port=", + file=sys.stderr, + ) + sys.exit(1) + return True + + +def cli_find_reme() -> None: + """Handle the `find_reme` CLI action: print HOST/PORT/PID or a hint.""" + found = asyncio.run(locate_reme()) + if found: + host, port, pid = found + print(f"HOST={host} PORT={port} PID={pid if pid is not None else 'unknown'}") + else: + print("reme not started. Try: reme start", file=sys.stderr) + sys.exit(1)