From bf7ca1770518f23a3e4da1a09c752ffc5a6bec10 Mon Sep 17 00:00:00 2001 From: jinliyl <6469360+jinliyl@users.noreply.github.com> Date: Mon, 13 Jul 2026 21:26:27 +0800 Subject: [PATCH] feat(benchmark): add LongMemEval golden answer validation (#335) * 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 --- .github/workflows/windows-smoke.yml | 38 ++ benchmark/longmemeval/clean_sample_outputs.py | 141 ++++++ benchmark/longmemeval/run_agentic_answer.py | 343 ++++++++++++++ benchmark/longmemeval/run_golden_check.py | 216 +++++++++ benchmark/longmemeval/run_session_review.py | 203 +++++++++ benchmark/longmemeval/stats_agentic_answer.py | 202 +++++++++ benchmark/longmemeval/stats_golden_check.py | 344 ++++++++++++++ benchmark/longmemeval/stats_session_review.py | 248 ++++++++++ pyproject.toml | 4 + reme/application.py | 6 +- .../agent_wrapper/as_agent_wrapper.py | 2 + reme/components/component_registry.py | 2 +- reme/config/config_parser.py | 2 +- reme/config/jinli_lme.yaml | 205 ++++++++- reme/steps/benchmark/__init__.py | 6 +- reme/steps/benchmark/lme/__init__.py | 14 +- reme/steps/benchmark/lme/agentic_answer.py | 87 ++++ reme/steps/benchmark/lme/agentic_answer.yaml | 40 ++ reme/steps/benchmark/lme/auto_memory.py | 428 ++++++++++++++++++ reme/steps/benchmark/lme/auto_memory.yaml | 45 ++ reme/steps/benchmark/lme/extract_session.py | 103 +++++ reme/steps/benchmark/lme/extract_session.yaml | 27 ++ reme/steps/benchmark/lme/golden_check.py | 201 ++++++++ reme/steps/benchmark/lme/golden_check.yaml | 31 ++ reme/steps/benchmark/lme/llm_judge.py | 81 ---- reme/steps/benchmark/lme/lme_llm_judge.py | 109 +++++ .../{llm_judge.yaml => lme_llm_judge.yaml} | 0 reme/steps/benchmark/lme/session_review.py | 358 +++++++++++++++ reme/steps/benchmark/lme/session_review.yaml | 32 ++ reme/steps/file_io/_daily_index.py | 2 +- reme/steps/file_io/_file_io.py | 2 +- reme/steps/file_io/_path.py | 2 +- reme/steps/index/__init__.py | 4 + reme/steps/index/_source_format.py | 29 ++ reme/steps/index/bm25_search.py | 12 +- reme/steps/index/clear_paths.py | 51 +++ reme/steps/index/vector_search.py | 12 +- reme/steps/index/wait_for_paths.py | 56 +++ tests/unit/test_logging_config.py | 60 +++ tests/unit/test_search_step.py | 24 +- 40 files changed, 3670 insertions(+), 102 deletions(-) create mode 100644 .github/workflows/windows-smoke.yml create mode 100644 benchmark/longmemeval/clean_sample_outputs.py create mode 100644 benchmark/longmemeval/run_agentic_answer.py create mode 100644 benchmark/longmemeval/run_golden_check.py create mode 100644 benchmark/longmemeval/run_session_review.py create mode 100644 benchmark/longmemeval/stats_agentic_answer.py create mode 100644 benchmark/longmemeval/stats_golden_check.py create mode 100644 benchmark/longmemeval/stats_session_review.py create mode 100644 reme/steps/benchmark/lme/agentic_answer.py create mode 100644 reme/steps/benchmark/lme/agentic_answer.yaml create mode 100644 reme/steps/benchmark/lme/auto_memory.py create mode 100644 reme/steps/benchmark/lme/auto_memory.yaml create mode 100644 reme/steps/benchmark/lme/extract_session.py create mode 100644 reme/steps/benchmark/lme/extract_session.yaml create mode 100644 reme/steps/benchmark/lme/golden_check.py create mode 100644 reme/steps/benchmark/lme/golden_check.yaml delete mode 100644 reme/steps/benchmark/lme/llm_judge.py create mode 100644 reme/steps/benchmark/lme/lme_llm_judge.py rename reme/steps/benchmark/lme/{llm_judge.yaml => lme_llm_judge.yaml} (100%) create mode 100644 reme/steps/benchmark/lme/session_review.py create mode 100644 reme/steps/benchmark/lme/session_review.yaml create mode 100644 reme/steps/index/_source_format.py create mode 100644 reme/steps/index/clear_paths.py create mode 100644 reme/steps/index/wait_for_paths.py create mode 100644 tests/unit/test_logging_config.py diff --git a/.github/workflows/windows-smoke.yml b/.github/workflows/windows-smoke.yml new file mode 100644 index 00000000..83ddcfe1 --- /dev/null +++ b/.github/workflows/windows-smoke.yml @@ -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 diff --git a/benchmark/longmemeval/clean_sample_outputs.py b/benchmark/longmemeval/clean_sample_outputs.py new file mode 100644 index 00000000..4fcd50ae --- /dev/null +++ b/benchmark/longmemeval/clean_sample_outputs.py @@ -0,0 +1,141 @@ +#!/usr/bin/env python3 +"""Remove generated LongMemEval files while keeping source inputs. + +For each ``datasets/longmemeval/`` 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()) diff --git a/benchmark/longmemeval/run_agentic_answer.py b/benchmark/longmemeval/run_agentic_answer.py new file mode 100644 index 00000000..f5bc1631 --- /dev/null +++ b/benchmark/longmemeval/run_agentic_answer.py @@ -0,0 +1,343 @@ +#!/usr/bin/env python3 +"""Drive the LongMemEval memory pipeline across all samples. + +For every workspace under ``datasets/longmemeval/`` this launches one or more +``reme start config=jinli_lme 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//.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/.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())) diff --git a/benchmark/longmemeval/run_golden_check.py b/benchmark/longmemeval/run_golden_check.py new file mode 100644 index 00000000..18e4d847 --- /dev/null +++ b/benchmark/longmemeval/run_golden_check.py @@ -0,0 +1,216 @@ +#!/usr/bin/env python3 +"""Run LongMemEval ``golden_check`` concurrently across samples. + +For every workspace under ``datasets/longmemeval/`` 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/.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())) diff --git a/benchmark/longmemeval/run_session_review.py b/benchmark/longmemeval/run_session_review.py new file mode 100644 index 00000000..c40b03b8 --- /dev/null +++ b/benchmark/longmemeval/run_session_review.py @@ -0,0 +1,203 @@ +#!/usr/bin/env python3 +"""Run LongMemEval ``session_review`` concurrently across samples. + +For every workspace under ``datasets/longmemeval/`` 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/.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())) diff --git a/benchmark/longmemeval/stats_agentic_answer.py b/benchmark/longmemeval/stats_agentic_answer.py new file mode 100644 index 00000000..37ddea6a --- /dev/null +++ b/benchmark/longmemeval/stats_agentic_answer.py @@ -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()) diff --git a/benchmark/longmemeval/stats_golden_check.py b/benchmark/longmemeval/stats_golden_check.py new file mode 100644 index 00000000..97f7f7f6 --- /dev/null +++ b/benchmark/longmemeval/stats_golden_check.py @@ -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()) diff --git a/benchmark/longmemeval/stats_session_review.py b/benchmark/longmemeval/stats_session_review.py new file mode 100644 index 00000000..345a9dee --- /dev/null +++ b/benchmark/longmemeval/stats_session_review.py @@ -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()) diff --git a/pyproject.toml b/pyproject.toml index 30ede1bc..caf23aa0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -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] diff --git a/reme/application.py b/reme/application.py index ef17349d..aeccc5b4 100644 --- a/reme/application.py +++ b/reme/application.py @@ -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__() diff --git a/reme/components/agent_wrapper/as_agent_wrapper.py b/reme/components/agent_wrapper/as_agent_wrapper.py index 41c85ca1..bbd6a529 100644 --- a/reme/components/agent_wrapper/as_agent_wrapper.py +++ b/reme/components/agent_wrapper/as_agent_wrapper.py @@ -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 diff --git a/reme/components/component_registry.py b/reme/components/component_registry.py index 8f141d43..2a7d044f 100644 --- a/reme/components/component_registry.py +++ b/reme/components/component_registry.py @@ -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.""" diff --git a/reme/config/config_parser.py b/reme/config/config_parser.py index 76b6c8b5..c97a84ec 100644 --- a/reme/config/config_parser.py +++ b/reme/config/config_parser.py @@ -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 diff --git a/reme/config/jinli_lme.yaml b/reme/config/jinli_lme.yaml index 6b6f76d6..5814e8b1 100644 --- a/reme/config/jinli_lme.yaml +++ b/reme/config/jinli_lme.yaml @@ -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 diff --git a/reme/steps/benchmark/__init__.py b/reme/steps/benchmark/__init__.py index 7130714a..cab5b241 100644 --- a/reme/steps/benchmark/__init__.py +++ b/reme/steps/benchmark/__init__.py @@ -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", ] diff --git a/reme/steps/benchmark/lme/__init__.py b/reme/steps/benchmark/lme/__init__.py index abf772b8..c9c5b4a3 100644 --- a/reme/steps/benchmark/lme/__init__.py +++ b/reme/steps/benchmark/lme/__init__.py @@ -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", ] diff --git a/reme/steps/benchmark/lme/agentic_answer.py b/reme/steps/benchmark/lme/agentic_answer.py new file mode 100644 index 00000000..fd90952d --- /dev/null +++ b/reme/steps/benchmark/lme/agentic_answer.py @@ -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/.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 diff --git a/reme/steps/benchmark/lme/agentic_answer.yaml b/reme/steps/benchmark/lme/agentic_answer.yaml new file mode 100644 index 00000000..36ccbf82 --- /dev/null +++ b/reme/steps/benchmark/lme/agentic_answer.yaml @@ -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. diff --git a/reme/steps/benchmark/lme/auto_memory.py b/reme/steps/benchmark/lme/auto_memory.py new file mode 100644 index 00000000..fcd33df6 --- /dev/null +++ b/reme/steps/benchmark/lme/auto_memory.py @@ -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 ``_(...)_