orchestrata_ultimate.verification.agents.verification_agent

Reference › orchestrata_ultimate › verification › agents › verification_agent

VerificationAgent (default id U:Vfy) is the deciding role: it checks constraints with Prolog, optionally analyses the results with an LLM, and makes the final call (Spec 035 US-005, US-007, US-009). The constructor takes a ConstraintEvaluator and a ProofVerifier (it builds defaults when none are given) and an LLM model, either directly or looked up by name in a ModelRegistry.

The flow is check_via_prolog, which delegates to ConstraintEvaluator.evaluate_all; then analyze_via_llm, which returns the model's analysis, a "No LLM available" note when no model is set, or an error note when the model raises (the failure is logged, not propagated); then decide, which counts satisfied results and returns the pass rate and an approved flag that is true only when every result is satisfied. validate_proof delegates to the proof verifier. Each decision is kept in memory and counted by decision_count. Models come from base_model.

At a glance

Class diagram

Dependency diagram

API

Classes

VerificationAgent

Decorators: traces(spec='035', story='US-005', reqs=['REQ-035-US-005-007', 'REQ-035-US-005-022', 'REQ-035-US-007-016', 'REQ-035-US-… · Line 48

Core verification agent.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/verification/agents/verification_agent.py

# mypy: ignore-errors
"""Verification agent (U:Vfy) — validates via Prolog + LLM.

Wiring status (2026-07-06, W6.2): the ``llm_model`` slot was previously
typed ``object | None`` (a pre-existing pyright error at the ``.analyze()``
call site below) with no connection to the ``LLMModel`` protocol or
``ModelRegistry`` it was built to consume. It is now typed against
``LLMModel`` directly, and an optional ``model_registry``/``model_name``
pair (REQ-035-US-005-005: select an appropriate model) resolves ``llm_model``
via registry lookup when no model is passed explicitly. Both new
constructor params default to ``None`` -- zero behavior change for
existing callers.
"""

import logging
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.verification.constraints.evaluator import (
    ConstraintEvaluator,
)
from orchestrata_ultimate.verification.models.base_model import (
    LLMModel,
    ModelRegistry,
)
from orchestrata_ultimate.verification.proofs.verifier import ProofVerifier
from orchestrata_ultimate.verification.types import (
    Constraint,
    Proof,
    ProofResult,
    VerificationResult,
)

logger = logging.getLogger(__name__)


@traces(
    spec="035",
    story="US-005",
    reqs=[
        "REQ-035-US-005-007",
        "REQ-035-US-005-022",
        "REQ-035-US-007-016",
        "REQ-035-US-007-020",
        "REQ-035-US-009-008",
    ],
)
class VerificationAgent:
    """Core verification agent.

    Checks constraints via Prolog, analyzes via LLM,
    and makes final decisions.
    """

    @traces(spec="035", story="US-005", reqs=["REQ-035-US-005-005"])
    def __init__(
        self,
        agent_id: str = "U:Vfy",
        evaluator: ConstraintEvaluator | None = None,
        proof_verifier: ProofVerifier | None = None,
        llm_model: LLMModel | None = None,
        model_registry: ModelRegistry | None = None,
        model_name: str | None = None,
    ) -> None:
        self._agent_id = agent_id
        self._evaluator = evaluator or ConstraintEvaluator()
        self._proof_verifier = proof_verifier or ProofVerifier()
        if llm_model is None and model_registry is not None and model_name is not None:
            llm_model = model_registry.get(model_name)
        self._llm = llm_model
        self._decisions: list[dict[str, Any]] = []

    @property
    def agent_id(self) -> str:
        return self._agent_id

    @traces(spec="035", story="US-009", reqs=["REQ-035-US-009-008"])
    def check_via_prolog(
        self, constraints: list[Constraint], context: dict[str, Any] | None = None
    ) -> list[VerificationResult]:
        """Check constraints via the Prolog-backed evaluator."""
        return self._evaluator.evaluate_all(constraints, context)

    @traces(spec="035", story="US-5", reqs=["REQ-035-US-005-005"])
    @traces(
        spec="035", story="US-005", reqs=["REQ-035-US-005-007", "REQ-035-US-005-022"]
    )
    async def analyze_via_llm(
        self, results: list[VerificationResult]
    ) -> dict[str, Any]:
        """Analyze results via LLM if available."""
        if not self._llm:
            return {"analysis": "No LLM available", "results_count": len(results)}
        try:
            return await self._llm.analyze([r.to_dict() for r in results])
        except Exception as exc:
            logger.warning("LLM analysis failed: %s", exc)
            return {"analysis": "LLM analysis failed", "error": str(exc)}

    def decide(
        self,
        prolog_results: list[VerificationResult],
        llm_analysis: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        """Make a verification decision."""
        satisfied = sum(1 for r in prolog_results if r.is_satisfied)
        total = len(prolog_results) if prolog_results else 1

        decision = {
            "agent_id": self._agent_id,
            "satisfied": satisfied,
            "total": total,
            "pass_rate": satisfied / total,
            "approved": satisfied == total,
            "llm_analysis": llm_analysis,
        }
        self._decisions.append(decision)
        return decision

    @traces(
        spec="035", story="US-007", reqs=["REQ-035-US-007-016", "REQ-035-US-007-020"]
    )
    def validate_proof(self, proof: Proof) -> ProofResult:
        """Validate a submitted proof."""
        return self._proof_verifier.verify(proof)

    @property
    def decision_count(self) -> int:
        return len(self._decisions)