scripts.evaluate_semantic_graph_qa

Reference › scripts › evaluate_semantic_graph_qa

Evaluates semantic-graph retrieval against a benchmark of 100 questions. A human runs it; no test imports it. It loads the environment through env_bootstrap and writes its report atomically with atomic_json. The graph is described in the semantic_graph narrative; enrich_semantic_graph_specs loads the data it queries.

Invocation: main is asynchronous and takes no options. It exits 1 if the questions file is missing. _parse_questions reads the questions and their expected keywords. _evaluate asks the semantic graph each question, scores how much of the expected keywords the answer covers and returns the per-question results, the average coverage and the number of low-quality answers. _write_report writes the report, and main prints the average keyword coverage, the low-quality count and the report path.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/evaluate_semantic_graph_qa.py

"""Evaluate semantic graph retrieval against the 100-question benchmark.

Reads questions from ``plans/semantic_graph_questions.txt``, retrieves the top-3
chunks for each question using the same hybrid pipeline as the ingestion script,
and reports keyword coverage. Results are written to
``plans/semantic_graph_qa_results_latest.md``.
"""

from __future__ import annotations

import asyncio
import os
import re
import sys
from pathlib import Path
from typing import Any, cast

from orchestrata_ultimate.core.env_bootstrap import load_env_contract
from orchestrata_ultimate.semantic_graph.adapters.memgraph import (
    MemgraphSemanticGraph,
)
from orchestrata_ultimate.semantic_graph.config import SemanticGraphConfig
from orchestrata_ultimate.semantic_graph.embedder_factory import (
    chunk_label,
    create_embedder,
    embed_model,
    vector_index_name,
)
from orchestrata_ultimate.semantic_graph.ingestion import (
    CodeChunker,
    SemanticGraphIngester,
)
from orchestrata_ultimate.semantic_graph.rerankers import OpenRouterReranker

PROJECT_ROOT = Path(__file__).resolve().parent.parent
QUESTIONS_PATH = PROJECT_ROOT / "plans" / "semantic_graph_questions.txt"
OUTPUT_PATH = PROJECT_ROOT / "plans" / "semantic_graph_qa_results_latest.md"

PROJECT_ID = "orchestrata-ultimate"
RERANK_MODEL = "cohere/rerank-4-pro"


def _load_dotenv() -> None:
    """Load the RUL-019 two-phase environment contract.

    Delegates to ``core.env_bootstrap.load_env_contract`` — THE canonical
    loader. What stood here loaded ``.env`` and stopped, and ``.env`` is the
    prod-BASE layer: it carries ``CXP_GRPC_HOST=localhost`` for the host
    user-units, so a script that stops after phase 1 resolves PRODUCTION
    endpoints while believing it is on the test fleet.

    ``override=False`` was the right instinct aimed at the wrong half. It was
    added on 2026-07-26 after a bootstrap intended for the test node wrote 592
    chunks into the production cluster, because ``.env``'s value had replaced
    an explicitly-exported one. The canonical loader keeps that property for
    phase 1 (``.env`` only fills keys the environment has not already set) and
    then applies the ``ORCHESTRATA_ENV_PROFILE`` overlay, which is the half
    that actually points the process at the right infrastructure — and honours
    ``declare_owned`` so it cannot reset a test worker's per-slot isolation.
    """
    # Rooted at this module's PROJECT_ROOT rather than letting the loader
    # resolve the repo itself. The missing-key tests redirect PROJECT_ROOT
    # at a tmp_path precisely so this call cannot re-read the real .env and
    # put OPENROUTER_API_KEY back; a self-resolving loader walked past the
    # redirect, reloaded the key, and turned a no-key assertion into a live
    # billed request to OpenRouter.
    load_env_contract(start=PROJECT_ROOT)


def _parse_questions(path: Path) -> list[dict[str, str | list[str]]]:
    """Parse questions and their implied keyword groups from the text file."""
    text = path.read_text(encoding="utf-8")
    lines = [line.strip() for line in text.splitlines()]

    questions: list[dict[str, str | list[str]]] = []
    current_category = "General"

    for line in lines:
        if not line:
            continue
        if line.endswith("?"):
            match = re.match(r"^(\d+)\.\s+(.*)$", line)
            if match:
                number = match.group(1)
                question = match.group(2)
                # Simple keyword extraction: content words longer than 3 chars.
                words = [
                    w.lower()
                    for w in re.findall(r"[A-Za-z0-9_]+", question)
                    if len(w) > 3
                    and w.lower()
                    not in {
                        "what",
                        "does",
                        "how",
                        "why",
                        "when",
                        "where",
                        "which",
                        "must",
                        "this",
                        "that",
                        "with",
                        "from",
                        "does",
                        "have",
                        "has",
                        "been",
                        "than",
                        "only",
                        "each",
                        "they",
                        "them",
                        "their",
                        "there",
                        "would",
                        "should",
                        "could",
                    }
                ]
                questions.append(
                    {
                        "number": number,
                        "category": current_category,
                        "question": question,
                        "keywords": list(dict.fromkeys(words))[:8],
                    }
                )
        elif not line.endswith("?") and line and not line[0].isdigit():
            current_category = line

    return questions


async def _evaluate() -> tuple[list[dict[str, object]], float, int]:
    _load_dotenv()

    api_key = os.environ.get("OPENROUTER_API_KEY", "")
    if not api_key:
        raise SystemExit("OPENROUTER_API_KEY is not set")

    config = SemanticGraphConfig.from_env()
    # Label and index follow the SELECTED backend: they are what say "these
    # vectors are comparable to each other", and this graph is shared with the
    # knowledge-wiki. Whatever wrote the index has to be what queries it.
    config.chunk_label = chunk_label()
    config.vector_index_name = vector_index_name()

    # OPENROUTER_EMBED_BASE_URL/OPENROUTER_RERANK_BASE_URL: read at call time,
    # defaulting to the real OpenRouter API (the classes' own defaults) --
    # lets tests redirect these two HTTP round trips at local test servers
    # without touching the real API (policy ruling 2026-08-04: tests make no
    # LLM/embedding network calls).
    # The backend is a choice, not a constant (semantic_graph.embedder_factory):
    # dev/test embed locally for free, prod keeps codestral via OpenRouter. The
    # model and dimension come from the same place so an index is always
    # queried by whatever wrote it.
    embedder = create_embedder(
        api_key=api_key, batch_size=64, max_tokens_per_batch=8_000
    )
    reranker = OpenRouterReranker(
        api_key=api_key,
        model=RERANK_MODEL,
        base_url=os.environ.get(
            "OPENROUTER_RERANK_BASE_URL", "https://openrouter.ai/api/v1"
        ),
    )
    graph = MemgraphSemanticGraph(config)

    connected = await graph.connect()
    if not connected:
        raise SystemExit("Failed to connect to Memgraph")

    ingester = SemanticGraphIngester(
        graph=graph,
        embedder=embedder,
        chunker=CodeChunker(lines_per_chunk=50, overlap_lines=5),
        project_id=PROJECT_ID,
        batch_size=64,
        reranker=reranker,
        rerank_candidates=20,
    )

    questions = _parse_questions(QUESTIONS_PATH)
    results: list[dict[str, object]] = []
    low_count = 0
    total_coverage = 0.0

    try:
        for item in questions:
            chunks, _ = await ingester.search(str(item["question"]), top_k=3)
            keywords = [kw.lower() for kw in item["keywords"]]
            found = 0
            for kw in keywords:
                for chunk in chunks:
                    if kw in chunk.content.lower():
                        found += 1
                        break
            coverage = (found / len(keywords) * 100) if keywords else 0.0
            total_coverage += coverage
            if coverage < 50:
                low_count += 1

            results.append(
                {
                    "number": item["number"],
                    "category": item["category"],
                    "question": item["question"],
                    "keywords": keywords,
                    "coverage": coverage,
                    "chunks": [
                        {
                            "path": chunk.metadata.get("file_path", "unknown"),
                            "start": chunk.metadata.get("start_line", 0),
                            "end": chunk.metadata.get("end_line", 0),
                            "snippet": chunk.content[:400].replace("\n", " "),
                        }
                        for chunk in chunks
                    ],
                }
            )
    finally:
        await graph.close()

    average = total_coverage / len(results) if results else 0.0
    return results, average, low_count


def _write_report(
    results: list[dict[str, object]], average: float, low_count: int
) -> None:
    lines: list[str] = [
        "# Semantic Graph Q&A Results (Latest: Hybrid Vector + Keyword + Module Graph)",
        "",
        f"Model: {embed_model()} | Reranker: {RERANK_MODEL} | Project: {PROJECT_ID}",
        "",
        f"**Average keyword coverage:** {average:.0f}%",
        f"**Low-quality answers (<50%):** {low_count}/{len(results)}",
        "",
    ]

    for r in results:
        lines.append(f"## Q{r['number']}: {r['question']}")
        lines.append(f"**Category:** {r['category']}")
        lines.append(f"**Keywords:** {', '.join(cast('list[str]', r['keywords']))}")
        lines.append(f"**Coverage:** {r['coverage']:.0f}%")
        lines.append("")
        # `r` is a heterogeneous result mapping, so `r["chunks"]` widens to
        # object; cast at the boundary instead of suppressing per-line. The
        # `# type: ignore[index]` on the snippet lookup became unused once the
        # element type was known, which mypy reports as its own error.
        chunks = cast("list[dict[str, Any]]", r["chunks"])
        for idx, chunk in enumerate(chunks, start=1):
            snippet = chunk["snippet"]
            lines.append(
                f"{idx}. `{chunk['path']}:{chunk['start']}-{chunk['end']}` — {snippet}..."
            )
        lines.append("")

    from orchestrata_ultimate.core.atomic_json import dump_text_atomic

    dump_text_atomic(OUTPUT_PATH, "\n".join(lines))


async def main() -> int:
    if not QUESTIONS_PATH.exists():
        print(f"Questions file not found: {QUESTIONS_PATH}", file=sys.stderr)
        return 1

    results, average, low_count = await _evaluate()
    _write_report(results, average, low_count)

    print(f"Average keyword coverage: {average:.0f}%")
    print(f"Low-quality answers: {low_count}/{len(results)}")
    print(f"Report written to: {OUTPUT_PATH}")
    return 0


if __name__ == "__main__":
    sys.exit(asyncio.run(main()))