Add Hierarchical two-stage intent variant

README.md

@@ -156,6 +156,14 @@ Padacioso ships as an OVOS pipeline plugin (`ovos-padacioso-pipeline-plugin`) an
 }
 ```
 
+## Hierarchical (domain) pipeline
+
+A second OPM entry point — `ovos-padacioso-domain-pipeline` — exposes a two-level
+variant backed by `DomainIntentContainer`. Intents are grouped by `skill_id` (taken from
+the `skill_id:intent` label prefix); the top-level classifier first picks the most likely
+domain, then the per-domain sub-container resolves the intent. See
+[docs/domain_pipeline.md](docs/domain_pipeline.md) for configuration and routing details.
+
 ## License
 
 Apache 2.0

docs/domain_pipeline.md

@@ -0,0 +1,110 @@
+# Domain Padacioso Pipeline
+
+This page documents two layers that ship together:
+
+* **`DomainPadaciosoPipeline`** — the OPM-discoverable pipeline class. Entry point: `ovos-padacioso-domain-pipeline`. Subclasses the flat `PadaciosoPipeline`; the only differences are the container shape (below) and that intents are routed to a domain == `skill_id` at registration time.
+* **`DomainIntentContainer`** — the parallel-argmax variant of `IntentContainer` used internally by that pipeline.
+
+A separate entry point (rather than a config flag on the flat pipeline) keeps the two pipelines independently selectable in `default_pipeline` ordering and lets each have its own `intents.<key>` config block.
+
+## Enabling
+
+Add it to your OVOS config and place it in your pipeline order alongside (or in place of) the flat plugin:
+
+```json
+{
+  "intents": {
+    "ovos-padacioso-domain-pipeline": {
+      "fuzz": true
+    },
+    "pipeline": [
+      "ovos-padacioso-domain-pipeline-high",
+      "ovos-padacioso-domain-pipeline-medium",
+      "ovos-padacioso-domain-pipeline-low"
+    ]
+  }
+}
+```
+
+Configuration keys are read from `intents.ovos_padacioso_domain_pipeline`. The pipeline accepts every key the flat plugin does (`fuzz`, `workers`, `conf_high`, `conf_med`, `conf_low`).
+
+## Domain container
+
+`DomainIntentContainer` groups intents into *domains* (one sub-container per domain) and evaluates every domain in parallel at inference time, returning the global argmax. This mirrors the parallel-argmax pattern shipped by sibling OVOS intent engines (adapt, `nebulento.DomainIntentContainer`, `ovos_padatious.DomainIntentContainer`, `palavreado.DomainIntentContainer`, `ovos_m2v_pipeline.DomainPrototypeIntentStore`).
+
+There is intentionally no top-level "router" container. Strict regex matching is the wrong tool for routing: paraphrases that don't match any router template would block the sub-stage from ever running, even when a domain has a perfect template hit. Parallel evaluation is strictly more permissive and — for padacioso — practically free.
+
+## Why grouped by domain
+
+Two concrete benefits over a flat container:
+
+1. **Targeted scoping** — `calc_intent(query, domain=...)` evaluates a single sub-container, useful for session/context-driven scoping where the caller already knows the active domain.
+2. **Cheap prefilter** — domains whose templates share zero literal tokens with the utterance are skipped before any regex runs.
+3. **Short-circuit on decisive match** — padacioso hits literal templates at 0.95 confidence; the first such hit ends the scan.
+
+## Routing
+
+Padatious intents follow the convention `<skill_id>:<intent_name>`. The domain pipeline extracts the `skill_id` prefix from each registered intent label and uses it as the domain name. Labels without a `:` use the whole name as the domain (graceful fallback).
+
+```
+              utterance
+                 │
+                 ▼
+   ┌─────────────────────────────────┐
+   │  prefilter by literal tokens    │
+   └─────────────────────────────────┘
+                 │
+       candidate domains
+                 │
+                 ▼
+   ┌─────────────────────────────────┐
+   │ domains[d1].calc_intent(query)  │
+   │ domains[d2].calc_intent(query)  │  parallel argmax
+   │ domains[d3].calc_intent(query)  │
+   └─────────────────────────────────┘
+                 │
+        best by confidence
+                 │
+                 ▼
+           PadaciosoIntent
+```
+
+Every `padatious:register_intent` adds the templates to the domain's `IntentContainer` (`domains[skill_id]`) under the full `skill_id:intent` label.
+
+Entities are shared across domains: a `padatious:register_entity` adds the entity to every sub-container.
+
+`detach_intent` removes only the named intent from its domain; if the domain is left empty, the domain entry is dropped. `detach_skill` removes the whole domain in one shot.
+
+## Programmatic usage
+
+```python
+from padacioso import DomainIntentContainer
+
+d = DomainIntentContainer()
+d.register_domain_intent("media", "play",
+                          ["play {song}", "put on {song}"])
+d.register_domain_intent("home", "lights_on",
+                          ["lights on", "turn on the lights"])
+
+d.calc_intent("play some jazz")
+# {'name': 'play', ..., 'conf': 0.95}
+```
+
+### Scoping to a single domain
+
+Pass `domain=...` to `calc_intent` to evaluate only that domain — useful for session/context-driven scoping where the caller already knows the active domain:
+
+```python
+d.calc_intent("play some jazz", domain="media")
+```
+
+### Top-K matches
+
+```python
+d.calc_intents("play some jazz", top_k=3)
+# [{'name': 'play', 'conf': 0.95, ...}, ...]
+```
+
+## See also
+
+- [Padacioso README](../README.md) — flat pipeline overview and template syntax.

padacioso/__init__.py

@@ -1,9 +1,9 @@
 import re
 from typing import List, Iterator, Optional
 
 import simplematch
 
 from ovos_spec_tools import expand, normalize_for_match
 
 
 def _normalize(text: str) -> str:
@@ -94,7 +94,7 @@
             _init_sm_word_type()
 
     @staticmethod
     def _get_fuzzed(sample: str) -> List[str]:
         fuzzed = []
         words = sample.split(" ")
         for idx in range(len(words)):
@@ -113,7 +113,7 @@
             if w != "*" and "{" not in w and "}" not in w
         )
 
     def add_intent(self, name: str, lines: List[str]):
         """
         Add an intent with examples.
         @param name: name of intent to add
@@ -162,7 +162,7 @@
                 self._fuzz_variants.pop(rx, None)
             self._cache_dirty = True  # Mark cache as needing rebuild
 
     def add_entity(self, name: str, lines: List[str]):
         """
         Add an entity with examples.
         @param name: name of entity to add
@@ -208,10 +208,10 @@
             if any(_kw_hit(s) for s in samples):
                 excluded_intents.append(intent_name)
         for intent_name, contexts in self.required_contexts.items():
             if intent_name not in self.available_contexts:
                 excluded_intents.append(intent_name)
             elif any(context not in self.available_contexts[intent_name] for context in contexts):
                 excluded_intents.append(intent_name)
         for intent_name, contexts in self.excluded_contexts.items():
             if intent_name not in self.available_contexts:
                 continue
@@ -436,3 +436,8 @@
     """
     regex = r"[a-zA-Z0-9]+"
     simplematch.register_type("word", regex)
+
+
+# Re-export DomainIntentContainer at the package root for parity with
+# nebulento and ovos-padatious.
+from padacioso.domain_engine import DomainIntentContainer  # noqa: E402, F401

padacioso/domain_engine.py

@@ -0,0 +1,176 @@
+"""Domain-aware intent container for hierarchical intent organisation.
+
+Mirrors the parallel-argmax design used by sibling intent engines
+(adapt, nebulento, ovos_padatious, palavreado): intents are grouped into
+*domains*, and at query time **every** domain's sub-container is
+evaluated in parallel; the global best confidence wins.
+
+There is intentionally no top-level "router" container. Strict regex
+matching (padacioso) is the wrong tool for routing: paraphrases that
+don't match any router template would block the sub-stage from ever
+running, even when a domain sub-container has a perfect template hit.
+Parallel evaluation is strictly more permissive and, for padacioso,
+practically free (regex matching is fast and we additionally prefilter
+domains by literal-token overlap and short-circuit on the first 0.95
+hit).
+"""
+
+import re
+from typing import Dict, List, Optional, Set
+
+from padacioso import IntentContainer, _normalize as normalize_example
+
+
+_TOKEN_RE = re.compile(r"[A-Za-z0-9_]+")
+
+
+def _literal_tokens(line: str) -> Set[str]:
+    """Return the set of literal word tokens in a padacioso template line.
+
+    Entity placeholders (``{name}``), wildcards (``*``) and bracketed
+    alternations are stripped so only fixed surface tokens remain.
+    """
+    try:
+        text = normalize_example(line)
+    except Exception:
+        text = line
+    # Drop entity placeholders
+    text = re.sub(r"\{[^}]*\}", " ", text)
+    # Drop bracket-alternation syntax (kept tokens are still extracted by regex)
+    text = text.replace("(", " ").replace(")", " ").replace("|", " ")
+    text = text.replace("*", " ")
+    return {t.lower() for t in _TOKEN_RE.findall(text)}
+
+
+class DomainIntentContainer:
+    """Parallel-argmax intent engine across per-domain sub-containers.
+
+    Intents are grouped into *domains*. At query time the engine asks
+    every sub-container to match and returns the global best.
+
+    Example::
+
+        from padacioso import DomainIntentContainer
+
+        d = DomainIntentContainer()
+        d.register_domain_intent("media", "play",
+                                  ["play {song}", "put on {song}"])
+        d.register_domain_intent("home", "lights_on",
+                                  ["lights on", "turn on the lights"])
+
+        result = d.calc_intent("play some jazz")
+        # result["name"] == "play"
+
+    Args:
+        fuzz: Forwarded to every :class:`IntentContainer` created
+            internally.  When ``True`` partial matching is enabled.
+        n_workers: Forwarded to every internal :class:`IntentContainer`.
+    """
+
+    #: Confidence at/above which the first matching domain short-circuits.
+    _GOOD_ENOUGH = 0.95
+
+    def __init__(self, fuzz: bool = False, n_workers: int = 4) -> None:
+        self.fuzz = fuzz
+        self.n_workers = n_workers
+        #: Per-domain intent containers, keyed by domain name.
+        self.domains: Dict[str, IntentContainer] = {}
+        #: Literal-token vocabulary per domain (for cheap prefilter).
+        self._domain_vocab: Dict[str, Set[str]] = {}
+
+    # ── domain management ──────────────────────────────────────────────────
+
+    def remove_domain(self, domain_name: str) -> None:
+        """Remove a domain and all its intents."""
+        self.domains.pop(domain_name, None)
+        self._domain_vocab.pop(domain_name, None)
+
+    # ── intent management ──────────────────────────────────────────────────
+
+    def register_domain_intent(self, domain_name: str, intent_name: str,
+                                lines: List[str]) -> None:
+        """Register an intent inside a domain.
+
+        Creates the domain's :class:`IntentContainer` on first use.
+
+        Args:
+            domain_name: Target domain (created if it does not exist).
+            intent_name: Unique intent name within the domain.
+            lines: Padacioso template lines for the intent.
+        """
+        if domain_name not in self.domains:
+            self.domains[domain_name] = IntentContainer(
+                fuzz=self.fuzz, n_workers=self.n_workers
+            )
+            self._domain_vocab[domain_name] = set()
+        self.domains[domain_name].add_intent(intent_name, lines)
+        for line in lines:
+            self._domain_vocab[domain_name] |= _literal_tokens(line)
+
+    def remove_domain_intent(self, domain_name: str, intent_name: str) -> None:
+        """Remove an intent from a domain."""
+        if domain_name in self.domains:
+            self.domains[domain_name].remove_intent(intent_name)
+            # Vocabulary is not pruned per-intent (cheap over-approximation
+            # only widens the prefilter; correctness is preserved).
+
+    # ── query API ──────────────────────────────────────────────────────────
+
+    def _candidate_domains(self, query: str) -> List[str]:
+        """Cheap literal-token prefilter: keep domains whose vocab overlaps."""
+        utt_tokens = {t.lower() for t in _TOKEN_RE.findall(query)}
+        if not utt_tokens:
+            return list(self.domains.keys())
+        candidates = []
+        for name, vocab in self._domain_vocab.items():
+            # Empty vocab (only entity/wildcard templates) -> can't prefilter.
+            if not vocab or utt_tokens & vocab:
+                candidates.append(name)
+        return candidates
+
+    def calc_intent(self, query: str,
+                     domain: Optional[str] = None) -> Optional[dict]:
+        """Return the global best intent match for *query*.
+
+        Args:
+            query: The utterance to match.
+            domain: If given, evaluate only inside this domain.
+
+        Returns:
+            The match dict from the winning domain's container, or ``None``.
+        """
+        if domain is not None:
+            sub = self.domains.get(domain)
+            return sub.calc_intent(query) if sub is not None else None
+
+        best: Optional[dict] = None
+        best_conf = 0.0
+        for name in self._candidate_domains(query):
+            sub = self.domains[name]
+            match = sub.calc_intent(query)
+            if not match or not match.get("name"):
+                continue
+            conf = match.get("conf", 0) or 0
+            if conf > best_conf:
+                best = match
+                best_conf = conf
+                # padacioso hits literal templates at 0.95; first such match
+                # is decisive enough to skip remaining domains.
+                if best_conf >= self._GOOD_ENOUGH:
+                    break
+        return best
+
+    def calc_intents(self, query: str, top_k: int = 5) -> List[dict]:
+        """Return the top-K intent matches across all domains.
+
+        Each sub-container contributes its best match; results are sorted
+        by confidence descending and truncated to ``top_k``.
+        """
+        results: List[dict] = []
+        for name in self._candidate_domains(query):
+            sub = self.domains[name]
+            for match in sub.calc_intents(query):
+                if match and match.get("name"):
+                    results.append(match)
+        results.sort(key=lambda m: m.get("conf", 0) or 0, reverse=True)
+        return results[:top_k]