orchestrata_ultimate.knowledge_graph.graph_context_validator

Reference › orchestrata_ultimate › knowledge_graph › graph_context_validator

GraphContextValidator is the programmatic check that an LLM response actually uses the graph context injected into its prompt (Spec 008 REQ-008-013 and REQ-008-056; Spec 016 REQ-016-038 and -039). It is constructed with the formatted context text, a compliance threshold (COMPLIANCE_THRESHOLD, documented as 50%) and an optional list of core entities. _extract_entities pulls class and function names out of the context and splits them into public names and private ones prefixed with an underscore. validate returns a ValidationResult and passes when there are no entities at all, or when at least the threshold fraction of public entities is referenced (word-boundary match via _check_references) and, with require_citations true, an explicit Entities used section, parsed by _extract_citations, lists every core entity (all public entities by default). Private entities are reported but do not affect the verdict. The result records found and missing entities, counts, the citations, invalid citations (cited names not in the context) and missing core citations, with the properties compliance_ratio, entities_found and entities_missing. context_validator_loop retries the call when validation fails. The validator reads no database.

At a glance

Class diagram

Dependency diagram

API

Classes

ValidationResult (dataclass)

Decorators: dataclass, traces(spec='008', story='US-003', reqs=['REQ-008-013']), traces(spec='016', story='US-003', reqs=['REQ-016-038']) · Line 35

Result of graph context validation against an LLM response.

GraphContextValidator

Decorators: traces(spec='008', story='US-003', reqs=['REQ-008-013', 'REQ-008-056']), traces(spec='016', story='US-003', reqs=['REQ-016-039']) · Line 87

Validates that LLM responses reference injected graph entities.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/knowledge_graph/graph_context_validator.py

"""Post-LLM response validator for graph context compliance.

Programmatic enforcement — not a prompt suggestion. After the LLM
responds, this validator checks that the response actually references
PUBLIC entities from the injected graph context. Private implementation
details (prefixed with ``_``) are tracked but not required.

Compliance threshold: 50% of public entities must be referenced.
If fewer are referenced, the response is REJECTED and retried
with explicit entity requirements.

Usage::

    validator = GraphContextValidator(graph_context)
    result = validator.validate(response_text)
    if not result.passed:
        retry_prompt = validator.build_retry_prompt(original, result)
"""

import logging
import re
from dataclasses import dataclass, field

from orchestrata_ultimate.core.base.traces import traces

_logger = logging.getLogger(__name__)

MAX_VALIDATION_RETRIES = 2
COMPLIANCE_THRESHOLD = 1.0  # 100% of public entities must be referenced


@dataclass
@traces(spec="008", story="US-003", reqs=["REQ-008-013"])
@traces(spec="016", story="US-003", reqs=["REQ-016-038"])
class ValidationResult:
    """Result of graph context validation against an LLM response."""

    passed: bool = False
    public_found: list[str] = field(default_factory=list)
    public_missing: list[str] = field(default_factory=list)
    private_found: list[str] = field(default_factory=list)
    private_missing: list[str] = field(default_factory=list)
    public_count: int = 0
    public_referenced: int = 0
    total_count: int = 0
    total_referenced: int = 0
    # Citation-aware fields (SPEC-106/Phase 0): explicit "Entities used" block.
    citations: list[str] = field(default_factory=list)
    invalid_citations: list[str] = field(default_factory=list)
    missing_core_citations: list[str] = field(default_factory=list)

    @property
    @traces(spec="028", reqs=["REQ-028-006"])
    def compliance_ratio(self) -> float:
        """Fraction of PUBLIC entities referenced in the response."""
        if self.public_count == 0:
            return 1.0  # No public entities = vacuously compliant
        return self.public_referenced / self.public_count

    @property
    @traces(spec="028", reqs=["REQ-028-006"])
    def entities_found(self) -> list[str]:
        """All referenced entities (public + private)."""
        return self.public_found + self.private_found

    @property
    @traces(spec="028", reqs=["REQ-028-006"])
    def entities_missing(self) -> list[str]:
        """All unreferenced entities (public + private)."""
        return self.public_missing + self.private_missing

    @property
    @traces(spec="028", reqs=["REQ-028-006"])
    def entity_count(self) -> int:
        """Total entity count (public + private)."""
        return self.total_count

    @property
    @traces(spec="028", reqs=["REQ-028-006"])
    def reference_count(self) -> int:
        """Total references found (public + private)."""
        return self.total_referenced


@traces(spec="008", story="US-003", reqs=["REQ-008-013", "REQ-008-056"])
@traces(spec="016", story="US-003", reqs=["REQ-016-039"])
class GraphContextValidator:
    """Validates that LLM responses reference injected graph entities.

    Separates entities into PUBLIC (classes, public functions) and
    PRIVATE (``_``-prefixed methods, internal helpers). Only public
    entities count toward the compliance threshold.

    Compliance rule: at least 50% of public entities must appear
    in the response as word-boundary matches. Private entities
    contribute to the score but are not required.

    Phase 0 citation awareness: responses are expected to contain an
    explicit "Entities used" section listing every class/function/entity
    they acted on. The validator parses that section, rejects citations to
    entities that were not injected, and (when ``require_citations`` is
    enabled) rejects responses that omit injected core entities from the
    citation list. Core defaults to all public entities; callers with
    tiered context can supply a narrower ``core_entities`` list.
    """

    def __init__(
        self,
        graph_context: str | None,
        threshold: float = COMPLIANCE_THRESHOLD,
        core_entities: list[str] | None = None,
    ) -> None:
        self._graph_context = graph_context or ""
        self._threshold = threshold
        all_entities = self._extract_entities(self._graph_context)
        self._public = [e for e in all_entities if not e.startswith("_")]
        self._private = [e for e in all_entities if e.startswith("_")]
        self._core = core_entities if core_entities is not None else list(self._public)
        self._all_entity_names = set(self._public + self._private)

    @staticmethod
    def _extract_entities(graph_context: str) -> list[str]:
        """Extract class and function names from formatted graph context.

        Parses the structured output from ContextInjector.to_prompt():
        - ``class ClassName(...): methods`` → ClassName
        - ``def function_name(...)`` → function_name
        - ``ClassName inherits from BaseName`` → ClassName, BaseName
        - ``caller() calls callee()`` → caller, callee
        - ``ClassName.method()`` → ClassName, method

        Returns:
            Deduplicated sorted list of entity names.
        """
        if not graph_context:
            return []

        names: set[str] = set()

        # class ClassName or class ClassName(Base):
        names.update(m.group(1) for m in re.finditer(r"class\s+(\w+)", graph_context))

        # def function_name(
        names.update(
            m.group(1) for m in re.finditer(r"def\s+(\w+)\s*\(", graph_context)
        )

        # X inherits from Y
        names.update(
            n
            for m in re.finditer(r"(\w+)\s+inherits\s+from\s+(\w+)", graph_context)
            for n in (m.group(1), m.group(2))
        )

        # X() calls Y()
        names.update(
            n
            for m in re.finditer(r"(\w+)\(\)\s+calls\s+(\w+)\(\)", graph_context)
            for n in (m.group(1), m.group(2))
        )

        # ClassName.method()
        names.update(
            n
            for m in re.finditer(r"(\w+)\.(\w+)\(\)", graph_context)
            for n in (m.group(1), m.group(2))
        )

        # Requirement ids (REQ-NNN-..., REQ-NNN-USn-NNN, etc.)
        names.update(
            m.group(1)
            for m in re.finditer(r"\b(REQ-[A-Z0-9]+(?:-[A-Z0-9]+)+)\b", graph_context)
        )

        # Decision ids: lines under "## Prior decisions..." are "  <id>: <rationale>"
        in_decisions = False
        for line in graph_context.splitlines():
            if "## Prior decisions" in line:
                in_decisions = True
                continue
            if in_decisions:
                if line.startswith("## "):
                    break
                decision_match = re.match(r"^\s+([A-Za-z0-9:_-]+):\s*", line)
                if decision_match:
                    names.add(decision_match.group(1))

        # Filter out common noise words
        noise = {
            "self",
            "cls",
            "None",
            "True",
            "False",
            "str",
            "int",
            "float",
            "bool",
            "list",
            "dict",
            "set",
            "tuple",
            "Any",
            "Optional",
            "def",
            "class",
            "from",
            "import",
            "return",
            "pass",
        }
        return sorted(names - noise)

    def _check_references(
        self, entities: list[str], response: str
    ) -> tuple[list[str], list[str]]:
        """Check which entities are referenced in the response.

        Returns:
            (found, missing) lists.
        """
        found: list[str] = []
        missing: list[str] = []
        for entity in entities:
            pattern = r"\b" + re.escape(entity) + r"\b"
            if re.search(pattern, response):
                found.append(entity)
            else:
                missing.append(entity)
        return found, missing

    def _extract_citations(self, response: str) -> list[str]:
        """Parse an explicit ``Entities used`` citation section from response.

        Accepts several markdown heading styles:
        - ``## Entities used``
        - ``### Entities used``
        - ``Entities used:``

        Bulleted items, numbered items, comma-separated items, and plain
        lines are all accepted. Returns a deduplicated, stripped list of
        cited entity names.
        """
        if not response:
            return []

        # Match a heading/label line, then capture the following block until
        # the next markdown heading or end of text.
        match = re.search(
            r"(?:^|\n)\s*#+\s*Entities used\s*(?::\s*)?\n(.*?)(?=\n\s*#+\s|\Z)",
            response,
            re.IGNORECASE | re.DOTALL,
        )
        if not match:
            # Fallback: look for an inline "Entities used:" label.
            match = re.search(
                r"(?:^|\n)\s*Entities used\s*:\s*\n?(.*?)(?=\n\s*\n|\Z)",
                response,
                re.IGNORECASE | re.DOTALL,
            )
        if not match:
            return []

        block = match.group(1)
        citations: set[str] = set()
        for raw_line in block.splitlines():
            line = raw_line.strip()
            if not line:
                continue
            # Strip leading bullets/numbers: "- Foo", "1. Foo", "* Foo"
            line = re.sub(r"^[-*\d.)\]]+\s*", "", line)
            # Split comma-separated entries
            for part in line.split(","):
                part = part.strip().strip("`'\"()")
                if part:
                    citations.add(part)
        return sorted(citations)

    def validate(
        self, response: str, require_citations: bool = True
    ) -> ValidationResult:
        """Check if the LLM response references enough public entities.

        Compliance passes when:
        - No public entities exist (vacuously compliant), OR
        - At least ``threshold`` fraction of public entities are
          referenced in the response (word-boundary match), AND
        - When ``require_citations`` is True, an explicit ``Entities used``
          section lists every core entity (defaults to all public entities).

        Private entities are checked for reporting but do not
        affect pass/fail.

        Args:
            response: The LLM response text.
            require_citations: If True, enforce an explicit citation section
                covering all core entities. Defaults to True.

        Returns:
            ValidationResult with pass/fail, entity details, and citation
            details.
        """
        if not self._public and not self._private:
            return ValidationResult(passed=True)

        pub_found, pub_missing = self._check_references(self._public, response)
        priv_found, priv_missing = self._check_references(self._private, response)

        public_count = len(self._public)
        public_referenced = len(pub_found)

        if public_count == 0:
            reference_passed = True
        else:
            reference_passed = (public_referenced / public_count) >= self._threshold

        citations: list[str] = []
        invalid_citations: list[str] = []
        missing_core_citations: list[str] = []
        citation_passed = True

        if require_citations and self._core:
            citations = self._extract_citations(response)
            cited_set = set(citations)
            invalid_citations = sorted(cited_set - self._all_entity_names)
            missing_core_citations = sorted(set(self._core) - cited_set)
            citation_passed = not invalid_citations and not missing_core_citations

        return ValidationResult(
            passed=reference_passed and citation_passed,
            public_found=pub_found,
            public_missing=pub_missing,
            private_found=priv_found,
            private_missing=priv_missing,
            public_count=public_count,
            public_referenced=public_referenced,
            total_count=public_count + len(self._private),
            total_referenced=public_referenced + len(priv_found),
            citations=citations,
            invalid_citations=invalid_citations,
            missing_core_citations=missing_core_citations,
        )

    def build_retry_prompt(
        self,
        original_prompt: str,
        validation: ValidationResult,
    ) -> str:
        """Build a retry prompt listing the PUBLIC entities that must be referenced.

        Only lists public missing entities — private internals are not
        forced. The directive is prepended to the original prompt.

        Args:
            original_prompt: The original user prompt.
            validation: The failed validation result.

        Returns:
            Modified prompt with mandatory entity references.
        """
        lines: list[str] = []
        ratio_pct = f"{validation.compliance_ratio:.0%}"
        threshold_pct = f"{self._threshold:.0%}"
        lines.append(
            f"YOUR PREVIOUS RESPONSE WAS REJECTED — compliance {ratio_pct} "
            f"(required: {threshold_pct})."
        )

        if validation.missing_core_citations or validation.invalid_citations:
            lines.append("")
            lines.append(
                "Your response MUST end with an explicit '## Entities used' "
                "section that lists every class/function/entity you acted on."
            )
            if validation.missing_core_citations:
                lines.append("")
                lines.append("Add these missing citations:")
                lines.extend(f"  - {e}" for e in validation.missing_core_citations)
            if validation.invalid_citations:
                lines.append("")
                lines.append(
                    "Remove these citations — they were not in the injected context:"
                )
                lines.extend(f"  - {e}" for e in validation.invalid_citations)

        if validation.public_missing:
            lines.append("")
            lines.append(
                "You MUST reference these existing classes and functions in your "
                "response. Use them, import them, call them, or inherit from them "
                "— do NOT create duplicates:"
            )
            lines.extend(f"  - {e}" for e in validation.public_missing)

        lines.append("")
        lines.append(
            "If an entity is genuinely irrelevant to this task, explicitly "
            "state WHY it is not applicable. Do not silently ignore it."
        )
        lines.append("")
        lines.append("---")
        lines.append("")
        return "\n".join(lines) + original_prompt

    @property
    @traces(spec="028", reqs=["REQ-028-006", "REQ-028-049"])
    def has_entities(self) -> bool:
        """True if graph context contains extractable public entities."""
        return len(self._public) > 0

    @property
    @traces(spec="028", reqs=["REQ-028-006"])
    def entity_names(self) -> list[str]:
        """List of all extracted entity names (public + private)."""
        return list(self._public + self._private)

    @property
    @traces(spec="028", reqs=["REQ-028-006"])
    def public_entity_names(self) -> list[str]:
        """List of public entity names (enforcement targets)."""
        return list(self._public)

    @property
    @traces(spec="028", reqs=["REQ-028-006"])
    def private_entity_names(self) -> list[str]:
        """List of private entity names (tracked, not enforced)."""
        return list(self._private)