""" FormFlow — Agentic Document-to-Action Pipeline with Human-in-the-Loop Approval ================================================================================ Three small, inspectable agents, chained together, with a mandatory human approval gate before anything is "actioned": 1. Extraction Agent -> pulls structured fields out of a document 2. Risk Assessment Agent -> runs explicit, visible rules over those fields 3. Decision Agent -> proposes Approve / Flag for Review / Reject Nothing is ever auto-executed. The human always clicks to confirm or override. No GPU required. No embedding model. No external API calls. Pure Python + regex for extraction and scoring, by design — every decision is traceable. """ import re from datetime import datetime import gradio as gr from pypdf import PdfReader # ----------------------------------------------------------------------- # In-memory audit log (resets each session — see README "What I'd build next" # for the note on persisting this to a real database in a production version) # ----------------------------------------------------------------------- AUDIT_LOG = [] # ----------------------------------------------------------------------- # Step 0: Document loading # ----------------------------------------------------------------------- def extract_text(file_path: str) -> str: if file_path.lower().endswith(".pdf"): reader = PdfReader(file_path) return "\n".join(page.extract_text() or "" for page in reader.pages) with open(file_path, "r", encoding="utf-8", errors="ignore") as f: return f.read() # ----------------------------------------------------------------------- # Step 1: Extraction Agent # ----------------------------------------------------------------------- def extraction_agent(text: str) -> dict: """ Rule-based structured field extraction. Deliberately regex/pattern-based rather than an LLM call, so every extracted value is directly traceable to a specific pattern in the source text. See README for the note on swapping this for an LLM-based extractor on messier documents. """ fields = {} def find(pattern, default="Not found", group=1, flags=re.IGNORECASE): m = re.search(pattern, text, flags) return m.group(group).strip() if m else default fields["vendor"] = find(r"From:\s*(.+)") fields["invoice_number"] = find(r"Invoice Number:\s*([A-Za-z0-9\-]+)") fields["invoice_date"] = find(r"Invoice Date:\s*([\d\-/]+)") fields["due_date"] = find(r"Due Date:\s*([\d\-/]+)") fields["payment_terms"] = find(r"Payment Terms:\s*(.+)") fields["total_due_raw"] = find(r"TOTAL DUE:\s*\$?([\d,]+\.\d{2})") fields["payment_method"] = find(r"Payment Method:\s*(.+)") fields["account_name"] = find(r"Account Name:\s*(.+)") fields["bank"] = find(r"Bank:\s*(.+)") fields["notes"] = find(r"Notes:\s*(.+?)(?:\n\n|\Z)", flags=re.IGNORECASE | re.DOTALL) # parse amount to a float for downstream risk scoring try: fields["total_due"] = float(fields["total_due_raw"].replace(",", "")) except (ValueError, AttributeError): fields["total_due"] = None # parse due date and invoice date to estimate the payment window in days fields["days_to_pay"] = None try: d1 = datetime.strptime(fields["invoice_date"], "%Y-%m-%d") d2 = datetime.strptime(fields["due_date"], "%Y-%m-%d") fields["days_to_pay"] = (d2 - d1).days except (ValueError, TypeError): pass return fields # ----------------------------------------------------------------------- # Step 2: Risk Assessment Agent — explicit, inspectable rules # ----------------------------------------------------------------------- RISK_RULES = [ { "name": "Large amount", "check": lambda f: f["total_due"] is not None and f["total_due"] > 20000, "weight": 2, "explain": "Invoice total exceeds $20,000 — above standard auto-approval threshold.", }, { "name": "Very short payment window", "check": lambda f: f["days_to_pay"] is not None and f["days_to_pay"] <= 2, "weight": 3, "explain": "Payment is due within 2 days or less of the invoice date — unusually urgent for a standard vendor invoice.", }, { "name": "Wire-transfer-only payment", "check": lambda f: "wire" in f["payment_method"].lower() and "only" in f["payment_method"].lower(), "weight": 2, "explain": "Payment method explicitly restricts to wire transfer only — a common fraud indicator, since wires are hard to reverse.", }, { "name": "Personal-sounding account name", "check": lambda f: bool(re.search(r"\bpersonal\b", f["account_name"], re.IGNORECASE)), "weight": 3, "explain": "Receiving account name is explicitly marked as a personal account rather than a registered business account.", }, { "name": "Urgency / avoid-contact language", "check": lambda f: bool(re.search(r"urgent|time-sensitive|do not contact|email only", f["notes"], re.IGNORECASE)), "weight": 3, "explain": "Notes field contains urgency or avoid-contact language — a classic social-engineering / invoice fraud pattern.", }, { "name": "Missing standard fields", "check": lambda f: f["total_due"] is None or f["vendor"] == "Not found", "weight": 1, "explain": "One or more standard invoice fields (vendor, total due) could not be confidently extracted.", }, { "name": "Offshore / unregistered bank", "check": lambda f: bool(re.search(r"offshore", f["bank"], re.IGNORECASE)), "weight": 2, "explain": "Receiving bank is described as offshore — higher scrutiny warranted for cross-border, less-traceable transfers.", }, ] def risk_assessment_agent(fields: dict): triggered = [] score = 0 for rule in RISK_RULES: try: if rule["check"](fields): triggered.append(rule) score += rule["weight"] except Exception: continue # a rule failing to evaluate (e.g. missing field) just doesn't fire return score, triggered # ----------------------------------------------------------------------- # Step 3: Decision Agent — proposes, never executes # ----------------------------------------------------------------------- def decision_agent(score: int): if score == 0: return "✅ APPROVE", "No risk indicators triggered. Recommended for standard approval." elif score <= 3: return "🟡 FLAG FOR MANUAL REVIEW", "Some risk indicators present. Recommend a second pair of eyes before approval." else: return "🔴 REJECT / ESCALATE", "Multiple strong risk indicators triggered. Recommend escalation to fraud/compliance review, not standard approval." # ----------------------------------------------------------------------- # Orchestration # ----------------------------------------------------------------------- def process_document(file_obj, sample_choice): try: if file_obj is not None: text = extract_text(file_obj.name) doc_label = f"Uploaded file: {file_obj.name.split('/')[-1]}" elif sample_choice == "Sample 2: Suspicious Invoice": text = extract_text("sample_invoice_suspicious.txt") doc_label = "Sample 2: Suspicious Invoice" else: text = extract_text("sample_invoice.txt") doc_label = "Sample 1: Normal Invoice" except Exception as e: return (f"❌ Could not read document: {e}", "", "", "", gr.update(visible=False), gr.update(visible=False), doc_label) fields = extraction_agent(text) score, triggered = risk_assessment_agent(fields) decision, decision_explain = decision_agent(score) # --- Extracted fields display --- fields_md = "### 📋 Extraction Agent — Structured Fields\n" fields_md += f"- **Vendor:** {fields['vendor']}\n" fields_md += f"- **Invoice #:** {fields['invoice_number']}\n" fields_md += f"- **Invoice date:** {fields['invoice_date']}\n" fields_md += f"- **Due date:** {fields['due_date']}\n" fields_md += f"- **Days to pay:** {fields['days_to_pay'] if fields['days_to_pay'] is not None else 'Could not calculate'}\n" fields_md += f"- **Total due:** ${fields['total_due']:,.2f}" if fields["total_due"] is not None else "- **Total due:** Not found" fields_md += f"\n- **Payment method:** {fields['payment_method']}\n" fields_md += f"- **Account name:** {fields['account_name']}\n" fields_md += f"- **Bank:** {fields['bank']}\n" # --- Risk assessment display --- risk_md = f"### ⚠️ Risk Assessment Agent — Score: {score}\n" if triggered: risk_md += f"**{len(triggered)} risk indicator(s) triggered:**\n\n" for rule in triggered: risk_md += f"- **{rule['name']}** (weight {rule['weight']}): {rule['explain']}\n" else: risk_md += "No risk indicators triggered.\n" # --- Decision display --- decision_md = f"### 🤖 Decision Agent — Proposed Action\n## {decision}\n\n{decision_explain}\n\n" decision_md += "**⏸️ This is a proposal only. Nothing has been logged or executed yet — confirm or override below.**" return ( fields_md, risk_md, decision_md, decision, # stored for the confirm/override buttons gr.update(visible=True), # show approval buttons gr.update(visible=True), # show audit log section doc_label, ) def confirm_decision(decision_text, human_choice, doc_label): timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") if human_choice == "confirm": entry = f"[{timestamp}] {doc_label} — AI recommended **{decision_text}** → Human **CONFIRMED** the recommendation." else: entry = f"[{timestamp}] {doc_label} — AI recommended **{decision_text}** → Human **OVERRODE** and approved anyway." AUDIT_LOG.insert(0, entry) log_display = "### 📝 Audit Log (this session)\n\n" + "\n\n".join(AUDIT_LOG) return log_display # ----------------------------------------------------------------------- # Gradio UI # ----------------------------------------------------------------------- with gr.Blocks(title="FormFlow") as demo: gr.Markdown( """ # 🧾 FormFlow — Agentic Document-to-Action Pipeline Upload an invoice (or try the samples below). Watch three agents work in sequence — **Extraction → Risk Assessment → Decision** — and notice that the AI only ever *proposes* an action. A human has to click to confirm or override before anything is logged. """ ) with gr.Row(): with gr.Column(scale=1): gr.Markdown("### 1. Load a document") sample_dropdown = gr.Dropdown( choices=["Sample 1: Normal Invoice", "Sample 2: Suspicious Invoice"], value="Sample 1: Normal Invoice", label="Try a sample", ) file_input = gr.File(label="...or upload your own .txt / .pdf", file_types=[".txt", ".pdf"]) run_btn = gr.Button("Run Pipeline", variant="primary") doc_label_box = gr.Textbox(label="Currently loaded", interactive=False) gr.Markdown( "**Tip:** run Sample 1 first to see a clean approval, then run " "Sample 2 to see the same pipeline catch five separate fraud " "indicators on a very different-looking invoice." ) with gr.Column(scale=2): fields_output = gr.Markdown() risk_output = gr.Markdown() decision_output = gr.Markdown() with gr.Row(visible=False) as approval_row: confirm_btn = gr.Button("✅ Confirm AI Recommendation", variant="primary") override_btn = gr.Button("⚠️ Override and Approve Anyway", variant="stop") with gr.Group(visible=False) as audit_group: audit_output = gr.Markdown("### 📝 Audit Log (this session)\n\n_No actions confirmed yet._") decision_state = gr.State("") def run_and_store(file_obj, sample_choice): fields_md, risk_md, decision_md, decision, btn_update, audit_update, doc_label = process_document(file_obj, sample_choice) return fields_md, risk_md, decision_md, decision, btn_update, audit_update, doc_label run_btn.click( fn=run_and_store, inputs=[file_input, sample_dropdown], outputs=[fields_output, risk_output, decision_output, decision_state, approval_row, audit_group, doc_label_box], ) confirm_btn.click( fn=lambda decision, doc_label: confirm_decision(decision, "confirm", doc_label), inputs=[decision_state, doc_label_box], outputs=[audit_output], ) override_btn.click( fn=lambda decision, doc_label: confirm_decision(decision, "override", doc_label), inputs=[decision_state, doc_label_box], outputs=[audit_output], ) if __name__ == "__main__": demo.launch()