From 67936d5a4333224a134f475a9b3bc5764ef8bc3d Mon Sep 17 00:00:00 2001 From: xyf2020 <75460675+xyf2020@users.noreply.github.com> Date: Wed, 30 Sep 2026 12:19:32 +0800 Subject: [PATCH] refactor(plugins): move BaseAgenticAnswerStep out of the core benchmark steps package into the lme/beam plugins (#576) * refactor(plugins): move BaseAgenticAnswerStep into lme/beam plugins Move the shared benchmark answer base class from the core reme.steps.benchmark package into each plugin's own src tree (reme_lme.base_agentic_answer / reme_beam.base_agentic_answer) with absolute imports, drop the core benchmark steps package, and update plugin READMEs accordingly. * refactor(plugins): simplify benchmark answer steps * docs(plugins): clarify benchmark answer step migration --------- Co-authored-by: jinli.yl --- AGENTS.md | 2 +- docs/en/framework.md | 1 - docs/zh/framework.md | 1 - plugins/beam/README.md | 6 +- plugins/beam/README_ZH.md | 9 +- plugins/beam/src/reme_beam/agentic_answer.py | 105 +++++++++++++-- plugins/lme/README.md | 6 +- plugins/lme/README_ZH.md | 9 +- plugins/lme/src/reme_lme/agentic_answer.py | 107 ++++++++++++++-- reme/steps/__init__.py | 3 +- reme/steps/benchmark/__init__.py | 5 - reme/steps/benchmark/base_agentic_answer.py | 128 ------------------- 12 files changed, 213 insertions(+), 169 deletions(-) delete mode 100644 reme/steps/benchmark/__init__.py delete mode 100644 reme/steps/benchmark/base_agentic_answer.py diff --git a/AGENTS.md b/AGENTS.md index a22b05b0..5329001e 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -47,7 +47,7 @@ and concise documentation together. - `reme/components/`: agent wrappers, model adapters, stores, catalogs, graphs, indexes, clients, tokenizers, and outbound proxies. - `reme/steps/`: registered job steps grouped by common, file I/O, index, evolve, cookbook, and transfer - concerns, plus shared benchmark base classes under `benchmark/`. + concerns. - `reme/utils/`: shared utilities, including service discovery, logging, web-static resolution, session I/O, token accounting, and wikilink handling. - `tests/unit/`: primary fast, isolated validation suite. diff --git a/docs/en/framework.md b/docs/en/framework.md index e2ec2b74..ab2b286e 100644 --- a/docs/en/framework.md +++ b/docs/en/framework.md @@ -77,7 +77,6 @@ reme/ steps/ base_step.py # BaseStep, Ref, dispatch_steps common/ # version, help, health_check, status, chat - benchmark/ # LongMemEval / BEAM evaluation steps file_io/ # read/write/edit/delete/move/frontmatter/daily index/ # watch/init/update/search/traverse evolve/ # auto_memory, auto_resource, auto_dream, proactive diff --git a/docs/zh/framework.md b/docs/zh/framework.md index d5a5adc7..2457806a 100644 --- a/docs/zh/framework.md +++ b/docs/zh/framework.md @@ -72,7 +72,6 @@ reme/ steps/ base_step.py # BaseStep、Ref、dispatch_steps common/ # version、help、health_check、status、chat - benchmark/ # LongMemEval / BEAM 评测步骤 file_io/ # read/write/edit/delete/move/frontmatter/daily index/ # watch/init/update/search/traverse evolve/ # auto_memory、auto_resource、auto_dream、proactive diff --git a/plugins/beam/README.md b/plugins/beam/README.md index 341cd79a..f17ca532 100644 --- a/plugins/beam/README.md +++ b/plugins/beam/README.md @@ -31,10 +31,12 @@ The existing `auto_memory`, `agentic_answer`, `answer_judge`, `bench` and `judge names and model environment variables are unchanged. Explicit application/CLI overrides still take precedence. Installing this plugin does not start an evaluation. -The shared answer base class lives in `reme.steps.benchmark.base_agentic_answer`. -The old core-owned `reme.steps.benchmark.beam` Python import path is removed. Custom Python callers should import memory, search and answer Steps from `reme_beam`, and install `beam-judge` before importing the judge Step from `judge_beam`. After uninstalling, Applications and CLI services must omit the plugin until it is installed again. +`BeamAgenticAnswerStep` now implements the answer behavior directly. The former +`reme.steps.benchmark.BaseAgenticAnswerStep` import is gone; custom subclasses can +extend `reme_beam.BeamAgenticAnswerStep` for BEAM behavior, or implement their own +Step using `reme.steps.base_step.BaseStep`. Uninstallation never removes datasets, workspaces or results. Restart an existing service after changing plugins. diff --git a/plugins/beam/README_ZH.md b/plugins/beam/README_ZH.md index 86558936..0e542ea6 100644 --- a/plugins/beam/README_ZH.md +++ b/plugins/beam/README_ZH.md @@ -28,8 +28,11 @@ editable 安装会注册 `beam` entry point，并让源码修改立即生效。r 均保持关闭。原有 `auto_memory`、`agentic_answer`、`answer_judge`、`bench`、`judge` 名称及模型环境变量 保持不变，显式应用参数和 CLI 覆盖仍优先。安装或启用插件不会自动开始评测。 -共享回答基类位于 `reme.steps.benchmark.base_agentic_answer`。 -原 `reme.steps.benchmark.beam` Python 导入路径已移除。自定义 Python 调用应从 `reme_beam` -导入记忆、搜索和回答 Step；安装 `beam-judge` 后再从 `judge_beam` 导入评判 Step。 +自定义 Python 调用应从 `reme_beam` 导入记忆、搜索和回答 Step；安装 `beam-judge` 后再从 +`judge_beam` 导入评判 Step。 +`BeamAgenticAnswerStep` 现在直接实现回答逻辑。原来的 +`reme.steps.benchmark.BaseAgenticAnswerStep` 导入路径已移除；自定义子类可继承 +`reme_beam.BeamAgenticAnswerStep` 以复用 BEAM 回答行为，或基于 +`reme.steps.base_step.BaseStep` 自行实现 Step。 卸载插件后，Application 和 CLI 服务必须移除插件选择，直到再次安装。 卸载不会删除数据集、工作区或结果。修改插件后需重启已有服务。 diff --git a/plugins/beam/src/reme_beam/agentic_answer.py b/plugins/beam/src/reme_beam/agentic_answer.py index d6452ec0..e852be3a 100644 --- a/plugins/beam/src/reme_beam/agentic_answer.py +++ b/plugins/beam/src/reme_beam/agentic_answer.py @@ -1,14 +1,101 @@ -"""BEAM agentic answer step – ReAct agent that answers questions using the search tool.""" +"""BEAM agentic-answer step.""" -from reme.steps.benchmark import BaseAgenticAnswerStep +import os + +from reme.enumeration import ChunkEnum +from reme.steps.base_step import BaseStep +from reme.steps.index._dedup import _ToolContextDedupMixin +from reme.utils.counter import global_counter_inc -class BeamAgenticAnswerStep(BaseAgenticAnswerStep): - """Answer a BEAM probing question via ReAct agent with access to the search tool. +class BeamAgenticAnswerStep(BaseStep): + """Answer a BEAM probing question with the ReAct agent and workspace tools.""" - The agent uses the ``agent_wrapper`` component in ReAct mode, calling the - ``search`` job tool to retrieve relevant memory chunks before generating - a final answer. - """ + # Reasoning-round budget. The AgentScope wrapper converts it to the + # backend's iteration-counting semantics; other backends ignore it. + MAX_ITERATION = 10 + TOOL_CONTEXT_PREFIX: str = "beam_agentic_answer" + JOB_TOOLS: list[str] = ["search", "add_draft", "read_all_draft", "read"] + INJECTED_JOB_KWARGS: dict = {"read_step_format_session": True} - TOOL_CONTEXT_PREFIX = "beam_agentic_answer" + def _injected_job_kwargs(self, query: str) -> dict: + """Enable query-aware session compression when requested.""" + injected = dict(self.INJECTED_JOB_KWARGS) + if self.context is not None and self.context.get("compress_session"): + injected["_search"] = { + "_compress": {"session": "true"}, + "queries": [query], + "type": "query-aware", + } + return injected + + async def execute(self): + """Answer the current query and store the result in the response.""" + assert self.context is not None + query: str = self.context.get("query", "") + query_time: str | None = self.context.get("query_time") + + if not query: + self.context.response.success = False + self.context.response.answer = "Skipped: empty query" + return self.context.response + + # Build system prompt with optional temporal context + sys_prompt = self.get_prompt("system_prompt") + if query_time: + sys_prompt += "\n" + self.prompt_format("temporal_hint", query_time=query_time) + + if self.app_context is not None: + tool_context_id = ( + f"{self.TOOL_CONTEXT_PREFIX}_{os.getpid()}_" + f"{global_counter_inc(self.app_context.metadata, [self.TOOL_CONTEXT_PREFIX])}" + ) + else: + tool_context_id = f"{self.TOOL_CONTEXT_PREFIX}_{os.getpid()}_local" + wrapper_kwargs = { + "system_prompt": sys_prompt, + "job_tools": list(self.JOB_TOOLS), + "react_config": {"max_iters": self.MAX_ITERATION}, + "tool_context_id": tool_context_id, + } + if injected_job_kwargs := self._injected_job_kwargs(f"{query}(query time: {query_time})"): + wrapper_kwargs["injected_job_kwargs"] = injected_job_kwargs + + if self.context.stream: + text = await self._stream_reply(query, **wrapper_kwargs) + else: + result = await self.agent_wrapper.reply(query, **wrapper_kwargs) + text = (result.get("result") or "").strip() + + self.logger.debug(f"[{self.name}] response: {text!r}") + + self.context.response.success = True + self.context.response.answer = text + self.context.response.metadata.update( + { + "query": query, + "query_time": query_time, + "sys_prompt": sys_prompt, + "response": text, + }, + ) + + if self.app_context is not None: + self.app_context.metadata.get(_ToolContextDedupMixin.TOOL_CONTEXTS_KEY, {}).pop(tool_context_id, None) + return self.context.response + + async def _stream_reply(self, query: str, **wrapper_kwargs) -> str: + """Stream unified chunks to the context stream queue.""" + assert self.context is not None + text_parts: list[str] = [] + + async for chunk in self.agent_wrapper.reply_stream(query, **wrapper_kwargs): + await self.context.add_stream_string(chunk.chunk, chunk.chunk_type) + + if chunk.chunk_type == ChunkEnum.CONTENT and isinstance(chunk.chunk, str): + text_parts.append(chunk.chunk) + + if chunk.session_id: + self.context.response.metadata["session_id"] = chunk.session_id + + return "".join(text_parts).strip() diff --git a/plugins/lme/README.md b/plugins/lme/README.md index a2310caf..ff5b88e8 100644 --- a/plugins/lme/README.md +++ b/plugins/lme/README.md @@ -31,10 +31,12 @@ The existing `auto_memory`, `agentic_answer`, `answer_judge`, `bench` and `judge names and model environment variables are unchanged. Explicit application/CLI overrides still take precedence. Installing this plugin does not start an evaluation. -The shared answer base class lives in `reme.steps.benchmark.base_agentic_answer`. -The old core-owned `reme.steps.benchmark.lme` Python import path is removed. Custom Python callers should import memory, search and answer Steps from `reme_lme`, and install `lme-judge` before importing the judge Step from `judge_lme`. After uninstalling, Applications and CLI services must omit the plugin until it is installed again. +`LmeAgenticAnswerStep` now implements the answer behavior directly. The former +`reme.steps.benchmark.BaseAgenticAnswerStep` import is gone; custom subclasses can +extend `reme_lme.LmeAgenticAnswerStep` for LongMemEval behavior, or implement their +own Step using `reme.steps.base_step.BaseStep`. Uninstallation never removes datasets, workspaces or results. Restart an existing service after changing plugins. diff --git a/plugins/lme/README_ZH.md b/plugins/lme/README_ZH.md index 9d0bbbb0..13ccb8a1 100644 --- a/plugins/lme/README_ZH.md +++ b/plugins/lme/README_ZH.md @@ -28,8 +28,11 @@ editable 安装会注册 `lme` entry point，并让源码修改立即生效。ru 均保持关闭。原有 `auto_memory`、`agentic_answer`、`answer_judge`、`bench`、`judge` 名称及模型环境变量 保持不变，显式应用参数和 CLI 覆盖仍优先。安装或启用插件不会自动开始评测。 -共享回答基类位于 `reme.steps.benchmark.base_agentic_answer`。 -原 `reme.steps.benchmark.lme` Python 导入路径已移除。自定义 Python 调用应从 `reme_lme` -导入记忆、搜索和回答 Step；安装 `lme-judge` 后再从 `judge_lme` 导入评判 Step。 +自定义 Python 调用应从 `reme_lme` 导入记忆、搜索和回答 Step；安装 `lme-judge` 后再从 +`judge_lme` 导入评判 Step。 +`LmeAgenticAnswerStep` 现在直接实现回答逻辑。原来的 +`reme.steps.benchmark.BaseAgenticAnswerStep` 导入路径已移除；自定义子类可继承 +`reme_lme.LmeAgenticAnswerStep` 以复用 LongMemEval 回答行为，或基于 +`reme.steps.base_step.BaseStep` 自行实现 Step。 卸载插件后，Application 和 CLI 服务必须移除插件选择，直到再次安装。 卸载不会删除数据集、工作区或结果。修改插件后需重启已有服务。 diff --git a/plugins/lme/src/reme_lme/agentic_answer.py b/plugins/lme/src/reme_lme/agentic_answer.py index e9017e5c..ffa36fd3 100644 --- a/plugins/lme/src/reme_lme/agentic_answer.py +++ b/plugins/lme/src/reme_lme/agentic_answer.py @@ -1,18 +1,101 @@ -"""LongMemEval agentic answer step – ReAct agent that answers questions using the search tool.""" +"""LongMemEval agentic-answer step.""" -from reme.steps.benchmark import BaseAgenticAnswerStep +import os + +from reme.enumeration import ChunkEnum +from reme.steps.base_step import BaseStep +from reme.steps.index._dedup import _ToolContextDedupMixin +from reme.utils.counter import global_counter_inc -class LmeAgenticAnswerStep(BaseAgenticAnswerStep): - """Answer a LongMemEval query via ReAct agent with access to the search tool. +class LmeAgenticAnswerStep(BaseStep): + """Answer a LongMemEval query with the ReAct agent and workspace tools.""" - The agent uses the ``agent_wrapper`` component in ReAct mode, calling the - ``search`` job tool to retrieve relevant memory chunks before generating - a final answer. + # Reasoning-round budget. The AgentScope wrapper converts it to the + # backend's iteration-counting semantics; other backends ignore it. + MAX_ITERATION = 10 + TOOL_CONTEXT_PREFIX: str = "lme_agentic_answer" + JOB_TOOLS: list[str] = ["search", "add_draft", "read_all_draft", "read"] + INJECTED_JOB_KWARGS: dict = {"read_step_format_session": True} - Session-transcript compression in the plugin's search Step is controlled by the - ``compress_session`` flag in the runtime context (set by the benchmark - runner from ``evaluation.compress_session``); it is off by default. - """ + def _injected_job_kwargs(self, query: str) -> dict: + """Enable query-aware session compression when requested.""" + injected = dict(self.INJECTED_JOB_KWARGS) + if self.context is not None and self.context.get("compress_session"): + injected["_search"] = { + "_compress": {"session": "true"}, + "queries": [query], + "type": "query-aware", + } + return injected - TOOL_CONTEXT_PREFIX = "lme_agentic_answer" + async def execute(self): + """Answer the current query and store the result in the response.""" + assert self.context is not None + query: str = self.context.get("query", "") + query_time: str | None = self.context.get("query_time") + + if not query: + self.context.response.success = False + self.context.response.answer = "Skipped: empty query" + return self.context.response + + # Build system prompt with optional temporal context + sys_prompt = self.get_prompt("system_prompt") + if query_time: + sys_prompt += "\n" + self.prompt_format("temporal_hint", query_time=query_time) + + if self.app_context is not None: + tool_context_id = ( + f"{self.TOOL_CONTEXT_PREFIX}_{os.getpid()}_" + f"{global_counter_inc(self.app_context.metadata, [self.TOOL_CONTEXT_PREFIX])}" + ) + else: + tool_context_id = f"{self.TOOL_CONTEXT_PREFIX}_{os.getpid()}_local" + wrapper_kwargs = { + "system_prompt": sys_prompt, + "job_tools": list(self.JOB_TOOLS), + "react_config": {"max_iters": self.MAX_ITERATION}, + "tool_context_id": tool_context_id, + } + if injected_job_kwargs := self._injected_job_kwargs(f"{query}(query time: {query_time})"): + wrapper_kwargs["injected_job_kwargs"] = injected_job_kwargs + + if self.context.stream: + text = await self._stream_reply(query, **wrapper_kwargs) + else: + result = await self.agent_wrapper.reply(query, **wrapper_kwargs) + text = (result.get("result") or "").strip() + + self.logger.debug(f"[{self.name}] response: {text!r}") + + self.context.response.success = True + self.context.response.answer = text + self.context.response.metadata.update( + { + "query": query, + "query_time": query_time, + "sys_prompt": sys_prompt, + "response": text, + }, + ) + + if self.app_context is not None: + self.app_context.metadata.get(_ToolContextDedupMixin.TOOL_CONTEXTS_KEY, {}).pop(tool_context_id, None) + return self.context.response + + async def _stream_reply(self, query: str, **wrapper_kwargs) -> str: + """Stream unified chunks to the context stream queue.""" + assert self.context is not None + text_parts: list[str] = [] + + async for chunk in self.agent_wrapper.reply_stream(query, **wrapper_kwargs): + await self.context.add_stream_string(chunk.chunk, chunk.chunk_type) + + if chunk.chunk_type == ChunkEnum.CONTENT and isinstance(chunk.chunk, str): + text_parts.append(chunk.chunk) + + if chunk.session_id: + self.context.response.metadata["session_id"] = chunk.session_id + + return "".join(text_parts).strip() diff --git a/reme/steps/__init__.py b/reme/steps/__init__.py index 80d1d8c0..d9bda4dd 100644 --- a/reme/steps/__init__.py +++ b/reme/steps/__init__.py @@ -1,11 +1,10 @@ """steps""" -from . import benchmark, common, evolve, file_io, index, transfer +from . import common, evolve, file_io, index, transfer from .base_step import BaseStep __all__ = [ "BaseStep", - "benchmark", "common", "evolve", "file_io", diff --git a/reme/steps/benchmark/__init__.py b/reme/steps/benchmark/__init__.py deleted file mode 100644 index c0d50414..00000000 --- a/reme/steps/benchmark/__init__.py +++ /dev/null @@ -1,5 +0,0 @@ -"""Shared benchmark steps; concrete implementations live in plugins.""" - -from .base_agentic_answer import BaseAgenticAnswerStep - -__all__ = ["BaseAgenticAnswerStep"] diff --git a/reme/steps/benchmark/base_agentic_answer.py b/reme/steps/benchmark/base_agentic_answer.py deleted file mode 100644 index bcd5c8f6..00000000 --- a/reme/steps/benchmark/base_agentic_answer.py +++ /dev/null @@ -1,128 +0,0 @@ -"""Shared base class for benchmark agentic-answer steps.""" - -import os - -from ..base_step import BaseStep -from ..index._dedup import _ToolContextDedupMixin -from ...enumeration import ChunkEnum -from ...utils.counter import global_counter_inc - - -class BaseAgenticAnswerStep(BaseStep): - """ReAct-agent answer implementation shared by benchmark plugins. - - Subclasses only need to set: - TOOL_CONTEXT_PREFIX (str): prefix used to build the unique tool_context_id. - JOB_TOOLS (list[str]): tools exposed to the ReAct agent; override to customize. - INJECTED_JOB_KWARGS (dict): server-owned kwargs injected into every job - tool call via ``injected_job_kwargs``; override the attribute or the - ``_injected_job_kwargs`` hook to customize. - - Concrete subclasses are registered by the plugin manifest. - - Inputs (from RuntimeContext): - query (str, required): The question to answer. - query_time (str, optional): ISO timestamp representing the query time, - used to ground the agent's temporal context. - - Output (written to context.response.answer): - The agent's final answer text. - """ - - # Reasoning-round budget. The AgentScope wrapper converts it to the - # backend's iteration-counting semantics; other backends ignore it. - MAX_ITERATION = 10 - TOOL_CONTEXT_PREFIX: str = "content_agentic_answer" - JOB_TOOLS: list[str] = ["search", "add_draft", "read_all_draft", "read"] - INJECTED_JOB_KWARGS: dict = {"read_step_format_session": True} - - def _injected_job_kwargs(self, query: str) -> dict: # pylint: disable=unused-argument - """Server-owned kwargs injected into every job tool call. - - Overridable hook: subclasses can extend the static - ``INJECTED_JOB_KWARGS`` with per-request values derived from ``query``. - - When the runtime context carries a truthy ``compress_session`` flag, - session-transcript compression is enabled in the benchmark plugin's search Step by - injecting a ``_search._compress.session`` marker plus the current - ``query`` as the query-aware relevance filter. Default (falsy) leaves - session chunks uncompressed. - """ - injected = dict(self.INJECTED_JOB_KWARGS) - if self.context is not None and self.context.get("compress_session"): - injected["_search"] = { - "_compress": {"session": "true"}, - "queries": [query], - "type": "query-aware", - } - return injected - - async def execute(self): - assert self.context is not None - query: str = self.context.get("query", "") - query_time: str | None = self.context.get("query_time") - - if not query: - self.context.response.success = False - self.context.response.answer = "Skipped: empty query" - return self.context.response - - # Build system prompt with optional temporal context - sys_prompt = self.get_prompt("system_prompt") - if query_time: - sys_prompt += "\n" + self.prompt_format("temporal_hint", query_time=query_time) - - if self.app_context is not None: - tool_context_id = ( - f"{self.TOOL_CONTEXT_PREFIX}_{os.getpid()}_" - f"{global_counter_inc(self.app_context.metadata, [self.TOOL_CONTEXT_PREFIX])}" - ) - else: - tool_context_id = f"{self.TOOL_CONTEXT_PREFIX}_{os.getpid()}_local" - wrapper_kwargs = { - "system_prompt": sys_prompt, - "job_tools": list(self.JOB_TOOLS), - "react_config": {"max_iters": self.MAX_ITERATION}, - "tool_context_id": tool_context_id, - } - if injected_job_kwargs := self._injected_job_kwargs(f"{query}(query time: {query_time})"): - wrapper_kwargs["injected_job_kwargs"] = injected_job_kwargs - - if self.context.stream: - text = await self._stream_reply(query, **wrapper_kwargs) - else: - result = await self.agent_wrapper.reply(query, **wrapper_kwargs) - text = (result.get("result") or "").strip() - - self.logger.debug(f"[{self.name}] response: {text!r}") - - self.context.response.success = True - self.context.response.answer = text - self.context.response.metadata.update( - { - "query": query, - "query_time": query_time, - "sys_prompt": sys_prompt, - "response": text, - }, - ) - - if self.app_context is not None: - self.app_context.metadata.get(_ToolContextDedupMixin.TOOL_CONTEXTS_KEY, {}).pop(tool_context_id, None) - return self.context.response - - async def _stream_reply(self, query: str, **wrapper_kwargs) -> str: - """Stream unified chunks to the context stream queue.""" - assert self.context is not None - text_parts: list[str] = [] - - async for chunk in self.agent_wrapper.reply_stream(query, **wrapper_kwargs): - await self.context.add_stream_string(chunk.chunk, chunk.chunk_type) - - if chunk.chunk_type == ChunkEnum.CONTENT and isinstance(chunk.chunk, str): - text_parts.append(chunk.chunk) - - if chunk.session_id: - self.context.response.metadata["session_id"] = chunk.session_id - - return "".join(text_parts).strip()