cortexm/bench/harness.py
Rama main · réplica de lectura del repositorio original.
"""BEAM-style harness runner.
Protocol (mirrors the plan's Phase 2 "Proof of God" benchmark):
1. Generate a seeded synthetic conversation at bucket scale
(128K / 500K / 1M / 10M estimated tokens).
2. Ingest under the μ=0 protocol — zero LLM calls, asserted.
3. Probe with questions across the 10 BEAM abilities.
4. Score with the deterministic nugget judge (offline, reproducible).
5. Compare against BM25-RAG and vector-only baselines.
6. Report per-ability accuracy, ingest throughput, storage, latency,
and trust metrics (provenance completeness, audit latency, cost).
A pluggable LLM reader/judge slot (canonical protocol: gpt-5 reader +
judge) is exposed via ``llm_judge=None``; the deterministic judge keeps
the whole harness honest and free.
"""
from __future__ import annotations
import json
import time
from dataclasses import dataclass, field
from datetime import datetime, timezone
from cortexm import metrics
from cortexm.api.memory import Memory
from cortexm.bench.abilities import (ABILITY_NAMES, ABILITIES, build_probes,
judge)
from cortexm.bench.baselines import BM25Index, bm25_context, vector_only_context
from cortexm.bench.generator import generate
@dataclass
class BucketResult:
bucket: str
n_questions: int = 0
per_ability: dict = field(default_factory=dict)
per_system: dict = field(default_factory=dict)
ingest: dict = field(default_factory=dict)
corpus: dict = field(default_factory=dict)
trust: dict = field(default_factory=dict)
def to_dict(self) -> dict:
return {
"bucket": self.bucket, "n_questions": self.n_questions,
"per_ability": self.per_ability, "per_system": self.per_system,
"ingest": self.ingest, "corpus": self.corpus, "trust": self.trust,
}
def run_bucket(bucket: str, seed: int = 42, systems=("context_m", "vector_only", "bm25"),
db_path: str = ":memory:", max_probes: int | None = None,
llm_judge=None) -> BucketResult:
res = BucketResult(bucket=bucket)
t0 = time.time()
corpus = generate(bucket, seed=seed)
res.corpus = {
"sessions": len(corpus.sessions),
"personas": len(corpus.personas),
"estimated_tokens": corpus.total_tokens,
"generation_seconds": round(corpus.generation_seconds, 2),
}
from cortexm.config import Config as _Cfg
cfg = _Cfg(db_path=db_path) if db_path != ":memory:" else _Cfg()
cfg.apply_rules_each_add = False # bulk mode: Datalog after ingest
memory = Memory(cfg)
metrics.reset_counters()
# ---- ingest (μ=0) ----------------------------------------------------
t_ingest = time.time()
n_msgs = 0
for user_id, date, msgs in corpus.sessions:
payload = [{"role": role, "content": text, "timestamp": date}
for role, text in msgs]
n_msgs += len(msgs)
memory.add(payload, user_id=user_id, timestamp=date)
n_derived = memory.apply_rules()
ingest_s = time.time() - t_ingest
stats = memory.stats()
res.ingest = {
"wall_seconds": round(ingest_s, 2),
"messages": n_msgs,
"tokens_per_second": int(corpus.total_tokens / max(ingest_s, 1e-9)),
"messages_per_second": round(n_msgs / max(ingest_s, 1e-9), 1),
"llm_calls": metrics.counters()["llm_calls"],
"u0_protocol": stats["u0_protocol"],
"facts": stats["facts"],
"active_facts": stats["active_facts"],
"chunks": stats["chunks"],
"commits": stats["commits"],
"derived_facts": stats["derived"],
"deferred_rule_pass": True,
}
# ---- probes -----------------------------------------------------------
probes = build_probes(corpus.personas, __import__("random").Random(seed))
by_ability: dict[str, list] = {a: [] for a in ABILITIES}
for p in probes:
by_ability[p.ability].append(p)
if max_probes:
for a in ABILITIES:
by_ability[a] = by_ability[a][:max_probes]
res.n_questions = sum(len(v) for v in by_ability.values())
# ---- baseline indexes ---------------------------------------------------
chunks_by_user: dict[str, list[dict]] = {}
for user_id, date, msgs in corpus.sessions:
docs = chunks_by_user.setdefault(user_id, [])
for i, (role, text) in enumerate(msgs):
if len(text) > 12: # skip tiny interjections
docs.append({"id": f"{user_id}:{len(docs)}", "text": text})
bm25 = {uid: BM25Index(docs) for uid, docs in chunks_by_user.items()}
# ---- evaluate ---------------------------------------------------------
per_system: dict[str, dict[str, float]] = {s: {a: 0.0 for a in ABILITIES}
for s in systems}
counts = {a: len(by_ability[a]) for a in ABILITIES}
details: list[dict] = []
latencies = {s: [] for s in systems}
provenance_ok = 0
provenance_checks = 0
for ability in ABILITIES:
for probe in by_ability[ability]:
for system in systems:
t_q = time.time()
if system == "context_m":
out = memory.search(probe.question, user_id=probe.user_id, k=12)
context = out["context_block"]
provenance_checks += 1
if out["provenance"]["verification"]:
provenance_ok += 1
elif system == "vector_only":
context = vector_only_context(memory, probe.question,
probe.user_id)
elif system == "bm25":
context = bm25_context(bm25[probe.user_id], probe.question)
latencies[system].append(time.time() - t_q)
score_fn = llm_judge if llm_judge else judge
score, detail = score_fn(probe, context)
per_system[system][ability] += score
if system == "context_m":
details.append({"ability": ability,
"question": probe.question,
"score": score, "detail": detail,
"context": context[:400]})
for system in systems:
res.per_system[system] = {
"overall": round(sum(per_system[system][a] for a in ABILITIES)
/ max(res.n_questions, 1), 4),
"per_ability": {
a: round(per_system[system][a] / max(counts[a], 1), 4)
for a in ABILITIES if counts[a]
},
"mean_latency_ms": round(
sum(latencies[system]) / max(len(latencies[system]), 1) * 1e3, 2),
}
res.per_ability = res.per_system.get("context_m", {}).get("per_ability", {})
res.trust = {
"provenance_completeness": round(
provenance_ok / max(provenance_checks, 1), 4),
"audit_latency_ms": round(
sum(latencies.get("context_m", [])) /
max(len(latencies.get("context_m", [])), 1) * 1e3, 2),
"u0_ingest_llm_calls": metrics.counters()["llm_calls"],
"storage": memory.storage_stats(),
"hash_provider": stats["hash_provider"],
"codec": stats["codec"],
"vsa_mode": stats["vsa_mode"],
"wall_seconds_total": round(time.time() - t0, 2),
}
memory.close()
res.details = details # type: ignore[attr-defined]
return res
def format_report(results: list[BucketResult]) -> str:
lines = ["# Context-M — BEAM-Style Benchmark Results", ""]
for r in results:
lines.append(f"## Bucket: {r.bucket.upper()} "
f"({r.corpus['estimated_tokens']:,} est. tokens, "
f"{r.n_questions} questions)")
lines.append("")
sys_names = list(r.per_system.keys())
header = "| System | Overall | " + " | ".join(ABILITIES) + " |"
sep = "|---" * (2 + len(ABILITIES)) + "|"
lines.append(header)
lines.append(sep)
for s in sys_names:
d = r.per_system[s]
row = [f"**{d['overall']:.1%}**" if s == "context_m" else f"{d['overall']:.1%}"]
for a in ABILITIES:
v = d["per_ability"].get(a)
row.append(f"{v:.0%}" if v is not None else "—")
lines.append(f"| {s} | " + " | ".join(row) + " |")
lines.append("")
lines.append(f"- Ingest: {r.ingest['wall_seconds']}s for "
f"{r.ingest['tokens_per_second']:,} tokens/s "
f"(μ=0: {r.ingest['u0_protocol']}, "
f"{r.ingest['llm_calls']} LLM calls)")
lines.append(f"- Memory: {r.ingest['facts']:,} facts / "
f"{r.ingest['chunks']:,} chunks / "
f"{r.ingest['commits']:,} commits "
f"({r.ingest['derived_facts']} derived by Datalog)")
lines.append(f"- Provenance completeness: "
f"{r.trust['provenance_completeness']:.1%} | "
f"retrieval latency p50≈{r.trust['audit_latency_ms']}ms")
lines.append("")
return "\n".join(lines)