pyrtan / sourcev0.19.0 · main

pyrtan / src / pyrtan / intent.py

15,815 bytes · last changed 2026-10-25 in Read sentences with rules: rewrites, typed slots, synonyms, typos, pyrtan match

1"""Triggers: how a sentence finds the tree that answers it.
2
3A trigger is a pattern with slots::
4
5 "weather in {city}"
6 "top {count:int} stories on hacker news"
7 "is {host:host} up"
8 "what's [the] latest release of {repo}"
9 "remind me {when:date} at {at:time}"
10
11A slot is ``{name}`` (any text) or ``{name:kind}``. The kinds: ``int``,
12``float``, ``word`` (no spaces), ``text``; and the read kinds, each with a
13fixed parser in :mod:`pyrtan.nlp`: ``number`` (digits or words),
14``date``, ``time``, ``duration``, ``host``, ``url``, ``email``. Square
15brackets mark optional words, round brackets a required choice, and ``|``
16separates alternatives inside either: ``"how (hot|cold) is it in {place}"``.
17
18Before matching, the sentence and the pattern both go through
19:func:`pyrtan.nlp.clean`: unicode folded, greetings and politeness dropped
20from the edges, contractions expanded, number words turned to digits.
21Matching is case-insensitive and a literal word matches its plural or
22singular. A tree may add ``synonyms`` and turn on ``typos`` in its
23``[language]`` table; both are applied here, and every rule that fires is
24reported in the :class:`Match` so the trace can show it.
25
26When several triggers match, the most specific wins: the one with the most
27literal text. A tie is reported as ambiguity rather than resolved by luck.
28"""
29
30from __future__ import annotations
31
32import datetime as _dt
33import re
34from dataclasses import dataclass, field
35from typing import Any
36
37from . import nlp
38
39KINDS = ("int", "float", "word", "text", "number", "date", "time", "duration", "host", "url", "email")
40SLOT = re.compile(r"\{([A-Za-z_][A-Za-z0-9_]*)(?::([a-z]+))?\}")
41SLOT_RE = {
42 "int": r"-?\d+",
43 "float": r"-?\d+(?:\.\d+)?",
44 "number": r"-?\d+(?:[.,]\d+)*",
45 "word": r"\S+",
46 "text": r".+?",
47 "date": r".+?",
48 "time": r"\S+(?:\s*[ap]m)?",
49 "duration": r".+?",
50 "host": r"\S+",
51 "url": r"\S+",
52 "email": r"\S+",
53 None: r".+?",
54}
55# The input type each slot kind produces, for the checker.
56PRODUCES = {"int": "int", "float": "float", "number": "number", "word": "str", "text": "str", "date": "str", "time": "str", "duration": "int", "host": "str", "url": "str", "email": "str", None: "str"}
57TYPO_MIN_LENGTH = 5
58
59
60class TriggerError(ValueError):
61 pass
62
63
64def normalise(text: str) -> str:
65 """The light form: trimmed, one space between words. Used where a
66 sentence is kept as a person wrote it (the unanswered log)."""
67 text = text.strip()
68 text = re.sub(r"^[\s\W_]+|[\s?!.,;:]+$", "", text)
69 text = re.sub(r"\s+", " ", text)
70 return text
71
72
73@dataclass
74class Language:
75 """A tree's ``[language]`` table: what it adds to the fixed rules."""
76
77 synonyms: dict[str, list[str]] = field(default_factory=dict)
78 typos: bool = False
79
80 @classmethod
81 def from_raw(cls, raw: Any, where: str = "language") -> "Language":
82 if raw is None:
83 return cls()
84 if not isinstance(raw, dict):
85 raise TriggerError(f"{where} must be a table")
86 syn = raw.get("synonyms", {})
87 if not isinstance(syn, dict):
88 raise TriggerError(f"{where}.synonyms must be a table of word = [words]")
89 synonyms: dict[str, list[str]] = {}
90 for word, alts in syn.items():
91 if isinstance(alts, str):
92 alts = [alts]
93 if not isinstance(alts, list) or not all(isinstance(a, str) and a.strip() for a in alts):
94 raise TriggerError(f"{where}.synonyms.{word} must be a list of words")
95 if not re.fullmatch(r"[A-Za-z][A-Za-z'-]*", word):
96 raise TriggerError(f"{where}.synonyms: {word!r} is not a word")
97 synonyms[word.lower()] = [a.strip().lower() for a in alts]
98 typos = raw.get("typos", False)
99 if not isinstance(typos, bool):
100 raise TriggerError(f"{where}.typos must be true or false")
101 for key in raw:
102 if key not in ("synonyms", "typos"):
103 raise TriggerError(f"{where}: unknown key {key!r}; one of synonyms, typos")
104 return cls(synonyms, typos)
105
106 def to_dict(self) -> dict[str, Any]:
107 out: dict[str, Any] = {}
108 if self.synonyms:
109 out["synonyms"] = self.synonyms
110 if self.typos:
111 out["typos"] = True
112 return out
113
114
115@dataclass
116class Trigger:
117 pattern: str
118 fixed_inputs: dict[str, Any] = field(default_factory=dict)
119 language: Language = field(default_factory=Language)
120 regex: re.Pattern[str] = field(init=False, repr=False)
121 slots: dict[str, str | None] = field(init=False, default_factory=dict)
122 specificity: int = field(init=False, default=0)
123 cleaned: str = field(init=False, default="")
124 vocabulary: set[str] = field(init=False, default_factory=set, repr=False)
125 edges: tuple[str, str] = field(init=False, default=("", ""), repr=False) # first and last literal word
126
127 def __post_init__(self) -> None:
128 if isinstance(self.language, dict):
129 self.language = Language.from_raw(self.language)
130 # The pattern gets the rewrites a sentence gets, minus the ones that
131 # only make sense for a sentence (fillers, number words): folded,
132 # contractions expanded, so "[what's the]" and "what is the" agree.
133 self.cleaned = "".join(part if part.startswith("{") else nlp.expand_contractions(part)[0] for part in re.split(r"(\{[^}]*\})", nlp.fold(self.pattern)))
134 self.regex, self.slots, self.specificity, self.vocabulary, self.edges = compile_pattern(self.cleaned, self.language)
135
136 def match(self, text: str, *, now: _dt.datetime | _dt.date | None = None, cleaned: bool = False) -> dict[str, Any] | None:
137 """The inputs this trigger reads from ``text``, or None. ``cleaned``
138 says the text already went through :func:`nlp.clean`."""
139 got = self.match_detail(text, now=now, cleaned=cleaned)
140 return got.inputs if got is not None else None
141
142 def match_detail(self, text: str, *, now: _dt.datetime | _dt.date | None = None, cleaned: bool = False) -> "Match | None":
143 sentence = text if cleaned else nlp.clean(text)[0]
144 notes: list[str] = []
145 if self.language.typos:
146 sentence, fixed = correct_typos(sentence, self.vocabulary)
147 notes.extend(f"typo: {a}->{b}" for a, b in fixed)
148 m = self.regex.fullmatch(sentence)
149 if not m:
150 return None
151 inputs: dict[str, Any] = dict(self.fixed_inputs)
152 for name, kind in self.slots.items():
153 raw = m.group(name)
154 if raw is None:
155 continue
156 try:
157 inputs[name] = read_slot(kind, raw.strip(), now)
158 except (ValueError, nlp.ParseError) as e:
159 notes.append(f"slot {name}: {e}")
160 return Match("", self, {}, self.specificity, notes, rejected=True)
161 if kind in nlp.PARSERS and str(inputs[name]) != raw.strip():
162 notes.append(f"{name}: {raw.strip()!r} read as {inputs[name]!r}")
163 return Match("", self, inputs, self.specificity, notes)
164
165
166def read_slot(kind: str | None, raw: str, now: _dt.datetime | _dt.date | None) -> Any:
167 if kind == "int":
168 return int(raw)
169 if kind == "float":
170 return float(raw)
171 if kind in (None, "word", "text"):
172 return raw
173 parser = nlp.PARSERS[kind]
174 return parser(raw, now) if kind == "date" else parser(raw)
175
176
177def correct_typos(sentence: str, vocabulary: set[str]) -> tuple[str, list[tuple[str, str]]]:
178 """Replace a word that is one edit from exactly one vocabulary word
179 (and is not itself one), for words of :data:`TYPO_MIN_LENGTH` letters
180 or more. Returns the sentence and what changed."""
181 if not vocabulary:
182 return sentence, []
183 fixed: list[tuple[str, str]] = []
184
185 def one(m: re.Match[str]) -> str:
186 word = m.group(0)
187 low = word.lower()
188 if len(low) < TYPO_MIN_LENGTH or low in vocabulary or any(low in nlp.forms(v) for v in vocabulary):
189 return word
190 close = [v for v in vocabulary if len(v) >= TYPO_MIN_LENGTH and nlp.distance(low, v, 1) <= 1]
191 if len(close) != 1:
192 return word
193 fixed.append((word, close[0]))
194 return close[0]
195
196 return re.sub(r"[A-Za-z]+", one, sentence), fixed
197
198
199def _word_regex(word: str, language: Language) -> str:
200 """A literal word matches itself, its singular or plural, and the same
201 for each of its synonyms."""
202 spellings = set(nlp.forms(word))
203 phrases: list[str] = []
204 for syn in language.synonyms.get(word.lower(), []):
205 if " " in syn:
206 phrases.append(r"\s+".join(re.escape(p) for p in syn.split()))
207 else:
208 spellings.update(nlp.forms(syn))
209 alternatives = [re.escape(s) for s in sorted(spellings, key=lambda s: (-len(s), s))] + phrases
210 return "(?:" + "|".join(alternatives) + ")"
211
212
213def compile_pattern(pattern: str, language: Language | None = None) -> tuple[re.Pattern[str], dict[str, str | None], int, set[str], tuple[str, str]]:
214 language = language or Language()
215 if not pattern or not pattern.strip():
216 raise TriggerError("a trigger pattern cannot be empty")
217 slots: dict[str, str | None] = {}
218 out: list[str] = []
219 vocabulary: set[str] = set()
220 ordered: list[str] = []
221 literal_chars = 0
222 pos = 0
223 depth = 0
224 text = pattern.strip().lower()
225 word: list[str] = []
226
227 def flush_word() -> None:
228 if word:
229 w = "".join(word)
230 out.append(_word_regex(w, language))
231 vocabulary.add(w)
232 ordered.append(w)
233 for syn in language.synonyms.get(w, []):
234 vocabulary.update(syn.split())
235 word.clear()
236
237 while pos < len(text):
238 ch = text[pos]
239 if ch.isalpha() or (word and ch in "'-" and pos + 1 < len(text) and text[pos + 1].isalpha()):
240 word.append(ch)
241 literal_chars += 1
242 pos += 1
243 continue
244 flush_word()
245 if ch == "{":
246 m = SLOT.match(text, pos)
247 if not m:
248 raise TriggerError(f"bad slot in trigger {pattern!r} at {pos}")
249 name, kind = m.group(1), m.group(2)
250 if kind is not None and kind not in KINDS:
251 raise TriggerError(f"slot {{{name}:{kind}}} in {pattern!r}: kind must be one of {', '.join(KINDS)}")
252 if name in slots:
253 raise TriggerError(f"slot {name!r} appears twice in {pattern!r}")
254 slots[name] = kind
255 out.append(f"(?P<{name}>{SLOT_RE[kind]})")
256 pos = m.end()
257 elif ch == "[":
258 depth += 1
259 out.append(r"(?:\s*")
260 pos += 1
261 elif ch == "]":
262 if depth == 0:
263 raise TriggerError(f"unbalanced ] in trigger {pattern!r}")
264 depth -= 1
265 out.append(r")?")
266 pos += 1
267 elif ch == "(":
268 depth += 1
269 out.append(r"(?:")
270 pos += 1
271 elif ch == ")":
272 if depth == 0:
273 raise TriggerError(f"unbalanced ) in trigger {pattern!r}")
274 depth -= 1
275 out.append(r")")
276 pos += 1
277 elif ch == "|" and depth > 0:
278 out.append("|")
279 pos += 1
280 elif ch.isspace():
281 while pos < len(text) and text[pos].isspace():
282 pos += 1
283 # Space is optional around optional groups and slots so that
284 # "[the] weather" matches "weather" without a leading space.
285 if out and out[-1] in (r"(?:\s*", r")?", r"(?:", r")", "|"):
286 out.append(r"\s*")
287 elif pos < len(text) and text[pos] in "[(|":
288 out.append(r"\s*")
289 else:
290 out.append(r"\s+")
291 else:
292 out.append(re.escape(ch))
293 literal_chars += 1
294 pos += 1
295 flush_word()
296 if depth != 0:
297 raise TriggerError(f"unbalanced [ in trigger {pattern!r}")
298 if literal_chars == 0:
299 raise TriggerError(f"trigger {pattern!r} has no literal words; it would match anything")
300 try:
301 regex = re.compile("".join(out), re.IGNORECASE)
302 except re.error as e:
303 raise TriggerError(f"trigger {pattern!r} does not compile: {e}") from e
304 return regex, slots, literal_chars, vocabulary, (ordered[0], ordered[-1]) if ordered else ("", "")
305
306
307@dataclass
308class Match:
309 tree: str
310 trigger: Trigger
311 inputs: dict[str, Any]
312 specificity: int
313 notes: list[str] = field(default_factory=list)
314 rejected: bool = False
315
316
317@dataclass
318class Reading:
319 """How a sentence was read: the rewrites, and every trigger's verdict."""
320
321 text: str
322 cleaned: str
323 steps: list[str]
324 matches: list[Match] # the winners: one, or several at a tie
325 considered: list[Match] # every trigger that matched, or matched and then refused a slot
326 now: str | None = None
327
328 @property
329 def winner(self) -> Match | None:
330 return self.matches[0] if len(self.matches) == 1 else None
331
332 def to_dict(self) -> dict[str, Any]:
333 return {
334 "text": self.text,
335 "cleaned": self.cleaned,
336 "steps": self.steps,
337 "now": self.now,
338 "matches": [_match_dict(m) for m in self.matches],
339 "considered": [_match_dict(m) for m in self.considered],
340 }
341
342 def step(self) -> dict[str, Any]:
343 """The ``intent`` step a trace starts with: what the sentence became
344 and which trigger took it."""
345 rewrite = f"{self.text!r}" + (f" -> {self.cleaned!r}" if self.cleaned != self.text else "")
346 if self.winner:
347 m = self.winner
348 summary = f"{rewrite}; {m.tree} via {m.trigger.pattern!r} ({m.specificity} literal chars)"
349 if m.notes:
350 summary += "; " + "; ".join(m.notes)
351 elif self.matches:
352 summary = f"{rewrite}; tie between {sorted({m.tree for m in self.matches})}"
353 else:
354 summary = f"{rewrite}; no trigger matched"
355 observed = {"cleaned": self.cleaned, "steps": self.steps}
356 if self.winner:
357 observed["trigger"] = self.winner.trigger.pattern
358 observed["inputs"] = self.winner.inputs
359 if self.winner.notes:
360 observed["notes"] = self.winner.notes
361 return {"node": "intent", "kind": "intent", "summary": summary, "observed": observed}
362
363
364def _match_dict(m: Match) -> dict[str, Any]:
365 return {"tree": m.tree, "trigger": m.trigger.pattern, "specificity": m.specificity, "inputs": m.inputs, "notes": m.notes, "rejected": m.rejected}
366
367
368def read(text: str, candidates: list[tuple[Any, Trigger]], now: _dt.datetime | _dt.date | None = None) -> Reading:
369 """Read ``text`` against every candidate and say what happened."""
370 keep = frozenset(w for _, trig in candidates for w in trig.edges if w)
371 cleaned, steps = nlp.clean(text, keep=keep)
372 considered: list[Match] = []
373 for tree, trig in candidates:
374 got = trig.match_detail(cleaned, now=now, cleaned=True)
375 if got is not None:
376 got.tree = tree
377 considered.append(got)
378 live = [m for m in considered if not m.rejected]
379 top = max((m.specificity for m in live), default=0)
380 matches = [m for m in live if m.specificity == top]
381 now_text = now.isoformat() if isinstance(now, (_dt.date, _dt.datetime)) else None
382 return Reading(text, cleaned, steps, matches, considered, now_text)
383
384
385def best_matches(text: str, candidates: list[tuple[Any, Trigger]], now: _dt.datetime | _dt.date | None = None) -> list[Match]:
386 """All matches at the highest specificity. One item is a clear winner;
387 several is ambiguity; none is no match."""
388 return read(text, candidates, now).matches