rdnpradipta/knowledge-base-mcp

1"""Flow A — deterministic error-code lookup against the relational `error_codes`2table. No LLM, no vector search: extract the code from the user's text, fetch the3row, and return it verbatim. Unknown codes get a fuzzy "did you mean?" ask-back,4mirroring the entity-validation pattern in orchestrator.validate_entities.5 6CLI: python error_lookup.py E9017"""8from __future__ import annotations9 10import re11import sys12from dataclasses import dataclass, field13 14from rapidfuzz import fuzz, process15 16from rag.db import get_conn17 18# Codes in this table are strictly one letter (C/E) + 3 digits.19_CODE_TOKEN = re.compile(r"\b([CE])?-?(\d{3})\b", re.IGNORECASE)20 21 22@dataclass23class ErrorCodeRow:24    code: str25    code_group: str26    possible_cause: str27    action: str28    guide_note: str29    source_file: str30 31 32@dataclass33class LookupResult:34    """rows is populated on a hit; message carries ask-back/guidance otherwise."""35    rows: list[ErrorCodeRow] = field(default_factory=list)36    message: str | None = None37 38 39def extract_codes(text: str) -> list[str]:40    """Pull candidate codes from free text and normalise to the table's form.41    A bare 3-digit number (e.g. "901") expands to both C and E variants."""42    codes: list[str] = []43    for prefix, digits in _CODE_TOKEN.findall(text):44        if prefix:45            codes.append(f"{prefix.upper()}{digits}")46        else:47            codes.extend([f"C{digits}", f"E{digits}"])48    # de-dupe, preserve order49    seen: dict[str, None] = {}50    for c in codes:51        seen.setdefault(c, None)52    return list(seen)53 54 55def _all_codes(conn) -> list[str]:56    return [r[0] for r in conn.execute("SELECT code FROM error_codes").fetchall()]57 58 59def lookup(text: str) -> LookupResult:60    candidates = extract_codes(text)61    if not candidates:62        return LookupResult(message=(63            "Please enter an error code, e.g. **E901** or **C911** "64            "(the code shown on the unit)."65        ))66 67    with get_conn(vector=False) as conn:68        found = conn.execute(69            """70            SELECT code, code_group, possible_cause, action, guide_note, source_file71            FROM error_codes WHERE code = ANY(%s)72            """,73            (candidates,),74        ).fetchall()75        if not found:76            lexicon = _all_codes(conn)77            suggestion = None78            if lexicon:79                match = process.extractOne(80                    candidates[0], lexicon, scorer=fuzz.WRatio, score_cutoff=7081                )82                if match:83                    suggestion = match[0]84            hint = f' Did you mean **{suggestion}**?' if suggestion else ""85            return LookupResult(message=(86                f'No entry for **{candidates[0]}** in the error-code database.{hint} '87                "Please check the code and try again."88            ))89 90    # Keep the order the user mentioned the codes in.91    by_code = {r[0]: r for r in found}92    rows = [ErrorCodeRow(*by_code[c]) for c in candidates if c in by_code]93    return LookupResult(rows=rows)94 95 96def format_row(row: ErrorCodeRow) -> str:97    """Render one error-code row as grounded markdown for the chat UI."""98    parts = [f"### Error code `{row.code}`  \n*Reported as: {row.code_group}*"]99    if row.possible_cause:100        parts.append(f"**Possible cause**\n\n{row.possible_cause}")101    if row.action:102        parts.append(f"**Action**\n\n{row.action}")103    if row.guide_note:104        parts.append(f"**Guide note**\n\n{row.guide_note}")105    return "\n\n".join(parts)106 107 108if __name__ == "__main__":109    if len(sys.argv) < 2:110        sys.exit('usage: python error_lookup.py "E901"')111    result = lookup(" ".join(sys.argv[1:]))112    if result.message:113        print(result.message)114    for row in result.rows:115        print(format_row(row))116        print(f"\n— source: {row.source_file}\n" + "-" * 60)117