From d0b7c8a4bbd8668ee0db0a631b1b68d4b08285c6 Mon Sep 17 00:00:00 2001 From: Johnny Huynh <27847622+johnnyhuy@users.noreply.github.com> Date: Sun, 8 Nov 2026 11:22:43 +0900 Subject: [PATCH] feat(rules): add keyword-rule matcher with config loading Operator rules pin a canned answer to a question pattern. The matcher supports contains, exact and regex modes with priority ordering, loads from OPERATOR_RULES_JSON or OPERATOR_RULES_FILE, and skips invalid rules instead of breaking startup. RAGEngine.generate_response checks rules before retrieval and generation, so a matched question returns without a Pi job. Adds unit tests and operator docs. Part of #6 --- .env.example | 12 +++ backend/app/config.py | 6 ++ backend/app/rag.py | 19 +++- backend/app/rules.py | 180 ++++++++++++++++++++++++++++++++++++ backend/tests/test_rules.py | 163 ++++++++++++++++++++++++++++++++ docs/operator-rules.md | 72 +++++++++++++++ mise.toml | 2 +- 7 files changed, 451 insertions(+), 3 deletions(-) create mode 100644 backend/app/rules.py create mode 100644 backend/tests/test_rules.py create mode 100644 docs/operator-rules.md diff --git a/.env.example b/.env.example index 5c1a0dc..1a5d829 100644 --- a/.env.example +++ b/.env.example @@ -132,6 +132,18 @@ CHUNK_OVERLAP=200 # Retrieval settings RETRIEVAL_TOP_K=5 +# ============================================================================ +# Operator Rules (deterministic canned answers, checked before RAG) +# ============================================================================ + +# Inline JSON list of rules. Each rule needs an id, keywords and a response. +# Optional per rule: match_type (contains, exact or regex), priority, enabled. +# Example: [{"id":"vpn-reset","keywords":["reset vpn"],"response":"Use the IT portal to reset VPN access."}] +OPERATOR_RULES_JSON= + +# Or a path to a JSON file with the same list shape. Wins over the inline value. +OPERATOR_RULES_FILE= + # ============================================================================ # Application Configuration # ============================================================================ diff --git a/backend/app/config.py b/backend/app/config.py index 2220447..659b52f 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -65,5 +65,11 @@ class Settings(BaseSettings): CONVERSATION_MEMORY_WINDOW: int = 10 # Number of recent messages to include CONVERSATION_TIMEOUT_MINUTES: int = 30 # How long to consider a conversation active + # Operator Rules (deterministic canned answers, checked before RAG) + # Inline JSON list of rules, or a path to a JSON file with the same shape. + # The file wins when both are set. Empty means no rules. + OPERATOR_RULES_JSON: Optional[str] = None + OPERATOR_RULES_FILE: Optional[str] = None + settings = Settings() diff --git a/backend/app/rag.py b/backend/app/rag.py index 8d3d6fe..404ec8c 100644 --- a/backend/app/rag.py +++ b/backend/app/rag.py @@ -9,6 +9,7 @@ from app.database import ConversationHistory, SessionLocal from app.vector_store import vector_store from app import agent_runtime +from app import rules logger = logging.getLogger(__name__) @@ -110,6 +111,10 @@ def _agent_payload(self, question: str, channel_id=None, conversation_id=None, payload["slack"] = slack return payload + def match_operator_rule(self, question: str): + """Return the operator rule that fires for a question, if any.""" + return rules.rule_engine.match(question) + def generate_response(self, question: str, channel_id: str = None, conversation_id: Optional[str] = None, user_id: Optional[str] = None, @@ -117,9 +122,19 @@ def generate_response(self, question: str, channel_id: str = None, request_id: Optional[str] = None) -> Dict[str, Any]: """Retrieve context, then wait for a persisted Pi Durable answer. - Legacy PostgreSQL history remains available for audit; Pi owns new turns. - API callers should supply a request ID to retry an interrupted wait safely. + Operator rules run first, so a pinned answer returns without + retrieval or a Pi job. Legacy PostgreSQL history remains available + for audit; Pi owns new turns. API callers should supply a request + ID to retry an interrupted wait safely. """ + matched = self.match_operator_rule(question) + if matched is not None: + logger.info("Operator rule %s answered a question", matched.rule.id) + return { + "answer": matched.rule.response, + "sources": [{"source": "operator_rule", "rule_id": matched.rule.id}], + "confidence": "high", + } return agent_runtime.generate(self._agent_payload( question, channel_id, conversation_id, request_id or (f"slack:{channel_id}:{message_ts}" if message_ts else None), diff --git a/backend/app/rules.py b/backend/app/rules.py new file mode 100644 index 0000000..ad41a89 --- /dev/null +++ b/backend/app/rules.py @@ -0,0 +1,180 @@ +""" +Deterministic operator rules for canned answers. + +Operators can pin an exact answer to a question pattern so Wingman replies +with approved wording instead of a generated answer. Rules are checked +before retrieval and generation. An empty or invalid config is a no-op, +and a bad rule is logged and skipped rather than breaking startup. +""" +import json +import logging +import re +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Iterable, List, Optional + +logger = logging.getLogger(__name__) + +MATCH_TYPES = ("contains", "exact", "regex") + + +@dataclass(frozen=True) +class OperatorRule: + """One canned answer pinned to one or more question patterns.""" + + id: str + keywords: tuple + response: str + match_type: str = "contains" + priority: int = 0 + enabled: bool = True + + +@dataclass(frozen=True) +class RuleMatch: + """A successful match, with the pattern that fired.""" + + rule: OperatorRule + matched_keyword: str + + +def _parse_rule(raw: Any) -> Optional[OperatorRule]: + """Validate one raw config entry. Return None when it is unusable.""" + if not isinstance(raw, dict): + logger.warning("Skipping operator rule that is not an object: %r", raw) + return None + + rule_id = raw.get("id") + keywords = raw.get("keywords") + response = raw.get("response") + match_type = raw.get("match_type", "contains") + priority = raw.get("priority", 0) + enabled = raw.get("enabled", True) + + if not isinstance(rule_id, str) or not rule_id.strip(): + logger.warning("Skipping operator rule with a missing id: %r", raw) + return None + if ( + not isinstance(keywords, list) + or not keywords + or not all(isinstance(k, str) and k.strip() for k in keywords) + ): + logger.warning("Skipping operator rule %s with no usable keywords", rule_id) + return None + if not isinstance(response, str) or not response.strip(): + logger.warning("Skipping operator rule %s with an empty response", rule_id) + return None + if match_type not in MATCH_TYPES: + logger.warning( + "Skipping operator rule %s with unknown match_type %r", rule_id, match_type + ) + return None + if not isinstance(priority, int) or isinstance(priority, bool): + logger.warning("Skipping operator rule %s with a non-integer priority", rule_id) + return None + if not isinstance(enabled, bool): + logger.warning("Skipping operator rule %s with a non-boolean enabled flag", rule_id) + return None + if match_type == "regex": + for keyword in keywords: + try: + re.compile(keyword) + except re.error as exc: + logger.warning("Skipping operator rule %s with bad regex %r: %s", rule_id, keyword, exc) + return None + + return OperatorRule( + id=rule_id.strip(), + keywords=tuple(k.strip() for k in keywords), + response=response, + match_type=match_type, + priority=priority, + enabled=enabled, + ) + + +def parse_rules(raw_rules: Any) -> List[OperatorRule]: + """Validate a decoded JSON value into rules, skipping bad entries.""" + if raw_rules is None: + return [] + if not isinstance(raw_rules, list): + logger.warning("Operator rules config is not a list, ignoring it") + return [] + rules = [] + for raw in raw_rules: + rule = _parse_rule(raw) + if rule is not None: + rules.append(rule) + return rules + + +def load_rules(json_text: Optional[str] = None, file_path: Optional[str] = None) -> List[OperatorRule]: + """Load rules from inline JSON or a JSON file. File wins when both are set.""" + text = json_text + if file_path: + path = Path(file_path) + if not path.is_file(): + logger.error("Operator rules file not found: %s", file_path) + return [] + text = path.read_text(encoding="utf-8") + if not text or not text.strip(): + return [] + try: + decoded = json.loads(text) + except json.JSONDecodeError as exc: + logger.error("Operator rules config is not valid JSON: %s", exc) + return [] + return parse_rules(decoded) + + +class RuleEngine: + """Matches a question against the enabled operator rules.""" + + def __init__(self, rules: Iterable[OperatorRule] = ()): + # Higher priority first. The sort is stable, so config order + # breaks ties between rules with equal priority. + self._rules = sorted( + (rule for rule in rules if rule.enabled), + key=lambda rule: -rule.priority, + ) + + @property + def rules(self) -> List[OperatorRule]: + return list(self._rules) + + @classmethod + def from_config(cls, json_text: Optional[str] = None, file_path: Optional[str] = None) -> "RuleEngine": + return cls(load_rules(json_text=json_text, file_path=file_path)) + + @classmethod + def from_settings(cls, settings_obj: Any = None) -> "RuleEngine": + if settings_obj is None: + from app.config import settings as settings_obj + return cls.from_config( + json_text=getattr(settings_obj, "OPERATOR_RULES_JSON", None), + file_path=getattr(settings_obj, "OPERATOR_RULES_FILE", None), + ) + + def match(self, question: Optional[str]) -> Optional[RuleMatch]: + """Return the first matching rule, or None when nothing fires.""" + if not question or not question.strip(): + return None + text = question.strip() + lowered = text.lower() + for rule in self._rules: + for keyword in rule.keywords: + if self._matches(rule, keyword, text, lowered): + return RuleMatch(rule=rule, matched_keyword=keyword) + return None + + @staticmethod + def _matches(rule: OperatorRule, keyword: str, text: str, lowered: str) -> bool: + if rule.match_type == "exact": + return lowered == keyword.lower() + if rule.match_type == "regex": + return re.search(keyword, text, re.IGNORECASE) is not None + return keyword.lower() in lowered + + +# Global instance, following the module-level singleton pattern in this codebase. +rule_engine = RuleEngine.from_settings() diff --git a/backend/tests/test_rules.py b/backend/tests/test_rules.py new file mode 100644 index 0000000..56ef97e --- /dev/null +++ b/backend/tests/test_rules.py @@ -0,0 +1,163 @@ +"""Unit tests for deterministic operator rules and the RAG short-circuit.""" +import json +from types import SimpleNamespace +from unittest.mock import Mock + +from app import rag as rag_module +from app.rules import OperatorRule, RuleEngine, load_rules, parse_rules + + +def engine(*rules): + return RuleEngine(list(rules)) + + +def rule(rule_id, keywords, response="canned", **kwargs): + return OperatorRule(id=rule_id, keywords=tuple(keywords), response=response, **kwargs) + + +class TestMatching: + def test_contains_is_case_insensitive(self): + match = engine(rule("vpn", ["reset vpn"])).match("How do I RESET VPN access?") + assert match is not None + assert match.rule.id == "vpn" + assert match.rule.response == "canned" + + def test_contains_needs_the_keyword(self): + assert engine(rule("vpn", ["reset vpn"])).match("How do I reset email?") is None + + def test_any_keyword_can_match(self): + match = engine(rule("hours", ["opening hours", "when open"])).match("When open?") + assert match is not None + assert match.matched_keyword == "when open" + + def test_exact_needs_the_whole_question(self): + eng = engine(rule("ping", ["ping"], match_type="exact")) + assert eng.match(" Ping ") is not None + assert eng.match("ping the server") is None + + def test_regex_match(self): + eng = engine(rule("ticket", [r"ticket #?\d+"], match_type="regex")) + assert eng.match("Where is ticket #4821?") is not None + assert eng.match("Where is my ticket?") is None + + def test_blank_question_never_matches(self): + eng = engine(rule("vpn", ["vpn"])) + assert eng.match("") is None + assert eng.match(" ") is None + assert eng.match(None) is None + + def test_no_rules_is_a_no_op(self): + assert RuleEngine().match("anything at all") is None + + +class TestOrderingAndEnabled: + def test_higher_priority_wins(self): + eng = engine( + rule("low", ["vpn"], response="low", priority=1), + rule("high", ["vpn"], response="high", priority=10), + ) + assert eng.match("vpn help").rule.response == "high" + + def test_config_order_breaks_priority_ties(self): + eng = engine( + rule("first", ["vpn"], response="first"), + rule("second", ["vpn"], response="second"), + ) + assert eng.match("vpn help").rule.response == "first" + + def test_disabled_rule_never_matches(self): + eng = engine(rule("off", ["vpn"], enabled=False)) + assert eng.match("vpn help") is None + assert eng.rules == [] + + +class TestConfigLoading: + def test_parse_valid_rules(self): + rules = parse_rules([ + {"id": "a", "keywords": ["vpn"], "response": "Use the portal."}, + {"id": "b", "keywords": ["hours"], "response": "9 to 5.", "priority": 3, "match_type": "exact"}, + ]) + assert [r.id for r in rules] == ["a", "b"] + assert rules[1].priority == 3 + assert rules[1].match_type == "exact" + + def test_parse_skips_invalid_entries(self): + rules = parse_rules([ + {"keywords": ["vpn"], "response": "no id"}, + {"id": "x", "keywords": [], "response": "no keywords"}, + {"id": "y", "keywords": ["vpn"], "response": ""}, + {"id": "z", "keywords": ["vpn"], "response": "ok", "match_type": "fuzzy"}, + {"id": "w", "keywords": ["vpn"], "response": "ok", "priority": "high"}, + "not an object", + {"id": "good", "keywords": ["vpn"], "response": "kept"}, + ]) + assert [r.id for r in rules] == ["good"] + + def test_parse_skips_bad_regex(self): + assert parse_rules([ + {"id": "bad", "keywords": ["[unclosed"], "response": "x", "match_type": "regex"}, + ]) == [] + + def test_parse_rejects_non_list(self): + assert parse_rules({"id": "a"}) == [] + assert parse_rules(None) == [] + + def test_load_from_json_text(self): + text = json.dumps([{"id": "a", "keywords": ["vpn"], "response": "portal"}]) + assert RuleEngine.from_config(json_text=text).match("vpn?").rule.response == "portal" + + def test_load_invalid_json_is_empty(self): + assert load_rules(json_text="{not json") == [] + assert load_rules(json_text="") == [] + assert load_rules() == [] + + def test_load_from_file_and_file_wins(self, tmp_path): + path = tmp_path / "rules.json" + path.write_text(json.dumps([ + {"id": "file", "keywords": ["vpn"], "response": "from file"}, + ]), encoding="utf-8") + inline = json.dumps([{"id": "inline", "keywords": ["vpn"], "response": "from inline"}]) + eng = RuleEngine.from_config(json_text=inline, file_path=str(path)) + assert eng.match("vpn?").rule.response == "from file" + + def test_load_missing_file_is_empty(self, tmp_path): + assert load_rules(file_path=str(tmp_path / "nope.json")) == [] + + def test_from_settings_reads_both_fields(self): + fake = SimpleNamespace( + OPERATOR_RULES_JSON=json.dumps([ + {"id": "a", "keywords": ["vpn"], "response": "portal"}, + ]), + OPERATOR_RULES_FILE=None, + ) + assert RuleEngine.from_settings(fake).match("vpn help") is not None + empty = SimpleNamespace(OPERATOR_RULES_JSON=None, OPERATOR_RULES_FILE=None) + assert RuleEngine.from_settings(empty).match("vpn help") is None + + +class TestRagShortCircuit: + def _install_engine(self, monkeypatch, eng): + monkeypatch.setattr(rag_module.rules, "rule_engine", eng) + + def test_matched_question_skips_retrieval_and_generation(self, monkeypatch): + self._install_engine(monkeypatch, engine(rule("vpn", ["reset vpn"], response="Use the portal."))) + search = Mock(side_effect=AssertionError("retrieval must not run")) + generate = Mock(side_effect=AssertionError("generation must not run")) + monkeypatch.setattr(rag_module.vector_store, "similarity_search", search) + monkeypatch.setattr(rag_module.agent_runtime, "generate", generate) + + result = rag_module.rag_engine.generate_response("How do I reset VPN access?") + + assert result == { + "answer": "Use the portal.", + "sources": [{"source": "operator_rule", "rule_id": "vpn"}], + "confidence": "high", + } + + def test_unmatched_question_still_uses_generation(self, monkeypatch): + self._install_engine(monkeypatch, engine(rule("vpn", ["reset vpn"]))) + monkeypatch.setattr(rag_module.vector_store, "similarity_search", lambda _: []) + expected = {"answer": "generated", "sources": [], "confidence": "low"} + monkeypatch.setattr(rag_module.agent_runtime, "generate", lambda _: expected) + + assert rag_module.rag_engine.generate_response("What is RAG?") == expected diff --git a/docs/operator-rules.md b/docs/operator-rules.md new file mode 100644 index 0000000..df98f28 --- /dev/null +++ b/docs/operator-rules.md @@ -0,0 +1,72 @@ +# Operator rules + +Operator rules pin an approved answer to a question pattern. When a rule +fires, Wingman returns the canned response straight away, without retrieval +or a generated answer. Use them when the generic answer is wrong or stale +and the correct wording is known. + +Rules are checked on the `/api/ask` path through `RAGEngine.generate_response` +in [backend/app/rag.py](../backend/app/rag.py). The matcher lives in +[backend/app/rules.py](../backend/app/rules.py). + +## Configure + +Set one of these in `.env`. The file wins when both are set. With neither +set, or with an empty value, rules are a no-op. + +```bash +# Inline JSON +OPERATOR_RULES_JSON=[{"id":"vpn-reset","keywords":["reset vpn"],"response":"Use the IT portal to reset VPN access."}] + +# Or a JSON file with the same list shape +OPERATOR_RULES_FILE=/path/to/operator-rules.json +``` + +## Rule shape + +| Field | Required | Notes | +| --- | --- | --- | +| `id` | yes | Short unique name, reported in the response sources | +| `keywords` | yes | One or more patterns; any one can fire the rule | +| `response` | yes | The exact answer returned | +| `match_type` | no | `contains` (default), `exact` or `regex` | +| `priority` | no | Higher wins first; config order breaks ties. Default 0 | +| `enabled` | no | Set `false` to park a rule without deleting it. Default true | + +Matching is case-insensitive. `contains` checks whether the pattern appears +anywhere in the question, `exact` needs the whole question to equal the +pattern, and `regex` treats the pattern as a regular expression searched +against the question. + +## Example + +```json +[ + { + "id": "vpn-reset", + "keywords": ["reset vpn", "vpn password"], + "response": "Use the IT portal at https://it.example.com/vpn to reset VPN access.", + "priority": 10 + }, + { + "id": "office-hours", + "keywords": ["office hours"], + "match_type": "exact", + "response": "The support desk runs 9am to 5pm, Monday to Friday." + } +] +``` + +## Failure behaviour + +A rule with a missing id, no usable keywords, an empty response, an unknown +match type, a non-integer priority or a bad regex is logged and skipped. +Invalid JSON, a missing file or a non-list config loads as no rules. Bad +config never stops the backend from starting. + +## Current limits + +Rules apply to the API ask path in this slice. Slack delivery still goes +through the Pi Durable job flow, so wiring rules into the Slack handlers is +a follow-up. Combining a rule with fetched source data before summarising +is also still open under the parent issue. diff --git a/mise.toml b/mise.toml index df04a3b..9b26183 100644 --- a/mise.toml +++ b/mise.toml @@ -51,7 +51,7 @@ run = "docker compose ps" [tasks.test-backend] description = "Run backend tests" dir = "backend" -run = "uv run pytest tests/test_file_extraction_units.py tests/test_logging_config.py tests/test_logging.py tests/test_agent_runtime.py -v" +run = "uv run pytest tests/test_file_extraction_units.py tests/test_logging_config.py tests/test_logging.py tests/test_agent_runtime.py tests/test_rules.py -v" [tasks.migrate] description = "Run database migrations (upgrade to latest)"