scriptoriumadmin/ReMe
1
0
Fork 0
mirror of https://github.com/agentscope-ai/ReMe.git synced 2026-08-28 05:25:04 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 11

feat(benchmark): add LongMemEval golden answer validation (#335)

Browse source 
* feat(benchmark): add golden answer validation and session review for LongMemEval

- Introduce GoldenCheckStep to validate LongMemEval golden answers using structured verdicts
- Add SessionReviewStep to extract query/answer-relevant evidence from all sessions
- Implement concurrent session processing with configurable concurrency limits
- Create check_golden job configuration with lme_review and lme_judge agent wrappers
- Add Qwen3.7-plus model configuration for enhanced processing capabilities
- Include python_execute tool integration for agent-based reasoning and date validation
- Generate comprehensive JSON output with session summaries and validation verdicts
- Add run_check_golden.py script for batch processing across all LongMemEval samples
- Configure proper logging initialization with console and file output options
- Update component registry and file I/O modules to support new benchmark features

* feat(scripts): add script to summarize LongMemEval check_golden verdicts

- Parse check_golden.json files across all LongMemEval samples
- Calculate accuracy metrics for golden answers and session IDs
- Provide breakdown by question type with percentage calculations
- Add command line options for listing bad samples and JSON output
- Include progress tracking showing completed vs pending samples
- Display confidence scores and date sanity checks statistics

* refactor(benchmark): move golden check scripts to longmemeval directory

- Moved run_check_golden.py from scripts/ to benchmark/longmemeval/
- Moved stats_check_golden.py from scripts/ to benchmark/longmemeval/
- Updated path resolution to use parents[2] instead of parent.parent
- Added new --list-run-failed option to stats script
- Added logging directory constant and functions for tracking launched samples
- Enhanced stats output with launched count and run failure information
- Improved error reporting with run failure details and log file paths

* feat(benchmark): add LongMemEval agentic answer workflow with session extraction

- Add LmeAgenticAnswerStep, LmeAutoMemoryStep, and LmeExtractSessionStep to __init__.py
- Create shared helper render_with_source for displaying search results with session_id
- Implement agentic_answer step with vector_search, bm25_search, and extract_session_by_id tools
- Add auto_memory step to convert each session into search-friendly daily notes
- Create extract_session step to retrieve and analyze raw session content by session_id
- Update jinli_lme.yaml with auto_memory, vector_search, bm25_search, and agentic_answer jobs
- Configure lme_memory, lme_extract, and lme_agentic_answer agent wrappers
- Enhance search steps with include_source option to show session_id metadata
- Add proper session_id tracking and collision handling in daily note generation

* feat(benchmark): add LongMemEval agentic answer evaluation pipeline

- Add session_id tracking to agentic_answer.py result metadata
- Introduce run_agentic_answer.py driver for complete pipeline execution
- Implement auto_memory, update_index, and agentic_answer job orchestration
- Add concurrent execution with configurable limits and staggering
- Create aggregation script for collecting tool-call trails and results
- Add stats_agentic_answer.py for comprehensive result analysis
- Implement resume capability with existing output detection
- Generate aggregate.json with per-sample breakdown and tool call summaries

* feat(steps): add ClearPathsStep for cleaning workspace outputs before rebuild

- Introduce ClearPathsStep to remove stale workspace files/directories
- Add support for specifying paths and config_keys as targets to clear
- Implement safety checks to prevent deletion of files outside workspace
- Add logging for cleared paths and warnings for invalid paths
- Configure clear_paths_step in jinli_lme.yaml to clean daily_dir
- Add clear_paths_step to clean mem_answer.json before rebuilds

* feat(benchmark): add resume functionality to agentic answer runner

- Replace --force flag with --resume flag for controlling job execution
- By default every job reruns with clean rebuild behavior using config clear steps
- Add --resume option to skip samples whose output already exists and continue interrupted batches
- Update documentation to reflect new default clean rebuild behavior
- Modify job skipping logic to honor resume flag instead of force flag
- Update dry-run output to show correct todo jobs based on resume status
- Change default example command to use --resume for continuing interrupted runs

* feat(benchmark): generate JSONL output for check golden records

- Add write_check_golden_list function to create JSONL file
- Write all readable check_golden records as JSONL format
- Include check_golden_list path in stats output
- Display generated JSONL file path in summary report
- Maintain UTF-8 encoding with non-ASCII character support

* refactor(benchmark): rename answer judge step and integrate LME LLM judge

- Rename AnswerJudgeStep to LmeLlmJudgeStep and update imports
- Add new llm_judge configuration in jinli_lme.yaml
- Update run_agentic_answer.py to include llm_judge in pipeline
- Modify LmeLlmJudgeStep to read from query.json and answer.json
- Write LLM judgement results back to mem_answer.json
- Add command line options for start/end sample range selection
- Update aggregate.json generation to include LLM judgement data
- Add resume capability for llm_judge job based on judgement presence

* refactor(benchmark): rename answer judge step and integrate LME LLM judge

- Rename AnswerJudgeStep to LmeLlmJudgeStep and update imports
- Add new llm_judge configuration in jinli_lme.yaml
- Update run_agentic_answer.py to include llm_judge in pipeline
- Modify LmeLlmJudgeStep to read from query.json and answer.json
- Write LLM judgement results back to mem_answer.json
- Add command line options for start/end sample range selection
- Update aggregate.json generation to include LLM judgement data
- Add resume capability for llm_judge job based on judgement presence

* feat(steps): add wait_for_paths_step to block until workspace files exist

- Introduce WaitForPathsStep class that polls for required workspace-relative paths
- Add step registration with 'wait_for_paths_step' backend identifier
- Implement path validation to ensure targets are within workspace boundaries
- Add polling mechanism with configurable intervals via poll_seconds parameter
- Include logging functionality with log_every_seconds parameter for status updates
- Add metadata tracking of waited paths and duration in response object
- Register step in index module and expose in public API
- Configure step in jinli_lme.yaml to wait for session_review.json before golden check
- Add script rename from run_check_golden.py to run_golden_check.py with enhanced options

* feat(benchmark): enhance longmemeval benchmarking with concurrency and progress tracking

- Add benchmark extra dependency group with portalocker requirement
- Introduce concurrent execution support for golden_check and session_review workflows
- Add progress reporting interval option with real-time status updates
- Implement global throttling mechanism for session review requests using file locks
- Enhance golden check validation with current schema verification
- Add active task tracking and graceful shutdown handling
- Rename check_golden scripts to golden_check for consistency
- Update statistics reporting with correct/incorrect terminology instead of reasonable
- Add stale format detection and compatibility handling for verdict fields
- Include both_correct rate calculation in accuracy metrics
- Add concurrency and staggering options for better resource management

* ci(workflow): add Windows smoke test workflow

- Create new workflow file .github/workflows/windows-smoke.yml
- Configure workflow to trigger on push and pull request events
- Set up Python environment with version 3.11
- Install package dependencies using pip
- Run version job as smoke test for CLI functionality
- Enable concurrency control to prevent duplicate runs
- Use matrix strategy for Python version testing

* feat(benchmark): add retry mechanism and health check for session review

- Added retry configuration options (retry_initial_seconds, retry_max_seconds, retry_max_attempts) to jinli_lme.yaml
- Implemented exponential backoff retry logic with configurable parameters in session_review step
- Added output_is_healthy function to verify session_review.json integrity and absence of failed reviews
- Updated resume functionality to skip only healthy outputs instead of all existing files
- Integrated JSON parsing and validation to check for failed reviews in output files
- Enhanced error handling and logging for retry attempts and recovery scenarios

* feat(benchmark): add LongMemEval session review statistics script

- Create stats_session_review.py to summarize session_review.json artifacts
- Add command line options for listing failed, missing, and run failed samples
- Implement JSON output mode for programmatic consumption
- Calculate and display health statistics including total samples, healthy outputs, failed sessions
- Provide detailed failure information with session IDs and error messages
- Generate re-run commands for samples with failed reviews
- Add percentage calculations for better statistical overview
- Include support for multiple output formats and detailed logging

* feat(benchmark): add LongMemEval output cleanup script and enhance golden check retry logic

- Added clean_sample_outputs.py script to remove generated LongMemEval files while preserving source inputs
- Implemented configurable retry mechanism in golden_check.py with exponential backoff strategy
- Added retry parameters (initial/max seconds and max attempts) to control failure recovery behavior
- Integrated asyncio support for asynchronous sleep during retry intervals
- Configured default retry settings in jinli_lme.yaml with 5s initial and 300s maximum intervals
- Preserved core files (query.json, answer.json, session/) while cleaning generated artifacts

* feat(benchmark): add AppleDouble file cleanup to sample output cleaner

- Remove AppleDouble files starting with '._' recursively including under session/
- Add is_under helper function to check if path is inside parent directory
- Track targets in set to avoid duplicate processing
- Include AppleDouble files in cleanup targets when not already covered by existing targets
- Maintain dry-run mode as default behavior with --apply flag for actual deletion

* refactor(benchmark): update LongMemEval sample output cleaning script

- Add time and Iterator imports for enhanced functionality
- Add --progress-every argument to control progress reporting frequency
- Replace is_under function with iter_sample_targets generator
- Implement detailed progress tracking with timing measurements
- Add sample-by-sample processing with elapsed time reporting
- Include AppleDouble file detection within session directory
- Update target counting and deletion statistics display
- Add conditional progress updates based on progress-every setting
- Improve dry-run mode with would-delete indication

* chore(benchmark): increase initial interval for session review step

- Changed START_INTERVAL_SECONDS from 1.0 to 3.0 seconds
- Adjusted timing parameters for better benchmark stability

* refactor(benchmark): implement coordinated retry mechanism for session reviews

- Add retry gate condition to coordinate concurrent review attempts
- Implement wait_for_healthy_start_slot to handle sequential retries
- Create mark_retrying and mark_recovered functions to track retry states
- Update reply_with_retry to accept index parameter for coordination
- Add has_prior_retry logic to prevent race conditions during recovery
- Ensure proper cleanup of retry state on success or failure
- Maintain backward compatibility while adding coordination features

* chore(benchmark): adjust session review start interval timeout

- Changed START_INTERVAL_SECONDS from 3.0 to 5.0 seconds
- Increased initial delay for session review benchmark step
- Updated timeout configuration for improved stability

* refactor(benchmark): update session review concurrency and throttling mechanism

- Replace global throttle with per-process concurrency control
- Add concurrency parameter with default value of 30 in config
- Add start_interval_seconds parameter with default value of 2 seconds
- Change default concurrency from 3 to 1 in command line interface
- Update documentation to reflect new throttling behavior
- Implement semaphore-based concurrency limiting for review tasks
- Modify retry mechanism to use local locking instead of global files
- Remove portalocker dependency for cross-process throttling

* refactor(config): update session review configuration and concurrency settings

- Removed deprecated retry configuration parameters from jinli_lme.yaml
- Increased MAX_CONCURRENCY from 30 to 60 in session_review.py
- Reduced START_INTERVAL_SECONDS from 2.0 to 1.0 in session_review.py
- Cleaned up redundant backend specifications in configuration file
- Simplified agent wrapper configurations by removing obsolete retry settings

* feat(benchmark): enhance LME auto memory step with advanced scheduling and error handling

- Add datetime parsing functionality for LongMemEval timestamps with regex pattern
- Implement configurable concurrency limits with MAX_CONCURRENCY of 60
- Introduce retry mechanism with exponential backoff for agent interactions
- Add session filtering based on date comparison with question_date validation
- Create rate limiting with start interval control between requests
- Implement sophisticated retry coordination using asyncio conditions
- Add comprehensive error tracking for failed and filtered session extracts
- Remove deprecated concurrency parameter from jinli_lme.yaml configuration
- Add structured output validation in session review step
- Include detailed metadata reporting with session statistics and errors

* fix(benchmark): adjust default concurrency for auto_memory job

- Changed default concurrency from 3 to 1 for auto_memory job to prevent API overload
- Updated help text to reflect new default value of 1 for concurrency parameter
- Modified documentation to clarify concurrency behavior varies by job type

* refactor(search): replace hardcoded candidate multiplier with constant

- Introduced _CANDIDATE_MULTIPLIER constant set to 10
- Replaced hardcoded factor of 5 with _CANDIDATE_MULTIPLIER in BM25 search
- Replaced hardcoded factor of 5 with _CANDIDATE_MULTIPLIER in vector search
- Updated test to verify both search steps use ten times limit for candidates
- Imported VectorSearchStep and Bm25SearchStep in test module
- Added comprehensive test case for candidate count calculation logic

* feat(lme): add data inspection error handling with fallback mechanism

- Implemented non-retryable data inspection error markers detection
- Added _is_data_inspection_error method to identify inspection failures
- Created fallback handling for data inspection errors in auto memory extraction
- Added fallback handling for data inspection errors in session review
- Extended failed extracts tracking with non-retryable and fallback flags
- Separated fallback extracts from regular failed extracts in reporting
- Enhanced error logging with specific data inspection failure messages
- Updated metrics to track fallback extractions and reviews separately
- Maintained existing retry logic for other exception types

* feat(benchmark): enhance session review statistics with fallback tracking

- Add support for identifying and listing non-retryable fallback reviews
- Introduce --list-fallback argument to display fallback review details
- Separate retryable failures from non-retryable fallbacks in reporting
- Track fallback samples and sessions separately from failed ones
- Update console output to show both retryable and non-retryable categories
- Include fallback details in JSON output with reasons and session info
- Modify failure counting logic to distinguish between retryable and fallback reviews

* feat(benchmark): add question_id tracking and enhanced fallback reporting

- Add question_id function to extract query.question_id from data
- Initialize question_id_by_id dictionary to store question IDs by index
- Store question_id for each sample during data processing
- Enhance fallback output to include question IDs and session information
- Format sample labels with question IDs when available
- Display session IDs associated with each fallback case

* feat(benchmark): add question_id support and improve bad sample reporting

- Add question_id_for function to extract question_id from multiple sources
- Add sample_label function to format samples as idx(question_id) when available
- Store question_id in data dictionary during processing
- Change bad_golden and bad_sessions to store full records instead of just indices
- Update list_bad output to show formatted labels with question_id information
- Improve error reporting with more detailed sample identification

* feat(benchmark): enhance golden check stats with structured output

- Add related_session_ids function to extract session IDs from verdict records
- Create grouped_records function to group records by question type
- Replace flat list output with JSON-formatted grouped records in list_bad option
- Replace flat list output with JSON-formatted grouped records in list_bad_sessions option
- Maintain Chinese labels while adding structured data presentation
- Improve readability of bad verdict record display with hierarchical grouping

* feat(benchmark): update data structure for question indexing

- Replace sample_label with _idx field for index tracking
- Add question_id field to store _question_id values
- Maintain backward compatibility with empty string defaults
- Preserve existing session_id functionality
- Update data mapping to include new fields in grouped results

* refactor(benchmark): streamline golden answer verification process

- Replace relevance filtering with comprehensive information extraction
- Remove is_relevant field and simplify session summary structure
- Change relevant_info to extracted_info for clarity
- Update golden check logic to work with full extractions instead of filtered summaries
- Simplify prompt instructions to focus on complete information extraction
- Remove redundant schema validation and structured output requirements
- Adjust statistics calculation to match new extraction approach
- Update metadata field names to reflect extraction rather than relevance checking

* feat(benchmark): add selective file deletion option to clean_sample_outputs

- Add --filename argument to delete only specific root-level files
- Modify iter_sample_targets function to accept optional filenames filter
- Implement validation for root-level filename constraints
- Update function calls to pass filenames parameter
- Add example usage for selective file deletion in documentation

* feat(benchmark): add error count metrics to golden check statistics

- Added golden_bad, session_bad, and both_bad calculation fields
- Updated console output format to include error counts per question type
- Modified table display to show both accuracy rates and error numbers
- Enhanced statistical summary with additional error breakdown metrics

* test(search): update search step tests with include_source parameter

- Added include_source=False parameter to VectorSearchStep initialization
- Added include_source=False parameter to Bm25SearchStep initialization
- Maintained existing RuntimeContext parameters for both search steps
- Updated test calls to match new constructor signature with include_source option
This commit is contained in:
jinliyl 2026-07-13 21:26:27 +08:00 • committed by GitHub
parent 90e7adc2d2
commit bf7ca17705
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
40 changed files with 3670 additions and 102 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

38
.github/workflows/windows-smoke.yml vendored Normal file
View file

@ -0,0 +1,38 @@
name: Windows Smoke
on:
push:
branches: [main, master, dev, develop]
pull_request:
branches: [main, master, dev, develop]
workflow_dispatch:
concurrency:
group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
cancel-in-progress: true
jobs:
cli-smoke:
name: CLI smoke - py${{ matrix.python-version }}
runs-on: windows-latest
strategy:
fail-fast: false
matrix:
python-version: ["3.11"]
steps:
- uses: actions/checkout@v4
- name: Set up Python ${{ matrix.python-version }}
uses: actions/setup-python@v5
with:
python-version: ${{ matrix.python-version }}
cache: 'pip'
- name: Install package
run: |
python -m pip install --upgrade pip setuptools wheel
pip install -e ".[core,benchmark]"
- name: Run version job
run: reme start service.backend=cli job=version

141
benchmark/longmemeval/clean_sample_outputs.py Normal file
View file

@ -0,0 +1,141 @@
#!/usr/bin/env python3
"""Remove generated LongMemEval files while keeping source inputs.
For each ``datasets/longmemeval/<idx>`` workspace, this keeps only:
- query.json
- answer.json
- session/
All other files or directories in the sample root are considered generated
artifacts and can be removed. AppleDouble files whose names start with ``._``
are also removed recursively, including under ``session/``. The script is
dry-run by default; pass ``--apply`` to actually delete. To delete only specific
root-level generated files, pass one or more ``--filename`` values.
Examples:
python benchmark/longmemeval/clean_sample_outputs.py
python benchmark/longmemeval/clean_sample_outputs.py --apply
python benchmark/longmemeval/clean_sample_outputs.py --start 36 --end 79 --apply
python benchmark/longmemeval/clean_sample_outputs.py --filename check_golden.json --apply
python benchmark/longmemeval/clean_sample_outputs.py --filename session_review.json --apply
"""
import argparse
import shutil
import time
from collections.abc import Iterator
from pathlib import Path
REPO = Path(__file__).resolve().parents[2]
DATA = REPO / "datasets" / "longmemeval"
KEEP = {"query.json", "answer.json", "session"}
def parse_args() -> argparse.Namespace:
"""Parse command-line arguments."""
p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
p.add_argument("--start", type=int, default=0, help="first numeric sample id to clean, inclusive (default 0)")
p.add_argument("--end", type=int, default=499, help="last numeric sample id to clean, inclusive (default 499)")
p.add_argument("--limit", type=int, default=0, help="only clean the first N selected samples (0 = all)")
p.add_argument("--progress-every", type=int, default=25, help="print progress every N samples when applying")
p.add_argument(
"--filename",
action="append",
default=[],
help="delete only this root-level file or directory name; can be passed multiple times",
)
p.add_argument("--apply", action="store_true", help="actually delete files; default is dry-run")
return p.parse_args()
def sample_ids() -> list[str]:
"""List all numeric sample IDs."""
ids = [p.name for p in DATA.iterdir() if p.is_dir() and p.name.isdigit()]
return sorted(ids, key=int)
def delete_path(path: Path) -> None:
"""Delete a file, symlink, or directory."""
if path.is_dir() and not path.is_symlink():
shutil.rmtree(path)
else:
path.unlink()
def iter_sample_targets(sample_dir: Path, filenames: set[str] | None = None) -> Iterator[Path]:
"""Yield generated artifacts for one sample.
Root-level generated directories are yielded as a whole, so there is no
need to recurse into them. AppleDouble files are only searched inside the
kept ``session/`` directory.
"""
if filenames:
for name in sorted(filenames):
path = sample_dir / name
if path.exists():
yield path
return
for path in sorted(sample_dir.iterdir(), key=lambda p: p.name):
if path.name not in KEEP:
yield path
session_dir = sample_dir / "session"
if session_dir.is_dir():
yield from session_dir.rglob("._*")
def main() -> int:
"""Main entry point."""
args = parse_args()
if args.end < args.start:
raise ValueError(f"--end ({args.end}) must be >= --start ({args.start})")
filenames = {name.strip() for name in args.filename if name.strip()}
invalid_filenames = [name for name in filenames if Path(name).name != name]
if invalid_filenames:
raise ValueError(f"--filename only accepts root-level names, got: {invalid_filenames}")
ids = [idx for idx in sample_ids() if args.start <= int(idx) <= args.end]
if args.limit:
ids = ids[: args.limit]
total_targets = 0
deleted = 0
started_at = time.time()
for ordinal, idx in enumerate(ids, start=1):
sample_dir = DATA / idx
sample_started_at = time.time()
targets = list(iter_sample_targets(sample_dir, filenames=filenames))
total_targets += len(targets)
print(f"[sample {ordinal}/{len(ids)}] {idx} targets={len(targets)}", flush=True)
for path in targets:
if args.apply:
target_started_at = time.time()
print(f"[delete] {path}", flush=True)
delete_path(path)
deleted += 1
print(f"[deleted] {path} elapsed={time.time() - target_started_at:.1f}s", flush=True)
else:
print(f"[would-delete] {path}")
if args.apply and args.progress_every > 0 and (int(idx) + 1) % args.progress_every == 0:
elapsed = time.time() - started_at
print(
f"[progress] processed={ordinal}/{len(ids)} through={idx} " f"deleted={deleted} elapsed={elapsed:.1f}s",
flush=True,
)
print(f"[sample-done] {idx} elapsed={time.time() - sample_started_at:.1f}s", flush=True)
mode = "DELETE" if args.apply else "DRY-RUN"
print(
f"{mode} LongMemEval generated artifacts: samples={len(ids)} "
f"targets={total_targets} deleted={deleted if args.apply else 0} range={args.start}..{args.end}",
flush=True,
)
if not args.apply:
print("No files deleted. Re-run with --apply to delete these paths.", flush=True)
return 0
if __name__ == "__main__":
raise SystemExit(main())

343
benchmark/longmemeval/run_agentic_answer.py Normal file
View file

@ -0,0 +1,343 @@
#!/usr/bin/env python3
"""Drive the LongMemEval memory pipeline across all samples.
For every workspace under ``datasets/longmemeval/<idx>`` this launches one or more
``reme start config=jinli_lme job=<job>`` runs with ``LME_WORKSPACE_DIR`` pointed
at that sample. The pipeline jobs, in order, are:
1. auto_memory — distil every raw session into a daily note (``daily/*.md``)
2. update_index — clear the store and rebuild the index over ``daily/*.md``
3. agentic_answer — read ``query.json`` and answer it, writing ``mem_answer.json``
4. llm_judge — judge ``mem_answer.json`` against ``answer.json``
Pick one with ``--job``, or ``--job all`` to run the full pipeline *serially per sample*.
Runs are capped at ``--concurrency`` (default 1 for ``--job auto_memory``, otherwise
3) samples at once and each launch is staggered by ``--stagger`` seconds so they
do not all hit the LLM API at once.
By default every selected job is rerun for every sample — each job's own clear
step (configured in jinli_lme.yaml) wipes stale output first, so a run is always
a clean rebuild. Pass ``--resume`` to instead skip samples whose output already
exists (``daily/`` for auto_memory, ``metadata/embedding_store/`` for
update_index, ``mem_answer.json`` for agentic_answer, ``mem_answer.json`` with
``llm_judge.judgement`` for llm_judge) and continue an interrupted batch. Each
sample's stdout/stderr goes to ``logs/agentic_answer/<job>/<idx>.log``.
After an agentic_answer run finishes, the driver aggregates every sample's query,
golden answer, predicted answer, LLM judgement and a best-effort tool-call trail
into one big JSON at ``logs/agentic_answer/aggregate.json``.
Examples:
python benchmark/longmemeval/run_agentic_answer.py # agentic_answer, all 500, conc 3
python benchmark/longmemeval/run_agentic_answer.py --job all # full pipeline serially per sample
python benchmark/longmemeval/run_agentic_answer.py --job auto_memory # just step 1
python benchmark/longmemeval/run_agentic_answer.py --job llm_judge # just judge existing answers
python benchmark/longmemeval/run_agentic_answer.py --limit 5 --dry-run # list what would run
python benchmark/longmemeval/run_agentic_answer.py --start 187 # samples 187..499
python benchmark/longmemeval/run_agentic_answer.py --start 187 --end 499 # samples 187..499
python benchmark/longmemeval/run_agentic_answer.py --job all --resume # continue an interrupted batch
"""
import argparse
import asyncio
import json
import os
import re
import time
from pathlib import Path
REPO = Path(__file__).resolve().parents[2]
DATA = REPO / "datasets" / "longmemeval"
LOGDIR = REPO / "logs" / "agentic_answer"
AGGREGATE = LOGDIR / "aggregate.json"
# Pipeline jobs in execution order.
JOB_ORDER = ["auto_memory", "update_index", "agentic_answer", "llm_judge"]
def parse_args() -> argparse.Namespace:
"""Parse command-line arguments."""
p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
p.add_argument(
"--job",
choices=[*JOB_ORDER, "all"],
default="agentic_answer",
help="which job to run per sample; 'all' runs the full pipeline serially (default: agentic_answer)",
)
p.add_argument("--concurrency", type=int, default=1, help="max samples running at once (default 3)")
p.add_argument("--stagger", type=float, default=1.0, help="seconds between consecutive launches (default 1)")
p.add_argument("--start", type=int, default=0, help="first numeric sample id to process, inclusive (default 0)")
p.add_argument(
"--end",
type=int,
default=0,
help="last numeric sample id to process, inclusive (0 = no upper bound)",
)
p.add_argument("--limit", type=int, default=0, help="only process the first N samples (0 = all)")
p.add_argument(
"--resume",
action="store_true",
help="skip a sample when the job's output already exists (resume an interrupted run); "
"by default every selected job is rerun so the config's clear step rebuilds cleanly",
)
p.add_argument("--dry-run", action="store_true", help="list what would run, launch nothing")
p.add_argument("--no-aggregate", action="store_true", help="skip writing aggregate.json after answer/judge jobs")
return p.parse_args()
def selected_jobs(job: str) -> list[str]:
"""Expand the --job choice into an ordered list of jobs."""
return list(JOB_ORDER) if job == "all" else [job]
def sample_ids() -> list[str]:
"""List all sample IDs (numeric workspace dirs), numerically sorted."""
ids = [p.name for p in DATA.iterdir() if p.is_dir() and p.name.isdigit()]
return sorted(ids, key=int)
def job_done(idx: str, job: str) -> bool:
"""Return True when ``job``'s expected output already exists for sample ``idx``."""
ws = DATA / idx
if job == "auto_memory":
daily = ws / "daily"
return daily.is_dir() and any(daily.rglob("*.md"))
if job == "update_index":
store = ws / "metadata" / "embedding_store"
return store.is_dir() and any(store.iterdir())
if job == "agentic_answer":
return (ws / "mem_answer.json").exists()
if job == "llm_judge":
judge = _load_json(ws / "mem_answer.json").get("llm_judge")
return isinstance(judge, dict) and bool(str(judge.get("judgement") or "").strip())
raise ValueError(f"unknown job: {job}")
async def run_job(idx: str, job: str, counters: dict) -> bool:
"""Run a single job for a single sample. Returns True on success."""
log = LOGDIR / job / f"{idx}.log"
log.parent.mkdir(parents=True, exist_ok=True)
env = dict(os.environ, LME_WORKSPACE_DIR=f"datasets/longmemeval/{idx}")
started = time.strftime("%H:%M:%S")
print(f"[start {started}] {idx}/{job}", flush=True)
with log.open("w", encoding="utf-8") as f:
proc = await asyncio.create_subprocess_exec(
"reme",
"start",
"config=jinli_lme",
f"job={job}",
cwd=str(REPO),
env=env,
stdout=f,
stderr=asyncio.subprocess.STDOUT,
)
rc = await proc.wait()
ok = rc == 0 and job_done(idx, job)
counters["done" if ok else "fail"] += 1
tag = "done" if ok else "fail"
print(f"[{tag}] {idx}/{job} rc={rc} ({counters['done']} done / {counters['fail']} fail)", flush=True)
return ok
async def run_one(idx: str, jobs: list[str], sem: asyncio.Semaphore, resume: bool, counters: dict) -> None:
"""Run the selected jobs for one sample, serially.
By default every selected job is rerun (the job's own clear step wipes stale
output first). With ``resume`` a job is skipped when its output already
exists, so an interrupted batch can continue without redoing finished work.
"""
async with sem:
for job in jobs:
if resume and job_done(idx, job):
counters["skip"] += 1
print(f"[skip] {idx}/{job} (output exists)", flush=True)
continue
ok = await run_job(idx, job, counters)
if not ok:
# Later jobs depend on earlier ones; don't waste a run on a broken workspace.
print(f"[abort] {idx}: {job} failed, skipping remaining jobs", flush=True)
break
# --------------------------------------------------------------------------- #
# Aggregation of agentic_answer results into one big JSON.
# --------------------------------------------------------------------------- #
# Match ``session_id=abc123`` headers and ``"...session_id": "abc123"`` fields in
# tool-result text, so we can list which sessions each search actually surfaced.
_SID_RE = re.compile(r'session_id["\s:=]+"?([A-Za-z0-9_\-]+)')
def _load_json(path: Path) -> dict:
"""Load a JSON object, returning {} on any error."""
try:
with path.open(encoding="utf-8") as f:
data = json.load(f)
return data if isinstance(data, dict) else {}
except (OSError, json.JSONDecodeError):
return {}
def parse_tool_calls(idx: str, session_id: str) -> list[dict]:
"""Best-effort: parse the agent trajectory into an ordered tool-call summary.
Reads ``mem_session/agentscope/<session_id>.jsonl`` — the trajectory the
agentic_answer run dumped — and pairs every ``tool_call`` (name + parsed
args) with the ``session_id`` hits found in its ``tool_result``. Returns an
empty list if the file is missing or unreadable (never raises).
"""
if not session_id:
return []
path = DATA / idx / "mem_session" / "agentscope" / f"{session_id}.jsonl"
if not path.exists():
return []
calls: dict[str, dict] = {}
order: list[str] = []
try:
for line in path.read_text(encoding="utf-8").splitlines():
line = line.strip()
if not line:
continue
try:
msg = json.loads(line)
except json.JSONDecodeError:
continue
for c in msg.get("content") or []:
if not isinstance(c, dict):
continue
cid = c.get("id")
if c.get("type") == "tool_call" and cid:
try:
args = json.loads(c.get("input") or "{}")
except (json.JSONDecodeError, TypeError):
args = c.get("input")
calls[cid] = {"name": c.get("name"), "args": args, "hit_session_ids": []}
order.append(cid)
elif c.get("type") == "tool_result" and cid in calls:
text = ""
for o in c.get("output") or []:
if isinstance(o, dict) and isinstance(o.get("text"), str):
text += o["text"]
hits = list(dict.fromkeys(_SID_RE.findall(text)))
calls[cid]["hit_session_ids"] = hits
except OSError:
return []
return [{"iter": i + 1, **calls[cid]} for i, cid in enumerate(order)]
def build_record(idx: str) -> dict:
"""Assemble one sample's aggregate record from its on-disk artifacts."""
ws = DATA / idx
query = _load_json(ws / "query.json")
golden = _load_json(ws / "answer.json")
mem = _load_json(ws / "mem_answer.json")
pred = str(mem.get("answer") or "").strip()
session_id = str(mem.get("session_id") or "")
llm_judge = mem.get("llm_judge") if isinstance(mem.get("llm_judge"), dict) else {}
tool_calls = parse_tool_calls(idx, session_id) if mem else []
if not mem:
status = "missing"
elif not pred:
status = "empty"
elif "not provided" in pred.lower():
status = "not_provided"
else:
status = "answered"
return {
"idx": idx,
"question_id": query.get("question_id"),
"question_type": query.get("question_type"),
"question": query.get("question"),
"question_date": query.get("question_date"),
"golden_answer": golden.get("answer"),
"golden_answer_session_ids": golden.get("answer_session_ids"),
"pred_answer": pred,
"session_id": session_id,
"status": status,
"llm_judge": llm_judge.get("judgement"),
"llm_judge_raw": llm_judge.get("raw_judgement"),
"num_tool_calls": len(tool_calls),
"tool_calls": tool_calls,
}
def write_aggregate(ids: list[str]) -> None:
"""Aggregate every sample's agentic_answer artifacts into one big JSON."""
records = [build_record(idx) for idx in ids]
finished = [r for r in records if r["status"] != "missing"]
by_status: dict[str, int] = {}
by_llm_judge: dict[str, int] = {}
for r in records:
by_status[r["status"]] = by_status.get(r["status"], 0) + 1
judgement = r.get("llm_judge") or "missing"
by_llm_judge[judgement] = by_llm_judge.get(judgement, 0) + 1
payload = {
"generated_at": time.strftime("%Y-%m-%d %H:%M:%S"),
"total": len(records),
"finished": len(finished),
"by_status": by_status,
"by_llm_judge": by_llm_judge,
"samples": records,
}
AGGREGATE.parent.mkdir(parents=True, exist_ok=True)
AGGREGATE.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
print(f"[aggregate] wrote {len(records)} samples ({len(finished)} finished) -> {AGGREGATE}", flush=True)
async def main() -> int:
"""Run the driver."""
args = parse_args()
LOGDIR.mkdir(parents=True, exist_ok=True)
jobs = selected_jobs(args.job)
ids = sample_ids()
if args.end and args.end < args.start:
raise ValueError(f"--end ({args.end}) must be >= --start ({args.start})")
ids = [i for i in ids if int(i) >= args.start and (not args.end or int(i) <= args.end)]
if args.limit:
ids = ids[: args.limit]
# Without --resume every job reruns; with --resume, jobs whose output exists are skipped.
def todo_jobs(i: str) -> list[str]:
return [j for j in jobs if not (args.resume and job_done(i, j))]
pending = [i for i in ids if todo_jobs(i)]
print(
f"jobs={jobs} resume={args.resume} samples total={len(ids)} pending={len(pending)} "
f"concurrency={args.concurrency} stagger={args.stagger}s",
flush=True,
)
if args.dry_run:
for i in pending:
print(f"[would-run] {i}: {todo_jobs(i)}")
return 0
sem = asyncio.Semaphore(args.concurrency)
counters = {"done": 0, "fail": 0, "skip": 0}
tasks: list[asyncio.Task] = []
for n, idx in enumerate(ids):
if n and args.stagger > 0:
await asyncio.sleep(args.stagger) # stagger each launch relative to the previous
tasks.append(asyncio.create_task(run_one(idx, jobs, sem, args.resume, counters)))
await asyncio.gather(*tasks, return_exceptions=True)
print(
f"ALL FINISHED done={counters['done']} fail={counters['fail']} skip={counters['skip']}",
flush=True,
)
if any(j in jobs for j in ("agentic_answer", "llm_judge")) and not args.no_aggregate:
write_aggregate(ids)
return 0 if counters["fail"] == 0 else 1
if __name__ == "__main__":
raise SystemExit(asyncio.run(main()))

216
benchmark/longmemeval/run_golden_check.py Normal file
View file

@ -0,0 +1,216 @@
#!/usr/bin/env python3
"""Run LongMemEval ``golden_check`` concurrently across samples.
For every workspace under ``datasets/longmemeval/<idx>`` in the selected numeric
range, this launches:
reme start config=jinli_lme job=golden_check
with ``LME_WORKSPACE_DIR`` pointed at that sample. Multiple samples can run at
once, capped by ``--concurrency``. The ``golden_check`` job itself waits for
``session_review.json`` when configured with ``wait_for_paths_step`` in
``jinli_lme.yaml``. Each sample's stdout/stderr goes to
``logs/golden_check/<idx>.log``.
By default the script processes samples 0..499 inclusive and reruns every sample
in that range. Pass ``--resume`` to skip samples whose ``check_golden.json``
already exists.
Examples:
python benchmark/longmemeval/run_golden_check.py
python benchmark/longmemeval/run_golden_check.py --start 187 --end 499
python benchmark/longmemeval/run_golden_check.py --concurrency 8 --stagger 1
python benchmark/longmemeval/run_golden_check.py --progress-interval 10
python benchmark/longmemeval/run_golden_check.py --resume
python benchmark/longmemeval/run_golden_check.py --limit 5 --dry-run
"""
import argparse
import asyncio
import json
import os
import time
from pathlib import Path
REPO = Path(__file__).resolve().parents[2]
DATA = REPO / "datasets" / "longmemeval"
LOGDIR = REPO / "logs" / "golden_check"
OUTPUT_FILENAME = "check_golden.json"
def parse_args() -> argparse.Namespace:
"""Parse command-line arguments."""
p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
p.add_argument("--start", type=int, default=0, help="first numeric sample id to process, inclusive (default 0)")
p.add_argument("--end", type=int, default=499, help="last numeric sample id to process, inclusive (default 499)")
p.add_argument("--limit", type=int, default=0, help="only process the first N selected samples (0 = all)")
p.add_argument("--concurrency", type=int, default=3, help="max samples running at once (default 3)")
p.add_argument("--stagger", type=float, default=1.0, help="seconds between consecutive launches (default 1)")
p.add_argument(
"--progress-interval",
type=float,
default=30.0,
help="seconds between progress reports while running (0 = disabled, default 30)",
)
p.add_argument(
"--resume",
action="store_true",
help=f"skip samples whose {OUTPUT_FILENAME} already exists",
)
p.add_argument("--dry-run", action="store_true", help="list what would run, launch nothing")
return p.parse_args()
def sample_ids() -> list[str]:
"""List all sample IDs (numeric workspace dirs), numerically sorted."""
ids = [p.name for p in DATA.iterdir() if p.is_dir() and p.name.isdigit()]
return sorted(ids, key=int)
def output_is_current(idx: str) -> bool:
"""Return True when the sample already has a current-schema golden-check artifact."""
path = DATA / idx / OUTPUT_FILENAME
try:
with path.open(encoding="utf-8") as f:
data = json.load(f)
except (OSError, json.JSONDecodeError):
return False
verdict = data.get("verdict") if isinstance(data, dict) else None
if not isinstance(verdict, dict):
return False
return isinstance(verdict.get("golden_answer_correct"), bool) and isinstance(
verdict.get("answer_session_ids_correct"),
bool,
)
def print_progress(counters: dict, active: set[str], selected_total: int, started_at: float) -> None:
"""Print a one-line progress snapshot."""
finished = counters["done"] + counters["fail"] + counters["skip"]
running = len(active)
outstanding = max(selected_total - finished - running, 0)
elapsed = time.monotonic() - started_at
print(
f"[progress] selected={selected_total} done={counters['done']} fail={counters['fail']} "
f"skip={counters['skip']} running={running} outstanding={outstanding} "
f"elapsed={elapsed:.0f}s",
flush=True,
)
async def progress_reporter(
counters: dict,
active: set[str],
selected_total: int,
started_at: float,
interval: float,
stop: asyncio.Event,
) -> None:
"""Periodically report progress until ``stop`` is set."""
if interval <= 0:
return
while not stop.is_set():
try:
await asyncio.wait_for(stop.wait(), timeout=interval)
except asyncio.TimeoutError:
print_progress(counters, active, selected_total, started_at)
async def run_one(idx: str, sem: asyncio.Semaphore, resume: bool, counters: dict, active: set[str]) -> None:
"""Run ``golden_check`` for one sample."""
if resume and output_is_current(idx):
counters["skip"] += 1
print(f"[skip] {idx} ({OUTPUT_FILENAME} exists)", flush=True)
return
async with sem:
active.add(idx)
log = LOGDIR / f"{idx}.log"
log.parent.mkdir(parents=True, exist_ok=True)
env = dict(os.environ, LME_WORKSPACE_DIR=f"datasets/longmemeval/{idx}")
started = time.strftime("%H:%M:%S")
print(f"[start {started}] {idx}", flush=True)
try:
with log.open("w", encoding="utf-8") as f:
proc = await asyncio.create_subprocess_exec(
"reme",
"start",
"config=jinli_lme",
"job=golden_check",
cwd=str(REPO),
env=env,
stdout=f,
stderr=asyncio.subprocess.STDOUT,
)
rc = await proc.wait()
ok = rc == 0 and output_is_current(idx)
counters["done" if ok else "fail"] += 1
tag = "done" if ok else "fail"
print(
f"[{tag}] {idx} rc={rc} log={log} ({counters['done']} done / {counters['fail']} fail)",
flush=True,
)
finally:
active.discard(idx)
async def main() -> int:
"""Run the concurrent driver."""
args = parse_args()
if args.end < args.start:
raise ValueError(f"--end ({args.end}) must be >= --start ({args.start})")
if args.concurrency < 1:
raise ValueError("--concurrency must be >= 1")
if args.progress_interval < 0:
raise ValueError("--progress-interval must be >= 0")
LOGDIR.mkdir(parents=True, exist_ok=True)
ids = [i for i in sample_ids() if args.start <= int(i) <= args.end]
if args.limit:
ids = ids[: args.limit]
pending = [i for i in ids if not (args.resume and output_is_current(i))]
print(
f"job=golden_check samples total={len(ids)} pending={len(pending)} "
f"range={args.start}..{args.end} resume={args.resume} "
f"concurrency={args.concurrency} stagger={args.stagger}s",
flush=True,
)
if args.dry_run:
for idx in pending:
print(f"[would-run] {idx}")
return 0
sem = asyncio.Semaphore(args.concurrency)
counters = {"done": 0, "fail": 0, "skip": 0}
active: set[str] = set()
started_at = time.monotonic()
stop_progress = asyncio.Event()
progress_task = asyncio.create_task(
progress_reporter(counters, active, len(ids), started_at, args.progress_interval, stop_progress),
)
tasks: list[asyncio.Task] = []
try:
for n, idx in enumerate(ids):
if n and args.stagger > 0:
await asyncio.sleep(args.stagger)
tasks.append(asyncio.create_task(run_one(idx, sem, args.resume, counters, active)))
await asyncio.gather(*tasks)
finally:
stop_progress.set()
await progress_task
print_progress(counters, active, len(ids), started_at)
print(
f"ALL FINISHED done={counters['done']} fail={counters['fail']} skip={counters['skip']}",
flush=True,
)
return 0 if counters["fail"] == 0 else 1
if __name__ == "__main__":
raise SystemExit(asyncio.run(main()))

203
benchmark/longmemeval/run_session_review.py Normal file
View file

@ -0,0 +1,203 @@
#!/usr/bin/env python3
"""Run LongMemEval ``session_review`` concurrently across samples.
For every workspace under ``datasets/longmemeval/<idx>`` in the selected numeric
range, this launches:
reme start config=jinli_lme job=session_review
with ``LME_WORKSPACE_DIR`` pointed at that sample. Multiple samples can run at
once, capped by ``--concurrency``. By default this runner launches one sample at
a time; request submission is throttled inside each ``session_review`` process.
Each sample's stdout/stderr goes to ``logs/session_review/<idx>.log``.
By default the script processes samples 0..499 inclusive and reruns every sample
in that range. Pass ``--resume`` to skip samples whose ``session_review.json``
already exists.
Examples:
python benchmark/longmemeval/run_session_review.py
python benchmark/longmemeval/run_session_review.py --start 187 --end 499
python benchmark/longmemeval/run_session_review.py --concurrency 2
python benchmark/longmemeval/run_session_review.py --resume
python benchmark/longmemeval/run_session_review.py --limit 5 --dry-run
"""
import argparse
import asyncio
import json
import os
import time
from pathlib import Path
REPO = Path(__file__).resolve().parents[2]
DATA = REPO / "datasets" / "longmemeval"
LOGDIR = REPO / "logs" / "session_review"
OUTPUT_FILENAME = "session_review.json"
def parse_args() -> argparse.Namespace:
"""Parse command-line arguments."""
p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
p.add_argument("--start", type=int, default=0, help="first numeric sample id to process, inclusive (default 0)")
p.add_argument("--end", type=int, default=499, help="last numeric sample id to process, inclusive (default 499)")
p.add_argument("--limit", type=int, default=0, help="only process the first N selected samples (0 = all)")
p.add_argument("--concurrency", type=int, default=1, help="max samples running at once (default 1)")
p.add_argument("--stagger", type=float, default=1.0, help="seconds between worker launches (default 1)")
p.add_argument(
"--resume",
action="store_true",
help=f"skip samples whose {OUTPUT_FILENAME} already exists",
)
p.add_argument("--dry-run", action="store_true", help="list what would run, launch nothing")
p.add_argument("--stop-on-fail", action="store_true", help="stop immediately after the first failed sample")
return p.parse_args()
def sample_ids() -> list[str]:
"""List all sample IDs (numeric workspace dirs), numerically sorted."""
ids = [p.name for p in DATA.iterdir() if p.is_dir() and p.name.isdigit()]
return sorted(ids, key=int)
def output_exists(idx: str) -> bool:
"""Return True when the sample already has a session review artifact."""
return (DATA / idx / OUTPUT_FILENAME).exists()
def output_is_healthy(idx: str) -> bool:
"""Return True when ``session_review.json`` exists and has no failed reviews."""
path = DATA / idx / OUTPUT_FILENAME
if not path.exists():
return False
try:
with path.open(encoding="utf-8") as f:
data = json.load(f)
except (OSError, json.JSONDecodeError):
return False
review = data.get("review") if isinstance(data, dict) else None
if not isinstance(review, dict):
return False
raw = review.get("num_failed_reviews")
if isinstance(raw, int):
return raw == 0
failed_reviews = review.get("failed_reviews")
return not failed_reviews
async def run_one(idx: str, active: set[str]) -> bool:
"""Run ``session_review`` for one sample. Returns True on success."""
log = LOGDIR / f"{idx}.log"
log.parent.mkdir(parents=True, exist_ok=True)
env = dict(os.environ, LME_WORKSPACE_DIR=f"datasets/longmemeval/{idx}")
started = time.strftime("%H:%M:%S")
print(f"[start {started}] {idx}", flush=True)
active.add(idx)
try:
with log.open("w", encoding="utf-8") as f:
proc = await asyncio.create_subprocess_exec(
"reme",
"start",
"config=jinli_lme",
"job=session_review",
cwd=str(REPO),
env=env,
stdout=f,
stderr=asyncio.subprocess.STDOUT,
)
rc = await proc.wait()
finally:
active.discard(idx)
ok = rc == 0 and output_exists(idx)
tag = "done" if ok else "fail"
print(f"[{tag}] {idx} rc={rc} log={log}", flush=True)
return ok
async def worker(
name: int,
queue: asyncio.Queue[str],
args: argparse.Namespace,
counters: dict[str, int],
active: set[str],
stop: asyncio.Event,
) -> None:
"""Run samples from ``queue`` until exhausted or fail-fast is triggered."""
if name and args.stagger > 0:
await asyncio.sleep(args.stagger * name)
while not stop.is_set():
try:
idx = queue.get_nowait()
except asyncio.QueueEmpty:
return
try:
if args.resume and output_is_healthy(idx):
counters["skip"] += 1
print(f"[skip] {idx} (healthy {OUTPUT_FILENAME} exists)", flush=True)
continue
if await run_one(idx, active):
counters["done"] += 1
else:
counters["fail"] += 1
if args.stop_on_fail:
stop.set()
finally:
queue.task_done()
async def main() -> int:
"""Run the concurrent driver."""
args = parse_args()
if args.end < args.start:
raise ValueError(f"--end ({args.end}) must be >= --start ({args.start})")
if args.concurrency < 1:
raise ValueError("--concurrency must be >= 1")
if args.stagger < 0:
raise ValueError("--stagger must be >= 0")
LOGDIR.mkdir(parents=True, exist_ok=True)
ids = [i for i in sample_ids() if args.start <= int(i) <= args.end]
if args.limit:
ids = ids[: args.limit]
pending = [i for i in ids if not (args.resume and output_exists(i))]
print(
f"job=session_review samples total={len(ids)} pending={len(pending)} "
f"range={args.start}..{args.end} resume={args.resume} "
f"concurrency={args.concurrency} stagger={args.stagger}s",
flush=True,
)
if args.dry_run:
for idx in pending:
print(f"[would-run] {idx}")
return 0
counters: dict[str, int] = {"done": 0, "fail": 0, "skip": 0}
active: set[str] = set()
stop = asyncio.Event()
queue: asyncio.Queue[str] = asyncio.Queue()
for idx in ids:
queue.put_nowait(idx)
workers = [
asyncio.create_task(worker(n, queue, args, counters, active, stop))
for n in range(min(args.concurrency, len(ids)))
]
await asyncio.gather(*workers)
print(
f"ALL FINISHED done={counters['done']} fail={counters['fail']} skip={counters['skip']}",
flush=True,
)
return 0 if counters["fail"] == 0 else 1
if __name__ == "__main__":
raise SystemExit(asyncio.run(main()))

202
benchmark/longmemeval/stats_agentic_answer.py Normal file
View file

@ -0,0 +1,202 @@
#!/usr/bin/env python3
"""Summarise the ``agentic_answer`` results across all LongMemEval samples.
Reports progress (how many of the 500 samples produced ``mem_answer.json``) and a
breakdown of answer *status*:
- answered — a non-empty answer that is not "not provided";
- not_provided — the agent gave up ("not provided");
- empty — ``mem_answer.json`` exists but the answer is blank;
- missing — no ``mem_answer.json`` yet.
Everything is broken down by ``question_type``. This script does NOT judge answer
correctness (there is no grader for ``mem_answer`` yet) — it only tracks progress
and collects predicted-vs-golden pairs. Tool-call statistics are read from the
aggregate written by ``run_agentic_answer.py`` when it is present.
Examples:
python benchmark/longmemeval/stats_agentic_answer.py
python benchmark/longmemeval/stats_agentic_answer.py --list-run-failed
python benchmark/longmemeval/stats_agentic_answer.py --list-unanswered
python benchmark/longmemeval/stats_agentic_answer.py --json
"""
import argparse
import json
from collections import defaultdict
from pathlib import Path
REPO = Path(__file__).resolve().parents[2]
DATA = REPO / "datasets" / "longmemeval"
LOGBASE = REPO / "logs" / "agentic_answer"
AGGREGATE = LOGBASE / "aggregate.json"
def parse_args() -> argparse.Namespace:
"""Parse command-line arguments."""
p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
p.add_argument("--list-unanswered", action="store_true", help="list samples answered 'not provided' or empty")
p.add_argument("--list-run-failed", action="store_true", help="list launched samples with no readable output")
p.add_argument("--json", action="store_true", help="emit the summary as JSON")
return p.parse_args()
def sample_ids() -> list[str]:
"""List all sample IDs (numeric workspace dirs), numerically sorted."""
ids = [p.name for p in DATA.iterdir() if p.is_dir() and p.name.isdigit()]
return sorted(ids, key=int)
def pct(num: int, den: int) -> str:
"""Format a percentage."""
return f"{(100.0 * num / den):.1f}%" if den else "n/a"
def logged_sample_ids() -> list[str]:
"""List sample IDs that have an agentic_answer launch log."""
logdir = LOGBASE / "agentic_answer"
if not logdir.exists():
return []
ids = [p.stem for p in logdir.glob("*.log") if p.stem.isdigit()]
return sorted(ids, key=int)
def answer_status(pred: str, has_file: bool) -> str:
"""Classify an answer into answered / not_provided / empty / missing."""
if not has_file:
return "missing"
if not pred:
return "empty"
if "not provided" in pred.lower():
return "not_provided"
return "answered"
def load_tool_calls() -> dict[str, int]:
"""Map idx -> num_tool_calls from the aggregate, if it exists."""
if not AGGREGATE.exists():
return {}
try:
with AGGREGATE.open(encoding="utf-8") as f:
data = json.load(f)
except (OSError, json.JSONDecodeError):
return {}
return {s["idx"]: s.get("num_tool_calls", 0) for s in data.get("samples", []) if "idx" in s}
def main() -> int:
"""Main entry point."""
args = parse_args()
ids = sample_ids()
total = len(ids)
tool_calls = load_tool_calls()
rows, unreadable = [], []
finished_ids = set()
for idx in ids:
query_path = DATA / idx / "query.json"
mem_path = DATA / idx / "mem_answer.json"
qtype = "(unknown)"
try:
with query_path.open(encoding="utf-8") as f:
qtype = json.load(f).get("question_type") or "(unknown)"
except (OSError, json.JSONDecodeError):
pass
has_file = mem_path.exists()
pred = ""
if has_file:
try:
with mem_path.open(encoding="utf-8") as f:
pred = str(json.load(f).get("answer") or "").strip()
finished_ids.add(idx)
except (OSError, json.JSONDecodeError):
unreadable.append(idx)
has_file = False
rows.append({"idx": idx, "type": qtype, "status": answer_status(pred, has_file)})
finished = [r for r in rows if r["status"] != "missing"]
n = len(finished)
launched = logged_sample_ids()
run_failed = [idx for idx in launched if idx not in finished_ids]
# Overall status tallies.
status_counts: dict[str, int] = defaultdict(int)
for r in rows:
status_counts[r["status"]] += 1
answered = status_counts["answered"]
unanswered = [r["idx"] for r in rows if r["status"] in ("not_provided", "empty")]
calls_vals = [tool_calls[i] for i in finished_ids if i in tool_calls]
avg_calls = sum(calls_vals) / len(calls_vals) if calls_vals else 0.0
# Per question_type breakdown.
by_type: dict[str, dict[str, int]] = defaultdict(lambda: {"n": 0, "answered": 0})
for r in finished:
by_type[r["type"]]["n"] += 1
by_type[r["type"]]["answered"] += 1 if r["status"] == "answered" else 0
if args.json:
print(
json.dumps(
{
"total": total,
"finished": n,
"pending": total - n - len(unreadable),
"unreadable": unreadable,
"launched": len(launched),
"run_failed": run_failed,
"status_counts": dict(status_counts),
"answered_rate": round(answered / n, 4) if n else None,
"avg_tool_calls": round(avg_calls, 2) if calls_vals else None,
"by_type": {
t: {**c, "answered_rate": round(c["answered"] / c["n"], 4)} for t, c in by_type.items()
},
"unanswered": unanswered,
"aggregate": str(AGGREGATE) if AGGREGATE.exists() else None,
},
ensure_ascii=False,
indent=2,
),
)
return 0
print("=" * 60)
print("LongMemEval agentic_answer 统计")
print("=" * 60)
print(f"样例总数 : {total}")
print(f"已完成 (有产出) : {n} ({pct(n, total)})")
print(f"未完成 : {total - n - len(unreadable)}")
if unreadable:
print(f"损坏/无法解析 : {len(unreadable)} {unreadable}")
print(f"已启动过 (有 log) : {len(launched)}")
print(f"运行失败/无可读产出 : {len(run_failed)}")
print("-" * 60)
print(f"已作答 (非 not provided): {answered} ({pct(answered, n)} of finished)")
print(f" 其中 not provided : {status_counts['not_provided']}")
print(f" 其中 空答案 : {status_counts['empty']}")
if calls_vals:
print(f"平均工具调用次数 : {avg_calls:.1f} (来自 {AGGREGATE.name})")
else:
print("平均工具调用次数 : n/a (先跑 run_agentic_answer.py 生成 aggregate.json)")
print("-" * 60)
print("按 question_type:")
print(f" {'type':<24} {'n':>4} {'已作答率':>12}")
for t in sorted(by_type):
c = by_type[t]
print(f" {t:<24} {c['n']:>4} {pct(c['answered'], c['n']):>12}")
if args.list_unanswered:
print("-" * 60)
print(f"not provided / 空答案的样例 ({len(unanswered)}): {unanswered}")
if args.list_run_failed:
print("-" * 60)
print(f"运行失败/无可读 mem_answer.json 的样例 ({len(run_failed)}): {run_failed}")
for idx in run_failed:
print(f" {idx}: {LOGBASE / 'agentic_answer' / f'{idx}.log'}")
print("=" * 60)
return 0
if __name__ == "__main__":
raise SystemExit(main())

344
benchmark/longmemeval/stats_golden_check.py Normal file
View file

@ -0,0 +1,344 @@
#!/usr/bin/env python3
"""Summarise the ``check_golden.json`` verdicts across all LongMemEval samples.
Reports progress (how many of the 500 samples have finished) and accuracy:
- golden answer accuracy = share of finished samples whose golden answer the
auditor judged correct (``verdict.golden_answer_correct``);
- answer_session_ids accuracy = share whose claimed answer sessions the auditor
judged exactly correct (``verdict.answer_session_ids_correct``).
Everything is also broken down by ``question_type``. Use ``--list-bad`` to print
the samples whose golden answer was judged NOT correct.
Examples:
python benchmark/longmemeval/stats_golden_check.py
python benchmark/longmemeval/stats_golden_check.py --list-bad
python benchmark/longmemeval/stats_golden_check.py --list-run-failed
python benchmark/longmemeval/stats_golden_check.py --json
"""
import argparse
import json
from collections import defaultdict
from pathlib import Path
REPO = Path(__file__).resolve().parents[2]
DATA = REPO / "datasets" / "longmemeval"
LOGDIR = REPO / "logs" / "golden_check"
def parse_args() -> argparse.Namespace:
"""Parse command-line arguments."""
p = argparse.ArgumentParser(description=__doc__)
p.add_argument("--list-bad", action="store_true", help="list samples whose golden answer is NOT correct")
p.add_argument(
"--list-bad-sessions",
action="store_true",
help="list samples whose answer_session_ids is NOT correct",
)
p.add_argument(
"--list-run-failed",
action="store_true",
help="list launched samples that did not produce readable output",
)
p.add_argument("--json", action="store_true", help="emit the summary as JSON")
return p.parse_args()
def sample_ids() -> list[str]:
"""List all sample IDs."""
ids = [p.name for p in DATA.iterdir() if p.is_dir() and p.name.isdigit()]
return sorted(ids, key=int)
def pct(num: int, den: int) -> str:
"""Format a percentage."""
return f"{(100.0 * num / den):.1f}%" if den else "n/a"
def logged_sample_ids() -> list[str]:
"""List all sample IDs that have been launched but not finished."""
if not LOGDIR.exists():
return []
ids = [p.stem for p in LOGDIR.glob("*.log") if p.stem.isdigit()]
return sorted(ids, key=int)
def load_json(path: Path) -> dict:
"""Load a JSON object, returning {} on any error."""
try:
with path.open(encoding="utf-8") as f:
data = json.load(f)
return data if isinstance(data, dict) else {}
except (OSError, json.JSONDecodeError):
return {}
def question_type_for(idx: str, data: dict) -> str:
"""Return question_type from the output, session review, or query.json."""
question_type = str(data.get("question_type") or "").strip()
if question_type:
return question_type
review_path_raw = str(data.get("session_review_path") or "").strip()
review_path = Path(review_path_raw) if review_path_raw else DATA / idx / "session_review.json"
if not review_path.is_absolute():
review_path = REPO / review_path
review = load_json(review_path)
review_question_type = str((review.get("query") or {}).get("question_type") or "").strip()
if review_question_type:
return review_question_type
query = load_json(DATA / idx / "query.json")
return str(query.get("question_type") or "(unknown)").strip() or "(unknown)"
def question_id_for(idx: str, data: dict) -> str:
"""Return question_id from the output, session review, or query.json."""
question_id = str(data.get("question_id") or "").strip()
if question_id:
return question_id
review_path_raw = str(data.get("session_review_path") or "").strip()
review_path = Path(review_path_raw) if review_path_raw else DATA / idx / "session_review.json"
if not review_path.is_absolute():
review_path = REPO / review_path
review = load_json(review_path)
review_question_id = str((review.get("query") or {}).get("question_id") or "").strip()
if review_question_id:
return review_question_id
query = load_json(DATA / idx / "query.json")
return str(query.get("question_id") or "").strip()
def sample_label(data: dict) -> str:
"""Format sample id as idx(question_id) when question_id is available."""
idx = str(data.get("_idx") or "")
qid = str(data.get("_question_id") or "").strip()
return f"{idx}({qid})" if qid else idx
def related_session_ids(data: dict) -> list[str]:
"""Return the best available session ids for a bad verdict record."""
verdict = data.get("verdict") if isinstance(data, dict) else None
if isinstance(verdict, dict):
true_ids = verdict.get("true_answer_session_ids")
if isinstance(true_ids, list):
ids = [str(session_id) for session_id in true_ids if str(session_id).strip()]
if ids:
return ids
summaries = data.get("session_summaries")
if isinstance(summaries, list):
return [
str(summary.get("session_id"))
for summary in summaries
if isinstance(summary, dict) and str(summary.get("session_id") or "").strip()
]
return []
def grouped_records(records: list[dict]) -> dict[str, list[dict]]:
"""Group records by question_type for human-readable list output."""
grouped: dict[str, list[dict]] = defaultdict(list)
for data in records:
question_type = str(data.get("_question_type") or "(unknown)")
grouped[question_type].append(
{
"index": str(data.get("_idx") or ""),
"question_id": str(data.get("_question_id") or ""),
"session_id": related_session_ids(data),
},
)
return dict(sorted(grouped.items()))
def verdict_bool(verdict: dict, new_key: str, old_key: str) -> bool:
"""Read a verdict boolean, accepting the old field name for compatibility."""
if verdict.get(new_key) is True:
return True
if verdict.get(new_key) is False:
return False
return verdict.get(old_key) is True
def has_current_verdict(data: dict) -> bool:
"""Return True when ``check_golden.json`` uses the current golden_check schema."""
verdict = data.get("verdict") if isinstance(data, dict) else None
if not isinstance(verdict, dict):
return False
return isinstance(verdict.get("golden_answer_correct"), bool) and isinstance(
verdict.get("answer_session_ids_correct"),
bool,
)
def write_golden_check_list(done: list[dict], output_path: Path) -> None:
"""Write all readable check_golden records as JSONL."""
with output_path.open("w", encoding="utf-8") as f:
for data in done:
f.write(json.dumps(data, ensure_ascii=False))
f.write("\n")
def main() -> int:
"""Main entry point."""
args = parse_args()
ids = sample_ids()
total = len(ids)
done, unreadable, stale = [], [], []
finished_ids = set()
for idx in ids:
path = DATA / idx / "check_golden.json"
if not path.exists():
continue
try:
with path.open(encoding="utf-8") as f:
data = json.load(f)
if not has_current_verdict(data):
stale.append(idx)
continue
data["_idx"] = idx
data["_question_type"] = question_type_for(idx, data)
data["_question_id"] = question_id_for(idx, data)
done.append(data)
finished_ids.add(idx)
except (OSError, json.JSONDecodeError):
unreadable.append(idx)
n = len(done)
output_path = Path.cwd() / "golden_check_list.jsonl"
write_golden_check_list(done, output_path)
launched = logged_sample_ids()
run_failed = [idx for idx in launched if idx not in finished_ids]
# Overall tallies.
golden_ok = sum(
1 for d in done if verdict_bool(d.get("verdict", {}), "golden_answer_correct", "golden_answer_reasonable")
)
sess_ok = sum(
1
for d in done
if verdict_bool(d.get("verdict", {}), "answer_session_ids_correct", "answer_session_ids_reasonable")
)
both_ok = sum(
1
for d in done
if verdict_bool(d.get("verdict", {}), "golden_answer_correct", "golden_answer_reasonable")
and verdict_bool(d.get("verdict", {}), "answer_session_ids_correct", "answer_session_ids_reasonable")
)
# Per question_type breakdown.
by_type: dict[str, dict[str, int]] = defaultdict(lambda: {"n": 0, "golden_ok": 0, "sess_ok": 0, "both_ok": 0})
for d in done:
v = d.get("verdict", {})
golden_is_ok = verdict_bool(v, "golden_answer_correct", "golden_answer_reasonable")
sess_is_ok = verdict_bool(v, "answer_session_ids_correct", "answer_session_ids_reasonable")
t = d.get("_question_type") or "(unknown)"
by_type[t]["n"] += 1
by_type[t]["golden_ok"] += 1 if golden_is_ok else 0
by_type[t]["sess_ok"] += 1 if sess_is_ok else 0
by_type[t]["both_ok"] += 1 if golden_is_ok and sess_is_ok else 0
bad_golden_records = [
d for d in done if not verdict_bool(d.get("verdict", {}), "golden_answer_correct", "golden_answer_reasonable")
]
bad_session_records = [
d
for d in done
if not verdict_bool(d.get("verdict", {}), "answer_session_ids_correct", "answer_session_ids_reasonable")
]
bad_golden = [d["_idx"] for d in bad_golden_records]
bad_sessions = [d["_idx"] for d in bad_session_records]
if args.json:
print(
json.dumps(
{
"total": total,
"finished": n,
"pending": total - n - len(unreadable),
"unreadable": unreadable,
"stale": stale,
"launched": len(launched),
"run_failed": run_failed,
"golden_answer_accuracy": round(golden_ok / n, 4) if n else None,
"answer_session_ids_accuracy": round(sess_ok / n, 4) if n else None,
"both_correct_rate": round(both_ok / n, 4) if n else None,
"golden_ok": golden_ok,
"sess_ok": sess_ok,
"both_ok": both_ok,
"by_type": {
t: {
**c,
"golden_bad": c["n"] - c["golden_ok"],
"session_bad": c["n"] - c["sess_ok"],
"both_bad": c["n"] - c["both_ok"],
"golden_acc": round(c["golden_ok"] / c["n"], 4),
"session_acc": round(c["sess_ok"] / c["n"], 4),
"both_acc": round(c["both_ok"] / c["n"], 4),
}
for t, c in by_type.items()
},
"bad_golden": bad_golden,
"bad_sessions": bad_sessions,
"golden_check_list": str(output_path),
},
ensure_ascii=False,
indent=2,
),
)
return 0
print("=" * 60)
print("LongMemEval golden_check 统计")
print("=" * 60)
print(f"样例总数 : {total}")
print(f"已完成 (有产出) : {n} ({pct(n, total)})")
print(f"未完成 : {total - n - len(unreadable)}")
if unreadable:
print(f"损坏/无法解析 : {len(unreadable)} {unreadable}")
if stale:
print(f"旧格式待重跑 : {len(stale)} {stale}")
print(f"已合并 JSONL : {output_path}")
print(f"已启动过 (有 log) : {len(launched)}")
print(f"运行失败/无可读产出 : {len(run_failed)}")
print("-" * 60)
print(f"golden answer 正确率 : {pct(golden_ok, n)} ({golden_ok}/{n})")
print(f"answer_session 正确率: {pct(sess_ok, n)} ({sess_ok}/{n})")
print(f"两者都正确 : {pct(both_ok, n)} ({both_ok}/{n})")
print("-" * 60)
print("按 question_type:")
print(
f" {'type':<24} {'n':>4} {'golden正确率':>14} {'golden错误':>10} "
f"{'session正确率':>14} {'session错误':>11} {'都正确':>10} {'都正确错误':>12}",
)
for t in sorted(by_type):
c = by_type[t]
print(
f" {t:<24} {c['n']:>4} {pct(c['golden_ok'], c['n']):>14} {c['n'] - c['golden_ok']:>10} "
f"{pct(c['sess_ok'], c['n']):>14} {c['n'] - c['sess_ok']:>11} "
f"{pct(c['both_ok'], c['n']):>10} {c['n'] - c['both_ok']:>12}",
)
if args.list_bad:
print("-" * 60)
print(f"golden answer 判为不正确的样例 ({len(bad_golden_records)}):")
print(json.dumps(grouped_records(bad_golden_records), ensure_ascii=False))
if args.list_bad_sessions:
print("-" * 60)
print(f"answer_session_ids 判为不正确的样例 ({len(bad_session_records)}):")
print(json.dumps(grouped_records(bad_session_records), ensure_ascii=False))
if args.list_run_failed:
print("-" * 60)
print(f"运行失败/无可读 check_golden.json 的样例 ({len(run_failed)}): {run_failed}")
for idx in run_failed:
print(f" {idx}: {LOGDIR / f'{idx}.log'}")
print("=" * 60)
return 0
if __name__ == "__main__":
raise SystemExit(main())

248
benchmark/longmemeval/stats_session_review.py Normal file
View file

@ -0,0 +1,248 @@
#!/usr/bin/env python3
"""Summarise LongMemEval ``session_review.json`` artifacts.
This script is for upstream health checks before running ``golden_check``.
Samples with retryable per-session failures should be rerun as a whole; samples
with non-retryable fallback reviews are reported separately.
Examples:
python benchmark/longmemeval/stats_session_review.py
python benchmark/longmemeval/stats_session_review.py --list-failed
python benchmark/longmemeval/stats_session_review.py --list-fallback
python benchmark/longmemeval/stats_session_review.py --json
"""
import argparse
import json
from pathlib import Path
REPO = Path(__file__).resolve().parents[2]
DATA = REPO / "datasets" / "longmemeval"
LOGDIR = REPO / "logs" / "session_review"
OUTPUT_FILENAME = "session_review.json"
def parse_args() -> argparse.Namespace:
"""Parse command-line arguments."""
p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
p.add_argument("--list-failed", action="store_true", help="list samples with retryable failed per-session reviews")
p.add_argument("--list-fallback", action="store_true", help="list non-retryable fallback reviews")
p.add_argument("--list-missing", action="store_true", help="list samples missing session_review.json")
p.add_argument("--list-run-failed", action="store_true", help="list launched samples without a healthy output")
p.add_argument("--json", action="store_true", help="emit the summary as JSON")
return p.parse_args()
def sample_ids() -> list[str]:
"""List all numeric sample IDs."""
ids = [p.name for p in DATA.iterdir() if p.is_dir() and p.name.isdigit()]
return sorted(ids, key=int)
def pct(num: int, den: int) -> str:
"""Format a percentage."""
return f"{(100.0 * num / den):.1f}%" if den else "n/a"
def load_json(path: Path) -> dict:
"""Load a JSON object, returning {} on any error."""
try:
with path.open(encoding="utf-8") as f:
data = json.load(f)
return data if isinstance(data, dict) else {}
except (OSError, json.JSONDecodeError):
return {}
def logged_sample_ids() -> list[str]:
"""List sample IDs that have a session_review runner log."""
if not LOGDIR.exists():
return []
ids = [p.stem for p in LOGDIR.glob("*.log") if p.stem.isdigit()]
return sorted(ids, key=int)
def review_block(data: dict) -> dict:
"""Return the review block when present."""
review = data.get("review") if isinstance(data, dict) else None
return review if isinstance(review, dict) else {}
def failure_details(data: dict) -> list[dict]:
"""Return retryable failed_reviews when present."""
failed_reviews = review_block(data).get("failed_reviews")
if not isinstance(failed_reviews, list):
return []
return [item for item in failed_reviews if isinstance(item, dict) and not item.get("fallback")]
def fallback_details(data: dict) -> list[dict]:
"""Return non-retryable fallback review details when present."""
review = review_block(data)
fallback_reviews = review.get("fallback_reviews")
if isinstance(fallback_reviews, list):
return [item for item in fallback_reviews if isinstance(item, dict)]
failed_reviews = review.get("failed_reviews")
if isinstance(failed_reviews, list):
return [item for item in failed_reviews if isinstance(item, dict) and item.get("fallback")]
return []
def failure_count(data: dict) -> int:
"""Return retryable failed review count."""
review = review_block(data)
raw = review.get("num_failed_reviews")
raw_fallback = review.get("num_fallback_reviews")
if isinstance(raw, int) and isinstance(raw_fallback, int):
return max(0, raw - raw_fallback)
return len(failure_details(data))
def fallback_count(data: dict) -> int:
"""Return non-retryable fallback review count."""
review = review_block(data)
raw = review.get("num_fallback_reviews")
if isinstance(raw, int):
return raw
return len(fallback_details(data))
def question_id(data: dict) -> str:
"""Return query.question_id when present."""
query = data.get("query") if isinstance(data, dict) else None
if not isinstance(query, dict):
return ""
return str(query.get("question_id") or "").strip()
def main() -> int:
"""Main entry point."""
args = parse_args()
ids = sample_ids()
total = len(ids)
healthy, failed, fallback, missing, unreadable = [], [], [], [], []
total_failed_sessions = 0
total_fallback_sessions = 0
failed_details_by_id: dict[str, list[dict]] = {}
fallback_details_by_id: dict[str, list[dict]] = {}
question_id_by_id: dict[str, str] = {}
for idx in ids:
path = DATA / idx / OUTPUT_FILENAME
if not path.exists():
missing.append(idx)
continue
data = load_json(path)
if not data:
unreadable.append(idx)
continue
question_id_by_id[idx] = question_id(data)
n_failed = failure_count(data)
n_fallback = fallback_count(data)
if n_failed:
failed.append(idx)
total_failed_sessions += n_failed
failed_details_by_id[idx] = failure_details(data)
if n_fallback:
fallback.append(idx)
total_fallback_sessions += n_fallback
fallback_details_by_id[idx] = fallback_details(data)
if not n_failed:
healthy.append(idx)
launched = logged_sample_ids()
healthy_set = set(healthy)
run_failed = [idx for idx in launched if idx not in healthy_set]
if args.json:
print(
json.dumps(
{
"total": total,
"healthy": len(healthy),
"failed_samples": failed,
"failed_sample_count": len(failed),
"failed_session_count": total_failed_sessions,
"fallback_samples": fallback,
"fallback_sample_count": len(fallback),
"fallback_session_count": total_fallback_sessions,
"missing": missing,
"unreadable": unreadable,
"launched": len(launched),
"run_failed_or_unhealthy": run_failed,
"failed_details": failed_details_by_id,
"fallback_details": fallback_details_by_id,
},
ensure_ascii=False,
indent=2,
),
)
return 0
print("=" * 60)
print("LongMemEval session_review 统计")
print("=" * 60)
print(f"样例总数 : {total}")
print(f"可继续产出 : {len(healthy)} ({pct(len(healthy), total)})")
print(f"有可重试失败 : {len(failed)}")
print(f"可重试失败 session : {total_failed_sessions}")
print(f"有不可重试 fallback : {len(fallback)}")
print(f"fallback session : {total_fallback_sessions}")
print(f"缺少 session_review : {len(missing)}")
print(f"损坏/无法解析 : {len(unreadable)}")
print(f"已启动过 (有 log) : {len(launched)}")
print(f"运行失败/非健康产出 : {len(run_failed)}")
print("-" * 60)
print("有可重试 failed_reviews 的样例需要整体重跑：")
if failed:
print(" ".join(failed))
print("重跑命令示例：")
print(f"python benchmark/longmemeval/run_session_review.py --start {failed[0]} --end {failed[0]}")
else:
print("(none)")
if fallback:
print("-" * 60)
print("不可重试 fallback 的样例不用重跑：")
for idx in fallback:
details = fallback_details_by_id.get(idx) or []
session_ids = [str(item.get("session_id") or "(unknown)") for item in details]
qid = question_id_by_id.get(idx)
sample_label = f"{idx}({qid})" if qid else idx
print(f"{sample_label}: {' '.join(session_ids) if session_ids else '(unknown)'}")
if args.list_failed and failed:
print("-" * 60)
for idx in failed:
details = failed_details_by_id.get(idx) or []
print(f"{idx}: {DATA / idx / OUTPUT_FILENAME} failed_sessions={len(details)}")
for item in details:
session_id = item.get("session_id", "(unknown)")
error = str(item.get("error") or "").replace("\n", " ")
print(f" - {session_id}: {error}")
if args.list_fallback and fallback:
print("-" * 60)
for idx in fallback:
details = fallback_details_by_id.get(idx) or []
print(f"{idx}: {DATA / idx / OUTPUT_FILENAME} fallback_sessions={len(details)}")
for item in details:
session_id = item.get("session_id", "(unknown)")
reason = str(item.get("fallback_reason") or "fallback")
error = str(item.get("error") or "").replace("\n", " ")
raw_saved = "yes" if item.get("raw_session") else "no"
print(f" - {session_id}: reason={reason} raw_session_saved={raw_saved} error={error}")
if args.list_missing and missing:
print("-" * 60)
print(f"缺少 session_review.json 的样例 ({len(missing)}): {missing}")
if args.list_run_failed and run_failed:
print("-" * 60)
print(f"运行失败/非健康产出的样例 ({len(run_failed)}): {run_failed}")
for idx in run_failed:
print(f" {idx}: {LOGDIR / f'{idx}.log'}")
print("=" * 60)
return 0
if __name__ == "__main__":
raise SystemExit(main())

4
pyproject.toml
View file

@ -55,9 +55,13 @@ dev = [
"pytest>=8.0",
"pytest-asyncio>=0.23",
]
benchmark = [
"portalocker>=2.10.1",
]
full = [
"reme-ai[core]",
"reme-ai[dev]",
"reme-ai[benchmark]",
]
[project.urls]

6
reme/application.py
View file

@ -29,7 +29,11 @@ class Application(BaseComponent):
if self.config.enable_logo:
print_logo(self.config)
logger = get_logger(log_to_console=self.config.log_to_console, log_to_file=self.config.log_to_file)
logger = get_logger(
log_to_console=self.config.log_to_console,
log_to_file=self.config.log_to_file,
force_init=True,
)
logger.info(f"Initializing {self.config.app_name} Application v{__version__}")
super().__init__()

2
reme/components/agent_wrapper/as_agent_wrapper.py
View file

@ -46,6 +46,7 @@ from agentscope.tool import (
Read,
ToolBase,
ToolChunk,
ToolChoice,
Toolkit,
Write,
)
@ -329,6 +330,7 @@ class AsAgentWrapper(BaseAgentWrapper):
res = await model.generate_structured_output(
messages=agent.state.context,
structured_model=output_schema,
tool_choice=ToolChoice(mode="auto"),
)
result["structured_output"] = res.content

2
reme/components/component_registry.py
View file

@ -18,7 +18,7 @@ class ComponentRegistry:
def __init__(self) -> None:
self._registry: dict[ComponentEnum, dict[str, type[BaseComponent]]] = {}
self.logger = get_logger()
self.logger = get_logger(log_to_file=False)
def _do_register(self, cls: type[T], name: str) -> type[T]:
"""Insert `cls` under its ``component_type`` group; warn on overwrite."""

2
reme/config/config_parser.py
View file

@ -207,7 +207,7 @@ def resolve_app_config(**kwargs) -> dict:
"""
from ..utils import get_logger
logger = get_logger()
logger = get_logger(log_to_file=False)
configs: list[dict] = []
# `config=path` arrives as a string here; `config.foo=bar` arrives as a

205
reme/config/jinli_lme.yaml
View file

@ -4,14 +4,26 @@ service:
workspace_dir: ${LME_WORKSPACE_DIR:-datasets/longmemeval/1}
session_dir: history_session
resource_dir: session
daily_dir: ""
daily_dir: daily
digest_dir: ""
jobs:
auto_memory:
backend: base
description: "Extract every raw session into a search-friendly daily note (one note per session)."
parameters:
type: object
properties: { }
steps:
- backend: clear_paths_step # wipe old daily notes so this is a clean rebuild
config_keys: [daily_dir]
- backend: lme_auto_memory_step
agent_wrapper: lme_memory
update_index:
backend: base
watch_dirs: [resource_dir]
watch_suffixes: [md, json, jsonl]
watch_dirs: [daily_dir]
watch_suffixes: [md]
steps:
- backend: clear_store_step
- backend: init_changes_step
@ -19,6 +31,73 @@ jobs:
monitor_name: default
dispatch_steps: [update_index_step]
extract_session_by_id:
backend: base
description: "Given a session_id shown in a search result, go back to the original raw session and extract everything in it that is relevant to the current question. Use this when a search hit looks relevant but the distilled note lacks an exact number, date, or wording."
parameters:
type: object
properties:
session_id:
type: string
description: "the session_id from a search result header"
required:
- session_id
steps:
- backend: lme_extract_session_step
agent_wrapper: lme_extract
vector_search:
backend: base
description: "Dense semantic search over the memory notes. Returns each hit with its source note path and session_id."
parameters:
type: object
properties:
query:
type: string
description: "search query"
required:
- query
steps:
- backend: vector_search_step
include_source: true
bm25_search:
backend: base
description: "Keyword (BM25) search over the memory notes. Returns each hit with its source note path and session_id."
parameters:
type: object
properties:
query:
type: string
description: "search query"
required:
- query
steps:
- backend: bm25_search_step
include_source: true
agentic_answer:
backend: base
description: "Answer query.json from indexed memory using vector/bm25 search + session extraction."
parameters:
type: object
properties: { }
steps:
- backend: clear_paths_step # drop the previous answer so each run rewrites cleanly
paths: [mem_answer.json]
- backend: lme_agentic_answer_step
agent_wrapper: lme_agentic_answer
llm_judge:
backend: base
description: "Judge mem_answer.json against answer.json and write the judgement back into mem_answer.json."
parameters:
type: object
properties: { }
steps:
- backend: lme_llm_judge_step
agent_wrapper: lme_judge
version:
backend: base
description: "return reme package version"
@ -92,6 +171,34 @@ jobs:
steps:
- backend: python_execute_step
session_review:
backend: base
description: "Review every session for query/answer-relevant evidence and write session_review.json."
parameters:
type: object
properties: { }
steps:
- backend: clear_paths_step # drop the previous review so each run rewrites cleanly
paths: [session_review.json]
- backend: lme_session_review_step
agent_wrapper: lme_review
golden_check:
backend: base
description: "Read session_review.json and judge whether the golden answer is reasonable."
parameters:
type: object
properties: { }
steps:
- backend: clear_paths_step # drop the previous verdict so each run rewrites cleanly
paths: [check_golden.json]
- backend: wait_for_paths_step # wait until session_review finishes; comment this step to fail fast instead
paths: [session_review.json]
poll_seconds: 5
log_every_seconds: 60
- backend: lme_golden_check_step
agent_wrapper: lme_judge
components:
tokenizer:
default:
@ -125,6 +232,18 @@ components:
parameters:
max_tokens: 65536
plus:
backend: ${LLM_BACKEND:-openai}
model: qwen3.7-plus
stream: true
context_size: 1000000
max_retries: 3
credential:
api_key: ${LLM_API_KEY:-}
base_url: ${LLM_BASE_URL:-}
parameters:
max_tokens: 65536
agent_wrapper:
default:
backend: agentscope
@ -160,6 +279,55 @@ components:
model_config:
max_retries: 3
lme_memory:
backend: agentscope
as_llm: plus
permission_mode: bypass
builtin_tools: false
react_config:
max_iters: 8
context_config:
trigger_ratio: 0.89
reserve_ratio: 0.1
tool_result_limit: 1000000
model_config:
max_retries: 3
lme_extract:
backend: agentscope
as_llm: plus
permission_mode: bypass
builtin_tools: false
react_config:
max_iters: 3
context_config:
trigger_ratio: 0.89
reserve_ratio: 0.1
tool_result_limit: 1000000
model_config:
max_retries: 3
lme_agentic_answer:
backend: agentscope
as_llm: default
cwd: session
permission_mode: bypass
builtin_tools: false
job_tools:
- vector_search
- bm25_search
- python_execute
- extract_session_by_id
sequential_tool_calls: true
react_config:
max_iters: 40
context_config:
trigger_ratio: 0.89
reserve_ratio: 0.1
tool_result_limit: 1000000
model_config:
max_retries: 3
claude_code:
backend: claude_code
model: ${CLAUDE_CODE_MODEL_NAME:-glm-5.2}
@ -167,6 +335,37 @@ components:
base_url: ${CLAUDE_CODE_BASE_URL:-https://dashscope.aliyuncs.com/apps/anthropic}
permission_mode: bypassPermissions
lme_review:
backend: agentscope
as_llm: plus
permission_mode: bypass
builtin_tools: false
react_config:
max_iters: 10
context_config:
trigger_ratio: 0.89
reserve_ratio: 0.1
tool_result_limit: 1000000
model_config:
max_retries: 3
lme_judge:
backend: agentscope
as_llm: default
permission_mode: bypass
builtin_tools: false
job_tools:
- python_execute
sequential_tool_calls: true
react_config:
max_iters: 50
context_config:
trigger_ratio: 0.89
reserve_ratio: 0.1
tool_result_limit: 1000000
model_config:
max_retries: 3
file_graph:
default:
backend: local

6
reme/steps/benchmark/__init__.py
View file

@ -1,10 +1,12 @@
"""Benchmark steps."""
from . import lme
from .lme import AnswerJudgeStep, ContextAnswerStep
from .lme import ContextAnswerStep, GoldenCheckStep, LmeLlmJudgeStep, SessionReviewStep
__all__ = [
"AnswerJudgeStep",
"ContextAnswerStep",
"GoldenCheckStep",
"LmeLlmJudgeStep",
"SessionReviewStep",
"lme",
]

14
reme/steps/benchmark/lme/__init__.py
View file

@ -1,9 +1,19 @@
"""LongMemEval benchmark steps."""
from .agentic_answer import LmeAgenticAnswerStep
from .auto_memory import LmeAutoMemoryStep
from .context_answer import ContextAnswerStep
from .llm_judge import AnswerJudgeStep
from .extract_session import LmeExtractSessionStep
from .golden_check import GoldenCheckStep
from .lme_llm_judge import LmeLlmJudgeStep
from .session_review import SessionReviewStep
__all__ = [
"AnswerJudgeStep",
"ContextAnswerStep",
"GoldenCheckStep",
"LmeAgenticAnswerStep",
"LmeAutoMemoryStep",
"LmeExtractSessionStep",
"LmeLlmJudgeStep",
"SessionReviewStep",
]

87
reme/steps/benchmark/lme/agentic_answer.py Normal file
View file

@ -0,0 +1,87 @@
"""agentic_answer — answer the LongMemEval question from the indexed memory.
Job #4 of the pipeline. Reads ``query.json`` and hands the question to an agent
equipped with ``vector_search`` / ``bm25_search`` / ``python_execute`` /
``extract_session_by_id``. The agent searches the daily-note index, pivots to
raw sessions by ``session_id`` when a hit is promising, and keeps trying until it
can answer or has searched too many times. The final answer is written to
``mem_answer.json`` in the workspace.
"""
import json
from ...base_step import BaseStep
from ....components import R
@R.register("lme_agentic_answer_step")
class LmeAgenticAnswerStep(BaseStep):
"""Drive the tool-using agent that answers from indexed memory."""
_OUTPUT_FILE = "mem_answer.json"
def _load_query(self) -> dict:
path = self.workspace_path / "query.json"
with path.open(encoding="utf-8") as f:
data = json.load(f)
if not isinstance(data, dict):
raise ValueError("query.json is not a JSON object")
return data
async def execute(self):
assert self.context is not None
if self.agent_wrapper is None:
raise ValueError("lme_agentic_answer_step requires agent_wrapper")
query = self._load_query()
question = str(query.get("question", "") or "").strip()
question_date = str(query.get("question_date", "") or "").strip()
question_id = str(query.get("question_id", "") or "").strip()
if not question:
raise ValueError("query.json requires a non-empty 'question'")
user_prompt = self.prompt_format(
"user_message",
question=question,
question_date=question_date or "(unknown)",
)
# A stable tool_context_id makes vector/bm25 dedup across this answer run,
# so repeated searches surface genuinely new chunks each time.
result = await self.agent_wrapper.reply(
user_prompt,
system_prompt=self.get_prompt("system_prompt"),
tool_context_id=question_id or question,
)
answer = (result.get("result") or "").strip()
# session_id names the trajectory file mem_session/agentscope/<session_id>.jsonl,
# so downstream tooling can locate this run's full tool-call trail.
session_id = str(result.get("session_id") or "")
out_path = self.workspace_path / self._OUTPUT_FILE
out_path.write_text(
json.dumps(
{
"question_id": question_id,
"question": question,
"answer": answer,
"session_id": session_id,
},
ensure_ascii=False,
indent=2,
),
encoding="utf-8",
)
self.logger.info(f"[{self.name}] answer for {question_id or question!r}: {answer!r}")
self.context.response.success = True
self.context.response.answer = answer
self.context.response.metadata.update(
{
"question_id": question_id,
"question": question,
"answer": answer,
"session_id": session_id,
"path": self._OUTPUT_FILE,
},
)
return self.context.response

40
reme/steps/benchmark/lme/agentic_answer.yaml Normal file
View file

@ -0,0 +1,40 @@
system_prompt: |
You answer a user's question using ONLY their long-term memory, retrieved through tools. The
memory is a set of daily notes, each distilled from one past chat session and tagged in its
header with a `session_id` pointing back to the raw session.
Available tools:
- `vector_search(query)`: dense semantic search over the memory notes.
- `bm25_search(query)`: keyword search over the memory notes.
- `extract_session_by_id(session_id)`: go back to the ORIGINAL raw session behind a note and
pull out its full content relevant to the question. Use the `session_id` shown in a search
result's header.
- `python_execute(code)`: run Python for any counting, date math, or reasoning over what you found.
Strategy:
1. Start by searching with the user's ORIGINAL question wording — call BOTH `vector_search` and
`bm25_search` with it.
2. If the results already fully support an answer, answer.
3. If a result looks relevant but the distilled note is not enough (missing a number, exact date,
or wording), take its `session_id` and call `extract_session_by_id` to read the raw session.
4. Do NOT give up early. Keep trying: reformulate with new keywords, aliases, entities, dates, and
short phrases from the question or from earlier results; search again; and call
`extract_session_by_id` on any additional relevant sessions.
5. Search results are deduplicated within this run, so a search returning nothing new means those
chunks were already seen — change your wording rather than repeating it.
6. Only after you have made MORE THAN 10 search attempts and still cannot find support, answer
exactly: not provided
Answer rules:
- Answer strictly from retrieved memory; never invent facts.
- Be direct and specific; include the exact value/date the question asks for.
- Your final message is the answer itself (no tool calls, no preamble).
user_message: |
Question date: {question_date}
Question: {question}
Find the answer in the user's long-term memory using the tools, following the strategy above.
Remember: search with the original question first (both vector and bm25), pivot to
`extract_session_by_id` for promising sessions, keep trying with new wording, and only answer
"not provided" after more than 10 search attempts have failed.

428
reme/steps/benchmark/lme/auto_memory.py Normal file
View file

@ -0,0 +1,428 @@
"""lme_auto_memory — turn every LongMemEval session into a search-friendly note.
For a workspace such as ``datasets/longmemeval/1`` this step walks each raw
session under ``resource_dir`` (files named ``<date>_(...)_<time>@<session_id>.json``
with ``haystack_date`` / ``haystack_session_id`` / ``messages``) and, one per
session, asks an agent to *completely* extract its content — entities, times,
numbers, preferences, events, causal links — into a daily note optimized for
both BM25 and vector retrieval.
Each note is written to ``<daily_dir>/<YYYY-MM-DD>/<name>.md`` via the shared
``daily_write`` job, so the frontmatter carries ``session_id`` for progressive
expansion (the agentic-answer flow pivots from a search hit back to the raw
session through this id). Filenames are LLM-generated topic stems; same-day
collisions are disambiguated by appending the session id.
"""
import asyncio
import json
import re
import time
from datetime import datetime
from pathlib import Path
import frontmatter
from ...base_step import BaseStep
from ...file_io import extract_daily_date
from ....components import R
START_INTERVAL_SECONDS = 1.0
MAX_CONCURRENCY = 60
RETRY_INITIAL_SECONDS = 5.0
RETRY_MAX_SECONDS = 300.0
_LME_DATETIME_RE = re.compile(r"(\d{4})/(\d{2})/(\d{2}).*?(\d{2}):(\d{2})")
_NON_RETRYABLE_DATA_INSPECTION_MARKERS = (
"data_inspection_failed",
"DataInspectionFailed",
"Input text data may contain inappropriate content",
)
# Structured extraction the memory agent must return per session.
_MEMORY_SCHEMA = {
"type": "object",
"properties": {
"name": {
"type": "string",
"description": "Concise, stable topic/event filename stem (kebab-case, no date, no slash or "
"reserved characters). E.g. 'daily-commute-details' or 'leather-boot-care'.",
},
"description": {
"type": "string",
"description": "Thorough one-paragraph summary of the note body — specific enough that this "
"description alone conveys all key facts. Used as a search-friendly abstract.",
},
"body": {
"type": "string",
"description": "Complete markdown extraction of every core fact in the session, written for "
"retrieval (natural-language statements, explicit entities, dates and numbers verbatim).",
},
},
"required": ["name", "description", "body"],
}
@R.register("lme_auto_memory_step")
class LmeAutoMemoryStep(BaseStep):
"""Extract each LME session into a daily note via a per-session agent."""
def __init__(self, **kwargs):
super().__init__(**kwargs)
self._reserve_lock = asyncio.Lock()
self._reserved: dict[tuple[str, str], str] = {}
def _resource_dir_name(self) -> str:
return self.app_context.app_config.resource_dir if self.app_context is not None else "session"
def _session_dir(self) -> Path:
return self.workspace_path / self._resource_dir_name()
@staticmethod
def _parse_lme_datetime(raw_date: str) -> datetime | None:
"""Parse LongMemEval timestamps like ``2023/05/20 (Sat) 03:29``."""
match = _LME_DATETIME_RE.search(raw_date.strip())
if match is None:
return None
try:
year, month, day, hour, minute = (int(part) for part in match.groups())
return datetime(year, month, day, hour, minute)
except ValueError:
return None
@staticmethod
def _parse_day(raw_date: str) -> str | None:
"""Parse a LongMemEval ``haystack_date`` (e.g. '2023/05/20 (Sat) 03:29') to YYYY-MM-DD."""
head = raw_date.strip()[:10].replace("/", "-")
return extract_daily_date(head)
@staticmethod
def _load_json(path: Path) -> dict:
with path.open(encoding="utf-8") as f:
data = json.load(f)
if not isinstance(data, dict):
raise ValueError("session file is not a JSON object")
return data
@staticmethod
def _format_messages(messages: list) -> str:
lines: list[str] = []
for msg in messages:
if not isinstance(msg, dict):
continue
role = str(msg.get("role", "")).strip() or "unknown"
content = msg.get("content", "")
if not isinstance(content, str):
content = json.dumps(content, ensure_ascii=False)
lines.append(f"[{role}]\n{content}")
return "\n\n".join(lines)
@staticmethod
def _is_data_inspection_error(exc: Exception) -> bool:
text = str(exc)
return any(marker in text for marker in _NON_RETRYABLE_DATA_INSPECTION_MARKERS)
async def _existing_session_id(self, rel_path: str) -> str:
note = self.workspace_path / rel_path
if not note.is_file():
return ""
try:
post = frontmatter.loads(note.read_text(encoding="utf-8"))
except Exception:
return ""
return str((post.metadata or {}).get("session_id", "") or "").strip()
async def _reserve_name(self, daily_dir: str, day: str, name: str, session_id: str) -> str:
"""Pick a collision-free filename stem for this session under ``day``."""
async with self._reserve_lock:
for cand in (name, f"{name}-{session_id}"):
key = (day, cand)
owner = self._reserved.get(key)
if owner == session_id:
return cand
if owner is not None:
continue
existing = await self._existing_session_id(f"{daily_dir}/{day}/{cand}.md")
if existing and existing != session_id:
continue
self._reserved[key] = session_id
return cand
# Extremely unlikely fallback (same topic AND same session id twice).
i = 2
while True:
cand = f"{name}-{session_id}-{i}"
key = (day, cand)
if key not in self._reserved:
self._reserved[key] = session_id
return cand
i += 1
# pylint: disable=too-many-statements
async def execute(self):
assert self.context is not None
if self.agent_wrapper is None:
raise ValueError("lme_auto_memory_step requires agent_wrapper")
query_data = self._load_json(self.workspace_path / "query.json")
question_date = str(query_data.get("question_date") or "").strip()
question_dt = self._parse_lme_datetime(question_date)
if question_dt is None:
raise ValueError(f"query.json has an invalid 'question_date': {question_date!r}")
session_dir = self._session_dir()
if not session_dir.is_dir():
raise FileNotFoundError(f"Session directory not found: {session_dir}")
session_files = sorted(p for p in session_dir.iterdir() if p.suffix == ".json")
sessions: list[tuple[dict, Path, str, str, str]] = []
filtered_sessions: list[dict] = []
session_ids_illegal: list[str] = []
for session_path in session_files:
try:
session = self._load_json(session_path)
except (ValueError, OSError) as exc:
self.logger.warning(f"[{self.name}] skip {session_path.name}: {exc}")
continue
session_id = str(session.get("haystack_session_id") or session_path.stem)
session_date = str(session.get("haystack_date") or "").strip()
session_dt = self._parse_lme_datetime(session_date)
if session_dt is not None and session_dt > question_dt:
session_ids_illegal.append(session_id)
filtered_sessions.append(
{
"session_id": session_id,
"session_date": session_date,
"session_file": session_path.name,
"reason": "session_date_after_question_date",
},
)
continue
if session_dt is None:
self.logger.warning(
f"[{self.name}] keep {session_id}: cannot parse haystack_date={session_date!r}",
)
day = session_dt.strftime("%Y-%m-%d") if session_dt is not None else (self._parse_day(session_date) or "")
sessions.append((session, session_path, session_id, session_date, day))
daily_dir = self.config_value("daily_dir")
resource_dir = self._resource_dir_name()
start_interval_seconds = float(self.kwargs.get("start_interval_seconds", START_INTERVAL_SECONDS))
if start_interval_seconds < 0:
start_interval_seconds = START_INTERVAL_SECONDS
concurrency = int(self.kwargs.get("concurrency", MAX_CONCURRENCY))
if concurrency <= 0:
concurrency = MAX_CONCURRENCY
concurrency = min(concurrency, MAX_CONCURRENCY)
total = len(sessions)
self.logger.info(
f"[{self.name}] extracting {total} sessions from {session_dir} "
f"(filtered {len(session_ids_illegal)} sessions after question_date, "
f"start_interval={start_interval_seconds}s, concurrency={concurrency})",
)
self._reserved.clear()
failed_extracts: list[dict] = []
retry_initial_seconds = float(self.kwargs.get("retry_initial_seconds", RETRY_INITIAL_SECONDS))
retry_max_seconds = float(self.kwargs.get("retry_max_seconds", RETRY_MAX_SECONDS))
retry_max_attempts_raw = self.kwargs.get("retry_max_attempts")
retry_max_attempts = int(retry_max_attempts_raw) if retry_max_attempts_raw not in (None, "") else 0
if retry_initial_seconds <= 0:
retry_initial_seconds = RETRY_INITIAL_SECONDS
retry_max_seconds = max(retry_max_seconds, retry_initial_seconds)
retry_gate = asyncio.Condition()
retry_sleeping_extract_idxs: set[int] = set()
submit_lock = asyncio.Lock()
last_submitted_at = 0.0
semaphore = asyncio.Semaphore(concurrency)
def has_prior_retry_sleeping(idx: int) -> bool:
return any(retry_idx < idx for retry_idx in retry_sleeping_extract_idxs)
async def wait_for_start_slot() -> None:
nonlocal last_submitted_at
async with submit_lock:
sleep_seconds = last_submitted_at + start_interval_seconds - time.monotonic()
if sleep_seconds > 0:
await asyncio.sleep(sleep_seconds)
last_submitted_at = time.monotonic()
async def wait_for_healthy_start_slot(idx: int, session_id: str) -> None:
while True:
async with retry_gate:
if has_prior_retry_sleeping(idx):
self.logger.info(
f"[{self.name}] ({idx}/{total}) {session_id} waits for earlier retry sleep",
)
await retry_gate.wait_for(lambda: not has_prior_retry_sleeping(idx))
await wait_for_start_slot()
async with retry_gate:
if not has_prior_retry_sleeping(idx):
return
async def mark_retry_sleeping(idx: int) -> None:
async with retry_gate:
retry_sleeping_extract_idxs.add(idx)
retry_gate.notify_all()
async def mark_retry_awake(idx: int) -> None:
async with retry_gate:
retry_sleeping_extract_idxs.discard(idx)
retry_gate.notify_all()
async def reply_with_retry(idx: int, user_prompt: str, session_id: str) -> dict:
attempt = 1
sleep_seconds = retry_initial_seconds
while True:
try:
await wait_for_healthy_start_slot(idx, session_id)
result = await self.agent_wrapper.reply(
user_prompt,
system_prompt=self.get_prompt("system_prompt"),
output_schema=_MEMORY_SCHEMA,
)
if not isinstance(result.get("structured_output"), dict):
raise ValueError("agent reply missing structured_output")
await mark_retry_awake(idx)
if attempt > 1:
self.logger.info(f"[{self.name}] extract recovered for {session_id} after {attempt} attempts")
return result
except Exception as exc:
if self._is_data_inspection_error(exc):
await mark_retry_awake(idx)
raise
if 0 < retry_max_attempts <= attempt:
await mark_retry_awake(idx)
raise
await mark_retry_sleeping(idx)
next_sleep = min(sleep_seconds, retry_max_seconds)
self.logger.warning(
f"[{self.name}] extract attempt {attempt} failed for {session_id}: {exc}; "
f"retrying in {next_sleep:.1f}s",
)
await asyncio.sleep(next_sleep)
await mark_retry_awake(idx)
sleep_seconds = min(sleep_seconds * 2, retry_max_seconds)
attempt += 1
async def extract_one(
idx: int,
session: dict,
session_path: Path,
session_id: str,
session_date: str,
day: str,
) -> dict | None:
if not day:
self.logger.warning(f"[{self.name}] skip {session_id}: unparseable date {session_date!r}")
return None
messages = session.get("messages") or []
user_prompt = self.prompt_format(
"user_message",
session_id=session_id,
session_date=session_date,
messages=self._format_messages(messages),
)
try:
result = await reply_with_retry(idx, user_prompt, session_id)
except Exception as exc: # noqa: BLE001 — one bad session must not abort the sweep
if self._is_data_inspection_error(exc):
self.logger.warning(
f"[{self.name}] extract fallback for {session_id}: non-retryable data inspection error",
)
failed_extracts.append(
{
"session_id": session_id,
"session_date": session_date,
"session_file": session_path.name,
"error": str(exc),
"non_retryable": True,
"fallback": True,
"fallback_reason": "data_inspection_failed",
"raw_session": session,
},
)
return None
self.logger.warning(f"[{self.name}] extract failed for {session_id}: {exc}")
failed_extracts.append(
{
"session_id": session_id,
"session_date": session_date,
"session_file": session_path.name,
"error": str(exc),
"non_retryable": False,
"fallback": False,
},
)
return None
extracted = result.get("structured_output")
name = description = body = ""
if isinstance(extracted, dict):
name = str(extracted.get("name") or "").strip()
description = str(extracted.get("description") or "").strip()
body = str(extracted.get("body") or "").strip()
if not isinstance(extracted, dict) or not name or not body:
if isinstance(extracted, dict):
self.logger.info(f"[{self.name}] empty extraction for {session_id}; skipping")
else:
self.logger.warning(f"[{self.name}] no structured output for {session_id}; skipping")
return None
unique_name = await self._reserve_name(daily_dir, day, name, session_id)
rel_path = f"{daily_dir}/{day}/{unique_name}.md"
post = frontmatter.Post(
body,
name=unique_name,
description=description,
session_id=session_id,
session_date=session_date,
source=f"[[{resource_dir}/{session_path.name}]]",
)
abs_path = self.workspace_path / rel_path
abs_path.parent.mkdir(parents=True, exist_ok=True)
abs_path.write_text(frontmatter.dumps(post), encoding="utf-8")
self.logger.info(f"[{self.name}] ({idx}/{total}) {session_id} -> {rel_path}")
return {"session_id": session_id, "date": day, "path": rel_path}
async def extract_one_limited(
idx: int,
session: dict,
session_path: Path,
session_id: str,
session_date: str,
day: str,
) -> dict | None:
async with semaphore:
return await extract_one(idx, session, session_path, session_id, session_date, day)
results = await asyncio.gather(
*(
extract_one_limited(idx, session, session_path, session_id, session_date, day)
for idx, (session, session_path, session_id, session_date, day) in enumerate(sessions, start=1)
),
)
written = [r for r in results if r is not None]
fallback_extracts = [e for e in failed_extracts if e.get("fallback")]
self.context.response.success = True
self.context.response.answer = f"wrote {len(written)}/{total} session notes"
self.context.response.metadata.update(
{
"num_sessions": total,
"num_session_files": len(session_files),
"num_written": len(written),
"num_failed_extracts": len(failed_extracts),
"num_fallback_extracts": len(fallback_extracts),
"num_filtered_sessions": len(session_ids_illegal),
"session_ids_illegal": session_ids_illegal,
"filtered_sessions": filtered_sessions,
"failed_extracts": failed_extracts,
"fallback_extracts": fallback_extracts,
"notes": written,
},
)
return self.context.response

45
reme/steps/benchmark/lme/auto_memory.yaml Normal file
View file

@ -0,0 +1,45 @@
system_prompt: |
You are a memory extractor for the LongMemEval benchmark. You are given ONE chat session
between a user and an assistant. Your job is to extract its content COMPLETELY into a single
daily memory note that will later be retrieved by both BM25 keyword search and dense vector
search. Completeness and retrievability are the only goals — do NOT compress or editorialize.
## What to extract — everything that could ever be asked later
Capture every core fact stated or clearly implied in the session, including but not limited to:
- Facts about the user: identity, preferences, habits, possessions, relationships, plans, goals.
- Events and actions: what happened, what was decided, what the user did or intends to do.
- Entities: people, places, organizations, products, titles, brands — with their exact names.
- Numbers and quantities: durations, distances, prices, counts, measurements — verbatim.
- Time information ABOVE ALL: absolute dates, weekdays, and relative expressions ("last week",
"since January 15th", "every morning", "for 3 years"). Always keep the fact together with its
time expression, and when possible also anchor it to the session date.
Do not invent anything. Only record what is actually in the session. If the session is pure
small talk with no durable facts, still produce a minimal faithful note (do not fabricate).
## How to write the body — optimize for search
- Write plain natural-language declarative sentences (one fact per sentence or bullet). Dense
retrievers embed sentences well; BM25 matches exact tokens — so both benefit from full,
unabbreviated wording.
- State entities, dates and numbers explicitly and verbatim; expand abbreviations and also
include common aliases/synonyms the user or assistant used, so keyword search can hit them.
- Prefer the user's own key phrasing at important points (quote short fragments verbatim).
- Use markdown structure (headings/bullets) freely, but never drop a fact for the sake of brevity.
## Frontmatter fields you return
- `name`: a concise, stable, kebab-case topic/event stem (no date, no slashes or reserved
characters). It is only a filename — the searchable content lives in `description` and `body`.
- `description`: a thorough, search-friendly abstract that on its own conveys all key facts.
- `body`: the complete extraction as described above.
user_message: |
Session id: {session_id}
Session date: {session_date}
--- Session messages ---
{messages}
--- End of session ---
Extract this session COMPLETELY into one search-friendly daily note. Keep every entity, number,
and (above all) every time expression verbatim, each attached to the fact it belongs to. Return
`name`, `description`, and `body`.

103
reme/steps/benchmark/lme/extract_session.py Normal file
View file

@ -0,0 +1,103 @@
"""extract_session_by_id — deep-read one raw session, keyed by its session_id.
This is the hand-written function tool the answering agent sees. Search results
surface a note's ``session_id``; when a hit looks relevant, the agent passes that
``session_id`` here. The step resolves the question/time from ``query.json``,
locates the raw session file (named ``<date>_(...)_<time>@<session_id>.json``
under ``resource_dir``), loads its messages, and asks an agent to extract —
completely and verbatim — every part of that session relevant to the question.
"""
import json
from pathlib import Path
from ...base_step import BaseStep
from ....components import R
@R.register("lme_extract_session_step")
class LmeExtractSessionStep(BaseStep):
"""Resolve a session_id to raw content, then deep-read it for the question."""
def _resource_dir_name(self) -> str:
return self.app_context.app_config.resource_dir if self.app_context is not None else "session"
def _load_query(self) -> dict:
path = self.workspace_path / "query.json"
with path.open(encoding="utf-8") as f:
data = json.load(f)
if not isinstance(data, dict):
raise ValueError("query.json is not a JSON object")
return data
def _find_session_file(self, session_id: str) -> Path | None:
session_dir = self.workspace_path / self._resource_dir_name()
if not session_dir.is_dir():
return None
# Files are named "<date>_(...)_<time>@<session_id>.json".
matches = list(session_dir.glob(f"*@{session_id}.json"))
if matches:
return matches[0]
# Fall back to a plain "<session_id>.json" naming.
direct = session_dir / f"{session_id}.json"
return direct if direct.is_file() else None
async def execute(self):
assert self.context is not None
if self.agent_wrapper is None:
raise ValueError("lme_extract_session_step requires agent_wrapper")
session_id: str = str(self.context.get("session_id", "") or "").strip()
if not session_id:
self.context.response.success = False
self.context.response.answer = "Error: session_id is required"
return self.context.response
try:
query = self._load_query()
except (OSError, ValueError) as exc:
self.context.response.success = False
self.context.response.answer = f"Error: cannot read query.json: {exc}"
return self.context.response
question = str(query.get("question", "") or "").strip()
question_time = str(query.get("question_date", "") or "").strip()
session_path = self._find_session_file(session_id)
if session_path is None:
self.context.response.success = False
self.context.response.answer = (
f"Error: no session file found for session_id={session_id!r}. "
"Use a session_id shown in a search result."
)
return self.context.response
try:
with session_path.open(encoding="utf-8") as f:
session = json.load(f)
except (OSError, ValueError) as exc:
self.context.response.success = False
self.context.response.answer = f"Error: cannot read session {session_path.name}: {exc}"
return self.context.response
messages = session.get("messages") if isinstance(session, dict) else None
session_content = json.dumps(messages or session, ensure_ascii=False, indent=2)
user_prompt = self.prompt_format(
"user_message",
question=question or "(unknown)",
question_time=question_time or "(unknown)",
session_content=session_content,
)
result = await self.agent_wrapper.reply(
user_prompt,
system_prompt=self.get_prompt("system_prompt"),
)
answer = (result.get("result") or "").strip()
self.logger.info(f"[{self.name}] extracted {len(answer)} chars for session_id={session_id!r}")
self.context.response.success = True
self.context.response.answer = answer
self.context.response.metadata.update(
{"session_id": session_id, "session_file": session_path.name},
)
return self.context.response

27
reme/steps/benchmark/lme/extract_session.yaml Normal file
View file

@ -0,0 +1,27 @@
system_prompt: |
You are a precise evidence extractor for the LongMemEval benchmark. You are given ONE full
chat session (raw messages) together with a target question and the time the question was asked.
Your job is to extract, COMPLETELY and VERBATIM, every part of this session that is relevant to
answering the question.
Rules:
- Extract the actual content from the session — quote the relevant user/assistant statements as
they appear. Do not summarize away details, and do not invent anything not in the session.
- Preserve all time information exactly: absolute dates, weekdays, and relative expressions
("last week", "since January 15th", "every day", durations, frequencies). Keep each fact
together with its time expression, and relate it to the question time when that matters
(e.g. a fact stated before the question date is valid evidence; note any date conflicts).
- Keep entities, names, and numbers verbatim.
- If, after reading the whole session, nothing in it is relevant to the question, reply with
exactly: NOT RELEVANT
user_message: |
Question: {question}
Question asked at: {question_time}
--- Full session content ---
{session_content}
--- End of session ---
Extract everything in this session that is relevant to answering the question, verbatim and with
all time information preserved. If nothing is relevant, reply exactly: NOT RELEVANT

201
reme/steps/benchmark/lme/golden_check.py Normal file
View file

@ -0,0 +1,201 @@
"""Judge whether the LongMemEval golden answer is reasonable.
Consumes ``session_review.json`` produced by ``lme_session_review_step`` and
hands its extracted session information to an agent that is equipped with the
``python_execute`` tool. The agent uses ``python_execute`` only as a scratchpad
for the hard reasoning (checking the golden answer and cross-checking the
filtered ``answer_session_ids``); the final verdict is not the
raw Python stdout but a *structured* object extracted from the whole conversation
via ``output_schema``. Sessions dated after ``question_date`` are filtered
upstream by ``lme_session_review_step`` and are not included in this
golden-check flow.
The output ``check_golden.json`` is intentionally slim: it does NOT duplicate the
query/golden/review fields already stored in ``session_review.json`` (referenced by
path), keeping only the relevant per-session ``session_summaries`` and the
structured verdict. It is written to the workspace root (e.g.
``datasets/longmemeval/1/check_golden.json``).
"""
import json
import asyncio
from uuid import uuid4
from ...base_step import BaseStep
from ....components import R
# File written under the workspace root with the full review + verdict payload.
OUTPUT_FILENAME = "check_golden.json"
SESSION_REVIEW_FILENAME = "session_review.json"
RETRY_INITIAL_SECONDS = 5.0
RETRY_MAX_SECONDS = 300.0
# Structured verdict the judge agent must produce (extracted from its reasoning).
_VERDICT_SCHEMA = {
"type": "object",
"properties": {
"reasoning": {
"type": "string",
"description": "用中文写出详细的推理过程：先说明证据支持的答案，再逐步判断 golden_answer "
"是否正确，以及 answer_session_ids 是否恰好正确。",
},
"golden_answer_correct": {
"type": "boolean",
"description": "golden_answer 是否正确。",
},
"true_answer": {
"type": "string",
"description": "仅当 golden_answer_correct 为 false 时填写：证据支持的正确答案（证据不足时填 "
"'unknown'）。golden_answer_correct 为 true 时填空字符串。",
},
"answer_session_ids_correct": {
"type": "boolean",
"description": "answer_session_ids 是否恰好是支持答案所需的会话（多、少、无关的 id 都算错误）。",
},
"true_answer_session_ids": {
"type": "array",
"items": {"type": "string"},
"description": "仅当 answer_session_ids_correct 为 false 时填写：真正支持答案的 session id 列表。"
"answer_session_ids_correct 为 true 时填空列表。",
},
},
"required": [
"reasoning",
"golden_answer_correct",
"true_answer",
"answer_session_ids_correct",
"true_answer_session_ids",
],
"additionalProperties": False,
}
@R.register("lme_golden_check_step")
class GoldenCheckStep(BaseStep):
"""Let a python-enabled agent decide whether the golden answer holds up."""
@staticmethod
def _compact_summary(summary: dict) -> dict:
"""Keep only the evidence fields used by the golden-check prompt."""
return {
"session_id": str(summary.get("session_id") or ""),
"session_date": str(summary.get("session_date") or ""),
"extracted_info": str(summary.get("extracted_info") or ""),
}
async def execute(self):
assert self.context is not None
if self.agent_wrapper is None:
raise ValueError("lme_golden_check_step requires agent_wrapper")
review_path = self.workspace_path / SESSION_REVIEW_FILENAME
if not review_path.is_file():
raise FileNotFoundError(
f"{SESSION_REVIEW_FILENAME} not found at {review_path}; run lme_session_review_step first",
)
try:
with review_path.open(encoding="utf-8") as f:
review_payload = json.load(f)
except json.JSONDecodeError as exc:
raise ValueError(f"Invalid JSON in {review_path}") from exc
if not isinstance(review_payload, dict):
raise ValueError(f"Expected a JSON object in {review_path}")
query = review_payload.get("query") or {}
golden = review_payload.get("golden") or {}
# session_review.json keeps one extraction per reviewed session.
session_summaries = [self._compact_summary(s) for s in review_payload.get("session_summaries") or []]
question = str(query.get("question") or "").strip()
question_type = str(query.get("question_type") or "").strip()
question_date = str(query.get("question_date") or "").strip()
golden_answer = str(golden.get("answer") or "").strip()
answer_session_ids = golden.get("answer_session_ids_filter_illegal") or []
if not question:
raise ValueError(f"{review_path} does not contain a question")
prompt_input = {
"question": question,
"question_type": question_type,
"question_date": question_date,
"golden_answer": golden_answer,
"answer_session_ids": answer_session_ids,
"session_summaries": session_summaries,
}
user_prompt = self.prompt_format(
"user_message",
question=question,
question_type=question_type,
question_date=question_date,
golden_answer=golden_answer,
answer_session_ids=", ".join(str(s) for s in answer_session_ids) or "(none)",
num_session_summaries=len(session_summaries),
payload_json=json.dumps(prompt_input, ensure_ascii=False, indent=2),
)
tool_context_id = str(self.context.get("tool_context_id") or f"lme-golden-{uuid4()}")
retry_initial_seconds = float(self.kwargs.get("retry_initial_seconds", RETRY_INITIAL_SECONDS))
retry_max_seconds = float(self.kwargs.get("retry_max_seconds", RETRY_MAX_SECONDS))
retry_max_attempts_raw = self.kwargs.get("retry_max_attempts")
retry_max_attempts = int(retry_max_attempts_raw) if retry_max_attempts_raw not in (None, "") else 0
if retry_initial_seconds <= 0:
retry_initial_seconds = RETRY_INITIAL_SECONDS
retry_max_seconds = max(retry_max_seconds, retry_initial_seconds)
attempt = 1
sleep_seconds = retry_initial_seconds
while True:
try:
result = await self.agent_wrapper.reply(
user_prompt,
system_prompt=self.get_prompt("system_prompt"),
tool_context_id=tool_context_id,
output_schema=_VERDICT_SCHEMA,
)
if attempt > 1:
self.logger.info(f"[{self.name}] golden check recovered after {attempt} attempts")
break
except Exception as exc:
if 0 < retry_max_attempts <= attempt:
raise
next_sleep = min(sleep_seconds, retry_max_seconds)
self.logger.warning(
f"[{self.name}] golden check attempt {attempt} failed: {exc}; " f"retrying in {next_sleep:.1f}s",
)
await asyncio.sleep(next_sleep)
sleep_seconds = min(sleep_seconds * 2, retry_max_seconds)
attempt += 1
# The structured verdict is the real output; the free-text reply is only the
# agent's closing narration and is kept as a fallback.
verdict = result.get("structured_output")
if not isinstance(verdict, dict):
self.logger.warning(f"[{self.name}] no structured verdict; falling back to free text")
verdict = {"reasoning": (result.get("result") or "").strip()}
# Slim output: do NOT duplicate session_review.json (referenced by path);
# keep only the compact session_summaries and the verdict.
output = {
"session_review_path": str(review_path),
"session_summaries": session_summaries,
"verdict": verdict,
}
output_path = self.workspace_path / OUTPUT_FILENAME
with output_path.open("w", encoding="utf-8") as f:
json.dump(output, f, ensure_ascii=False, indent=2)
self.logger.info(f"[{self.name}] wrote verdict to {output_path}")
self.context.response.success = True
self.context.response.answer = json.dumps(verdict, ensure_ascii=False, indent=2)
self.context.response.metadata.update(
{
"num_session_summaries": len(session_summaries),
"session_review_path": str(review_path),
"tool_context_id": tool_context_id,
"agent_session_id": result.get("session_id"),
"output_path": str(output_path),
"verdict": verdict,
},
)
return self.context.response

31
reme/steps/benchmark/lme/golden_check.yaml Normal file
View file

@ -0,0 +1,31 @@
system_prompt: |
你是 LongMemEval 基准测试的审核员。你要根据从用户聊天记录中提取的证据，判断某个问题的
golden_answer 是否正确，以及 answer_session_ids 是否恰好正确。
你可以使用 python_execute 工具作为推理草稿本：统计相关会话、抽取答案、交叉核对
answer_session_ids。把给定的数据以字面量形式直接嵌入 Python 代码，使计算可复现；把中间结果
以 JSON 打印出来，便于审计。
python 的 stdout 不是你的最终答案，只是草稿。计算充分、确信之后，停止调用 python，用中文
给出结论。最终的结构化结果会从整段对话中自动抽取，所以务必把推理和结论清楚表达。
结构化输出要求：
- reasoning 必须是详细的中文推理过程。
- true_answer 仅在 golden_answer_correct 为 false 时填写，为空字符串否则。
- true_answer_session_ids 仅在 answer_session_ids_correct 为 false 时填写，为空列表否则。
user_message: |
Question: {question}
Question type: {question_type}
Question date: {question_date}
Golden answer: {golden_answer}
Answer session ids: {answer_session_ids}
Number of session extractions included: {num_session_summaries}
证据（JSON）。下方 answer_session_ids 已是 session_review.json 中的
answer_session_ids_filter_illegal；session_summaries 含上游审核过的会话，每条只有 session_id、
session_date、extracted_info：
{payload_json}
用 python_execute 统计和推理，过程中打印中间 JSON。然后用中文给出最终结论：golden_answer
是否正确（不正确时给出 true_answer），answer_session_ids 是否正确（不正确时给出真正的 id）。

81
reme/steps/benchmark/lme/llm_judge.py
View file

@ -1,81 +0,0 @@
"""Judge whether an agent answer matches the golden answer."""
import re
from ...base_step import BaseStep
from ....components import R
@R.register("answer_judge_step")
class AnswerJudgeStep(BaseStep):
"""Evaluate whether an agent answer is correct against a golden answer."""
PROMPT_KEYS_BY_QUESTION_TYPE = {
"temporal_reasoning": "temporal_reasoning_system_prompt",
"knowledge_update": "knowledge_update_system_prompt",
"single_session_preference": "single_session_preference_system_prompt",
}
@classmethod
def _judge_prompt_key(cls, question_type: str) -> str:
normalized = question_type.strip().lower().replace("-", "_").replace(" ", "_")
return cls.PROMPT_KEYS_BY_QUESTION_TYPE.get(normalized, "other_question_types_system_prompt")
@staticmethod
def _normalize_judgement(raw_answer: str) -> str:
match = re.match(r"\s*(yes|no)\b", raw_answer, re.IGNORECASE)
if match:
return match.group(1).lower()
return raw_answer.strip().lower()
async def execute(self):
assert self.context is not None
query: str = self.context.get("query", "")
agent_answer: str = self.context.get("agent_answer", "")
golden_answer: str = self.context.get("golden_answer", "")
question_type: str = self.context.get("question_type", "")
if not query:
raise ValueError("answer_judge_step requires non-empty query")
if not agent_answer:
raise ValueError("answer_judge_step requires non-empty agent_answer")
if not golden_answer:
raise ValueError("answer_judge_step requires non-empty golden_answer")
if self.agent_wrapper is None:
raise RuntimeError("answer_judge_step requires agent_wrapper")
judge_prompt_key = self._judge_prompt_key(question_type)
user_prompt_key = (
"preference_judge_user_message"
if judge_prompt_key == "single_session_preference_system_prompt"
else "answer_judge_user_message"
)
user_prompt = self.prompt_format(
user_prompt_key,
query=query,
golden_answer=golden_answer,
agent_answer=agent_answer,
)
result = await self.agent_wrapper.reply(
user_prompt,
system_prompt=self.prompt_format(judge_prompt_key),
)
raw_answer = (result.get("result") or "").strip()
answer = self._normalize_judgement(raw_answer)
self.logger.info(f"[{self.name}] answer judgement: {answer}")
self.context["answer_judgement"] = answer
self.context.response.success = True
self.context.response.answer = answer
self.context.response.metadata.update(
{
"query": query,
"agent_answer": agent_answer,
"golden_answer": golden_answer,
"question_type": question_type,
"answer_judgement": answer,
"raw_answer_judgement": raw_answer,
},
)
return self.context.response

109
reme/steps/benchmark/lme/lme_llm_judge.py Normal file
View file

@ -0,0 +1,109 @@
"""LME file-based LLM judge step."""
import json
import re
from ...base_step import BaseStep
from ....components import R
@R.register("lme_llm_judge_step")
class LmeLlmJudgeStep(BaseStep):
"""Judge ``mem_answer.json`` against ``answer.json`` and update it in place."""
PROMPT_KEYS_BY_QUESTION_TYPE = {
"temporal_reasoning": "temporal_reasoning_system_prompt",
"knowledge_update": "knowledge_update_system_prompt",
"single_session_preference": "single_session_preference_system_prompt",
}
@classmethod
def _judge_prompt_key(cls, question_type: str) -> str:
normalized = question_type.strip().lower().replace("-", "_").replace(" ", "_")
return cls.PROMPT_KEYS_BY_QUESTION_TYPE.get(normalized, "other_question_types_system_prompt")
@staticmethod
def _user_prompt_key(judge_prompt_key: str) -> str:
if judge_prompt_key == "single_session_preference_system_prompt":
return "preference_judge_user_message"
return "answer_judge_user_message"
@staticmethod
def _normalize_judgement(raw_answer: str) -> str:
match = re.match(r"\s*(yes|no)\b", raw_answer, re.IGNORECASE)
if match:
return match.group(1).lower()
return raw_answer.strip().lower()
def _load_json_object(self, filename: str) -> dict:
path = self.workspace_path / filename
if not path.exists():
raise FileNotFoundError(f"{filename} does not exist in {self.workspace_path}")
with path.open(encoding="utf-8") as f:
data = json.load(f)
if not isinstance(data, dict):
raise ValueError(f"{filename} is not a JSON object")
return data
async def execute(self):
assert self.context is not None
if self.agent_wrapper is None:
raise RuntimeError("lme_llm_judge_step requires agent_wrapper")
query_data = self._load_json_object("query.json")
golden_data = self._load_json_object("answer.json")
mem_answer = self._load_json_object("mem_answer.json")
query = str(query_data.get("question", "") or "").strip()
agent_answer = str(mem_answer.get("answer", "") or "").strip()
golden_answer = str(golden_data.get("answer", "") or "").strip()
question_type = str(query_data.get("question_type", "") or "")
if not query:
raise ValueError("query.json requires a non-empty 'question'")
if not agent_answer:
raise ValueError("mem_answer.json requires a non-empty 'answer'")
if not golden_answer:
raise ValueError("answer.json requires a non-empty 'answer'")
judge_prompt_key = self._judge_prompt_key(question_type)
user_prompt = self.prompt_format(
self._user_prompt_key(judge_prompt_key),
query=query,
golden_answer=golden_answer,
agent_answer=agent_answer,
)
result = await self.agent_wrapper.reply(
user_prompt,
system_prompt=self.prompt_format(judge_prompt_key),
)
raw_answer = (result.get("result") or "").strip()
answer = self._normalize_judgement(raw_answer)
mem_answer["llm_judge"] = {
"judgement": answer,
"raw_judgement": raw_answer,
"golden_answer": golden_answer,
"question_type": question_type,
}
out_path = self.workspace_path / "mem_answer.json"
out_path.write_text(json.dumps(mem_answer, ensure_ascii=False, indent=2), encoding="utf-8")
question_id = str(query_data.get("question_id") or mem_answer.get("question_id") or "")
self.logger.info(f"[{self.name}] llm judgement for {question_id or query!r}: {answer}")
self.context.response.success = True
self.context.response.answer = answer
self.context.response.metadata.update(
{
"question_id": question_id,
"query": query,
"agent_answer": agent_answer,
"golden_answer": golden_answer,
"question_type": question_type,
"answer_judgement": answer,
"raw_answer_judgement": raw_answer,
"path": "mem_answer.json",
},
)
return self.context.response

0
reme/steps/benchmark/lme/llm_judge.yaml → reme/steps/benchmark/lme/lme_llm_judge.yaml
View file

358
reme/steps/benchmark/lme/session_review.py Normal file
View file

@ -0,0 +1,358 @@
"""Review every LongMemEval session and extract its information.
For a workspace such as ``datasets/longmemeval/1`` this step loads ``query.json``
and ``answer.json``, filters out sessions dated after ``question_date``, then
walks each remaining session under ``resource_dir`` one by one. An agent wrapper
extracts the complete information in each session, with extra care not to omit
anything related to the question or golden answer.
The collected per-session extractions are written to ``session_review.json`` for
the downstream golden-answer check.
"""
import asyncio
import json
import re
import time
from datetime import datetime
from pathlib import Path
from ...base_step import BaseStep
from ....components import R
START_INTERVAL_SECONDS = 1.0
MAX_CONCURRENCY = 60
RETRY_INITIAL_SECONDS = 5.0
RETRY_MAX_SECONDS = 300.0
OUTPUT_FILENAME = "session_review.json"
_LME_DATETIME_RE = re.compile(r"(\d{4})/(\d{2})/(\d{2}).*?(\d{2}):(\d{2})")
_NON_RETRYABLE_DATA_INSPECTION_MARKERS = (
"data_inspection_failed",
"DataInspectionFailed",
"Input text data may contain inappropriate content",
)
@R.register("lme_session_review_step")
class SessionReviewStep(BaseStep):
"""Extract complete information from every eligible session."""
def _load_json(self, path: Path | str) -> dict:
if not isinstance(path, Path):
path = self.workspace_path / path
try:
with path.open(encoding="utf-8") as f:
data = json.load(f)
except OSError as exc:
raise FileNotFoundError(f"Cannot read LongMemEval file: {path}") from exc
except json.JSONDecodeError as exc:
raise ValueError(f"Invalid JSON in LongMemEval file: {path}") from exc
if not isinstance(data, dict):
raise ValueError(f"Expected a JSON object in {path}")
return data
def _session_dir(self) -> Path:
resource_dir = self.app_context.app_config.resource_dir if self.app_context is not None else "session"
return self.workspace_path / resource_dir
@staticmethod
def _parse_lme_datetime(raw_date: str) -> datetime | None:
"""Parse LongMemEval timestamps like ``2023/05/20 (Sat) 03:29``."""
match = _LME_DATETIME_RE.search(raw_date.strip())
if match is None:
return None
try:
year, month, day, hour, minute = (int(part) for part in match.groups())
return datetime(year, month, day, hour, minute)
except ValueError:
return None
@staticmethod
def _is_data_inspection_error(exc: Exception) -> bool:
text = str(exc)
return any(marker in text for marker in _NON_RETRYABLE_DATA_INSPECTION_MARKERS)
# pylint: disable=too-many-statements
async def execute(self):
assert self.context is not None
if self.agent_wrapper is None:
raise ValueError("lme_session_review_step requires agent_wrapper")
query_data = self._load_json("query.json")
answer_data = self._load_json("answer.json")
question = str(query_data.get("question") or "").strip()
question_type = str(query_data.get("question_type") or "").strip()
question_date = str(query_data.get("question_date") or "").strip()
if not question:
raise ValueError("query.json requires a non-empty 'question'")
question_dt = self._parse_lme_datetime(question_date)
if question_dt is None:
raise ValueError(f"query.json has an invalid 'question_date': {question_date!r}")
golden_answer = str(answer_data.get("answer") or "").strip()
answer_session_ids = [str(s) for s in (answer_data.get("answer_session_ids") or [])]
session_dir = self._session_dir()
if not session_dir.is_dir():
raise FileNotFoundError(f"Session directory not found: {session_dir}")
session_files = sorted(p for p in session_dir.iterdir() if p.suffix == ".json")
sessions: list[tuple[dict, str, str]] = []
filtered_sessions: list[dict] = []
session_ids_illegal: list[str] = []
answer_session_ids_illegal: list[str] = []
answer_session_id_set = set(answer_session_ids)
for session_path in session_files:
try:
session = self._load_json(session_path)
except (ValueError, FileNotFoundError) as exc:
self.logger.warning(f"[{self.name}] skip {session_path.name}: {exc}")
continue
session_id = str(session.get("haystack_session_id") or session_path.stem)
session_date = str(session.get("haystack_date") or "").strip()
session_dt = self._parse_lme_datetime(session_date)
if session_dt is not None and session_dt > question_dt:
session_ids_illegal.append(session_id)
filtered_sessions.append(
{
"session_id": session_id,
"session_date": session_date,
"session_file": session_path.name,
"reason": "session_date_after_question_date",
},
)
if session_id in answer_session_id_set:
answer_session_ids_illegal.append(session_id)
continue
if session_dt is None:
self.logger.warning(
f"[{self.name}] keep {session_id}: cannot parse haystack_date={session_date!r}",
)
sessions.append((session, session_id, session_date))
illegal_answer_session_ids = set(answer_session_ids_illegal)
answer_session_ids_filter_illegal = [
session_id for session_id in answer_session_ids if session_id not in illegal_answer_session_ids
]
total = len(sessions)
start_interval_seconds = float(self.kwargs.get("start_interval_seconds", START_INTERVAL_SECONDS))
if start_interval_seconds < 0:
start_interval_seconds = START_INTERVAL_SECONDS
concurrency = int(self.kwargs.get("concurrency", MAX_CONCURRENCY))
if concurrency <= 0:
concurrency = MAX_CONCURRENCY
concurrency = min(concurrency, MAX_CONCURRENCY)
self.logger.info(
f"[{self.name}] reviewing {total} sessions from {session_dir} "
f"(filtered {len(session_ids_illegal)} sessions after question_date, "
f"start_interval={start_interval_seconds}s, concurrency={concurrency})",
)
failed_reviews: list[dict] = []
retry_initial_seconds = float(self.kwargs.get("retry_initial_seconds", RETRY_INITIAL_SECONDS))
retry_max_seconds = float(self.kwargs.get("retry_max_seconds", RETRY_MAX_SECONDS))
retry_max_attempts_raw = self.kwargs.get("retry_max_attempts")
retry_max_attempts = int(retry_max_attempts_raw) if retry_max_attempts_raw not in (None, "") else 0
if retry_initial_seconds <= 0:
retry_initial_seconds = RETRY_INITIAL_SECONDS
retry_max_seconds = max(retry_max_seconds, retry_initial_seconds)
retry_gate = asyncio.Condition()
retry_sleeping_review_idxs: set[int] = set()
submit_lock = asyncio.Lock()
last_submitted_at = 0.0
def has_prior_retry_sleeping(idx: int) -> bool:
return any(retry_idx < idx for retry_idx in retry_sleeping_review_idxs)
async def wait_for_start_slot() -> None:
nonlocal last_submitted_at
async with submit_lock:
sleep_seconds = last_submitted_at + start_interval_seconds - time.monotonic()
if sleep_seconds > 0:
await asyncio.sleep(sleep_seconds)
last_submitted_at = time.monotonic()
async def wait_for_healthy_start_slot(idx: int, session_id: str) -> None:
while True:
async with retry_gate:
if has_prior_retry_sleeping(idx):
self.logger.info(
f"[{self.name}] ({idx}/{total}) {session_id} waits for earlier retry sleep",
)
await retry_gate.wait_for(lambda: not has_prior_retry_sleeping(idx))
await wait_for_start_slot()
async with retry_gate:
if not has_prior_retry_sleeping(idx):
return
async def mark_retry_sleeping(idx: int) -> None:
async with retry_gate:
retry_sleeping_review_idxs.add(idx)
retry_gate.notify_all()
async def mark_retry_awake(idx: int) -> None:
async with retry_gate:
retry_sleeping_review_idxs.discard(idx)
retry_gate.notify_all()
async def reply_with_retry(idx: int, user_prompt: str, session_id: str) -> dict:
attempt = 1
sleep_seconds = retry_initial_seconds
while True:
try:
await wait_for_healthy_start_slot(idx, session_id)
result = await self.agent_wrapper.reply(
user_prompt,
system_prompt=self.get_prompt("system_prompt"),
)
await mark_retry_awake(idx)
if attempt > 1:
self.logger.info(f"[{self.name}] review recovered for {session_id} after {attempt} attempts")
return result
except Exception as exc:
if self._is_data_inspection_error(exc):
await mark_retry_awake(idx)
raise
if 0 < retry_max_attempts <= attempt:
await mark_retry_awake(idx)
raise
await mark_retry_sleeping(idx)
next_sleep = min(sleep_seconds, retry_max_seconds)
self.logger.warning(
f"[{self.name}] review attempt {attempt} failed for {session_id}: {exc}; "
f"retrying in {next_sleep:.1f}s",
)
await asyncio.sleep(next_sleep)
await mark_retry_awake(idx)
sleep_seconds = min(sleep_seconds * 2, retry_max_seconds)
attempt += 1
async def review_one(idx: int, session: dict, session_id: str, session_date: str) -> dict | None:
user_prompt = self.prompt_format(
"user_message",
question=question,
question_type=question_type,
question_date=question_date,
golden_answer=golden_answer,
session_id=session_id,
session_date=session_date,
session_content=json.dumps(session.get("messages", []), ensure_ascii=False, indent=2),
)
try:
result = await reply_with_retry(idx, user_prompt, session_id)
except Exception as exc: # noqa: BLE001 — one bad session must not abort the sweep
if self._is_data_inspection_error(exc):
error = str(exc)
self.logger.warning(
f"[{self.name}] review fallback for {session_id}: non-retryable data inspection error",
)
failed_reviews.append(
{
"session_id": session_id,
"session_date": session_date,
"error": error,
"non_retryable": True,
"fallback": True,
"fallback_reason": "data_inspection_failed",
"raw_session": session,
},
)
return {
"session_id": session_id,
"session_date": session_date,
"extracted_info": "",
"review_status": "fallback",
"fallback_reason": "data_inspection_failed",
"error": error,
"raw_session": session,
}
self.logger.warning(f"[{self.name}] review failed for {session_id}: {exc}")
failed_reviews.append(
{
"session_id": session_id,
"session_date": session_date,
"error": str(exc),
"non_retryable": False,
"fallback": False,
},
)
return None
extracted_info = str(result.get("result") or "").strip()
summary = {
"session_id": session_id,
"session_date": session_date,
"extracted_info": extracted_info,
}
self.logger.info(f"[{self.name}] ({idx}/{total}) extracted {session_id}")
return summary
review_semaphore = asyncio.Semaphore(concurrency)
async def review_one_limited(idx: int, session: dict, session_id: str, session_date: str) -> dict | None:
async with review_semaphore:
return await review_one(idx, session, session_id, session_date)
# gather preserves input order, so summaries stay chronological.
results = await asyncio.gather(
*(
review_one_limited(idx, session, session_id, session_date)
for idx, (session, session_id, session_date) in enumerate(sessions, start=1)
),
)
summaries: list[dict] = [s for s in results if s is not None]
non_empty_summaries = [s for s in summaries if str(s.get("extracted_info") or "").strip()]
fallback_summaries = [s for s in summaries if s.get("review_status") == "fallback"]
reviewed_session_ids = [str(s.get("session_id")) for s in summaries if s.get("session_id")]
output = {
"query": {
"question_id": query_data.get("question_id"),
"question": question,
"question_type": question_type,
"question_date": question_date,
},
"golden": {
"answer": golden_answer,
"answer_session_ids": answer_session_ids,
"answer_session_ids_filter_illegal": answer_session_ids_filter_illegal,
"answer_session_ids_illegal": answer_session_ids_illegal,
},
"review": {
"num_session_files": len(session_files),
"num_reviewed_sessions": len(summaries),
"num_extracted_sessions": len(non_empty_summaries),
"num_empty_extractions": len(summaries) - len(non_empty_summaries),
"num_failed_reviews": len(failed_reviews),
"num_fallback_reviews": len(fallback_summaries),
"num_filtered_sessions": len(session_ids_illegal),
"reviewed_session_ids": reviewed_session_ids,
"session_ids_illegal": session_ids_illegal,
"filtered_sessions": filtered_sessions,
"failed_reviews": failed_reviews,
"fallback_reviews": fallback_summaries,
},
"session_summaries": summaries,
}
output_path = self.workspace_path / OUTPUT_FILENAME
with output_path.open("w", encoding="utf-8") as f:
json.dump(output, f, ensure_ascii=False, indent=2)
self.logger.info(f"[{self.name}] wrote session review to {output_path}")
self.context.response.success = True
self.context.response.answer = f"reviewed {len(summaries)} sessions"
self.context.response.metadata.update(
{
"num_session_files": len(session_files),
"num_reviewed_sessions": len(summaries),
"num_failed_reviews": len(failed_reviews),
"num_fallback_reviews": len(fallback_summaries),
"num_filtered_sessions": len(session_ids_illegal),
"output_path": str(output_path),
},
)
return self.context.response

32
reme/steps/benchmark/lme/session_review.yaml Normal file
View file

@ -0,0 +1,32 @@
system_prompt: |
You are an information extractor for the LongMemEval benchmark. You are given ONE chat session,
a target question, and its golden answer. Your job is to extract the complete information in
this session, especially anything related to the question or golden answer.
Rules:
- Extract all facts, names, entities, numbers, preferences, constraints, corrections, updates,
contradictions, plans, events, and outcomes that appear in the session.
- Be especially careful not to omit any information related to the question or golden answer.
- Keep time expressions inline with the fact they modify, including dates, weekdays, relative
times such as "last week" or "since January 15th", durations, and frequencies.
- Do not invent facts. Only extract what is actually present in the session.
- Do not judge whether answer_session_ids are correct. That is handled by the downstream
golden check.
- Output only the extracted information as plain text. Do not output JSON, markdown fences, or
relevance labels.
user_message: |
Question: {question}
Question type: {question_type}
Question date: {question_date}
Golden answer: {golden_answer}
--- Session under review ---
Session id: {session_id}
Session date: {session_date}
Session messages (JSON):
{session_content}
--- End of session ---
Extract the complete information from this session. Do not summarize away details, and do not
omit information related to the question or golden answer.

2
reme/steps/file_io/_daily_index.py
View file

@ -8,7 +8,7 @@ import frontmatter
from ._path import resolve_path, validate_filename_component
from ...utils import get_logger
logger = get_logger()
logger = get_logger(log_to_file=False)
def validate_session_id(session_id: str) -> str | None:

2
reme/steps/file_io/_file_io.py
View file

@ -10,7 +10,7 @@ import aiofiles.os
from ...constants import DEFAULT_MAX_BYTES, MAX_FILE_READ_BYTES, TRUNCATION_NOTICE_MARKER
from ...utils import get_logger
logger = get_logger()
logger = get_logger(log_to_file=False)
# ---------------------------------------------------------------------------
# In-process per-path write lock.

2
reme/steps/file_io/_path.py
View file

@ -5,7 +5,7 @@ from pathlib import Path
from ...utils import get_logger
logger = get_logger()
logger = get_logger(log_to_file=False)
NON_MD_WARNING = (
"non-markdown file detected; CRUD operations are recommended on markdown files. "

4
reme/steps/index/__init__.py
View file

@ -1,6 +1,7 @@
"""Index steps."""
from .bm25_search import Bm25SearchStep
from .clear_paths import ClearPathsStep
from .clear_store import ClearStoreStep
from .draft import AddDraftStep, ReadAllDraftStep
from .log_changes import LogChangesStep
@ -10,6 +11,7 @@ from .search import SearchStep
from .traverse import TraverseStep
from .update_changes import ChangeApplyStep, UpdateCatalogStep, UpdateIndexStep
from .vector_search import VectorSearchStep
from .wait_for_paths import WaitForPathsStep
from .watch_changes import (
DEFAULT_LOW_POWER_POLL_MS,
DEFAULT_WATCH_DEBOUNCE_MS,
@ -21,6 +23,7 @@ __all__ = [
"AddDraftStep",
"Bm25SearchStep",
"ChangeApplyStep",
"ClearPathsStep",
"ClearStoreStep",
"DEFAULT_LOW_POWER_POLL_MS",
"DEFAULT_WATCH_DEBOUNCE_MS",
@ -34,5 +37,6 @@ __all__ = [
"UpdateCatalogStep",
"UpdateIndexStep",
"VectorSearchStep",
"WaitForPathsStep",
"WatchChangesStep",
]

29
reme/steps/index/_source_format.py Normal file
View file

@ -0,0 +1,29 @@
"""Shared helper: render search results with each hit's originating session_id.
Plain ``vector_search``/``bm25_search`` return only ``chunk.text``. The
LongMemEval agentic-answer flow needs each hit's ``session_id`` so the agent can
pivot back to the raw session via ``extract_session_by_id``. This helper reads
that ``session_id`` from the note's frontmatter and prefixes it to the text.
"""
from pathlib import Path
import frontmatter
from ...schema import FileChunk
def render_with_source(chunks: list[FileChunk], workspace_path: Path) -> str:
"""Render each chunk as ``[session_id=<sid>]`` header + text."""
lines: list[str] = []
for c in chunks:
sid = ""
if c.path:
try:
post = frontmatter.loads((workspace_path / c.path).read_text(encoding="utf-8"))
sid = str((post.metadata or {}).get("session_id", "") or "").strip()
except Exception:
sid = ""
header = f"[session_id={sid}]" if sid else "[session_id: unknown]"
lines.append(f"{header}\n{c.text}")
return "\n\n".join(lines)

12
reme/steps/index/bm25_search.py
View file

@ -4,10 +4,12 @@ import datetime
from typing import Final
from ..base_step import BaseStep
from ._source_format import render_with_source
from ...components import R
from ...schema import FileChunk
_MAX_CANDIDATES: Final = 200
_CANDIDATE_MULTIPLIER: Final = 10
@R.register("bm25_search_step")
@ -17,9 +19,10 @@ class Bm25SearchStep(BaseStep):
TOOL_CONTEXTS_KEY: Final[str] = "tool_contexts"
SEARCH_SEEN_KEY: Final[str] = "search_seen_chunk_ids"
def __init__(self, *args, seen_ttl_hours: float = 24, **kwargs):
def __init__(self, *args, seen_ttl_hours: float = 24, include_source: bool = True, **kwargs):
super().__init__(*args, **kwargs)
self.seen_ttl_hours = seen_ttl_hours
self.include_source = include_source
def _tool_context_store(self, tool_context_id: str) -> dict:
"""Return the mutable state bucket for a tool context."""
@ -56,7 +59,7 @@ class Bm25SearchStep(BaseStep):
return self.context.response
assert limit > 0, f"limit must be positive, got {limit}"
candidates = min(_MAX_CANDIDATES, max(1, limit * 5))
candidates = min(_MAX_CANDIDATES, max(1, limit * _CANDIDATE_MULTIPLIER))
results = await self.file_store.keyword_search(query, candidates, {})
self.logger.info(f"[{self.name}] query={query!r} candidates={candidates} hits={len(results)}")
@ -68,7 +71,10 @@ class Bm25SearchStep(BaseStep):
else:
results = results[:limit]
self.context.response.answer = "\n\n".join(c.text for c in results)
if self.include_source:
self.context.response.answer = render_with_source(results, self.workspace_path)
else:
self.context.response.answer = "\n\n".join(c.text for c in results)
self.context.response.metadata["results"] = [
c.model_dump(exclude_none=True, exclude={"embedding"}) for c in results
]

51
reme/steps/index/clear_paths.py Normal file
View file

@ -0,0 +1,51 @@
"""Delete stale workspace outputs before a job rebuilds them.
A tiny, generic housekeeping step: given workspace-relative ``paths`` (and/or
``config_keys`` resolved against the app config, e.g. ``daily_dir``), it removes
each target — file or directory — so the following steps start from a clean
slate. Missing targets are ignored, and anything resolving outside the workspace
is refused, so a misconfigured path can never wipe unrelated files.
"""
import shutil
from pathlib import Path
from ..base_step import BaseStep
from ...components import R
@R.register("clear_paths_step")
class ClearPathsStep(BaseStep):
"""Remove configured files/dirs under the workspace before a rebuild."""
def _targets(self) -> list[str]:
"""Collect workspace-relative targets from ``paths`` + ``config_keys``."""
rels: list[str] = list(self.kwargs.get("paths") or [])
for key in self.kwargs.get("config_keys") or []:
value = self.config_value(key)
if value:
rels.append(str(value))
return rels
async def execute(self):
assert self.context is not None
root = self.workspace_path.resolve()
removed: list[str] = []
for rel in self._targets():
target = (self.workspace_path / rel).resolve()
# Refuse anything outside the workspace, or the workspace root itself.
if target == root or root not in target.parents:
self.logger.warning(f"[{self.name}] refusing to clear out-of-workspace path: {rel!r}")
continue
if target.is_dir() and not target.is_symlink():
shutil.rmtree(target, ignore_errors=True)
removed.append(rel)
elif target.exists() or target.is_symlink():
Path(target).unlink(missing_ok=True)
removed.append(rel)
self.context.response.metadata["cleared_paths"] = removed
if removed:
self.logger.info(f"[{self.name}] cleared {removed}")
return self.context.response

12
reme/steps/index/vector_search.py
View file

@ -4,10 +4,12 @@ import datetime
from typing import Final
from ..base_step import BaseStep
from ._source_format import render_with_source
from ...components import R
from ...schema import FileChunk
_MAX_CANDIDATES: Final = 200
_CANDIDATE_MULTIPLIER: Final = 10
@R.register("vector_search_step")
@ -17,9 +19,10 @@ class VectorSearchStep(BaseStep):
TOOL_CONTEXTS_KEY: Final[str] = "tool_contexts"
SEARCH_SEEN_KEY: Final[str] = "search_seen_chunk_ids"
def __init__(self, *args, seen_ttl_hours: float = 24, **kwargs):
def __init__(self, *args, seen_ttl_hours: float = 24, include_source: bool = True, **kwargs):
super().__init__(*args, **kwargs)
self.seen_ttl_hours = seen_ttl_hours
self.include_source = include_source
def _tool_context_store(self, tool_context_id: str) -> dict:
"""Return the mutable state bucket for a tool context."""
@ -56,7 +59,7 @@ class VectorSearchStep(BaseStep):
return self.context.response
assert limit > 0, f"limit must be positive, got {limit}"
candidates = min(_MAX_CANDIDATES, max(1, limit * 5))
candidates = min(_MAX_CANDIDATES, max(1, limit * _CANDIDATE_MULTIPLIER))
results = await self.file_store.vector_search(query, candidates, {})
self.logger.info(f"[{self.name}] query={query!r} candidates={candidates} hits={len(results)}")
@ -68,7 +71,10 @@ class VectorSearchStep(BaseStep):
else:
results = results[:limit]
self.context.response.answer = "\n\n".join(c.text for c in results)
if self.include_source:
self.context.response.answer = render_with_source(results, self.workspace_path)
else:
self.context.response.answer = "\n\n".join(c.text for c in results)
self.context.response.metadata["results"] = [
c.model_dump(exclude_none=True, exclude={"embedding"}) for c in results
]

56
reme/steps/index/wait_for_paths.py Normal file
View file

@ -0,0 +1,56 @@
"""Wait for required workspace files before continuing a job."""
import asyncio
import time
from pathlib import Path
from ..base_step import BaseStep
from ...components import R
@R.register("wait_for_paths_step")
class WaitForPathsStep(BaseStep):
"""Block until configured workspace-relative paths exist."""
def _targets(self) -> list[str]:
"""Collect workspace-relative targets from ``paths`` + ``config_keys``."""
rels: list[str] = list(self.kwargs.get("paths") or [])
for key in self.kwargs.get("config_keys") or []:
value = self.config_value(key)
if value:
rels.append(str(value))
return rels
async def execute(self):
assert self.context is not None
poll_seconds = float(self.kwargs.get("poll_seconds", 5.0))
log_every_seconds = float(self.kwargs.get("log_every_seconds", 60.0))
root = self.workspace_path.resolve()
targets: list[tuple[str, Path]] = []
for rel in self._targets():
target = (self.workspace_path / rel).resolve()
if target == root or root not in target.parents:
raise ValueError(f"wait_for_paths_step refuses out-of-workspace path: {rel!r}")
targets.append((rel, target))
if not targets:
return self.context.response
start = time.monotonic()
last_log_at = 0.0
while True:
missing = [rel for rel, target in targets if not target.exists()]
if not missing:
waited_seconds = time.monotonic() - start
self.context.response.metadata["waited_for_paths"] = [rel for rel, _ in targets]
self.context.response.metadata["waited_seconds"] = waited_seconds
self.logger.info(f"[{self.name}] required paths are ready after {waited_seconds:.1f}s")
return self.context.response
now = time.monotonic()
if now - last_log_at >= log_every_seconds:
waited_seconds = now - start
self.logger.info(f"[{self.name}] waiting for {missing} ({waited_seconds:.1f}s elapsed)")
last_log_at = now
await asyncio.sleep(poll_seconds)

60
tests/unit/test_logging_config.py Normal file
View file

@ -0,0 +1,60 @@
"""Tests for logging configuration handoff during app startup."""
from reme.application import Application
from reme.config.config_parser import resolve_app_config
class DummyLogger:
"""Minimal logger used to capture initialization without touching sinks."""
def bind(self, **_kwargs):
"""No-op."""
return self
def info(self, *_args, **_kwargs):
"""No-op."""
return None
def test_resolve_app_config_does_not_create_file_logger(monkeypatch):
"""Config-loading messages should not create empty run log files."""
calls = []
def fake_get_logger(**kwargs):
calls.append(kwargs)
return DummyLogger()
monkeypatch.setattr("reme.utils.get_logger", fake_get_logger)
resolve_app_config(config="demo")
assert calls[0]["log_to_file"] is False
def test_application_reinitializes_logger_from_final_config(monkeypatch, tmp_path):
"""Application startup should install sinks from the resolved ApplicationConfig."""
calls = []
def fake_get_logger(**kwargs):
calls.append(kwargs)
return DummyLogger()
monkeypatch.setattr("reme.application.get_logger", fake_get_logger)
monkeypatch.setattr("reme.components.base_component.get_logger", lambda **_kwargs: DummyLogger())
monkeypatch.setattr(Application, "_init_service", lambda self: setattr(self.context, "service", None))
monkeypatch.setattr(Application, "_init_components", lambda self: None)
monkeypatch.setattr(Application, "_init_jobs", lambda self: None)
Application(
enable_logo=False,
log_to_console=False,
log_to_file=True,
workspace_dir=str(tmp_path / "workspace"),
service={"backend": "unused"},
)
assert calls[0] == {
"log_to_console": False,
"log_to_file": True,
"force_init": True,
}

24
tests/unit/test_search_step.py
View file

@ -110,6 +110,26 @@ def test_search_step_rrf_merges_vector_and_keyword_by_chunk_id():
asyncio.run(run())
def test_lme_plain_search_steps_use_ten_times_limit_candidates():
"""LME vector/bm25 tools fetch a wider candidate pool before truncating."""
async def run():
store = FakeSearchStore()
vector = VectorSearchStep(file_store=store, include_source=False)
bm25 = Bm25SearchStep(file_store=store, include_source=False)
await vector(RuntimeContext(query="alpha", limit=3))
await bm25(RuntimeContext(query="alpha", limit=3))
assert store.calls == [
("vector", "alpha", 30, {}),
("keyword", "alpha", 30, {}),
]
asyncio.run(run())
def test_draft_steps_accumulate_by_tool_context_id():
"""Drafts are stored in app metadata and isolated by injected tool_context_id."""
@ -168,10 +188,10 @@ def test_plain_search_steps_apply_min_score_before_truncation():
],
)
vector = await VectorSearchStep(file_store=vector_store)(
vector = await VectorSearchStep(file_store=vector_store, include_source=False)(
RuntimeContext(query="alpha", limit=5, min_score=0.5),
)
keyword = await Bm25SearchStep(file_store=keyword_store)(
keyword = await Bm25SearchStep(file_store=keyword_store, include_source=False)(
RuntimeContext(query="alpha", limit=5, min_score=1.0),
)