From 3924f89bb4abd75e671b9942cfa10f3be931bc83 Mon Sep 17 00:00:00 2001 From: imrewce Date: Tue, 11 Aug 2026 16:37:54 +0800 Subject: [PATCH] feat(bench): adding eval adapter for proactiveness on Pi-Bench (#439) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(bench): adding eval adapter for proactiveness on Pi-Bench * Revise README for π-Bench evaluation suite Updated the README to reflect the new project name and description. * fix(bench): refining pi-bench scripts according to cr comments * fix(bench): restore agent builtin tools in prebuilt toolkit --- .gitignore | 4 + benchmark/pibench/.gitignore | 14 + benchmark/pibench/README.md | 327 ++++++ benchmark/pibench/README_ZH.md | 284 +++++ benchmark/pibench/bridge_reme.py | 1039 +++++++++++++++++ .../bench/evaluation/trace_history.yaml | 53 + benchmark/pibench/config/models/reme.yaml | 40 + benchmark/pibench/env.sh.example | 57 + benchmark/pibench/fix_trace_logs.py | 198 ++++ benchmark/pibench/resume.py | 332 ++++++ benchmark/pibench/run_all.sh | 119 ++ benchmark/pibench/run_persona.sh | 301 +++++ 12 files changed, 2768 insertions(+) create mode 100644 benchmark/pibench/.gitignore create mode 100644 benchmark/pibench/README.md create mode 100644 benchmark/pibench/README_ZH.md create mode 100755 benchmark/pibench/bridge_reme.py create mode 100644 benchmark/pibench/config/bench/evaluation/trace_history.yaml create mode 100644 benchmark/pibench/config/models/reme.yaml create mode 100644 benchmark/pibench/env.sh.example create mode 100755 benchmark/pibench/fix_trace_logs.py create mode 100755 benchmark/pibench/resume.py create mode 100755 benchmark/pibench/run_all.sh create mode 100755 benchmark/pibench/run_persona.sh diff --git a/.gitignore b/.gitignore index d9c98bac..b5f00ce2 100644 --- a/.gitignore +++ b/.gitignore @@ -56,6 +56,10 @@ docs/_build/ site/ evaluation/ +# The pi-Bench suite ships its own trace-history render config, which must +# stay in git even though it lives under an evaluation/ directory. +!benchmark/pibench/config/bench/evaluation/ +!benchmark/pibench/config/bench/evaluation/** datasets/ # Claude Code skills (local only) diff --git a/benchmark/pibench/.gitignore b/benchmark/pibench/.gitignore new file mode 100644 index 00000000..6af97fd4 --- /dev/null +++ b/benchmark/pibench/.gitignore @@ -0,0 +1,14 @@ +# 含真实 API key，绝不入库 +env.sh + +# 运行时产物（含对话内容，勿入库） +logs/ +outputs/ +reme_workspace/ +nanobot_workspace/ + +# 数据符号链接（指向外部 π-Bench 仓库） +data + +__pycache__/ +*.pyc diff --git a/benchmark/pibench/README.md b/benchmark/pibench/README.md new file mode 100644 index 00000000..c62249a0 --- /dev/null +++ b/benchmark/pibench/README.md @@ -0,0 +1,327 @@ +[中文版 / Chinese version](./README_ZH.md) + +# π-Bench Evaluation Suite + +A glue layer that connects the **ReMe agent (with persistent memory)** to +**π-Bench** (Proactive Personal Assistant Benchmark). This directory contains +only the minimal code and configuration needed for the integration: the +π-Bench framework (`src/`), evaluation data (`data/`), the AppWorld tool +environment, and ReMe itself are all **external third-party dependencies**, +referenced in place via symlink and environment variables and never bundled +with this suite. + +- π-Bench: https://github.com/Simplified-Reasoning/Pi-Bench (arXiv: 2605.14678) +- ReMe: the root of the ReMe repository this suite lives in (recommended + location: `ReMe/benchmark/pibench/`) + +## 1. Architecture + +``` +π-Bench runner (src.main --mode run) + │ user_agent (simulated-user LLM) walks data/{persona}/episode.yaml + │ task by task, chatting with the agent over multiple turns and judging + │ hidden intents (PROC) during the run phase + ▼ +test server (π-Bench scripts/test_server.py, HTTP long-polling) + ▲ /send │ /poll + │ ▼ +bridge_reme.py ──────────────► ReMe Application (embedded as a library) + │ ├─ agent_wrapper: agent under test (AgentScope) + │ ├─ jobs: search / auto_memory / daily_write + │ └─ workspace: reme_workspace/{persona}/ + │ (isolated persistent memory per persona) + └──── MCP ────► AppWorld MCP ────► AppWorld APIs (tool/app environment) + +π-Bench runner (src.main --mode eval) + judger (judge LLM) reads the traces and scores each checklist item (COMP) +``` + +Key points: +- The bridge runs on **ReMe's own venv python** and uses ReMe as a library + (`resolve_app_config` + `Application`); **no ReMe source modification** is + required. +- Every incoming user message automatically triggers a ReMe memory `search` + and injects the matched memories (tuning knobs in §8); on task end (reset) + the session is distilled into daily notes by `auto_memory`. +- Tool calls executed by the agent (AppWorld MCP + ReMe job tools) are + captured per turn into the trace as `tool_steps`, so π-Bench + `tools_evaluation_path` scripts can score tool behavior (§7). +- π-Bench's `data/`, `src/` and AppWorld are not part of this suite; install + π-Bench first (§3.1). + +## 2. Directory layout + +``` +pibench/ +├── README.md / README_ZH.md # this document (English / Chinese) +├── env.sh.example # environment template (copy to env.sh, fill TODOs) +├── bridge_reme.py # ReMe ↔ test server bridge (memory inject/save, +│ # profile injection, tool-trace capture) +├── run_persona.sh # full pipeline for ONE persona (5 services + run + eval) +├── run_all.sh # batch over 5 personas (fresh/resume, default parallel=2) +├── resume.py # checkpoint resume: completion detection + surgical +│ # cleanup of interrupted tasks' residual memory +├── fix_trace_logs.py # run outputs → ~/.nanobot/trace_logs conversion, +│ # merging tool sidecars into turn files (pre-eval) +├── .gitignore # excludes env.sh and all runtime artifacts +└── config/ + ├── models/reme.yaml # runner model config (model_id=reme) + └── bench/evaluation/trace_history.yaml # trace render policy (shipped with + # the suite; passed via --history-config-path) +``` + +Generated at runtime (all git-ignored): `data` (symlink), `logs/`, `outputs/`, +`reme_workspace/`, `nanobot_workspace/`. + +## 3. Prerequisites (third-party, install first) + +### 3.1 π-Bench repository (with AppWorld) + +```bash +git clone https://github.com/Simplified-Reasoning/Pi-Bench.git +cd +python3.11 -m venv .venv # scripts expect exactly this venv name +source .venv/bin/activate +pip install -e . # pibench runner (src.main) +bash scripts/setup_appworld.sh # install AppWorld and download its data (large) +``` + +Post-install sanity checks: +```bash +ls data/ # should contain researcher marketer pharmacist law_trainee Financier +.venv/bin/python -c "import src" && echo OK +.venv/bin/appworld --help >/dev/null && echo OK +``` + +### 3.2 ReMe repository + +```bash +cd # ReMe repository root (contains the reme/ package) +python3.11 -m venv .venv # scripts expect exactly this venv name +source .venv/bin/activate +pip install -e . # or ReMe's own install flow; `import reme` must work +``` + +Sanity check: `.venv/bin/python -c "import reme; print('ok')"` + +## 4. Install this suite (step by step) + +1. **Place the suite** (recommended inside the ReMe repo so `REME_DIR` is + inferred automatically): + ```bash + cp -r pibench /benchmark/pibench + cd /benchmark/pibench + ``` + If placed elsewhere, set `REME_DIR` explicitly in env.sh later. + +2. **Create the environment file and fill in the custom parameters**: + ```bash + cp env.sh.example env.sh + ``` + Open `env.sh`; required items (marked TODO): + | Variable | Description | + |---|---| + | `PI_BENCH_ROOT` | π-Bench repo root (contains `src/` `data/` `.venv` `third_party/appworld`) | + | `USER_API_KEY` | API key of the simulated-user LLM (run phase, hidden-intent judging) | + | `JUDGER_API_KEY` | API key of the judger LLM (eval phase, checklist scoring) | + | `BRAVE_SEARCH_API_KEY` | optional; for the agent's web_search tool, `dummy` when unused | + + Optional tuning: `REME_MODEL_NAME` (base model of the agent under test), + `REME_DIR`, `REME_LLM_BASE_URL` (default: DashScope OpenAI-compatible + endpoint). + +3. **Link the evaluation data** (referenced in place, never copied): + ```bash + ln -s "$PI_BENCH_ROOT/data" data + ``` + +4. **(Optional) adjust model config** `config/models/reme.yaml`: + - `user_agent.model` / `judger.model`: model names for the simulated user + and the judger (literal values; π-Bench only expands `${ENV}` in + base_url/api_key). + - `run.turn_timeout`, `max_tool_iterations`, etc. as needed. + +5. **Smoke check** (does not start the evaluation): + ```bash + bash -n run_all.sh && bash -n run_persona.sh + source env.sh && "$REME_DIR/.venv/bin/python" -c "import reme; print('reme ok')" + ``` + +## 5. Run the evaluation + +> ⚠️ For long runs use `screen`, **not nohup** (nohup loses the permission +> context in sandboxed/restricted environments and breaks child processes). + +```bash +# Full official run: wipe ALL personas' memory/outputs/traces first (default +# fresh mode, parallel=2) +mkdir -p logs # on a fresh deployment logs/ does not exist yet +screen -dmS pibench_suite bash -c "cd $(pwd) && bash run_all.sh > logs/run_all_master.log 2>&1" + +# Checkpoint continuation (after an interruption; no wipe, completed tasks skipped) +bash run_all.sh --resume + +# Other usages +bash run_all.sh --parallel 1 # sequential +bash run_all.sh --resume --skip-eval # run phase only +bash run_persona.sh researcher # single persona (default --resume semantics) +bash run_persona.sh researcher --fresh +``` + +Time reference: 5 personas × 20 tasks, parallel=2, fresh full run ≈ 12–14 hours. + +`run_all.sh` exits non-zero when any persona fails, so upstream automation +cannot mistake a partially failed suite run for a success. + +## 6. Port allocation (parallel personas never collide) + +| persona | AppWorld API | AppWorld MCP | Test Server | ReMe internal service | +|-------------|------|-------|------|-------| +| marketer | 9001 | 10001 | 9998 | 18766 | +| law_trainee | 9002 | 10002 | 9997 | 18767 | +| pharmacist | 9003 | 10003 | 9996 | 18768 | +| researcher | 9004 | 10004 | 9995 | 18765 | +| Financier | 9005 | 10005 | 9994 | 18769 | + +## 7. Outputs and scores + +- **Results**: `outputs/reme/{persona}/{task}/eval/results/*_result.json` + - `overall_average_score`: checklist completeness (COMP; the judger scores + each criterion YES/NO, weighted across dependency groups) + - `overall_proactiveness_average_score`: proactiveness (PROC; the + user_agent judges hidden-intent coverage during the run phase; each task + file also carries the global average) +- **Traces**: `~/.nanobot/trace_logs/reme/{persona}/{task}/...` (the scoring + input of the eval phase) +- **Logs**: `logs/` (`suite_.log` per persona; `bridge_*`, + `runner_run/eval_*`, `appworld_*`, `test_server_*` per service) +- **Memory store**: `reme_workspace/{persona}/` (daily/digest notes, raw + session dialogs, BM25 index, etc.; persistent across runs, wiped only in + fresh mode) + +Score summary: +```bash +grep -h "overall_average_score\|overall_proactiveness" \ + outputs/reme/*/*/eval/results/*_result.json | head +``` + +### Tool-trace capture (tools_evaluation support) + +Some tasks define `objectives.tools_evaluation_path`: Python scripts that +score tool behavior (e.g. "the temporary Todoist board was created and +removed"). They need the executed tool calls in the trace. The pipeline: + +1. During `reply()`, the bridge reads the persisted AgentScope session state + after each turn and extracts the new `tool_call` / `tool_result` blocks + (tool name, arguments, result). +2. Records are appended to + `outputs/reme/{persona}/{task}/history/{ts}-tools.jsonl`, tagged with the + turn number; AgentScope MCP names (`mcp__AppWorld__`) are normalized + to the π-Bench convention (`mcp_appworld_`). +3. `fix_trace_logs.py` pairs each `{ts}-messages.jsonl` run with the + temporally closest tools sidecar and merges the records into the generated + `turn_N.json` files under the `tool_steps` key — one of the two + tool-history formats understood by π-Bench's `collect_tool_history()`. +4. The eval phase then feeds `tool_steps` to both the tools_evaluation + scripts and the rendered `` seen by the judger. + +## 8. Memory mechanism (core design of this suite) + +- **Persona isolation**: each persona has its own workspace + (`reme_workspace/{persona}/`); the bridge takes an exclusive + `.bridge.lock` on it at startup, so two bridges can never share one memory + store, and one persona's memory search can never reach another's memories. +- **Writes**: on task end (runner sends reset), the session is distilled by + the `auto_memory` job into daily notes and indexed by the background + watcher (BM25). Saves are non-blocking background tasks; the first message + of a new session waits for in-flight writes before searching. +- **Reads**: on every incoming user message the bridge runs one `search` and + injects matched memories (`[Relevant memories from previous sessions]` + prefix); without matches the message passes through unchanged. Retrieval + tuning (bridge CLI flags, adjustable in run_persona.sh): + - `--search-limit 3`: at most 3 memory chunks injected per message; + - `--search-min-score 2.0`: weak BM25 hits are filtered out; + - `tool_context_id` rotates per task: chunks already injected within the + same task are not re-injected (ReMe's seen-chunk dedup, 24h TTL); normal + recall resumes after task boundaries. +- **No self-leakage**: the in-progress session is not in the store yet + (saves happen on reset), so a task can never retrieve its own unfinished + content. +- The agent also holds `search`/`daily_write` tools and can retrieve/record + proactively. +- **System prompt**: `bridge_reme.py:build_system_prompt()` embeds the + HIDDEN-NEEDS protocol (proactiveness-oriented) and injects the persona + profile from `data/{persona}/profile.yaml` into every turn's system prompt. + +## 9. Checkpoint resume and memory-cleanup semantics + +- **Completion detection** (resume.py): scans + `outputs/reme/{persona}/**/history/*-log.jsonl` and + `outputs/reme/{persona}/run/*-log.jsonl` for + `Task finished task_id=X status=Y`. The status with the **newest event + timestamp** wins per task (record `timestamp`, falling back to + `timestamp_iso`, then to the timestamp embedded in the log file name) — + file category and read order alone can never override a newer record, so an + old run-level SUCCESS cannot mask a newer per-task ERROR. `SUCCESS / + MAX_TURNS / TIMEOUT` count as completed; `ERROR` and never-started tasks + are re-run (passed to the runner as repeated `--task-id` flags in episode + order). +- **Answer-leak prevention**: an interrupted task may already have been + distilled into daily notes during graceful shutdown; re-running it with + that memory injected would inflate scores. Before resuming, + `resume.py cleanup` therefore removes residual memory **only for tasks + about to be re-run** (daily/digest notes, session/dialog, mem_session; + matched via `session_id = pibench_{task}_*`). Completed tasks' memories are + never touched. Daily index files are refreshed **only for the dates that + lost notes**, by full workspace-relative wikilink path — and when the ReMe + package is importable, the refresh reuses ReMe's own daily-index rebuild + logic (`refresh_day_index`), so same-named notes on other dates are never + modified. +- **fresh vs resume are mutually exclusive**: a full memory wipe belongs to + fresh mode only (`run_all.sh` default, executed before any service starts); + resume never wipes. + +## 10. Customization entry points + +| Goal | Location | +|---|---| +| Base model of the agent under test | `REME_MODEL_NAME` in `env.sh` | +| user_agent / judger models | `config/models/reme.yaml` | +| Agent system prompt | `bridge_reme.py` `build_system_prompt()` | +| Memory retrieval limit/threshold | `--search-limit/--search-min-score` on the bridge command in `run_persona.sh` | +| ReMe internal parameters | **Do not modify ReMe source**; write a dedicated config modeled on `reme/config/beam.yaml` and override via `resolve_app_config(config=...)` (see bridge `_init_reme_app`) | +| Turn timeout / tool iteration cap | `config/models/reme.yaml` `run.turn_timeout`, `model.max_tool_iterations` | + +## 11. Troubleshooting + +- **Port already in use**: the scripts auto-kill residual processes on the + four port groups above; if another suite (e.g. a different π-Bench + experiment) holds them, stop it first or change the port table in + run_persona.sh. +- **Bridge exits immediately with workspace locked**: another bridge already + holds the same workspace; make sure each persona uses its own + `--workspace-dir` (the scripts allocate one per persona). +- **Runner reports `${USER_API_KEY} ... empty`**: env.sh is unfilled or not + sourced; run_persona.sh sources env.sh automatically — when running the + runner manually, `source env.sh` first. +- **`Cannot import 'reme'`**: the bridge must run with + `${REME_DIR}/.venv/bin/python` (run_persona.sh already does); otherwise + check that `REME_DIR` points at the ReMe repository root. +- **AppWorld fails to start**: run `bash scripts/setup_appworld.sh` in the + π-Bench repo first (downloads data); inspect + `logs/appworld_*_.log`. +- **trace_history.yaml not found**: the runner needs + `config/bench/evaluation/trace_history.yaml`; this suite ships the file and + passes it explicitly via `--history-config-path`, and run_persona.sh fails + fast with a clear error if it is missing. Always launch run_persona.sh / + run_all.sh from the suite directory. + +## 12. Privacy and security + +- The suite code and config templates contain **no real API keys, user names + or absolute paths**; real keys live only in your local `env.sh` + (git-ignored). +- `logs/`, `outputs/`, `reme_workspace/` and `nanobot_workspace/` contain + full conversations and model outputs; never commit or share them. +- The `data` symlink points at the official π-Bench evaluation data; respect + its data license terms. diff --git a/benchmark/pibench/README_ZH.md b/benchmark/pibench/README_ZH.md new file mode 100644 index 00000000..0a8b58d6 --- /dev/null +++ b/benchmark/pibench/README_ZH.md @@ -0,0 +1,284 @@ +# π-Bench 评测说明 + +[English version](./README.md) + +将 **ReMe agent（带持久记忆）** 接入 **π-Bench**（Proactive Personal Assistant +Benchmark）的胶水层评测套件。只含对接所需的最小代码与配置；π-Bench 框架 +（`src/`）、评测数据（`data/`）、AppWorld 工具环境、ReMe 本体均为**外部第三方 +依赖**，通过符号链接与环境变量原位引用，不随本套件分发。 + +- π-Bench: https://github.com/Simplified-Reasoning/Pi-Bench （arXiv: 2605.14678） +- ReMe: 你所在 ReMe 仓库的根目录（本套件推荐放在 `ReMe/benchmark/pibench/`） + +## 1. 架构总览 + +``` +π-Bench runner (src.main --mode run) + │ user_agent（模拟用户 LLM）按 data/{persona}/episode.yaml 顺序 + │ 逐任务、多轮地与 agent 对话，并在 run 阶段判定隐藏意图(PROC) + ▼ +test server (π-Bench scripts/test_server.py, HTTP 长轮询) + ▲ /send │ /poll + │ ▼ +bridge_reme.py ──────────────► ReMe Application（以库方式内嵌启动） + │ ├─ agent_wrapper: 被测 agent（AgentScope） + │ ├─ jobs: search / auto_memory / daily_write + │ └─ workspace: reme_workspace/{persona}/ + │ （每 persona 独立持久记忆库，互不可见） + └──── MCP ────► AppWorld MCP ────► AppWorld API（工具/应用环境） + +π-Bench runner (src.main --mode eval) + judger（裁判 LLM）读取 trace，按 checklist 逐条 YES/NO 打分(COMP) +``` + +要点： +- bridge 用 **ReMe 自己的 venv python** 运行，把 ReMe 当库用（`resolve_app_config` + + `Application`），**ReMe 源码零改动**。 +- 每条用户消息都会自动触发一次 ReMe memory `search` 并把命中记忆注入当前消息 + （参数见 §8）；任务结束（reset）时会话被 `auto_memory` 提炼为 daily 笔记落盘。 +- agent 执行的每一轮工具调用（AppWorld MCP + ReMe job 工具）都会被采集并以 + `tool_steps` 形式写入 trace，供 π-Bench 的 `tools_evaluation_path` 脚本 + 对工具行为评分（§7）。 +- π-Bench 的 `data/`、`src/`、AppWorld 均不属于本套件，需先装好 π-Bench（§3.1）。 + +## 2. 目录结构 + +``` +pibench/ +├── README.md / README_ZH.md # 本文档（英文 / 中文） +├── env.sh.example # 环境配置模板（复制为 env.sh 后填写 TODO 项） +├── bridge_reme.py # ReMe ↔ test server 桥接（记忆注入/保存、 +│ # profile 注入、工具调用轨迹采集） +├── run_persona.sh # 单 persona 全流程（5 个服务 + run + eval） +├── run_all.sh # 5 个 persona 批跑（fresh/resume，默认 2 并行） +├── resume.py # 断点续跑：完成判定 + 中断任务残留记忆的外科清理 +├── fix_trace_logs.py # run 输出 → ~/.nanobot/trace_logs 转换， +│ # 并把工具轨迹合并进 turn 文件（eval 前置） +├── .gitignore # 排除 env.sh 与全部运行产物 +└── config/ + ├── models/reme.yaml # runner 模型配置（model_id=reme） + └── bench/evaluation/trace_history.yaml # trace 渲染策略（随套件提供， + # 经 --history-config-path 显式传入） +``` + +运行时自动生成（均被 .gitignore 排除）：`data`（符号链接）、`logs/`、 +`outputs/`、`reme_workspace/`、`nanobot_workspace/`。 + +## 3. 前置依赖（第三方，先装好） + +### 3.1 π-Bench 仓库（含 AppWorld） + +```bash +git clone https://github.com/Simplified-Reasoning/Pi-Bench.git +cd +python3.11 -m venv .venv # 脚本约定使用 .venv 这个目录名 +source .venv/bin/activate +pip install -e . # pibench runner（src.main） +bash scripts/setup_appworld.sh # 安装 AppWorld 并下载其数据（体积较大，需网络） +``` + +装完自检： +```bash +ls data/ # 应含 researcher marketer pharmacist law_trainee Financier +.venv/bin/python -c "import src" && echo OK +.venv/bin/appworld --help >/dev/null && echo OK +``` + +### 3.2 ReMe 仓库 + +```bash +cd # ReMe 仓库根目录（含 reme/ 包） +python3.11 -m venv .venv # 脚本约定使用 .venv 这个目录名 +source .venv/bin/activate +pip install -e . # 或按 ReMe 自身安装方式，保证 `import reme` 可用 +``` + +自检：`.venv/bin/python -c "import reme; print('ok')"` + +## 4. 安装本套件（逐步） + +1. **放置套件**（推荐放进 ReMe 仓库，`REME_DIR` 可自动推断）： + ```bash + cp -r pibench /benchmark/pibench + cd /benchmark/pibench + ``` + 若放在其他位置，稍后在 env.sh 中显式设置 `REME_DIR`。 + +2. **创建环境文件并填写自定义参数**： + ```bash + cp env.sh.example env.sh + ``` + 打开 `env.sh`，必填项（标 TODO 的）： + | 变量 | 说明 | + |---|---| + | `PI_BENCH_ROOT` | π-Bench 仓库根目录（含 `src/` `data/` `.venv` `third_party/appworld`） | + | `USER_API_KEY` | 模拟用户 LLM 的 API key（run 阶段判定隐藏意图） | + | `JUDGER_API_KEY` | 裁判 LLM 的 API key（eval 阶段 checklist 打分） | + | `BRAVE_SEARCH_API_KEY` | 可选；agent 的 web_search 工具用，不用填 `dummy` | + + 可选调整：`REME_MODEL_NAME`（被测 agent 基模）、`REME_DIR`、 + `REME_LLM_BASE_URL`（默认 DashScope OpenAI 兼容端点）。 + +3. **链接评测数据**（π-Bench 数据原位引用，不复制）： + ```bash + ln -s "$PI_BENCH_ROOT/data" data + ``` + +4. **（可选）调整模型配置** `config/models/reme.yaml`： + - `user_agent.model` / `judger.model`：模拟用户与裁判的模型名（字面量， + π-Bench 仅对 base_url/api_key 做 `${ENV}` 展开）。 + - `run.turn_timeout`、`max_tool_iterations` 等按需。 + +5. **冒烟自检**（不启动评测）： + ```bash + bash -n run_all.sh && bash -n run_persona.sh + source env.sh && "$REME_DIR/.venv/bin/python" -c "import reme; print('reme ok')" + ``` + +## 5. 运行评测 + +> ⚠️ 长时间运行请放进 `screen`，**不要用 nohup**（nohup 在沙箱/受限环境下 +> 会丢失权限上下文导致子进程异常）。 + +```bash +# 完整正式评测：先清空全部 persona 的记忆/输出/trace，再从头跑（默认 fresh，2 并行） +mkdir -p logs # 全新部署时 logs/ 尚不存在，先建再重定向 +screen -dmS pibench_suite bash -c "cd $(pwd) && bash run_all.sh > logs/run_all_master.log 2>&1" + +# 断点续跑（中断后继续；不清记忆，跳过已完成任务） +bash run_all.sh --resume + +# 其他用法 +bash run_all.sh --parallel 1 # 串行 +bash run_all.sh --resume --skip-eval # 只跑 run 阶段 +bash run_persona.sh researcher # 单 persona（默认 --resume 语义） +bash run_persona.sh researcher --fresh +``` + +耗时参考：5 persona × 20 任务、2 并行，fresh 全量约 12–14 小时。 + +任一 persona 失败时 `run_all.sh` 以非零状态退出，上层自动化不会把部分失败 +的评测误判为成功。 + +## 6. 端口分配（多 persona 并行互不冲突） + +| persona | AppWorld API | AppWorld MCP | Test Server | ReMe 内部服务 | +|-------------|------|-------|------|-------| +| marketer | 9001 | 10001 | 9998 | 18766 | +| law_trainee | 9002 | 10002 | 9997 | 18767 | +| pharmacist | 9003 | 10003 | 9996 | 18768 | +| researcher | 9004 | 10004 | 9995 | 18765 | +| Financier | 9005 | 10005 | 9994 | 18769 | + +## 7. 输出与分数 + +- **结果**：`outputs/reme/{persona}/{task}/eval/results/*_result.json` + - `overall_average_score`：checklist 完整度（COMP，judger 逐条 YES/NO 按依赖组加权） + - `overall_proactiveness_average_score`：主动性（PROC，run 阶段 user_agent + 判定隐藏意图覆盖率；每个任务文件同时携带全局均值） +- **trace**：`~/.nanobot/trace_logs/reme/{persona}/{task}/...`（eval 的判分输入） +- **日志**：`logs/`（`suite_.log` 为每 persona 总日志，`bridge_*`、 + `runner_run/eval_*`、`appworld_*`、`test_server_*` 分服务） +- **记忆库**：`reme_workspace/{persona}/`（daily/digest 笔记、session 原始对话、 + BM25 索引等；跨运行持久，fresh 才清空） + +查看汇总： +```bash +grep -h "overall_average_score\|overall_proactiveness" \ + outputs/reme/*/*/eval/results/*_result.json | head +``` + +### 工具轨迹采集（tools_evaluation 支持） + +部分任务定义了 `objectives.tools_evaluation_path`：用 Python 脚本对工具行为 +打分（例如"临时 Todoist 看板已创建并被删除"）。这些脚本需要 trace 里有真实 +的工具调用记录。采集链路： + +1. 每轮 `reply()` 之后，bridge 读取 AgentScope 落盘的会话状态，提取本轮新增 + 的 `tool_call` / `tool_result` 块（工具名、参数、结果）。 +2. 记录按 turn 编号追加写入 + `outputs/reme/{persona}/{task}/history/{ts}-tools.jsonl`；AgentScope 的 + MCP 工具名（`mcp__AppWorld__`）会规范化为 π-Bench 约定 + （`mcp_appworld_`）。 +3. `fix_trace_logs.py` 将每个 `{ts}-messages.jsonl` 运行与时间上最接近的 + tools 旁路文件配对，把记录合并进生成的 `turn_N.json` 的 `tool_steps` + 字段——这是 π-Bench `collect_tool_history()` 支持的两种工具轨迹格式之一。 +4. eval 阶段 `tool_steps` 既提供给 tools_evaluation 脚本，也会被渲染为 + judger 可见的 ``。 + +## 8. 记忆机制（本套件的核心设计） + +- **persona 隔离**：每个 persona 独立 workspace（`reme_workspace/{persona}/`）， + bridge 启动时对 workspace 加 `.bridge.lock` 排他锁，两个 bridge 不可能共用 + 同一记忆库；一个 persona 的 memory search 永远接触不到其他 persona 的记忆。 +- **写入**：任务结束（runner 发送 reset）时，会话经 `auto_memory` job 提炼为 + daily 笔记落盘，后台 watcher 建 BM25 索引。保存为非阻塞后台任务， + 新会话首条消息会先等待在途写入完成再检索。 +- **读取**：bridge 每收到一条用户消息自动 `search` 一次并注入命中记忆 + （`[Relevant memories from previous sessions]` 前缀），无命中则原样透传。 + 检索参数（bridge 命令行，可在 run_persona.sh 中调整）： + - `--search-limit 3`：每条消息最多注入 3 个记忆块； + - `--search-min-score 2.0`：过滤弱 BM25 命中； + - `tool_context_id` 按任务轮换：同一任务内已注入的记忆块不重复注入 + （ReMe 自带 seen-chunk 去重，24h TTL），任务边界后恢复正常召回。 +- **无自泄漏**：进行中的会话尚未入库（save 发生在 reset），任务不会检索到 + 自己未完成的内容。 +- agent 同时持有 `search`/`daily_write` 工具，可主动检索/记录。 +- **system prompt**：`bridge_reme.py:build_system_prompt()` 内置 + HIDDEN-NEEDS 协议（面向 proactiveness），并把 `data/{persona}/profile.yaml` + 的 persona profile 注入每轮 system prompt。 + +## 9. 断点续跑与记忆清理语义 + +- **完成判定**（resume.py）：扫描 `outputs/reme/{persona}/**/history/*-log.jsonl` + 与 `outputs/reme/{persona}/run/*-log.jsonl` 中的 + `Task finished task_id=X status=Y`。每个任务以**事件时间最新**的记录为准 + （优先取记录的 `timestamp`，回退 `timestamp_iso`，再回退日志文件名中的 + 时间戳）——文件类别与读取顺序本身不能覆盖更新的记录，因此旧的 run 级 + SUCCESS 不会掩盖更新的 per-task ERROR。`SUCCESS/MAX_TURNS/TIMEOUT` 记为 + 完成，`ERROR`/未开始的任务重跑（按 episode 顺序以 `--task-id` 传给 runner）。 +- **防答案泄漏**：被中断的任务可能已在优雅退出时提炼成 daily 笔记，直接重跑会 + 把答案注入、抬高分数。因此 resume 启动前 `resume.py cleanup` **只删除待重跑 + 任务**的残留记忆（daily/digest 笔记、session/dialog、mem_session，按 + `session_id = pibench_{task}_*` 匹配），已完成任务的记忆一律不动。daily + 索引**只刷新实际发生删除的日期**，按完整的 workspace 相对 wikilink 路径 + 匹配；当 ReMe 包可导入时，刷新直接复用 ReMe 自带的 daily 索引重建逻辑 + （`refresh_day_index`），不会误改其他日期下的同名笔记条目。 +- **fresh vs resume 互斥**：全量清记忆只属于 fresh 模式（`run_all.sh` 默认， + 在任何服务启动前执行）；resume 永不清全量。 + +## 10. 自定义与调优入口 + +| 目标 | 位置 | +|---|---| +| 被测 agent 基模 | `env.sh` 的 `REME_MODEL_NAME` | +| user_agent / judger 模型 | `config/models/reme.yaml` | +| agent system prompt | `bridge_reme.py` `build_system_prompt()` | +| 记忆检索条数/阈值 | `run_persona.sh` bridge 启动命令的 `--search-limit/--search-min-score` | +| ReMe 内部参数 | **不要改 ReMe 源码**；仿照 `reme/config/beam.yaml` 写专有配置，经 `resolve_app_config(config=...)` 覆盖（见 bridge `_init_reme_app`） | +| 轮超时/工具迭代上限 | `config/models/reme.yaml` `run.turn_timeout`、`model.max_tool_iterations` | + +## 11. 故障排查 + +- **端口被占用**：脚本会自动 kill 上述 4 组端口上的残留进程；若与其他套件 + （如别的 π-Bench 实验）冲突，请先停掉对方或改 run_persona.sh 的端口表。 +- **bridge 启动即退出，提示 workspace locked**：另一个 bridge 正占用同一 + workspace；确认每个 persona 用各自的 `--workspace-dir`（脚本已按 persona 分配）。 +- **runner 报 `${USER_API_KEY} ... empty`**：env.sh 未填写或未生效； + run_persona.sh 会自动 source env.sh，手动运行 runner 时请先 `source env.sh`。 +- **`Cannot import 'reme'`**：bridge 必须用 `${REME_DIR}/.venv/bin/python` 运行 + （run_persona.sh 已如此），或检查 `REME_DIR` 是否指向 ReMe 仓库根目录。 +- **AppWorld 启动失败**：先在 π-Bench 仓库执行 `bash scripts/setup_appworld.sh` + 下载数据；查看 `logs/appworld_*_.log`。 +- **trace_history.yaml 找不到**：runner 需要 + `config/bench/evaluation/trace_history.yaml`；本套件已随附该文件并通过 + `--history-config-path` 显式传入，run_persona.sh 启动前会做存在性检查， + 缺失时立即报出清晰错误。请始终从套件目录启动 run_persona.sh / run_all.sh。 + +## 12. 隐私与安全 + +- 套件代码与配置模板中**不含任何真实 API key、用户名或绝对路径**； + 真实 key 只存在于你本地的 `env.sh`（已被 .gitignore 排除）。 +- `logs/`、`outputs/`、`reme_workspace/`、`nanobot_workspace/` 含完整对话内容 + 与模型输出，请勿提交仓库或外传。 +- `data` 符号链接指向 π-Bench 官方评测数据，请遵守其数据许可条款。 diff --git a/benchmark/pibench/bridge_reme.py b/benchmark/pibench/bridge_reme.py new file mode 100755 index 00000000..dadd7877 --- /dev/null +++ b/benchmark/pibench/bridge_reme.py @@ -0,0 +1,1039 @@ +#!/usr/bin/env python3 +""" +Bridge script: Connects ReMe agent to Pi-Bench Test Server. + +Uses ReMe's AgentScope-based agent wrapper directly as a library, +with MCP integration to AppWorld and cross-session memory support. + +Flow: +1. Poll Test Server /poll for user messages +2. Forward to ReMe agent (via AgentScope) +3. Extract reply text +4. Send reply back to Test Server POST /send +5. On session end (reset), save conversation as ReMe daily memory (non-blocking) +6. On every incoming user message, trigger a ReMe memory search and inject + the relevant memories retrieved from previous sessions +7. After every agent reply, capture the turn's tool calls (tool name, + arguments, result) from the persisted AgentScope session state and append + them to outputs////history/-tools.jsonl; + fix_trace_logs.py merges these into the per-turn traces as tool_steps so + π-Bench tools_evaluation scripts can score tool behavior. + +Key design decisions: +- Memory saves are non-blocking (fire-and-forget asyncio tasks) so reset + acknowledgments are sent immediately and don't time out. +- A pending-save tracker ensures the first message of a new session waits + for any in-flight memory writes to complete before searching. +- User profile is loaded from data/{user_id}/profile.yaml and injected + into every turn's system prompt. +- AgentScope session state is maintained via `resume` within a task, + and cleared on reset for cross-task isolation. +- Memory search tuning: each search is capped at `--search-limit` + results (default 3), weak BM25 hits below `--search-min-score` + (default 2.0) are filtered, and a per-task `tool_context_id` + enables ReMe's seen-chunk dedup so the same memory chunk is not + re-injected on every turn of the same task. +- Persona isolation: the workspace defaults to a per-user subdirectory + and an exclusive lock file guarantees that no two bridges can share + one memory store at runtime. + +Usage: + python bridge_reme.py [--test-server-url URL] [--reme-dir DIR] +""" + +import argparse +import asyncio +import fcntl +import json +import logging +import os +import re +import signal +import sys +from datetime import datetime +from pathlib import Path +from typing import Any, Dict, List, Optional + +import httpx +import yaml + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", +) +logger = logging.getLogger("bridge_reme") + + +# ─── User Profile Loading ───────────────────────────────────────────── + + +def load_user_profile(data_root: str, user_id: str) -> str: + """Load user profile YAML and return as formatted text for the system prompt. + + Handles the full Pi-Bench profile schema: role (with sub-sections), + preferences, and long_term_goals. + """ + profile_path = Path(data_root) / user_id / "profile.yaml" + if not profile_path.exists(): + logger.warning("User profile not found: %s", profile_path) + return "" + + with open(profile_path, "r", encoding="utf-8") as f: + profile = yaml.safe_load(f) + + if not profile: + return "" + + parts = [] + + # Role section: contains the full persona description + if "role" in profile and profile["role"]: + role_text = str(profile["role"]).strip() + if role_text: + parts.append(f"## User Profile\n{role_text}") + + # Preferences section + if "preferences" in profile and profile["preferences"]: + prefs = profile["preferences"] + if isinstance(prefs, dict): + pref_lines = [] + for k, v in prefs.items(): + if v is not None and str(v).strip(): + pref_lines.append(f"- {k}: {v}") + if pref_lines: + parts.append("## Preferences\n" + "\n".join(pref_lines)) + elif isinstance(prefs, str): + parts.append(f"## Preferences\n{prefs}") + + # Long-term goals + if "long_term_goals" in profile and profile["long_term_goals"]: + goals = profile["long_term_goals"] + if isinstance(goals, list): + goal_lines = [f"- {g}" for g in goals if g] + if goal_lines: + parts.append("## Long-term Goals\n" + "\n".join(goal_lines)) + elif isinstance(goals, str): + parts.append(f"## Long-term Goals\n{goals}") + + result = "\n\n".join(parts) + logger.info( + "Loaded profile for %s: %d chars, sections: %s", + user_id, + len(result), + [k for k in ["role", "preferences", "long_term_goals"] if k in profile], + ) + return result + + +def build_system_prompt(user_profile: str) -> str: + """Build the system prompt for the ReMe agent with profile context.""" + base_prompt = """\ +You are a proactive personal assistant agent in a long-horizon evaluation. Be thorough, anticipatory, +detail-oriented; use the user's profile, memory and tools proactively +(AppWorld via MCP; memory `search`/`daily_write`; file tools). + +## HIDDEN-NEEDS PROTOCOL (MANDATORY) +Every task carries implicit needs the user does not state. Before each substantive response: +1. Derive the implicit needs of THIS task (method below), plus what the user's profile and past sessions imply. +2. Cover EVERY need explicitly and specifically in this response. +3. Anything you cannot cover now, you MUST still raise explicitly: one precise question or a concrete +next step targeting exactly that need. Generic closers do not count. + +## HOW TO DERIVE IMPLICIT NEEDS +- Entities: for every item the task involves (a paper, product, person, account, case, event), cover the +attributes this user would need: what it is + key details, availability or cost, suitability/evaluation, +how to proceed, risks, and alternatives. +- Action completeness: if the task implies an action chain (prepare → execute → verify), cover every +stage, including verification and closing the loop. +- Context: apply everything the user's profile, constraints and past sessions imply (budget, size, format, +style, tools, deadlines) without being reminded. +- Structure: provide the format or verdict the user would expect (table, overall rating, pass/fail, +conclusion-first) whenever applicable. + +## DELIVERABLE STRUCTURE +What (conclusion first) → Why → How → Risks (limits, fallbacks) → Next steps. + +## STRICTNESS +An implicit need counts only with specific, detailed content or a concrete action — vague or generic +scores nothing. Deliver specifics in your FIRST response. +""" + + if user_profile: + base_prompt += f"\n\n---\n\n{user_profile}\n" + + base_prompt += ( + "\n\n---\n\nAlways respond in the same language as the user's message. Use tools proactively to help the user." + ) + return base_prompt + + +# ─── ReMe Bridge ────────────────────────────────────────────────────── + + +class ReMeBridge: + """Bridge between Pi-Bench Test Server and ReMe agent.""" + + def __init__( + self, + test_server_url: str = "http://localhost:9999", + appworld_mcp_url: str = "http://localhost:10000/mcp", + reme_dir: str = "", + data_root: str = "data", + user_id: str = "researcher", + poll_timeout: int = 30, + workspace_dir: str = "", + model_name: str = "qwen3.6-plus", + model_base_url: str = "", + model_api_key: str = "", + reme_port: int = 18765, + search_limit: int = 3, + search_min_score: float = 2.0, + outputs_dir: str = "", + model_id: str = "reme", + ): + self.test_server_url = test_server_url.rstrip("/") + self.appworld_mcp_url = appworld_mcp_url + self.reme_dir = Path(reme_dir).resolve() if reme_dir else None + self.data_root = Path(data_root) + self.user_id = user_id + self.poll_timeout = poll_timeout + if workspace_dir: + self.workspace_dir = Path(workspace_dir) + else: + # Per-persona default so two bridges can never share a memory store. + root = os.environ.get("REME_WORKSPACE_ROOT", "/tmp/reme_pibench_workspaces") + self.workspace_dir = Path(root) / user_id + self.model_name = model_name + self.model_base_url = model_base_url + self.model_api_key = model_api_key + self.reme_port = reme_port + self.search_limit = search_limit + self.search_min_score = search_min_score + # Runner outputs root; the tool-trace sidecar files are written next + # to the runner's *-messages.jsonl history files. + if outputs_dir: + self.outputs_dir = Path(outputs_dir).resolve() + else: + self.outputs_dir = (self.data_root.parent / "outputs").resolve() + self.model_id = model_id + # Task generation counter: rotated on every reset so the search dedup + # context (tool_context_id) is scoped to a single task. + self.task_seq = 0 + self._workspace_lock_fd: Optional[int] = None + + # Tool-trace capture state (per bridge lifetime): + # - turn counter per chat (each user message = one π-Bench turn) + # - already-seen session content block ids (tool_call / tool_result) + # - tool_call blocks waiting for their tool_result block + # - sidecar file timestamp per chat (fixed at first capture) + self._turn_by_chat: Dict[str, int] = {} + self._seen_tool_block_ids: set = set() + self._pending_tool_calls: Dict[str, Dict] = {} + self._tools_file_ts: Dict[str, str] = {} + + self.client: Optional[httpx.AsyncClient] = None + self.running = False + + # ReMe components + self.app = None + self.agent_wrapper = None + self.auto_memory_job = None + self.search_job = None + + # Session state + self.user_profile_text = "" + self.session_messages: Dict[str, List[Dict]] = {} + self.agent_session_id: Optional[str] = None + + # Non-blocking memory save tracking + self._pending_memory_tasks: List[asyncio.Task] = [] + + async def start(self): + """Initialize the ReMe application and bridge components.""" + self.client = httpx.AsyncClient(timeout=300.0, trust_env=False) + self.running = True + + # Enforce per-persona workspace isolation before anything else: an + # exclusive lock guarantees no other bridge can use this memory store. + self._acquire_workspace_lock() + if self.workspace_dir.name != self.user_id: + logger.warning( + "Workspace basename %r != user_id %r; cross-persona isolation " + "relies on each bridge having its own workspace_dir", + self.workspace_dir.name, + self.user_id, + ) + logger.info( + "Memory isolation: user=%s workspace=%s reme_port=%d", + self.user_id, + self.workspace_dir, + self.reme_port, + ) + + # Load user profile + self.user_profile_text = load_user_profile(str(self.data_root), self.user_id) + logger.info("User profile loaded: %d chars", len(self.user_profile_text)) + + # Initialize ReMe application + await self._init_reme_app() + + logger.info( + "Bridge started: test_server=%s appworld_mcp=%s user=%s model=%s", + self.test_server_url, + self.appworld_mcp_url, + self.user_id, + self.model_name, + ) + + def _acquire_workspace_lock(self): + """Take an exclusive lock on the workspace (persona isolation guard).""" + self.workspace_dir.mkdir(parents=True, exist_ok=True) + lock_path = self.workspace_dir / ".bridge.lock" + fd = os.open(str(lock_path), os.O_CREAT | os.O_RDWR) + try: + fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError as exc: + os.close(fd) + raise SystemExit( + f"Workspace {self.workspace_dir} is already locked by another " + f"bridge process; each persona needs its own workspace_dir.", + ) from exc + os.ftruncate(fd, 0) + os.write(fd, f"pid={os.getpid()} user={self.user_id}\n".encode()) + self._workspace_lock_fd = fd + + def _release_workspace_lock(self): + if self._workspace_lock_fd is not None: + try: + fcntl.flock(self._workspace_lock_fd, fcntl.LOCK_UN) + os.close(self._workspace_lock_fd) + except OSError: + pass + self._workspace_lock_fd = None + + async def _init_reme_app(self): + """Initialize the ReMe application with proper configuration.""" + # Add reme to Python path so imports work + if self.reme_dir: + reme_str = str(self.reme_dir) + if reme_str not in sys.path: + sys.path.insert(0, reme_str) + + try: + from reme.config import resolve_app_config + from reme.application import Application + except ImportError as exc: + raise RuntimeError( + "Cannot import 'reme'. Run the bridge with the ReMe venv " + "python, or pass --reme-dir pointing to the ReMe repo root.", + ) from exc + + # Set environment variables for ReMe LLM config expansion + os.environ["LLM_MODEL_NAME"] = self.model_name + if self.model_base_url: + os.environ["LLM_BASE_URL"] = self.model_base_url + if self.model_api_key: + os.environ["LLM_API_KEY"] = self.model_api_key + # Ensure BRAVE_SEARCH_API_KEY is set (required by some tools) + if not os.environ.get("BRAVE_SEARCH_API_KEY"): + os.environ["BRAVE_SEARCH_API_KEY"] = "dummy" + + # Ensure workspace exists + self.workspace_dir.mkdir(parents=True, exist_ok=True) + + # Load .env from reme dir if available + environment = {} + if self.reme_dir: + env_path = self.reme_dir / ".env" + if env_path.exists(): + with open(env_path, "r", encoding="utf-8") as f: + for line in f: + line = line.strip() + if line and not line.startswith("#") and "=" in line: + key, _, value = line.partition("=") + environment[key.strip()] = value.strip() + + # Override with explicit values + if self.model_base_url: + environment["LLM_BASE_URL"] = self.model_base_url + if self.model_api_key: + environment["LLM_API_KEY"] = self.model_api_key + environment["LLM_MODEL_NAME"] = self.model_name + + # Use resolve_app_config to load default.yaml and merge overrides + reme_config = resolve_app_config( + log_config=False, + workspace_dir=str(self.workspace_dir), + service={"backend": "http", "host": "127.0.0.1", "port": self.reme_port}, + environment=environment, + ) + + try: + self.app = Application(**reme_config) + await self.app.start() + + # Get components + self.agent_wrapper = self.app.context.components.get("agent_wrapper", {}).get("default") + if self.agent_wrapper is None: + raise RuntimeError("agent_wrapper component 'default' not found") + + # Get jobs for memory operations + self.auto_memory_job = self.app.context.jobs.get("auto_memory") + self.search_job = self.app.context.jobs.get("search") + + logger.info("ReMe initialized OK") + logger.info(" agent_wrapper: %s", getattr(self.agent_wrapper, "name", "default")) + logger.info(" auto_memory: %s", "yes" if self.auto_memory_job else "no") + logger.info(" search: %s", "yes" if self.search_job else "no") + logger.info(" total jobs: %d", len(self.app.context.jobs)) + + except Exception as e: + logger.exception("Failed to initialize ReMe: %s", e) + raise + + async def stop(self): + """Stop the bridge and cleanup.""" + self.running = False + self._release_workspace_lock() + + # Wait for pending memory saves + if self._pending_memory_tasks: + logger.info("Waiting for %d pending memory saves...", len(self._pending_memory_tasks)) + for task in self._pending_memory_tasks: + try: + await asyncio.wait_for(task, timeout=60.0) + except (asyncio.TimeoutError, Exception) as e: + logger.warning("Pending memory save timed out or failed: %s", e) + + if self.app: + try: + await self.app.close() + except Exception as e: + logger.warning("Error closing ReMe app: %s", e) + if self.client: + await self.client.aclose() + self.client = None + logger.info("Bridge stopped") + + def _create_mcp_client(self): + """Create an MCP client for AppWorld.""" + from agentscope.mcp import MCPClient, HttpMCPConfig + + return MCPClient( + name="AppWorld", + is_stateful=False, + mcp_config=HttpMCPConfig( + url=self.appworld_mcp_url, + timeout=120.0, + ), + ) + + def _build_agent_toolkit(self, mcp_client, job_tool_names: List[str]): + """Build the agent toolkit so AppWorld MCP tools are really registered. + + agent_wrapper.reply() accepts a prebuilt ``toolkit`` kwarg but does + not wire a bare ``mcps`` kwarg into the agent, so the toolkit is + assembled here: ReMe job tools (search / auto_memory / daily_write) + plus the AppWorld MCP client. Returns None when the wrapper lacks + the required hooks; the caller then falls back to plain kwargs. + """ + try: + from agentscope.tool import Toolkit + except ImportError as exc: + logger.warning("Cannot import agentscope Toolkit: %s", exc) + return None + + make_tool = getattr(type(self.agent_wrapper), "_make_tool", None) + if make_tool is None: + logger.warning( + "agent_wrapper %s cannot wrap jobs as tools; falling back to kwargs (MCP tools may be unavailable)", + type(self.agent_wrapper).__name__, + ) + return None + + tools = [] + for name in job_tool_names: + job = self.app.context.jobs.get(name) if self.app is not None else None + if job is None: + continue + try: + tools.append(make_tool(job, None, None)) + except Exception as exc: + logger.warning("Failed to wrap job '%s' as agent tool: %s", name, exc) + + # AgentScope builtin file tools (bash/read/write/edit/glob/grep). + # A prebuilt toolkit bypasses _build_agent's builtin-tools branch, + # and since ReMe commit e05b201d builtins are opt-in, they must be + # added here explicitly — the agent needs them to read task asset + # files (e.g. instrument_booking_brief.md) from its workspace. + builtin = getattr(self.agent_wrapper, "_builtin_tools", None) + if builtin is not None: + try: + tools.extend(builtin("all", sequential_tool_calls=True)) + except Exception as exc: + logger.warning("Failed to add builtin tools: %s", exc) + + try: + return Toolkit(tools=tools, mcps=[mcp_client]) + except Exception as exc: + logger.warning("Failed to build agent toolkit: %s", exc) + return None + + # ─── Memory Operations ────────────────────────────────────────── + + async def _wait_for_pending_memory_saves(self): + """Wait for all in-flight memory save tasks to complete.""" + if not self._pending_memory_tasks: + return + logger.info( + "Waiting for %d pending memory saves before search...", + len(self._pending_memory_tasks), + ) + tasks = self._pending_memory_tasks[:] + self._pending_memory_tasks.clear() + for task in tasks: + try: + await asyncio.wait_for(task, timeout=120.0) + except asyncio.TimeoutError: + logger.warning("Memory save task timed out (120s)") + except Exception as e: + logger.warning("Memory save task failed: %s", e) + + async def _search_memory(self, query: str, tool_context_id: str = "") -> str: + """Search ReMe memory for relevant context from previous sessions.""" + if not self.search_job: + return "" + try: + response = await self.search_job( + query=query, + limit=self.search_limit, + min_score=self.search_min_score, + tool_context_id=tool_context_id, + ) + if response.success and response.answer: + returned = response.metadata.get("counts", {}).get("returned", "?") + logger.info( + "Memory search: hits=%s limit=%d min_score=%s ctx=%s", + returned, + self.search_limit, + self.search_min_score, + tool_context_id or "-", + ) + return response.answer + except Exception as e: + logger.warning("Memory search failed: %s", e) + return "" + + async def _do_save_session_memory(self, chat_id: str, messages: List[Dict]): + """Actually perform the session memory save (runs as background task).""" + if not self.auto_memory_job or not messages: + return + + session_id = f"pibench_{chat_id}_{datetime.now().strftime('%Y%m%d_%H%M%S')}" + + # Convert messages to auto_memory format + memory_messages = [] + for msg in messages: + role = msg.get("role", "user") + memory_messages.append( + { + "role": role, + "name": msg.get("name", "user" if role == "user" else "assistant"), + "content": msg.get("content", ""), + "created_at": msg.get("timestamp", datetime.now().isoformat()), + }, + ) + + try: + logger.info( + "Saving session memory: chat_id=%s messages=%d", + chat_id, + len(memory_messages), + ) + response = await self.auto_memory_job( + messages=memory_messages, + session_id=session_id, + memory_hint=( + f"Pi-Bench evaluation session for task {chat_id}. " + f"User persona: {self.user_id}. " + f"Save key decisions, actions taken, important outcomes, " + f"and any user preferences or context that may be useful " + f"for future sessions." + ), + ) + if response.success: + preview = (response.answer or "OK")[:200] + logger.info("Session memory saved: %s", preview) + else: + logger.warning("Memory save returned unsuccessful: %s", response.answer) + except Exception as e: + logger.exception("Error saving session memory: %s", e) + + def _schedule_memory_save(self, chat_id: str, messages: List[Dict]): + """Schedule a non-blocking memory save task.""" + if not self.auto_memory_job or not messages: + return + + task = asyncio.create_task( + self._do_save_session_memory(chat_id, messages), + name=f"memory_save_{chat_id}", + ) + self._pending_memory_tasks.append(task) + + # Clean up completed tasks from the tracking list + self._pending_memory_tasks = [t for t in self._pending_memory_tasks if not t.done()] + + # ─── Tool Trace Capture ───────────────────────────────────────── + + MCP_TOOL_NAME_RE = re.compile(r"^mcp__(?P[A-Za-z0-9_-]+?)__(?P.+)$") + + @classmethod + def _normalize_tool_name(cls, name: str) -> str: + """Map AgentScope MCP tool names to the π-Bench / nanobot convention. + + AgentScope registers MCP tools as ``mcp____`` while + π-Bench task objectives and tools_evaluation scripts expect + ``mcp__`` (lower-case client, single underscores). + """ + match = cls.MCP_TOOL_NAME_RE.match(name) + if match: + return f"mcp_{match.group('client').lower()}_{match.group('tool')}" + return name + + @staticmethod + def _tool_result_text(output: Any) -> str: + """Flatten an AgentScope tool-result payload into plain text.""" + if isinstance(output, str): + return output + if isinstance(output, list): + parts = [ + str(item.get("text") or "") for item in output if isinstance(item, dict) and item.get("type") == "text" + ] + return "\n".join(parts) + return "" + + def _tools_file_for(self, chat_id: str) -> Path: + """Return (and lazily name) the tool-trace sidecar file of a task.""" + task_dir = self.outputs_dir / self.model_id / self.user_id / chat_id / "history" + task_dir.mkdir(parents=True, exist_ok=True) + if chat_id not in self._tools_file_ts: + self._tools_file_ts[chat_id] = datetime.now().strftime("%Y%m%d_%H%M%S") + return task_dir / f"{self._tools_file_ts[chat_id]}-tools.jsonl" + + def _capture_tool_calls(self, chat_id: str, session_id: str) -> None: + """Record the current turn's tool calls from the AgentScope session. + + After every reply, the agent wrapper dumps the full session context to + ``/mem_session/agentscope/.jsonl``. This method + scans that dump for tool_call / tool_result content blocks that were + not seen before and appends the completed pairs to the per-task + sidecar file consumed by fix_trace_logs.py. + """ + if not session_id: + return + mem_session_dir = "mem_session" + app_config = getattr(getattr(self.app, "context", None), "app_config", None) + if app_config is not None and getattr(app_config, "mem_session_dir", None): + mem_session_dir = app_config.mem_session_dir + state_path = self.workspace_dir / mem_session_dir / "agentscope" / f"{session_id}.jsonl" + if not state_path.is_file(): + return + try: + lines = state_path.read_text(encoding="utf-8").splitlines() + except OSError as exc: + logger.warning("Cannot read agent session state %s: %s", state_path, exc) + return + + turn = self._turn_by_chat.get(chat_id, 0) + records: List[Dict] = [] + for line in lines[1:]: # line 1 is the state header, not a message + try: + msg = json.loads(line) + except json.JSONDecodeError: + continue + content = msg.get("content") + if not isinstance(content, list): + continue + for block in content: + if not isinstance(block, dict): + continue + block_id = str(block.get("id") or "") + block_type = block.get("type") + if block_type not in ("tool_call", "tool_result"): + continue + # A tool_result block reuses its tool_call's id, so dedup + # must be keyed on (type, id), not id alone. + seen_key = (block_type, block_id) + if not block_id or seen_key in self._seen_tool_block_ids: + continue + if block_type == "tool_call": + self._seen_tool_block_ids.add(seen_key) + arguments = block.get("input") or "" + if isinstance(arguments, str): + try: + arguments = json.loads(arguments) + except json.JSONDecodeError: + arguments = {"raw_input": arguments} + self._pending_tool_calls[block_id] = { + "turn": turn, + "name": self._normalize_tool_name(str(block.get("name") or "")), + "arguments": arguments, + } + elif block_type == "tool_result": + self._seen_tool_block_ids.add(seen_key) + call = self._pending_tool_calls.pop(block_id, None) + if call is None: + continue + call["result"] = self._tool_result_text(block.get("output")) + records.append(call) + + if records: + tools_path = self._tools_file_for(chat_id) + with open(tools_path, "a", encoding="utf-8") as f: + for record in records: + f.write(json.dumps(record, ensure_ascii=False) + "\n") + logger.info( + "Tool trace: chat=%s turn=%d captured=%d -> %s", + chat_id, + turn, + len(records), + tools_path.name, + ) + + # ─── Message Processing ───────────────────────────────────────── + + async def process_message(self, _sender_id: str, chat_id: str, content: str) -> Optional[str]: + """Process a user message through the ReMe agent.""" + # Track session messages for later memory save + if chat_id not in self.session_messages: + self.session_messages[chat_id] = [] + + self.session_messages[chat_id].append( + { + "role": "user", + "name": "user", + "content": content, + "timestamp": datetime.now().isoformat(), + }, + ) + + # Each user message is one π-Bench turn; tool records captured after + # the reply below are tagged with this turn number. + self._turn_by_chat[chat_id] = self._turn_by_chat.get(chat_id, 0) + 1 + + # Build system prompt with user profile + system_prompt = build_system_prompt(self.user_profile_text) + + # Create MCP client for AppWorld + mcp_client = self._create_mcp_client() + + # Determine which reme jobs to expose as tools + job_tools = [] + if self.search_job: + job_tools.append("search") + if self.auto_memory_job: + job_tools.extend(["auto_memory", "daily_write"]) + + # On EVERY incoming user message, automatically trigger a ReMe + # memory search and inject the relevant memories retrieved from + # previous sessions. Memory is only surfaced through search + # (relevance-filtered), never dumped wholesale. The in-progress + # session is not in the store yet (saves happen on reset), so a + # task can never retrieve its own partial content. + memory_context = "" + # Wait for any in-flight memory saves so the store is complete + # before searching (no-op when nothing is pending). + await self._wait_for_pending_memory_saves() + if self.search_job: + try: + memory_context = await self._search_memory( + content, + tool_context_id=f"pibench_{self.user_id}_task_{self.task_seq}", + ) + if memory_context: + logger.info("Found relevant memory: %d chars", len(memory_context)) + except Exception as e: + logger.warning("Memory search failed: %s", e) + + try: + # Prepend memory context if available + user_message = content + if memory_context: + user_message = ( + f"[Relevant memories from previous sessions]\n" + f"{memory_context}\n\n" + f"[Current user message]\n{content}" + ) + + # Call ReMe agent with MCP tools and memory tools + reply_kwargs = { + "system_prompt": system_prompt, + "permission_mode": "bypass", + } + toolkit = self._build_agent_toolkit(mcp_client, job_tools) + if toolkit is not None: + # Prebuilt toolkit: registers AppWorld MCP tools AND job tools. + reply_kwargs["toolkit"] = toolkit + else: + # Fallback path (kept for wrappers without toolkit support). + reply_kwargs["mcps"] = [mcp_client] + if job_tools: + reply_kwargs["job_tools"] = job_tools + + # Resume existing session for multi-turn continuity within same task + if self.agent_session_id: + reply_kwargs["resume"] = self.agent_session_id + + result = await self.agent_wrapper.reply(user_message, **reply_kwargs) + + reply_text = result.get("result", "") + session_id = result.get("session_id", "") + + if session_id: + self.agent_session_id = session_id + + # Capture the turn's tool calls for π-Bench tools_evaluation. + self._capture_tool_calls(chat_id, session_id or self.agent_session_id or "") + + # Track the assistant reply + self.session_messages[chat_id].append( + { + "role": "assistant", + "name": "assistant", + "content": reply_text, + "timestamp": datetime.now().isoformat(), + }, + ) + + logger.info("Reply: %d chars, session=%s", len(reply_text), session_id) + return reply_text + + except Exception as e: + logger.exception("Error processing message: %s", e) + return None + + async def handle_reset(self, chat_id: str): + """Handle session reset: schedule non-blocking memory save and clear state.""" + messages = self.session_messages.pop(chat_id, []) + if messages: + self._schedule_memory_save(chat_id, messages) + # Clear agent session for cross-task isolation + self.agent_session_id = None + # New task boundary: rotate the search dedup context so memories can be + # recalled again in the next task while repeats within a task are filtered. + self.task_seq += 1 + + # ─── Test Server Communication ────────────────────────────────── + + async def poll_test_server(self) -> Optional[List[Dict[str, Any]]]: + """Poll Test Server for pending messages. + + Returns None on connection/response errors so the caller can back off; + an empty list means a successful poll with no pending messages. + """ + try: + resp = await self.client.get( + f"{self.test_server_url}/poll", + params={"timeout": self.poll_timeout}, + ) + if resp.is_success: + data = resp.json() + messages = data.get("messages", []) + if messages: + logger.info("Received %d messages", len(messages)) + return messages + logger.warning("Poll returned HTTP %s", resp.status_code) + except Exception as e: + logger.warning("Poll error: %s", e) + return None + + async def send_to_test_server(self, chat_id: str, content: str) -> bool: + """Send reply back to Test Server.""" + payload = { + "chat_id": chat_id, + "content": content, + "media": [], + "meta": {}, + } + try: + resp = await self.client.post( + f"{self.test_server_url}/send", + json=payload, + ) + if resp.is_success: + logger.info("Sent reply: chat_id=%s len=%d", chat_id, len(content)) + return True + logger.error("Failed to send: %s", resp.status_code) + except Exception: + logger.exception("Error sending to Test Server") + return False + + # ─── Main Loop ────────────────────────────────────────────────── + + def _install_signal_handlers(self): + """Install SIGTERM/SIGINT handlers for graceful shutdown.""" + loop = asyncio.get_running_loop() + for sig_name in ("SIGTERM", "SIGINT"): + sig = getattr(signal, sig_name, None) + if sig is not None: + loop.add_signal_handler(sig, self._handle_shutdown_signal, sig_name) + + def _handle_shutdown_signal(self, sig_name: str): + """Handle shutdown signal: stop the bridge loop gracefully.""" + logger.info("Received %s, initiating graceful shutdown...", sig_name) + self.running = False + + async def run(self): + """Main bridge loop.""" + await self.start() + self._install_signal_handlers() + + consecutive_poll_errors = 0 + try: + while self.running: + messages = await self.poll_test_server() + + if messages is None: + # Back off on poll failure to avoid a tight error loop. + consecutive_poll_errors += 1 + if consecutive_poll_errors in (1, 5, 20) or consecutive_poll_errors % 50 == 0: + logger.warning( + "Poll failure #%d, backing off", + consecutive_poll_errors, + ) + await asyncio.sleep(min(2 ** min(consecutive_poll_errors, 5), 30)) + continue + consecutive_poll_errors = 0 + + for msg in messages: + sender_id = msg.get("sender_id", "unknown") + chat_id = msg.get("chat_id", "default") + content = msg.get("content", "") + + if not content: + continue + + logger.info( + "Processing: sender=%s chat=%s len=%d", + sender_id, + chat_id, + len(content), + ) + + # Check for reset/new-session signal + if content.strip().lower() in ("reset", "new session", "/new"): + logger.info("Reset signal: chat_id=%s", chat_id) + await self.handle_reset(chat_id) + await self.send_to_test_server(chat_id, "New session started") + continue + + # Forward to ReMe agent + reply = await self.process_message(sender_id, chat_id, content) + + if reply: + await self.send_to_test_server(chat_id, reply) + else: + logger.warning("No reply for chat_id=%s", chat_id) + await self.send_to_test_server( + chat_id, + "[Error: Agent failed to generate response]", + ) + + except KeyboardInterrupt: + logger.info("Interrupted by user") + finally: + # Save any remaining session memories + for cid in list(self.session_messages.keys()): + messages = self.session_messages.pop(cid, []) + if messages: + self._schedule_memory_save(cid, messages) + # Wait for all pending memory saves to complete + if self._pending_memory_tasks: + logger.info( + "Waiting for %d pending memory saves to complete...", + len(self._pending_memory_tasks), + ) + await self._wait_for_pending_memory_saves() + await self.stop() + + +async def main(): + """CLI entrypoint: parse arguments and run the ReMe bridge.""" + # Ignore SIGHUP to prevent bridge from being killed (same fix as qwenpaw) + signal.signal(signal.SIGHUP, signal.SIG_IGN) + + parser = argparse.ArgumentParser( + description="Bridge between Pi-Bench Test Server and ReMe agent", + ) + parser.add_argument("--test-server-url", default="http://localhost:9999") + parser.add_argument("--appworld-mcp-url", default="http://localhost:10000/mcp") + parser.add_argument( + "--reme-dir", + default="", + help="ReMe repo root. Optional when 'reme' is already importable " + "(e.g. running inside the ReMe repo with its own venv).", + ) + parser.add_argument("--data-root", default="data") + parser.add_argument("--user-id", default="researcher") + parser.add_argument("--poll-timeout", type=int, default=30) + parser.add_argument("--workspace-dir", default="") + parser.add_argument("--model-name", default="qwen3.6-plus") + parser.add_argument("--model-base-url", default="") + parser.add_argument("--model-api-key", default="") + parser.add_argument( + "--reme-port", + type=int, + default=18765, + help="Port for ReMe's internal HTTP service (must be unique per concurrently running bridge).", + ) + parser.add_argument( + "--search-limit", + type=int, + default=3, + help="Max memory chunks injected per user message.", + ) + parser.add_argument( + "--search-min-score", + type=float, + default=2.0, + help="Min BM25 score for injected memory chunks.", + ) + parser.add_argument( + "--outputs-dir", + default="", + help="Runner outputs root for tool-trace sidecar files " + "(default: /../outputs, matching the runner layout).", + ) + parser.add_argument( + "--model-id", + default="reme", + help="model_id used under outputs//...; must match " + "config/models/reme.yaml so traces align with the runner.", + ) + + args = parser.parse_args() + + bridge = ReMeBridge( + test_server_url=args.test_server_url, + appworld_mcp_url=args.appworld_mcp_url, + reme_dir=args.reme_dir, + data_root=args.data_root, + user_id=args.user_id, + poll_timeout=args.poll_timeout, + workspace_dir=args.workspace_dir, + model_name=args.model_name, + model_base_url=args.model_base_url, + model_api_key=args.model_api_key, + reme_port=args.reme_port, + search_limit=args.search_limit, + search_min_score=args.search_min_score, + outputs_dir=args.outputs_dir, + model_id=args.model_id, + ) + + await bridge.run() + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/benchmark/pibench/config/bench/evaluation/trace_history.yaml b/benchmark/pibench/config/bench/evaluation/trace_history.yaml new file mode 100644 index 00000000..2a42c6ea --- /dev/null +++ b/benchmark/pibench/config/bench/evaluation/trace_history.yaml @@ -0,0 +1,53 @@ +version: 1 + +format: + root_tag: trace + turn_tag: turn + message_tag: message + file_tag: file + tool_call_tag_prefix: tool_call + tool_result_tag_prefix: tool_result + +text_policy: + default: + truncate_chars: 1200 + mask_newlines: false + field_overrides: + files_read: + truncate_chars: 40000 + assistant_content: + truncate_chars: 40000 + tool_result_content: + truncate_chars: 40000 + +fields: + turn: + include_session_key: false + + files: + enabled: true + + messages: + enabled: true + include_message_role_attr: true + include_message_index_attr: false + include_system: false + include_user: true + include_assistant_thinking_content: false + include_assistant_thinking_reasoning: false + include_assistant_content: true + include_assistant_reasoning: false + include_assistant_tool_calls: false + require_matching_tool_call: true + + tool_calls: + include_tool_call_id: false + tools: + web_fetch: + enabled: true + include_tool_call_keys: [url] + include_tool_result: false + web_search: + enabled: true + include_tool_call_keys: [query] + include_tool_result: false diff --git a/benchmark/pibench/config/models/reme.yaml b/benchmark/pibench/config/models/reme.yaml new file mode 100644 index 00000000..4573dffc --- /dev/null +++ b/benchmark/pibench/config/models/reme.yaml @@ -0,0 +1,40 @@ +# ReMe model configuration for Pi-Bench +# Uses ReMe's AgentScope agent with Dashscope as the LLM backend + +model: + model: reme + base_url: "http://localhost:8088" + api_key: "dummy" + provider: custom + max_tokens: 16384 + max_tool_iterations: 120 + memory_window: 100 + +user_agent: + model: qwen3.8-max + base_url: "${USER_BASE_URL}" + api_key: "${USER_API_KEY}" + temperature: 0.0 + request_timeout: 360.0 + +judger: + model: qwen3.8-max + base_url: "${JUDGER_BASE_URL}" + api_key: "${JUDGER_API_KEY}" + temperature: 0.0 + request_timeout: 360.0 + +tools: + brave_search_api_key: "${BRAVE_SEARCH_API_KEY}" + web_search_max_results: 10 + +nanobot: + trace_logs_dir: "~/.nanobot/trace_logs" + workspace_dir: "~/.nanobot/workspace" + copy_task_assets_to_workspace: true + +run: + output_dir: outputs + log_level: INFO + user_mode: llm + turn_timeout: 2400.0 diff --git a/benchmark/pibench/env.sh.example b/benchmark/pibench/env.sh.example new file mode 100644 index 00000000..62c35575 --- /dev/null +++ b/benchmark/pibench/env.sh.example @@ -0,0 +1,57 @@ +#!/bin/bash +# ═══════════════════════════════════════════════════════════════════════ +# pibench evaluation suite - environment configuration template +# Usage: cp env.sh.example env.sh, then fill in the TODO items below. +# ⚠️ env.sh contains real API keys; never commit or share it +# (already excluded via .gitignore). +# ═══════════════════════════════════════════════════════════════════════ + +SUITE_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" + +# ─── TODO: π-Bench repository root ──────────────────────────────────── +# Must contain src/, data/, scripts/test_server.py, third_party/appworld +# and .venv (see README setup). +export PI_BENCH_ROOT="" + +# ─── ReMe repository ────────────────────────────────────────────────── +# Defaults to two levels above this directory (the layout this suite uses +# when placed at ReMe/benchmark/pibench); point it at the actual ReMe +# repository root if the suite lives elsewhere. +export REME_DIR="${REME_DIR:-$(cd "${SUITE_DIR}/../.." && pwd)}" + +# ─── Base model of the agent under test (LLM used by the ReMe agent) ── +export REME_MODEL_NAME="${REME_MODEL_NAME:-qwen3.6-plus}" + +# ─── LLM service endpoint (default: DashScope OpenAI-compatible; any +# OpenAI-compatible endpoint works) ──────────────────────────────── +DASHSCOPE_BASE_URL="https://dashscope.aliyuncs.com/compatible-mode/v1" +export REME_LLM_BASE_URL="${REME_LLM_BASE_URL:-${DASHSCOPE_BASE_URL}}" + +# ─── TODO: API keys ─────────────────────────────────────────────────── +# USER_API_KEY : drives the simulated user LLM (run phase; judges whether +# hidden intents are satisfied and asks follow-ups) +# JUDGER_API_KEY: drives the judger LLM (eval phase; scores the checklist) +# The two may be identical; one strong model is recommended for both. +export USER_BASE_URL="${DASHSCOPE_BASE_URL}" +export USER_API_KEY="TODO-fill-in-user-agent-api-key" + +export JUDGER_BASE_URL="${DASHSCOPE_BASE_URL}" +export JUDGER_API_KEY="TODO-fill-in-judger-api-key" + +# The ReMe agent's key reuses USER_API_KEY by default (no need to repeat +# it when both use the same service and key). +export REME_LLM_API_KEY="${REME_LLM_API_KEY:-${USER_API_KEY}}" + +# Brave Search (optional; used by the agent's web_search tool - use +# "dummy" when not needed). +export BRAVE_SEARCH_API_KEY="TODO-optional-brave-search-key-or-dummy" + +# ─── Persistent memory workspaces (one subdirectory per persona, +# created automatically) ─────────────────────────────────────────── +export REME_WORKSPACE_ROOT="${REME_WORKSPACE_ROOT:-${SUITE_DIR}/reme_workspace}" + +# ─── Variables consumed by ReMe's default.yaml model config expansion; +# do not remove ──────────────────────────────────────────────────── +export LLM_MODEL_NAME="${REME_MODEL_NAME}" +export LLM_BASE_URL="${REME_LLM_BASE_URL}" +export LLM_API_KEY="${REME_LLM_API_KEY}" diff --git a/benchmark/pibench/fix_trace_logs.py b/benchmark/pibench/fix_trace_logs.py new file mode 100755 index 00000000..2a855e17 --- /dev/null +++ b/benchmark/pibench/fix_trace_logs.py @@ -0,0 +1,198 @@ +#!/usr/bin/env python3 +"""Convert reme_eval run outputs into eval-compatible trace logs. + +outputs/{model_id}/{user_id}/{task_id}/history/{ts}-messages.jsonl + -> ~/.nanobot/trace_logs/{model_id}/{user_id}/{task_id}/{ts}/turn_N.json + +The bridge additionally writes {ts}-tools.jsonl sidecar files next to the +message histories: one JSON object per executed tool call with fields +{turn, name, arguments, result}. Each messages run is paired with the +temporally closest sidecar, and the records are merged into the generated +turn files under the "tool_steps" key, which is one of the tool-history +formats π-Bench's collect_tool_history() understands. Without this step, +tools_evaluation scripts would see no tool evidence at all. + +Usage: python fix_trace_logs.py [user_id ...] (no args = all users) +""" + +import json +import re +import sys +from datetime import datetime +from pathlib import Path + +SUITE_DIR = Path(__file__).resolve().parent +OUTPUTS_DIR = SUITE_DIR / "outputs" +TRACE_LOGS_DIR = Path.home() / ".nanobot" / "trace_logs" + +MESSAGES_FILE_RE = re.compile(r"^(\d{8}_\d{6})-messages\.jsonl$") +TOOLS_FILE_RE = re.compile(r"^(\d{8}_\d{6})-tools\.jsonl$") +TIME_FORMAT = "%Y%m%d_%H%M%S" +# A tool sidecar belongs to the messages run that started at most this many +# seconds earlier (the bridge stamps the sidecar when the task's first user +# message arrives, shortly after the runner opened the messages file). +MAX_PAIR_DELTA_SECONDS = 6 * 3600 + + +def _to_epoch(timestamp: str) -> float: + """Parse a YYYYMMDD_HHMMSS timestamp into epoch seconds.""" + try: + return datetime.strptime(timestamp, TIME_FORMAT).timestamp() + except ValueError: + return 0.0 + + +def load_tool_records(tools_file: Path) -> dict: + """Group sidecar tool records by turn number.""" + by_turn: dict = {} + try: + with open(tools_file, "r", encoding="utf-8") as f: + for line in f: + line = line.strip() + if not line: + continue + try: + record = json.loads(line) + except json.JSONDecodeError: + continue + if not isinstance(record, dict) or not record.get("name"): + continue + turn = int(record.get("turn") or 0) + by_turn.setdefault(turn, []).append( + { + "name": record["name"], + "arguments": record.get("arguments", {}), + "result": record.get("result", ""), + }, + ) + except OSError as exc: + print(f" WARNING: cannot read tool sidecar {tools_file}: {exc}") + return by_turn + + +def pair_tool_sidecars(message_runs: list, tool_runs: list) -> dict: + """Pair each messages run with the temporally closest unused tool sidecar. + + Fresh runs produce exactly one messages file and one sidecar per task; + re-runs append matching pairs, so sorted greedy nearest-timestamp + matching is stable. Sidecars farther away than MAX_PAIR_DELTA_SECONDS + (e.g. leftovers of a crashed bridge) stay unpaired. + """ + pairing: dict = {} + unused = list(tool_runs) + for msg_ts, _ in message_runs: + best_delta = None + best_item = None + for tool_ts, tool_path in unused: + delta = abs(_to_epoch(tool_ts) - _to_epoch(msg_ts)) + if best_delta is None or delta < best_delta: + best_delta = delta + best_item = (tool_ts, tool_path) + if best_delta is not None and best_item is not None and best_delta <= MAX_PAIR_DELTA_SECONDS: + pairing[msg_ts] = best_item[1] + unused.remove(best_item) + return pairing + + +def build_turns(messages: list) -> list: + """Split the flat message list into per-turn [user, assistant] groups.""" + turns = [] + i = 0 + while i < len(messages): + turn_msgs = [] + if messages[i]["role"] == "user": + turn_msgs.append({"role": "user", "content": messages[i]["message"]}) + i += 1 + if i < len(messages) and messages[i]["role"] == "assistant": + turn_msgs.append({"role": "assistant", "content": messages[i]["message"]}) + i += 1 + if not turn_msgs: + i += 1 # defensive: never spin on unexpected roles + continue + turns.append(turn_msgs) + return turns + + +def convert_task(model_id: str, user_id: str, task_dir: Path) -> None: + """Convert one task's history dir into trace turn files with tool_steps.""" + history_dir = task_dir / "history" + if not history_dir.is_dir(): + return + + message_runs = [] + tool_runs = [] + for msg_file in history_dir.glob("*-messages.jsonl"): + match = MESSAGES_FILE_RE.match(msg_file.name) + if match: + message_runs.append((match.group(1), msg_file)) + for tools_file in history_dir.glob("*-tools.jsonl"): + match = TOOLS_FILE_RE.match(tools_file.name) + if match: + tool_runs.append((match.group(1), tools_file)) + if not message_runs: + return + + message_runs.sort(key=lambda item: item[0]) + tool_runs.sort(key=lambda item: item[0]) + pairing = pair_tool_sidecars(message_runs, tool_runs) + + print(f"\n{model_id}/{user_id}/{task_dir.name}") + for timestamp, msg_file in message_runs: + trace_dir = TRACE_LOGS_DIR / model_id / user_id / task_dir.name / timestamp + trace_dir.mkdir(parents=True, exist_ok=True) + + messages = [] + with open(msg_file, "r", encoding="utf-8") as f: + for line in f: + line = line.strip() + if not line: + continue + msg = json.loads(line) + if msg.get("role") == "user" and msg.get("message") == "/new": + continue + messages.append(msg) + + tools_file = pairing.get(timestamp) + tools_by_turn = load_tool_records(tools_file) if tools_file else {} + if tools_file is not None: + print(f" {timestamp}: paired tool sidecar {tools_file.name}") + + turns = build_turns(messages) + for turn_idx, turn_msgs in enumerate(turns, start=1): + turn_data = {"messages": turn_msgs} + tool_steps = tools_by_turn.get(turn_idx) + if tool_steps: + turn_data["tool_steps"] = tool_steps + turn_file = trace_dir / f"turn_{turn_idx}.json" + with open(turn_file, "w", encoding="utf-8") as f: + json.dump(turn_data, f, indent=2, ensure_ascii=False) + tool_total = sum(len(steps) for steps in tools_by_turn.values()) + print(f" {timestamp}: {len(turns)} turns, {tool_total} tool step(s) -> {trace_dir}") + + +def convert_outputs(user_filter=None): + """Convert message history JSONL files into per-turn trace JSON files.""" + if not OUTPUTS_DIR.exists(): + print(f"outputs dir not found: {OUTPUTS_DIR}") + return + + for model_dir in sorted(OUTPUTS_DIR.iterdir()): + if not model_dir.is_dir(): + continue + model_id = model_dir.name + + for user_dir in sorted(model_dir.iterdir()): + if not user_dir.is_dir(): + continue + user_id = user_dir.name + if user_filter and user_id not in user_filter: + continue + + for task_dir in sorted(user_dir.iterdir()): + if task_dir.is_dir(): + convert_task(model_id, user_id, task_dir) + + +if __name__ == "__main__": + convert_outputs(set(sys.argv[1:]) or None) + print("\ndone") diff --git a/benchmark/pibench/resume.py b/benchmark/pibench/resume.py new file mode 100755 index 00000000..d40b4317 --- /dev/null +++ b/benchmark/pibench/resume.py @@ -0,0 +1,332 @@ +#!/usr/bin/env python3 +"""Checkpoint-resume support for the reme_eval suite. + +Completion source of truth: + - outputs/reme///history/*-log.jsonl (per-task logs, + flushed incrementally, survive mid-run kills) + - outputs/reme//run/*-log.jsonl (run-level logs, + may be truncated if the process was killed before flush) + lines: "Task finished task_id= status=" + A task counts as COMPLETED when its latest terminal status is one of + SUCCESS / MAX_TURNS / TIMEOUT. ERROR or never-started tasks stay pending. + + "Latest" is decided by EVENT TIME, not by file category or read order: + each record's "timestamp" (epoch seconds, or "timestamp_iso" as fallback) + is compared across per-task and run-level logs alike, with the timestamp + embedded in the log file name as a last-resort fallback. This keeps an + old run-level SUCCESS from overriding a newer per-task ERROR when the + re-run died before the new run-level log captured the task. + +Commands: + remaining [--json] + Print task_ids still to run, in data//episode.yaml order + (one per line; --json prints {"completed": [...], "remaining": [...]}). + + cleanup [--dry-run] + Surgically remove residual memory artifacts of tasks that are about + to be RE-RUN (i.e. pending tasks that left partial state because a + previous run was interrupted). This prevents answer leakage: an + interrupted task's conversation may already have been distilled into + daily notes during graceful shutdown, and re-running the task with + that memory injected would inflate scores. + + Removed artifacts (only for pending tasks with residual state): + - daily//.md whose frontmatter session_id matches + pibench__*, plus a refresh of ONLY the daily index of + the affected date(s) (daily/.md), matched by the full + workspace-relative note path, never by bare file name + - digest notes with matching session_id + - session/dialog/pibench__*.jsonl + - mem_session/**.jsonl files containing pibench__ + When the ReMe package is importable, the daily index refresh reuses + ReMe's own rebuild logic (reme.steps.file_io._daily_index. + refresh_day_index); otherwise index lines are dropped by exact + wikilink path match. Either way, indexes of other dates are never + touched. The ReMe watcher (init_changes_step) detects the deleted + daily notes on next bridge startup and removes them from the BM25 + index itself. + + Completed tasks' memories are NEVER touched by this command. + +Design note (resume vs memory-wipe conflict): + A full memory wipe is a suite-level action of fresh mode (run_all.sh + without --resume) and happens before any service starts. Resume mode + never wipes; it only performs the surgical cleanup above. The two modes + are mutually exclusive, so a resumed run can never lose the cross-session + memory accumulated by completed tasks. +""" + +import asyncio +import json +import os +import re +import sys +from datetime import datetime +from pathlib import Path + +import yaml + +try: # Reuse ReMe's daily-index rebuild when running inside the ReMe venv. + from reme.steps.file_io._daily_index import refresh_day_index +except ImportError: # pragma: no cover - depends on runtime venv + refresh_day_index = None + +SUITE_DIR = Path(__file__).resolve().parent +DATA_DIR = Path(os.environ.get("REME_EVAL_DATA_DIR", SUITE_DIR / "data")).resolve() +OUTPUTS_DIR = Path(os.environ.get("REME_EVAL_OUTPUTS_DIR", SUITE_DIR / "outputs")) / "reme" +WORKSPACE_ROOT = Path( + os.environ.get("REME_WORKSPACE_ROOT", SUITE_DIR / "reme_workspace"), +).resolve() + +COMPLETED_STATUSES = {"SUCCESS", "MAX_TURNS", "TIMEOUT"} +TASK_FINISHED_RE = re.compile(r"Task finished task_id=(\S+) status=(\S+)") +SESSION_ID_RE = re.compile(r"^session_id:\s*(\S+)", re.MULTILINE) +NOTE_COUNT_RE = re.compile(r"(description:\s*)\d+(\s*note\(s\) today)") +LOG_FILE_TS_RE = re.compile(r"^(\d{8}_\d{6})-log\.jsonl$") +TIME_FORMAT = "%Y%m%d_%H%M%S" + + +def log(msg: str) -> None: + """Print a status message to stderr.""" + print(msg, file=sys.stderr) + + +def episode_task_order(persona: str) -> list[str]: + """Return the ordered task ids from the persona's episode.yaml.""" + episode_path = DATA_DIR / persona / "episode.yaml" + with open(episode_path, "r", encoding="utf-8") as f: + episode = yaml.safe_load(f) + return [task["task_id"] for task in episode.get("tasks", [])] + + +def _event_time(record: dict, file_ts: str) -> float: + """Best-effort event time (epoch seconds) of one log record. + + Prefers the record's own timestamp fields; falls back to the timestamp + embedded in the log file name so that even stripped records keep a + meaningful order. Returns 0.0 when nothing is parseable. + """ + timestamp = record.get("timestamp") + if isinstance(timestamp, (int, float)) and not isinstance(timestamp, bool): + return float(timestamp) + iso = record.get("timestamp_iso") + if isinstance(iso, str): + try: + return datetime.fromisoformat(iso).timestamp() + except ValueError: + pass + if file_ts: + try: + return datetime.strptime(file_ts, TIME_FORMAT).timestamp() + except ValueError: + pass + return 0.0 + + +def latest_task_statuses(persona: str) -> dict[str, str]: + """Scan per-task and run-level logs; the newest EVENT TIME wins per task. + + Every "Task finished" record across both log categories is keyed by + (event_time, file timestamp, file order, line number); the record with + the highest key decides the task's status. File category and read order + alone can never override a newer record from the other category. + """ + persona_dir = OUTPUTS_DIR / persona + if not persona_dir.is_dir(): + return {} + + log_files = sorted(persona_dir.glob("*/history/*-log.jsonl")) + log_files += sorted(persona_dir.glob("run/*-log.jsonl")) + + best: dict[str, tuple[tuple, str]] = {} + for file_order, log_file in enumerate(log_files): + ts_match = LOG_FILE_TS_RE.match(log_file.name) + file_ts = ts_match.group(1) if ts_match else "" + try: + with open(log_file, "r", encoding="utf-8") as f: + for line_no, line in enumerate(f): + if "Task finished" not in line: + continue + try: + record = json.loads(line) + except json.JSONDecodeError: + continue + match = TASK_FINISHED_RE.search(str(record.get("message", ""))) + if not match: + continue + task_id, status = match.group(1), match.group(2) + sort_key = (_event_time(record, file_ts), file_ts, file_order, line_no) + current = best.get(task_id) + if current is None or sort_key > current[0]: + best[task_id] = (sort_key, status) + except OSError: + continue + return {task_id: status for task_id, (_, status) in best.items()} + + +def split_tasks(persona: str) -> tuple[list[str], list[str]]: + """Split the episode task order into completed and remaining tasks.""" + order = episode_task_order(persona) + statuses = latest_task_statuses(persona) + completed = [t for t in order if statuses.get(t) in COMPLETED_STATUSES] + remaining = [t for t in order if t not in set(completed)] + return completed, remaining + + +def _daily_note_session_id(note_path: Path) -> str: + try: + text = note_path.read_text(encoding="utf-8") + except OSError: + return "" + match = SESSION_ID_RE.search(text) + return match.group(1) if match else "" + + +class _WorkspaceFileStoreShim: + """Structural stand-in for ReMe's file store; only workspace_path is read.""" + + def __init__(self, workspace_path: Path): + self.workspace_path = workspace_path + + +def _refresh_daily_indexes( + workspace: Path, + removed_by_date: dict[str, set[str]], + removed: list[str], +) -> None: + """Rebuild the daily index of each affected date via ReMe's own logic.""" + for date in sorted(removed_by_date): + result = asyncio.run( + refresh_day_index(_WorkspaceFileStoreShim(workspace), date, "daily"), + ) + if result.get("error"): + log(f"[resume] WARNING: daily index refresh failed for {date}: {result['error']}") + continue + removed.append(f"daily/{date}.md (refreshed, {len(removed_by_date[date])} note(s) removed)") + + +def _strip_index_lines( + workspace: Path, + removed_by_date: dict[str, set[str]], + removed: list[str], + dry_run: bool, +) -> None: + """Fallback index edit: drop lines that reference removed notes by full + workspace-relative wikilink path, and fix the note count. Only the index + files of affected dates are touched.""" + for date in sorted(removed_by_date): + index_path = workspace / "daily" / f"{date}.md" + if not index_path.is_file(): + continue + wikilinks = [f"[[{rel_path}]]" for rel_path in sorted(removed_by_date[date])] + lines = index_path.read_text(encoding="utf-8").splitlines() + kept = [line for line in lines if not any(link in line for link in wikilinks)] + if len(kept) == len(lines): + continue + note_count = sum(1 for line in kept if line.startswith("- [[daily/")) + kept = [NOTE_COUNT_RE.sub(rf"\g<1>{note_count}\2", line) for line in kept] + removed.append(f"{index_path.relative_to(workspace)} (rewritten)") + if not dry_run: + index_path.write_text("\n".join(kept) + "\n", encoding="utf-8") + + +def cleanup_partial_memory(persona: str, remaining: list[str], dry_run: bool = False) -> list[str]: + """Remove partial memory artifacts of remaining tasks so they can be re-run cleanly.""" + workspace = WORKSPACE_ROOT / persona + removed: list[str] = [] + if not workspace.is_dir() or not remaining: + return removed + + prefixes = tuple(f"pibench_{task_id}_" for task_id in remaining) + + def act(path: Path, label: str) -> None: + removed.append(label) + if not dry_run: + path.unlink() + + # 1) daily / digest notes distilled from interrupted sessions. For daily + # notes, remember the full workspace-relative path grouped by date so only + # the affected daily indexes are refreshed below. + removed_by_date: dict[str, set[str]] = {} + for section in ("daily", "digest"): + section_root = workspace / section + if not section_root.is_dir(): + continue + for note_path in section_root.rglob("*.md"): + if note_path.parent == section_root: + continue # index files handled below + session_id = _daily_note_session_id(note_path) + if session_id.startswith(prefixes): + rel_path = note_path.relative_to(workspace).as_posix() + act(note_path, rel_path) + if section == "daily": + removed_by_date.setdefault(note_path.parent.name, set()).add(rel_path) + + # 2) daily index files: refresh only the dates that lost notes, matching + # notes by their full wikilink path instead of their bare file name. + if removed_by_date: + if dry_run: + for date in sorted(removed_by_date): + removed.append(f"daily/{date}.md (would refresh index)") + elif refresh_day_index is not None: + _refresh_daily_indexes(workspace, removed_by_date, removed) + else: + _strip_index_lines(workspace, removed_by_date, removed, dry_run) + + # 3) raw dialog logs of interrupted sessions + dialog_dir = workspace / "session" / "dialog" + if dialog_dir.is_dir(): + for task_id in remaining: + for dialog_path in dialog_dir.glob(f"pibench_{task_id}_*.jsonl"): + act(dialog_path, str(dialog_path.relative_to(workspace))) + + # 4) agent-scope session states that contain interrupted-task sessions + mem_session_dir = workspace / "mem_session" + if mem_session_dir.is_dir(): + for session_path in mem_session_dir.rglob("*.jsonl"): + try: + content = session_path.read_text(encoding="utf-8", errors="ignore") + except OSError: + continue + if any(prefix in content for prefix in prefixes): + act(session_path, str(session_path.relative_to(workspace))) + + return removed + + +def main() -> int: + """CLI entrypoint: run 'remaining' or 'cleanup' action for a persona.""" + args = sys.argv[1:] + if len(args) < 2 or args[0] not in {"remaining", "cleanup"}: + print(__doc__, file=sys.stderr) + return 2 + + command, persona = args[0], args[1] + completed, remaining = split_tasks(persona) + + if command == "remaining": + if "--json" in args: + print(json.dumps({"completed": completed, "remaining": remaining})) + else: + for task_id in remaining: + print(task_id) + log( + f"[resume] {persona}: completed={len(completed)} " + f"({', '.join(completed) if completed else '-'}) remaining={len(remaining)}", + ) + return 0 + + dry_run = "--dry-run" in args + removed = cleanup_partial_memory(persona, remaining, dry_run=dry_run) + if removed: + verb = "would remove" if dry_run else "removed" + log(f"[resume] {persona}: {verb} {len(removed)} partial-memory artifact(s):") + for item in removed: + log(f" - {item}") + else: + log(f"[resume] {persona}: no partial-memory artifacts to clean") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/benchmark/pibench/run_all.sh b/benchmark/pibench/run_all.sh new file mode 100755 index 00000000..e6ecd3b1 --- /dev/null +++ b/benchmark/pibench/run_all.sh @@ -0,0 +1,119 @@ +#!/bin/bash +# Run all 5 personas with the ReMe agent, PARALLEL at a time (default 2). +# Each persona's tasks follow data/{persona}/episode.yaml order. +# +# Usage: +# bash run_all.sh # FRESH official run: wipes ALL personas' +# # ReMe memory/outputs/trace logs first, +# # then runs everything from scratch. +# bash run_all.sh --resume # Checkpoint continuation: no wipe; every +# # persona skips already-completed tasks. +# bash run_all.sh --parallel 1 # sequential (original behavior) +# bash run_all.sh --skip-eval # run phase only +# +# Memory-wipe vs resume conflict resolution: +# The full ReMe memory wipe happens ONLY here, ONLY in fresh mode (the +# default), and ONLY before any service/bridge starts. --resume never +# wipes; run_persona.sh then additionally performs a surgical cleanup of +# residual memory belonging to interrupted (to-be-re-run) tasks, so a +# resumed run keeps all completed-task memory but never inherits a partial +# task's own answer. The two modes are mutually exclusive. +set -uo pipefail + +SUITE_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +PERSONAS=(researcher marketer law_trainee pharmacist Financier) +TRACE_ROOT="${HOME}/.nanobot/trace_logs" + +PARALLEL=2 +MODE="fresh" +PASS_ARGS=() +while [[ $# -gt 0 ]]; do + case $1 in + --parallel) + PARALLEL="${2:-}"; shift 2 || true + case "$PARALLEL" in (""|*[!0-9]*) echo "--parallel needs a positive integer"; exit 2 ;; esac + [ "$PARALLEL" -lt 1 ] && PARALLEL=1 + [ "$PARALLEL" -gt ${#PERSONAS[@]} ] && PARALLEL=${#PERSONAS[@]} + ;; + --resume) + if [ "$MODE" = "fresh_set" ]; then echo "--fresh and --resume are mutually exclusive"; exit 2; fi + MODE="resume"; shift ;; + --fresh) + if [ "$MODE" = "resume" ]; then echo "--fresh and --resume are mutually exclusive"; exit 2; fi + MODE="fresh_set"; shift ;; + --skip-eval) PASS_ARGS+=(--skip-eval); shift ;; + *) echo "Unknown option: $1"; exit 1 ;; + esac +done +[ "$MODE" = "fresh_set" ] && MODE="fresh" + +START_TS=$(date +%Y%m%d_%H%M%S) +SUMMARY_LOG="${SUITE_DIR}/logs/run_all_${START_TS}.summary" +mkdir -p "${SUITE_DIR}/logs" + +echo "############################################################" +echo "# reme_eval suite | mode=${MODE} parallel=${PARALLEL} | ${START_TS}" +echo "############################################################" + +# ─── Fresh mode: suite-level wipe BEFORE anything starts ────────────── +if [ "$MODE" = "fresh" ]; then + echo "[fresh] wiping ALL personas' memory workspaces, outputs and trace logs..." + for persona in "${PERSONAS[@]}"; do + rm -rf "${SUITE_DIR}/reme_workspace/${persona}" + rm -rf "${SUITE_DIR}/outputs/reme/${persona}" + rm -rf "${TRACE_ROOT}/reme/${persona}" + rm -rf "${SUITE_DIR}/nanobot_workspace/${persona}" + done + echo "[fresh] wipe done." +else + echo "[resume] no memory wipe; personas resume after their last completed task." +fi + +# ─── Run personas in batches of PARALLEL ────────────────────────────── +STATUS_LIST=() +ANY_FAILED=0 +OVERALL_START=$(date +%s) +TOTAL=${#PERSONAS[@]} + +for ((i = 0; i < TOTAL; i += PARALLEL)); do + BATCH=("${PERSONAS[@]:i:PARALLEL}") + BATCH_PIDS=() + BATCH_NAMES=() + echo "" + echo "============================================================" + echo "# BATCH $(( i / PARALLEL + 1 )): ${BATCH[*]} started $(date '+%F %T')" + echo "============================================================" + for persona in "${BATCH[@]}"; do + bash "${SUITE_DIR}/run_persona.sh" "${persona}" --resume ${PASS_ARGS[@]+"${PASS_ARGS[@]}"} \ + > "${SUITE_DIR}/logs/suite_${persona}.log" 2>&1 & + BATCH_PIDS+=($!) + BATCH_NAMES+=("$persona") + done + for j in $(seq 0 $(( ${#BATCH[@]} - 1 ))); do + pid=${BATCH_PIDS[$j]} + persona=${BATCH_NAMES[$j]} + if wait "$pid"; then + STATUS_LIST+=("${persona}: OK") + else + rc=$? + ANY_FAILED=1 + STATUS_LIST+=("${persona}: FAILED rc=${rc}") + echo "[run_all] ${persona} FAILED (rc=${rc}); see logs/suite_${persona}.log" + fi + done +done + +total=$(( $(date +%s) - OVERALL_START )) +echo "" +echo "================ FINAL SUMMARY (${total}s total) ================" | tee -a "${SUMMARY_LOG}" +for line in "${STATUS_LIST[@]}"; do + echo " ${line}" | tee -a "${SUMMARY_LOG}" +done +echo "Summary: ${SUMMARY_LOG}" + +if [ "${ANY_FAILED}" -ne 0 ]; then + FAILED_COUNT=$(printf '%s\n' "${STATUS_LIST[@]}" | grep -c "FAILED") + echo "[run_all] ${FAILED_COUNT} persona(s) FAILED; suite run is marked as failed." | tee -a "${SUMMARY_LOG}" + exit 1 +fi +exit 0 diff --git a/benchmark/pibench/run_persona.sh b/benchmark/pibench/run_persona.sh new file mode 100755 index 00000000..78f7d5ca --- /dev/null +++ b/benchmark/pibench/run_persona.sh @@ -0,0 +1,301 @@ +#!/bin/bash +# Run the full pi-bench evaluation for ONE persona with the ReMe agent. +# Tasks follow data/{persona}/episode.yaml order (runner-native). +# +# Usage: bash run_persona.sh [--fresh|--resume] [--skip-eval] +# +# Modes (default: --resume): +# --resume Checkpoint continuation. Never wipes memory. Tasks already +# finished (SUCCESS/MAX_TURNS/TIMEOUT in the task history logs) +# are skipped via repeated --task-id flags. Before starting, any +# residual memory of tasks that are about to be RE-RUN (partial +# sessions from an interrupted run) is surgically removed by +# resume.py cleanup, so re-runs don't inherit leaked answers. +# --fresh Wipes THIS persona's ReMe memory, outputs and trace logs first, +# then runs all tasks from scratch. +# The two flags are mutually exclusive. A full multi-persona memory wipe is a +# suite-level action of `run_all.sh` (fresh mode), never done here implicitly. +set -uo pipefail + +SUITE_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +TRACE_ROOT="${HOME}/.nanobot/trace_logs" + +# ─── External dependencies (pi-bench / ReMe are NOT bundled; see README) ── +if [ ! -f "${SUITE_DIR}/env.sh" ]; then + echo "env.sh not found. Run: cp env.sh.example env.sh (then fill in the TODO items)" + exit 1 +fi +source "${SUITE_DIR}/env.sh" + +PIBENCH_DIR="${PI_BENCH_ROOT:-}" +if [ -z "${PIBENCH_DIR}" ] || [ ! -f "${PIBENCH_DIR}/src/main.py" ]; then + echo "PI_BENCH_ROOT is unset or invalid (src/main.py not found). Set it in env.sh." + exit 1 +fi +if [ ! -x "${PIBENCH_DIR}/.venv/bin/python" ] || [ ! -x "${PIBENCH_DIR}/.venv/bin/appworld" ]; then + echo "pi-bench venv incomplete: ${PIBENCH_DIR}/.venv must provide python + appworld (see README setup)." + exit 1 +fi +if [ ! -x "${REME_DIR}/.venv/bin/python" ]; then + echo "ReMe venv not found: ${REME_DIR}/.venv/bin/python (check REME_DIR in env.sh)" + exit 1 +fi +if [ ! -e "${SUITE_DIR}/data" ]; then + echo 'Benchmark data not linked. Run: ln -s "$PI_BENCH_ROOT/data" data' + exit 1 +fi + +# ─── Pre-flight: files the runner needs before any service starts ───── +MODEL_CONFIG="${SUITE_DIR}/config/models/reme.yaml" +HISTORY_CONFIG="${SUITE_DIR}/config/bench/evaluation/trace_history.yaml" +if [ ! -f "${MODEL_CONFIG}" ]; then + echo "Model config not found: ${MODEL_CONFIG} (see README directory layout)." + exit 1 +fi +if [ ! -f "${HISTORY_CONFIG}" ]; then + echo "Trace history config not found: ${HISTORY_CONFIG}" + echo "pi-bench requires config/bench/evaluation/trace_history.yaml; see README." + exit 1 +fi + +APPWORLD_DIR="${PIBENCH_DIR}/third_party/appworld" +PI_PYTHON="${PIBENCH_DIR}/.venv/bin/python" +APPWORLD_BIN="${PIBENCH_DIR}/.venv/bin/appworld" +# resume.py runs on the ReMe venv so it can reuse ReMe's daily-index rebuild. +REME_PYTHON="${REME_DIR}/.venv/bin/python" + +PERSONA="${1:-}" +if [ -z "$PERSONA" ]; then + echo "Usage: $0 [--fresh|--resume] [--skip-eval]" + exit 1 +fi +shift + +MODE="resume" +SKIP_EVAL=false +while [[ $# -gt 0 ]]; do + case $1 in + --fresh) + if [ "$MODE" = "resume_set" ]; then echo "--fresh and --resume are mutually exclusive"; exit 2; fi + MODE="fresh"; shift ;; + --resume) + if [ "$MODE" = "fresh" ]; then echo "--fresh and --resume are mutually exclusive"; exit 2; fi + MODE="resume_set"; shift ;; + --skip-eval) SKIP_EVAL=true; shift ;; + *) echo "Unknown option: $1"; exit 1 ;; + esac +done +[ "$MODE" = "resume_set" ] && MODE="resume" + +# ─── Per-persona ports (pi-bench AGENTS.md convention) ──────────────── +# REME_PORT: ReMe's internal HTTP service; must be unique per concurrent bridge. +case "$PERSONA" in + marketer) API_PORT=9001; MCP_PORT=10001; TEST_PORT=9998; REME_PORT=18766 ;; + law_trainee) API_PORT=9002; MCP_PORT=10002; TEST_PORT=9997; REME_PORT=18767 ;; + pharmacist) API_PORT=9003; MCP_PORT=10003; TEST_PORT=9996; REME_PORT=18768 ;; + researcher) API_PORT=9004; MCP_PORT=10004; TEST_PORT=9995; REME_PORT=18765 ;; + Financier) API_PORT=9005; MCP_PORT=10005; TEST_PORT=9994; REME_PORT=18769 ;; + *) echo "Unknown persona: $PERSONA"; exit 1 ;; +esac + +API_URL="http://127.0.0.1:${API_PORT}" +MCP_URL="http://127.0.0.1:${MCP_PORT}/mcp" +TEST_URL="http://127.0.0.1:${TEST_PORT}" +LOG_DIR="${SUITE_DIR}/logs" +mkdir -p "${LOG_DIR}" + +# ─── Environment (env.sh already sourced at the top) ────────────────── +WORKSPACE_DIR="${REME_WORKSPACE_ROOT}/${PERSONA}" +NANOBOT_WORKSPACE_DIR="${SUITE_DIR}/nanobot_workspace/${PERSONA}" +mkdir -p "${WORKSPACE_DIR}" "${NANOBOT_WORKSPACE_DIR}" + +echo "=========================================" +echo "ReMe x Pi-Bench | persona=${PERSONA} | mode=${MODE}" +echo " api=${API_PORT} mcp=${MCP_PORT} test=${TEST_PORT} reme=${REME_PORT}" +echo " model=${REME_MODEL_NAME}" +echo " memory workspace=${WORKSPACE_DIR} (persistent)" +echo "=========================================" + +# ─── Fresh mode: wipe this persona's state ──────────────────────────── +if [ "$MODE" = "fresh" ]; then + echo "[fresh] wiping persona state: memory workspace, outputs, trace logs" + rm -rf "${WORKSPACE_DIR}" + rm -rf "${SUITE_DIR}/outputs/reme/${PERSONA}" + rm -rf "${TRACE_ROOT}/reme/${PERSONA}" + rm -rf "${NANOBOT_WORKSPACE_DIR}" + mkdir -p "${WORKSPACE_DIR}" "${NANOBOT_WORKSPACE_DIR}" +fi + +# ─── Resume: determine remaining tasks + clean partial memories ─────── +TASK_ARGS=() +RUN_PHASE_NEEDED=true +if [ "$MODE" = "resume" ]; then + REMAINING_JSON="$("${REME_PYTHON}" "${SUITE_DIR}/resume.py" remaining "${PERSONA}" --json)" + if [ -z "$REMAINING_JSON" ]; then + echo "Failed to compute remaining tasks"; exit 1 + fi + echo "[resume] ${REMAINING_JSON}" + REMAINING_TASKS=() + while IFS= read -r tid_line; do + [ -n "$tid_line" ] && REMAINING_TASKS+=("$tid_line") + done < <("${REME_PYTHON}" "${SUITE_DIR}/resume.py" remaining "${PERSONA}" 2>/dev/null) + if [ ${#REMAINING_TASKS[@]} -eq 0 ]; then + RUN_PHASE_NEEDED=false + echo "[resume] all tasks already completed; skipping run phase" + else + # Remove residual memory of interrupted (to-be-re-run) tasks so + # re-runs don't get their own partial answers injected. + "${REME_PYTHON}" "${SUITE_DIR}/resume.py" cleanup "${PERSONA}" + for tid in "${REMAINING_TASKS[@]}"; do + TASK_ARGS+=(--task-id "$tid") + done + echo "[resume] running ${#REMAINING_TASKS[@]} remaining task(s): ${REMAINING_TASKS[*]}" + fi +fi + +# ─── Port cleanup from previous runs ────────────────────────────────── +for port in ${API_PORT} ${MCP_PORT} ${TEST_PORT} ${REME_PORT}; do + pids=$(lsof -ti :${port} 2>/dev/null || true) + if [ -n "$pids" ]; then + echo "Killing stale processes on port ${port}: ${pids}" + kill -9 $pids 2>/dev/null || true + fi +done +sleep 2 + +PIDS=() +cleanup() { + echo "[${PERSONA}] cleaning up services..." + for pid in "${PIDS[@]:-}"; do + kill "$pid" 2>/dev/null || true + done + wait 2>/dev/null || true +} +trap cleanup EXIT INT TERM + +wait_for_service() { + local url="$1" name="$2" port="$3" timeout="${4:-180}" + echo -n " waiting for ${name}..." + local start=$(date +%s) + while true; do + if curl -sf --max-time 5 "${url}" > /dev/null 2>&1; then + echo " ready"; return 0 + fi + if [ -n "$port" ] && lsof -ti :${port} > /dev/null 2>&1; then + local elapsed=$(( $(date +%s) - start )) + if [ "$elapsed" -ge 10 ]; then echo " ready (port)"; return 0; fi + fi + if [ $(( $(date +%s) - start )) -ge "$timeout" ]; then + echo " TIMEOUT"; return 1 + fi + sleep 2 + done +} + +# ─── [1/5] AppWorld API ──────────────────────────────────────────────── +echo "[1/5] AppWorld API (:${API_PORT})" +(cd "${APPWORLD_DIR}" && exec "${APPWORLD_BIN}" serve apis --root . \ + --port ${API_PORT}) > "${LOG_DIR}/appworld_api_${PERSONA}.log" 2>&1 & +PIDS+=($!) +if ! wait_for_service "${API_URL}/docs" "AppWorld API" "${API_PORT}" 180; then + tail -20 "${LOG_DIR}/appworld_api_${PERSONA}.log"; exit 1 +fi + +# ─── [2/5] AppWorld MCP ──────────────────────────────────────────────── +echo "[2/5] AppWorld MCP (:${MCP_PORT})" +TOOLS_CONFIG="${SUITE_DIR}/data/${PERSONA}/tools.yaml" +(cd "${APPWORLD_DIR}" && exec "${APPWORLD_BIN}" serve mcp http --root . \ + --remote-apis-url "${API_URL}" --port ${MCP_PORT} \ + --tools-config-file "${TOOLS_CONFIG}") > "${LOG_DIR}/appworld_mcp_${PERSONA}.log" 2>&1 & +PIDS+=($!) +if ! wait_for_service "${MCP_URL}" "AppWorld MCP" "${MCP_PORT}" 180; then + tail -20 "${LOG_DIR}/appworld_mcp_${PERSONA}.log"; exit 1 +fi + +# ─── [3/5] Test Server ───────────────────────────────────────────────── +echo "[3/5] Test Server (:${TEST_PORT})" +PORT=${TEST_PORT} "${PI_PYTHON}" "${PIBENCH_DIR}/scripts/test_server.py" \ + > "${LOG_DIR}/test_server_${PERSONA}.log" 2>&1 & +PIDS+=($!) +if ! wait_for_service "${TEST_URL}/sent?after=-1" "Test Server" "${TEST_PORT}" 30; then + tail -20 "${LOG_DIR}/test_server_${PERSONA}.log"; exit 1 +fi + +# ─── [4/5] ReMe Bridge (ReMe venv) ───────────────────────────────────── +echo "[4/5] ReMe Bridge (reme service port ${REME_PORT})" +"${REME_DIR}/.venv/bin/python" "${SUITE_DIR}/bridge_reme.py" \ + --test-server-url "${TEST_URL}" \ + --appworld-mcp-url "${MCP_URL}" \ + --reme-dir "${REME_DIR}" \ + --data-root "${SUITE_DIR}/data" \ + --user-id "${PERSONA}" \ + --workspace-dir "${WORKSPACE_DIR}" \ + --reme-port "${REME_PORT}" \ + --model-name "${REME_MODEL_NAME}" \ + --model-base-url "${REME_LLM_BASE_URL}" \ + --model-api-key "${REME_LLM_API_KEY}" \ + > "${LOG_DIR}/bridge_${PERSONA}.log" 2>&1 & +BRIDGE_PID=$! +PIDS+=(${BRIDGE_PID}) +sleep 5 +if ! kill -0 "${BRIDGE_PID}" 2>/dev/null; then + echo "Bridge failed to start:"; tail -30 "${LOG_DIR}/bridge_${PERSONA}.log"; exit 1 +fi +for i in $(seq 1 12); do + if grep -q "Bridge started:" "${LOG_DIR}/bridge_${PERSONA}.log" 2>/dev/null; then + echo " bridge initialized"; break + fi + sleep 5 +done +grep -q "Bridge started:" "${LOG_DIR}/bridge_${PERSONA}.log" 2>/dev/null || { + echo "WARNING: bridge may not be ready:"; tail -20 "${LOG_DIR}/bridge_${PERSONA}.log"; } + +# ─── [5/5] Runner (run phase) ────────────────────────────────────────── +if [ "$RUN_PHASE_NEEDED" = true ]; then + echo "[5/5] Runner: run phase (episode order from data/${PERSONA}/episode.yaml)" + cd "${SUITE_DIR}" + BENCH_TEST_SERVER_URL="${TEST_URL}" PYTHONPATH="${PIBENCH_DIR}" \ + "${PI_PYTHON}" -m src.main \ + --model-config "${MODEL_CONFIG}" \ + --history-config-path "${HISTORY_CONFIG}" \ + --mode run --user-id "${PERSONA}" \ + --workspace-dir "${NANOBOT_WORKSPACE_DIR}" \ + ${TASK_ARGS[@]+"${TASK_ARGS[@]}"} \ + 2>&1 | tee "${LOG_DIR}/runner_run_${PERSONA}.log" + RUN_EXIT=${PIPESTATUS[0]} + if [ ${RUN_EXIT} -ne 0 ]; then + echo "Run phase failed (exit ${RUN_EXIT}). Logs: ${LOG_DIR}/" + exit ${RUN_EXIT} + fi +else + echo "[5/5] Runner: run phase skipped (all tasks completed)" +fi + +if [ "$SKIP_EVAL" = true ]; then + echo "Skipping eval (--skip-eval)" + exit 0 +fi + +# ─── Trace conversion + eval phase (always over all available traces) ── +echo "Converting trace logs..." +"${PI_PYTHON}" "${SUITE_DIR}/fix_trace_logs.py" "${PERSONA}" + +echo "Runner: eval phase" +cd "${SUITE_DIR}" +BENCH_TEST_SERVER_URL="${TEST_URL}" PYTHONPATH="${PIBENCH_DIR}" \ + "${PI_PYTHON}" -m src.main \ + --model-config "${MODEL_CONFIG}" \ + --history-config-path "${HISTORY_CONFIG}" \ + --mode eval --user-id "${PERSONA}" \ + --workspace-dir "${NANOBOT_WORKSPACE_DIR}" \ + 2>&1 | tee "${LOG_DIR}/runner_eval_${PERSONA}.log" +EVAL_EXIT=${PIPESTATUS[0]} + +echo "" +echo "=========================================" +echo "persona=${PERSONA} finished (eval exit=${EVAL_EXIT})" +echo " results : ${SUITE_DIR}/outputs/reme/${PERSONA}/" +echo " memory : ${WORKSPACE_DIR}/" +echo " logs : ${LOG_DIR}/" +echo "=========================================" +exit ${EVAL_EXIT}