core-labs/core
1
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

parallel eval runner (#46)

Browse source
This commit is contained in:
Shay 2026-05-19 23:51:59 -07:00 committed by GitHub
parent 37c0ea1835
commit 0eaba474ed
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
9 changed files with 329 additions and 85 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

59
core/cli.py
View file

@ -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")

83
evals/_parallel.py Normal file
View file

@ -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))

55
evals/anchor_lens_tour/run_tour.py
View file

@ -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)

33
evals/cognition/runner.py
View file

@ -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:

3
evals/framework.py
View file

@ -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,

57
evals/orthogonality_tour/run_tour.py
View file

@ -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)

62
evals/parallel.py
View file

@ -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"]

28
evals/realizer_guard/run_holdout.py
View file

@ -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"]

34
evals/run_cognition_eval.py
View file

@ -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: