orchestrata_ultimate.orchestrator.nlp

Reference › orchestrata_ultimate › orchestrator › nlp

The text-processing layer under the natural-language intent path. Its only importer is intent_classifier, and it is traced to spec 003 story US-001. It has no external dependencies beyond the traceability decorator; everything is regular-expression based.

Tokenizer.tokenize splits text into Token records classified by TokenType (word, number, path, operator, punctuation, whitespace, string, glob, url, flag); tokenize_sentences splits sentences and get_token_types groups tokens by type. NamedEntityRecognizer.recognize finds typed NamedEntity spans (EntityType: file path, directory, package, command, agent, number, duration, size, percentage, url, git ref, variable) with a confidence score, running URL, file path, agent alias (the U: volk aliases), git ref, measurement and variable recognizers in turn and skipping any span that overlaps one already found. extract_by_type filters by entity type.

Context handling is in SessionContext, a dataclass holding the session id, current directory, recent files and commands (each capped at 20), active agent, variables and a bounded history (50 by default), and ContextAwareParser. Its parse_with_context tokenizes, recognizes entities, and replaces pronoun references such as "it" with the most recent file or command from the session, returning a dictionary with the original and resolved text, the tokens, the entities, the references resolved and a context_used flag. has_context_references tells a caller whether the text needs that step.

At a glance

Class diagram

Dependency diagram

API

Classes

TokenType (enum)

Bases: Enum · Decorators: traces(spec='003', story='US-001', reqs=['REQ-003-US-001-012']) · Line 24

Token classification types.

Token (dataclass)

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

A single token from text input.

Tokenizer

Decorators: traces(spec='003', story='US-001', reqs=['REQ-003-US-001-008', 'REQ-003-US-001-012', 'REQ-003-US-001-019', 'REQ-003-US-… · Line 66

Tokenizes natural language input for intent parsing.

EntityType (enum)

Bases: Enum · Decorators: traces(spec='003', story='US-001', reqs=['REQ-003-US-001-015']) · Line 160

Named entity types for intent extraction.

NamedEntity (dataclass)

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

A recognized named entity from text.

NamedEntityRecognizer

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

Extracts named entities from natural language intent text.

SessionContext (dataclass)

Decorators: dataclass, traces(spec='003', story='US-001', reqs=['REQ-003-US-001-009', 'REQ-003-US-001-020']) · Line 423

Session state for context-aware parsing.

ContextAwareParser

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

Enhances intent parsing with session state context.

Source

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

"""NLP processing for intent parsing.

Provides tokenization, named entity recognition, and context-aware
parsing for natural language intent extraction.

Spec: 003-intent / US-001
"""

import logging
import re
from dataclasses import dataclass, field
from enum import Enum
from typing import Any

from orchestrata_ultimate.core.base.traces import traces

logger = logging.getLogger(__name__)


# ── Tokenization (REQ-003-US-001-012) ────────────────────────────────────


@traces(spec="003", story="US-001", reqs=["REQ-003-US-001-012"])
class TokenType(Enum):
    """Token classification types."""

    WORD = "word"
    NUMBER = "number"
    PATH = "path"
    OPERATOR = "operator"
    PUNCTUATION = "punctuation"
    WHITESPACE = "whitespace"
    STRING = "string"
    GLOB = "glob"
    URL = "url"
    FLAG = "flag"


@dataclass
@traces(spec="003", story="US-001", reqs=["REQ-003-US-001-012"])
class Token:
    """A single token from text input."""

    text: str
    token_type: TokenType
    start: int
    end: int
    normalized: str = ""

    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-012"])
    def __post_init__(self) -> None:
        if not self.normalized:
            self.normalized = self.text.lower().strip()


@traces(
    spec="003",
    story="US-001",
    reqs=[
        "REQ-003-US-001-008",
        "REQ-003-US-001-012",
        "REQ-003-US-001-019",
        "REQ-003-US-001-024",
    ],
)
class Tokenizer:
    """Tokenizes natural language input for intent parsing.

    Recognizes domain-specific tokens: file paths, glob patterns,
    URLs, CLI flags, numbers with units, and natural language words.
    """

    # Ordered patterns — first match wins
    _PATTERNS = [
        (TokenType.URL, re.compile(r"https?://[^\s]+")),
        (
            TokenType.PATH,
            re.compile(
                r"(?:~|\.{1,2})?/(?:[a-zA-Z0-9_.\-]+/)*[a-zA-Z0-9_.\-]+(?:\.[a-zA-Z0-9]+)?"
            ),
        ),
        (TokenType.GLOB, re.compile(r"(?:\*\*?/[^\s]+|[^\s]*\*[^\s]*\.[a-zA-Z0-9]+)")),
        (TokenType.FLAG, re.compile(r"--?[a-zA-Z][a-zA-Z0-9\-]*(?:=[^\s]+)?")),
        (TokenType.STRING, re.compile(r'"[^"]*"|\'[^\']*\'')),
        (
            TokenType.NUMBER,
            re.compile(
                r"\d+(?:\.\d+)?(?:\s*(?:ms|seconds?|secs?|minutes?|mins?|hours?|hrs?|days?|bytes?|[kKmMgGtT][bB]|%))?"
            ),
        ),
        (TokenType.OPERATOR, re.compile(r"[=<>!]=?|&&|\|\||[|&;]")),
        (TokenType.PUNCTUATION, re.compile(r"[.,;:!?(){}\[\]]")),
        (TokenType.WORD, re.compile(r"[a-zA-Z_][a-zA-Z0-9_]*")),
        (TokenType.WHITESPACE, re.compile(r"\s+")),
    ]

    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-012"])
    def tokenize(self, text: str) -> list[Token]:
        """Tokenize text into classified tokens.

        Args:
            text: Raw input text to tokenize.

        Returns:
            List of Token objects (whitespace tokens excluded).
        """
        if not text:
            return []

        tokens = []
        pos = 0

        while pos < len(text):
            matched = False
            for token_type, pattern in self._PATTERNS:
                m = pattern.match(text, pos)
                if m:
                    if token_type != TokenType.WHITESPACE:
                        tokens.append(
                            Token(
                                text=m.group(),
                                token_type=token_type,
                                start=m.start(),
                                end=m.end(),
                            )
                        )
                    pos = m.end()
                    matched = True
                    break

            if not matched:
                # Skip unrecognized character
                pos += 1

        return tokens

    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-012"])
    def tokenize_sentences(self, text: str) -> list[str]:
        """Split text into sentences."""
        if not text:
            return []
        # Split on sentence-ending punctuation followed by space or end
        sentences = re.split(r"(?<=[.!?])\s+", text.strip())
        return [s.strip() for s in sentences if s.strip()]

    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-012"])
    def get_token_types(self, text: str) -> dict[TokenType, list[str]]:
        """Group tokens by their type."""
        tokens = self.tokenize(text)
        groups: dict[TokenType, list[str]] = {}
        for token in tokens:
            groups.setdefault(token.token_type, []).append(token.text)
        return groups


# ── Named Entity Recognition (REQ-003-US-001-015) ───────────────────


@traces(spec="003", story="US-001", reqs=["REQ-003-US-001-015"])
class EntityType(Enum):
    """Named entity types for intent extraction."""

    FILE_PATH = "file_path"
    DIRECTORY = "directory"
    PACKAGE = "package"
    COMMAND = "command"
    AGENT = "agent"
    NUMBER = "number"
    DURATION = "duration"
    SIZE = "size"
    PERCENTAGE = "percentage"
    URL = "url"
    GIT_REF = "git_ref"
    VARIABLE = "variable"


@dataclass
@traces(spec="003", story="US-001", reqs=["REQ-003-US-001-015"])
class NamedEntity:
    """A recognized named entity from text."""

    text: str
    entity_type: EntityType
    start: int
    end: int
    confidence: float = 1.0
    metadata: dict[str, Any] = field(default_factory=dict)


@traces(
    spec="003",
    story="US-001",
    reqs=["REQ-003-US-001-015", "REQ-003-US-001-024"],
)
class NamedEntityRecognizer:
    """Extracts named entities from natural language intent text.

    Uses pattern-based recognition for domain-specific entities:
    file paths, packages, commands, agents, and value types.
    """

    # Agent aliases
    _AGENT_PATTERN = re.compile(r"\bU:[A-Z][a-z]+\b")
    _AGENT_NAMES = frozenset(
        {"U:Ox", "U:PS", "U:Cd", "U:Ux", "U:PM", "U:Ar", "U:Qa", "U:Cx"}
    )

    # Git refs
    _GIT_REF_PATTERN = re.compile(
        r"\b(?:main|master|develop|HEAD|[a-f0-9]{7,40}|v\d+\.\d+(?:\.\d+)?)\b"
    )

    # Package names (PyPI-style)
    _PACKAGE_PATTERN = re.compile(r"\b[a-z][a-z0-9]*(?:[-_][a-z0-9]+)*\b")

    # Command patterns
    _COMMAND_KEYWORDS = frozenset(
        {
            "run",
            "execute",
            "start",
            "stop",
            "restart",
            "install",
            "uninstall",
            "build",
            "test",
            "deploy",
            "lint",
            "format",
            "check",
            "migrate",
            "create",
            "delete",
            "update",
            "list",
            "show",
            "describe",
        }
    )

    # Duration patterns
    _DURATION_RE = re.compile(
        r"\b(\d+(?:\.\d+)?)\s*(ms|seconds?|secs?|minutes?|mins?|hours?|hrs?|days?)\b",
        re.IGNORECASE,
    )

    # Size patterns
    _SIZE_RE = re.compile(
        r"\b(\d+(?:\.\d+)?)\s*([kKmMgGtT][bB]|bytes?)\b", re.IGNORECASE
    )

    # Percentage
    _PERCENT_RE = re.compile(r"\b(\d+(?:\.\d+)?)\s*%")

    # URL
    _URL_RE = re.compile(r"https?://[^\s]+")

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

    # Variable/identifier (UPPER_SNAKE_CASE)
    _VARIABLE_RE = re.compile(r"\b[A-Z][A-Z0-9_]{2,}\b")

    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-015"])
    def recognize(self, text: str) -> list[NamedEntity]:
        """Extract all named entities from text.

        Args:
            text: Natural language text to analyze.

        Returns:
            List of NamedEntity objects sorted by position.
        """
        if not text:
            return []

        entities: list[NamedEntity] = []

        # URLs first so later file-path matches inside a URL are skipped.
        self._recognize_urls(text, entities)
        self._recognize_file_paths(text, entities)
        self._recognize_agents(text, entities)
        self._recognize_git_refs(text, entities)
        self._recognize_measurements(text, entities)
        self._recognize_variables(text, entities)

        entities.sort(key=lambda e: e.start)
        return entities

    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-015"])
    def _recognize_urls(self, text: str, entities: list[NamedEntity]) -> None:
        """Recognize URLs (matched first to shadow paths inside them)."""
        for m in self._URL_RE.finditer(text):
            entities.append(
                NamedEntity(
                    text=m.group(),
                    entity_type=EntityType.URL,
                    start=m.start(),
                    end=m.end(),
                )
            )

    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-015"])
    def _recognize_file_paths(self, text: str, entities: list[NamedEntity]) -> None:
        """Recognize file paths, skipping spans already captured (e.g. URLs)."""
        for m in self._FILE_PATH_RE.finditer(text):
            if not self._overlaps(entities, m.start(), m.end()):
                entities.append(
                    NamedEntity(
                        text=m.group(),
                        entity_type=EntityType.FILE_PATH,
                        start=m.start(),
                        end=m.end(),
                    )
                )

    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-015"])
    def _recognize_agents(self, text: str, entities: list[NamedEntity]) -> None:
        """Recognize known agent aliases (e.g. U:Ox)."""
        for m in self._AGENT_PATTERN.finditer(text):
            if m.group() in self._AGENT_NAMES:
                entities.append(
                    NamedEntity(
                        text=m.group(),
                        entity_type=EntityType.AGENT,
                        start=m.start(),
                        end=m.end(),
                    )
                )

    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-015"])
    def _recognize_git_refs(self, text: str, entities: list[NamedEntity]) -> None:
        """Recognize git refs not already captured as another entity."""
        for m in self._GIT_REF_PATTERN.finditer(text):
            if not self._overlaps(entities, m.start(), m.end()):
                entities.append(
                    NamedEntity(
                        text=m.group(),
                        entity_type=EntityType.GIT_REF,
                        start=m.start(),
                        end=m.end(),
                        confidence=0.8,
                    )
                )

    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-015"])
    def _recognize_measurements(self, text: str, entities: list[NamedEntity]) -> None:
        """Recognize durations, sizes, and percentages with numeric metadata."""
        for m in self._DURATION_RE.finditer(text):
            entities.append(
                NamedEntity(
                    text=m.group(),
                    entity_type=EntityType.DURATION,
                    start=m.start(),
                    end=m.end(),
                    metadata={"value": float(m.group(1)), "unit": m.group(2)},
                )
            )

        for m in self._SIZE_RE.finditer(text):
            entities.append(
                NamedEntity(
                    text=m.group(),
                    entity_type=EntityType.SIZE,
                    start=m.start(),
                    end=m.end(),
                    metadata={"value": float(m.group(1)), "unit": m.group(2)},
                )
            )

        for m in self._PERCENT_RE.finditer(text):
            entities.append(
                NamedEntity(
                    text=m.group(),
                    entity_type=EntityType.PERCENTAGE,
                    start=m.start(),
                    end=m.end(),
                    metadata={"value": float(m.group(1))},
                )
            )

    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-015"])
    def _recognize_variables(self, text: str, entities: list[NamedEntity]) -> None:
        """Recognize UPPER_SNAKE_CASE variables not already captured."""
        for m in self._VARIABLE_RE.finditer(text):
            if not self._overlaps(entities, m.start(), m.end()):
                entities.append(
                    NamedEntity(
                        text=m.group(),
                        entity_type=EntityType.VARIABLE,
                        start=m.start(),
                        end=m.end(),
                        confidence=0.7,
                    )
                )

    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-015"])
    def _overlaps(self, entities: list[NamedEntity], start: int, end: int) -> bool:
        """Check if a span overlaps with any existing entity."""
        for e in entities:
            if start < e.end and end > e.start:
                return True
        return False

    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-015"])
    def extract_by_type(self, text: str, entity_type: EntityType) -> list[NamedEntity]:
        """Extract entities of a specific type."""
        return [e for e in self.recognize(text) if e.entity_type == entity_type]


# ── Context-Aware Parsing (REQ-003-US-001-020) ──────────────────────


@dataclass
@traces(
    spec="003",
    story="US-001",
    reqs=["REQ-003-US-001-009", "REQ-003-US-001-020"],
)
class SessionContext:
    """Session state for context-aware parsing."""

    session_id: str = ""
    current_directory: str = ""
    recent_files: list[str] = field(default_factory=list)
    recent_commands: list[str] = field(default_factory=list)
    active_agent: str = ""
    variables: dict[str, Any] = field(default_factory=dict)
    history: list[str] = field(default_factory=list)
    max_history: int = 50

    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-020"])
    def add_to_history(self, text: str) -> None:
        self.history.append(text)
        if len(self.history) > self.max_history:
            self.history = self.history[-self.max_history :]

    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-020"])
    def add_recent_file(self, path: str) -> None:
        if path not in self.recent_files:
            self.recent_files.append(path)
        if len(self.recent_files) > 20:
            self.recent_files = self.recent_files[-20:]

    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-020"])
    def add_recent_command(self, cmd: str) -> None:
        self.recent_commands.append(cmd)
        if len(self.recent_commands) > 20:
            self.recent_commands = self.recent_commands[-20:]


@traces(
    spec="003",
    story="US-001",
    reqs=["REQ-003-US-001-009", "REQ-003-US-001-020"],
)
class ContextAwareParser:
    """Enhances intent parsing with session state context.

    Resolves ambiguous references using session history, recent files,
    and current working directory. Expands pronouns like "it", "that file",
    "the same" to concrete references.
    """

    # Pronoun patterns that reference previous context
    _PRONOUN_PATTERNS = [
        re.compile(r"\b(?:it|this|that)\b", re.IGNORECASE),
        re.compile(r"\bthat\s+file\b", re.IGNORECASE),
        re.compile(r"\bthe\s+same\s+(?:file|command|thing)\b", re.IGNORECASE),
        re.compile(r"\bthere\b", re.IGNORECASE),
        re.compile(r"\blast\s+(?:file|command|result)\b", re.IGNORECASE),
    ]

    # Relative path indicators
    _RELATIVE_PATTERNS = [
        re.compile(r"\bhere\b", re.IGNORECASE),
        re.compile(r"\bcurrent\s+(?:dir|directory|folder)\b", re.IGNORECASE),
        re.compile(r"\bthis\s+(?:dir|directory|folder)\b", re.IGNORECASE),
    ]

    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-020"])
    def __init__(self) -> None:
        self._tokenizer = Tokenizer()
        self._ner = NamedEntityRecognizer()

    @traces(
        spec="003",
        story="US-001",
        reqs=["REQ-003-US-001-009", "REQ-003-US-001-020"],
    )
    def parse_with_context(
        self,
        text: str,
        context: SessionContext,
    ) -> dict[str, Any]:
        """Parse text with session context for disambiguation.

        Args:
            text: Natural language input.
            context: Current session state.

        Returns:
            Dict with resolved references and extracted info.
        """
        result: dict[str, Any] = {
            "original_text": text,
            "resolved_text": text,
            "tokens": [],
            "entities": [],
            "resolved_references": [],
            "context_used": False,
        }

        # Tokenize
        tokens = self._tokenizer.tokenize(text)
        result["tokens"] = [
            {"text": t.text, "type": t.token_type.value} for t in tokens
        ]

        # NER
        entities = self._ner.recognize(text)
        result["entities"] = [
            {"text": e.text, "type": e.entity_type.value, "confidence": e.confidence}
            for e in entities
        ]

        # Resolve pronoun references
        resolved_text = text
        for pattern in self._PRONOUN_PATTERNS:
            if pattern.search(resolved_text):
                replacement = self._resolve_pronoun(pattern, resolved_text, context)
                if replacement:
                    resolved_text = pattern.sub(replacement, resolved_text, count=1)
                    result["resolved_references"].append(
                        {
                            "pattern": pattern.pattern,
                            "resolved_to": replacement,
                        }
                    )
                    result["context_used"] = True

        # Resolve relative paths
        for pattern in self._RELATIVE_PATTERNS:
            if pattern.search(resolved_text) and context.current_directory:
                resolved_text = pattern.sub(
                    context.current_directory, resolved_text, count=1
                )
                result["resolved_references"].append(
                    {
                        "pattern": pattern.pattern,
                        "resolved_to": context.current_directory,
                    }
                )
                result["context_used"] = True

        result["resolved_text"] = resolved_text

        # Add to history
        context.add_to_history(text)

        return result

    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-020"])
    def _resolve_pronoun(
        self,
        pattern: re.Pattern,
        text: str,
        context: SessionContext,
    ) -> str | None:
        """Resolve a pronoun reference using session context."""
        # Find what was actually matched in text to determine resolution target
        m = pattern.search(text)
        matched_text = m.group().lower() if m else ""

        # "that file" / "the same file" / "last file" -> most recent file
        if "file" in matched_text:
            if context.recent_files:
                return context.recent_files[-1]
            return None

        # "last command" / "the same command" -> most recent command
        if "command" in matched_text:
            if context.recent_commands:
                return context.recent_commands[-1]
            return None

        # Generic "it" / "this" / "that" / "there" -> try recent file, then command
        if context.recent_files:
            return context.recent_files[-1]
        if context.recent_commands:
            return context.recent_commands[-1]

        return None

    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-020"])
    def has_context_references(self, text: str) -> bool:
        """Check if text contains references that need context resolution."""
        for pattern in self._PRONOUN_PATTERNS + self._RELATIVE_PATTERNS:
            if pattern.search(text):
                return True
        return False