feat: implementation of the read step for reme (markdown) (#245)

* feat: implementation of the read step for reme (markdown)

* fix(step): markdown read step fixing pr comments

* fix(step): more  fixes for pr comments

* further fix for better review adaptation

* accept absolute path

* fixing base job exception
This commit is contained in:
imrewce 2026-05-19 16:19:13 +08:00 committed by jinli.yl
parent 2847d81200
commit d00eac5d53
10 changed files with 627 additions and 1 deletions

View file

@ -42,6 +42,7 @@ reme4 version
| 🔎 search | 🔍 `search` (`search_step`) | `call_server("search", query=…, …)` | 📥 `query:str` 🎚️ `limit:int=5`(>0) 🎚️ `min_score:float=0.0` ⚖️ `vector_weight:float=0.7` ∈[0,1]keyword 权 = 1-vw 🔀 `candidate_multiplier:float=3.0`candidates = min(200, limit×mult) 🔗 `expand_links:bool=True` 🔢 `max_links_per_direction:int=10` 🎚️ `search_filter:dict={}` 📤 `answer` 每命中一行 `path:start-end [score=… vector=… keyword=…] text` + 缩进的 `→ outlinks (n)` / `← inlinks (n)` + `via predicate=… anchor=#…` 📊 `metadata.results` / `metadata.link_expansion` / `metadata.counts={vector,keyword,returned,hybrid}` 🛠️ 并行 `vector_search` + `keyword_search` → RRF 融合K=60按 chunk.id 合并)→ `min_score` 过滤 → `limit` 截断 → 邻居 meta 注入 |
| 🧪 demo | 🪄 `demo_echo` (`demo_echo_step1` + `step2`) | `call_server("demo_echo", query=…, min_score=…)` | 📥 `query:str=""` 🎚️ `min_score:float=0.5` 🛠️ step1`processed_query = query.strip().lower()``adjusted_min_score = min_score * 0.9`,写回 context 📤 step2`answer = "echo: {processed_query} (min_score={adjusted_min_score})"` 📊 `metadata = {step, query, min_score, processed_query, adjusted_min_score}` |
| 🌊 demo | 🌊 `stream_demo` (`stream_demo_step1` + `step2`) | `call_server("stream_demo", query=…, repeat=…, interval=…)` | 📥 `query:str=""` 🎚️ `repeat:int=10` 🎚️ `interval:float=0.1`(秒/字符)| 🛠️ step1`stream_text = query * repeat` 写回 context 📤 step2按字符 `add_stream_string(ch, ChunkEnum.CONTENT)` 流式输出,`asyncio.sleep(interval)` 节流 |
| 📂 crud | 📖 `read` (`read_step`) | `call_server("read", path=…, …)` | 📥 `path:str` ⭐(**完整相对路径**,相对于 `working_dir`;绝对路径会被拒绝;非 `.md` 后缀拒绝)| 🎚️ `start_line:int=null`1-based, 含端点)| 🎚️ `end_line:int=null`1-based, 含端点)| 🎚️ `max_bytes:int=51200`(截断阈值)| 📤 `answer = 选中的行内容`,超过 `max_bytes` 时附加 `--- TRUNCATED ---` 续读指引(`start_line=…` 📊 `metadata.path` / `metadata.total_lines`(出错路径才会附带)| 🛠️ 流程:`BaseStep.resolve_path(raw, require_md=True)``aiofiles.os.stat``read_file_safe`utf-8-sig BOM 容忍、UnicodeDecodeError fallback `errors=ignore`)→ `split("\n")` 切片 `[s-1:e]``truncate_text_output` 按字节截断保行 |
使用示例:
@ -66,6 +67,11 @@ reme4 version
reme4 reindex
reme4 search query="latency 问题" limit=10 min_score=0.2 vector_weight=0.6
# 读取 working_dir 下的 markdown完整相对路径无后缀自动补 .md可按行切片或限制字节
reme4 read path=Templates/Recipe.md
reme4 read path=Notes start_line=1 end_line=20
reme4 read path=Big.md max_bytes=4096
# 通过 MCP backend 调用
reme4 search query="..." backend=mcp
```

View file

@ -49,5 +49,6 @@ class BaseJob(BaseComponent):
await step(context)
except Exception as e:
self.logger.exception(f"Failed to execute job: {e}")
context.response.success = False
context.response.answer = str(e)
return context.response

View file

@ -96,6 +96,26 @@ jobs:
steps:
- backend: search_step
- backend: base
name: read
description: "read a markdown file (relative path under working_dir)"
parameters:
type: object
properties:
path:
type: string
description: "relative path under the working_dir (no absolute paths); markdown only"
start_line:
type: integer
description: "Optional, first line to read (1-based, inclusive)"
end_line:
type: integer
description: "Optional, last line to read (1-based, inclusive)"
required:
- path
steps:
- backend: read_step
- backend: stream
name: stream_demo
description: "stream demo job: repeat query 10x and stream char-by-char"

View file

@ -5,3 +5,8 @@ REME_SERVICE_INFO = "REME_SERVICE_INFO"
REME_DEFAULT_HOST = "127.0.0.1"
REME_DEFAULT_PORT = 2333
# CRUD steps: file IO limits and truncation marker (shared across CRUD steps).
DEFAULT_MAX_BYTES = 50 * 1024
MAX_FILE_READ_BYTES = 200 * 1024 * 1024
TRUNCATION_NOTICE_MARKER = "<<TRUNCATION_NOTICE>>"

View file

@ -1,9 +1,11 @@
"""steps"""
from . import common
from . import crud
from .base_step import BaseStep
__all__ = [
"common",
"crud",
"BaseStep",
]

View file

@ -2,6 +2,7 @@
import copy
from abc import abstractmethod, ABC
from pathlib import Path
from typing import TypeVar, TYPE_CHECKING
from agentscope.formatter import FormatterBase
@ -83,7 +84,20 @@ class BaseStep(ABC):
self.context.apply_mapping(self.output_mapping)
return result
def _resolve(self, key: str, base_cls: type[T], comp_enum: ComponentEnum, attr: str | None = None) -> T:
@property
def working_path(self) -> Path:
"""Resolved working directory from app context or cwd."""
if self.app_context is None:
return Path.cwd()
return Path(self.app_context.app_config.working_dir)
def _resolve(
self,
key: str,
base_cls: type[T],
comp_enum: ComponentEnum,
attr: str | None = None,
) -> T:
"""Return a kwargs-supplied instance, or look one up by name in the app registry."""
# 1. Step init kwargs, 2. Runtime context (run_job kwargs), 3. App registry by name.
for source in (self.kwargs, self.context or {}):

View file

@ -0,0 +1,7 @@
"""CRUD steps for markdown files under the working_dir."""
from .read import ReadStep
__all__ = [
"ReadStep",
]

View file

@ -0,0 +1,109 @@
"""Shared filesystem helpers for CRUD steps (path gating, safe read, truncation)."""
from pathlib import Path
import aiofiles
import aiofiles.os
from ...constants import DEFAULT_MAX_BYTES, MAX_FILE_READ_BYTES, TRUNCATION_NOTICE_MARKER
from ...utils import get_logger
logger = get_logger()
def resolve_path(working_path: Path, raw: str) -> tuple[Path | None, str | None]:
"""Resolve a relative `path=` argument under self.working_path.
Rules:
- the caller supplies the full relative path under ``self.working_path``;
absolute paths are rejected.
Returns ``(abs_path, None)`` on success, or ``(None, error_message)`` on failure.
Filetype-specific gating (e.g. markdown-only / suffix auto-append) is
layered on top by callers see ``reme4/steps/crud/_file_io.py::gate_md``.
"""
if not raw or not str(raw).strip():
return None, "`path` is required"
s = str(raw).strip()
p = Path(s)
if p.is_absolute():
logger.info("absolute path detected, recommmending relative paths")
return p, None
return working_path / p, None
def gate_md(target: Path, raw: str) -> tuple[Path | None, str | None]:
"""Markdown-only gate: auto-append `.md` when no suffix; reject any non-`.md` suffix.
Layered on top of ``BaseStep.resolve_path`` to keep filetype-specific rules
out of the generic path resolver.
"""
if target.suffix == "":
return target.with_suffix(".md"), None
if target.suffix.lower() != ".md":
return None, (f"path {raw!r} is not a markdown file; this command only supports .md files")
return target, None
async def read_file_safe(file_path, max_bytes: int = MAX_FILE_READ_BYTES) -> str:
"""Read file with utf-8-sig (BOM-tolerant), fallback to errors='ignore'."""
stat = await aiofiles.os.stat(str(file_path))
read_size = min(stat.st_size, max_bytes)
try:
async with aiofiles.open(str(file_path), "r", encoding="utf-8-sig") as f:
return await f.read(read_size)
except UnicodeDecodeError:
async with aiofiles.open(
str(file_path),
"r",
encoding="utf-8-sig",
errors="ignore",
) as f:
return await f.read(read_size)
def truncate_text_output(
text: str,
*,
start_line: int = 1,
total_lines: int = 0,
max_bytes: int = DEFAULT_MAX_BYTES,
file_path: str | None = None,
encoding: str = "utf-8",
) -> str:
"""Truncate text by bytes preserving line integrity; append a continuation notice.
See qwenpaw `tools/utils.py` for the same semantics. Returns text unchanged when
it fits within max_bytes, when max_bytes <= 0, or when the last line itself
exceeds max_bytes (unhandled edge case).
"""
if not text or max_bytes <= 0:
return text
try:
text_bytes = text.encode(encoding)
if len(text_bytes) <= max_bytes:
return text
truncated = text_bytes[:max_bytes]
result = truncated.decode(encoding, errors="ignore")
newline_count = result.count("\n")
next_line = start_line + max(1, newline_count)
if next_line <= total_lines:
read_from = next_line
elif start_line < total_lines:
read_from = total_lines
else:
return result
notice = (
TRUNCATION_NOTICE_MARKER + f"\nThe output above was truncated."
f"\nThe full content is saved to the file and contains {total_lines} lines in total."
f"\nThis excerpt starts at line {start_line} and covers the next {max_bytes} bytes."
f"\nIf the current content is not enough, call `read` with file={file_path or ''} "
f"start_line={read_from} to read more."
)
return result + notice
except Exception:
logger.warning("truncate_text_output failed, returning original text", exc_info=True)
return text

91
reme4/steps/crud/read.py Normal file
View file

@ -0,0 +1,91 @@
"""Read a markdown file from the vault, with line-range slicing and byte-truncation."""
from ._file_io import (
gate_md,
resolve_path,
read_file_safe,
truncate_text_output,
)
from ..base_step import BaseStep
from ...components import R
@R.register("read_step")
class ReadStep(BaseStep):
"""Read a markdown file. Optional `start_line`/`end_line` for ranged reads."""
def _fail(self, message: str, **meta) -> None:
assert self.context is not None
self.context.response.success = False
self.context.response.answer = f"Error: {message}"
if meta:
self.context.response.metadata.update(meta)
async def execute(self): # pylint: disable=too-many-return-statements
assert self.context is not None
raw = str(self.context.get("path") or "")
start_line = self.context.get("start_line")
end_line = self.context.get("end_line")
target, err = resolve_path(self.working_path, raw)
if err:
self._fail(err)
return None
target, err = gate_md(target, raw)
if err:
self._fail(err)
return None
for label, value in (("start_line", start_line), ("end_line", end_line)):
if value is None:
continue
try:
int(value)
except (TypeError, ValueError):
self._fail(f"{label} must be an integer, got {value!r}")
return None
if not target.exists():
self._fail(f"file {target} does not exist", path=str(target))
return None
if not target.is_file():
self._fail(f"path {target} is not a file", path=str(target))
return None
try:
content = await read_file_safe(target)
except Exception as e:
self._fail(f"read failed: {e}", path=str(target))
return None
all_lines = content.split("\n")
total = len(all_lines)
s = max(1, int(start_line) if start_line is not None else 1)
e = min(total, int(end_line) if end_line is not None else total)
if s > total:
self._fail(
f"start_line {s} exceeds file length ({total} lines)",
path=str(target),
total_lines=total,
)
return None
if s > e:
self._fail(f"start_line ({s}) > end_line ({e})", path=str(target))
return None
selected = "\n".join(all_lines[s - 1 : e])
text = truncate_text_output(
selected,
start_line=s,
total_lines=total,
file_path=str(target),
)
self.context.response.success = True
self.context.response.answer = text
self.logger.info(
f"[{self.name}] read path={target} lines={s}-{e}/{total} bytes={len(text.encode('utf-8'))}",
)
return self.context.response

View file

@ -0,0 +1,371 @@
"""End-to-end tests for reme4 crud_md steps: spawn `reme4 start`, drive via HTTP,
verify responses, then shut down. Each test uses an isolated cwd so the working_dir
(.reme by default) does not collide.
CLI rule: `path=` is relative-only, rooted at the reme working_dir. A bare path with
no suffix auto-appends `.md`; non-`.md` suffix is rejected. Absolute paths are
rejected.
"""
import asyncio
import os
import tempfile
import warnings
from pathlib import Path
from reme4.utils import call_action, call_and_check, mock_reme_server
warnings.filterwarnings("ignore", category=DeprecationWarning, module="jieba")
warnings.filterwarnings("ignore", category=DeprecationWarning, module="pkg_resources")
class _temp_chdir:
"""chdir to path for the duration of the block; restore on exit."""
def __init__(self, path):
self.path = path
self._old = None
def __enter__(self):
self._old = os.getcwd()
os.chdir(self.path)
return self
def __exit__(self, *exc):
os.chdir(self._old)
def _run(coro):
"""Run an async coroutine on a fresh isolated event loop."""
asyncio.run(coro)
def _seed_md(working_dir: Path, rel: str, body: str) -> Path:
target = working_dir / rel
target.parent.mkdir(parents=True, exist_ok=True)
target.write_text(body, encoding="utf-8")
return target
# ---------------------------------------------------------------------------
# Individual job tests
# ---------------------------------------------------------------------------
def test_read_relative_path():
"""`reme4 read path=Templates/Recipe.md` returns the file body from .reme/."""
async def run():
with tempfile.TemporaryDirectory() as tmp, _temp_chdir(tmp):
working = Path(tmp) / ".reme"
working.mkdir(parents=True, exist_ok=True)
body = "# Recipe\n\nMix flour and water.\n"
_seed_md(working, "Templates/Recipe.md", body)
async with mock_reme_server() as (host, port):
await call_and_check(
"read",
host=host,
port=port,
path="Templates/Recipe.md",
validator=lambda r: (
isinstance(r, dict)
and r.get("success") is True
and "# Recipe" in str(r.get("answer", ""))
and "flour and water" in str(r.get("answer", ""))
),
)
print("✓ test_read_relative_path passed")
_run(run())
def test_read_no_suffix_autoappends_md():
"""A bare path with no suffix auto-appends `.md`."""
async def run():
with tempfile.TemporaryDirectory() as tmp, _temp_chdir(tmp):
working = Path(tmp) / ".reme"
working.mkdir(parents=True, exist_ok=True)
_seed_md(working, "Templates/Recipe.md", "auto-md\n")
async with mock_reme_server() as (host, port):
await call_and_check(
"read",
host=host,
port=port,
path="Templates/Recipe",
validator=lambda r: (
isinstance(r, dict) and r.get("success") is True and "auto-md" in str(r.get("answer", ""))
),
)
print("✓ test_read_no_suffix_autoappends_md passed")
_run(run())
def test_read_line_range():
"""start_line / end_line slice the file 1-based, inclusive."""
async def run():
with tempfile.TemporaryDirectory() as tmp, _temp_chdir(tmp):
working = Path(tmp) / ".reme"
working.mkdir(parents=True, exist_ok=True)
_seed_md(working, "Notes.md", "L1\nL2\nL3\nL4\nL5\n")
async with mock_reme_server() as (host, port):
await call_and_check(
"read",
host=host,
port=port,
path="Notes.md",
start_line=2,
end_line=4,
validator=lambda r: (
isinstance(r, dict)
and r.get("success") is True
and "L2" in str(r["answer"])
and "L3" in str(r["answer"])
and "L4" in str(r["answer"])
and "L1" not in str(r["answer"])
and "L5" not in str(r["answer"])
),
)
print("✓ test_read_line_range passed")
_run(run())
def test_read_absolute_path_rejected():
"""Absolute paths are rejected (relative-only after refactor)."""
async def run():
with tempfile.TemporaryDirectory() as tmp, _temp_chdir(tmp):
working = Path(tmp) / ".reme"
working.mkdir(parents=True, exist_ok=True)
target = _seed_md(working, "Abs.md", "x\n")
async with mock_reme_server() as (host, port):
result = await call_action(
"read",
host=host,
port=port,
path=str(target.resolve()),
)
if not (
isinstance(result, dict)
and result.get("success") is False
and "absolute" in str(result.get("answer", "")).lower()
):
raise AssertionError(f"expected absolute-path rejection, got {result!r}")
print("✓ test_read_absolute_path_rejected passed")
_run(run())
def test_read_non_md_rejected():
"""Paths whose suffix is not `.md` are rejected."""
async def run():
with tempfile.TemporaryDirectory() as tmp, _temp_chdir(tmp):
working = Path(tmp) / ".reme"
working.mkdir(parents=True, exist_ok=True)
async with mock_reme_server() as (host, port):
result = await call_action(
"read",
host=host,
port=port,
path="data/foo.txt",
)
if not (
isinstance(result, dict)
and result.get("success") is False
and "markdown" in str(result.get("answer", "")).lower()
):
raise AssertionError(f"expected markdown-only rejection, got {result!r}")
print("✓ test_read_non_md_rejected passed")
_run(run())
def test_read_missing_file():
"""Reading a non-existent file should fail with a clear error."""
async def run():
with tempfile.TemporaryDirectory() as tmp, _temp_chdir(tmp):
working = Path(tmp) / ".reme"
working.mkdir(parents=True, exist_ok=True)
async with mock_reme_server() as (host, port):
result = await call_action(
"read",
host=host,
port=port,
path="NotThere.md",
)
if not (
isinstance(result, dict)
and result.get("success") is False
and "does not exist" in str(result.get("answer", "")).lower()
):
raise AssertionError(f"expected missing-file rejection, got {result!r}")
print("✓ test_read_missing_file passed")
_run(run())
def test_read_start_after_end():
"""start_line > end_line is invalid."""
async def run():
with tempfile.TemporaryDirectory() as tmp, _temp_chdir(tmp):
working = Path(tmp) / ".reme"
working.mkdir(parents=True, exist_ok=True)
_seed_md(working, "Range.md", "a\nb\nc\n")
async with mock_reme_server() as (host, port):
result = await call_action(
"read",
host=host,
port=port,
path="Range.md",
start_line=3,
end_line=1,
)
if not (
isinstance(result, dict)
and result.get("success") is False
and "start_line" in str(result.get("answer", ""))
):
raise AssertionError(f"expected start>end rejection, got {result!r}")
print("✓ test_read_start_after_end passed")
_run(run())
def test_read_start_line_exceeds_total():
"""start_line beyond total line count is invalid."""
async def run():
with tempfile.TemporaryDirectory() as tmp, _temp_chdir(tmp):
working = Path(tmp) / ".reme"
working.mkdir(parents=True, exist_ok=True)
_seed_md(working, "Short.md", "only-one-line\n")
async with mock_reme_server() as (host, port):
result = await call_action(
"read",
host=host,
port=port,
path="Short.md",
start_line=99,
)
if not (
isinstance(result, dict)
and result.get("success") is False
and "exceeds" in str(result.get("answer", "")).lower()
):
raise AssertionError(f"expected exceeds-length rejection, got {result!r}")
print("✓ test_read_start_line_exceeds_total passed")
_run(run())
def test_read_truncation():
"""A small max_bytes triggers truncation with a continuation notice."""
async def run():
with tempfile.TemporaryDirectory() as tmp, _temp_chdir(tmp):
working = Path(tmp) / ".reme"
working.mkdir(parents=True, exist_ok=True)
body = "\n".join(f"line {i}" for i in range(200)) + "\n"
_seed_md(working, "Big.md", body)
async with mock_reme_server() as (host, port):
await call_and_check(
"read",
host=host,
port=port,
path="Big.md",
max_bytes=64,
validator=lambda r: (
isinstance(r, dict)
and r.get("success") is True
and "truncated" in str(r["answer"])
and "start_line=" in str(r["answer"])
),
)
print("✓ test_read_truncation passed")
_run(run())
def test_read_empty_path_rejected():
"""An empty `path` should be rejected."""
async def run():
with tempfile.TemporaryDirectory() as tmp, _temp_chdir(tmp):
working = Path(tmp) / ".reme"
working.mkdir(parents=True, exist_ok=True)
async with mock_reme_server() as (host, port):
result = await call_action("read", host=host, port=port, path="")
if not (
isinstance(result, dict)
and result.get("success") is False
and "required" in str(result.get("answer", "")).lower()
):
raise AssertionError(f"expected `path` required rejection, got {result!r}")
print("✓ test_read_empty_path_rejected passed")
_run(run())
# ---------------------------------------------------------------------------
# Aggregate test: reuse one server instance for all read cases (faster).
# ---------------------------------------------------------------------------
def test_all_read_cases_one_server():
"""Run multiple read scenarios against a single shared server for efficiency."""
async def run():
with tempfile.TemporaryDirectory() as tmp, _temp_chdir(tmp):
working = Path(tmp) / ".reme"
working.mkdir(parents=True, exist_ok=True)
_seed_md(working, "Templates/Recipe.md", "# Recipe\nbody\n")
_seed_md(working, "Notes.md", "L1\nL2\nL3\n")
async with mock_reme_server() as (host, port):
await call_and_check(
"read",
host=host,
port=port,
path="Templates/Recipe.md",
validator=lambda r: r.get("success") is True and "# Recipe" in r["answer"],
)
await call_and_check(
"read",
host=host,
port=port,
path="Notes",
validator=lambda r: r.get("success") is True and "L1" in r["answer"],
)
await call_and_check(
"read",
host=host,
port=port,
path="Notes.md",
start_line=2,
end_line=2,
validator=lambda r: r.get("success") is True and r["answer"].strip() == "L2",
)
print("✓ test_all_read_cases_one_server passed")
_run(run())
if __name__ == "__main__":
print("\n=== reme4 crud_md (read) E2E tests ===")
test_read_relative_path()
test_read_no_suffix_autoappends_md()
test_read_line_range()
test_read_absolute_path_rejected()
test_read_non_md_rejected()
test_read_missing_file()
test_read_start_after_end()
test_read_start_line_exceeds_total()
test_read_truncation()
test_read_empty_path_rejected()
test_all_read_cases_one_server()
print("\n所有测试通过!")