diff --git a/core/cli.py b/core/cli.py index 56c85cbc..c671c5cb 100644 --- a/core/cli.py +++ b/core/cli.py @@ -1148,7 +1148,14 @@ def cmd_doctor(args: argparse.Namespace) -> int: def cmd_eval(args: argparse.Namespace) -> int: """Run an eval lane by name, or list available lanes.""" - from evals.framework import discover_lanes, get_lane, run_lane, write_result + from evals._parallel import normalize_workers + from evals.framework import ( + discover_lanes, + get_lane, + load_cases, + run_lane, + write_result, + ) if args.list_lanes: lanes = discover_lanes() @@ -1171,8 +1178,27 @@ def cmd_eval(args: argparse.Namespace) -> int: version = args.version or (lane.versions[0] if lane.versions else "v1") split = args.split + if not args.json and lane_name == "cognition": + if split == "dev": + cases_path = lane.dev_cases_path() + elif split == "public": + cases_path = lane.public_cases_path(version) + else: + cases_path = lane.holdout_cases_path(version) + cases = load_cases(cases_path) + effective_workers = normalize_workers( + args.workers if args.workers is not None else 4, + len(cases), + ) + print(f"workers : {effective_workers}") + try: - result = run_lane(lane, version=version, split=split) + result = run_lane( + lane, + version=version, + split=split, + workers=args.workers, + ) except FileNotFoundError as exc: _die(str(exc)) @@ -1761,7 +1787,7 @@ For the central evidence index: _ALL_PREAMBLE = """ ================================================================================ - core demo all — Run Every Demo, End-to-End + core demo all — Combined Demo, End-to-End ================================================================================ Runs the full demo suite in sequence and prints a consolidated PASS/FAIL @@ -1976,7 +2002,7 @@ def cmd_demo(args: argparse.Namespace) -> int: if target == "anchor-lens-tour": from evals.anchor_lens_tour.run_tour import run_tour as run_lens_tour - result = run_lens_tour(emit_json=args.json) + result = run_lens_tour(emit_json=args.json, workers=args.workers) if args.json: print(json.dumps(result, indent=2, sort_keys=True, default=str)) return 0 if result.get("all_claims_supported", False) else 1 @@ -1984,7 +2010,7 @@ def cmd_demo(args: argparse.Namespace) -> int: if target == "orthogonality-tour": from evals.orthogonality_tour.run_tour import run_tour as run_ortho_tour - result = run_ortho_tour(emit_json=args.json) + result = run_ortho_tour(emit_json=args.json, workers=args.workers) if args.json: print(json.dumps(result, indent=2, sort_keys=True, default=str)) return 0 if result.get("all_claims_supported", False) else 1 @@ -2258,13 +2284,14 @@ def _run_demo_all(emit_json: bool) -> int: print(json.dumps(consolidated, indent=2, sort_keys=True, default=str)) else: print("\n" + "═" * 76) - print(" core demo all — consolidated summary") + print(" core demo all — Combined demo summary") print("═" * 76) for name, ok in passed.items(): mark = "✓ PASS" if ok else "✗ FAIL" print(f" {mark} {name}") print() print(f" all_demos_passed : {all_passed}") + print(" load-bearing claim of the ADR-0024 chain") print() _write_results_index() @@ -2929,6 +2956,16 @@ def build_parser() -> argparse.ArgumentParser: ), ) demo.add_argument("--json", action="store_true", help="emit machine-readable JSON") + demo.add_argument( + "--workers", + type=int, + default=4, + metavar="N", + help=( + "parallel worker count for supported demos " + "(0/1 => sequential; default 4)" + ), + ) demo.add_argument( "--no-stream", dest="no_stream", @@ -2945,6 +2982,16 @@ def build_parser() -> argparse.ArgumentParser: eval_cmd.add_argument("--list", dest="list_lanes", action="store_true", help="list available eval lanes") eval_cmd.add_argument("--version", help="version to evaluate (default: latest)") eval_cmd.add_argument("--split", default="public", choices=["dev", "public", "holdout"], help="which split to score (default: public)") + eval_cmd.add_argument( + "--workers", + type=int, + default=4, + metavar="N", + help=( + "parallel worker count for cognition lane " + "(0/1 => sequential; default 4)" + ), + ) eval_cmd.add_argument("--json", action="store_true", help="emit machine-readable JSON") eval_cmd.add_argument("--save", action="store_true", help="write result to lane results/ directory") eval_cmd.add_argument("--report", metavar="PATH", help="write JSON report to file") diff --git a/evals/_parallel.py b/evals/_parallel.py new file mode 100644 index 00000000..97e19ad6 --- /dev/null +++ b/evals/_parallel.py @@ -0,0 +1,83 @@ +"""Process-parallel eval runner with per-worker warm-up. + +The eval lanes in this repository are deliberately embarrassingly +parallel: each case gets a fresh runtime in its own process, so there +is no shared mutable state and no race risk. The expensive part is +worker-local pack loading, so this helper uses a ``Pool`` initializer +to warm the relevant caches once per worker before any cases run. + +The builder passed to :func:`run_cases_parallel` is invoked once per +worker and must return a callable that scores a single case with a +fresh runtime. Typical builders do two things: + +1. Construct one or more warm-up runtimes to populate process-local + caches. +2. Return a per-case function that instantiates a new runtime for each + case and computes the case result deterministically. + +The helper preserves input order in its returned list. +""" + +from __future__ import annotations + +import multiprocessing as mp +import os +from collections.abc import Callable, Sequence +from typing import Any, TypeVar + +_R = TypeVar("_R") +_CaseRunner = Callable[[Any], _R] +_CaseRunnerBuilder = Callable[[], Callable[[Any], _R]] + +_MP_CONTEXT = mp.get_context("spawn") +_WORKER_CASE_RUNNER: _CaseRunner[Any] | None = None + + +def _default_workers() -> int: + detected = os.cpu_count() or 4 + return max(1, min(detected, 8)) + + +def normalize_workers(n_workers: int, case_count: int) -> int: + """Clamp worker count to the active CPU budget and case count.""" + cpu_cap = os.cpu_count() or 1 + return max(1, min(int(n_workers), cpu_cap, max(1, int(case_count)))) + + +def _worker_init(build_runtime_fn: _CaseRunnerBuilder[_R]) -> None: + """Build the worker-local case runner after caches are warm.""" + global _WORKER_CASE_RUNNER + _WORKER_CASE_RUNNER = build_runtime_fn() + + +def _run_case_in_worker(case: Any) -> _R: + if _WORKER_CASE_RUNNER is None: # pragma: no cover - defensive guard + raise RuntimeError("worker case runner was not initialized") + return _WORKER_CASE_RUNNER(case) + + +def run_cases_parallel( + cases: Sequence[Any], + build_runtime_fn: _CaseRunnerBuilder[_R], + n_workers: int = 4, +) -> list[_R]: + """Run ``cases`` in parallel using a worker-initialized process pool. + + ``build_runtime_fn`` is called once per worker. It should warm any + worker-local caches and return a callable that scores a single case + using a fresh runtime. + """ + if not cases: + return [] + + effective_workers = normalize_workers(n_workers, len(cases)) + if effective_workers <= 1: + case_runner = build_runtime_fn() + return [case_runner(case) for case in cases] + + with _MP_CONTEXT.Pool( + processes=effective_workers, + initializer=_worker_init, + initargs=(build_runtime_fn,), + ) as pool: + return list(pool.imap(_run_case_in_worker, cases)) diff --git a/evals/anchor_lens_tour/run_tour.py b/evals/anchor_lens_tour/run_tour.py index b410a25c..b554be51 100644 --- a/evals/anchor_lens_tour/run_tour.py +++ b/evals/anchor_lens_tour/run_tour.py @@ -38,11 +38,13 @@ orthogonality seam claimed by ADR-0073. from __future__ import annotations import json -from typing import Any +from functools import partial +from typing import Any, Callable from chat.runtime import ChatRuntime from core.cognition.pipeline import CognitiveTurnPipeline from core.config import RuntimeConfig +from evals._parallel import normalize_workers, run_cases_parallel _LENSES = ( @@ -129,6 +131,32 @@ def _run_one_lens(lens_id: str) -> list[dict[str, Any]]: return cells +def _build_case_runner() -> Callable[[dict[str, Any]], dict[str, Any]]: + """Warm every lens pack once, then return a per-case scorer.""" + for lens_id in _LENSES: + ChatRuntime(config=RuntimeConfig(anchor_lens_id=lens_id)) + + def _run(case: dict[str, Any]) -> dict[str, Any]: + lens_id = case["lens_id"] + prompt = case["prompt"] + runtime = ChatRuntime(config=RuntimeConfig(anchor_lens_id=lens_id)) + pipeline = CognitiveTurnPipeline(runtime=runtime) + result = pipeline.run(prompt) + turn_event = runtime.turn_log[-1] + return { + "prompt": prompt, + "surface": turn_event.surface, + "grounding_source": getattr(turn_event, "grounding_source", ""), + "trace_hash": result.trace_hash, + "anchor_lens_id": getattr(turn_event, "anchor_lens_id", ""), + "anchor_lens_mode_label": getattr( + turn_event, "anchor_lens_mode_label", "" + ), + } + + return _run + + def _print_grid(grid: dict[str, list[dict[str, Any]]]) -> None: for prompt_idx, prompt in enumerate(_PROMPTS): _say(f" P{prompt_idx + 1}: {prompt!r}") @@ -214,7 +242,7 @@ def _check_claims( } -def run_tour(*, emit_json: bool = False) -> dict[str, Any]: +def run_tour(*, emit_json: bool = False, workers: int | None = None) -> dict[str, Any]: """Run the anchor-lens tour end-to-end and return a structured report.""" global _VERBOSE _VERBOSE = not emit_json @@ -222,11 +250,26 @@ def run_tour(*, emit_json: bool = False) -> dict[str, Any]: if not emit_json: _print_header() - grid: dict[str, list[dict[str, Any]]] = {} - for lens_id in _LENSES: - if not emit_json: + cases = [ + {"lens_id": lens_id, "prompt": prompt} + for lens_id in _LENSES + for prompt in _PROMPTS + ] + effective_workers = normalize_workers(workers if workers is not None else 4, len(cases)) + if not emit_json: + for lens_id in _LENSES: _say(f" Running lens: {lens_id}") - grid[lens_id] = _run_one_lens(lens_id) + _say(f" workers: {effective_workers}") + + cells = run_cases_parallel( + cases, + partial(_build_case_runner), + n_workers=effective_workers, + ) + + grid: dict[str, list[dict[str, Any]]] = {lens_id: [] for lens_id in _LENSES} + for cell in cells: + grid[cell["anchor_lens_id"]].append(cell) if not emit_json: _say() _say("-" * 72) diff --git a/evals/cognition/runner.py b/evals/cognition/runner.py index 19d9711a..1b1ffa7f 100644 --- a/evals/cognition/runner.py +++ b/evals/cognition/runner.py @@ -6,11 +6,14 @@ where report has ``.metrics`` (dict) and ``.case_details`` (list[dict]). from __future__ import annotations from dataclasses import dataclass, field +from functools import partial +from typing import Callable from typing import Any from chat.runtime import ChatRuntime from core.config import RuntimeConfig from core.cognition.pipeline import CognitiveTurnPipeline +from evals._parallel import run_cases_parallel from generate.intent import IntentTag @@ -71,10 +74,28 @@ def _run_case(case: dict[str, Any], pipeline: CognitiveTurnPipeline) -> CaseResu ) +def _build_case_runner( + config: RuntimeConfig | None = None, +) -> Callable[[dict[str, Any]], CaseResult]: + """Warm worker-local caches once, then return a per-case scorer.""" + if config is None: + ChatRuntime() + else: + ChatRuntime(config=config) + + def _run(case: dict[str, Any]) -> CaseResult: + runtime = ChatRuntime(config=config) if config else ChatRuntime() + pipeline = CognitiveTurnPipeline(runtime) + return _run_case(case, pipeline) + + return _run + + def run_lane( cases: list[dict[str, Any]], *, config: RuntimeConfig | None = None, + workers: int | None = None, ) -> LaneReport: """Run all cases through CognitiveTurnPipeline and return metrics + details.""" total = 0 @@ -85,10 +106,14 @@ def run_lane( versor_closures = 0 case_details: list[dict[str, Any]] = [] - for case in cases: - runtime = ChatRuntime(config=config) if config else ChatRuntime() - pipeline = CognitiveTurnPipeline(runtime) - cr = _run_case(case, pipeline) + case_runner_builder = partial(_build_case_runner, config=config) + case_results = run_cases_parallel( + cases, + case_runner_builder, + n_workers=workers if workers is not None else 4, + ) + + for cr in case_results: total += 1 if cr.intent_correct: diff --git a/evals/framework.py b/evals/framework.py index dbb58cf8..4a9b1a1b 100644 --- a/evals/framework.py +++ b/evals/framework.py @@ -153,6 +153,7 @@ def run_lane( version: str = "v1", split: str = "public", config: Any = None, + workers: int | None = None, ) -> LaneResult: """Run a single lane on a given version and split.""" if split == "dev": @@ -172,7 +173,7 @@ def run_lane( cases = load_cases(cases_path) runner_module = load_lane_runner(lane) - report = runner_module.run_lane(cases, config=config) + report = runner_module.run_lane(cases, config=config, workers=workers) return LaneResult( lane=lane.name, diff --git a/evals/orthogonality_tour/run_tour.py b/evals/orthogonality_tour/run_tour.py index f1ce5ff3..a6b817f2 100644 --- a/evals/orthogonality_tour/run_tour.py +++ b/evals/orthogonality_tour/run_tour.py @@ -37,11 +37,13 @@ Exit code 0 iff every claim holds. from __future__ import annotations import json -from typing import Any +from functools import partial +from typing import Any, Callable from chat.runtime import ChatRuntime from core.cognition.pipeline import CognitiveTurnPipeline from core.config import RuntimeConfig +from evals._parallel import normalize_workers, run_cases_parallel _REGISTERS = ( @@ -130,13 +132,39 @@ def _run_one_cell(register_id: str, lens_id: str, prompt: str) -> dict[str, Any] } -def _build_grid() -> list[dict[str, Any]]: - cells: list[dict[str, Any]] = [] +def _build_case_runner() -> Callable[[dict[str, Any]], dict[str, Any]]: + """Warm all register/lens pack combinations once, then score cases.""" for register_id in _REGISTERS: for lens_id in _LENSES: - for prompt in _PROMPTS: - cells.append(_run_one_cell(register_id, lens_id, prompt)) - return cells + ChatRuntime(config=RuntimeConfig( + register_pack_id=register_id, + anchor_lens_id=lens_id, + )) + + def _run(case: dict[str, Any]) -> dict[str, Any]: + register_id = case["register_id"] + lens_id = case["lens_id"] + prompt = case["prompt"] + runtime = ChatRuntime(config=RuntimeConfig( + register_pack_id=register_id, + anchor_lens_id=lens_id, + )) + pipeline = CognitiveTurnPipeline(runtime=runtime) + result = pipeline.run(prompt) + turn_event = runtime.turn_log[-1] + return { + "prompt": prompt, + "register_id": register_id, + "lens_id": lens_id, + "surface": turn_event.surface, + "trace_hash": result.trace_hash, + "grounding_source": getattr(turn_event, "grounding_source", ""), + "register_variant_id": getattr(turn_event, "register_variant_id", ""), + "anchor_lens_id": getattr(turn_event, "anchor_lens_id", ""), + "anchor_lens_mode_label": getattr(turn_event, "anchor_lens_mode_label", ""), + } + + return _run def _cells_by( @@ -273,7 +301,7 @@ def _print_grid(cells: list[dict[str, Any]]) -> None: _say() -def run_tour(*, emit_json: bool = False) -> dict[str, Any]: +def run_tour(*, emit_json: bool = False, workers: int | None = None) -> dict[str, Any]: """Run the orthogonality tour and return a structured report.""" global _VERBOSE _VERBOSE = not emit_json @@ -285,7 +313,20 @@ def run_tour(*, emit_json: bool = False) -> dict[str, Any]: f"{len(_REGISTERS) * len(_LENSES) * len(_PROMPTS)} cells") _say() - cells = _build_grid() + cases = [ + {"register_id": register_id, "lens_id": lens_id, "prompt": prompt} + for register_id in _REGISTERS + for lens_id in _LENSES + for prompt in _PROMPTS + ] + effective_workers = normalize_workers(workers if workers is not None else 4, len(cases)) + if not emit_json: + _say(f" workers: {effective_workers}") + cells = run_cases_parallel( + cases, + partial(_build_case_runner), + n_workers=effective_workers, + ) if not emit_json: _say("-" * 76) diff --git a/evals/parallel.py b/evals/parallel.py index 66bb4989..2c608af4 100644 --- a/evals/parallel.py +++ b/evals/parallel.py @@ -1,47 +1,21 @@ -"""Parallel case-runner helper for embarrassingly-parallel eval lanes. +"""Compatibility helper for legacy parallel eval runners. -The per-case lanes (provenance, calibration, symbolic-logic, -adversarial-identity) each build a fresh ``ChatRuntime`` per case with -no shared state, so they parallelize cleanly across OS processes. - -Threading does not help here because the dominant per-case cost is -``ChatRuntime.__init__`` — pure-Python pack loading that holds the GIL. -``multiprocessing.Pool`` gives one runtime per worker and yields ~5–7× -wall-clock speedup on an 8-core machine. - -Determinism: each case is independent and the per-case scoring is a -deterministic function of the case spec. Parallel execution preserves -the same per-case results as serial execution; only the *order* of -returned results may differ, so callers should re-sort by case id or -by the input order before computing ordered metrics. - -Usage: - from evals.parallel import run_cases_parallel - - details = run_cases_parallel(cases, _run_case, workers=None) - # details is a list ordered to match cases input. - -The worker function ``run_case_fn`` must be importable at module level -(picklable). Closures and lambdas will not work. +This preserves the original ``workers=`` API used by the older lanes +while the new worker-initialized helper lives in :mod:`evals._parallel`. """ from __future__ import annotations import multiprocessing as mp import os -from typing import Any, Callable, TypeVar +from collections.abc import Callable +from typing import Any, TypeVar _R = TypeVar("_R") - -# Use 'spawn' so worker processes get a fresh Python interpreter — avoids -# forking heavy parent state (loaded numpy/torch backends, vault caches, -# language pack manifolds) into every child. _MP_CONTEXT = mp.get_context("spawn") def _default_workers() -> int: - # Cap default at a reasonable number; very high parallelism increases - # per-worker pack-load cost without proportional speedup. detected = os.cpu_count() or 4 return max(1, min(detected, 8)) @@ -52,26 +26,7 @@ def run_cases_parallel( *, workers: int | None = None, ) -> list[_R]: - """Run cases in parallel using a multiprocessing.Pool. - - Parameters - ---------- - cases - List of case dicts. Each is passed individually to - ``run_case_fn``. - run_case_fn - Module-level (importable, picklable) function that takes one - case dict and returns a per-case detail dict. - workers - Number of worker processes. Defaults to - ``min(os.cpu_count(), 8)``. Set to 1 to force serial execution - (useful for debugging). - - Returns - ------- - list[dict] - Per-case details, in the same order as the input ``cases``. - """ + """Run cases in parallel with the legacy per-case callable API.""" if not cases: return [] @@ -80,6 +35,7 @@ def run_cases_parallel( return [run_case_fn(c) for c in cases] with _MP_CONTEXT.Pool(processes=n) as pool: - # imap preserves input ordering and starts yielding before all - # tasks finish, which keeps memory bounded on large lanes. return list(pool.imap(run_case_fn, cases)) + + +__all__ = ["run_cases_parallel"] diff --git a/evals/realizer_guard/run_holdout.py b/evals/realizer_guard/run_holdout.py index c211b900..ec7af505 100644 --- a/evals/realizer_guard/run_holdout.py +++ b/evals/realizer_guard/run_holdout.py @@ -24,11 +24,13 @@ from __future__ import annotations import json import sys -from typing import Any +from functools import partial +from typing import Any, Callable from chat.runtime import ChatRuntime from core.cognition.pipeline import CognitiveTurnPipeline from core.config import RuntimeConfig +from evals._parallel import normalize_workers, run_cases_parallel from generate.realizer_guard import check_surface @@ -104,12 +106,32 @@ def _run_synthetic_one(candidate: str, expected_rule: str) -> dict[str, Any]: } -def run_holdout(*, emit_json: bool = False) -> dict[str, Any]: +def _build_runtime_case_runner() -> Callable[[str], dict[str, Any]]: + """Warm the holdout runtime once, then return a per-prompt scorer.""" + _build_runtime() + + def _run(prompt: str) -> dict[str, Any]: + return _run_runtime_one(prompt) + + return _run + + +def run_holdout(*, emit_json: bool = False, workers: int | None = None) -> dict[str, Any]: synthetic_cells = [ _run_synthetic_one(candidate, rule) for candidate, rule in _SYNTHETIC_ILLEGAL_CANDIDATES ] - runtime_cells = [_run_runtime_one(p) for p in _HOLDOUT_PROMPTS] + effective_workers = normalize_workers( + workers if workers is not None else 4, + len(_HOLDOUT_PROMPTS), + ) + if not emit_json: + print(f" workers : {effective_workers}") + runtime_cells = run_cases_parallel( + list(_HOLDOUT_PROMPTS), + partial(_build_runtime_case_runner), + n_workers=effective_workers, + ) failures = [ c for c in (*synthetic_cells, *runtime_cells) if not c["cell_supported"] diff --git a/evals/run_cognition_eval.py b/evals/run_cognition_eval.py index 2e6776c7..56891694 100644 --- a/evals/run_cognition_eval.py +++ b/evals/run_cognition_eval.py @@ -8,12 +8,15 @@ metrics. Each case gets a fresh pipeline instance for isolation. from __future__ import annotations import json +from functools import partial from pathlib import Path +from typing import Callable from chat.runtime import ChatRuntime from core.config import RuntimeConfig from core.cognition.pipeline import CognitiveTurnPipeline from evals.metrics import CaseResult, EvalReport +from evals._parallel import run_cases_parallel from generate.intent import IntentTag _CASES_PATH = Path(__file__).parent / "cognition_cases.jsonl" @@ -65,19 +68,42 @@ def _run_case(case: dict, pipeline: CognitiveTurnPipeline) -> CaseResult: ) +def _build_case_runner( + config: RuntimeConfig | None = None, +) -> Callable[[dict], CaseResult]: + """Warm worker-local caches once, then return a per-case scorer.""" + if config is None: + ChatRuntime() + else: + ChatRuntime(config=config) + + def _run(case: dict) -> CaseResult: + runtime = ChatRuntime(config=config) if config else ChatRuntime() + pipeline = CognitiveTurnPipeline(runtime) + return _run_case(case, pipeline) + + return _run + + def run_eval( cases: list[dict] | None = None, config: RuntimeConfig | None = None, + *, + workers: int | None = None, ) -> EvalReport: if cases is None: cases = load_cases() report = EvalReport() - for case in cases: - runtime = ChatRuntime(config=config) if config else ChatRuntime() - pipeline = CognitiveTurnPipeline(runtime) - case_result = _run_case(case, pipeline) + case_runner_builder = partial(_build_case_runner, config=config) + case_results = run_cases_parallel( + cases, + case_runner_builder, + n_workers=workers if workers is not None else 4, + ) + + for case_result in case_results: report.total += 1 if case_result.intent_correct: