"""Deterministic, grounded answerer over the brain graph. Routes a natural-language question to a graph traversal based on the resolved entity + intent keywords, and returns markdown with clickable wiki links. Every answer is grounded in graph edges; nothing is invented. Falls back to keyword search when no entity is recognised. """ from __future__ import annotations import re from query.retriever import get_brain, page_path def _link(ent) -> str: return f"[{ent['name']}](/wiki/{page_path(ent)})" def _cite(ent) -> dict: return {"name": ent["name"], "type": ent["type"], "path": "/wiki/" + page_path(ent)} def _bullets(ents, cap=60): ents = sorted(ents, key=lambda e: e["name"]) lines = [f"- {_link(e)} — {_short(e)}" for e in ents[:cap]] if len(ents) > cap: lines.append(f"- …and {len(ents) - cap} more (open the page to see all).") return "\n".join(lines) def _short(e) -> str: a = e.get("attrs", {}) if e["type"] == "sp": return (a.get("purpose") or (a.get("csv_tasks") or ["stored procedure"])[0]).strip() if e["type"] == "error": return (a.get("message", "")[:80]) if e["type"] == "table": return f"{a.get('column_count', 0)} columns" if e["type"] == "activity": return a.get("desc", "") if e["type"] == "api": return f"v{'/'.join(a.get('versions', []))}" return a.get("desc", "") def _has(q, *words): return any(re.search(rf"\b{w}", q) for w in words) def _unresolved_note(refs): if not refs: return "" shown = ", ".join(f"`{r}`" for r in sorted(set(refs))[:12]) return (f"\n\n_(plus non-core relations — views / cross-module tables not in PO/Table: " f"{shown})_") def answer(q: str) -> dict: b = get_brain() ql = q.lower().strip() # explicit error-id lookup err = b.resolve_error_id(ql) mentions = b.resolve_mentions(ql) if err and err not in mentions: mentions = [err] + mentions if not mentions: mentions = b.resolve_intent(ql) if not mentions: return _search_answer(b, q) primary = b.entities[mentions[0]] t = primary["type"] cites = [_cite(primary)] def respond(lead, ents): for e in ents: cites.append(_cite(e)) bod = _bullets(ents) if ents else "_(none found in the graph)_" return {"answer": f"{lead}\n\n{bod}", "citations": cites, "matched": _cite(primary)} if t == "table": if _has(ql, "writ", "insert", "updat", "save", "modif"): return respond(f"Stored procedures that **write** {_link(primary)}:", b.neighbors(primary["id"], "writes", reverse=True)) if _has(ql, "read", "select", "fetch", "use"): return respond(f"Stored procedures that **read** {_link(primary)}:", b.neighbors(primary["id"], "reads", reverse=True)) w = b.neighbors(primary["id"], "writes", reverse=True) r = b.neighbors(primary["id"], "reads", reverse=True) a = primary["attrs"] lead = (f"**{primary['name']}** — {a.get('column_count',0)} columns. " f"Written by {len(w)} SP(s), read by {len(r)} SP(s).") return respond(lead, w + r) if t == "sp": pa = primary["attrs"] if _has(ql, "writ", "insert", "updat"): ents = b.neighbors(primary["id"], "writes") extra = _unresolved_note(pa.get("unresolved_writes", [])) return respond(f"Tables {_link(primary)} **writes**:{extra}", ents) if _has(ql, "read", "select", "fetch"): ents = b.neighbors(primary["id"], "reads") extra = _unresolved_note(pa.get("unresolved_reads", [])) return respond(f"Tables {_link(primary)} **reads**:{extra}", ents) if _has(ql, "call", "invoke", "exec"): return respond(f"SPs {_link(primary)} calls / is called by:", b.neighbors(primary["id"], "calls") + b.neighbors(primary["id"], "calls", reverse=True)) if _has(ql, "error", "raise", "fail", "message"): return respond(f"Errors {_link(primary)} can raise:", b.neighbors(primary["id"], "raises")) if _has(ql, "activit", "screen", "journ", "where", "used", "belong"): return respond(f"Where {_link(primary)} is reached from:", b.neighbors(primary["id"], "runs", reverse=True) + b.neighbors(primary["id"], "invokes", reverse=True)) # overview w = b.neighbors(primary["id"], "writes") r = b.neighbors(primary["id"], "reads") acts = b.neighbors(primary["id"], "runs", reverse=True) errs = b.neighbors(primary["id"], "raises") lead = (f"**{primary['name']}** — {_short(primary)}. " f"Writes {len(w)}, reads {len(r)} table(s); raises {len(errs)} error(s); " f"reached by {len(acts)} activity(ies).") return respond(lead, w + r + acts) if t == "activity": if _has(ql, "screen"): return respond(f"Screens in **{primary['name']}** ({_short(primary)}):", b.neighbors(primary["id"], "shown_on")) if _has(ql, "api", "endpoint", "rest"): return respond(f"API(s) corresponding to **{primary['name']}** (inferred):", b.neighbors(primary["id"], "backed_by", reverse=True)) if _has(ql, "sp", "procedure", "proc"): return respond(f"Stored procedures in **{primary['name']}**:", b.neighbors(primary["id"], "runs")) sc = b.neighbors(primary["id"], "shown_on") sp = b.neighbors(primary["id"], "runs") api = b.neighbors(primary["id"], "backed_by", reverse=True) lead = (f"**{primary['name']}** — {_short(primary)}. " f"{len(sc)} screen(s), {len(sp)} SP(s)" + (f", API {api[0]['name']}" if api else "") + ".") return respond(lead, sc + api) if t == "api": if _has(ql, "field"): flds = primary["attrs"].get("fields", []) body = ", ".join(f"`{x}`" for x in flds[:80]) return {"answer": f"**{primary['name']}** exposes {len(flds)} fields:\n\n{body}", "citations": cites, "matched": _cite(primary)} acts = b.neighbors(primary["id"], "backed_by") a = primary["attrs"] lead = (f"**{primary['name']}** — REST endpoint v{'/'.join(a.get('versions', []))}, " f"{len(a.get('fields', []))} fields" + (f". Corresponds to activity {acts[0]['name']} (inferred)" if acts else "") + ".") return respond(lead, acts) if t == "screen": if _has(ql, "activit", "journ"): return respond(f"Activities that show **{primary['name']}**:", b.neighbors(primary["id"], "shown_on", reverse=True)) return respond(f"Stored procedures invoked by screen **{primary['name']}**:", b.neighbors(primary["id"], "invokes")) if t == "error": a = primary["attrs"] sps = b.neighbors(primary["id"], "raises", reverse=True) lead = (f"**Error {primary['name']}** ({a.get('severity','')}):\n\n" f"> {a.get('message','')}\n\nRaised by:") return respond(lead, sps) return _search_answer(b, q) _TYPE_HINT = [("procedure", "sp"), (r"\bsp\b", "sp"), ("proc", "sp"), ("table", "table"), ("column", "table"), ("screen", "screen"), (r"\bapi\b", "api"), ("endpoint", "api"), ("error", "error"), ("activit", "activity"), ("journey", "activity")] def _search_answer(b, q) -> dict: ql = q.lower() want = {etype for pat, etype in _TYPE_HINT if re.search(pat, ql)} hits = [b.entities[i] for i in b.search(q, limit=40)] if want: biased = [e for e in hits if e["type"] in want] if biased: hits = biased hits = hits[:12] if not hits: return {"answer": "I couldn't find anything matching that in the PO brain. " "Try a stored-procedure name, a table, an activity (e.g. PoCrt), " "an API (e.g. CreatePO), or an error id.", "citations": [], "matched": None} return {"answer": "Closest matches in the brain:", "citations": [_cite(e) for e in hits], "matched": None, "results_md": _bullets(hits)} | { "answer": "Closest matches in the brain:\n\n" + _bullets(hits)} if __name__ == "__main__": import sys print(answer(" ".join(sys.argv[1:]) or "which SPs write PO_POMAS_PUR_ORDER_HDR")["answer"][:1200])