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

feat(file_graph): add optional paths parameter to get_nodes method

Browse source 
Add support for `paths=None` in `get_nodes` method to return all real
nodes as the graph's "scan everything" entry point. When `paths` is
explicitly passed as an empty list, returns empty list. Virtual
placeholders are filtered out in both cases.

BREAKING CHANGE: The `get_nodes` method signature changed from
requiring a list of paths to accepting an optional list or None.

refactor(memory): move runtime_response imports to steps package

Move runtime_response imports from memory package to steps package to
improve code organization and reduce circular dependencies.

refactor(config): update obsidian configuration table structure

Update the obsidian configuration markdown table to improve clarity
and fix incorrect section headings. Change 'stat/list' to 'file' and
'return specific tag info' to 'tags' section with proper commands.

chore(memory): remove deprecated toolkit modules

Remove deprecated agent_toolkit.py, file_toolkit.py, and
graph_toolkit.py modules as functionality has been moved to steps
package.

feat(steps): add new CRUD file operations

Add new steps for file operations including file_download and
file_list operations with proper path resolution and filtering
capabilities.
This commit is contained in:
huangsen 2026-05-15 14:55:08 +08:00
parent 30971af5ec
commit 1852d05f7b
27 changed files with 1033 additions and 499 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

11
reme2/component/file_graph/base_file_graph.py
View file

@ -26,8 +26,15 @@ class BaseFileGraph(BaseComponent):
"""Delete nodes from the graph."""
@abstractmethod
async def get_nodes(self, paths: list[str]) -> list[FileNode]:
"""Get nodes from the graph."""
async def get_nodes(self, paths: list[str] | None = None) -> list[FileNode]:
"""Get nodes from the graph.
``paths=None`` (the default) returns every real node — the
graph's "scan everything" entry point. Pass an explicit list
for path-known lookups; ``[]`` returns ``[]`` (empty input,
empty output). Virtual placeholders (nodes that exist only
because something links to them) are filtered out either way.
"""
@abstractmethod
async def rebuild_links(self) -> None:

4
reme2/component/file_graph/local_file_graph.py
View file

@ -98,7 +98,9 @@ class LocalFileGraph(BaseFileGraph):
if demoted:
self._pending.setdefault(path, set()).update(demoted)
async def get_nodes(self, paths: list[str]) -> list[FileNode]:
async def get_nodes(self, paths: list[str] | None = None) -> list[FileNode]:
if paths is None:
return list(self._nodes.values())
return [self._nodes[p] for p in paths if p in self._nodes]
async def rebuild_links(self) -> None:

38
reme2/component/file_graph/neo4j_file_graph.py
View file

@ -231,20 +231,34 @@ class Neo4jFileGraph(BaseFileGraph):
paths=paths,
)
async def get_nodes(self, paths: list[str]) -> list[FileNode]:
"""Return only real nodes (virtual placeholders are filtered)."""
if not paths:
async def get_nodes(self, paths: list[str] | None = None) -> list[FileNode]:
"""Return real nodes (virtual placeholders filtered).
``paths=None`` streams every real node ordered by path. An
explicit ``[]`` returns ``[]`` without hitting the database.
"""
if paths is not None and not paths:
return []
async with self._session() as session:
rec = await session.run(
"""
UNWIND $paths AS p
MATCH (f:File {path: p})
WHERE f.links_json IS NOT NULL
RETURN f
""",
paths=list(paths),
)
if paths is None:
rec = await session.run(
"""
MATCH (f:File)
WHERE f.links_json IS NOT NULL
RETURN f
ORDER BY f.path ASC
""",
)
else:
rec = await session.run(
"""
UNWIND $paths AS p
MATCH (f:File {path: p})
WHERE f.links_json IS NOT NULL
RETURN f
""",
paths=list(paths),
)
rows = [row["f"] async for row in rec]
return [self._row_to_node(row) for row in rows]

8
reme2/component/file_graph/nx_file_graph.py
View file

@ -95,8 +95,14 @@ class NxFileGraph(BaseFileGraph):
if self._graph.in_degree(path) == 0:
self._graph.remove_node(path)
async def get_nodes(self, paths: list[str]) -> list[FileNode]:
async def get_nodes(self, paths: list[str] | None = None) -> list[FileNode]:
nodes_view = self._graph.nodes
if paths is None:
return [
data["node"]
for _, data in nodes_view(data=True)
if "node" in data
]
return [
nodes_view[path]["node"]
for path in paths

10
reme2/config/obsidian.md
View file

@ -8,14 +8,16 @@
| 通用 | version | |
| search | search | query="search term" limit=10 tag="[]" score=0.1 copy=true |
| 通用 | tags | |@sen
@sen
| tags | stat | 返回特定tag信息 |
| tags | list | 返回所有tag列表 |
| crud | upload/download | 其他文件 |
| stat/list | stat | path |
| stat/list | list | path |
| file | stat | path |
| file | list | path |
| property | property:read | |
| property | property:update | path="My Note" status=done xx=xxx |
| property | property:delete | keys="[xxxx, xxxx]" |
| link | walk | path="My Note" directtion=forward/backward depth=1 predicat=xxx |
| graph | traverse | path="My Note" directtion=forward/backward depth=1 predicat=xxx |
@wangce
| crud | create | path="New Note" content="# Hello" title="xxx" tags="[]" status="" |

78
reme2/memory/agent_toolkit.py
View file

@ -1,78 +0,0 @@
"""Agent toolkit — composition entry point for the agent's tool surface.
The actual tool implementations are split by category across four
modules, each cohesive and self-contained:
memory_toolkit.py memory_get / memory_create / memory_update_body /
memory_update_meta / memory_search
+ memory_graph_search (MCP-only)
+ schema policy helpers (path template, status
state machine, create_with_schema, update_status)
file_toolkit.py file_download / file_upload / file_delete /
file_list / file_move
+ path resolution + session temp dir
graph_toolkit.py graph_traverse + adjacency BFS
event_toolkit.py event_open / event_complete
— uses memory_toolkit's schema gates so the
event index follows the memory schema
This module just composes them into one ``Toolkit`` for hosts that
want all 13 tools bound at once. Hosts that want a subset can import
the per-category builders and pass through a shared ``Toolkit``.
Lint tools (``check_dangling`` / ``check_orphans`` / ``check_collisions``
/ ``check_schema``) live in ``lint_toolkit.py`` — separate factory,
not bound here by default.
"""
from __future__ import annotations
from agentscope.tool import Toolkit
from ..component import R
from ..enumeration import ComponentEnum
from .event_toolkit import EVENT_TOOL_NAMES
from .file_toolkit import FILE_TOOL_NAMES
from .graph_toolkit import GRAPH_TOOL_NAMES
from .memory_toolkit import MEMORY_TOOL_NAMES
# Order: memory → file → graph → event. Within memory, search comes
# last (retrieval is a separate concern from CRUD); the rest follow
# their CRUD order. memory_graph_search stays out — it's MCP-only.
AGENT_TOOL_NAMES: tuple[str, ...] = (
*MEMORY_TOOL_NAMES,
*FILE_TOOL_NAMES,
*GRAPH_TOOL_NAMES,
*EVENT_TOOL_NAMES,
)
def build_agent_toolkit(
app_context,
audit: list[dict] | None = None,
toolkit: Toolkit | None = None,
) -> Toolkit:
"""Bind every agent tool's method to an agentscope ``Toolkit``.
For each name in ``AGENT_TOOL_NAMES``, instantiates the registered
BaseStep against ``app_context``, attaches the shared ``audit``
list, and registers the same-named class method as a tool function.
agentscope introspects the method signature directly — no separate
JSON schema layer.
"""
toolkit = toolkit or Toolkit()
for name in AGENT_TOOL_NAMES:
step_cls = R.get(ComponentEnum.STEP, name)
if step_cls is None:
continue
instance = step_cls(app_context=app_context)
instance.audit = audit # type: ignore[attr-defined]
toolkit.register_tool_function(
getattr(instance, name),
namesake_strategy="override",
)
return toolkit

2
reme2/memory/event_toolkit.py
View file

@ -39,7 +39,7 @@ from ..component import R
from ..component.base_step import BaseStep
from .file_toolkit import resolve_vault_path
from .memory_toolkit import create_with_schema, update_status
from .runtime_response import _set_answer, _tool_response
from ..steps.runtime_response import _set_answer, _tool_response
# ===========================================================================

248
reme2/memory/file_toolkit.py
View file

@ -1,248 +0,0 @@
"""File category — type-agnostic vault transport + directory operations.
Five tools:
file_download copy vault file → session temp dir
file_upload copy local file → vault path (any type)
file_delete remove vault file (universal entry point)
file_list enumerate vault files with optional filters
file_move rename / relocate; optional inbound-link rewrite
Session-scoped temp dir is lazy: created on first ``file_download``,
auto-cleaned on process exit. Each download lands in a fresh
sub-directory so concurrent agents can freely modify the temp file
without trampling each other.
"""
from __future__ import annotations
import mimetypes
import shutil
import tempfile
from pathlib import Path
from agentscope.tool import ToolResponse
from . import memory_io
from ..component import R
from ..component.base_step import BaseStep
from .runtime_response import _set_answer, _tool_response
# ===========================================================================
# Section 1 — Helpers (session temp dir + path resolution)
# ===========================================================================
_TEMP_ROOT: Path | None = None
def _get_temp_root() -> Path:
"""Lazy session-scoped temp dir. Auto-cleaned on process exit."""
global _TEMP_ROOT
if _TEMP_ROOT is None:
_TEMP_ROOT = Path(tempfile.mkdtemp(prefix="reme2-files-"))
return _TEMP_ROOT
def resolve_vault_path(file_store, vault_path: str) -> Path:
"""Compose the absolute on-disk path for a vault-relative entry.
Exposed (no underscore) so ``event_toolkit`` can reuse the same
path-resolution logic when computing event workspace locations.
"""
working_dir = getattr(file_store, "working_dir", None) or "."
p = Path(vault_path)
if p.is_absolute():
return p.resolve()
return (Path(working_dir) / p).resolve()
# ===========================================================================
# Section 2 — File tools
# ===========================================================================
@R.register("file_download")
class FileDownload(BaseStep):
"""Copy a vault file to a session temp dir; return the local path."""
audit: list[dict] | None = None
async def execute(self):
assert self.context is not None
vault_path: str = self.context.get("vault_path", "") or ""
assert vault_path, "vault_path is required"
payload = self._download(vault_path)
self.context.response.success = "error" not in payload
_set_answer(self.context, payload)
async def file_download(self, vault_path: str) -> ToolResponse:
"""Copy a vault file to session temp dir; return the local path."""
payload = self._download(vault_path)
ok = "error" not in payload
return _tool_response("file_download", ok, payload, audit=self.audit)
def _download(self, vault_path: str) -> dict:
src = resolve_vault_path(self.file_store, vault_path)
if not src.is_file():
return {"vault_path": vault_path, "error": "not found"}
dst_dir = Path(tempfile.mkdtemp(prefix="dl-", dir=_get_temp_root()))
dst = dst_dir / src.name
shutil.copy2(src, dst)
return {
"vault_path": vault_path,
"local_path": str(dst),
"size": dst.stat().st_size,
}
@R.register("file_upload")
class FileUpload(BaseStep):
"""Copy a local file into the vault. Watcher / parser register the
FileNode asynchronously."""
audit: list[dict] | None = None
async def execute(self):
assert self.context is not None
local_path: str = self.context.get("local_path", "") or ""
vault_path: str = self.context.get("vault_path", "") or ""
overwrite: bool = bool(self.context.get("overwrite", True))
assert local_path and vault_path, "local_path and vault_path are required"
payload = self._upload(local_path, vault_path, overwrite)
self.context.response.success = "error" not in payload
_set_answer(self.context, payload)
async def file_upload(
self, local_path: str, vault_path: str, overwrite: bool = True,
) -> ToolResponse:
"""Copy local_path into the vault at vault_path."""
payload = self._upload(local_path, vault_path, overwrite)
ok = "error" not in payload
return _tool_response("file_upload", ok, payload, audit=self.audit)
def _upload(self, local_path: str, vault_path: str, overwrite: bool) -> dict:
src = Path(local_path)
if not src.is_file():
return {"local_path": local_path, "error": "source not found"}
dst = resolve_vault_path(self.file_store, vault_path)
if dst.exists() and not overwrite:
return {"vault_path": vault_path, "error": "destination exists; pass overwrite=True"}
dst.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(src, dst)
return {
"vault_path": vault_path,
"size": dst.stat().st_size,
"mime": mimetypes.guess_type(dst.name)[0] or "application/octet-stream",
}
@R.register("file_delete")
class FileDelete(BaseStep):
"""Delete a vault file. Universal entry point for any file type."""
audit: list[dict] | None = None
async def execute(self):
assert self.context is not None
vault_path: str = self.context.get("vault_path", "") or ""
assert vault_path, "vault_path is required"
target = resolve_vault_path(self.file_store, vault_path)
ok, payload = memory_io.delete_file(target)
self.context.response.success = ok
_set_answer(self.context, payload)
async def file_delete(self, vault_path: str) -> ToolResponse:
"""Delete a vault file."""
target = resolve_vault_path(self.file_store, vault_path)
ok, payload = memory_io.delete_file(target)
return _tool_response("file_delete", ok, payload, audit=self.audit)
@R.register("file_list")
class FileList(BaseStep):
"""Enumerate vault files with optional frontmatter filters."""
audit: list[dict] | None = None
async def execute(self):
assert self.context is not None
result = memory_io.list_files(
self.file_store,
path_prefix=self.context.get("prefix") or self.context.get("path_prefix"),
tags=self.context.get("tags") or [],
metadata=self.context.get("metadata") or {},
limit=int(self.context.get("limit") or 100),
)
_set_answer(self.context, result)
async def file_list(
self,
prefix: str | None = None,
tags: list[str] | None = None,
metadata: dict | None = None,
limit: int = 100,
) -> ToolResponse:
"""List vault files. Filters: path prefix, frontmatter tags / fields."""
result = memory_io.list_files(
self.file_store,
path_prefix=prefix,
tags=tags or [],
metadata=metadata or {},
limit=limit,
)
return _tool_response("file_list", True, result, audit=self.audit)
@R.register("file_move")
class FileMove(BaseStep):
"""Rename / relocate. Default leaves inbound wikilinks untouched
(maintainer cleans dangling refs); pass ``update_refs=True`` to
rewrite ``[[old]] → [[new]]`` in every referencing file."""
audit: list[dict] | None = None
async def execute(self):
assert self.context is not None
src: str = self.context.get("src") or self.context.get("old_path") or ""
dst: str = self.context.get("dst") or self.context.get("new_path") or ""
update_refs: bool = bool(self.context.get("update_refs", False))
assert src and dst, "src and dst are required"
payload = self._move(src, dst, update_refs)
self.context.response.success = payload.get("ok", False)
_set_answer(self.context, payload)
async def file_move(
self, src: str, dst: str, update_refs: bool = False,
) -> ToolResponse:
"""Rename / relocate. update_refs=True rewrites [[old]] → [[new]]."""
payload = self._move(src, dst, update_refs)
ok = payload.get("ok", False)
return _tool_response("file_move", ok, payload, audit=self.audit)
def _move(self, src: str, dst: str, update_refs: bool) -> dict:
src_abs = resolve_vault_path(self.file_store, src)
dst_abs = resolve_vault_path(self.file_store, dst)
if not src_abs.is_file():
return {"ok": False, "src": src, "error": "source not found"}
if update_refs:
working_dir = Path(getattr(self.file_store, "working_dir", None) or ".").resolve()
ok, payload = memory_io.rename_file(
self.file_store, working_dir,
old_path=src_abs, new_path=dst_abs,
)
payload["ok"] = ok
return payload
dst_abs.parent.mkdir(parents=True, exist_ok=True)
shutil.move(str(src_abs), str(dst_abs))
return {"ok": True, "src": str(src_abs), "dst": str(dst_abs), "refs_updated": 0}
FILE_TOOL_NAMES: tuple[str, ...] = (
"file_download",
"file_upload",
"file_delete",
"file_list",
"file_move",
)

148
reme2/memory/graph_toolkit.py
View file

@ -1,148 +0,0 @@
"""Graph category — relationship exploration via BFS.
Single tool: ``graph_traverse``. depth=1 covers the trivial outlinks /
inlinks lookups (just set direction); multi-hop covers exploration.
Output is one record per edge traversed (not per node), so the same
node may appear multiple times if reached via different predicates or
paths — agents dedupe at the call site if they want a flat node set.
Adjacency source: walks ``file_store.file_nodes`` directly (each
FileNode carries its ``links`` list). Inbound is a linear scan over
all nodes; cheap for vault sizes. Swap for a precomputed reverse
index on the file_graph component if it ever becomes a hot path.
"""
from __future__ import annotations
from collections import deque
from agentscope.tool import ToolResponse
from ..component import R
from ..component.base_step import BaseStep
from .runtime_response import _set_answer, _tool_response
# ===========================================================================
# Section 1 — Adjacency lookups
# ===========================================================================
def _outlinks(file_store, path: str) -> list[tuple[str, str | None, str | None]]:
"""Outgoing edges from ``path`` — [(target_path, predicate, anchor)]."""
node = file_store.file_nodes.get(path)
if node is None:
return []
return [(link.path, link.predicate, link.anchor) for link in node.links if link.path]
def _inlinks(file_store, path: str) -> list[tuple[str, str | None, str | None]]:
"""Incoming edges to ``path`` — linear scan over all nodes' links."""
out: list[tuple[str, str | None, str | None]] = []
for src_path, src_node in file_store.file_nodes.items():
if src_path == path:
continue
for link in src_node.links:
if link.path == path:
out.append((src_path, link.predicate, link.anchor))
return out
def _bfs(
file_store,
seeds: list[str],
max_depth: int,
direction: str,
predicate: str | None,
) -> list[dict]:
"""BFS from each seed. One record per edge traversed."""
visited_edges: set[tuple[str, str, str | None]] = set()
results: list[dict] = []
queue: deque[tuple[str, int]] = deque((s, 0) for s in seeds)
while queue:
current, depth = queue.popleft()
if depth >= max_depth:
continue
edges: list[tuple[str, str | None, str | None]] = []
if direction in ("out", "both"):
for tgt, pred, anchor in _outlinks(file_store, current):
if predicate is not None and pred != predicate:
continue
edges.append((tgt, pred, anchor))
if direction in ("in", "both"):
for src, pred, anchor in _inlinks(file_store, current):
if predicate is not None and pred != predicate:
continue
edges.append((src, pred, 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))
return results
# ===========================================================================
# Section 2 — Graph tool
# ===========================================================================
@R.register("graph_traverse")
class GraphTraverse(BaseStep):
"""BFS from seed(s) to explore relationships in the memory graph."""
audit: list[dict] | None = None
async def execute(self):
assert self.context is not None
seeds_raw = self.context.get("seeds") or []
if isinstance(seeds_raw, str):
seeds = [seeds_raw]
else:
seeds = list(seeds_raw)
max_depth: int = int(self.context.get("max_depth") or 1)
direction: str = self.context.get("direction", "out") or "out"
predicate = self.context.get("predicate")
assert seeds, "seeds is required (single path or list of paths)"
assert direction in ("out", "in", "both"), \
f"direction must be 'out' | 'in' | 'both', got {direction!r}"
results = _bfs(self.file_store, seeds, max_depth, direction, predicate)
_set_answer(self.context, results)
async def graph_traverse(
self,
seeds: str | list[str],
max_depth: int = 1,
direction: str = "out",
predicate: str | None = None,
) -> ToolResponse:
"""BFS from seed(s) to explore relationships.
Args:
seeds: single path or list of paths to start from.
max_depth: hops to expand (default 1 = immediate neighbors).
direction: "out" / "in" / "both".
predicate: filter edges by predicate (None = no filter).
"""
if isinstance(seeds, str):
seeds_list = [seeds]
else:
seeds_list = list(seeds)
assert direction in ("out", "in", "both"), \
f"direction must be 'out' | 'in' | 'both', got {direction!r}"
results = _bfs(self.file_store, seeds_list, max_depth, direction, predicate)
return _tool_response("graph_traverse", True, results, audit=self.audit)
GRAPH_TOOL_NAMES: tuple[str, ...] = (
"graph_traverse",
)

2
reme2/memory/ingestor.py
View file

@ -28,7 +28,7 @@ from agentscope.message import Msg
from agentscope.tool import Toolkit
from pydantic import BaseModel, Field
from .runtime_response import _set_answer, _to_jsonable
from ..steps.runtime_response import _set_answer, _to_jsonable
from . import memory_io
from .memory_io import create_file
from .agent_toolkit import build_agent_toolkit

2
reme2/memory/lint_toolkit.py
View file

@ -28,7 +28,7 @@ from agentscope.tool import Toolkit, ToolResponse
from ..component import R
from ..component.base_step import BaseStep
from .runtime_response import _set_answer, _tool_response
from ..steps.runtime_response import _set_answer, _tool_response
from ..enumeration import ComponentEnum

2
reme2/memory/maintainer.py
View file

@ -69,7 +69,7 @@ from pydantic import BaseModel, Field
from ..component import R
from ..component.base_step import BaseStep
from .runtime_response import _set_answer
from ..steps.runtime_response import _set_answer
from . import memory_io
from .schema import parse_frontmatter

2
reme2/memory/memory_toolkit.py
View file

@ -29,7 +29,7 @@ from ..component import R
from ..component.base_step import BaseStep
from ..enumeration import ComponentEnum
from .retriever import BaseRetriever, HybridRetriever
from .runtime_response import _set_answer, _tool_response
from ..steps.runtime_response import _set_answer, _tool_response
# ===========================================================================

89
reme2/steps/crud/download.py Normal file
View file

@ -0,0 +1,89 @@
"""``file_download`` — copy a vault file to a session temp dir.
Agent flow: agent calls ``file_download(path)``, gets back a
local path under a fresh per-call temp directory, then opens / parses
the file with whatever tooling it likes. The vault copy is untouched.
The temp root is lazy and session-scoped — created on first download,
left for the OS to clean up at process exit. Each download lands in
its own subdirectory so concurrent agents don't trample each other.
Also exports ``resolve_path`` — the shared helper for turning
a vault-relative or absolute path into an absolute on-disk path.
``upload`` and ``list`` import it from here.
"""
from __future__ import annotations
import shutil
import tempfile
from pathlib import Path
from agentscope.tool import ToolResponse
from ..base_step import BaseStep
from ..runtime_response import _set_answer, _tool_response
from ...component import R
from ...enumeration import ComponentEnum
_TEMP_ROOT: Path | None = None
def _get_temp_root() -> Path:
"""Lazy session-scoped temp dir. Auto-cleaned on process exit."""
global _TEMP_ROOT
if _TEMP_ROOT is None:
_TEMP_ROOT = Path(tempfile.mkdtemp(prefix="reme2-files-"))
return _TEMP_ROOT
def resolve_path(file_store, path: str) -> Path:
"""Compose the absolute on-disk path for relative entry.
Public so sibling steps (``upload``, ``list``, event tools) can
reuse the same path-resolution rule. Absolute paths pass through;
relative paths join under ``file_store.working_dir``.
"""
working_dir = getattr(file_store, "working_dir", None) or "."
p = Path(path)
if p.is_absolute():
return p.resolve()
return (Path(working_dir) / p).resolve()
@R.register("file_download")
class FileDownload(BaseStep):
"""Copy a vault file to a session temp dir; return the local path."""
component_type = ComponentEnum.STEP
audit: list[dict] | None = None
async def execute(self):
assert self.context is not None
path: str = self.context.get("path", "") or ""
assert path, "path is required"
payload = self._download(path)
self.context.response.success = "error" not in payload
_set_answer(self.context, payload)
async def file_download(self, path: str) -> ToolResponse:
"""Copy a vault file to session temp dir; return the local path."""
payload = self._download(path)
ok = "error" not in payload
return _tool_response("file_download", ok, payload, audit=self.audit)
def _download(self, path: str) -> dict:
src = resolve_path(self.file_store, path)
if not src.is_file():
return {"path": path, "error": "not found"}
dst_dir = Path(tempfile.mkdtemp(prefix="dl-", dir=_get_temp_root()))
dst = dst_dir / src.name
shutil.copy2(src, dst)
return {
"path": path,
"local_path": str(dst),
"size": dst.stat().st_size,
}

101
reme2/steps/crud/list.py Normal file
View file

@ -0,0 +1,101 @@
"""``file_list`` — enumerate vault files with optional frontmatter filters.
Walks the file_graph in one shot via ``get_nodes(None)`` — no filesystem
scan, no per-file frontmatter parse, no per-file graph round-trip.
Filters compose:
path_prefix — prefix match against the indexed path string
tags — every requested tag must be present in ``tags``
metadata — frontmatter must equal each ``{key: value}`` pair
The ``metadata`` filter sees the full frontmatter dict — typed fields
(title / description / tags) merged with any ``extra=allow`` extras —
so agents can filter on schema-known keys or arbitrary extras alike.
"""
from __future__ import annotations
from agentscope.tool import ToolResponse
from ..base_step import BaseStep
from ..runtime_response import _set_answer, _tool_response
from ...component import R
from ...enumeration import ComponentEnum
def _matches(
path: str,
md: dict,
*,
path_prefix: str | None,
tags: list[str],
metadata: dict,
) -> bool:
if path_prefix and not path.startswith(path_prefix):
return False
if metadata and any(md.get(k) != v for k, v in metadata.items()):
return False
if tags:
file_tags = set(md.get("tags") or [])
if not all(t in file_tags for t in tags):
return False
return True
async def _list(
file_store,
*,
path_prefix: str | None,
tags: list[str],
metadata: dict,
limit: int,
) -> dict:
if not file_store.file_graph:
return {"items": [], "count": 0}
items: list[dict] = []
for node in await file_store.file_graph.get_nodes():
md = node.front_matter.model_dump()
if not _matches(node.path, md, path_prefix=path_prefix, tags=tags, metadata=metadata):
continue
items.append({"path": node.path, "metadata": md})
if len(items) >= limit:
break
return {"items": items, "count": len(items)}
@R.register("file_list")
class FileList(BaseStep):
"""Enumerate vault files with optional frontmatter filters."""
component_type = ComponentEnum.STEP
audit: list[dict] | None = None
async def execute(self):
assert self.context is not None
result = await _list(
self.file_store,
path_prefix=self.context.get("prefix") or self.context.get("path_prefix"),
tags=self.context.get("tags") or [],
metadata=self.context.get("metadata") or {},
limit=int(self.context.get("limit") or 100),
)
_set_answer(self.context, result)
async def file_list(
self,
prefix: str | None = None,
tags: list[str] | None = None,
metadata: dict | None = None,
limit: int = 100,
) -> ToolResponse:
"""List vault files. Filters: path prefix, frontmatter tags / fields."""
result = await _list(
self.file_store,
path_prefix=prefix,
tags=tags or [],
metadata=metadata or {},
limit=limit,
)
return _tool_response("file_list", True, result, audit=self.audit)

17
reme2/steps/crud/property/__init__.py Normal file
View file

@ -0,0 +1,17 @@
"""Property steps — frontmatter-only CRUD on memory files.
Three Steps:
property:read — return the frontmatter dict
property:update — merge a patch into the frontmatter
property:delete — drop the listed keys
Body content stays untouched; use the ``crud`` siblings (``read``,
``edit``, ``append``, ``prepend``) for body-level operations. Each
Step here is a pure disk read-modify-write — the watcher / parser
notices the change and refreshes the projections asynchronously.
"""
from . import read # noqa: F401 -- @R.register("property:read")
from . import update # noqa: F401 -- @R.register("property:update")
from . import delete # noqa: F401 -- @R.register("property:delete")

79
reme2/steps/crud/property/delete.py Normal file
View file

@ -0,0 +1,79 @@
"""``property:delete`` — remove keys from a memory file's frontmatter.
Returns both ``deleted`` (keys that were present and removed) and
``missing`` (keys that weren't there) so the agent can tell whether
a no-op happened. The file is rewritten only when at least one key
is actually removed — calling delete with all-missing keys is a
zero-side-effect read.
"""
from __future__ import annotations
import frontmatter
from agentscope.tool import ToolResponse
from ..download import resolve_path
from ...base_step import BaseStep
from ...runtime_response import _set_answer, _tool_response
from ....component import R
from ....enumeration import ComponentEnum
def _delete(file_store, path: str, keys: list[str]) -> dict:
target = resolve_path(file_store, path)
if not target.is_file():
return {"path": path, "error": "not found"}
if not keys:
return {"path": path, "error": "keys is empty"}
raw = target.read_text(encoding="utf-8")
post = frontmatter.loads(raw)
deleted: list[str] = []
missing: list[str] = []
for k in keys:
if k in post.metadata:
del post.metadata[k]
deleted.append(k)
else:
missing.append(k)
if deleted:
target.write_text(frontmatter.dumps(post), encoding="utf-8")
return {
"path": path,
"deleted": deleted,
"missing": missing,
"frontmatter": dict(post.metadata),
}
@R.register("property:delete")
class PropertyDelete(BaseStep):
"""Remove keys from a memory file's frontmatter."""
component_type = ComponentEnum.STEP
audit: list[dict] | None = None
async def execute(self):
assert self.context is not None
path: str = self.context.get("path") or self.context.get("path") or ""
assert path, "path is required"
keys = self.context.get("keys") or []
if isinstance(keys, str):
keys = [keys]
payload = _delete(self.file_store, path, list(keys))
self.context.response.success = "error" not in payload
_set_answer(self.context, payload)
async def property_delete(
self,
path: str,
keys: list[str] | str,
) -> ToolResponse:
"""Remove the listed ``keys`` from the frontmatter at ``path``."""
if isinstance(keys, str):
keys = [keys]
payload = _delete(self.file_store, path, list(keys))
ok = "error" not in payload
return _tool_response("property:delete", ok, payload, audit=self.audit)

51
reme2/steps/crud/property/read.py Normal file
View file

@ -0,0 +1,51 @@
"""``property:read`` — return the frontmatter dict of a memory file.
Cheap structured read — frontmatter only, no body. Use ``crud:read``
when you need the body too. Returns ``{exists: false}`` when the
target doesn't exist; otherwise ``{exists: true, frontmatter: {...}}``.
"""
from __future__ import annotations
import frontmatter
from agentscope.tool import ToolResponse
from ..download import resolve_path
from ...base_step import BaseStep
from ...runtime_response import _set_answer, _tool_response
from ....component import R
from ....enumeration import ComponentEnum
def _read(file_store, path: str) -> dict:
target = resolve_path(file_store, path)
if not target.is_file():
return {"path": path, "exists": False}
raw = target.read_text(encoding="utf-8")
meta = dict(frontmatter.loads(raw).metadata)
return {"path": path, "exists": True, "frontmatter": meta}
@R.register("property:read")
class PropertyRead(BaseStep):
"""Read a memory file's frontmatter (YAML metadata only)."""
component_type = ComponentEnum.STEP
audit: list[dict] | None = None
async def execute(self):
assert self.context is not None
path: str = self.context.get("path") or self.context.get("path") or ""
assert path, "path is required"
payload = _read(self.file_store, path)
self.context.response.success = payload.get("exists", False)
_set_answer(self.context, payload)
async def property_read(self, path: str) -> ToolResponse:
"""Return the frontmatter dict of the memory file at ``path``."""
payload = _read(self.file_store, path)
ok = payload.get("exists", False)
return _tool_response("property:read", ok, payload, audit=self.audit)

93
reme2/steps/crud/property/update.py Normal file
View file

@ -0,0 +1,93 @@
"""``property:update`` — merge a patch into a memory file's frontmatter.
Patch semantics are merge-write: keys in ``patch`` overwrite existing
keys, keys not in ``patch`` are left untouched. To remove a key,
use ``property:delete`` — passing ``None`` here just sets the literal
None (which is rarely what you want).
Per the obsidian config convention the agent calls this with
arbitrary keyword arguments — ``path="My Note" status=done xx=xxx``.
The Step packs everything except ``path`` into the patch dict, so
the tool method accepts ``**fields`` directly.
"""
from __future__ import annotations
import frontmatter
from agentscope.tool import ToolResponse
from ..download import resolve_path
from ...base_step import BaseStep
from ...runtime_response import _set_answer, _tool_response
from ....component import R
from ....enumeration import ComponentEnum
# Context keys that belong to plumbing (request envelope, response
# slot, the path itself) and must never be promoted into a free-form
# frontmatter patch.
_RESERVED_CONTEXT_KEYS = {
"path", "path", "patch", "response", "request", "data",
}
def _update(file_store, path: str, patch: dict) -> dict:
target = resolve_path(file_store, path)
if not target.is_file():
return {"path": path, "error": "not found"}
if not patch:
return {"path": path, "error": "patch is empty"}
raw = target.read_text(encoding="utf-8")
post = frontmatter.loads(raw)
post.metadata.update(patch)
target.write_text(frontmatter.dumps(post), encoding="utf-8")
return {
"path": path,
"applied": dict(patch),
"frontmatter": dict(post.metadata),
}
@R.register("property:update")
class PropertyUpdate(BaseStep):
"""Merge a patch into a memory file's frontmatter."""
component_type = ComponentEnum.STEP
audit: list[dict] | None = None
async def execute(self):
assert self.context is not None
path: str = self.context.get("path") or self.context.get("path") or ""
assert path, "path is required"
patch = self.context.get("patch")
if patch is None:
# Free-form mode — every other context key is treated as a
# patch entry, matching the obsidian convention
# `path=... key=val key=val`.
patch = {
k: v for k, v in self.context.data.items()
if k not in _RESERVED_CONTEXT_KEYS
}
payload = _update(self.file_store, path, dict(patch or {}))
self.context.response.success = "error" not in payload
_set_answer(self.context, payload)
async def property_update(
self,
path: str,
patch: dict | None = None,
**fields,
) -> ToolResponse:
"""Merge ``patch`` (or free-form ``**fields``) into the
frontmatter at ``path``."""
merged: dict = {}
if patch:
merged.update(patch)
if fields:
merged.update(fields)
payload = _update(self.file_store, path, merged)
ok = "error" not in payload
return _tool_response("property:update", ok, payload, audit=self.audit)

94
reme2/steps/crud/stat.py Normal file
View file

@ -0,0 +1,94 @@
"""``file_stat`` — peek at vault file metadata without copying it.
Cheap inspection alternative to ``file_download``: the agent gets
size, mtime, mime type, and (for markdown files) the parsed
frontmatter — enough to decide whether to download / parse / skip
without paying the copy cost.
Returns a uniform envelope:
{exists, type, size, mtime, ctime, mime, frontmatter}
``exists=False`` short-circuits everything else to ``None``. ``type``
is ``"file"`` / ``"dir"`` (covers event workspace probes too).
``frontmatter`` is populated only for ``.md`` files and only when
parsing succeeds — schema validity is a lint concern.
"""
from __future__ import annotations
import mimetypes
from datetime import datetime
from pathlib import Path
import frontmatter
from agentscope.tool import ToolResponse
from .download import resolve_path
from ..base_step import BaseStep
from ..runtime_response import _set_answer, _tool_response
from ...component import R
from ...enumeration import ComponentEnum
def _iso(ts: float) -> str:
"""Filesystem timestamp → ISO 8601 string."""
return datetime.fromtimestamp(ts).isoformat()
def _stat(file_store, path: str) -> dict:
target = resolve_path(file_store, path)
if not target.exists():
return {"path": path, "exists": False}
st = target.stat()
out: dict = {
"path": path,
"path": str(target),
"exists": True,
"type": "dir" if target.is_dir() else "file",
"mtime": _iso(st.st_mtime),
"ctime": _iso(st.st_ctime),
}
if target.is_file():
out["size"] = st.st_size
out["mime"] = mimetypes.guess_type(target.name)[0] or "application/octet-stream"
if target.suffix == ".md":
try:
meta = dict(frontmatter.loads(target.read_text(encoding="utf-8")).metadata)
except Exception:
meta = {}
out["frontmatter"] = meta
return out
@R.register("file_stat")
class FileStat(BaseStep):
"""Return metadata for a vault file or directory."""
component_type = ComponentEnum.STEP
audit: list[dict] | None = None
async def execute(self):
assert self.context is not None
path: str = self.context.get("path", "") or ""
assert path, "path is required"
payload = _stat(self.file_store, path)
self.context.response.success = payload.get("exists", False)
_set_answer(self.context, payload)
async def file_stat(self, path: str) -> ToolResponse:
"""Return metadata for a vault file or directory.
Returns ``{exists, type, size, mtime, ctime, mime, frontmatter}``.
``size`` / ``mime`` / ``frontmatter`` are file-only;
``frontmatter`` is markdown-only.
"""
payload = _stat(self.file_store, path)
ok = payload.get("exists", False)
return _tool_response("file_stat", ok, payload, audit=self.audit)

72
reme2/steps/crud/upload.py Normal file
View file

@ -0,0 +1,72 @@
"""``file_upload`` — copy a local file into the vault.
Agent flow: agent has a file at some local path (a download result, a
freshly generated artifact, a user-supplied attachment), and wants it
to live inside the vault at a known location. This is type-agnostic
— text, binary, anything — so it's the entry point for non-markdown
materials too.
The watcher / parser pick up the new file asynchronously; this step
just performs the copy. ``overwrite=True`` is the default since most
upload flows are intentional replacements (re-uploading the same
material after re-generation).
"""
from __future__ import annotations
import mimetypes
import shutil
from pathlib import Path
from agentscope.tool import ToolResponse
from .download import resolve_path
from ..base_step import BaseStep
from ..runtime_response import _set_answer, _tool_response
from ...component import R
from ...enumeration import ComponentEnum
@R.register("file_upload")
class FileUpload(BaseStep):
"""Copy a local file into the vault. Watcher / parser register the
FileNode asynchronously."""
component_type = ComponentEnum.STEP
audit: list[dict] | None = None
async def execute(self):
assert self.context is not None
local_path: str = self.context.get("local_path", "") or ""
path: str = self.context.get("path", "") or ""
overwrite: bool = bool(self.context.get("overwrite", True))
assert local_path and path, "local_path and path are required"
payload = self._upload(local_path, path, overwrite)
self.context.response.success = "error" not in payload
_set_answer(self.context, payload)
async def file_upload(
self, local_path: str, path: str, overwrite: bool = True,
) -> ToolResponse:
"""Copy ``local_path`` into the vault at ``path``."""
payload = self._upload(local_path, path, overwrite)
ok = "error" not in payload
return _tool_response("file_upload", ok, payload, audit=self.audit)
def _upload(self, local_path: str, path: str, overwrite: bool) -> dict:
src = Path(local_path)
if not src.is_file():
return {"local_path": local_path, "error": "source not found"}
dst = resolve_path(self.file_store, path)
if dst.exists() and not overwrite:
return {"path": path, "error": "destination exists; pass overwrite=True"}
dst.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(src, dst)
return {
"path": path,
"size": dst.stat().st_size,
"mime": mimetypes.guess_type(dst.name)[0] or "application/octet-stream",
}

15
reme2/steps/graph/__init__.py Normal file
View file

@ -0,0 +1,15 @@
"""Graph steps — wikilink relationship exploration.
Single Step:
graph:traverse — BFS from a seed file over wikilink edges.
Direction vocabulary matches the obsidian convention
(``forward`` / ``backward`` / ``both``); the engine vocab
(``out`` / ``in`` / ``both``) is accepted as alias. Inbound
traversal walks the vault to reconstruct source paths since the
file_graph contract's ``get_inlinks`` only returns target-shaped
FileLinks.
"""
from . import traverse # noqa: F401 -- @R.register("graph:traverse")

190
reme2/steps/graph/traverse.py Normal file
View file

@ -0,0 +1,190 @@
"""``graph:traverse`` — 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 obsidian convention
(``forward`` / ``backward`` / ``both``) and the engine convention
(``out`` / ``in`` / ``both``).
"""
from __future__ import annotations
from collections import deque
from pathlib import Path
from agentscope.tool import ToolResponse
from ..crud.download import resolve_path
from ..base_step import BaseStep
from ..runtime_response import _set_answer, _tool_response
from ...component import R
from ...enumeration import ComponentEnum
from ...schema import FileLink
_FORWARD = {"out", "forward"}
_BACKWARD = {"in", "backward"}
_BOTH = {"both"}
_VALID_DIRECTIONS = _FORWARD | _BACKWARD | _BOTH
async def _build_indexes(
file_store,
) -> tuple[
dict[str, list[tuple[str, FileLink]]],
dict[str, list[tuple[str, FileLink]]],
]:
"""One ``get_nodes(None)`` call → (outbound, inbound) adjacency dicts.
Each dict is keyed by node path; values are ``(neighbor_path, link)``
tuples. Source paths land in the inbound index alongside the link
object — solving the contract gap where ``get_inlinks`` returns
target-shaped FileLinks without source attribution.
"""
outbound: dict[str, list[tuple[str, FileLink]]] = {}
inbound: dict[str, list[tuple[str, FileLink]]] = {}
if not file_store.file_graph:
return outbound, inbound
for node in await file_store.file_graph.get_nodes():
for link in node.links:
if not link.path:
continue
outbound.setdefault(node.path, []).append((link.path, link))
inbound.setdefault(link.path, []).append((node.path, link))
return outbound, inbound
def _bfs(
seeds: list[str],
max_depth: int,
direction: str,
predicate: str | None,
outbound: dict[str, list[tuple[str, FileLink]]],
inbound: dict[str, list[tuple[str, FileLink]]],
) -> list[dict]:
"""In-memory BFS. One record per edge traversed."""
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((s, 0) for s in seeds)
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.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.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))
return results
def _normalize_seeds(file_store, raw) -> list[str]:
"""Coerce a single path or list of paths into resolved absolute strings."""
if isinstance(raw, (str, Path)):
items = [raw]
else:
items = list(raw or [])
return [str(resolve_path(file_store, str(p))) for p in items if p]
@R.register("graph:traverse")
class GraphTraverse(BaseStep):
"""BFS from a seed file to explore wikilink relationships.
Parameters (per obsidian convention):
path — single seed (str). ``seeds`` accepted as alias for batch mode.
direction — ``forward`` / ``backward`` / ``both`` (or ``out`` / ``in`` / ``both``).
depth — hop limit (default 1 = immediate neighbors).
predicate — optional edge-type filter; ``None`` = no filter.
"""
component_type = ComponentEnum.STEP
audit: list[dict] | None = None
async def execute(self):
assert self.context is not None
path = self.context.get("path")
seeds_raw = self.context.get("seeds") if path is None else path
seeds = _normalize_seeds(self.file_store, seeds_raw)
depth = int(self.context.get("depth") or self.context.get("max_depth") or 1)
direction = (self.context.get("direction") or "forward").lower()
predicate = self.context.get("predicate")
assert seeds, "path (or seeds) is required"
assert direction in _VALID_DIRECTIONS, (
f"direction must be one of {sorted(_VALID_DIRECTIONS)}, got {direction!r}"
)
outbound, inbound = await _build_indexes(self.file_store)
results = _bfs(seeds, depth, direction, predicate, outbound, inbound)
_set_answer(self.context, results)
async def graph_traverse(
self,
path: str | list[str],
direction: str = "forward",
depth: int = 1,
predicate: str | None = None,
) -> ToolResponse:
"""BFS from ``path`` over wikilink edges.
Args:
path: seed file (or list of seeds).
direction: ``forward`` (outbound) / ``backward`` (inbound) /
``both``. Aliases ``out`` / ``in`` accepted.
depth: hops to expand (default 1 = immediate neighbors).
predicate: filter edges by predicate (None = no filter).
"""
direction = direction.lower()
assert direction in _VALID_DIRECTIONS, (
f"direction must be one of {sorted(_VALID_DIRECTIONS)}, got {direction!r}"
)
seeds = _normalize_seeds(self.file_store, path)
assert seeds, "path is required"
outbound, inbound = await _build_indexes(self.file_store)
results = _bfs(seeds, depth, direction, predicate, outbound, inbound)
return _tool_response("graph:traverse", True, results, audit=self.audit)

0
reme2/memory/runtime_response.py → reme2/steps/runtime_response.py
View file

15
reme2/steps/tags/__init__.py Normal file
View file

@ -0,0 +1,15 @@
"""Tags steps — frontmatter-tag enumeration and per-tag statistics.
Two Steps:
tags:list — distinct tags + document counts across the vault.
tags:stat — count + file list for one specific tag.
Both consume what ``file_store`` already has in memory: path
discovery from ``file_store.file_chunks``, tag lookup from a single
batched ``file_graph.get_nodes(paths)`` call. No filesystem walk —
the cost is one dict iteration plus one graph round-trip.
"""
from . import list # noqa: F401 -- @R.register("tags:list")
from . import stat # noqa: F401 -- @R.register("tags:stat")

84
reme2/steps/tags/list.py Normal file
View file

@ -0,0 +1,84 @@
"""``tags:list`` — enumerate every tag declared in the vault.
Single ``file_graph.get_nodes(None)`` call streams every real node;
we accumulate tag counts in one pass. No filesystem walk, no
per-file frontmatter parse.
Returns:
{tags: [{tag, count}, ...], total: int}
Per-tag file lists live on ``tags:stat`` so this listing stays cheap.
"""
from __future__ import annotations
from collections import Counter
from agentscope.tool import ToolResponse
from ..base_step import BaseStep
from ..runtime_response import _set_answer, _tool_response
from ...component import R
from ...enumeration import ComponentEnum
async def _iter_tagged(file_store):
"""Yield ``(path, tags_list)`` for every indexed node that has tags.
Owns the tag-extraction rule so ``tags:stat`` shares it via import —
single source of truth for "what counts as tagged".
"""
if not file_store.file_graph:
return
for node in await file_store.file_graph.get_nodes():
tags = node.front_matter.tags or []
if tags:
yield node.path, [str(t) for t in tags if t]
async def _list(file_store, sort: str, limit: int | None) -> dict:
counter: Counter[str] = Counter()
async for _, tags in _iter_tagged(file_store):
counter.update(tags)
if sort == "alpha":
items = [{"tag": t, "count": c} for t, c in sorted(counter.items())]
else: # "count" (default) — desc count then alpha tiebreak
items = [
{"tag": t, "count": c}
for t, c in sorted(counter.items(), key=lambda kv: (-kv[1], kv[0]))
]
if limit is not None and limit > 0:
items = items[:limit]
return {"tags": items, "total": len(counter)}
@R.register("tags:list")
class TagsList(BaseStep):
"""List every distinct tag in the vault with its document count."""
component_type = ComponentEnum.STEP
audit: list[dict] | None = None
async def execute(self):
assert self.context is not None
sort = (self.context.get("sort") or "count").lower()
limit = self.context.get("limit")
payload = await _list(self.file_store, sort, int(limit) if limit else None)
_set_answer(self.context, payload)
async def tags_list(
self,
sort: str = "count",
limit: int | None = None,
) -> ToolResponse:
"""List every tag in the vault with its document count.
Args:
sort: ``count`` (default, descending) or ``alpha``.
limit: cap the number of returned tags; ``None`` = all.
"""
payload = await _list(self.file_store, sort.lower(), limit)
return _tool_response("tags:list", True, payload, audit=self.audit)

77
reme2/steps/tags/stat.py Normal file
View file

@ -0,0 +1,77 @@
"""``tags:stat`` — usage statistics for a single tag.
Reuses ``_iter_tagged`` from ``tags:list`` so node discovery and tag
extraction follow the same rule. Filters the stream by the requested
tag and collects matching paths.
Returns:
{tag, exists, count, paths, truncated}
``exists=False`` short-circuits to ``count=0`` and ``paths=[]`` —
useful for "does the agent need to introduce this tag?" probes.
``limit`` caps the path list (default 100); ``count`` is always
the true total.
"""
from __future__ import annotations
from agentscope.tool import ToolResponse
from .list import _iter_tagged
from ..base_step import BaseStep
from ..runtime_response import _set_answer, _tool_response
from ...component import R
from ...enumeration import ComponentEnum
async def _stat(file_store, tag: str, limit: int) -> dict:
matches: list[str] = []
count = 0
async for path, tags in _iter_tagged(file_store):
if tag in tags:
count += 1
if len(matches) < limit:
matches.append(path)
return {
"tag": tag,
"exists": count > 0,
"count": count,
"paths": matches,
"truncated": count > len(matches),
}
@R.register("tags:stat")
class TagsStat(BaseStep):
"""Usage statistics + file list for a single tag."""
component_type = ComponentEnum.STEP
audit: list[dict] | None = None
async def execute(self):
assert self.context is not None
tag: str = self.context.get("tag", "") or ""
assert tag, "tag is required"
limit = int(self.context.get("limit") or 100)
payload = await _stat(self.file_store, tag, limit)
self.context.response.success = True
_set_answer(self.context, payload)
async def tags_stat(
self,
tag: str,
limit: int = 100,
) -> ToolResponse:
"""Return usage statistics for ``tag``: count + file list.
Args:
tag: the tag to look up.
limit: cap the number of returned paths (count is always
the true total). Default 100.
"""
payload = await _stat(self.file_store, tag, limit)
return _tool_response("tags:stat", True, payload, audit=self.audit)