orchestrata_ultimate.orchestrator.ambiguity_resolver

Reference › orchestrata_ultimate › orchestrator › ambiguity_resolver

Decides between competing readings of the same intent. No other reference page imports it. It is traced to spec 003 story US-001 (REQ-003-US-001-007, 017, 018 and 032).

An AmbiguityCandidate is an action, a target, a confidence and optional metadata. AmbiguityResolver.detect_ambiguity returns an AmbiguityResult: with no candidates the reason is no_candidates; with one it is resolved immediately; otherwise _score_candidates scores them against the text, sorts by confidence, and the intent is ambiguous when the gap between the top two is below AMBIGUITY_THRESHOLD. resolve goes further for an ambiguous case. It first tries Prolog rules through an optional bridge (_try_prolog_resolution, reason prolog_resolved), then a context dictionary (_try_context_resolution, reason context_resolved), and finally falls back to the highest-scoring candidate. AmbiguityResult.confidence exposes the confidence of the resolved candidate.

At a glance

Class diagram

Dependency diagram

API

Classes

AmbiguityCandidate

Decorators: traces(spec='003', story='US-001', reqs=['REQ-003-US-001-017']) · Line 18

A candidate interpretation of an ambiguous intent.

AmbiguityResult

Decorators: traces(spec='003', story='US-001', reqs=['REQ-003-US-001-032']) · Line 45

Result of ambiguity detection/resolution.

AmbiguityResolver

Decorators: traces(spec='003', story='US-001', reqs=['REQ-003-US-001-007', 'REQ-003-US-001-017', 'REQ-003-US-001-018', 'REQ-003-US-… · Line 80

Detects and resolves ambiguous intent interpretations.

Source

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

"""Ambiguity detection and resolution for intents.

Detects when an intent maps to multiple valid interpretations
and resolves via scoring and optional Prolog rules.
"""

# Spec: 003 - Intent Broker

import logging
from typing import Any

from orchestrata_ultimate.core.base.traces import traces

logger = logging.getLogger(__name__)


@traces(spec="003", story="US-001", reqs=["REQ-003-US-001-017"])
class AmbiguityCandidate:
    """A candidate interpretation of an ambiguous intent."""

    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-032"])
    def __init__(
        self,
        action: str,
        target: str,
        confidence: float = 0.0,
        metadata: dict | None = None,
    ) -> None:
        self.action = action
        self.target = target
        self.confidence = confidence
        self.metadata = metadata or {}

    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-032"])
    def to_dict(self) -> dict[str, Any]:
        return {
            "action": self.action,
            "target": self.target,
            "confidence": self.confidence,
            "metadata": self.metadata,
        }


@traces(spec="003", story="US-001", reqs=["REQ-003-US-001-032"])
class AmbiguityResult:
    """Result of ambiguity detection/resolution."""

    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-032"])
    def __init__(
        self,
        is_ambiguous: bool,
        candidates: list[AmbiguityCandidate],
        resolved: AmbiguityCandidate | None = None,
        reason: str = "",
    ) -> None:
        self.is_ambiguous = is_ambiguous
        self.candidates = candidates
        self.resolved = resolved
        self.reason = reason

    @property
    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-017"])
    def confidence(self) -> float:
        if self.resolved:
            return self.resolved.confidence
        return 0.0


@traces(
    spec="003",
    story="US-001",
    reqs=[
        "REQ-003-US-001-007",
        "REQ-003-US-001-017",
        "REQ-003-US-001-018",
        "REQ-003-US-001-031",
        "REQ-003-US-001-032",
    ],
)
class AmbiguityResolver:
    """Detects and resolves ambiguous intent interpretations."""

    AMBIGUITY_THRESHOLD = 0.3  # Max gap between top two for ambiguity

    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-017"])
    def __init__(self, prolog_bridge: Any = None) -> None:
        self._prolog = prolog_bridge

    @traces(
        spec="003",
        story="US-001",
        reqs=["REQ-003-US-001-007", "REQ-003-US-001-017", "REQ-003-US-001-032"],
    )
    def detect_ambiguity(
        self, text: str, candidates: list[AmbiguityCandidate]
    ) -> AmbiguityResult:
        """Detect if candidates represent an ambiguous intent."""
        if not candidates:
            return AmbiguityResult(
                is_ambiguous=False,
                candidates=[],
                reason="no_candidates",
            )

        if len(candidates) == 1:
            return AmbiguityResult(
                is_ambiguous=False,
                candidates=candidates,
                resolved=candidates[0],
                reason="single_candidate",
            )

        # Score all candidates
        scored = self._score_candidates(text, candidates)

        # Sort by confidence descending
        scored.sort(key=lambda c: c.confidence, reverse=True)

        # Check if top two are too close
        gap = scored[0].confidence - scored[1].confidence
        is_ambiguous = gap < self.AMBIGUITY_THRESHOLD

        return AmbiguityResult(
            is_ambiguous=is_ambiguous,
            candidates=scored,
            resolved=scored[0] if not is_ambiguous else None,
            reason="ambiguous" if is_ambiguous else "clear_winner",
        )

    @traces(
        spec="003",
        story="US-001",
        reqs=["REQ-003-US-001-007", "REQ-003-US-001-018", "REQ-003-US-001-031"],
    )
    def resolve(
        self,
        text: str,
        candidates: list[AmbiguityCandidate],
        context: dict | None = None,
    ) -> AmbiguityResult:
        """Resolve ambiguity using scoring and optional Prolog rules."""
        if not candidates:
            return AmbiguityResult(
                is_ambiguous=False,
                candidates=[],
                reason="no_candidates",
            )

        detection = self.detect_ambiguity(text, candidates)

        if not detection.is_ambiguous:
            return detection

        # Try Prolog resolution
        if self._prolog is not None:
            prolog_result = self._try_prolog_resolution(text, candidates, context)
            if prolog_result is not None:
                return AmbiguityResult(
                    is_ambiguous=False,
                    candidates=detection.candidates,
                    resolved=prolog_result,
                    reason="prolog_resolved",
                )

        # Context-based resolution
        if context:
            context_result = self._try_context_resolution(candidates, context)
            if context_result is not None:
                return AmbiguityResult(
                    is_ambiguous=False,
                    candidates=detection.candidates,
                    resolved=context_result,
                    reason="context_resolved",
                )

        # Fall back to highest scoring
        best = detection.candidates[0]
        return AmbiguityResult(
            is_ambiguous=True,
            candidates=detection.candidates,
            resolved=best,
            reason="best_effort",
        )

    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-017"])
    def _score_candidates(
        self, text: str, candidates: list[AmbiguityCandidate]
    ) -> list[AmbiguityCandidate]:
        """Score candidates based on text similarity."""
        text_lower = text.lower()
        text_words = set(text_lower.split())

        for candidate in candidates:
            score = candidate.confidence  # Start with existing confidence

            # Action word match
            if candidate.action.lower() in text_lower:
                score += 0.4

            # Target word match
            if candidate.target.lower() in text_lower:
                score += 0.3

            # Word overlap
            action_words = set(candidate.action.lower().split("_"))
            target_words = set(candidate.target.lower().split("_"))
            all_candidate_words = action_words | target_words
            overlap = text_words & all_candidate_words
            if all_candidate_words:
                score += len(overlap) / len(all_candidate_words) * 0.3

            candidate.confidence = round(score, 4)

        return candidates

    @traces(spec="003", story="US-003", reqs=["REQ-003-US-003-012"])
    def _try_prolog_resolution(
        self, text: str, candidates: list[AmbiguityCandidate], context: dict | None
    ) -> AmbiguityCandidate | None:
        """Try resolving via Prolog rules."""
        try:
            for candidate in candidates:
                query = f"resolve_ambiguity('{candidate.action}', '{candidate.target}', Priority)"
                result = self._prolog.query(query)
                if result.success and result.results:
                    priority = result.results[0].get("Priority", 0)
                    candidate.confidence += float(priority) * 0.1
                    candidate.metadata["prolog_priority"] = priority

            # Re-sort after Prolog scoring
            candidates.sort(key=lambda c: c.confidence, reverse=True)
            if len(candidates) >= 2:
                gap = candidates[0].confidence - candidates[1].confidence
                if gap >= self.AMBIGUITY_THRESHOLD:
                    return candidates[0]
        except Exception as e:
            logger.warning("Prolog resolution failed: %s", e)

        return None

    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-020"])
    def _try_context_resolution(
        self, candidates: list[AmbiguityCandidate], context: dict
    ) -> AmbiguityCandidate | None:
        """Try resolving using context hints."""
        preferred_action = context.get("preferred_action")
        preferred_target = context.get("preferred_target")

        for candidate in candidates:
            if preferred_action and candidate.action == preferred_action:
                candidate.confidence += 0.5
            if preferred_target and candidate.target == preferred_target:
                candidate.confidence += 0.5

        candidates.sort(key=lambda c: c.confidence, reverse=True)
        if len(candidates) >= 2:
            gap = candidates[0].confidence - candidates[1].confidence
            if gap >= self.AMBIGUITY_THRESHOLD:
                return candidates[0]

        return None