1066 lines
40 KiB
Python
1066 lines
40 KiB
Python
#!/usr/bin/env python3
|
|
"""Run the repository-native Meeting Lab benchmark pipeline.
|
|
|
|
The runner intentionally orchestrates the existing production stages without
|
|
changing their prompts or semantic behavior. It adds pre-flight checks,
|
|
artifact preservation and benchmark metadata around those stages.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import platform
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from collections import Counter
|
|
from dataclasses import asdict
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from queue import Empty, Queue
|
|
from threading import Thread
|
|
from typing import Any, Callable
|
|
|
|
import requests
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
if str(REPO_ROOT) not in sys.path:
|
|
sys.path.insert(0, str(REPO_ROOT))
|
|
|
|
from src.meeting_lab.chunking.chunk_transcript import ( # noqa: E402
|
|
build_chunks,
|
|
read_transcript_blocks,
|
|
write_chunks,
|
|
)
|
|
from src.meeting_lab.consolidation.canonicalize import ( # noqa: E402
|
|
canonicalize_extractions,
|
|
write_canonicalized,
|
|
)
|
|
from src.meeting_lab.consolidation.consolidate_facts import ( # noqa: E402
|
|
build_consolidated_output,
|
|
build_consolidation_prompt,
|
|
call_ollama_with_repetition_retry,
|
|
fact_items,
|
|
parse_model_json,
|
|
repair_model_group_coverage,
|
|
resolve_num_predict,
|
|
validate_consolidated_output,
|
|
validate_group_shapes,
|
|
validate_model_groups,
|
|
write_json,
|
|
write_report as write_consolidator_report,
|
|
)
|
|
from src.meeting_lab.extraction.extract_chunks import ( # noqa: E402
|
|
DEFAULT_ENDPOINT,
|
|
build_prompt as build_extraction_prompt,
|
|
call_ollama as call_extraction_ollama,
|
|
find_normalized_chunks,
|
|
normalize_current_schema,
|
|
parse_json_response,
|
|
)
|
|
from src.meeting_lab.models.meeting_context import ( # noqa: E402
|
|
MeetingContext,
|
|
load_meeting_context,
|
|
)
|
|
from src.meeting_lab.normalization.normalize_transcript import ( # noqa: E402
|
|
normalize_block,
|
|
split_blocks,
|
|
)
|
|
from src.meeting_lab.protocol.render_working_protocol import ( # noqa: E402
|
|
DEFAULT_NUM_PREDICT as RENDERER_NUM_PREDICT,
|
|
render_working_protocol,
|
|
)
|
|
|
|
|
|
DEFAULT_MODEL = "qwen3.5:9b"
|
|
DEFAULT_TARGET_CHARS = 4500
|
|
DEFAULT_MAX_CHARS = 5500
|
|
DEFAULT_MIN_CHARS = 2500
|
|
DEFAULT_OVERLAP_BLOCKS = 0
|
|
DEFAULT_NUM_CTX = 32768
|
|
DEFAULT_TEMPERATURE = 0.0
|
|
DEFAULT_TIMEOUT = 1800
|
|
DEFAULT_PROGRESS_INTERVAL = 30
|
|
BENCHMARK_ROOT = Path("samples") / "benchmarks"
|
|
|
|
|
|
class RunnerError(RuntimeError):
|
|
"""Raised for actionable runner failures."""
|
|
|
|
def __init__(self, message: str, details: dict[str, Any] | None = None) -> None:
|
|
super().__init__(message)
|
|
self.details = details or {}
|
|
|
|
|
|
class ProgressReporter:
|
|
"""Reusable progress and timing reporter for benchmark runs."""
|
|
|
|
STAGES = (
|
|
"preflight",
|
|
"chunking",
|
|
"normalization",
|
|
"extraction",
|
|
"canonicalizer",
|
|
"semantic_consolidator",
|
|
"renderer",
|
|
)
|
|
|
|
DISPLAY_NAMES = {
|
|
"preflight": "Preflight",
|
|
"chunking": "Chunking",
|
|
"normalization": "Normalization",
|
|
"extraction": "Extraction",
|
|
"canonicalizer": "Canonicalizer",
|
|
"semantic_consolidator": "Semantic Consolidator",
|
|
"renderer": "Renderer",
|
|
}
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
stream: Any = sys.stdout,
|
|
monotonic: Callable[[], float] = time.perf_counter,
|
|
wall_clock: Callable[[], datetime] = datetime.now,
|
|
stages: tuple[str, ...] = STAGES,
|
|
) -> None:
|
|
self.stream = stream
|
|
self.monotonic = monotonic
|
|
self.wall_clock = wall_clock
|
|
self.stages = stages
|
|
self.completed: set[str] = set()
|
|
self.current_stage: str | None = None
|
|
self.pipeline_start = self.monotonic()
|
|
self.stage_starts: dict[str, float] = {}
|
|
self.stage_wall_starts: dict[str, datetime] = {}
|
|
self.stage_durations: dict[str, float] = {}
|
|
|
|
def stage_name(self, stage: str) -> str:
|
|
return self.DISPLAY_NAMES.get(stage, stage.replace("_", " ").title())
|
|
|
|
def format_hms(self, seconds: float) -> str:
|
|
seconds_int = max(0, int(round(seconds)))
|
|
hours, remainder = divmod(seconds_int, 3600)
|
|
minutes, secs = divmod(remainder, 60)
|
|
return f"{hours:02d}:{minutes:02d}:{secs:02d}"
|
|
|
|
def format_mmss(self, seconds: float) -> str:
|
|
seconds_int = max(0, int(round(seconds)))
|
|
minutes, secs = divmod(seconds_int, 60)
|
|
return f"{minutes:02d}:{secs:02d}"
|
|
|
|
def format_duration_seconds(self, seconds: float) -> str:
|
|
return f"{seconds:.1f} s"
|
|
|
|
def extraction_eta(self, completed_chunks: int, total_chunks: int, elapsed: float) -> tuple[float | None, float | None]:
|
|
if completed_chunks <= 0 or total_chunks <= 0:
|
|
return None, None
|
|
average = elapsed / completed_chunks
|
|
remaining = average * max(total_chunks - completed_chunks, 0)
|
|
return average, remaining
|
|
|
|
def pipeline_percent(self) -> int:
|
|
if not self.stages:
|
|
return 100
|
|
completed = len(self.completed)
|
|
if self.current_stage and self.current_stage not in self.completed:
|
|
completed += 0.5
|
|
return min(100, int((completed / len(self.stages)) * 100))
|
|
|
|
def progress_bar(self, percent: int, width: int = 20) -> str:
|
|
filled = int(width * percent / 100)
|
|
return "[" + "#" * filled + "-" * (width - filled) + f"] {percent}%"
|
|
|
|
def render_pipeline_progress(self) -> str:
|
|
lines = [
|
|
"=" * 60,
|
|
"Meeting Lab Benchmark",
|
|
"",
|
|
self.progress_bar(self.pipeline_percent()),
|
|
"",
|
|
]
|
|
for stage in self.stages:
|
|
marker = "x" if stage in self.completed else ">" if stage == self.current_stage else " "
|
|
lines.append(f"[{marker}] {self.stage_name(stage)}")
|
|
lines.extend(
|
|
[
|
|
"",
|
|
f"Elapsed: {self.format_hms(self.monotonic() - self.pipeline_start)}",
|
|
"=" * 60,
|
|
]
|
|
)
|
|
return "\n".join(lines)
|
|
|
|
def print_pipeline_progress(self) -> None:
|
|
print(self.render_pipeline_progress(), file=self.stream, flush=True)
|
|
|
|
def start_stage(self, stage: str) -> None:
|
|
self.current_stage = stage
|
|
self.stage_starts[stage] = self.monotonic()
|
|
self.stage_wall_starts[stage] = self.wall_clock()
|
|
self.print_pipeline_progress()
|
|
|
|
def finish_stage(self, stage: str, status: str) -> float:
|
|
now = self.monotonic()
|
|
elapsed = now - self.stage_starts.get(stage, now)
|
|
self.stage_durations[stage] = elapsed
|
|
if status == "passed":
|
|
self.completed.add(stage)
|
|
if self.current_stage == stage:
|
|
self.current_stage = None
|
|
self.print_pipeline_progress()
|
|
return elapsed
|
|
|
|
def format_extraction_progress(self, completed_chunks: int, total_chunks: int, elapsed: float) -> str:
|
|
percent = int((completed_chunks / total_chunks) * 100) if total_chunks else 100
|
|
average, remaining = self.extraction_eta(completed_chunks, total_chunks, elapsed)
|
|
average_text = self.format_mmss(average) if average is not None else "--:--"
|
|
remaining_text = "~" + self.format_hms(remaining) if remaining is not None else "unknown"
|
|
return "\n".join(
|
|
[
|
|
"Extraction",
|
|
f"Chunk {completed_chunks} / {total_chunks} ({percent}%)",
|
|
"",
|
|
f"Elapsed: {self.format_hms(elapsed)}",
|
|
f"Average: {average_text} / chunk",
|
|
f"Remaining: {remaining_text}",
|
|
]
|
|
)
|
|
|
|
def print_extraction_progress(self, completed_chunks: int, total_chunks: int, elapsed: float) -> None:
|
|
print(
|
|
self.format_extraction_progress(completed_chunks, total_chunks, elapsed),
|
|
file=self.stream,
|
|
flush=True,
|
|
)
|
|
|
|
def format_stage_heartbeat(self, stage: str, elapsed: float) -> str:
|
|
started = self.stage_wall_starts.get(stage)
|
|
started_text = started.isoformat(timespec="seconds") if started else "unknown"
|
|
return "\n".join(
|
|
[
|
|
self.stage_name(stage),
|
|
"",
|
|
"Running...",
|
|
f"Started: {started_text}",
|
|
f"Elapsed: {self.format_mmss(elapsed)}",
|
|
]
|
|
)
|
|
|
|
def print_stage_heartbeat(self, stage: str) -> None:
|
|
started = self.stage_starts.get(stage, self.monotonic())
|
|
print(
|
|
self.format_stage_heartbeat(stage, self.monotonic() - started),
|
|
file=self.stream,
|
|
flush=True,
|
|
)
|
|
|
|
def format_stage_timing_summary(self, timings: dict[str, Any]) -> str:
|
|
rows = []
|
|
for stage in self.stages:
|
|
data = timings.get(stage)
|
|
if not isinstance(data, dict):
|
|
continue
|
|
seconds = data.get("runtime_seconds")
|
|
if isinstance(seconds, (int, float)):
|
|
rows.append((self.stage_name(stage), float(seconds)))
|
|
total = sum(seconds for _stage, seconds in rows)
|
|
lines = ["Stage Timing Summary", "", f"{'Stage':<24} Duration", ""]
|
|
lines.extend(f"{stage:<24} {self.format_duration_seconds(seconds):>10}" for stage, seconds in rows)
|
|
lines.extend(["", f"{'Total':<24} {self.format_duration_seconds(total):>10}"])
|
|
return "\n".join(lines)
|
|
|
|
|
|
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(
|
|
description="Run the Meeting Lab production benchmark pipeline."
|
|
)
|
|
parser.add_argument("--input", required=True, type=Path, help="Normalized Whisper JSON input.")
|
|
parser.add_argument("--context", required=True, type=Path, help="Meeting Context YAML.")
|
|
parser.add_argument("--model", default=DEFAULT_MODEL, help=f"Ollama model tag (default: {DEFAULT_MODEL}).")
|
|
parser.add_argument(
|
|
"--benchmark-label",
|
|
default="meeting_run",
|
|
help="Output directory label prefix under samples/benchmarks.",
|
|
)
|
|
parser.add_argument("--endpoint", default=DEFAULT_ENDPOINT, help=f"Ollama generate endpoint (default: {DEFAULT_ENDPOINT}).")
|
|
parser.add_argument("--target-chars", type=int, default=DEFAULT_TARGET_CHARS)
|
|
parser.add_argument("--max-chars", type=int, default=DEFAULT_MAX_CHARS)
|
|
parser.add_argument("--min-chars", type=int, default=DEFAULT_MIN_CHARS)
|
|
parser.add_argument("--overlap-blocks", type=int, default=DEFAULT_OVERLAP_BLOCKS)
|
|
parser.add_argument("--temperature", type=float, default=DEFAULT_TEMPERATURE)
|
|
parser.add_argument("--num-ctx", type=int, default=DEFAULT_NUM_CTX)
|
|
parser.add_argument(
|
|
"--num-predict",
|
|
type=int,
|
|
default=None,
|
|
help="Extraction num_predict. Default uses the committed extraction default.",
|
|
)
|
|
parser.add_argument(
|
|
"--consolidator-num-predict",
|
|
type=int,
|
|
default=None,
|
|
help="Semantic Consolidator num_predict. Default uses committed adaptive sizing.",
|
|
)
|
|
parser.add_argument(
|
|
"--renderer-num-predict",
|
|
type=int,
|
|
default=RENDERER_NUM_PREDICT,
|
|
help=f"Renderer num_predict (default: {RENDERER_NUM_PREDICT}).",
|
|
)
|
|
parser.add_argument("--timeout", type=int, default=DEFAULT_TIMEOUT)
|
|
parser.add_argument("--progress-interval", type=int, default=DEFAULT_PROGRESS_INTERVAL)
|
|
parser.add_argument(
|
|
"--allow-loaded-models",
|
|
action="store_true",
|
|
help="Allow execution when `ollama ps` reports already loaded models.",
|
|
)
|
|
parser.add_argument(
|
|
"--output-root",
|
|
type=Path,
|
|
default=BENCHMARK_ROOT,
|
|
help="Directory where benchmark run directories are created.",
|
|
)
|
|
return parser.parse_args(argv)
|
|
|
|
|
|
def sanitize_label(label: str) -> str:
|
|
cleaned = re.sub(r"[^A-Za-z0-9_.-]+", "_", label.strip())
|
|
return cleaned.strip("._-") or "meeting_run"
|
|
|
|
|
|
def create_unique_output_dir(output_root: Path, label: str, now: Callable[[], datetime] = datetime.now) -> Path:
|
|
timestamp = now().strftime("%Y%m%d_%H%M%S")
|
|
base = output_root / f"{sanitize_label(label)}_{timestamp}"
|
|
candidate = base
|
|
suffix = 1
|
|
while candidate.exists():
|
|
candidate = output_root / f"{base.name}_{suffix:02d}"
|
|
suffix += 1
|
|
candidate.mkdir(parents=True)
|
|
return candidate
|
|
|
|
|
|
def read_json_object(path: Path) -> dict[str, Any]:
|
|
try:
|
|
data = json.loads(path.read_text(encoding="utf-8-sig"))
|
|
except json.JSONDecodeError as exc:
|
|
raise RunnerError(f"Input is not valid JSON: {path}: {exc}") from exc
|
|
if not isinstance(data, dict):
|
|
raise RunnerError(f"Input JSON must contain an object: {path}")
|
|
return data
|
|
|
|
|
|
def validate_whisper_input(path: Path) -> dict[str, Any]:
|
|
if not path.is_file():
|
|
raise RunnerError(f"Input file does not exist: {path}")
|
|
data = read_json_object(path)
|
|
segments = data.get("segments")
|
|
if not isinstance(segments, list) or not segments:
|
|
raise RunnerError("Input JSON must contain a non-empty 'segments' list.")
|
|
bad = [
|
|
index
|
|
for index, segment in enumerate(segments)
|
|
if not isinstance(segment, dict) or not isinstance(segment.get("text"), str)
|
|
]
|
|
if bad:
|
|
raise RunnerError(f"Input segments must be objects with string text; first bad index: {bad[0]}")
|
|
return data
|
|
|
|
|
|
def ollama_base_url(endpoint: str) -> str:
|
|
return endpoint.rsplit("/api/", 1)[0] if "/api/" in endpoint else endpoint.rstrip("/")
|
|
|
|
|
|
def require_ollama(
|
|
endpoint: str,
|
|
model: str,
|
|
timeout: int = 10,
|
|
allow_loaded_models: bool = False,
|
|
) -> dict[str, Any]:
|
|
base = ollama_base_url(endpoint)
|
|
try:
|
|
tags_response = requests.get(f"{base}/api/tags", timeout=timeout)
|
|
tags_response.raise_for_status()
|
|
except requests.RequestException as exc:
|
|
raise RunnerError(f"Ollama endpoint is not reachable at {base}: {exc}") from exc
|
|
|
|
tags = tags_response.json()
|
|
models = tags.get("models")
|
|
if not isinstance(models, list):
|
|
raise RunnerError(f"Ollama /api/tags returned an unexpected payload from {base}.")
|
|
installed = {item.get("name") for item in models if isinstance(item, dict)}
|
|
if model not in installed:
|
|
raise RunnerError(f"Requested model is not installed in Ollama: {model}")
|
|
|
|
active_models: list[str] = []
|
|
try:
|
|
ps_response = requests.get(f"{base}/api/ps", timeout=timeout)
|
|
ps_response.raise_for_status()
|
|
ps_models = ps_response.json().get("models", [])
|
|
if isinstance(ps_models, list):
|
|
active_models = [
|
|
str(item.get("name"))
|
|
for item in ps_models
|
|
if isinstance(item, dict) and item.get("name")
|
|
]
|
|
except requests.RequestException:
|
|
active_models = ["unavailable"]
|
|
|
|
result = {
|
|
"base_url": base,
|
|
"installed_models": sorted(installed),
|
|
"active_models": active_models,
|
|
"allow_loaded_models": allow_loaded_models,
|
|
}
|
|
if active_models and active_models != ["unavailable"] and not allow_loaded_models:
|
|
raise RunnerError(
|
|
"Ollama has loaded model(s). Stop them before a reference benchmark "
|
|
"or pass --allow-loaded-models to record and allow this condition: "
|
|
+ ", ".join(active_models),
|
|
details={"preflight": {"ollama": result}},
|
|
)
|
|
return result
|
|
|
|
|
|
def system_metadata(model: str, endpoint: str) -> dict[str, Any]:
|
|
return {
|
|
"branch": run_command_text(["git", "rev-parse", "--abbrev-ref", "HEAD"]),
|
|
"commit": run_command_text(["git", "rev-parse", "HEAD"]),
|
|
"operating_system": platform.platform(),
|
|
"cpu": platform.processor() or os.environ.get("PROCESSOR_IDENTIFIER"),
|
|
"ram_bytes": total_ram_bytes(),
|
|
"python_version": sys.version.replace("\n", " "),
|
|
"ollama_version": run_command_text(["ollama", "--version"]),
|
|
"model": model,
|
|
"endpoint": endpoint,
|
|
}
|
|
|
|
|
|
def run_command_text(command: list[str]) -> str | None:
|
|
try:
|
|
completed = subprocess.run(
|
|
command,
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
errors="replace",
|
|
)
|
|
except OSError:
|
|
return None
|
|
text = (completed.stdout or completed.stderr).strip()
|
|
return text or None
|
|
|
|
|
|
def total_ram_bytes() -> int | None:
|
|
try:
|
|
import psutil # type: ignore[import-not-found]
|
|
except ModuleNotFoundError:
|
|
return None
|
|
return int(psutil.virtual_memory().total)
|
|
|
|
|
|
def run_stage(
|
|
name: str,
|
|
timings: dict[str, Any],
|
|
func: Callable[[], Any],
|
|
reporter: ProgressReporter,
|
|
*,
|
|
heartbeat: bool = False,
|
|
heartbeat_interval: int = DEFAULT_PROGRESS_INTERVAL,
|
|
) -> Any:
|
|
reporter.start_stage(name)
|
|
print(f"[{datetime.now().isoformat(timespec='seconds')}] Starting {name}", flush=True)
|
|
try:
|
|
if heartbeat:
|
|
result = run_with_heartbeat(
|
|
name,
|
|
func,
|
|
reporter=reporter,
|
|
heartbeat_interval=heartbeat_interval,
|
|
)
|
|
else:
|
|
result = func()
|
|
except Exception:
|
|
elapsed = reporter.finish_stage(name, "failed")
|
|
timings[name] = {"runtime_seconds": round(elapsed, 3), "status": "failed"}
|
|
print(f"[{datetime.now().isoformat(timespec='seconds')}] FAILED {name} after {elapsed:.3f}s", flush=True)
|
|
raise
|
|
elapsed = reporter.finish_stage(name, "passed")
|
|
timings[name] = {"runtime_seconds": round(elapsed, 3), "status": "passed"}
|
|
print(f"[{datetime.now().isoformat(timespec='seconds')}] Finished {name} in {elapsed:.3f}s", flush=True)
|
|
return result
|
|
|
|
|
|
def run_with_heartbeat(
|
|
name: str,
|
|
func: Callable[[], Any],
|
|
*,
|
|
reporter: ProgressReporter,
|
|
heartbeat_interval: int,
|
|
) -> Any:
|
|
results: Queue[tuple[str, Any]] = Queue()
|
|
|
|
def worker() -> None:
|
|
try:
|
|
results.put(("result", func()))
|
|
except BaseException as exc: # noqa: BLE001 - forwarded to caller.
|
|
results.put(("error", exc))
|
|
|
|
thread = Thread(target=worker, daemon=True)
|
|
thread.start()
|
|
while True:
|
|
try:
|
|
status, payload = results.get(timeout=max(heartbeat_interval, 1))
|
|
except Empty:
|
|
reporter.print_stage_heartbeat(name)
|
|
continue
|
|
if status == "error":
|
|
raise payload
|
|
return payload
|
|
|
|
|
|
def write_normalized_chunk(input_path: Path, output_path: Path, changes_path: Path) -> dict[str, Any]:
|
|
source = input_path.read_text(encoding="utf-8-sig")
|
|
blocks = split_blocks(source)
|
|
normalized_blocks = []
|
|
changes = []
|
|
for block_number, block in enumerate(blocks, start=1):
|
|
normalized, change = normalize_block(block, block_number)
|
|
normalized_blocks.append(normalized)
|
|
if change is not None:
|
|
changes.append(change)
|
|
output_path.parent.mkdir(parents=True, exist_ok=True)
|
|
output_path.write_text("\n\n".join(normalized_blocks).strip() + "\n", encoding="utf-8")
|
|
report = {
|
|
"source_file": input_path.name,
|
|
"output_file": output_path.name,
|
|
"blocks_total": len(blocks),
|
|
"blocks_changed": len(changes),
|
|
"changes": [asdict(change) for change in changes],
|
|
}
|
|
changes_path.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
return report
|
|
|
|
|
|
def extraction_output_path(chunk_path: Path, output_dir: Path) -> Path:
|
|
name = chunk_path.name.replace("_normalized.txt", "_extraction.json")
|
|
return output_dir / name
|
|
|
|
|
|
def run_extraction_stage(
|
|
normalized_dir: Path,
|
|
output_dir: Path,
|
|
model: str,
|
|
endpoint: str,
|
|
timeout: int,
|
|
temperature: float,
|
|
num_predict: int | None,
|
|
num_ctx: int,
|
|
meeting_context: MeetingContext,
|
|
progress_reporter: ProgressReporter | None = None,
|
|
) -> list[Path]:
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
output_paths = []
|
|
chunk_paths = find_normalized_chunks(normalized_dir)
|
|
total_chunks = len(chunk_paths)
|
|
start = progress_reporter.monotonic() if progress_reporter else time.perf_counter()
|
|
|
|
def elapsed() -> float:
|
|
clock = progress_reporter.monotonic if progress_reporter else time.perf_counter
|
|
return clock() - start
|
|
|
|
for chunk_index, chunk_path in enumerate(chunk_paths, start=1):
|
|
if progress_reporter:
|
|
progress_reporter.print_extraction_progress(chunk_index - 1, total_chunks, elapsed())
|
|
output_path = extraction_output_path(chunk_path, output_dir)
|
|
raw_path = output_path.with_suffix(".raw.txt")
|
|
metadata_path = output_path.with_name(output_path.stem + "_metadata.json")
|
|
print(f"Extracting {chunk_path.name} -> {output_path.name}", flush=True)
|
|
transcript = chunk_path.read_text(encoding="utf-8-sig").strip()
|
|
if not transcript:
|
|
raise RunnerError(f"Normalized chunk is empty: {chunk_path}")
|
|
prompt = build_extraction_prompt(chunk_path.name, transcript, meeting_context=meeting_context)
|
|
raw_text, metadata = call_extraction_ollama(
|
|
endpoint=endpoint,
|
|
model=model,
|
|
prompt=prompt,
|
|
timeout=timeout,
|
|
temperature=temperature,
|
|
num_predict=num_predict,
|
|
num_ctx=num_ctx,
|
|
)
|
|
raw_path.write_text(raw_text + "\n", encoding="utf-8")
|
|
metadata_path.write_text(json.dumps(metadata, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
parsed = parse_json_response(raw_text)
|
|
extraction = normalize_current_schema(parsed)
|
|
extraction["context"] = meeting_context.provenance()
|
|
output_path.write_text(json.dumps(extraction, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
output_paths.append(output_path)
|
|
if progress_reporter:
|
|
progress_reporter.print_extraction_progress(chunk_index, total_chunks, elapsed())
|
|
return output_paths
|
|
|
|
|
|
def source_coverage_report_from_groups(
|
|
groups: Any,
|
|
expected_source_ids: set[str],
|
|
) -> dict[str, Any]:
|
|
seen = []
|
|
violations = []
|
|
if not isinstance(groups, list):
|
|
return {
|
|
"valid": False,
|
|
"expected_source_fact_count": len(expected_source_ids),
|
|
"observed_source_fact_occurrence_count": 0,
|
|
"group_count": None,
|
|
"violations": [{"type": "groups_not_list"}],
|
|
}
|
|
for group_index, group in enumerate(groups):
|
|
source_ids = group.get("source_item_ids") if isinstance(group, dict) else None
|
|
if not isinstance(source_ids, list) or not source_ids:
|
|
violations.append({"type": "invalid_group_source_ids", "group_index": group_index})
|
|
continue
|
|
string_ids = [item_id for item_id in source_ids if isinstance(item_id, str)]
|
|
seen.extend(string_ids)
|
|
for unknown in sorted(set(string_ids) - expected_source_ids):
|
|
violations.append({"type": "unknown_id", "id": unknown, "group_index": group_index})
|
|
counts = Counter(seen)
|
|
for item_id, count in sorted(counts.items()):
|
|
if count > 1:
|
|
violations.append({"type": "duplicate_id", "id": item_id, "count": count})
|
|
for missing in sorted(expected_source_ids - set(seen)):
|
|
violations.append({"type": "missing_id", "id": missing})
|
|
return {
|
|
"valid": not violations,
|
|
"expected_source_fact_count": len(expected_source_ids),
|
|
"observed_source_fact_occurrence_count": len(seen),
|
|
"group_count": len(groups),
|
|
"violations": violations,
|
|
}
|
|
|
|
|
|
def run_semantic_consolidator_stage(
|
|
canonicalized_path: Path,
|
|
output_dir: Path,
|
|
model: str,
|
|
endpoint: str,
|
|
timeout: int,
|
|
num_ctx: int,
|
|
num_predict: int | None,
|
|
progress_interval: int,
|
|
) -> dict[str, Any]:
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
canonicalized = read_json_object(canonicalized_path)
|
|
facts = fact_items(canonicalized)
|
|
prompt = build_consolidation_prompt(facts)
|
|
prompt_token_estimate = (len(prompt) + 3) // 4
|
|
resolved_num_predict = resolve_num_predict(
|
|
requested_num_predict=num_predict,
|
|
facts=facts,
|
|
prompt_token_estimate=prompt_token_estimate,
|
|
num_ctx=num_ctx,
|
|
)
|
|
raw_text, response_data, runtime, attempts = call_ollama_with_repetition_retry(
|
|
endpoint=endpoint,
|
|
model=model,
|
|
prompt=prompt,
|
|
timeout=timeout,
|
|
num_ctx=num_ctx,
|
|
num_predict=resolved_num_predict,
|
|
think=False,
|
|
progress_interval=progress_interval,
|
|
output_dir=output_dir,
|
|
)
|
|
output_path = output_dir / "consolidated_extractions.json"
|
|
report_path = output_dir / "report.md"
|
|
repair_metadata_path = output_dir / "repair_metadata.json"
|
|
model_output = parse_model_json(raw_text)
|
|
expected_fact_ids = {item["item_id"] for item in facts}
|
|
validator_before = source_coverage_report_from_groups(
|
|
model_output.get("groups"),
|
|
expected_fact_ids,
|
|
)
|
|
write_json(output_dir / "validator_before.json", validator_before)
|
|
model_output, repair_changes = repair_model_group_coverage(model_output, facts)
|
|
if repair_changes:
|
|
write_json(
|
|
repair_metadata_path,
|
|
{
|
|
"scope": "semantic_consolidator_v0_source_coverage",
|
|
"llm_used": False,
|
|
"repair_count": len(repair_changes),
|
|
"repairs": repair_changes,
|
|
},
|
|
)
|
|
else:
|
|
write_json(
|
|
repair_metadata_path,
|
|
{
|
|
"scope": "semantic_consolidator_v0_source_coverage",
|
|
"llm_used": False,
|
|
"repair_count": 0,
|
|
"repairs": [],
|
|
},
|
|
)
|
|
write_json(output_dir / "repaired_model_groups.json", model_output)
|
|
groups = validate_model_groups(model_output, expected_fact_ids)
|
|
validate_group_shapes(groups)
|
|
output = build_consolidated_output(canonicalized, groups)
|
|
validate_consolidated_output(canonicalized, output)
|
|
write_json(output_path, output)
|
|
validator_after = source_coverage_report_from_groups(
|
|
[item for item in output["items"] if item.get("category") == "fact"],
|
|
expected_fact_ids,
|
|
)
|
|
# Output fact items use source_item_ids and should be covered by the same checker.
|
|
write_json(output_dir / "validator_after.json", validator_after)
|
|
write_consolidator_report(
|
|
path=report_path,
|
|
model=model,
|
|
runtime=runtime,
|
|
prompt_chars=len(prompt),
|
|
prompt_token_estimate=prompt_token_estimate,
|
|
num_predict=resolved_num_predict,
|
|
fact_count=len(facts),
|
|
groups=groups,
|
|
output_path=output_path,
|
|
repair_changes=repair_changes,
|
|
llm_call_count=len(attempts),
|
|
)
|
|
return {
|
|
"runtime_seconds": runtime,
|
|
"llm_call_count": len(attempts),
|
|
"attempts": attempts,
|
|
"fact_count": len(facts),
|
|
"num_predict": resolved_num_predict,
|
|
"validator_before": validator_before,
|
|
"validator_after": validator_after,
|
|
"repair_count": len(repair_changes),
|
|
"response_metadata": response_data,
|
|
}
|
|
|
|
|
|
def category_counts_from_extractions(extractions_dir: Path) -> dict[str, int]:
|
|
categories = ("facts", "decisions", "todos", "questions", "positions", "technical")
|
|
counts = {category: 0 for category in categories}
|
|
for path in sorted(extractions_dir.glob("chunk_*_extraction.json")):
|
|
data = read_json_object(path)
|
|
for category in categories:
|
|
values = data.get(category, [])
|
|
if isinstance(values, list):
|
|
counts[category] += len(values)
|
|
return counts
|
|
|
|
|
|
def chunk_distribution(chunks_dir: Path) -> dict[str, Any]:
|
|
manifest = read_json_object(chunks_dir / "manifest.json")
|
|
chunks = manifest.get("chunks", [])
|
|
sizes = [chunk["chars"] for chunk in chunks if isinstance(chunk, dict) and isinstance(chunk.get("chars"), int)]
|
|
return {
|
|
"chunk_count": manifest.get("chunk_count"),
|
|
"min_chars": min(sizes) if sizes else None,
|
|
"max_chars": max(sizes) if sizes else None,
|
|
"avg_chars": round(sum(sizes) / len(sizes), 1) if sizes else None,
|
|
"chunks": sizes,
|
|
}
|
|
|
|
|
|
def write_benchmark_report(path: Path, metadata: dict[str, Any]) -> None:
|
|
renderer = metadata.get("renderer", {})
|
|
semantic = metadata.get("semantic_consolidator", {})
|
|
timings = metadata.get("runtime_seconds_by_stage", {})
|
|
ollama = metadata.get("preflight", {}).get("ollama", {})
|
|
reporter = ProgressReporter()
|
|
lines = [
|
|
"# Meeting Lab Benchmark Report",
|
|
"",
|
|
f"- Status: `{metadata.get('status')}`",
|
|
f"- Input: `{metadata.get('input')}`",
|
|
f"- Meeting Context: `{metadata.get('context')}`",
|
|
f"- Model: `{metadata.get('model')}`",
|
|
f"- Output directory: `{metadata.get('output_dir')}`",
|
|
"",
|
|
"## Configuration",
|
|
"",
|
|
f"- Chunking: target={metadata['configuration']['target_chars']}, max={metadata['configuration']['max_chars']}, min={metadata['configuration']['min_chars']}, overlap={metadata['configuration']['overlap_blocks']}",
|
|
f"- LLM: think=false, temperature={metadata['configuration']['temperature']}, num_ctx={metadata['configuration']['num_ctx']}",
|
|
f"- Allow loaded Ollama models: {metadata['configuration'].get('allow_loaded_models')}",
|
|
"",
|
|
"## Ollama Cleanliness",
|
|
"",
|
|
f"- Detected loaded models: `{ollama.get('active_models')}`",
|
|
f"- Loaded-model override: `{ollama.get('allow_loaded_models', metadata['configuration'].get('allow_loaded_models'))}`",
|
|
"",
|
|
"## Runtime",
|
|
"",
|
|
"| Stage | Seconds | Status |",
|
|
"|---|---:|---|",
|
|
]
|
|
for stage, data in timings.items():
|
|
lines.append(f"| {stage} | {data.get('runtime_seconds')} | {data.get('status')} |")
|
|
lines.extend(
|
|
[
|
|
"",
|
|
"## Stage Timing Summary",
|
|
"",
|
|
"```text",
|
|
reporter.format_stage_timing_summary(timings),
|
|
"```",
|
|
"",
|
|
f"- Total wall-clock runtime: {metadata.get('total_runtime_seconds')} seconds",
|
|
f"- Average extraction time per chunk: {metadata.get('average_extraction_seconds_per_chunk')} seconds",
|
|
"",
|
|
"## Counts",
|
|
"",
|
|
f"- Chunk distribution: `{metadata.get('chunk_distribution')}`",
|
|
f"- Extraction counts: `{metadata.get('extraction_counts')}`",
|
|
f"- Canonicalizer stats: `{metadata.get('canonicalizer_stats')}`",
|
|
"",
|
|
"## Validation",
|
|
"",
|
|
f"- Validator before: valid={semantic.get('validator_before', {}).get('valid')}, violations={len(semantic.get('validator_before', {}).get('violations', []))}",
|
|
f"- Validator after: valid={semantic.get('validator_after', {}).get('valid')}, violations={len(semantic.get('validator_after', {}).get('violations', []))}",
|
|
f"- Deterministic repair count: {semantic.get('repair_count')}",
|
|
f"- Renderer contract valid: {renderer.get('valid')}",
|
|
f"- Renderer violations: {len(renderer.get('violations', [])) if isinstance(renderer.get('violations'), list) else 'n/a'}",
|
|
]
|
|
)
|
|
if metadata.get("failure"):
|
|
lines.extend(["", "## Failure", "", str(metadata["failure"])])
|
|
path.write_text("\n".join(lines).rstrip() + "\n", encoding="utf-8")
|
|
|
|
|
|
def preflight(args: argparse.Namespace, output_dir: Path) -> tuple[dict[str, Any], MeetingContext]:
|
|
input_data = validate_whisper_input(args.input)
|
|
if not args.context.is_file():
|
|
raise RunnerError(f"Meeting Context file does not exist: {args.context}")
|
|
meeting_context = load_meeting_context(args.context)
|
|
ollama = require_ollama(
|
|
args.endpoint,
|
|
args.model,
|
|
allow_loaded_models=args.allow_loaded_models,
|
|
)
|
|
if args.target_chars <= 0 or args.max_chars <= 0 or args.min_chars < 0:
|
|
raise RunnerError("Chunk size parameters must be positive.")
|
|
if args.min_chars > args.target_chars:
|
|
raise RunnerError("--min-chars must not exceed --target-chars.")
|
|
if args.target_chars > args.max_chars:
|
|
raise RunnerError("--target-chars must not exceed --max-chars.")
|
|
if args.overlap_blocks < 0:
|
|
raise RunnerError("--overlap-blocks must not be negative.")
|
|
probe = output_dir / ".write_test"
|
|
probe.write_text("ok", encoding="utf-8")
|
|
probe.unlink()
|
|
return {
|
|
"input_segments": len(input_data["segments"]),
|
|
"input_segment_text_chars": sum(len(segment.get("text", "")) for segment in input_data["segments"]),
|
|
"input_text_chars": len(str(input_data.get("text", ""))),
|
|
"ollama": ollama,
|
|
}, meeting_context
|
|
|
|
|
|
def run_pipeline(args: argparse.Namespace) -> tuple[int, Path, Path, Path | None]:
|
|
output_dir = create_unique_output_dir(args.output_root, args.benchmark_label)
|
|
report_path = output_dir / "benchmark_report.md"
|
|
metadata_path = output_dir / "run_metadata.json"
|
|
protocol_path: Path | None = None
|
|
timings: dict[str, Any] = {}
|
|
start = time.perf_counter()
|
|
reporter = ProgressReporter()
|
|
metadata: dict[str, Any] = {
|
|
"status": "running",
|
|
"input": str(args.input),
|
|
"context": str(args.context),
|
|
"model": args.model,
|
|
"output_dir": str(output_dir),
|
|
"configuration": {
|
|
"target_chars": args.target_chars,
|
|
"max_chars": args.max_chars,
|
|
"min_chars": args.min_chars,
|
|
"overlap_blocks": args.overlap_blocks,
|
|
"temperature": args.temperature,
|
|
"num_ctx": args.num_ctx,
|
|
"num_predict": args.num_predict,
|
|
"consolidator_num_predict": args.consolidator_num_predict,
|
|
"renderer_num_predict": args.renderer_num_predict,
|
|
"think": False,
|
|
"semantic_retries": 0,
|
|
"manual_intervention": False,
|
|
"allow_loaded_models": args.allow_loaded_models,
|
|
},
|
|
"system": system_metadata(args.model, args.endpoint),
|
|
"runtime_seconds_by_stage": timings,
|
|
}
|
|
try:
|
|
preflight_data, meeting_context = run_stage(
|
|
"preflight",
|
|
timings,
|
|
lambda: preflight(args, output_dir),
|
|
reporter,
|
|
)
|
|
metadata["preflight"] = preflight_data
|
|
|
|
chunks_dir = output_dir / "chunks"
|
|
run_stage(
|
|
"chunking",
|
|
timings,
|
|
lambda: write_chunks(
|
|
build_chunks(
|
|
read_transcript_blocks(args.input),
|
|
target_chars=args.target_chars,
|
|
max_chars=args.max_chars,
|
|
min_chars=args.min_chars,
|
|
overlap_blocks=args.overlap_blocks,
|
|
),
|
|
chunks_dir,
|
|
args.input.name,
|
|
),
|
|
reporter,
|
|
)
|
|
metadata["chunk_distribution"] = chunk_distribution(chunks_dir)
|
|
|
|
normalized_dir = output_dir / "normalized_chunks"
|
|
|
|
def normalize_all() -> None:
|
|
normalized_dir.mkdir(parents=True, exist_ok=True)
|
|
for chunk_path in sorted(chunks_dir.glob("chunk_*.txt")):
|
|
base = chunk_path.stem
|
|
write_normalized_chunk(
|
|
chunk_path,
|
|
normalized_dir / f"{base}_normalized.txt",
|
|
normalized_dir / f"{base}_changes.json",
|
|
)
|
|
shutil.copyfile(chunks_dir / "manifest.json", normalized_dir / "manifest.json")
|
|
|
|
run_stage("normalization", timings, normalize_all, reporter)
|
|
|
|
extractions_dir = output_dir / "extractions"
|
|
extraction_paths = run_stage(
|
|
"extraction",
|
|
timings,
|
|
lambda: run_extraction_stage(
|
|
normalized_dir=normalized_dir,
|
|
output_dir=extractions_dir,
|
|
model=args.model,
|
|
endpoint=args.endpoint,
|
|
timeout=args.timeout,
|
|
temperature=args.temperature,
|
|
num_predict=args.num_predict,
|
|
num_ctx=args.num_ctx,
|
|
meeting_context=meeting_context,
|
|
progress_reporter=reporter,
|
|
),
|
|
reporter,
|
|
)
|
|
metadata["extraction_files"] = len(extraction_paths)
|
|
metadata["extraction_counts"] = category_counts_from_extractions(extractions_dir)
|
|
|
|
canonicalized_path = output_dir / "canonicalizer" / "canonicalized_extractions.json"
|
|
canonicalized = run_stage(
|
|
"canonicalizer",
|
|
timings,
|
|
lambda: canonicalize_extractions(
|
|
extractions_dir,
|
|
merge_duplicates=True,
|
|
meeting_context_path=args.context,
|
|
),
|
|
reporter,
|
|
heartbeat=True,
|
|
heartbeat_interval=args.progress_interval,
|
|
)
|
|
write_canonicalized(canonicalized, canonicalized_path)
|
|
metadata["canonicalizer_stats"] = canonicalized.get("stats")
|
|
|
|
semantic_metadata = run_stage(
|
|
"semantic_consolidator",
|
|
timings,
|
|
lambda: run_semantic_consolidator_stage(
|
|
canonicalized_path=canonicalized_path,
|
|
output_dir=output_dir / "semantic_consolidator",
|
|
model=args.model,
|
|
endpoint=args.endpoint,
|
|
timeout=args.timeout,
|
|
num_ctx=args.num_ctx,
|
|
num_predict=args.consolidator_num_predict,
|
|
progress_interval=max(args.timeout, args.progress_interval) + 1,
|
|
),
|
|
reporter,
|
|
heartbeat=True,
|
|
heartbeat_interval=args.progress_interval,
|
|
)
|
|
metadata["semantic_consolidator"] = semantic_metadata
|
|
|
|
renderer_metadata = run_stage(
|
|
"renderer",
|
|
timings,
|
|
lambda: render_working_protocol(
|
|
input_path=output_dir / "semantic_consolidator" / "consolidated_extractions.json",
|
|
output_dir=output_dir / "working_protocol",
|
|
model=args.model,
|
|
endpoint=args.endpoint,
|
|
timeout=args.timeout,
|
|
num_ctx=args.num_ctx,
|
|
num_predict=args.renderer_num_predict,
|
|
think=False,
|
|
),
|
|
reporter,
|
|
heartbeat=True,
|
|
heartbeat_interval=args.progress_interval,
|
|
)
|
|
metadata["renderer"] = {
|
|
"valid": renderer_metadata.get("valid"),
|
|
"runtime_seconds": renderer_metadata.get("runtime_seconds"),
|
|
"violations": read_json_object(output_dir / "working_protocol" / "validation_report.json").get("violations", []),
|
|
"metadata": renderer_metadata,
|
|
}
|
|
if renderer_metadata.get("output_path"):
|
|
protocol_path = Path(str(renderer_metadata["output_path"]))
|
|
metadata["status"] = "completed"
|
|
# Renderer contract invalid is a completed benchmark with failed final protocol.
|
|
exit_code = 0
|
|
except Exception as exc: # noqa: BLE001 - preserve artifacts and report.
|
|
metadata["status"] = "failed"
|
|
metadata["failure"] = f"{type(exc).__name__}: {exc}"
|
|
if isinstance(exc, RunnerError) and exc.details:
|
|
metadata.update(exc.details)
|
|
exit_code = 2
|
|
print(f"Error: {metadata['failure']}", file=sys.stderr)
|
|
|
|
total_runtime = time.perf_counter() - start
|
|
metadata["total_runtime_seconds"] = round(total_runtime, 3)
|
|
chunk_count = metadata.get("chunk_distribution", {}).get("chunk_count")
|
|
extraction_runtime = timings.get("extraction", {}).get("runtime_seconds")
|
|
metadata["average_extraction_seconds_per_chunk"] = (
|
|
round(extraction_runtime / chunk_count, 3)
|
|
if isinstance(extraction_runtime, (int, float)) and isinstance(chunk_count, int) and chunk_count
|
|
else None
|
|
)
|
|
metadata_path.write_text(json.dumps(metadata, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
write_benchmark_report(report_path, metadata)
|
|
return exit_code, output_dir, report_path, protocol_path
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
args = parse_args(argv)
|
|
try:
|
|
exit_code, output_dir, report_path, protocol_path = run_pipeline(args)
|
|
except RunnerError as exc:
|
|
print(f"Error: {exc}", file=sys.stderr)
|
|
return 2
|
|
print(f"Output directory: {output_dir}")
|
|
print(f"Benchmark report: {report_path}")
|
|
if protocol_path is not None and protocol_path.exists():
|
|
print(f"Protocol: {protocol_path}")
|
|
else:
|
|
print("Protocol: not written")
|
|
return exit_code
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|