orchestrata_ultimate.orchestrator.impact.classifier

Reference › orchestrata_ultimate › orchestrator › impact › classifier

The first step of impact analysis: it classifies the intent text into an IntentType using keyword patterns. It is used by analyzer and re-exported by the package facade. Do not confuse it with the natural-language classifier in intent_classifier, which assigns work types to user requests.

IntentClassifier compiles one regular expression per intent type from keyword lists in its constructor. classify lowercases the text and _score_intent_types counts the matches for each type. No matches gives UNKNOWN with confidence zero; otherwise the highest count wins, ties are broken by a fixed precedence (so a sentence about refactoring something in order to add a feature stays a refactor), and confidence is the count divided by five, capped at one. The result is an IntentClassification that also carries the entities, the target package and module and the keywords found by the _extract_* helpers.

At a glance

Class diagram

Dependency diagram

API

Classes

IntentClassifier

Decorators: traces(spec='019', story='US-001', reqs=['REQ-019-001']) · Line 114

Classifies intents into types and extracts entities.

Source

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

"""Intent classifier - analyzes text to classify intent type.

Uses keyword matching and pattern recognition to classify intents as
NEW_REQUIREMENT, NEW_SPEC, FEATURE, BUGFIX, or REFACTOR.
"""

import re

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.orchestrator.impact.types import (
    IntentClassification,
    IntentType,
)

REQUIREMENT_KEYWORDS = [
    "need",
    "require",
    "requirement",
    "must have",
    "should have",
    "want",
    "wish",
    "demand",
    "specification",
    "user story",
    "as a",
    "i want to",
    "so that",
]

SPEC_KEYWORDS = [
    "spec",
    "specification",
    "standard",
    "protocol",
    "interface",
    "api",
    "contract",
    "define",
    "schema",
    "structure",
]

FEATURE_KEYWORDS = [
    "add",
    "implement",
    "create",
    "build",
    "develop",
    "new functionality",
    "new feature",
    "enhance",
    "extend",
    "introduce",
    "support",
    "enable",
]

BUGFIX_KEYWORDS = [
    "fix",
    "bug",
    "error",
    "issue",
    "problem",
    "broken",
    "not working",
    "fails",
    "failing",
    "wrong",
    "incorrect",
    "repair",
    "resolve",
    "solve",
]

REFACTOR_KEYWORDS = [
    "refactor",
    "restructure",
    "reorganize",
    "rewrite",
    "clean up",
    "simplify",
    "improve code",
    "technical debt",
    "redesign",
    "rearchitect",
    "optimize",
]

# Tie-break precedence when several intent types match the same number of
# keywords. An explicit operation verb (refactor) outranks the generic feature
# verbs ("add"/"create"/"new ..."): "refactor X to add Y" is a refactor, not a
# feature. NEW_REQUIREMENT/NEW_SPEC stay highest (meta-intents); FEATURE keeps
# its precedence over BUGFIX (both unchanged). Replaces the old dict-insertion
# tie-break, which let FEATURE silently beat REFACTOR.
_TIE_PRECEDENCE = [
    IntentType.NEW_REQUIREMENT,
    IntentType.NEW_SPEC,
    IntentType.REFACTOR,
    IntentType.FEATURE,
    IntentType.BUGFIX,
]

ENTITY_PATTERNS = [
    (r"(?:file|module|class|function)\s+(\w+)", "code_entity"),
    (r"packages?/orchestrata-(\w+)", "package"),
    (r"(\w+)\.py", "file"),
    (r"def\s+(\w+)", "function"),
    (r"class\s+(\w+)", "class"),
]


@traces(spec="019", story="US-001", reqs=["REQ-019-001"])
class IntentClassifier:
    """Classifies intents into types and extracts entities.

    Uses regex-based keyword matching against five intent
    categories with confidence scoring.
    """

    def __init__(self) -> None:
        """Initialize classifier with compiled patterns."""
        self._patterns = {
            IntentType.NEW_REQUIREMENT: self._build_pattern(REQUIREMENT_KEYWORDS),
            IntentType.NEW_SPEC: self._build_pattern(SPEC_KEYWORDS),
            IntentType.FEATURE: self._build_pattern(FEATURE_KEYWORDS),
            IntentType.BUGFIX: self._build_pattern(BUGFIX_KEYWORDS),
            IntentType.REFACTOR: self._build_pattern(REFACTOR_KEYWORDS),
        }

    def _build_pattern(self, keywords: list[str]) -> re.Pattern:
        """Build case-insensitive regex from keyword list.

        Args:
            keywords: List of keyword strings.

        Returns:
            Compiled regex pattern.
        """
        escaped = [re.escape(k) for k in keywords]
        return re.compile("|".join(escaped), re.IGNORECASE)

    def classify(self, intent_text: str) -> IntentClassification:
        """Classify an intent and extract entities.

        Args:
            intent_text: Raw intent text from user.

        Returns:
            IntentClassification with type, confidence,
            entities, and target package/module.
        """
        text_lower = intent_text.lower()
        intent_type, confidence = self._score_intent_types(text_lower)

        return IntentClassification(
            type=intent_type,
            confidence=confidence,
            entities=self._extract_entities(intent_text),
            target_package=self._extract_package(intent_text),
            target_module=self._extract_module(intent_text),
            keywords=self._extract_keywords(text_lower),
        )

    def _score_intent_types(self, text_lower: str) -> tuple[IntentType, float]:
        """Score text against each intent type pattern.

        Args:
            text_lower: Lowercased intent text.

        Returns:
            Tuple of (best IntentType, confidence float).
        """
        scores = {
            itype: len(pattern.findall(text_lower))
            for itype, pattern in self._patterns.items()
        }
        max_score = max(scores.values())

        if max_score == 0:
            return IntentType.UNKNOWN, 0.0

        # Break ties by explicit precedence (refactor > feature/bugfix) rather
        # than dict-insertion order, so "refactor X to add Y" stays a refactor.
        best_type = next(t for t in _TIE_PRECEDENCE if scores[t] == max_score)
        confidence = min(max_score / 5.0, 1.0)
        return best_type, confidence

    def _extract_entities(self, text: str) -> list[str]:
        """Extract code entities from intent text.

        Args:
            text: Raw intent text.

        Returns:
            Deduplicated list of entity names.
        """
        entities = []
        for pattern, _ in ENTITY_PATTERNS:
            matches = re.findall(pattern, text, re.IGNORECASE)
            entities.extend(matches)
        return list(set(entities))

    def _extract_package(self, text: str) -> str | None:
        """Extract package name from intent text.

        Args:
            text: Raw intent text.

        Returns:
            Package name string or None.
        """
        match = re.search(r"orchestrata-(\w+)", text, re.IGNORECASE)
        if match:
            return f"orchestrata-{match.group(1)}"

        match = re.search(r"(?:package|module)\s+(\w+)", text, re.IGNORECASE)
        if match:
            return match.group(1)
        return None

    def _extract_module(self, text: str) -> str | None:
        """Extract module name from intent text.

        Args:
            text: Raw intent text.

        Returns:
            Module path string or None.
        """
        match = re.search(r"(?:in|from)\s+([\w.]+)(?:\s|$)", text)
        if match:
            return match.group(1)
        return None

    def _extract_keywords(self, text: str) -> list[str]:
        """Extract all matched keywords from text.

        Args:
            text: Lowercased intent text.

        Returns:
            List of matched keywords, max 10.
        """
        all_keywords = (
            REQUIREMENT_KEYWORDS
            + SPEC_KEYWORDS
            + FEATURE_KEYWORDS
            + BUGFIX_KEYWORDS
            + REFACTOR_KEYWORDS
        )
        found = [kw for kw in all_keywords if kw.lower() in text]
        return found[:10]