orchestrata_ultimate.orchestrator.intent_normalizer

Reference › orchestrata_ultimate › orchestrator › intent_normalizer

The first stage of natural-language classification: it cleans a raw user turn into one plain sentence before the grammar tier of intent_classifier (its only importer) sees it. It is traced to spec 003 story US-001 (REQ-003-US-001-060 to 062).

IntentNormalizer.normalize is asynchronous. An empty input is returned as is. Otherwise it looks up a key built by cache_key over the text and any context turns in an LRU cache from core.lru_cache, and on a miss calls the language model through llm.client.call_llm with a fixed system prompt, a user message built by build_user_message and temperature 0. The client is imported inside the method so callers that never normalize do not load the provider stack. The model name and maximum reply length come from the environment variables INTENT_NORMALIZER_MODEL and INTENT_NORMALIZER_MAX_CHARS, with defaults in the module.

The stage is designed never to block classification. Any exception from the model is logged and the original text is returned. clean_reply takes the first non-empty line of the reply, strips quotes and whitespace, and falls back to the original if the reply is empty, longer than the limit, or not grounded. Grounding means the reply shares at least one word of three or more characters with the user's turn or its context turns; the docstring records that this check was added after a degenerate reply had been accepted. stats reports the counters for calls, cache hits and pass-throughs.

At a glance

Class diagram

Dependency diagram

API

Classes

IntentNormalizer

Decorators: traces(spec='003', story='US-001', reqs=['REQ-003-US-001-060', 'REQ-003-US-001-061', 'REQ-003-US-001-062']) · Line 147

Normalize a raw user turn into one clean sentence for the DCG tier.

Source

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

"""Haiku normalization stage — clean the sentence BEFORE Prolog sees it.

Spec: 003-intent / US-001 extension (REQ-003-US-001-060..064). Design:
plans/intent-dcg-accuracy-design.md §5.

WHY THIS EXISTS, measured rather than argued. The DCG grammar in
``prolog/intent_grammar.pl`` is a keyword-driven parser: it matches stems
against a lexicon. Real people do not type lexicon words. Against the 76 real
user turns in ``tests/orchestrator/data/intent_corpus_raw_turns.json``, mined
from 14,978 captured conversation turns, the grammar scored:

    raw user text            24/76 = 31.6%
    rephrased user text      29/76 = 38.2%    <- the headroom
    THIS normalizer          29/76 = 38.2%    <- at it

The gap is spelling and phrasing, not vocabulary. "Why is my test suite runnig
in 124v mins, did yiou forget the threads again?" parses as ANALYSIS on the raw
text (the interrogative `why` wins) and as TEST once it reads "run the test
suite". "check k3 work" parses as nothing at all until it reads "review the
work".

Those raw/normalizer numbers were 19/76 and 26/92 before two changes landed
together, and they are only comparable at the same grammar version: the lexicon
now stems both sides of a match, and work_type_of/3 resolves by majority vote
rather than by position. The vote is what makes THIS prompt's long, faithful
rewrites beat the terse ones — see the note above the prompt, where four
variants and two of my own refuted rules are recorded with their numbers.

29/76 is itself optimistic: the agent that produced those rewrites also
produced the labels they are scored against. Prompt changes here must be
justified by re-measuring against that corpus, not by reasoning about what
reads better — three rules that read better have now been measured and
removed.

So a small, fast model normalizes first and the grammar always parses clean
input. The division of labour is the point: the LLM does what LLMs are good
at — messy human text, typos, pronouns, "the damn thing" — and Prolog keeps
what only it gives you, a deterministic and explainable verdict that the
governance layer can reason about.

FOUR PROPERTIES THIS MUST HAVE, all of them because normalization sits on the
hot path of every intent:

1. NEVER BLOCKS. Any failure — no provider, rate limit, timeout, empty reply
   — returns the original text. A normalizer outage degrades accuracy back to
   the 31.6% raw baseline; it never fails a classification.
2. CACHED. Identical (text, context) returns the identical normalization
   without a second call, so the same turn replayed through the audit path
   cannot produce a different verdict, and a retry costs nothing.
3. TRANSPARENT. The normalized sentence is returned to the caller and stored
   on the ClassifiedIntent, so a verdict can always be reconstructed from
   what the grammar actually read rather than from what the user typed.
4. BOUNDED. One sentence in, one sentence out. The model is instructed not to
   summarise, and output longer than the configured ceiling is rejected in
   favour of the original — an elaborating model is a worse input than a
   messy one.
"""

from __future__ import annotations

import hashlib
import logging
import os
import re
from collections.abc import Mapping, Sequence

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.lru_cache import LRUCache

logger = logging.getLogger(__name__)

# The model is told what the output is FOR. Withholding that (an earlier cut
# of this prompt did) buys nothing and costs accuracy: it produced
# "the rtm shows ghost rows." from "why is the rtm showing ghost rows",
# deleting the interrogative that carried the whole classification.
#
# It is NOT given the WorkType/IntentTarget taxonomy. Not on principle — the
# reason is practical. A model handed the categories starts writing the
# category name into the sentence, which turns a normalizer into a classifier
# with none of a classifier's evaluation, and the Prolog tier's verdict stops
# being reconstructable from the user's own words.
# Chosen by measurement over tests/orchestrator/data/intent_corpus_raw_turns.json,
# against three rivals and the raw text itself. Full corpus, n=76:
#
#     raw user text                          24/76
#     THIS prompt                            29/76      <- the 29/76 ceiling
#
# (28/76 measured before the grounding check in clean_reply, 29/76 with it. One
# turn, so read it as "grounding costs nothing", not as a gain.)
#
# and on the 20-turn stride sample the rivals were measured against:
#
#     raw user text                           8/20
#     previous shipped prompt                 8/20      <- bought nothing
#     terse imperative (<=10 words)           7/20      <- WORSE than raw
#     "<verb> the <thing>" (avg 4.0 words)    8/20
#     THIS prompt (avg 13.9 words)           11/20
#
# TWO OF MY OWN RULES DIED HERE, and the order matters.
#
# The previous prompt ended "Do NOT name a category of work" — a blindfold I
# invented so the model could not become the classifier. It cost the entire
# gain: 8/20, exactly the raw baseline.
#
# I then predicted terse imperatives would win, reasoning that a leftmost-wins
# grammar sees fewer decoys in a short sentence. Measured, terse is the WORST
# variant of the five. The reasoning was stale: work_type_of/3 resolves by
# MAJORITY VOTE now (REQ-003-US-001-065), so compressing a sentence to a single
# decisive word leaves the vote nothing to count. Naming the kind of work and
# KEEPING every mention of it is what feeds the vote — hence the long average.
#
# One further variant added "Do NOT add facts the user did not state" back on
# top of this text and scored 10/20 against this one's 11/20. One point on
# twenty turns is weak evidence, but a rule has to earn its place and that one
# did not; the containment is structural instead — ClassifiedIntent keeps
# .normalized_text beside the raw turn, clean_reply/2 rejects an over-long
# reply, and an invented fact can only move the verdict if it happens to be a
# word in the bounded lexicon.
#
# Re-measure before editing this string. Every rule in it that sounded right
# and was not is listed above.
NORMALIZER_SYSTEM_PROMPT = """You rewrite a developer's message into one \
sentence for a parser.

Rules:
- Fix spelling and grammar.
- Resolve "it", "this", "that" against the two preceding turns.
- Name the KIND of work being asked for using a plain word: fix, bug, test,
  document, optimize, slow, refactor, add, deploy, review, design.
- Name the component or file the work is about.
- A complaint or a question about something broken IS a request to fix it.
- Keep every work-kind word the user already used. If the user named the same
  kind of work more than once, keep every mention -- do not compress them.
- Output the sentence only. No preamble, no quotes, no explanation."""

_DEFAULT_MODEL = "haiku"
_DEFAULT_MAX_CHARS = 300
_DEFAULT_CACHE_SIZE = 2048
_DEFAULT_CACHE_TTL_SECONDS = 3600.0


@traces(
    spec="003",
    story="US-001",
    reqs=["REQ-003-US-001-060", "REQ-003-US-001-061", "REQ-003-US-001-062"],
)
class IntentNormalizer:
    """Normalize a raw user turn into one clean sentence for the DCG tier.

    Construction is cheap and side-effect free; the provider is resolved on
    first use, so an unconfigured environment costs nothing until a
    normalization is actually attempted (and then degrades to passthrough).
    """

    def __init__(
        self,
        model: str | None = None,
        max_chars: int | None = None,
        cache: LRUCache[str] | None = None,
        env: Mapping[str, str] | None = None,
    ) -> None:
        environ = env if env is not None else os.environ
        self.model = model or environ.get("INTENT_NORMALIZER_MODEL", _DEFAULT_MODEL)
        self.max_chars = max_chars or int(
            environ.get("INTENT_NORMALIZER_MAX_CHARS", str(_DEFAULT_MAX_CHARS))
        )
        self._cache: LRUCache[str] = cache or LRUCache(
            max_size=_DEFAULT_CACHE_SIZE, default_ttl=_DEFAULT_CACHE_TTL_SECONDS
        )
        self._calls = 0
        self._cache_hits = 0
        self._passthroughs = 0

    # ── cache key ───────────────────────────────────────────────────────
    @staticmethod
    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-064"])
    def cache_key(text: str, context: Sequence[str] | None) -> str:
        """Stable key over the exact input the model would see.

        Hashed rather than concatenated because a turn can be long and the
        cache is bounded by entry count, not by bytes.
        """
        digest = hashlib.sha256()
        digest.update(text.encode("utf-8"))
        for prior in context or ():
            digest.update(b"\x00")
            digest.update(prior.encode("utf-8"))
        return digest.hexdigest()

    # ── prompt ──────────────────────────────────────────────────────────
    @staticmethod
    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-060"])
    def build_user_message(text: str, context: Sequence[str] | None) -> str:
        """Render the turn plus its preceding context for the model."""
        if not context:
            return text
        lines = ["Preceding turns, oldest first:"]
        lines.extend(f"- {prior}" for prior in context)
        lines.append("")
        lines.append("Message to rewrite:")
        lines.append(text)
        return "\n".join(lines)

    # ── result shaping ──────────────────────────────────────────────────
    @traces(
        spec="003",
        story="US-001",
        reqs=["REQ-003-US-001-061", "REQ-003-US-001-067"],
    )
    def clean_reply(
        self, reply: str, original: str, context: Sequence[str] | None = None
    ) -> str:
        """Reduce a model reply to one sentence, or fall back to ``original``.

        A model that answers with a paragraph, a preamble, or an empty string
        has failed the contract, and its output is a worse parser input than
        the user's own text.

        The last check is GROUNDING, and it was added after a live failure
        rather than in anticipation. This method used to accept any non-empty,
        short-enough string as "the normalization", which means the tier
        trusted a model to have rewritten the user's sentence instead of
        verifying it. Under the suite's cassette-synth seam the model returns
        canned text that has nothing to do with the input, and
        "fix the crash in /tmp/x_widget.py" — unambiguous FIX work on the raw
        string — came back through the normalizer classifying as `feature` at
        0.30 confidence, below the dispatch threshold. The synth reply was a
        test artifact; ACCEPTING it was not. Any degenerate reply (a refusal,
        an answer to the wrong question, a truncated stream) fails the same way
        in production, and silently, because the raw turn still looks fine in
        the audit trail.

        A rewrite of the user's words shares words with them. The check is
        therefore: at least one token of three or more characters in common
        with what the user actually wrote — the turn AND the context turns,
        since resolving "it" against a preceding turn legitimately imports
        every content word from there and none from the turn itself. Zero
        overlap means the model answered a different question.
        """
        if not reply:
            return original
        first_line = next(
            (line.strip() for line in reply.splitlines() if line.strip()), ""
        )
        candidate = " ".join(first_line.split()).strip().strip('"').strip("'")
        if not candidate or len(candidate) > self.max_chars:
            return original
        if not self._is_grounded(candidate, original, context):
            return original
        return candidate

    @staticmethod
    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-067"])
    def _tokens(text: str) -> set[str]:
        """Case-folded word tokens of three or more characters.

        Three is the floor because shorter tokens ("a", "it", "of", "is") are
        function words shared by almost any pair of English sentences, so
        counting them would make the grounding check pass on everything.
        """
        return {
            token for token in re.findall(r"[a-z0-9_]+", text.lower()) if len(token) > 2
        }

    @classmethod
    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-067"])
    def _is_grounded(
        cls, candidate: str, original: str, context: Sequence[str] | None
    ) -> bool:
        """True when the rewrite shares vocabulary with what the user wrote."""
        source = cls._tokens(original)
        for prior in context or ():
            source |= cls._tokens(prior)
        if not source:
            # The user typed nothing a token check can see (punctuation, emoji,
            # CJK). Nothing to ground against, so do not reject on that basis —
            # the length and emptiness checks above still apply.
            return True
        return bool(cls._tokens(candidate) & source)

    # ── the call ────────────────────────────────────────────────────────
    @traces(spec="003", story="US-001", reqs=["REQ-003-US-001-060"])
    async def normalize(self, text: str, context: Sequence[str] | None = None) -> str:
        """Return a cleaned sentence, or ``text`` unchanged on any failure."""
        if not text or not text.strip():
            return text

        key = self.cache_key(text, context)
        cached = self._cache.get(key)
        if cached is not None:
            self._cache_hits += 1
            return cached

        # Imported here, not at module scope: importing the LLM client pulls
        # in the whole provider stack, and a caller that never normalizes
        # (no provider configured, tests exercising the grammar directly)
        # should not pay for it.
        from orchestrata_ultimate.llm.client import call_llm

        try:
            self._calls += 1
            reply = await call_llm(
                system_prompt=NORMALIZER_SYSTEM_PROMPT,
                user_message=self.build_user_message(text, context),
                model=self.model,
                temperature=0.0,
            )
        except Exception as exc:  # noqa: BLE001 — degrade, never block
            self._passthroughs += 1
            logger.warning(
                "intent_normalizer_unavailable: %s: %s — classifying the raw "
                "text instead",
                type(exc).__name__,
                exc,
            )
            return text

        normalized = self.clean_reply(reply, text, context)
        if normalized is text or normalized == text:
            self._passthroughs += 1
        self._cache.put(key, normalized)
        return normalized

    # ── observability ───────────────────────────────────────────────────
    def stats(self) -> dict[str, int]:
        """Counters for the normalization stage.

        ``passthroughs`` is the one to watch: a rising passthrough rate with a
        flat call count means the provider is failing and classification has
        quietly reverted to raw-text accuracy.
        """
        return {
            "calls": self._calls,
            "cache_hits": self._cache_hits,
            "passthroughs": self._passthroughs,
        }