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

up

Browse source
This commit is contained in:
jinli.yl 2026-05-27 14:47:09 +08:00
parent 11b32e64de
commit 4f620324e5
4 changed files with 75 additions and 235 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
reme4/steps/__init__.py
View file

@ -38,14 +38,12 @@ from . import index # noqa: F401 -- registers update_catalog_step / update_ind
# from . import jobs # noqa: F401 -- registers synchronizer / digester
from .base_step import BaseStep
from . import graph # noqa: F401
__all__ = [
"background",
"common",
"crud",
"transfer",
"graph",
"index",
"frontmatter",
"daily",

7
reme4/steps/graph/__init__.py
View file

@ -1,7 +0,0 @@
"""Graph steps."""
from .traverse import GraphTraverseStep
__all__ = [
"GraphTraverseStep",
]

119
reme4/steps/graph/traverse.py
View file

@ -1,119 +0,0 @@
"""``graph_traverse_step`` — BFS over wikilink edges from a seed file.
Single tool for relationship browsing. ``depth=1`` covers the trivial
"what does this link to / what links here" lookups (set ``direction``
accordingly); higher depth opens up multi-hop exploration.
Output is one record per edge traversed (not per node), so the same
target can appear multiple times if reached via different predicates
or paths — agents dedupe at the call site if they want a flat node
set. Each record carries ``via`` (the predecessor) and the link's
``predicate`` / ``anchor`` so the agent can reconstruct the path.
Adjacency is loaded once via ``file_graph.get_nodes(None)`` — every
real node arrives with its full ``links`` payload, and we build both
the outbound and the inbound index in a single pass. The BFS then
runs purely in memory: no per-frontier-node graph round-trips, no
filesystem walk. The ``get_inlinks`` / ``get_outlinks`` contract
methods stay unused here because they'd add network round-trips for
data we already have.
Direction vocabulary accepts both the standard convention
(``forward`` / ``backward`` / ``both``) and the engine convention
(``out`` / ``in`` / ``both``).
The seed ``path`` is taken as-is (vault-relative). A seed that doesn't
match any graph node yields an empty result (no error).
"""
from collections import deque
from ..base_step import BaseStep
from ...components import R
from ...schema import FileLink
_FORWARD = {"out", "forward"}
_BACKWARD = {"in", "backward"}
_BOTH = {"both"}
_VALID_DIRECTIONS = _FORWARD | _BACKWARD | _BOTH
@R.register("graph_traverse_step")
class GraphTraverseStep(BaseStep):
"""BFS from a seed file to explore wikilink relationships.
Parameters:
path — seed path (vault-relative).
direction — ``forward`` / ``backward`` / ``both`` (or ``out`` / ``in`` / ``both``).
depth — hop limit (default 1 = immediate neighbors).
predicate — optional edge-type filter; ``None`` = no filter.
"""
async def execute(self):
"""BFS from ``path`` and emit one record per traversed edge."""
assert self.context is not None
seed = str(self.context.get("path") or "").strip()
assert seed, "path is required"
max_depth = int(self.context.get("depth") or 1)
direction = (self.context.get("direction") or "both").lower()
predicate = self.context.get("predicate")
assert (
direction in _VALID_DIRECTIONS
), f"direction must be one of {sorted(_VALID_DIRECTIONS)}, got {direction!r}"
# Build outbound / inbound adjacency in one pass over all nodes.
outbound: dict[str, list[tuple[str, FileLink]]] = {}
inbound: dict[str, list[tuple[str, FileLink]]] = {}
if self.file_store.file_graph:
for node in await self.file_store.file_graph.get_nodes():
for link in node.links:
if not link.target_path:
continue
outbound.setdefault(node.path, []).append((link.target_path, link))
inbound.setdefault(link.target_path, []).append((node.path, link))
walk_out = direction in _FORWARD or direction in _BOTH
walk_in = direction in _BACKWARD or direction in _BOTH
visited_edges: set[tuple[str, str, str | None]] = set()
results: list[dict] = []
queue: deque[tuple[str, int]] = deque([(seed, 0)])
while queue:
current, depth = queue.popleft()
if depth >= max_depth:
continue
edges: list[tuple[str, str | None, str | None]] = []
if walk_out:
for tgt, link in outbound.get(current, ()):
if predicate is not None and link.predicate != predicate:
continue
edges.append((tgt, link.predicate, link.target_anchor))
if walk_in:
for src, link in inbound.get(current, ()):
if predicate is not None and link.predicate != predicate:
continue
edges.append((src, link.predicate, link.target_anchor))
for next_path, pred, anchor in edges:
edge_key = (current, next_path, pred)
if edge_key in visited_edges:
continue
visited_edges.add(edge_key)
results.append(
{
"path": next_path,
"depth": depth + 1,
"via": current,
"predicate": pred,
"anchor": anchor,
},
)
if depth + 1 < max_depth:
queue.append((next_path, depth + 1))
self.context.response.success = True
self.context.response.answer = f"Traversed {len(results)} edge(s) from {seed}"
self.context.response.metadata.update({"edges": results, "count": len(results)})

182
tests4/unittest/test_common_steps.py
View file

@ -2,10 +2,10 @@
Two surfaces share this file:
* **HTTP / MCP E2E tests** (top half) spawn ``reme4 start`` via
``mock_reme_server`` and drive ``version`` / ``help`` / ``search`` /
``init`` / ``demo`` over the wire. Each test uses an isolated cwd so
the vault (``.reme`` by default) does not collide.
* **In-process job tests** (top half) build an ``Application`` from the
default config and call ``run_job`` directly — no subprocess, no HTTP.
Each test uses an isolated cwd so the vault (``.reme`` by default)
does not collide.
* **Direct unit tests** (bottom half) exercise ``TraverseStep``
(registered as ``traverse_step``) — BFS over wikilink edges from a
seed file, forward / backward / both — against a freshly built
@ -19,11 +19,12 @@ import os
import tempfile
import warnings
from reme4 import __version__ as REME_VERSION
from reme4 import Application, __version__ as REME_VERSION
from reme4.components.file_store import LocalFileStore
from reme4.config import resolve_app_config
from reme4.schema import FileLink, FileNode
from reme4.steps.common import traverse as traverse_mod
from reme4.utils import call_action, call_and_check, mock_reme_server
from reme4.utils import load_env
warnings.filterwarnings("ignore", category=DeprecationWarning, module="jieba")
warnings.filterwarnings("ignore", category=DeprecationWarning, module="pkg_resources")
@ -50,6 +51,15 @@ def _run(coro):
asyncio.run(coro)
async def _make_app() -> Application:
"""Build and start an Application with the default config, logging silenced."""
load_env()
cfg = resolve_app_config(log_to_console=False, log_to_file=False, enable_logo=False)
app = Application(**cfg)
await app.start()
return app
def _node(path: str, links: list[tuple[str, str | None, str | None]] | None = None) -> FileNode:
"""Build a FileNode with (target_path, target_anchor, predicate) outgoing edges."""
return FileNode(
@ -73,7 +83,7 @@ def _edges(step) -> list[dict]:
# ===========================================================================
# HTTP / MCP E2E tests: version / help / search / init / demo
# In-process job tests: version / help / health_check / search / reindex
# ===========================================================================
@ -82,18 +92,14 @@ def test_version_job():
async def run():
with tempfile.TemporaryDirectory() as tmp, _temp_chdir(tmp):
async with mock_reme_server() as (host, port):
await call_and_check(
"version",
host=host,
port=port,
validator=lambda r: (
isinstance(r, dict)
and r.get("success") is True
and r.get("answer") == REME_VERSION
and r.get("metadata", {}).get("version") == REME_VERSION
),
)
app = await _make_app()
try:
resp = await app.run_job("version")
assert resp.success is True
assert resp.answer == REME_VERSION
assert resp.metadata.get("version") == REME_VERSION
finally:
await app.close()
print("✓ test_version_job passed")
_run(run())
@ -104,24 +110,17 @@ def test_help_job():
async def run():
with tempfile.TemporaryDirectory() as tmp, _temp_chdir(tmp):
async with mock_reme_server() as (host, port):
result = await call_and_check(
"help",
host=host,
port=port,
validator=lambda r: (
isinstance(r, dict)
and r.get("success") is True
and isinstance(r.get("answer"), str)
and r.get("metadata", {}).get("job_count", 0) > 0
and "`help`" not in r["answer"]
),
)
# Spot-check that a couple of known jobs appear in the listing.
answer = result["answer"]
app = await _make_app()
try:
resp = await app.run_job("help")
assert resp.success is True
assert isinstance(resp.answer, str)
assert resp.metadata.get("job_count", 0) > 0
assert "`help`" not in resp.answer
for expected_job in ("version", "health_check", "search"):
if expected_job not in answer:
raise AssertionError(f"help output missing job {expected_job!r}: {answer!r}")
assert expected_job in resp.answer, f"help missing {expected_job!r}: {resp.answer!r}"
finally:
await app.close()
print("✓ test_help_job passed")
_run(run())
@ -132,92 +131,61 @@ def test_search_job_empty_store():
async def run():
with tempfile.TemporaryDirectory() as tmp, _temp_chdir(tmp):
async with mock_reme_server() as (host, port):
await call_and_check(
"search",
host=host,
port=port,
query="hello world",
limit=5,
validator=lambda r: (
isinstance(r, dict)
and r.get("success") is True
and isinstance(r.get("metadata"), dict)
and isinstance(r["metadata"].get("counts"), dict)
and r["metadata"]["counts"].get("returned", -1) == 0
),
)
app = await _make_app()
try:
resp = await app.run_job("search", query="hello world", limit=5)
assert resp.success is True
counts = resp.metadata.get("counts", {})
assert isinstance(counts, dict)
assert counts.get("returned", -1) == 0
finally:
await app.close()
print("✓ test_search_job_empty_store passed")
_run(run())
def test_search_job_missing_query():
"""search without a query should surface the assertion error in `answer`."""
"""search with empty query returns success=False and a query-related error in answer."""
async def run():
with tempfile.TemporaryDirectory() as tmp, _temp_chdir(tmp):
async with mock_reme_server() as (host, port):
result = await call_action("search", host=host, port=port, query="")
if not isinstance(result, dict):
raise AssertionError(f"expected dict response, got {result!r}")
if "query" not in str(result.get("answer", "")).lower():
raise AssertionError(f"expected query-related error in answer, got {result!r}")
app = await _make_app()
try:
resp = await app.run_job("search", query="")
assert resp.success is False
assert "query" in str(resp.answer).lower()
finally:
await app.close()
print("✓ test_search_job_missing_query passed")
_run(run())
# -- aggregate: reuse one server instance for all jobs -------------------
def test_all_jobs_one_server():
"""Run every common job against a single shared server for efficiency."""
def test_all_jobs_single_app():
"""Run every common job against one shared in-process Application for efficiency."""
async def run():
with tempfile.TemporaryDirectory() as tmp, _temp_chdir(tmp):
async with mock_reme_server() as (host, port):
# version
await call_and_check(
"version",
host=host,
port=port,
validator=lambda r: isinstance(r, dict) and r.get("answer") == REME_VERSION,
)
# help
await call_and_check(
"help",
host=host,
port=port,
validator=lambda r: isinstance(r, dict) and r.get("metadata", {}).get("job_count", 0) > 0,
)
# health_check
await call_and_check(
"health_check",
host=host,
port=port,
validator=lambda r: isinstance(r, dict)
and isinstance(
r.get("metadata", {}).get("health"),
dict,
),
)
# search (empty store)
await call_and_check(
"search",
host=host,
port=port,
query="anything",
validator=lambda r: isinstance(r, dict) and r.get("success") is True,
)
# reindex
await call_and_check(
"reindex",
host=host,
port=port,
validator=lambda r: isinstance(r, dict) and isinstance(r.get("metadata", {}).get("counts"), dict),
)
print("✓ test_all_jobs_one_server passed")
app = await _make_app()
try:
resp = await app.run_job("version")
assert resp.answer == REME_VERSION
resp = await app.run_job("help")
assert resp.metadata.get("job_count", 0) > 0
resp = await app.run_job("health_check")
assert isinstance(resp.metadata.get("health"), dict)
resp = await app.run_job("search", query="anything")
assert resp.success is True
resp = await app.run_job("reindex")
assert isinstance(resp.metadata.get("counts"), dict)
finally:
await app.close()
print("✓ test_all_jobs_single_app passed")
_run(run())
@ -365,12 +333,12 @@ def test_traverse_both_directions():
if __name__ == "__main__":
print("\n=== reme4 common steps E2E tests ===")
print("\n=== reme4 common steps in-process tests ===")
test_version_job()
test_help_job()
test_search_job_empty_store()
test_search_job_missing_query()
test_all_jobs_one_server()
test_all_jobs_single_app()
print("\n=== traverse step tests ===")
test_traverse_forward_depth_1()
test_traverse_backward_returns_inlinks()