orchestrata_ultimate.orchestrator.parameter_extractor

Reference › orchestrata_ultimate › orchestrator › parameter_extractor

Pulls structured parameters out of the free text of an intent. Its only importer is intent_parser, which accepts a ParameterExtractor through its constructor. It is traced to spec 003 story US-001.

The single class has one method per kind of parameter: extract_file_params (file paths, directories and glob patterns), extract_numeric_params (numbers with optional units), extract_constraint_params (clauses such as must or only, returned as key and value dictionaries), extract_list_params (comma or "and" separated lists) and extract_boolean_params (flags). extract_all runs them all and merges the results into one dictionary. resolve_ambiguity takes the text and a list of candidate interpretations and scores each one, returning the result as a dictionary. A separate module, ambiguity_resolver, chooses between whole intent readings. The module reads nothing from the environment and touches no storage.

At a glance

Class diagram

Dependency diagram

API

Classes

ParameterExtractor

Decorators: traces(spec='003', story='US-001', reqs=['REQ-003-US-001-005', 'REQ-003-US-001-006', 'REQ-003-US-001-015', 'REQ-003-US-… · Line 27

Extracts structured parameters from natural language intent text.

Source

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

# mypy: ignore-errors
"""Advanced parameter extraction from natural language.

Extracts file paths, numeric values, constraints, lists,
and resolves ambiguous parameters from intent text.
"""

import logging
import re
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-005",
        "REQ-003-US-001-006",
        "REQ-003-US-001-015",
        "REQ-003-US-001-016",
    ],
)
class ParameterExtractor:
    """Extracts structured parameters from natural language intent text."""

    # File path patterns
    _FILE_PATH_RE = re.compile(
        r"(?:^|\s)(/(?:[a-zA-Z0-9_.\-]+/)*[a-zA-Z0-9_.\-]+(?:\.[a-zA-Z0-9]+)?)"
    )
    _GLOB_PATTERN_RE = re.compile(
        r"(?:^|\s)(\*\*?/[^\s]+|[^\s]*\*[^\s]*\.[a-zA-Z0-9]+)"
    )
    _RELATIVE_PATH_RE = re.compile(r"(?:^|\s)(\.{1,2}/[^\s]+)")

    # Numeric patterns with units
    _NUMERIC_RE = re.compile(
        r"(\d+(?:\.\d+)?)\s*(bytes?|[kKmMgGtT][bB]|ms|seconds?|secs?|minutes?|mins?|hours?|hrs?|days?|%|count)?",
        re.IGNORECASE,
    )

    # Constraint patterns
    _CONSTRAINT_PATTERNS = [
        (re.compile(r"must\s+(?:be\s+)?(.+?)(?:\.|$)", re.IGNORECASE), "must"),
        (re.compile(r"within\s+(\d+\s*\w+)", re.IGNORECASE), "within"),
        (re.compile(r"before\s+(.+?)(?:\.|$)", re.IGNORECASE), "before"),
        (re.compile(r"at\s+most\s+(\d+\s*\w+)", re.IGNORECASE), "at_most"),
        (re.compile(r"at\s+least\s+(\d+\s*\w+)", re.IGNORECASE), "at_least"),
        (re.compile(r"no\s+more\s+than\s+(\d+\s*\w+)", re.IGNORECASE), "no_more_than"),
        (re.compile(r"timeout\s+(?:of\s+)?(\d+\s*\w+)", re.IGNORECASE), "timeout"),
    ]

    # Boolean indicators
    _ENABLE_WORDS = frozenset({"enable", "on", "yes", "true", "activate", "start"})
    _DISABLE_WORDS = frozenset({"disable", "off", "no", "false", "deactivate", "stop"})

    # List separators
    _LIST_SEP_RE = re.compile(r",\s*(?:and\s+)?|\s+and\s+")

    # Unit multipliers (to bytes)
    _UNIT_BYTES = {
        "b": 1,
        "byte": 1,
        "bytes": 1,
        "kb": 1024,
        "mb": 1024**2,
        "gb": 1024**3,
        "tb": 1024**4,
    }

    # Unit multipliers (to seconds)
    _UNIT_SECONDS = {
        "ms": 0.001,
        "s": 1,
        "sec": 1,
        "secs": 1,
        "second": 1,
        "seconds": 1,
        "m": 60,
        "min": 60,
        "mins": 60,
        "minute": 60,
        "minutes": 60,
        "h": 3600,
        "hr": 3600,
        "hrs": 3600,
        "hour": 3600,
        "hours": 3600,
        "d": 86400,
        "day": 86400,
        "days": 86400,
    }

    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-005"])
    def extract_file_params(self, text: str) -> dict[str, Any]:
        """Extract file paths, directories, and glob patterns."""
        result: dict[str, Any] = {}

        # Absolute paths
        paths = self._FILE_PATH_RE.findall(text)
        if paths:
            result["paths"] = paths

        # Glob patterns
        globs = self._GLOB_PATTERN_RE.findall(text)
        if globs:
            result["glob_patterns"] = [g.strip() for g in globs]

        # Relative paths
        relatives = self._RELATIVE_PATH_RE.findall(text)
        if relatives:
            result["relative_paths"] = [r.strip() for r in relatives]

        return result

    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-005"])
    def extract_numeric_params(self, text: str) -> dict[str, Any]:
        """Extract numeric values with optional units."""
        result: dict[str, Any] = {}
        values = []

        for match in self._NUMERIC_RE.finditer(text):
            raw_value = float(match.group(1))
            unit = (match.group(2) or "").lower().rstrip("s")

            entry: dict[str, Any] = {"raw": raw_value, "unit": match.group(2) or ""}

            # Normalize to base units
            if unit in self._UNIT_BYTES or unit + "s" in self._UNIT_BYTES:
                key = unit if unit in self._UNIT_BYTES else unit + "s"
                entry["bytes"] = int(raw_value * self._UNIT_BYTES.get(key, 1))
                entry["type"] = "size"
            elif unit in self._UNIT_SECONDS or unit + "s" in self._UNIT_SECONDS:
                key = unit if unit in self._UNIT_SECONDS else unit + "s"
                entry["seconds"] = raw_value * self._UNIT_SECONDS.get(key, 1)
                entry["type"] = "duration"
            elif unit == "%":
                entry["type"] = "percentage"
            elif unit == "count" or unit == "":
                entry["type"] = "count"

            values.append(entry)

        if values:
            result["numeric_values"] = values

        return result

    @traces(
        spec="003",
        story="US-001",
        reqs=["REQ-003-US-001-006", "REQ-003-US-001-016"],
    )
    def extract_constraint_params(self, text: str) -> list[dict[str, str]]:
        """Extract constraint clauses (must, within, before, at most, etc.)."""
        constraints = [
            {
                "type": constraint_type,
                "value": match.group(1).strip(),
            }
            for pattern, constraint_type in self._CONSTRAINT_PATTERNS
            for match in pattern.finditer(text)
        ]
        return constraints

    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-005"])
    def extract_list_params(self, text: str) -> dict[str, list[str]]:
        """Extract comma/and-separated lists."""
        result: dict[str, list[str]] = {}

        # Look for "install X, Y, and Z" or similar patterns
        list_patterns = [
            re.compile(
                r"(?:install|add|include|import|use)\s+(.+?)(?:\s*$|\s*\.)",
                re.IGNORECASE,
            ),
            re.compile(
                r"(?:files?|packages?|modules?|items?):\s*(.+?)(?:\s*$|\s*\.)",
                re.IGNORECASE,
            ),
        ]

        for pattern in list_patterns:
            match = pattern.search(text)
            if match:
                items_text = match.group(1)
                items = self._LIST_SEP_RE.split(items_text)
                items = [item.strip() for item in items if item.strip()]
                if len(items) > 1:
                    result["items"] = items

        return result

    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-005"])
    def extract_boolean_params(self, text: str) -> dict[str, bool]:
        """Extract boolean flags from text."""
        result: dict[str, bool] = {}
        words = text.lower().split()

        for i, word in enumerate(words):
            if word in self._ENABLE_WORDS:
                # Look for the thing being enabled
                if i + 1 < len(words):
                    key = words[i + 1]
                    result[key] = True
            elif word in self._DISABLE_WORDS:
                if i + 1 < len(words):
                    key = words[i + 1]
                    result[key] = False

        return result

    @traces(
        spec="003",
        story="US-001",
        reqs=["REQ-003-US-001-005", "REQ-003-US-001-015"],
    )
    def extract_all(self, text: str) -> dict[str, Any]:
        """Extract all parameter types from text."""
        if not text:
            return {}

        result: dict[str, Any] = {}

        file_params = self.extract_file_params(text)
        if file_params:
            result["files"] = file_params

        numeric_params = self.extract_numeric_params(text)
        if numeric_params:
            result["numeric"] = numeric_params

        constraints = self.extract_constraint_params(text)
        if constraints:
            result["constraints"] = constraints

        list_params = self.extract_list_params(text)
        if list_params:
            result["lists"] = list_params

        boolean_params = self.extract_boolean_params(text)
        if boolean_params:
            result["booleans"] = boolean_params

        return result

    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-017"])
    def resolve_ambiguity(self, text: str, candidates: list[str]) -> dict[str, Any]:
        """Score multiple interpretation candidates and pick best."""
        if not candidates:
            return {"best": None, "scores": {}}

        scores: dict[str, float] = {}
        text_lower = text.lower()

        for candidate in candidates:
            score = 0.0
            candidate_lower = candidate.lower()

            # Exact match bonus
            if candidate_lower in text_lower:
                score += 1.0

            # Word overlap score
            text_words = set(text_lower.split())
            candidate_words = set(candidate_lower.split())
            overlap = text_words & candidate_words
            if text_words:
                score += len(overlap) / len(text_words) * 0.5

            # Length similarity
            len_ratio = min(len(candidate), len(text)) / max(
                len(candidate), len(text), 1
            )
            score += len_ratio * 0.3

            scores[candidate] = round(score, 4)

        # `scores[key]` (not `scores.get`, which types as `-> float | None`
        # and isn't a valid `SupportsRichComparison` key function): every
        # candidate scored in the loop above is a key in `scores`.
        best = max(scores, key=lambda k: scores[k])
        return {"best": best, "scores": scores, "confidence": scores[best]}