""" agent.py — the banking agent assembled as a graph (Knowledge Module 03). The graph: START -> route ─┬─(policy)──> rag ────────────────> END ├─(balance)─> balance ─────────────> END ├─(transfer)> prepare_transfer -> human_review -*-> execute -> END └─(refuse)──> refuse ──────────────> END (*) interrupt_before=['human_review'] Safety properties demonstrated: - money facts come from tools (bank.py), never the LLM; - transfers PAUSE for human approval (interrupt) and resume; - tools enforce least-privilege (an injected cross-account transfer fails); - every step is appended to an immutable `audit` list in the state. PRODUCTION (LangGraph) equivalent shown inline in `route()` and `rag_node()`. """ from __future__ import annotations import re from agentgraph import END, START, Graph, StateGraph, Checkpointer from bank import AuthorizationError, CoreBanking # --- a tiny grounded policy responder (Lab 01 does this properly via RAG) --- POLICIES = { "dispute": "You can dispute a card transaction within 60 days of the statement date. [dispute-policy]", "fee": "The current account monthly fee is 25 AED, waived above a 3000 AED average balance. [fees]", "transfer_fee": "Local transfers are free; an international SWIFT transfer costs 50 AED. [fees-transfers]", "iban": "A UAE IBAN has 23 characters and is shown on your statement and in the app. [iban-info]", } def _intent(text: str) -> str: """Deterministic intent router — the stand-in for an LLM classification call. PRODUCTION: replace with a structured-output call to a cheap model, e.g. AzureChatOpenAI(model='gpt-4o-mini').with_structured_output(IntentSchema) returning one of {'policy','balance','transfer','refuse'}.""" t = text.lower() if any(w in t for w in ["balance", "how much do i have", "available funds"]): return "balance" if any(w in t for w in ["transfer", "send", "pay ", "move "]) and re.search(r"\d", t): return "transfer" if any(w in t for w in ["dispute", "fee", "charge", "iban", "swift", "waiv"]): return "policy" # out-of-scope / advice / unsafe -> refuse if any(w in t for w in ["invest", "stock", "legal advice", "tax", "loan should i"]): return "refuse" return "policy" # default to grounded policy lookup (which itself can abstain) class BankingAgent: def __init__(self, core: CoreBanking | None = None) -> None: self.core = core or CoreBanking() self.ckpt = Checkpointer() self.graph = self._build() # ---- nodes ---- def route(self, state: dict) -> dict: intent = _intent(state["message"]) return {"intent": intent, "audit": [{"step": "route", "intent": intent}]} def rag_node(self, state: dict) -> dict: """Grounded policy answer. PRODUCTION: call Lab 01's RagPipeline.ask(). Here we keyword-match the canned policies and ABSTAIN otherwise.""" t = state["message"].lower() for key, ans in POLICIES.items(): hit = ("dispute" in t and key == "dispute") or \ (("swift" in t or "international" in t) and key == "transfer_fee") or \ (("fee" in t or "charge" in t or "waiv" in t) and key == "fee") or \ ("iban" in t and key == "iban") if hit: return {"answer": ans, "audit": [{"step": "rag", "source": key}]} return {"answer": ("I don't have that in our policy documents; let me connect " "you with a human agent."), "abstained": True, "audit": [{"step": "rag", "source": None}]} def balance_node(self, state: dict) -> dict: """Money fact via TOOL CALL — never the LLM. Enforces ownership.""" try: res = self.core.get_balance(state["account_id"], state["user"]) ans = f"Your available balance is {res['balance']:.2f} {res['currency']}." return {"answer": ans, "tool_result": res, "audit": [{"step": "tool:get_balance", "args": state["account_id"], "ok": True}]} except AuthorizationError as e: return {"answer": "I can't access that account for you.", "audit": [{"step": "tool:get_balance", "ok": False, "error": str(e)}]} def prepare_transfer(self, state: dict) -> dict: """Parse the requested transfer into a pending_action. Does NOT execute.""" m = re.search(r"(\d+(?:\.\d+)?)", state["message"]) to = "AE_external" m_to = re.search(r"to\s+([A-Z]{2}\w+)", state["message"]) if m_to: to = m_to.group(1) amount = float(m.group(1)) if m else 0.0 pending = {"from": state["account_id"], "to": to, "amount": amount} return {"pending_action": pending, "answer": f"Please confirm: transfer {amount:.2f} AED to {to}.", "audit": [{"step": "prepare_transfer", "pending": pending}]} def human_review(self, state: dict) -> dict: """The node guarded by interrupt_before. By the time it RUNS, a human has approved (state['approved'] set on resume). If not approved, decline.""" if not state.get("approved"): return {"answer": "Transfer cancelled — not approved.", "audit": [{"step": "human_review", "approved": False, "approver": state.get("approver")}]} return {"audit": [{"step": "human_review", "approved": True, "approver": state.get("approver")}]} def execute_node(self, state: dict) -> dict: if not state.get("approved"): return {} # human_review already produced the cancellation message p = state["pending_action"] try: res = self.core.initiate_transfer(p["from"], state["user"], p["to"], p["amount"]) return {"answer": f"Done. Transferred {p['amount']:.2f} AED. " f"New balance {res['new_balance']:.2f} AED.", "tool_result": res, "audit": [{"step": "tool:initiate_transfer", "ok": True, "result": res}]} except (AuthorizationError, ValueError) as e: return {"answer": f"Transfer failed: {e}", "audit": [{"step": "tool:initiate_transfer", "ok": False, "error": str(e)}]} def refuse_node(self, state: dict) -> dict: return {"answer": ("I'm sorry, I can't help with that. I can assist with your " "accounts, payments, and our products. For investment, legal, " "or tax advice please speak to a qualified specialist."), "refused": True, "audit": [{"step": "refuse"}]} # ---- wiring ---- def _build(self) -> Graph: g = StateGraph() g.add_node("route", self.route) g.add_node("rag", self.rag_node) g.add_node("balance", self.balance_node) g.add_node("prepare_transfer", self.prepare_transfer) g.add_node("human_review", self.human_review) g.add_node("execute", self.execute_node) g.add_node("refuse", self.refuse_node) g.add_edge(START, "route") g.add_conditional_edges("route", lambda s: s["intent"], {"policy": "rag", "balance": "balance", "transfer": "prepare_transfer", "refuse": "refuse"}) g.add_edge("rag", END) g.add_edge("balance", END) g.add_edge("prepare_transfer", "human_review") g.add_edge("human_review", "execute") g.add_edge("execute", END) g.add_edge("refuse", END) # interrupt BEFORE human_review: pause for approval, resume after. return g.compile(checkpointer=self.ckpt, interrupt_before=["human_review"]) # ---- public API ---- def handle(self, message: str, user: str, account_id: str, thread_id: str) -> dict: return self.graph.invoke( {"message": message, "user": user, "account_id": account_id, "messages": [{"role": "user", "content": message}], "audit": []}, thread_id=thread_id) def approve(self, thread_id: str, approved: bool, approver: str) -> dict: return self.graph.resume(thread_id, {"approved": approved, "approver": approver})