"""Step 4 — the conversational answer engine. question -> (target on the spine, intent) -> grounded facts -> 2-4 sentences of English + the verifiable chain (screen -> service/API -> SP -> tables) + source citations. The composer is deterministic and fully grounded. query.llm.polish() optionally smooths the prose with Gemini using ONLY these facts; it never changes the grounding or the chain. """ from __future__ import annotations import json import re import urllib.parse import config from compiler.common import eid from query.retriever import get_brain from query.docstore import get_docstore from query import llm # ---------------------------------------------------------------- loading class V2Brain: def __init__(self): self.spine = json.loads((config.GRAPH_DIR / "spine.json").read_text())["activities"] self.glosses = json.loads((config.GRAPH_DIR / "glosses.json").read_text()) self.brain = get_brain() self.docs = get_docstore() self.xref = {} if config.JOURNEY_XREF_JSON.exists(): self.xref = json.loads(config.JOURNEY_XREF_JSON.read_text()) # quick lookup: lowercase activity desc/name -> activity name self.activity_by_word = {} for name, a in self.spine.items(): for w in re.findall(r"[a-z]+", (a.get("desc", "") + " " + name).lower()): self.activity_by_word.setdefault(w, name) def gloss(self, node_id): return self.glosses.get(node_id, {}) _V2 = None def v2() -> V2Brain: global _V2 if _V2 is None: _V2 = V2Brain() return _V2 # ---------------------------------------------------------------- intent VERB_ACT = {"creat": "PoCrt", "raise": "PoCrt", "amend": "PoAmnd", "approv": "PoApp", "edit": "PoEdt", "hold": "PoHold", "copy": "PoCopy", "view": "PoViewDtls", "display": "PoViewDtls", "schedul": "PoScl"} def classify(q: str) -> str: s = q.lower() if re.search(r"\bcompare\b|difference between|\bvs\.?\b|versus", s): return "COMPARE" # runtime/journey status — check before the transactional rule ("live ... today") if re.search(r"implemented|runtime|chatbot|journey|in the bot|live (today|now|in)", s): return "META_JOURNEY" # transactional / live-data questions the brain (a metadata index) cannot answer if (re.search(r"\b(last|this|next)\s+(month|week|year|quarter|day)\b|yesterday|today" r"|total spend|how much did|\bactual\b", s) or re.search(r"how many (pos|purchase orders|orders|transactions)\b", s)): return "META_LIMITS" if re.search(r"can('| n)?o?t? (answer|do)|not answer|boundaries|limitation|where would i look", s): return "META_LIMITS" if re.search(r"variant|types? of|capital|dropship|consignment|general po|kinds of", s): return "VARIANTS" if re.search(r"valid|error|rule|reject|mandatory|required|fail|stop me|cannot be|blank", s): return "VALIDATIONS" if re.search(r"which (api|service|endpoint)|what api|endpoint|payload|fires? when", s): return "API" if re.search(r"which (tables?|sps?|stored proc)|tables? (get|are) |populat|written|stored|" r"writes? to|data footprint|lands? in", s): return "TABLES" if re.search(r"\btrace\b|chain|from .* to |every step|between clicking", s): return "TRACE" if re.search(r"screens?|sub-?screens?", s): return "SCREENS" if re.search(r"what happens|walk me|explain the|how do i|process|overview|what does .* do|" r"what is the purpose", s): return "WHAT_HAPPENS" if re.search(r"what is|what does|meaning|mean|stand for|explain|describe|tell me about", s): return "WHAT_IS" return "WHAT_IS" def resolve_activity(q: str) -> str | None: s = q.lower() # 1) explicit action verb (create / amend / approve / edit / hold / copy / view / schedule) for verb, act in VERB_ACT.items(): if re.search(rf"\b{verb}", s) and act in v2().spine: return act # 2) a full activity description mentioned verbatim ("amend purchase order") best = None for name, a in v2().spine.items(): desc = (a.get("desc") or "").lower() # require a distinctive multi-word match, not just "purchase order" if desc and desc in s and desc not in ("purchase order",): if best is None or len(desc) > len(v2().spine[best]["desc"]): best = name return best # None -> caller defaults generic PO questions to the canonical create flow # ---------------------------------------------------------------- helpers def _cite(*items): out, seen = [], set() for it in items: if not it: continue key = (it.get("source"), it.get("loc")) if key not in seen: seen.add(key) out.append({k: it[k] for k in ("source", "loc", "title") if k in it}) return out def _g_text(node_id): g = v2().gloss(node_id) return g.get("gloss", "") def _table_phrase(tname): g = v2().gloss(eid("table", tname)) txt = g.get("gloss", tname) return re.sub(r"^Stores (the )?", "", txt).rstrip(".") def build_chain(act_name) -> list[dict]: a = v2().spine[act_name] chain = [{"step": "screen", "label": "Screens", "items": a["screens"]}] sections = [{"name": s["label"], "id": s["section_id"], "sps": [x["sp"] for x in s["save_sps"]], "api": s.get("sub_screen_api"), "tables": s["tables"]} for s in a["sections"]] chain.append({"step": "section", "label": "Sub-screen sections", "items": sections}) if a.get("primary_api"): chain.append({"step": "api", "label": "API", "items": [a["primary_api"]["name"]]}) chain.append({"step": "sp", "label": "Save procedures", "items": sorted({x["sp"] for s in a["sections"] for x in s["save_sps"]})}) chain.append({"step": "table", "label": "Tables written", "items": a["tables"]}) return chain # ---------------------------------------------------------------- composers def answer(q: str, use_llm: bool | None = None) -> dict: vb = v2() intent = classify(q) mentions = vb.brain.resolve_mentions(q.lower()) primary = vb.brain.entities[mentions[0]] if mentions else None act = resolve_activity(q) # generic PO action question with no explicit target -> default to the canonical create flow if (not act and not (primary and primary["type"] in ("table", "sp", "api")) and re.search(r"purchase order|\bpos?\b", q.lower()) and intent in {"WHAT_HAPPENS", "TABLES", "API", "VALIDATIONS", "VARIANTS", "TRACE", "SCREENS"}): act = "PoCrt" # entity-first intents (table/sp/api meaning, traceability) if primary and primary["type"] in ("table", "sp", "api") and intent in ( "WHAT_IS", "TABLES", "TRACE", "API", "VALIDATIONS"): res = _entity_answer(primary, intent, q) elif intent == "META_JOURNEY": res = _meta_journey(q) elif intent == "META_LIMITS": res = _meta_limits() elif intent == "COMPARE": res = _compare(q) elif act: res = _activity_answer(act, intent, q) elif primary: res = _entity_answer(primary, intent, q) else: res = _doc_fallback(q) res.setdefault("intent", intent) # handoff: if the answer is about an activity that IS a live journey, offer to run it tgt = res.get("target") if tgt and tgt.get("type") == "activity": xr = vb.xref.get(eid("activity", tgt["name"])) if xr and xr.get("implemented_in_runtime"): seed = urllib.parse.quote(xr.get("activity_desc") or tgt["name"]) res["handoff"] = {"label": "Do this in the assistant", "url": f"{config.RAMCO_CHAT_URL}/?start={seed}", "journey": xr.get("activity_desc", tgt["name"])} # LLM composition: rewrite the grounded sections into a tailored, natural answer. # Grounded material comes from the deterministic layer; the LLM may only rephrase it. want_llm = llm.available() if use_llm is None else use_llm if want_llm and res.get("sections"): composed = llm.compose(q, res["sections"], res.get("facts", {})) if composed: res["sections_grounded"] = res["sections"] # keep the deterministic source res["sections"] = composed res["answer"] = _clean_join(composed) res["llm"] = True return res # ---- section builders (rich, grounded, multi-paragraph) ------------------- def _human_list(items): items = [i for i in dict.fromkeys([x for x in items if x])] if not items: return "" if len(items) == 1: return items[0] return ", ".join(items[:-1]) + " and " + items[-1] def _sec(h, b): return {"heading": h, "body": _clean(b)} def _activity_errors(act_name): b = v2().brain a = v2().spine[act_name] save = {eid("sp", x["sp"]) for s in a["sections"] for x in s["save_sps"]} seen, examples, total = set(), [], 0 for sp in save: for e in b.fwd[sp].get("raises", []): ent = b.entities[e] if ent["name"] in seen: continue seen.add(ent["name"]); total += 1 msg = (ent["attrs"].get("message") or "").strip() if msg and not msg.startswith("^") and len(msg) < 110 and len(examples) < 4: examples.append(msg) return total, examples def _sec_overview(act): g = v2().gloss(eid("activity", act)) txt = re.sub(r"^(Process\s+Overview|Overview|Introduction)\s+", "", g.get("gloss", "")) return _sec("What it is", txt), g.get("sources", []) def _sec_fill(act): a = v2().spine[act] secs = [s["label"].lower() for s in a["sections"]] body = (f"It begins on the main screen, which captures the core of the order — the supplier, " f"the buyer, the purchase-order type, the currency, and the individual line items being " f"ordered. From there, {len(secs)} optional sub-screens capture the rest of the document: " f"{_human_list(secs)}. You fill only the sub-screens that are relevant to your order, so a " f"simple PO might use just the main screen while a complex one touches several.") return _sec("What you fill in", body), [{"source": a["source"]}] def _sec_save(act): a = v2().spine[act] api = a.get("primary_api") nsp = a["counts"]["save_sps"] subs = list(dict.fromkeys([s["sub_screen_api"] for s in a["sections"] if s.get("sub_screen_api")])) body = (f"When you save, the values on screen are assembled into a single payload and sent to the " + (f"{api['name']} service (versions {', '.join(api['versions'])}, {api['fields']} request " f"and response fields)" if api else "Purchase Order service") + f". Behind that service, {nsp} stored procedures run: they first validate the document " f"and then write each part of the order to its own set of tables — the header and line " f"items first, then each sub-screen's data. ") if subs: body += f"Several sub-screens save through their own service as well, such as {_human_list(subs)}." return _sec("What happens when you save", body), ([{"source": api["source"]}] if api else []) def _sec_tables(act): a = v2().spine[act] anchors = {"PO_POMAS_PUR_ORDER_HDR", "PO_POITM_ITEM_DETAIL"} groups = [] for s in a["sections"]: d = [t for t in s["tables"] if t not in anchors] if d: groups.append(f"{s['label'].lower()} in {_human_list(d)}") groups = list(dict.fromkeys(groups)) body = (f"In all, saving writes to {a['counts']['tables']} tables. The order header is held in " f"PO_POMAS_PUR_ORDER_HDR — one row per purchase order — and every ordered line in " f"PO_POITM_ITEM_DETAIL. Beyond those, each part of the document has its own tables: " + "; ".join(groups) + ". This is why a single PO touches so many tables: it is one logical " "document spread across header, lines, schedules, taxes, terms and notes.") return _sec("Where the data is stored", body), [{"source": a["source"]}] def _sec_variants(act): d = _types_sentence() drop = "Dropship" in str(v2().spine[act]["sections"]) base = (d["text"] + " ") if d and d.get("text") else \ "Purchase orders come in several types — General, Capital, Consignment and Dropship. " body = (base + "Each type becomes its own variant with different required inputs: a General PO is " "the standard procurement of stockable or non-stockable items; a Capital PO is for capital " "assets and needs capital-proposal details; a Consignment PO covers supplier-owned stock; " "and a Dropship PO ships straight to a customer and therefore needs a customer / dropship " "address" + (", which is captured on a dedicated Dropship sub-screen here" if drop else "") + ". The combination of type plus the sub-screens you fill defines the specific variant.") cites = [{k: d[k] for k in ("source", "loc", "title") if k in d}] if d else [] return _sec("Types and variants", body), cites def _sec_screens(act): a = v2().spine[act] main = next((s for s in a["screens"] if s.lower().endswith("main")), None) items = [] for sc in a["screens"]: if sc == main: continue g = (v2().gloss(eid("screen", sc)).get("gloss") or "").strip() items.append(f"{sc} ({g})" if g else sc) main_g = (v2().gloss(eid("screen", main)).get("gloss") or "the main order") if main else "the main order" body = (f"{a['desc']} is made up of {len(a['screens'])} screens. You start on the main screen" + (f" — {main} ({main_g})" if main else "") + f", which captures the core order. The sub-screens, each capturing one part of the " f"document, are: " + _human_list(items) + ". You open only the sub-screens your order " "actually needs.") return _sec("Screens and sub-screens", body), [{"source": a["source"]}] def _sec_validations(act): a = v2().spine[act] total, ex = _activity_errors(act) examples = " ".join(f"“{e}”;" for e in ex).rstrip(";") body = (f"Saving is not unconditional. Before the order is written, the stored procedures run a " f"large body of validation and will stop the save if something is inconsistent. Across " f"{a['desc'].lower()} there are roughly {total} distinct error checks drawn from Ramco's " f"own error catalogue. Typical examples: {examples}." if ex else f"Before saving, the procedures run roughly {total} distinct validation checks and stop " f"the save on any failure.") return _sec("Validations and checks", body), [{"source": "PO/ModelInfo/PO_Design_ErrorMessage.xlsx"}] _SECTION_ORDER = { "WHAT_HAPPENS": ["overview", "fill", "save", "tables", "variants"], "WHAT_IS": ["overview", "fill", "save", "tables", "variants"], "TABLES": ["tables", "save", "overview"], "TRACE": ["tables", "save", "fill"], "API": ["save", "fill", "tables"], "SCREENS": ["screens", "overview", "save"], "VARIANTS": ["variants", "overview", "fill"], "VALIDATIONS": ["validations", "save", "overview"], } _BUILDERS = {"overview": _sec_overview, "fill": _sec_fill, "save": _sec_save, "tables": _sec_tables, "variants": _sec_variants, "validations": _sec_validations, "screens": _sec_screens} def _activity_answer(act_name, intent, q) -> dict: a = v2().spine[act_name] order = _SECTION_ORDER.get(intent, _SECTION_ORDER["WHAT_HAPPENS"]) sections, cites = [], [] for key in order: sec, c = _BUILDERS[key](act_name) if sec["body"]: sections.append(sec); cites += c cites += [{"source": a["source"]}] if a.get("primary_api"): cites.append({"source": a["primary_api"]["source"]}) return {"answer": _clean_join(sections), "sections": sections, "target": {"name": act_name, "type": "activity"}, "chain": build_chain(act_name), "citations": _cite(*cites), "facts": {"activity": act_name, "desc": a["desc"], "tables": a["tables"]}} def _clean_join(sections) -> str: return "\n\n".join(f"{s['heading']}. {s['body']}" for s in sections) def _entity_answer(ent, intent, q) -> dict: nid = ent["id"] gloss = _g_text(nid) if ent["type"] == "table": writers = vb_writers(ent) acts = sorted({a for a, _ in writers}) g = v2().gloss(nid) detail = g.get("detail", "") cols = ent["attrs"].get("columns", []) secs = [_sec("What it stores", f"{gloss} {detail}" if detail else gloss)] secs.append(_sec("Where it fits in the flow", (f"This table sits on the save path of {_human_list(acts)} — those activities write to it " f"when their purchase order is saved. " if acts else "It is part of the PO data model. ") + (f"{len(writers)} stored procedures write to it in total." if writers else ""))) if cols: names = ", ".join(c["name"] for c in cols[:18]) secs.append(_sec(f"Columns ({len(cols)})", f"Its columns include {names}" + (" …and more." if len(cols) > 18 else "."))) chain = [{"step": "table", "label": "Table", "items": [ent["name"]]}, {"step": "sp", "label": "Written by SPs", "items": sorted({sp for _, sp in writers})[:30]}, {"step": "activity", "label": "Via activities", "items": acts}] return {"answer": _clean_join(secs), "sections": secs, "target": {"name": ent["name"], "type": "table"}, "chain": chain, "citations": _cite(*g.get("sources", []), {"source": ent.get("source")}), "facts": {"table": ent["name"], "written_by": acts}} if ent["type"] == "api": a = ent["attrs"] ans = (f"{gloss} It exposes {len(a.get('fields', []))} request/response fields " f"across versions {', '.join(a.get('versions', []))}.") return {"answer": _clean(ans), "target": {"name": ent["name"], "type": "api"}, "chain": [{"step": "api", "label": "API", "items": [ent["name"]]}], "citations": _cite({"source": ent.get("source")}), "facts": {"api": ent["name"], "fields": len(a.get("fields", []))}} # sp writes = sorted({d.split(":")[1].upper() for d in vb_sp_writes(ent)}) purpose = ent["attrs"].get("purpose") or (ent["attrs"].get("csv_tasks") or [""])[0] ans = (f"{ent['name']} — {purpose or 'a Purchase Order stored procedure'}. " + (f"It writes to {', '.join(writes[:6])}." if writes else "It performs validation/fetch logic and writes no core PO table.")) return {"answer": _clean(ans), "target": {"name": ent["name"], "type": "sp"}, "chain": [{"step": "sp", "label": "Procedure", "items": [ent["name"]]}, {"step": "table", "label": "Writes", "items": writes}], "citations": _cite({"source": ent.get("source")}), "facts": {"sp": ent["name"], "writes": writes}} def vb_writers(table_ent): """[(activity, sp)] that write a table, from the spine.""" out = [] tname = table_ent["name"] for act, a in v2().spine.items(): for s in a["sections"]: if tname in s["tables"]: for x in s["save_sps"]: if tname in x["writes"]: out.append((act, x["sp"])) return out def vb_sp_writes(sp_ent): b = v2().brain return [d for d in b.fwd[sp_ent["id"]].get("writes", [])] def _validations_for_activity(act_name, q) -> dict: b = v2().brain a = v2().spine[act_name] save_sps = {eid("sp", x["sp"]) for s in a["sections"] for x in s["save_sps"]} errs = [] for sp in save_sps: for e in b.fwd[sp].get("raises", []): errs.append(b.entities[e]) seen, uniq = set(), [] for e in errs: if e["name"] not in seen: seen.add(e["name"]); uniq.append(e) top = uniq[:6] lines = "; ".join(f"“{e['attrs'].get('message','')[:90]}”" for e in top) ans = (f"When saving {a['desc'].lower()}, the system runs validation in its save procedures " f"and can raise {len(uniq)} distinct errors. Examples: {lines}.") return {"answer": _clean(ans), "intent": "VALIDATIONS", "target": {"name": act_name, "type": "activity"}, "chain": [{"step": "error", "label": "Sample validations", "items": [e["name"] for e in top]}], "citations": _cite({"source": "PO/ModelInfo/PO_Design_ErrorMessage.xlsx"}), "facts": {"activity": act_name, "error_count": len(uniq), "examples": [e["attrs"].get("message", "") for e in top]}} def _types_sentence(): """A sentence that actually lists >=2 PO types, from the manuals (or the PoCrt gloss).""" from query.docstore import sentences, is_clean_sentence TYPES = ("general", "capital", "consignment", "dropship") for c in v2().docs.search("types of purchase order general capital consignment dropship", k=20, prefer_types=["arm", "procurement_manual"]): for s in sentences(c["text"]): if is_clean_sentence(s) and sum(t in s.lower() for t in TYPES) >= 2: return {"text": re.sub(r"\s+", " ", s)[:300].strip(), "source": c["source"], "loc": c["loc"], "title": c["doc_title"]} g = v2().gloss(eid("activity", "PoCrt")) return {"text": g.get("gloss", ""), **({"source": g["sources"][0]["source"], "loc": g["sources"][0].get("loc")} if g.get("sources") else {})} def _variants(act_name, q) -> dict: d = _types_sentence() drop = "Dropship" in str(v2().spine[act_name]["sections"]) base = (d["text"] if d and d.get("text") else "Purchase orders come in several types — General, Capital, Consignment and Dropship.") ans = (f"{base} Each type unrolls into a variant: e.g. a Capital PO needs capital-proposal " f"details, and a Dropship PO needs a customer/dropship address" + (" (the brain sees a dedicated Dropship sub-screen on this activity)." if drop else ".")) return {"answer": _clean(ans), "intent": "VARIANTS", "target": {"name": act_name, "type": "activity"}, "chain": build_chain(act_name), "citations": _cite(d) if d else [], "facts": {"activity": act_name}} def _compare(q) -> dict: acts = [] for verb, a in VERB_ACT.items(): if re.search(rf"\b{verb}", q.lower()) and a in v2().spine and a not in acts: acts.append(a) if len(acts) < 2: acts = ["PoCrt", "PoAmnd"] a1, a2 = (v2().spine[acts[0]], v2().spine[acts[1]]) t1, t2 = set(a1["tables"]), set(a2["tables"]) only1 = sorted(t1 - t2)[:5]; only2 = sorted(t2 - t1)[:5]; both = len(t1 & t2) ans = (f"{a1['desc']} writes {len(t1)} tables; {a2['desc']} writes {len(t2)}. " f"They share {both} core tables (header, items, schedule…). " f"Only {a1['desc']}: {', '.join(only1) or 'none'}. " f"Only {a2['desc']}: {', '.join(only2) or 'none'}.") return {"answer": _clean(ans), "intent": "COMPARE", "target": {"name": f"{acts[0]} vs {acts[1]}", "type": "compare"}, "chain": [{"step": "table", "label": f"{acts[0]} tables", "items": a1["tables"]}, {"step": "table", "label": f"{acts[1]} tables", "items": a2["tables"]}], "citations": _cite({"source": a1["source"]}), "facts": {"compare": acts}} def _meta_journey(q) -> dict: xref = {} p = config.JOURNEY_XREF_JSON if p.exists(): xref = json.loads(p.read_text()) impl = [v["activity_name"] for v in xref.values() if v.get("implemented_in_runtime")] modeled = [v["activity_name"] for v in xref.values()] ans = (f"Of the {len(modeled)} PO activities, the ramco-chat runtime currently implements " f"{len(impl)} as a live, executable journey: {', '.join(impl) or 'none'} " f"(Create Direct Purchase Order). The others are recognised but not yet built. " f"This brain answers structural questions about all of them.") return {"answer": _clean(ans), "intent": "META_JOURNEY", "target": None, "chain": [], "citations": _cite({"source": "ramco-chat/kb/PO/journeys.json"}), "facts": {"implemented": impl}} def _meta_limits() -> dict: ans = ("I answer questions about the Purchase Order *system* — its processes, screens, " "APIs, stored procedures, the tables they populate, validations, and what each means " "— all grounded in the Ramco artifacts and manuals. I am not connected to a live " "database, so I can't return actual transactions (e.g. 'how many POs last month'). " "For live data, query the PO tables directly or the reporting layer.") return {"answer": ans, "intent": "META_LIMITS", "target": None, "chain": [], "citations": [], "facts": {}} def _doc_fallback(q) -> dict: d = v2().docs.best_definition(q, cues=q.lower().split()[:4], prefer_types=["arm", "procurement_manual", "system_manual"]) if d: return {"answer": _clean(d["text"]), "intent": "WHAT_IS", "target": None, "chain": [], "citations": _cite(d), "facts": {}} hits = v2().docs.search(q, k=3) if hits: return {"answer": "Closest documentation: " + hits[0]["text"][:280], "intent": "WHAT_IS", "target": None, "chain": [], "citations": _cite(*hits), "facts": {}} return {"answer": "I couldn't find that in the PO brain. Try an activity (create/amend/" "approve a PO), a table, an API, or ask what happens during a process.", "intent": "WHAT_IS", "target": None, "chain": [], "citations": [], "facts": {}} _PDF_FIX = [(r"\bt he\b", "the"), (r"\bo f\b", "of"), (r"\ba nd\b", "and"), (r"\bP O\b", "PO"), (r"\bi s\b", "is"), (r"\bt o\b", "to")] def _clean(s: str) -> str: s = re.sub(r"\s+", " ", s).strip() for pat, rep in _PDF_FIX: s = re.sub(pat, rep, s) return re.sub(r"\s+", " ", s).strip() if __name__ == "__main__": import sys qs = [" ".join(sys.argv[1:])] if len(sys.argv) > 1 else [ "What happens when someone creates a purchase order?", "Which tables get populated when I create a PO?", "Which API fires when I save a new PO?", "What are the variants of creating a purchase order?", "What validations run when I save a PO?", "What does PO_POSHD_SCHEDULE_DTL mean in business terms?", "Which SPs write to PO_POMAS_PUR_ORDER_HDR?", "Compare the data footprint of Create vs Amend PO", "Which PO activities are implemented in the chatbot runtime today?", "How many POs did we raise last month?", ] for q in qs: r = answer(q) print("Q:", q) print("→", r["answer"]) print(" [intent]", r["intent"], "| citations:", len(r["citations"]), "| chain steps:", len(r["chain"])) print()