from __future__ import annotations import argparse import json import os import re import sqlite3 import sys import time import urllib.request import zipfile from datetime import date, datetime, timedelta, timezone from pathlib import Path from typing import Any ROOT = Path(__file__).resolve().parents[1] RUNS_DIR = ROOT / "runs" WHEEL = ROOT / "wheels" / "sibyl_memory_client-0.4.9-py3-none-any.whl" if WHEEL.exists(): extracted_wheel = Path("/tmp") / "sibyl-memory-client-0.4.9-wheel" if not extracted_wheel.exists(): extracted_wheel.mkdir(parents=True, exist_ok=True) with zipfile.ZipFile(WHEEL) as wheel: wheel.extractall(extracted_wheel) sys.path.insert(0, str(extracted_wheel)) from sibyl_memory_client import MemoryClient, NotFoundError # noqa: E402 RUN_ID = "run-2026-06-06-sibyl-365d-500c-50q-category-sonnet46" START_DATE = date(2026, 1, 5) COMPANY_COUNT = 500 TIMELINE_DAYS = 365 QUESTIONS_PER_CATEGORY = 50 STAKEHOLDERS_PER_COMPANY = 3 TENANT_ID = "365d-500c-50q-sibyl-benchmark" DEFAULT_CREDENTIALS = Path.home() / ".sibyl-memory" / "credentials.json" ANTHROPIC_ENV = ROOT / ".env.anthropic" ANTHROPIC_BASE_URL = os.environ.get("ANTHROPIC_BASE_URL", "https://api.anthropic.com") ANTHROPIC_MODEL = os.environ.get("ANTHROPIC_MODEL", "claude-sonnet-4-6") MAX_TOKENS = int(os.environ.get("SIBYL_365_BENCH_MAX_TOKENS", "800")) ROLES = ("account owner", "technical owner", "finance reviewer") TOPICS = ( "contract review", "data migration", "billing reconciliation", "API rollout", "support escalation", "weekly steering", "renewal planning", "risk review", "security approval", "implementation note", ) SEGMENTS = ( "logistics", "healthcare", "retail", "payments", "biotech", "construction", "insurance", "energy", "education", "manufacturing", ) REGIONS = ( "North America", "Western Europe", "APAC", "LATAM", "Middle East", ) MILESTONE_DAYS = (1, 30, 60, 90, 120, 180, 240, 300, 365) MARKER_DAYS = (1, 15, 30, 45, 60, 90, 120, 150, 180, 210, 240, 270, 300, 330, 365) PRICING_USD_PER_M = { "claude-sonnet-4-6": {"input_cache_miss": 3.0, "input_cache_hit": 0.30, "output": 15.0}, } def iso_now() -> str: return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") def load_env(path: Path) -> None: if not path.exists(): return for line in path.read_text(encoding="utf8").splitlines(): if not line or line.strip().startswith("#") or "=" not in line: continue key, value = line.split("=", 1) os.environ.setdefault(key.strip(), value.strip()) def write_json(path: Path, value: Any) -> None: path.write_text(json.dumps(value, indent=2, sort_keys=True, default=str) + "\n", encoding="utf8") def load_credentials(path: Path) -> dict[str, Any]: if not path.exists(): raise SystemExit(f"Sibyl credentials missing: {path}") credentials = json.loads(path.read_text(encoding="utf8")) required = ("account_id", "session_token", "signature", "tier") missing = [key for key in required if not credentials.get(key)] if missing: raise SystemExit(f"Sibyl credentials missing required fields: {', '.join(missing)}") claim = {key: value for key, value in credentials.items() if key != "signature"} return { "tier": str(credentials["tier"]), "account_id": str(credentials["account_id"]), "session_token": str(credentials.get("bearer_token") or credentials["session_token"]), "credentials_claim": claim, "credentials_signature": str(credentials["signature"]), } def slug(index: int) -> str: return f"scale365-company-{index:03d}" def display(index: int) -> str: return f"Scale365 Company {index:03d}" def company_key(index: int) -> str: return f"run365-{slug(index)}" def person_key(index: int, role_index: int) -> str: return f"run365-person-{index:03d}-{role_index:02d}" def relation_key(index: int, role_index: int) -> str: return f"run365-rel-{index:03d}-{role_index:02d}" def day_date(day_index: int) -> str: return (START_DATE + timedelta(days=day_index - 1)).isoformat() def topic(day_index: int) -> str: return TOPICS[(day_index - 1) % len(TOPICS)] def segment(index: int) -> str: return SEGMENTS[(index - 1) % len(SEGMENTS)] def region(index: int) -> str: return REGIONS[(index - 1) % len(REGIONS)] def marker(index: int, day_index: int) -> str: return f"scale365{slug(index).replace('-', '')}day{day_index:03d}" def approx_tokens(value: Any) -> int: return max(1, len(json.dumps(value, sort_keys=True, default=str)) // 4) def spread_indices(count: int, maximum: int, *, offset: int = 0) -> list[int]: if count <= 0 or maximum <= 0: return [] if count >= maximum: return list(range(1, maximum + 1)) values: list[int] = [] seen: set[int] = set() denominator = max(count - 1, 1) for position in range(count): value = 1 + round(position * (maximum - 1) / denominator) value = ((value + offset - 1) % maximum) + 1 if value not in seen: seen.add(value) values.append(value) candidate = 1 while len(values) < count: value = ((candidate + offset - 1) % maximum) + 1 if value not in seen: seen.add(value) values.append(value) candidate += 1 return values def result_row(row: dict[str, Any]) -> dict[str, Any]: return { "tier": row.get("tier"), "key": row.get("key") or row.get("name"), "category": row.get("category"), "body": row.get("body"), "snippet": row.get("snippet"), } def add_unique(rows: list[dict[str, Any]], row: dict[str, Any] | None) -> None: if row is None: return identity = (row.get("tier"), row.get("key"), row.get("category")) if identity not in {(item.get("tier"), item.get("key"), item.get("category")) for item in rows}: rows.append(row) def get_entity_safe(client: MemoryClient, category: str, name: str) -> dict[str, Any] | None: try: return result_row({"tier": "entity", **client.get_entity(category, name)}) except (NotFoundError, KeyError): return None def get_state_safe(client: MemoryClient, key: str) -> dict[str, Any] | None: state = client.get_state(key) if not state: return None return {"tier": "state", "key": key, "category": None, "body": state["body"], "snippet": None} def company_index_from_case(case: dict[str, Any]) -> int | None: text = " ".join(str(case.get(key, "")) for key in ("id", "query", "retrieval_query")) match = re.search(r"(?:company[_ ]|Company )(\d{3})", text) return int(match.group(1)) if match else None def day_from_case(case: dict[str, Any]) -> int | None: text = " ".join(str(case.get(key, "")) for key in ("id", "query", "retrieval_query")) match = re.search(r"(?:day[_ ]|day )(\d{1,3})", text) return int(match.group(1)) if match else None def role_index_from_case(case: dict[str, Any]) -> int | None: match = re.search(r"role_company_\d{3}_(\d)", str(case.get("id", ""))) if match: return int(match.group(1)) role_text = str(case.get("query", "")).lower() for index, role in enumerate(ROLES, start=1): if role in role_text: return index return None def table_counts(db_path: Path) -> dict[str, int]: with sqlite3.connect(db_path) as conn: return { "entities": conn.execute("SELECT COUNT(*) FROM entities").fetchone()[0], "state_documents": conn.execute("SELECT COUNT(*) FROM state_documents").fetchone()[0], "journal_events": conn.execute("SELECT COUNT(*) FROM journal_events").fetchone()[0], "reference_documents": conn.execute("SELECT COUNT(*) FROM reference_documents").fetchone()[0], } def ingest_dataset(client: MemoryClient, *, progress: bool) -> dict[str, Any]: started = time.perf_counter() writes = {"entities": 0, "state_sets": 0, "journal_events": 0, "failed": 0} for index in range(1, COMPANY_COUNT + 1): company_slug = slug(index) company_display = display(index) try: client.set_entity( "company", company_key(index), { "display_name": company_display, "segment": segment(index), "region": region(index), "timeline_start": START_DATE.isoformat(), "timeline_days": TIMELINE_DAYS, "stakeholder_count": STAKEHOLDERS_PER_COMPANY, "context_stats": { "simulated_days": TIMELINE_DAYS, "topic_cycle_size": len(TOPICS), "milestone_count": len(MILESTONE_DAYS), }, }, ) writes["entities"] += 1 for role_index, role in enumerate(ROLES, start=1): person = person_key(index, role_index) client.set_entity( "person", person, { "display_name": f"{company_display} Contact {role_index}", "role": role, "company": company_key(index), "company_display": company_display, }, ) client.set_entity( "relationship", relation_key(index, role_index), { "source": person, "target": company_key(index), "type": "works_on_account", "role": role, "description": f"{person} is the {role} for {company_display}.", }, ) writes["entities"] += 2 for day_index in range(1, TIMELINE_DAYS + 1): event_marker = marker(index, day_index) event_topic = topic(day_index) event_date = day_date(day_index) client.write_event( acted={ "kind": "scale365_chronology_daily_update", "body": { "company": company_display, "company_slug": company_slug, "day_index": day_index, "date": event_date, "topic": event_topic, "marker": event_marker, "summary": f"{company_display} day {day_index:03d}: {event_topic} update recorded.", }, }, extra={ "category": "chronology", "name": f"run365-{company_slug}-day-{day_index:03d}", "company": company_key(index), "marker": event_marker, }, ) writes["journal_events"] += 1 if day_index in MILESTONE_DAYS: client.set_entity( "timeline_milestone", f"run365-{company_slug}-day-{day_index:03d}", { "company": company_key(index), "company_display": company_display, "day_index": day_index, "date": event_date, "topic": event_topic, "marker": event_marker, "decision": f"{company_display} checkpoint day {day_index:03d} accepted for {event_topic}.", }, ) writes["entities"] += 1 if day_index % 10 == 0 or day_index == TIMELINE_DAYS: client.set_state( f"run365-current-status-{company_slug}", { "company": company_display, "last_day_index": day_index, "last_date": event_date, "last_marker": event_marker, "status": "active", }, ) writes["state_sets"] += 1 except Exception as exc: writes["failed"] += 1 print(f"[ingest] failed company={index:03d} error={type(exc).__name__}: {exc}", file=sys.stderr, flush=True) if progress and (index == 1 or index % 25 == 0): elapsed = time.perf_counter() - started print( f"[ingest] company {index}/{COMPANY_COUNT} elapsed={elapsed:.1f}s " f"events={writes['journal_events']} entities={writes['entities']}", file=sys.stderr, flush=True, ) writes["elapsed_seconds"] = round(time.perf_counter() - started, 3) writes["total_write_calls"] = writes["entities"] + writes["state_sets"] + writes["journal_events"] + writes["failed"] return writes def build_questions() -> list[dict[str, Any]]: cases: list[dict[str, Any]] = [] for index in spread_indices(QUESTIONS_PER_CATEGORY, COMPANY_COUNT, offset=0): cases.append( { "category": "status", "id": f"status_company_{index:03d}", "query": f"What is the current status for {display(index)} after the 365 day chronology?", "retrieval_query": f"{display(index)} current status day 365", "expected_contains": [display(index), str(TIMELINE_DAYS), day_date(TIMELINE_DAYS), "active"], } ) for position, index in enumerate(spread_indices(QUESTIONS_PER_CATEGORY, COMPANY_COUNT, offset=7)): day_index = MILESTONE_DAYS[position % len(MILESTONE_DAYS)] cases.append( { "category": "milestone", "id": f"milestone_company_{index:03d}_day_{day_index:03d}", "query": f"What was the {display(index)} day {day_index} milestone accepted for?", "retrieval_query": f"{display(index)} day {day_index} milestone", "expected_contains": [display(index), str(day_index), day_date(day_index), topic(day_index)], } ) for index in spread_indices(QUESTIONS_PER_CATEGORY, COMPANY_COUNT, offset=13): cases.append( { "category": "context_stat", "id": f"context_company_{index:03d}", "query": f"What segment and region context is stored for {display(index)}?", "retrieval_query": display(index), "expected_contains": [display(index), segment(index), region(index), str(TIMELINE_DAYS)], } ) for position, index in enumerate(spread_indices(QUESTIONS_PER_CATEGORY, COMPANY_COUNT, offset=19)): role_index = (position % len(ROLES)) + 1 cases.append( { "category": "role", "id": f"role_company_{index:03d}_{role_index}", "query": f"Who is the {ROLES[role_index - 1]} for {display(index)}?", "retrieval_query": f"{display(index)} {ROLES[role_index - 1]}", "expected_contains": [f"{display(index)} Contact {role_index}", ROLES[role_index - 1]], } ) for position, index in enumerate(spread_indices(QUESTIONS_PER_CATEGORY, COMPANY_COUNT, offset=29)): day_index = MARKER_DAYS[position % len(MARKER_DAYS)] cases.append( { "category": "marker", "id": f"marker_company_{index:03d}_day_{day_index:03d}", "query": f"What marker is associated with {display(index)} day {day_index}?", "retrieval_query": marker(index, day_index), "expected_contains": [marker(index, day_index)], } ) for position, index in enumerate(spread_indices(QUESTIONS_PER_CATEGORY, COMPANY_COUNT, offset=37)): day_index = 1 + ((position * 17) % TIMELINE_DAYS) cases.append( { "category": "temporal_topic", "id": f"topic_company_{index:03d}_day_{day_index:03d}", "query": f"What topic was recorded for {display(index)} on day {day_index}?", "retrieval_query": f"{display(index)} day {day_index} topic", "expected_contains": [display(index), str(day_index), day_date(day_index), topic(day_index)], } ) negative_names = [f"Phantom Prospect {index:03d}" for index in range(1, QUESTIONS_PER_CATEGORY + 1)] for name in negative_names: safe = re.sub(r"[^a-z0-9]+", "_", name.lower()).strip("_") cases.append( { "category": "negative_trap", "id": f"negative_{safe}", "query": f"What is the current status for {name} after the 365 day chronology?", "retrieval_query": f"{name} current status day 365", "expected_unknown": True, "forbidden_contains": ["active", str(TIMELINE_DAYS), day_date(TIMELINE_DAYS)], } ) return cases def hybrid_context(client: MemoryClient, case: dict[str, Any]) -> list[dict[str, Any]]: rows: list[dict[str, Any]] = [] index = company_index_from_case(case) if index is None: for row in client.search(str(case.get("retrieval_query", "")), limit=8): add_unique(rows, result_row(row)) return rows company_slug = slug(index) case_id = str(case["id"]) if case_id.startswith("status_company"): add_unique(rows, get_state_safe(client, f"run365-current-status-{company_slug}")) add_unique(rows, get_entity_safe(client, "company", company_key(index))) elif case_id.startswith("context_company"): add_unique(rows, get_entity_safe(client, "company", company_key(index))) add_unique(rows, get_state_safe(client, f"run365-current-status-{company_slug}")) elif case_id.startswith("role_company"): role_index = role_index_from_case(case) if role_index is not None: add_unique(rows, get_entity_safe(client, "person", person_key(index, role_index))) add_unique(rows, get_entity_safe(client, "relationship", relation_key(index, role_index))) add_unique(rows, get_entity_safe(client, "company", company_key(index))) elif case_id.startswith("milestone_company"): day_index = day_from_case(case) if day_index is not None: add_unique(rows, get_entity_safe(client, "timeline_milestone", f"run365-{company_slug}-day-{day_index:03d}")) for row in client.search(marker(index, day_index), limit=4, tiers=("journal",)): add_unique(rows, result_row(row)) add_unique(rows, get_entity_safe(client, "company", company_key(index))) elif case_id.startswith("marker_company") or case_id.startswith("topic_company"): day_index = day_from_case(case) if day_index is not None: add_unique(rows, get_entity_safe(client, "timeline_milestone", f"run365-{company_slug}-day-{day_index:03d}")) for row in client.search(marker(index, day_index), limit=5, tiers=("entity", "journal", "state")): add_unique(rows, result_row(row)) add_unique(rows, get_entity_safe(client, "company", company_key(index))) else: for row in client.search(str(case.get("retrieval_query", "")), limit=8): add_unique(rows, result_row(row)) return rows def score_context(case: dict[str, Any], context: list[dict[str, Any]]) -> dict[str, Any]: text = json.dumps(context, sort_keys=True, default=str).lower() expected = [str(item).lower() for item in case.get("expected_contains", [])] forbidden = [str(item).lower() for item in case.get("forbidden_contains", [])] if case.get("expected_unknown") is True: return {"passed": not any(item in text for item in forbidden), "contains_ok": True, "forbidden_ok": not any(item in text for item in forbidden)} contains_ok = all(item in text for item in expected) forbidden_ok = not any(item in text for item in forbidden) return {"passed": contains_ok and forbidden_ok, "contains_ok": contains_ok, "forbidden_ok": forbidden_ok} def build_prompt(question: str, context: list[dict[str, Any]]) -> tuple[str, str]: system = ( "You answer strictly from the provided Sibyl memory context. " "Answer in plain English in one or two short sentences. " "Include exact names, dates, day numbers, status, topic, segment, region, role, person, or marker when relevant. " "If the answer is not supported by the context, say exactly: " "I don't have that information in the provided memory." ) return system, json.dumps({"question": question, "context": context}, sort_keys=True, default=str) def ask_anthropic(question: str, context: list[dict[str, Any]]) -> dict[str, Any]: system, user_content = build_prompt(question, context) payload = { "model": ANTHROPIC_MODEL, "system": system, "messages": [{"role": "user", "content": user_content}], "temperature": 0, "max_tokens": MAX_TOKENS, } request = urllib.request.Request( f"{ANTHROPIC_BASE_URL.rstrip('/')}/v1/messages", data=json.dumps(payload).encode("utf8"), headers={ "x-api-key": os.environ["ANTHROPIC_API_KEY"], "anthropic-version": "2023-06-01", "Content-Type": "application/json", }, method="POST", ) with urllib.request.urlopen(request, timeout=120) as response: raw = json.loads(response.read().decode("utf8")) content = "".join(part.get("text", "") for part in raw.get("content", []) if isinstance(part, dict)) return {"response": {"answer": content, "unknown": "i don't have that information" in content.lower()}, "usage": raw.get("usage", {})} def estimate_cost(usage: dict[str, Any]) -> dict[str, Any]: pricing = PRICING_USD_PER_M[ANTHROPIC_MODEL] prompt_tokens = int(usage.get("prompt_tokens") or usage.get("input_tokens") or 0) completion_tokens = int(usage.get("completion_tokens") or usage.get("output_tokens") or 0) prompt_cache_hit = int(usage.get("prompt_cache_hit_tokens") or usage.get("cache_read_input_tokens") or 0) prompt_cache_miss = int(usage.get("prompt_cache_miss_tokens") or max(prompt_tokens - prompt_cache_hit, 0)) input_cost = (prompt_cache_hit / 1_000_000 * pricing["input_cache_hit"]) + ( prompt_cache_miss / 1_000_000 * pricing["input_cache_miss"] ) output_cost = completion_tokens / 1_000_000 * pricing["output"] return { "prompt_tokens": prompt_tokens, "completion_tokens": completion_tokens, "prompt_cache_hit_tokens": prompt_cache_hit, "prompt_cache_miss_tokens": prompt_cache_miss, "estimated_input_usd": input_cost, "estimated_output_usd": output_cost, "estimated_total_usd": input_cost + output_cost, } def add_cost(total: dict[str, Any], cost: dict[str, Any]) -> None: for key in total: total[key] += cost.get(key, 0) def score_answer(case: dict[str, Any], answer: dict[str, Any], context: list[dict[str, Any]]) -> dict[str, Any]: combined = json.dumps(answer, sort_keys=True, default=str).lower() expected = [str(item).lower() for item in case.get("expected_contains", [])] forbidden = [str(item).lower() for item in case.get("forbidden_contains", [])] contains_ok = all(item in combined for item in expected) forbidden_ok = not any(item in combined for item in forbidden) unknown_ok = True if case.get("expected_unknown") is True: unknown_ok = any( marker_text in combined for marker_text in ("i don't have that information", "not supported", "provided memory", "unknown", "no information") ) retrieval = score_context(case, context) return { "passed": contains_ok and forbidden_ok and unknown_ok, "contains_ok": contains_ok, "forbidden_ok": forbidden_ok, "unknown_ok": unknown_ok, "retrieval_has_expected": retrieval["passed"], } def summarize_questions(questions: list[dict[str, Any]], key: str) -> dict[str, Any]: summary: dict[str, Any] = {} for item in questions: category = item["category"] bucket = summary.setdefault(category, {"passed": 0, "total": 0}) bucket["total"] += 1 bucket["passed"] += int(bool(item[key]["passed"])) return summary def build_markdown(raw: dict[str, Any]) -> str: summary = raw["summary"] retrieval_rows = "\n".join( f"| {category} | {row['passed']} / {row['total']} |" for category, row in summary["retrieval_by_category"].items() ) answer_rows = "\n".join( f"| {category} | {row['passed']} / {row['total']} |" for category, row in summary.get("answer_by_category", {}).items() ) failures = [ { "id": item["id"], "category": item["category"], "query": item["query"], "expected_contains": item.get("expected_contains", []), "retrieval_score": item["retrieval_score"], "answer_score": item.get("answer_score"), "context_sample": item["context"][:2], } for item in raw["questions"] if not item["retrieval_score"]["passed"] or (item.get("answer_score") and not item["answer_score"]["passed"]) ] answer_summary = "Skipped" if raw["answer_mode"] == "sonnet": answer_summary = f"""{summary["answer_passed"]} / {summary["question_total"]} | Category | Sonnet answer passed | | --- | ---: | {answer_rows} """ return f"""# Sibyl 365d 500c Category Benchmark Run: `{raw["run_id"]}` ## Scope - System: Sibyl local memory, sync tier. - Corpus: {raw["dataset"]["company_count"]} companies, {raw["dataset"]["stakeholder_count"]} stakeholders, {raw["dataset"]["timeline_days"]} simulated days. - Stored records: {raw["dataset"]["record_count"]} final records. - Write calls: {raw["ingest"]["total_write_calls"]}. - Questions: {summary["question_total"]}, {raw["dataset"]["questions_per_category"]} per category. - Categories: {", ".join(summary["retrieval_by_category"].keys())}. - Retrieval strategy: app-side hybrid exact fallback by question intent. - Answer mode: `{raw["answer_mode"]}`. ## Summary | Metric | Value | | --- | ---: | | Retrieval passed | {summary["retrieval_passed"]} / {summary["question_total"]} | | Avg context rows | {summary["avg_context_rows"]:.2f} | | Avg context tokens | {summary["avg_context_tokens"]:.2f} | | DB size bytes | {raw["dataset"]["db_size_bytes"]} | | Ingest elapsed seconds | {raw["ingest"]["elapsed_seconds"]} | | Sonnet answers passed | {answer_summary if raw["answer_mode"] != "sonnet" else f'{summary["answer_passed"]} / {summary["question_total"]}'} | | Sonnet estimated cost USD | {summary.get("estimated_cost", {}).get("estimated_total_usd", 0):.6f} | ## Retrieval By Category | Category | Retrieval passed | | --- | ---: | {retrieval_rows} ## Answer By Category {answer_summary} ## Failures ```json {json.dumps(failures[:25], indent=2, sort_keys=True, default=str)} ``` """ def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Run Sibyl 365d/500-company category benchmark.") parser.add_argument("--run-id", default=RUN_ID) parser.add_argument("--data-root", default=f"/tmp/{RUN_ID}") parser.add_argument("--skip-ingest", action="store_true") parser.add_argument("--skip-llm", action="store_true") parser.add_argument("--question-limit", type=int, default=None) parser.add_argument("--credentials", default=str(DEFAULT_CREDENTIALS)) parser.add_argument("--progress", action="store_true") return parser.parse_args() def main() -> None: args = parse_args() load_env(ANTHROPIC_ENV) data_root = Path(args.data_root) data_root.mkdir(parents=True, exist_ok=True) db_path = data_root / "memory.db" credentials = load_credentials(Path(args.credentials)) client = MemoryClient.local(db_path, tenant_id=TENANT_ID, **credentials) raw: dict[str, Any] = { "run_id": args.run_id, "ts_started": iso_now(), "source_db": str(db_path), "tenant_id": TENANT_ID, "credentials_source": str(Path(args.credentials)), "sibyl_tier": credentials["tier"], "model": ANTHROPIC_MODEL if not args.skip_llm else None, "max_tokens": MAX_TOKENS, } if args.skip_ingest: ingest = {"skipped": True, "total_write_calls": None, "elapsed_seconds": 0} else: ingest = ingest_dataset(client, progress=args.progress) if ingest.get("failed"): raise SystemExit(f"Ingestion failed for {ingest['failed']} companies; refusing to score an incomplete corpus.") questions = build_questions() if args.question_limit is not None: questions = questions[: args.question_limit] question_rows: list[dict[str, Any]] = [] total_cost = { "prompt_tokens": 0, "completion_tokens": 0, "prompt_cache_hit_tokens": 0, "prompt_cache_miss_tokens": 0, "estimated_input_usd": 0.0, "estimated_output_usd": 0.0, "estimated_total_usd": 0.0, } answer_mode = "skipped" if not args.skip_llm: if not os.environ.get("ANTHROPIC_API_KEY"): raise SystemExit("ANTHROPIC_API_KEY missing in environment or .env.anthropic") answer_mode = "sonnet" started_questions = time.perf_counter() for position, case in enumerate(questions, start=1): context = hybrid_context(client, case) retrieval_score = score_context(case, context) row = { **case, "context_count": len(context), "context_tokens_estimate": approx_tokens(context), "context": context, "retrieval_score": retrieval_score, } if answer_mode == "sonnet": llm = ask_anthropic(case["query"], context) cost = estimate_cost(llm["usage"]) add_cost(total_cost, cost) row["answer"] = llm["response"] row["usage"] = llm["usage"] row["cost"] = cost row["answer_score"] = score_answer(case, llm["response"], context) question_rows.append(row) if args.progress and (position == 1 or position % 25 == 0): print(f"[questions] {position}/{len(questions)} elapsed={time.perf_counter() - started_questions:.1f}s", file=sys.stderr, flush=True) counts = table_counts(db_path) record_count = sum(counts.values()) raw["ts_completed"] = iso_now() raw["ingest"] = ingest raw["dataset"] = { "company_count": COMPANY_COUNT, "stakeholder_count": COMPANY_COUNT * STAKEHOLDERS_PER_COMPANY, "timeline_days": TIMELINE_DAYS, "questions_per_category": QUESTIONS_PER_CATEGORY, "table_counts": counts, "record_count": record_count, "db_size_bytes": db_path.stat().st_size if db_path.exists() else None, } raw["answer_mode"] = answer_mode raw["questions"] = question_rows retrieval_passed = sum(1 for item in question_rows if item["retrieval_score"]["passed"]) summary = { "question_total": len(question_rows), "retrieval_passed": retrieval_passed, "retrieval_failed": len(question_rows) - retrieval_passed, "retrieval_by_category": summarize_questions(question_rows, "retrieval_score"), "avg_context_rows": sum(item["context_count"] for item in question_rows) / len(question_rows), "avg_context_tokens": sum(item["context_tokens_estimate"] for item in question_rows) / len(question_rows), } if answer_mode == "sonnet": answer_passed = sum(1 for item in question_rows if item["answer_score"]["passed"]) summary.update( { "answer_passed": answer_passed, "answer_failed": len(question_rows) - answer_passed, "answer_by_category": summarize_questions(question_rows, "answer_score"), "estimated_cost": total_cost, } ) raw["summary"] = summary raw["status"] = "PASS" if retrieval_passed == len(question_rows) and (answer_mode != "sonnet" or summary["answer_passed"] == len(question_rows)) else "FAIL" RUNS_DIR.mkdir(parents=True, exist_ok=True) raw_path = RUNS_DIR / f"{args.run_id}.raw_result.json" md_path = RUNS_DIR / f"{args.run_id}.md" write_json(raw_path, raw) md_path.write_text(build_markdown(raw), encoding="utf8") print(json.dumps({"status": raw["status"], "summary": summary, "raw": str(raw_path), "md": str(md_path)}, indent=2)) if __name__ == "__main__": main()