โ ๏ธ No invoice state provided.
"
+
+ if isinstance(state, dict):
+ try:
+ state = InvoiceProcessingState(**state)
+ except Exception:
+ pass
+
+ # --- Extract fields safely ---
+ invoice = self._safe_invoice_dict(state) or {}
+ validation = self._safe_validation(state) or {}
+ risk = self._safe_risk(state) or {}
+ payment = (
+ state.payment_decision.model_dump(exclude_none=True)
+ if getattr(state, "payment_decision", None)
+ and hasattr(state.payment_decision, "model_dump")
+ else getattr(state, "payment_decision", {}) or {}
+ )
+
+ discrepancies = validation.get("discrepencies", []) # per schema
+
+ inv_id = invoice.get("invoice_number") or invoice.get("file_name") or ""
+ vendor = invoice.get("customer_name") or invoice.get("vendor_name") or "Unknown"
+ total = invoice.get("total") or invoice.get("amount") or 0
+
+ status = getattr(state, "overall_status", "unknown")
+ status_val = status.value if hasattr(status, "value") else str(status)
+
+ # --- Interpret status fields ---
+ risk_level = risk.get("risk_level")
+ if hasattr(risk_level, "value"):
+ risk_level = risk_level.value
+ risk_score = risk.get("risk_score", 0) or 0.0
+
+ val_status = validation.get("validation_status")
+ if hasattr(val_status, "value"):
+ val_status = val_status.value
+
+ payment_status = payment.get("status")
+ if hasattr(payment_status, "value"):
+ payment_status = payment_status.value
+
+ # --- Badge colors ---
+ colors = {
+ "VALIDATION": "#ffc107",
+ "RISK": (
+ "#ff1744" if str(risk_level).lower() == "critical"
+ else "#ff9800" if str(risk_level).lower() == "medium"
+ else "#4caf50"
+ ),
+ "PAYMENT": "#4caf50",
+ "AUDIT": "#2196f3",
+ }
+
+ # --- Header layout ---
+ header_html = f"""
+
+
+ Validation
+
+
+ Risk
+
+
+ Payment
+
+
+ Audit
+
+
Invoice: {inv_id}
",
+ f"Vendor: {vendor}
",
+ f"Amount: {_fmt(total)}
",
+ f"Status: {status_val}
",
+ "
",
+ f"Validation: {val_status or 'unknown'}
",
+ f"Risk Level: {risk_level or 'low'} ({risk_score})
",
+ f"Payment: {payment.get('decision', 'N/A')} ({payment_status or 'pending'})
",
+ ]
+
+ if discrepancies:
+ lines.append("Discrepancies Found:
")
+ for d in discrepancies:
+ field = d.get("field", "unknown")
+ expected = d.get("expected", "")
+ actual = d.get("actual", "")
+ lines.append(f"- {field}: expected
{expected}, got {actual} ")
+ lines.append("
")
+
+ # --- Recommendations ---
+ advice = []
+ if str(val_status).lower() == "invalid":
+ advice.append("โ Invoice failed validation โ requires manual review.")
+ elif str(val_status).lower() in ("partial", "partial_match"):
+ advice.append("โ ๏ธ Partial validation โ check mismatched fields.")
+ if str(risk_level).lower() == "critical":
+ advice.append("๐จ Critical risk detected โ immediate escalation required.")
+ elif str(risk_level).lower() == "medium":
+ advice.append("โ ๏ธ Medium risk โ consider manual review.")
+ if not advice:
+ advice.append("โ
No major issues detected. Proceed as usual.")
+
+ lines.append("Recommendation:
")
+ for a in advice:
+ lines.append(f"- {a}
")
+ lines.append("
")
+
+ explanation_html = header_html + "\n".join(lines)
+
+ # --- Gemini polishing (using your API key) ---
+ if self.use_gemini:
+ try:
+ import google.generativeai as genai
+ model = genai.GenerativeModel("models/gemini-2.0-flash")
+
+ prompt = f"""
+ You are a professional financial analyst.
+ Here is structured invoice data and an auto-generated explanation.
+
+ Invoice summary:
+ {json.dumps(invoice, indent=2)}
+
+ Validation details: {json.dumps(validation, indent=2)}
+ Risk assessment: {json.dumps(risk, indent=2)}
+ Payment info: {json.dumps(payment, indent=2)}
+
+ Rewrite the following explanation to sound executive-level, clear, and concise.
+ Use HTML for sections but do not remove any factual details.
+
+ Existing summary:
+ {explanation_html}
+ """
+ response = model.generate_content(prompt)
+ if response and getattr(response, "text", None):
+ return response.text.strip()
+ except Exception as e:
+ return explanation_html + f"Gemini explanation failed: {e}
"
+
+ return explanation_html
diff --git a/Project/agents/validation_agent.py b/Project/agents/validation_agent.py
new file mode 100644
index 0000000000000000000000000000000000000000..a2838cec0f91c9013ed09b773021a9439a364b58
--- /dev/null
+++ b/Project/agents/validation_agent.py
@@ -0,0 +1,357 @@
+
+"""Validation Agent for Invoice Processing"""
+
+# TODO: Implement agent
+import asyncio
+import os
+import pandas as pd
+from typing import Dict, Any, List, Tuple
+from fuzzywuzzy import fuzz
+import numpy as np
+import time
+from agents.base_agent import BaseAgent
+from state import (
+ InvoiceProcessingState, ValidationResult, ValidationStatus,
+ ProcessingStatus
+)
+from datetime import datetime, timedelta
+
+from utils.logger import StructuredLogger
+from difflib import SequenceMatcher
+
+class ValidationAgent(BaseAgent):
+ """Agent responsible for validating invoice data against purchase orders"""
+
+ health_history: List[Dict[str, Any]] = [] # global history for metrics
+
+ def __init__(self, config: Dict[str, Any] = None):
+ # pass
+ super().__init__(agent_name="validation_agent",config=config or {})
+ self.logger = StructuredLogger(__name__)
+ self.po_file = self.config.get("po_file","data/purchase_orders.csv")
+ self.tolerance = self.config.get("tolerance",0.05)
+ self.successful_executions = 0
+ self.failed_executions = 0
+ self.total_duration = 0.0
+ self.total_executions = 0
+ self.last_run = None
+ # self.match_threshold = self.config.get("match_threshold",80)
+
+ def _validate_preconditions(self, state: InvoiceProcessingState, workflow_type) -> bool:
+ # pass
+ if not state.invoice_data:
+ self.logger.logger.error("No invoice data available for validation.")
+ return False
+ return True
+
+ def _validate_postconditions(self, state: InvoiceProcessingState) -> bool:
+ # pass
+ return hasattr(state,'validation_result') and state.validation_result is not None
+
+ async def execute(self, state: InvoiceProcessingState, workflow_type) -> InvoiceProcessingState:
+ # pass
+ self.logger.logger.info(f"[ValidationAgent] Starting validation for {state.file_name}")
+ start_time = time.time()
+ try:
+ if not self._validate_preconditions(state, workflow_type):
+ state.status = ProcessingStatus.FAILED
+ self._log_decision(state,"Validation Failed","Precondition not met",confidence = 0.0)
+ return state
+ invoice_data = state.invoice_data
+ matching_pos = await self._find_matching_pos(invoice_data)
+ validation_result = await self._validate_against_pos(invoice_data,matching_pos)
+ state.validation_result = validation_result
+ state.current_agent = "validation_agent"
+ state.overall_status = ProcessingStatus.IN_PROGRESS
+
+ if self._should_escalate_validation(validation_result, invoice_data):
+ state.escalation_required = True
+ self._validate_postconditions(state)
+ self.successful_executions += 1
+ self.last_run = datetime.utcnow().isoformat()
+ # print("ValidationResult().confidence_score", state.validation_result.confidence_score)
+ self._log_decision(
+ state,
+ "Validation Successful",
+ "PDF text successfully validated and checked by AI",
+ state.validation_result.confidence_score,
+ state.process_id
+ )
+ return state
+ except Exception as e:
+ self.logger.logger.error(f"[ValidationAgent] Execution failed: {e}")
+ self.failed_executions += 1
+ state.overall_status = ProcessingStatus.FAILED
+ return state
+
+ finally:
+ duration = (time.time() - start_time) * 1000 # ms
+ self.total_executions += 1
+ self.total_duration += duration
+ self._record_health_metrics(duration)
+
+ def _load_purchase_orders(self) -> pd.DataFrame:
+ # pass
+ """load po data from csv"""
+ try:
+ df = pd.read_csv(self.po_file)
+ self.logger.logger.info(f"[ValidationAgent] Loaded {len(df)} purchase orders")
+ return df
+ except Exception as e:
+ self.logger.logger.error(f"[ValidationAgent] failed to load purchase order: {e}")
+ raise
+
+ async def _find_matching_pos(self, invoice_data) -> List[Dict[str, Any]]:
+ """find POs matching invoice order_id or fuzzy customer/items"""
+ po_df = self._load_purchase_orders()
+ matches = []
+ for _,po in po_df.iterrows():
+ customer_score = fuzz.token_sort_ratio(po["customer_name"], invoice_data.customer_name)
+ order_id_score = fuzz.token_sort_ratio(po["order_id"], invoice_data.order_id)
+ for item in invoice_data.item_details:
+ item_score = fuzz.token_sort_ratio(po["item_name"],item.item_name)
+ print(f"Compairing PO item {po['item_name']} with invoice item {item.item_name}: score = {item_score}")
+
+ if (customer_score >= 80) and (item_score >=80) and (order_id_score >=80) and (po['invoice_number'] == int(invoice_data.invoice_number)):
+ matches.append(po.to_dict())
+
+ print("matches.....", matches)
+ return matches
+
+
+ async def _validate_against_pos(self, invoice_data, matching_pos: List[Dict[str, Any]]) -> ValidationResult:
+ # pass
+
+ if not matching_pos:
+ return ValidationResult(po_found=False, validation_status='missing_po',validation_result='No matching purchase order found',
+ discrepancies = [],
+ confidence_score = 0.0)
+ po_data = matching_pos[0]
+ discrepancies = self._validate_item_against_po(invoice_data,po_data)
+ discrepancies += self._validate_totals(invoice_data,po_data)
+ actual_amount = [item.amount for item in invoice_data.item_details][0]
+ actual_quantity = [item.quantity for item in invoice_data.item_details][0]
+ actual_rate = [item.rate for item in invoice_data.item_details][0]
+ amount_diff = abs(actual_amount - po_data.get('expected_amount',0))
+ tolerance_limit = po_data.get('expected_amount',0)*self.tolerance
+ amount_match = amount_diff <= tolerance_limit
+
+ validation_result = ValidationResult(
+ po_found=True,
+ quantity_match=actual_quantity == po_data.get('quantity'),
+ rate_match=abs(actual_rate - po_data.get('rate', 0)) <= tolerance_limit,
+ amount_match=amount_match,
+ validation_status=ValidationStatus.NOT_STARTED, # temporary
+ validation_result="; ".join(discrepancies) if discrepancies else "All checks passed",
+ discrepencies=discrepancies,
+ confidence_score=0.0, # temporary
+ expected_amount=po_data.get('amount'),
+ po_data=po_data
+ )
+ validation_result.validation_status = self._determine_validation_status(validation_result)
+ validation_result.confidence_score = self._calculate_validation_confidence(validation_result, matching_pos, invoice_data)
+ return validation_result
+
+ def _validate_item_against_po(self, item, po_data: Dict[str, Any]) -> List[str]:
+ # pass
+ # print("itemmmmmmmmm", item.item_details.quantity)
+ print("po_-------------", po_data)
+ discrepancies = []
+ for item in item.item_details:
+ if item.quantity != po_data.get('quantity'):
+ discrepancies.append(f"Quantity mismatch: Expected {po_data['quantity']}, Found {item.quantity}")
+ if abs(item.rate - po_data.get('rate',0)) > po_data.get('rate',0)*self.tolerance:
+ discrepancies.append(f"Rate mismatch: Expected {po_data['rate']}, Found {item.rate}")
+ return discrepancies
+
+ def _validate_totals(self, invoice_data, po_data: Dict[str, Any]) -> List[str]:
+ # pass
+ discrepancies = []
+ expected = po_data.get('expected_amount',0)
+ actual = [item.amount for item in invoice_data.item_details][0]
+ diff = abs(expected-actual)
+ if diff > expected*self.tolerance:
+ discrepancies.append(f"Total amount mismatch: Expected {expected}, Actual {actual} (Difference:{diff:.2f})")
+ return discrepancies
+
+ def _calculate_validation_confidence(self, validation_result: ValidationResult,
+ matching_pos: List[Dict[str, Any]], invoice_data) -> float:
+ """
+ Compute an intelligent, weighted confidence score across 7 key dimensions:
+ invoice_number, order_id, customer_name, item_name, amount, rate, quantity.
+ Each field contributes based on importance.
+ """
+
+ if not validation_result.po_found or not matching_pos:
+ return 0.0
+
+ po_data = matching_pos[0]
+
+ # Extract PO (expected) values
+ expected = {
+ "invoice_number": po_data.get("invoice_number", ""),
+ "order_id": po_data.get("order_id", ""),
+ "customer_name": po_data.get("customer_name", ""),
+ "item_name": po_data.get("item_name", ""),
+ "amount": float(po_data.get("expected_amount", po_data.get("amount", 0))),
+ "rate": float(po_data.get("rate", 0)),
+ "quantity": float(po_data.get("quantity", 0))
+ }
+
+ # Extract actual (from invoice)
+ actual = {
+ "invoice_number": invoice_data.invoice_number,
+ "order_id": invoice_data.order_id,
+ "customer_name": invoice_data.customer_name,
+ }
+
+ # Handle line-item level (assuming single dominant item)
+ if invoice_data.item_details:
+ item = invoice_data.item_details[0]
+ actual.update({
+ "item_name": item.item_name,
+ "amount": float(item.amount or 0),
+ "rate": float(item.rate or 0),
+ "quantity": float(item.quantity or 0)
+ })
+
+ # Define weights intelligently (sum = 1)
+ weights = {
+ "invoice_number": 0.20,
+ "order_id": 0.15,
+ "customer_name": 0.05,
+ "item_name": 0.05,
+ "amount": 0.25,
+ "rate": 0.15,
+ "quantity": 0.15
+ }
+
+ # --- Similarity functions ---
+ def numeric_similarity(expected_val, actual_val):
+ if expected_val == 0:
+ return 1.0 if actual_val == 0 else 0.0
+ diff_ratio = abs(expected_val - actual_val) / (abs(expected_val) + 1e-6)
+ return max(0.0, 1.0 - diff_ratio)
+
+ def text_similarity(a, b):
+ return SequenceMatcher(None, str(a).lower(), str(b).lower()).ratio()
+
+ # --- Compute weighted similarities ---
+ weighted_scores = []
+ for field, weight in weights.items():
+ exp_val, act_val = expected.get(field), actual.get(field)
+
+ if isinstance(exp_val, (int, float)) and isinstance(act_val, (int, float)):
+ score = numeric_similarity(exp_val, act_val)
+ else:
+ score = text_similarity(exp_val, act_val)
+
+ weighted_scores.append(weight * score)
+
+ # Combine to final confidence
+ confidence = sum(weighted_scores)
+ confidence = round(confidence * 100, 2) # convert to %
+ confidence = max(0.0, min(confidence, 100.0)) # clamp 0โ100
+
+ self.logger.logger.debug(f"Validation Confidence (weighted): {confidence}%")
+ return confidence
+
+
+
+ def _determine_validation_status(self, validation_result: ValidationResult) -> ValidationStatus:
+ """
+ Determine the final validation status based on PO existence, discrepancies, and amount match.
+ """
+ if not validation_result.po_found:
+ return ValidationStatus.MISSING_PO
+
+ discrepancies_count = len(validation_result.discrepencies)
+
+ if discrepancies_count == 0 and validation_result.amount_match:
+ return ValidationStatus.VALID
+
+ if validation_result.amount_match and discrepancies_count <= 2:
+ return ValidationStatus.PARTIAL_MATCH
+
+ return ValidationStatus.INVALID
+
+
+ def _should_escalate_validation(self, validation_result: ValidationResult, invoice_data) -> bool:
+ # pass
+ return validation_result.validation_status in ['invalid','missing_po']
+
+ def _record_health_metrics(self, duration: float):
+ """Record the health metrics after each execution"""
+ success_rate = (
+ (self.successful_executions / self.total_executions) * 100
+ if self.total_executions > 0 else 0
+ )
+ avg_duration = (
+ self.total_duration / self.total_executions
+ if self.total_executions > 0 else 0
+ )
+
+ metrics = {
+ "Agent": "Validation Agent โ
",
+ "Executions": self.total_executions,
+ "Success Rate (%)": round(success_rate, 2),
+ "Avg Duration (ms)": round(avg_duration, 2),
+ "Total Failures": self.failed_executions,
+ # "Timestamp": datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S UTC"),
+ }
+ metrics_data = {}
+ executions = 0
+ success_rate = 0.0
+ avg_duration = 0.0
+ failures = 0
+ last_run = None
+
+ if self.metrics:
+ print("self.metrics from validation agent", self.metrics)
+ executions = self.metrics["processed"]
+ print("executions.....", executions)
+ avg_duration = self.metrics["avg_latency_ms"]
+ failures = self.metrics["errors"]
+ last_run = self.metrics["last_run_at"]
+ print("last_run.....", last_run)
+ success_rate = (executions - failures) / (executions + 1e-6)
+
+ # if last_run == None:
+ last_run = self.last_run
+
+ # 3. Health logic
+ overall_status = "๐ข Healthy"
+ if failures > 3:
+ overall_status = "๐ Degraded"
+ if executions > 0 and success_rate < 0.5:
+ overall_status = "๐ด Unhealthy"
+
+ print("metrics from val---....1", metrics)
+
+ metrics.update({
+ "Last Run": str(last_run) if last_run else "Not applicable",
+ "Overall Health": overall_status,
+ })
+ print("metrics from val---....", metrics)
+ # maintain up to last 50 records
+ ValidationAgent.health_history.append(metrics)
+ # ValidationAgent.health_history = ValidationAgent.health_history[-50:]
+
+ async def health_check(self) -> Dict[str, Any]:
+ """
+ Returns the health metrics summary for UI display.
+ """
+ await asyncio.sleep(0.05)
+ if not ValidationAgent.health_history:
+ return {
+ "Agent": "Validation Agent โ
",
+ "Executions": 0,
+ "Success Rate (%)": 0.0,
+ "Avg Duration (ms)": 0.0,
+ "Total Failures": 0,
+ }
+
+
+ latest = ValidationAgent.health_history[-1]
+ print("latest.....", latest)
+ return latest
diff --git a/Project/bounding_box.py b/Project/bounding_box.py
new file mode 100644
index 0000000000000000000000000000000000000000..f3ff8f97245c57e4b47f7b2de1db24f10cc43243
--- /dev/null
+++ b/Project/bounding_box.py
@@ -0,0 +1,138 @@
+
+import fitz # PyMuPDF
+import pandas as pd
+import os
+import re
+
+# === File paths ===
+DATA_DIR = os.path.join(os.getcwd(), "data")
+PDF_PATH = os.path.join(DATA_DIR, "invoices/Invoice-26.pdf") # Update for new PDF if needed
+CSV_PATH = os.path.join(DATA_DIR, "purchase_orders.csv")
+OUTPUT_PATH = os.path.join(DATA_DIR, "annotated_invoice.pdf")
+
+# === Field coordinate map (from your data) ===
+FIELD_BOXES = {
+ "invoice_number": (525, 55, 575, 75),
+ "order_id": (45, 470, 230, 490),
+ "customer_name": (40, 135, 100, 155),
+ "quantity": (370, 235, 385, 250),
+ "rate": (450, 235, 500, 250),
+ "expected_amount": (520, 360, 570, 375),
+}
+
+# === Step 1: Open PDF and extract text ===
+pdf = fitz.open(PDF_PATH)
+page = pdf[0]
+pdf_text = page.get_text()
+
+# === Step 2: Helper to extract fields ===
+def extract_field(pattern, text, group=1):
+ match = re.search(pattern, text, re.IGNORECASE)
+ return match.group(group).strip() if match else None
+
+# Extract key identifiers
+invoice_number_pdf = extract_field(r"#\s*(\d+)", pdf_text)
+order_id_pdf = extract_field(r"Order ID\s*[:\-]?\s*(\S+)", pdf_text)
+customer_name_pdf = extract_field(r"Bill To:\s*(.*)", pdf_text)
+
+# === Step 3: Read CSV and match correct row ===
+po_df = pd.read_csv(CSV_PATH)
+
+matched_row = po_df[
+ (po_df['invoice_number'].astype(str) == str(invoice_number_pdf))
+ | (po_df['order_id'] == order_id_pdf)
+]
+
+if matched_row.empty:
+ raise ValueError(f"No matching CSV row found for Invoice {invoice_number_pdf} / Order {order_id_pdf}")
+
+expected = matched_row.iloc[0].to_dict()
+expected = {k.lower(): str(v).strip() for k, v in expected.items()}
+
+print("โ
Loaded expected data from CSV for this PDF:")
+for k, v in expected.items():
+ print(f" {k}: {v}")
+
+# === Step 4: Extract fields from PDF ===
+invoice_data = {
+ "invoice_number": invoice_number_pdf,
+ "customer_name": customer_name_pdf,
+ "order_id": order_id_pdf,
+}
+
+# Numeric fields
+amounts = re.findall(r"\$?([\d,]+\.\d{2})", pdf_text)
+invoice_data["expected_amount"] = amounts[-1] if amounts else None
+
+# Extract first item (quantity, rate)
+item_lines = re.findall(
+ r"([A-Za-z0-9 ,\-]+)\s+(\d+)\s+\$?([\d,]+\.\d{2})\s+\$?([\d,]+\.\d{2})",
+ pdf_text,
+)
+if item_lines:
+ invoice_data["quantity"] = item_lines[0][1]
+ invoice_data["rate"] = item_lines[0][2]
+
+print("\nโ
Extracted data from PDF:")
+for k, v in invoice_data.items():
+ print(f" {k}: {v}")
+
+# === Step 5: Compare PDF vs CSV ===
+discrepancies = []
+
+def add_discrepancy(field, expected_val, found_val):
+ discrepancies.append({"field": field, "expected": expected_val, "found": found_val})
+
+# Compare string fields
+for field in ["invoice_number", "order_id", "customer_name"]:
+ if str(invoice_data.get(field, "")).strip() != str(expected.get(field, "")).strip():
+ add_discrepancy(field, expected.get(field, ""), invoice_data.get(field, ""))
+
+# Compare numeric fields
+for field in ["quantity", "rate", "expected_amount"]:
+ try:
+ found_val = float(str(invoice_data.get(field, 0)).replace(",", "").replace("$", ""))
+ expected_val = float(str(expected.get(field, 0)).replace(",", "").replace("$", ""))
+ if round(found_val, 2) != round(expected_val, 2):
+ add_discrepancy(field, expected_val, found_val)
+ except:
+ if str(invoice_data.get(field, "")) != str(expected.get(field, "")):
+ add_discrepancy(field, expected.get(field, ""), invoice_data.get(field, ""))
+
+# === Step 6: Annotate mismatched fields using fixed coordinates ===
+for d in discrepancies:
+ field = d["field"]
+ if field not in FIELD_BOXES:
+ print(f"โ ๏ธ No coordinates found for field '{field}' โ skipping annotation.")
+ continue
+
+ rect_coords = FIELD_BOXES[field]
+ rect = fitz.Rect(rect_coords)
+ expected_text = (
+ f"{float(d['expected']):,.2f}"
+ if field in ["quantity", "rate", "expected_amount"]
+ else str(d["expected"])
+ )
+
+ # Draw red bounding box
+ page.draw_rect(rect, color=(1, 0, 0), width=1.5)
+
+ # Add expected value below box
+ page.insert_text(
+ (rect.x0, rect.y1 + 10),
+ expected_text,
+ fontsize=9,
+ color=(1, 0, 0),
+ )
+
+pdf.save(OUTPUT_PATH)
+pdf.close()
+
+print("\nโ
Annotated invoice saved at:", OUTPUT_PATH)
+
+if discrepancies:
+ print("\nโ ๏ธ Mismatches found:")
+ for d in discrepancies:
+ print(f" - {d['field']}: expected {d['expected']}, found {d['found']}")
+else:
+ print("\nโ
No mismatches found! Invoice matches CSV.")
diff --git a/Project/data/annotated_invoice.pdf b/Project/data/annotated_invoice.pdf
new file mode 100644
index 0000000000000000000000000000000000000000..a20e05135c4536da162a516d493e6e18128009b5
Binary files /dev/null and b/Project/data/annotated_invoice.pdf differ
diff --git a/Project/data/invoices/Invoice-01.pdf b/Project/data/invoices/Invoice-01.pdf
new file mode 100644
index 0000000000000000000000000000000000000000..ecfdf68222f1e900880bc15e51eaf214fa276b8c
Binary files /dev/null and b/Project/data/invoices/Invoice-01.pdf differ
diff --git a/Project/data/invoices/Invoice-02.pdf b/Project/data/invoices/Invoice-02.pdf
new file mode 100644
index 0000000000000000000000000000000000000000..af39020d574e8806478b1ca8eae374c7d46898f8
Binary files /dev/null and b/Project/data/invoices/Invoice-02.pdf differ
diff --git a/Project/data/invoices/Invoice-03.pdf b/Project/data/invoices/Invoice-03.pdf
new file mode 100644
index 0000000000000000000000000000000000000000..5603fb506eb5ec9dbc834ef01ca66dd96fb5e19d
Binary files /dev/null and b/Project/data/invoices/Invoice-03.pdf differ
diff --git a/Project/data/invoices/Invoice-04.pdf b/Project/data/invoices/Invoice-04.pdf
new file mode 100644
index 0000000000000000000000000000000000000000..7feeced3584889b6bf57fbf9bbc6bc8204b4ac1a
Binary files /dev/null and b/Project/data/invoices/Invoice-04.pdf differ
diff --git a/Project/data/invoices/Invoice-05.pdf b/Project/data/invoices/Invoice-05.pdf
new file mode 100644
index 0000000000000000000000000000000000000000..adaf933183c519f0a456db7cca2a1ca51fedc118
Binary files /dev/null and b/Project/data/invoices/Invoice-05.pdf differ
diff --git a/Project/data/invoices/Invoice-06.pdf b/Project/data/invoices/Invoice-06.pdf
new file mode 100644
index 0000000000000000000000000000000000000000..26f48dd42d5b837187a204a5f4e5220d2f13b803
Binary files /dev/null and b/Project/data/invoices/Invoice-06.pdf differ
diff --git a/Project/data/invoices/Invoice-07.pdf b/Project/data/invoices/Invoice-07.pdf
new file mode 100644
index 0000000000000000000000000000000000000000..ab4f2f05613c5d30a131e8cfda681ee46cbf99e5
Binary files /dev/null and b/Project/data/invoices/Invoice-07.pdf differ
diff --git a/Project/data/invoices/Invoice-08.pdf b/Project/data/invoices/Invoice-08.pdf
new file mode 100644
index 0000000000000000000000000000000000000000..ad40433bbb70cbf264ffa55c816473d6bdd23ba6
Binary files /dev/null and b/Project/data/invoices/Invoice-08.pdf differ
diff --git a/Project/data/invoices/Invoice-09.pdf b/Project/data/invoices/Invoice-09.pdf
new file mode 100644
index 0000000000000000000000000000000000000000..e6948d8bae8968946feaae6c85d25cbef1b4063c
Binary files /dev/null and b/Project/data/invoices/Invoice-09.pdf differ
diff --git a/Project/data/invoices/Invoice-10.pdf b/Project/data/invoices/Invoice-10.pdf
new file mode 100644
index 0000000000000000000000000000000000000000..ced3999cee25b765cd9e8ada1811f916eb8b81e5
Binary files /dev/null and b/Project/data/invoices/Invoice-10.pdf differ
diff --git a/Project/data/invoices/Invoice-11.pdf b/Project/data/invoices/Invoice-11.pdf
new file mode 100644
index 0000000000000000000000000000000000000000..9385825ca388da1438aa39bdddb1dd64bcc54cf1
Binary files /dev/null and b/Project/data/invoices/Invoice-11.pdf differ
diff --git a/Project/data/invoices/Invoice-12.pdf b/Project/data/invoices/Invoice-12.pdf
new file mode 100644
index 0000000000000000000000000000000000000000..17d155ea17c3bba1824a0d2160e2dfbea3562e64
Binary files /dev/null and b/Project/data/invoices/Invoice-12.pdf differ
diff --git a/Project/data/invoices/Invoice-13.pdf b/Project/data/invoices/Invoice-13.pdf
new file mode 100644
index 0000000000000000000000000000000000000000..a621cffdf971687b23a08cffddd689aef8bbd689
Binary files /dev/null and b/Project/data/invoices/Invoice-13.pdf differ
diff --git a/Project/data/invoices/Invoice-14.pdf b/Project/data/invoices/Invoice-14.pdf
new file mode 100644
index 0000000000000000000000000000000000000000..75e2d36460d64abb542313c664fad631b0bfca7c
Binary files /dev/null and b/Project/data/invoices/Invoice-14.pdf differ
diff --git a/Project/data/invoices/Invoice-15.pdf b/Project/data/invoices/Invoice-15.pdf
new file mode 100644
index 0000000000000000000000000000000000000000..28789650578358643d112260521d6e43ecee7504
Binary files /dev/null and b/Project/data/invoices/Invoice-15.pdf differ
diff --git a/Project/data/invoices/Invoice-16.pdf b/Project/data/invoices/Invoice-16.pdf
new file mode 100644
index 0000000000000000000000000000000000000000..06f36701897c8771759e617b78befd2921a6ad74
Binary files /dev/null and b/Project/data/invoices/Invoice-16.pdf differ
diff --git a/Project/data/invoices/Invoice-17.pdf b/Project/data/invoices/Invoice-17.pdf
new file mode 100644
index 0000000000000000000000000000000000000000..71617e61b831fa7a074fb8381f46cec5710f8a9b
Binary files /dev/null and b/Project/data/invoices/Invoice-17.pdf differ
diff --git a/Project/data/invoices/Invoice-18.pdf b/Project/data/invoices/Invoice-18.pdf
new file mode 100644
index 0000000000000000000000000000000000000000..39e3d0ace78d0b82593cb6b3588bfd5b3a3678af
Binary files /dev/null and b/Project/data/invoices/Invoice-18.pdf differ
diff --git a/Project/data/invoices/Invoice-19.pdf b/Project/data/invoices/Invoice-19.pdf
new file mode 100644
index 0000000000000000000000000000000000000000..d73458b574c054b574be69524a7e0b3b710000df
Binary files /dev/null and b/Project/data/invoices/Invoice-19.pdf differ
diff --git a/Project/data/invoices/Invoice-20.pdf b/Project/data/invoices/Invoice-20.pdf
new file mode 100644
index 0000000000000000000000000000000000000000..dc2ecdc80c912aaca920cccdbf47a5f2d58fe700
Binary files /dev/null and b/Project/data/invoices/Invoice-20.pdf differ
diff --git a/Project/data/invoices/Invoice-21.pdf b/Project/data/invoices/Invoice-21.pdf
new file mode 100644
index 0000000000000000000000000000000000000000..65d3410764a8142ff633219d8855d91c4943925d
Binary files /dev/null and b/Project/data/invoices/Invoice-21.pdf differ
diff --git a/Project/data/invoices/Invoice-22.pdf b/Project/data/invoices/Invoice-22.pdf
new file mode 100644
index 0000000000000000000000000000000000000000..e466db6fd8d551db07a3a3bc79664342d95c73b6
Binary files /dev/null and b/Project/data/invoices/Invoice-22.pdf differ
diff --git a/Project/data/invoices/Invoice-23.pdf b/Project/data/invoices/Invoice-23.pdf
new file mode 100644
index 0000000000000000000000000000000000000000..63591097449a7710ddd5a1f62f3233ab968f1593
Binary files /dev/null and b/Project/data/invoices/Invoice-23.pdf differ
diff --git a/Project/data/invoices/Invoice-24.pdf b/Project/data/invoices/Invoice-24.pdf
new file mode 100644
index 0000000000000000000000000000000000000000..c273d2281738262282320df6a0a3de0b1600e865
Binary files /dev/null and b/Project/data/invoices/Invoice-24.pdf differ
diff --git a/Project/data/invoices/Invoice-25.pdf b/Project/data/invoices/Invoice-25.pdf
new file mode 100644
index 0000000000000000000000000000000000000000..35a6e9d60acc756a73d0bbd4087f8958c0100cb7
Binary files /dev/null and b/Project/data/invoices/Invoice-25.pdf differ
diff --git a/Project/data/invoices/Invoice-26.pdf b/Project/data/invoices/Invoice-26.pdf
new file mode 100644
index 0000000000000000000000000000000000000000..d3ec4bf4986ca498f45b1730f7fcd875cae1e160
Binary files /dev/null and b/Project/data/invoices/Invoice-26.pdf differ
diff --git a/Project/data/invoices/Invoice-27.pdf b/Project/data/invoices/Invoice-27.pdf
new file mode 100644
index 0000000000000000000000000000000000000000..ce9a2aea421990762b0a48a557bd2bc0a4f4fee1
Binary files /dev/null and b/Project/data/invoices/Invoice-27.pdf differ
diff --git a/Project/data/invoices/Invoice-28.pdf b/Project/data/invoices/Invoice-28.pdf
new file mode 100644
index 0000000000000000000000000000000000000000..4e6200886dbf354840302735c39124ec77e177c8
Binary files /dev/null and b/Project/data/invoices/Invoice-28.pdf differ
diff --git a/Project/data/invoices/Invoice-29.pdf b/Project/data/invoices/Invoice-29.pdf
new file mode 100644
index 0000000000000000000000000000000000000000..09d0d0d9d8f6d4b8f82b486ce66b6b40b659601d
Binary files /dev/null and b/Project/data/invoices/Invoice-29.pdf differ
diff --git a/Project/data/invoices/Invoice-30.pdf b/Project/data/invoices/Invoice-30.pdf
new file mode 100644
index 0000000000000000000000000000000000000000..c1c6e7b027cb027268c97e838e18c439399b06b7
Binary files /dev/null and b/Project/data/invoices/Invoice-30.pdf differ
diff --git a/Project/data/invoices/Invoice-31.pdf b/Project/data/invoices/Invoice-31.pdf
new file mode 100644
index 0000000000000000000000000000000000000000..4a9030fe15ee4bc2720edd2ff1ecce8c6d885c65
Binary files /dev/null and b/Project/data/invoices/Invoice-31.pdf differ
diff --git a/Project/data/invoices/Invoice-32.pdf b/Project/data/invoices/Invoice-32.pdf
new file mode 100644
index 0000000000000000000000000000000000000000..f5f569c4b55b39f462f2168a6ba21d0cf8739f76
Binary files /dev/null and b/Project/data/invoices/Invoice-32.pdf differ
diff --git a/Project/data/invoices/Invoice-33.pdf b/Project/data/invoices/Invoice-33.pdf
new file mode 100644
index 0000000000000000000000000000000000000000..92736aa3e768a0c28e833e93e32f10dde085a561
Binary files /dev/null and b/Project/data/invoices/Invoice-33.pdf differ
diff --git a/Project/data/invoices/Invoice-34.pdf b/Project/data/invoices/Invoice-34.pdf
new file mode 100644
index 0000000000000000000000000000000000000000..894fa7f945db3d963abd91f7463717e38533a1a1
Binary files /dev/null and b/Project/data/invoices/Invoice-34.pdf differ
diff --git a/Project/data/invoices/Invoice-35.pdf b/Project/data/invoices/Invoice-35.pdf
new file mode 100644
index 0000000000000000000000000000000000000000..9bfc25e9eef27c1efe1863dc7e8a863620f9b5e5
Binary files /dev/null and b/Project/data/invoices/Invoice-35.pdf differ
diff --git a/Project/data/invoices/Invoice-36.pdf b/Project/data/invoices/Invoice-36.pdf
new file mode 100644
index 0000000000000000000000000000000000000000..b450493b58abae5521bc423b6875a0d15ee8d7ac
Binary files /dev/null and b/Project/data/invoices/Invoice-36.pdf differ
diff --git a/Project/data/invoices/Invoice-37.pdf b/Project/data/invoices/Invoice-37.pdf
new file mode 100644
index 0000000000000000000000000000000000000000..c782b45a4be1dbdf5a3dc4f132537a3e56b0b28b
Binary files /dev/null and b/Project/data/invoices/Invoice-37.pdf differ
diff --git a/Project/data/invoices/Invoice-38.pdf b/Project/data/invoices/Invoice-38.pdf
new file mode 100644
index 0000000000000000000000000000000000000000..a688a8442c69c534f0b409c9781ee5b2bc348d53
Binary files /dev/null and b/Project/data/invoices/Invoice-38.pdf differ
diff --git a/Project/data/invoices/Invoice-39.pdf b/Project/data/invoices/Invoice-39.pdf
new file mode 100644
index 0000000000000000000000000000000000000000..0f40061c0296c2e90c6316436bd10573020f6f8d
Binary files /dev/null and b/Project/data/invoices/Invoice-39.pdf differ
diff --git a/Project/data/invoices/Invoice-40.pdf b/Project/data/invoices/Invoice-40.pdf
new file mode 100644
index 0000000000000000000000000000000000000000..c5edbede858a675e3e45e144d590a10378ce8c24
Binary files /dev/null and b/Project/data/invoices/Invoice-40.pdf differ
diff --git a/Project/data/invoices/Invoice-41.pdf b/Project/data/invoices/Invoice-41.pdf
new file mode 100644
index 0000000000000000000000000000000000000000..04c75ac6b1998098274cd4964911b6f116d523fc
Binary files /dev/null and b/Project/data/invoices/Invoice-41.pdf differ
diff --git a/Project/data/invoices/Invoice-42.pdf b/Project/data/invoices/Invoice-42.pdf
new file mode 100644
index 0000000000000000000000000000000000000000..8f81787e0a617a41b201d99a2aad37b5f6232747
Binary files /dev/null and b/Project/data/invoices/Invoice-42.pdf differ
diff --git a/Project/data/invoices/Invoice-43.pdf b/Project/data/invoices/Invoice-43.pdf
new file mode 100644
index 0000000000000000000000000000000000000000..3c0ef0c6fe119a042f0009a8c95fcf6f568c6d08
Binary files /dev/null and b/Project/data/invoices/Invoice-43.pdf differ
diff --git a/Project/data/invoices/Invoice-44.pdf b/Project/data/invoices/Invoice-44.pdf
new file mode 100644
index 0000000000000000000000000000000000000000..864221ce79c3fed877686aaa45edf41828f99c4a
Binary files /dev/null and b/Project/data/invoices/Invoice-44.pdf differ
diff --git a/Project/data/invoices/Invoice-45.pdf b/Project/data/invoices/Invoice-45.pdf
new file mode 100644
index 0000000000000000000000000000000000000000..27c2b02e52b9f4b0a97d106fae7e1e0daa3d0928
Binary files /dev/null and b/Project/data/invoices/Invoice-45.pdf differ
diff --git a/Project/data/invoices/Invoice-46.pdf b/Project/data/invoices/Invoice-46.pdf
new file mode 100644
index 0000000000000000000000000000000000000000..43129ec8ef92695e52c8fb4cc778712b7fbd3b69
Binary files /dev/null and b/Project/data/invoices/Invoice-46.pdf differ
diff --git a/Project/data/purchase_orders.csv b/Project/data/purchase_orders.csv
new file mode 100644
index 0000000000000000000000000000000000000000..89f7c57deecc26d33a8caf9e96cfb5c6cb71e665
--- /dev/null
+++ b/Project/data/purchase_orders.csv
@@ -0,0 +1,26 @@
+invoice_number,order_id,customer_name,item_name,quantity,rate,expected_amount
+14021,ES-2025-BE11335139-41340,Bill Eplett,"Canon Wireless Fax, Laser Copiers, Technology, TEC-CO-3710",5,1893.30,9466.5
+6459,MX-2025-DK1298539-41339,Darren Koutras,"KitchenAid Stove, Silver Appliances, Office Supplies, OFF-AP-4966",4,1903.0,7612.0
+24450,IN-2025-GM146807-41338,Greg Matthias,"Hon Executive Leather Armchair, Adjustable Chairs, Furniture, FUR-CH-4654",1,409.24,409.24
+14130,ES-2025-LT1711045-41340,Liz Thompson,"Safco 3-Shelf Cabinet, Mobile Bookcases, Furniture, FUR-BO-5746",8,1079.57,8636.56
+24429,IN-2025-MZ173357-41340,Maria Zettner,"Hon Rocking Chair, Black Chairs, Furniture, FUR-CH-4682",4,461.48,1845.94
+10963,ES-2025-SW202458-41340,Scot Wooten,"Hewlett Fax Machine, Color Copiers, Technology, TEC-CO-4575",3,1285.44,3856.32
+22091,IN-2025-GB145307-41338,George Bell,"Safco Classic Bookcase, Metal Bookcases, Furniture, FUR-BO-5760",9,2756.94,24812.46
+36552,CA-2025-AH10195140-41338,Alan Haines,"36X48 HARDFLOOR CHAIRMAT Furnishings, Furniture, FUR-FU-2864",2,33.57,67.14
+36551,CA-2025-AH10195140-41338,Alan Haines,"Situations Contoured Folding Chairs, 4/Set Chairs, Furniture, FUR-CH-6016",2,99.37,198.74
+9063,MX-2025-PO1885051-41338,Patrick O'Brill,"Novimex Executive Leather Armchair, Adjustable Chairs, Furniture, FUR-CH-5378",2,607.36,1214.72
+28557,IN-2025-BT1153058-41331,Bradley Talbott,"Barricks Conference Table, Fully Assembled Tables, Furniture, FUR-TA-3344",3,5451.3,16353.9
+11001,ES-2025-RA19285120-41335,Ralph Arnett,"StarTech Inkjet, Wireless Machines, Technology, TEC-MA-6142",3,815.1,2445.31
+11000,ES-2025-RA19285120-41335,Ralph Arnett,"Canon Fax and Copier, Laser Copiers, Technology, TEC-CO-3685",11,2104.74,23152.14
+18509,ES-2025-JC15340139-41334,Jasper Cacioppo,"Cisco Audio Dock, Full Size Phones, Technology, TEC-PH-3785",4,733.44,2933.76
+49650,SG-2025-EH4005111-41331,Erica Hernandez,"Brother Fax Machine, High-Speed Copiers, Technology, TEC-CO-3597",2,633.48,1266.96
+22988,IN-2025-RE194507-41329,Richard Eichhorn,"Safco Library with Doors, Metal Bookcases, Furniture, FUR-BO-5785",2,700.65,1401.3
+22999,IN-2025-BP1123058-41329,Benjamin Patterson,"Office Star Executive Leather Armchair, Red Chairs, Furniture, FUR-CH-5443",4,1878.72,7514.88
+22711,IN-2025-DL1333058-41328,Denise Leinenbach,"Samsung Smart Phone, Full Size Phones, Technology, TEC-PH-5840",5,3187.2,15936.0
+11502,ES-2025-PB19210139-41327,Phillip Breyer,"Enermax Keyboard, Erganomic Accessories, Technology, TEC-AC-4156",6,487.62,2925.72
+49674,UP-2025-AH10030137-41325,Aaron Hawkins,"Hon Rocking Chair, Black Chairs, Furniture, FUR-CH-4682",8,1025.52,8204.16
+35513,CA-2025-AB10600140-41322,Ann Blume,"Smead Alpha-Z Color-Coded Second Alphabetical Labels and Starter Set Labels, Office Supplies, OFF-LA-6022",3,9.24,27.72
+20145,ES-2025-BD11560139-41327,Brendan Dodson,"Hon Executive Leather Armchair, Black Chairs, Furniture, FUR-CH-4655",5,2285.70,11428.50
+18517,ES-2025-FO1430545-41322,Frank Olsen,"Belkin Numeric Keypad, Bluetooth Accessories, Technology, TEC-AC-3396",10,579.60,5796.00
+28863,IN-2025-ME1801027-41322,Michelle Ellison,"Office Star Executive Leather Armchair, Adjustable Chairs, Furniture, FUR-CH-5441",1,930.00,930.00
+23981,IN-2025-NG1843058-41319,Nathan Gelder,"SAFCO Chairmat, Set of Two Chairs, Furniture, FUR-CH-5759",4,248.28,993.12
\ No newline at end of file
diff --git a/Project/graph.py b/Project/graph.py
new file mode 100644
index 0000000000000000000000000000000000000000..f609e3500b6ff6100054e444a00eb8bd89e8abb8
--- /dev/null
+++ b/Project/graph.py
@@ -0,0 +1,487 @@
+"""LangGraph workflow orchestrator"""
+# TODO: Implement graph workflow
+
+import asyncio
+import uuid # extra import
+from typing import Dict, Any, List, Optional, Literal
+from datetime import datetime
+from langgraph.graph import StateGraph, END
+from langgraph.checkpoint.memory import MemorySaver
+
+from state import (
+ InvoiceProcessingState, ProcessingStatus, ValidationStatus,
+ RiskLevel, PaymentStatus, WORKFLOW_CONFIGS
+)
+from agents.base_agent import agent_registry
+from agents.document_agent import DocumentAgent
+from agents.validation_agent import ValidationAgent
+from agents.risk_agent import RiskAgent
+from agents.payment_agent import PaymentAgent
+from agents.audit_agent import AuditAgent
+from agents.escalation_agent import EscalationAgent
+from utils.logger import StructuredLogger
+
+
+class InvoiceProcessingGraph:
+ """Graph orchestrator"""
+
+ def __init__(self, config: Dict[str, Any] = None):
+ self.logger = StructuredLogger("InvoiceProcessingGraph")
+ self.config = config or {}
+ #Simple in-memory store for process states (process_id -> InvoiceProcessingState)
+ self._process_store: Dict[str, InvoiceProcessingState] = {}
+ #Register and initialize agents
+ self._initialize_agents()
+ try:
+ self.graph = self._create_workflow_graph()
+ self.compiled_graph = self.graph.compile(checkpointer=MemorySaver())
+ self.logger.logger.info("InvoiceProcessingGraph initialized successfully with compiled graph.")
+ except Exception as e:
+ self.logger.logger.warning(f"Failed to fully build graph nodes: {e} โ exposing empty StateGraph")
+ self.graph = StateGraph("invoice_processing_graph_fallback")
+
+ def _initialize_agents(self):
+ """Instantiate and register agent instances in the global registry"""
+ #create agent instances (idempotent - replace if already registered)
+ agents = [
+ DocumentAgent(),
+ ValidationAgent(),
+ RiskAgent(),
+ PaymentAgent(),
+ AuditAgent(),
+ EscalationAgent(),
+ ]
+ for agent in agents:
+ agent_registry.register(agent)
+ self.logger.logger.info(f"Registered agents: {agent_registry.list_agents()}")
+
+ def _create_workflow_graph(self) -> StateGraph:
+ """
+ Build a LangGraph StateGraph with conditional routing:
+ Each node executes its corresponding agent and determines
+ the next node based on runtime logic (risk, validation, etc.)
+ """
+
+ graph = StateGraph("invoice_processing_graph")
+
+ # NODE DEFINITIONS
+ async def node_document(state: InvoiceProcessingState):
+ state = await self._document_agent_node(state)
+ next_node = self._route_after_document(state)
+ return next_node, state
+
+ async def node_validation(state: InvoiceProcessingState):
+ state = await self._validation_agent_node(state)
+ next_node = self._route_after_validation(state)
+ return next_node, state
+
+ async def node_risk(state: InvoiceProcessingState):
+ state = await self._risk_agent_node(state)
+ next_node = self._route_after_risk(state)
+ return next_node, state
+
+ async def node_payment(state: InvoiceProcessingState):
+ state = await self._payment_agent_node(state)
+ next_node = self._route_after_payment(state)
+ return next_node, state
+
+ async def node_audit(state: InvoiceProcessingState):
+ state = await self._audit_agent_node(state)
+ next_node = self._route_after_audit(state)
+ return next_node, state
+
+ async def node_escalation(state: InvoiceProcessingState):
+ state = await self._escalation_agent_node(state)
+ next_node = self._route_after_escalation(state)
+ return next_node, state
+
+ async def node_human_review(state: InvoiceProcessingState):
+ state = await self._human_review_node(state)
+ next_node = self._route_after_human_review(state)
+ return next_node, state
+
+ async def node_end(state: InvoiceProcessingState):
+ self.logger.logger.info(f"Invoice {state.invoice_id} completed at {state.updated_at}")
+ return "end", state
+
+ # REGISTER NODES
+ for name, func in {
+ "document": node_document,
+ "validation": node_validation,
+ "risk": node_risk,
+ "payment": node_payment,
+ "audit": node_audit,
+ "escalation": node_escalation,
+ "human_review": node_human_review,
+ "end": node_end,
+ }.items():
+ try:
+ graph.add_node(name, func)
+ except Exception:
+ # fallback if add_node signature differs
+ setattr(graph, name, func)
+
+ # ADD EDGES (DEFAULT PATHS)
+ try:
+ graph.add_edge("document", "validation")
+ graph.add_edge("validation", "risk")
+ graph.add_edge("risk", "payment")
+ graph.add_edge("payment", "audit")
+ graph.add_edge("audit", "end")
+ # Alternative / exception flows
+ graph.add_edge("document", "escalation")
+ graph.add_edge("validation", "escalation")
+ graph.add_edge("risk", "escalation")
+ graph.add_edge("escalation", "human_review")
+ graph.add_edge("human_review", "end")
+
+ graph.set_entry_point("document")
+ except Exception as ex:
+ self.logger.logger.warning(f"Edge registration failed: {ex}")
+
+ self.logger.logger.info("Conditional workflow graph built successfully.")
+ return graph
+
+
+ async def _document_agent_node(self, state: InvoiceProcessingState, workflow_type) -> InvoiceProcessingState:
+ agent: DocumentAgent = agent_registry.get("document_agent")
+ print("agent from doc", agent)
+ if not agent:
+ agent = DocumentAgent()
+ agent_registry.register(agent)
+ print("Registry instance ID in graph:", id(agent_registry))
+
+ return await agent.run(state, workflow_type)
+
+ async def _validation_agent_node(self, state: InvoiceProcessingState, workflow_type) -> InvoiceProcessingState:
+ agent: ValidationAgent = agent_registry.get("validation_agent")
+ print("agent from val", agent)
+ if not agent:
+ agent = ValidationAgent()
+ agent_registry.register(agent)
+ return await agent.run(state, workflow_type)
+
+ async def _risk_agent_node(self, state: InvoiceProcessingState, workflow_type) -> InvoiceProcessingState:
+ agent: RiskAgent = agent_registry.get("risk_agent")
+ if not agent:
+ agent = RiskAgent()
+ agent_registry.register(agent)
+ return await agent.run(state, workflow_type)
+
+ async def _payment_agent_node(self, state: InvoiceProcessingState, workflow_type) -> InvoiceProcessingState:
+ agent: PaymentAgent = agent_registry.get("payment_agent")
+ if not agent:
+ agent = PaymentAgent()
+ agent_registry.register(agent)
+ return await agent.run(state, workflow_type)
+
+ async def _audit_agent_node(self, state: InvoiceProcessingState, workflow_type) -> InvoiceProcessingState:
+ agent: AuditAgent = agent_registry.get("audit_agent")
+ if not agent:
+ agent = AuditAgent()
+ agent_registry.register(agent)
+ return await agent.run(state, workflow_type)
+
+ async def _escalation_agent_node(self, state: InvoiceProcessingState, workflow_type) -> InvoiceProcessingState:
+ agent: EscalationAgent = agent_registry.get("escalation_agent")
+ if not agent:
+ agent = EscalationAgent()
+ agent_registry.register(agent)
+ return await agent.run(state, workflow_type)
+
+ async def _human_review_node(self, state: InvoiceProcessingState, workflow_type) -> InvoiceProcessingState:
+ #Reusing escalation agent's human-in-the-loop or simply marking for manual review
+ agent: EscalationAgent = agent_registry.get("escalation_agent")
+ if not agent:
+ agent = EscalationAgent()
+ agent_registry.register(agent)
+ return await agent.run(state, workflow_type)
+
+ def _route_after_document(self, state: InvoiceProcessingState) -> Literal["validation", "escalation", "end"]:
+ """Route decision after document extraction"""
+ #if extraction yielded no invoice_data or low confidence -> escalate
+ if not state.invoice_data:
+ return "escalation"
+ #if extraction confidence exists and is low -> escalate
+ conf = getattr(state.invoice_data, "extraction_confidence", None)
+ if conf is not None and conf<0.6:
+ return "escalation"
+ return "validation"
+
+
+ def _route_after_validation(self, state: InvoiceProcessingState) -> Literal["risk", "escalation", "end"]:
+ """Route decision after document validation"""
+ vr = state.validation_result
+ if not vr:
+ return "escalation"
+ #if missing PO or invalid -> escalate
+ try:
+ status = vr.validation_status
+ #ValidationStatus maybe enum or str
+ if isinstance(status,ValidationStatus):
+ status_val = status
+ else:
+ status_val = ValidationStatus(status) if isinstance(status,str) else None
+ if status_val == ValidationStatus.NO_MATCH or status_val == ValidationStatus.PARTIAL_MATCH and (not vr.amount_match):
+ return "escalation"
+ except Exception:
+ #fallback: if discrepancies exist -> escalation
+ if vr and getattr(vr,"discrepancies",None):
+ return "escalation"
+ return "risk"
+
+ def _route_after_risk(self, state: InvoiceProcessingState) -> Literal["payment", "escalation", "human_review", "end"]:
+ """Route decision after risk assessment"""
+ ra = state.risk_assessment
+ if not ra:
+ return "escalation"
+ #ra.risk_level is an enum RiskLevel
+ rl = getattr(ra,"risk_level",None)
+ #handle strings or enums
+ rl_val = rl.value if hasattr(rl,"value") else str(rl).lower()
+ try:
+ if rl_val in (RiskLevel.CRITICAL.value, RiskLevel.HIGH.value):
+ #For critical-> human review; for high->escalate
+ if rl_val == RiskLevel.CRITICAL.value:
+ return "human_review"
+ return "escalation"
+ else:
+ #low or medium -> payment
+ return "payment"
+ except Exception:
+ return "payment"
+
+ def _route_after_payment(self, state: InvoiceProcessingState) -> Literal["audit", "escalation", "end"]:
+ pd = getattr(state,"payment_decision",None)
+ if not pd:
+ return "escalation"
+ #If approved (or scheduled) -> audit
+ try:
+ status = pd.payment_status
+ #Accept enum or str
+ status_val = status if isinstance(status,str) else getattr(status,"value",str(status))
+ if status_val in (PaymentStatus.APPROVED.value, PaymentStatus.SCHEDULED.value, PaymentStatus.PENDING_APPROVAL.value):
+ return "audit"
+ else:
+ return "escalation"
+ except Exception:
+ return "audit"
+
+ def _route_after_audit(self, state: InvoiceProcessingState) -> Literal["escalation", "end"]:
+ cr = getattr(state, "compliance_report",None)
+ if not cr:
+ return "end"
+ #If any compliance issues ->escalate
+ issues = cr.get("issues",{}) if isinstance(cr, dict) else {}
+ has_issues = any(issues.get(k) for k in issues)
+ return "escalation" if has_issues else "end"
+
+ async def _handle_escalation_chain(self, state: "InvoiceProcessingState", workflow_type):
+ """Common handler for escalation โ human review โ complete"""
+ state = await self._escalation_agent_node(state, workflow_type)
+ self._process_store[state.process_id] = state
+ state = await self._human_review_node(state, workflow_type)
+ state.overall_status = ProcessingStatus.COMPLETED
+ self._process_store[state.process_id] = state
+ return state
+
+ async def process_invoice(self, file_name: str, workflow_type: str = "standard",
+ config: Dict[str, Any] = None) -> InvoiceProcessingState:
+ """
+ Orchestrate processing for a single invoice file.
+ Supports 3 workflow types: standard, high_value, and expedited.
+ """
+ process_id = f"proc_{datetime.utcnow().strftime('%Y%m%d_%H%M%S')}_{uuid.uuid4().hex[:6]}"
+ initial_state = InvoiceProcessingState(
+ process_id=process_id,
+ file_name=file_name,
+ overall_status=ProcessingStatus.PENDING,
+ current_agent=None,
+ workflow_type=workflow_type,
+ created_at=datetime.utcnow(),
+ updated_at=datetime.utcnow(),
+ )
+
+ self._process_store[process_id] = initial_state
+ start_ts = datetime.utcnow()
+ state = initial_state
+ worked_agents = []
+ try:
+ # STEP 1๏ธ Document Extraction
+ state = await self._document_agent_node(state, workflow_type)
+ self._process_store[process_id] = state
+ route = self._route_after_document(state)
+ print("state agent anme ::::::::::::::", state.agent_name)
+ worked_agents.append(state.agent_name)
+ if route == "escalation":
+ state = await self._handle_escalation_chain(state, workflow_type)
+ worked_agents.append(state.agent_name)
+ return state, worked_agents
+
+ # ---- Workflow branching ----
+ if workflow_type == "expedited":
+ # Fast lane - skip validation if AI confidence is high
+ # if getattr(state, "extraction_confidence", 0.0) < 0.85:
+ state = await self._validation_agent_node(state, workflow_type)
+ self._process_store[process_id] = state
+ worked_agents.append(state.agent_name)
+ route = self._route_after_validation(state)
+ if route == "escalation":
+ state = await self._handle_escalation_chain(state, workflow_type)
+ worked_agents.append(state.agent_name)
+ return state, worked_agents
+
+ # Directly go to Payment, minimal audit
+ state = await self._payment_agent_node(state, workflow_type)
+ worked_agents.append(state.agent_name)
+ self._process_store[process_id] = state
+ if getattr(state, "payment_decision", {}).decision == "auto-pay":
+ state = await self._audit_agent_node(state, workflow_type)
+ worked_agents.append(state.agent_name)
+ self._process_store[process_id] = state
+
+ elif workflow_type == "high_value":
+ # 2๏ธ Validation (twice for accuracy)
+ state = await self._validation_agent_node(state, workflow_type)
+ self._process_store[process_id] = state
+ state = await self._validation_agent_node(state, workflow_type)
+ worked_agents.append(state.agent_name)
+ route = self._route_after_validation(state)
+ if route == "escalation":
+ state = await self._handle_escalation_chain(state, workflow_type)
+ worked_agents.append(state.agent_name)
+ return state, worked_agents
+
+ # 3๏ธ Risk
+ state = await self._risk_agent_node(state, workflow_type)
+ worked_agents.append(state.agent_name)
+ self._process_store[process_id] = state
+ route = self._route_after_risk(state)
+ if route in ["escalation", "human_review"]:
+ state = await self._handle_escalation_chain(state, workflow_type)
+ worked_agents.append(state.agent_name)
+ return state, worked_agents
+
+ # 4 Audit
+ state = await self._audit_agent_node(state, workflow_type)
+ worked_agents.append(state.agent_name)
+ self._process_store[process_id] = state
+
+ # 5 Mandatory human review for high-value invoices
+ state = await self._human_review_node(state, workflow_type)
+ worked_agents.append(state.agent_name)
+ self._process_store[process_id] = state
+
+ else: # STANDARD workflow
+ # 2๏ธ Validation
+ state = await self._validation_agent_node(state, workflow_type)
+ self._process_store[process_id] = state
+ worked_agents.append(state.agent_name)
+ route = self._route_after_validation(state)
+ if route == "escalation":
+ state = await self._handle_escalation_chain(state, workflow_type)
+ worked_agents.append(state.agent_name)
+ return state, worked_agents
+
+ # 3๏ธ Risk
+ state = await self._risk_agent_node(state, workflow_type)
+ self._process_store[process_id] = state
+ worked_agents.append(state.agent_name)
+ route = self._route_after_risk(state)
+ if route in ["escalation", "human_review"]:
+ state = await self._handle_escalation_chain(state, workflow_type)
+ worked_agents.append(state.agent_name)
+ return state, worked_agents
+
+ # 4๏ธ Payment
+ state = await self._payment_agent_node(state, workflow_type)
+ self._process_store[process_id] = state
+ worked_agents.append(state.agent_name)
+ route = self._route_after_payment(state)
+ if route == "escalation":
+ state = await self._handle_escalation_chain(state, workflow_type)
+ worked_agents.append(state.agent_name)
+ return state, worked_agents
+
+ # 5๏ธ Audit
+ state = await self._audit_agent_node(state, workflow_type)
+ worked_agents.append(state.agent_name)
+ self._process_store[process_id] = state
+
+ # Success completion
+ state.overall_status = ProcessingStatus.COMPLETED
+ state.updated_at = datetime.utcnow()
+ elapsed = (datetime.utcnow() - start_ts).total_seconds()
+ self.logger.logger.info(f"Process {process_id} ({workflow_type}) completed in {elapsed:.2f}s")
+ self._process_store[process_id] = state
+ # print("from graph worked agents::::", worked_agents)
+ return state, worked_agents
+
+ except Exception as e:
+ self.logger.logger.exception(f"Error processing invoice {file_name}: {e}")
+ state.overall_status = ProcessingStatus.FAILED
+ self._process_store[process_id] = state
+ return state, worked_agents
+
+
+ # async def process_batch(self, file_names: List[str], workflow_type: str = "standard",
+ # max_concurrent: int = 5) -> List[InvoiceProcessingState]:
+ # """Process a batch of files with limit concurrency"""
+ # sem = asyncio.Semaphore(max_concurrent)
+ # results: List[InvoiceProcessingState] = []
+
+ # async def _worker(fn: str):
+ # async with sem:
+ # return await self.process_invoice(fn, workflow_type=workflow_type)
+
+ # tasks = [asyncio.create_task(_worker(f)) for f in file_names]
+ # completed = await asyncio.gather(*tasks)
+ # for st in completed:
+ # results.append(st)
+ # return results
+ async def process_batch(self, file_names: List[str], workflow_type: str = "standard",
+ max_concurrent: int = 5):
+ sem = asyncio.Semaphore(max_concurrent)
+ results = [] # will store: {"state": ..., "worked_agents": [...]}
+
+ async def _worker(fn: str):
+ async with sem:
+ return await self.process_invoice(fn, workflow_type=workflow_type)
+
+ tasks = [asyncio.create_task(_worker(f)) for f in file_names]
+ completed = await asyncio.gather(*tasks)
+
+ for result in completed:
+ state, worked_agents = result # unpack the tuple
+ results.append({
+ "state": state,
+ "worked_agents": worked_agents
+ })
+
+ return results
+
+
+ async def get_workflow_status(self, process_id: str) -> Optional[Dict[str, Any]]:
+ """Return the stored workflow status dictionary for a given process_id"""
+ state = self._process_store.get(process_id)
+ if not state:
+ return None
+ return {"process_id":process_id, "status":state.overall_status, "updated_at": getattr(state,"updated_at",None), "state":state.model_dump()}
+
+ async def health_check(self) -> Dict[str, Any]:
+ """Aggregate health check across agents and the orchestrator itself"""
+ agents_health = await agent_registry.health_check_all()
+ return {"orchestrator":"Healthy","timestamp":datetime.utcnow().isoformat(),"agent":agents_health}
+
+ def _extract_final_state(self, result, initial_state: InvoiceProcessingState) -> InvoiceProcessingState:
+ """Compatibility helper (returns invoice processing state)"""
+ return result
+
+
+invoice_workflow: Optional[InvoiceProcessingGraph] = None
+
+def get_workflow(config: Dict[str, Any] = None) -> InvoiceProcessingGraph:
+ global invoice_workflow
+ if invoice_workflow is None:
+ invoice_workflow = InvoiceProcessingGraph(config=config)
+ return invoice_workflow
\ No newline at end of file
diff --git a/Project/main.py b/Project/main.py
new file mode 100644
index 0000000000000000000000000000000000000000..b56ba8f116087343b3f6b107e0c1c5dd94660f5a
--- /dev/null
+++ b/Project/main.py
@@ -0,0 +1,1312 @@
+"""Main Streamlit UI for invoice processing"""
+# TODO: Build Streamlit dashboard
+import os
+import asyncio
+import pandas as pd
+import streamlit as st
+from datetime import datetime
+import plotly.express as px
+import plotly.graph_objects as go
+from typing import Dict, Any, List
+from enum import Enum
+import fitz # PyMuPDF
+import re
+
+from graph import get_workflow
+from state import InvoiceProcessingState, ProcessingStatus, ValidationStatus, RiskLevel, PaymentStatus
+from utils.logger import setup_logging, get_logger
+
+import json
+import google.generativeai as genai
+from agents.smart_explainer_agent import SmartExplainerAgent
+from agents.insights_agent import InsightAgent
+from agents.forecast_agent import ForecastAgent
+
+
+# Logging Setup
+setup_logging()
+logger = get_logger("InvoiceProcessingApp")
+
+def make_arrow_safe(df: pd.DataFrame) -> pd.DataFrame:
+ """
+ Convert any DataFrame to be Streamlit/Arrow compatible:
+ - Converts Enums to string values
+ - Replaces None/NaN with 'Not applicable'
+ - Ensures all columns are strings (avoids Arrow conversion errors)
+ - Capitalizes column headers
+ """
+ if df.empty:
+ return df
+
+ # Convert Enums to strings
+ df = df.applymap(lambda x: x.value if isinstance(x, Enum) else x)
+
+ # Replace None/NaN and make all values string
+ df = df.fillna("Not applicable").astype(str)
+
+ # Capitalize column names nicely
+ df.columns = [col.capitalize() for col in df.columns]
+ return df
+
+import ast
+import re
+def parse_escalation_details(s):
+ if isinstance(s, dict):
+ return s
+ if not isinstance(s, str) or not s.strip():
+ return {}
+
+ # Convert datetime.datetime(YYYY,MM,DD,HH,MM,SS) โ "YYYY-MM-DD HH:MM:SS"
+ def repl(match):
+ parts = match.group(1).split(',')
+ parts = [p.strip() for p in parts]
+ # convert to ISO style
+ return f"'{parts[0]}-{parts[1]}-{parts[2]} {parts[3]}:{parts[4]}:{parts[5]}'"
+
+ s_clean = re.sub(r"datetime\.datetime\((.*?)\)", repl, s)
+
+ try:
+ return ast.literal_eval(s_clean)
+ except:
+ return {}
+
+def serialize_state(state):
+ # Pydantic v2
+ if hasattr(state, "model_dump"):
+ return state.model_dump()
+
+ # Pydantic v1 fallback
+ if hasattr(state, "dict"):
+ return state.dict()
+
+ # Normal python object
+ if hasattr(state, "__dict__"):
+ return state.__dict__
+
+ # Already a dict
+ if isinstance(state, dict):
+ return state
+
+ # string, int, None, etc
+ return {"value": state}
+
+
+class InvoiceProcessingApp:
+ """Main application class for AI Invoice Processing Dashboard"""
+
+ def __init__(self):
+ self.workflow = None
+ self.initialize_session_state()
+ self.initialize_workflow()
+ self.smart_explainer = SmartExplainerAgent()
+ self.insights = InsightAgent()
+ self.forecast = ForecastAgent()
+ self.gemini_api_key = os.getenv("GEMINI_API_KEY_7")
+
+ # INITIALIZATION
+ def initialize_session_state(self):
+ if "selected_files" not in st.session_state:
+ st.session_state.selected_files = []
+ if "results" not in st.session_state:
+ st.session_state.results = []
+ if "last_run" not in st.session_state:
+ st.session_state.last_run = None
+ if "workflow_type" not in st.session_state:
+ st.session_state.workflow_type = "standard"
+ if "max_concurrent" not in st.session_state:
+ st.session_state.max_concurrent = 1
+ if "annotated_pdfs" not in st.session_state:
+ st.session_state.annotated_pdfs = {}
+ # if "priority_level" not in st.session_state:
+ # st.session_state.priority_level = 1
+
+ def initialize_workflow(self):
+ try:
+ self.workflow = get_workflow()
+ logger.info("Workflow initialized successfully.")
+ except Exception as e:
+ logger.exception("Workflow initialization failed: %s", e)
+ st.error("Failed to initialize workflow. Check logs for details.")
+
+ # SIDEBAR + HEADER
+ def render_header(self):
+ st.markdown(
+ """
+
+
๐งพ Invoice AgenticAI - LangGraph
+
AI-Powered Invoice Processing with Intelligent Agent Workflows
+
+ # ๐ Processing {len(selected_files)} invoice(s) via
+ # {workflow_type} workflow...
+ #
+ # """,
+ # unsafe_allow_html=True,
+ # )
+
+ # for i, file in enumerate(selected_files):
+ # st.markdown(f"### ๐ `{os.path.basename(file)}` ({i+1}/{total_files})")
+
+ # try:
+ # # Process one file and get the agent flow
+ # with st.spinner("๐ค Processing Invoice(s) with AI agents..."):
+ # start_time = datetime.utcnow()
+ # state, worked_agents = await self.workflow.process_invoice(file, workflow_type=workflow_type)
+ # duration += (datetime.utcnow() - start_time).total_seconds()
+ # total_stages += len(worked_agents)
+ # results.append(state)
+ # # Loop through dynamic stages
+ # for j, stage in enumerate(worked_agents):
+ # with pipeline_placeholder:
+ # self.show_agent_pipeline(stage, workflow_type, worked_agents)
+
+ # status_placeholder.markdown(
+ # f"๐ง Running {stage}...
",
+ # unsafe_allow_html=True
+ # )
+
+ # # fake progress animation
+ # for pct in range(0, 101, 25):
+ # await asyncio.sleep(0.15)
+
+ # stage_index += 1
+ # progress_bar.progress(int((stage_index/total_stages)*100))
+ # # Step 2๏ธ: Update to "Completed" state
+ # status_placeholder.markdown(
+ # f"""
+ #
+ # โ
All the above Agents have been processed successfully!
+ #
+ # """,
+ # unsafe_allow_html=True
+ # )
+
+ # await asyncio.sleep(0.3) # Small delay to let user see the completion
+
+ # except Exception as e:
+ # logger.exception(f"Error processing {file}: {e}")
+ # st.error(f"โ Failed: {os.path.basename(file)}")
+ # continue
+ # progress_bar.progress(100)
+ # banner_placeholder.empty()
+ # # duration = (datetime.utcnow() - start_time).total_seconds()
+ # st.balloons()
+ # banner_placeholder.markdown(
+ # f"""
+ #
+ # โ
Processed {len(results)} invoice(s) successfully via
+ # {workflow_type} workflow in {duration:.2f}s!
+ #
+ # """,
+ # unsafe_allow_html=True,
+ # )
+
+ # # Final summary
+ # st.success(f"๐ All {len(results)} invoices processed in {duration:.2f} seconds")
+
+ # st.markdown(
+ # f""
+ # f"โ
{workflow_type.capitalize()} Workflow Completed"
+ # f"
", unsafe_allow_html=True
+ # )
+
+ # st.session_state.results = [r.model_dump() if hasattr(r, 'model_dump') else r for r in results]
+ # st.session_state.last_workflow_type = workflow_type
+ async def process_invoices_async(self, selected_files, workflow_type, max_concurrent):
+ if not self.workflow:
+ st.error("Workflow not initialized.")
+ return
+
+ total_files = len(selected_files)
+ progress_bar = st.progress(0)
+ pipeline_placeholder = st.empty()
+ status_placeholder = st.empty()
+ banner_placeholder = st.empty()
+ duration = 0
+ results = []
+
+ # Initial banner
+ banner_placeholder.markdown(
+ f"""
+
+ ๐ Processing {len(selected_files)} invoice(s) via
+ {workflow_type} workflow...
+
+ """,
+ unsafe_allow_html=True,
+ )
+
+ # # ----------------------------------------------------------------------
+ # # ๐ PARALLEL / BATCH PROCESSING MODE (max_concurrent > 1)
+ # # ----------------------------------------------------------------------
+ # if max_concurrent > 1:
+ # st.info(f"โก Running in parallel mode with max_concurrent = {max_concurrent}")
+
+ # with st.spinner("๐ค Processing invoices in parallel..."):
+ # start_time = datetime.utcnow()
+
+ # # Run all invoices concurrently via process_batch()
+ # batch_states = await self.workflow.process_batch(
+ # selected_files,
+ # workflow_type=workflow_type,
+ # max_concurrent=max_concurrent
+ # )
+
+ # duration = (datetime.utcnow() - start_time).total_seconds()
+
+ # # Update progress bar as each file completes
+ # for idx, state in enumerate(batch_states):
+ # progress_bar.progress(int(((idx + 1) / total_files) * 100))
+ # await asyncio.sleep(0.1)
+
+ # results = batch_states
+
+ # st.success(f"๐ Processed {len(results)} invoices in {duration:.2f} seconds (parallel mode)")
+
+ # banner_placeholder.markdown(
+ # f"""
+ #
+ # โ
Processed {len(results)} invoice(s) successfully via
+ # {workflow_type} workflow in {duration:.2f}s!
+ #
+ # """,
+ # unsafe_allow_html=True,
+ # )
+
+ # st.balloons()
+
+ # # Store in session
+ # clean_results = []
+ # for r in results:
+ # state = r["state"]
+ # worked = r["worked_agents"]
+ # clean_results.append({
+ # "state": state.model_dump(),
+ # "worked_agents": worked
+ # })
+ # st.session_state.results = clean_results
+ # st.session_state.last_workflow_type = workflow_type
+ # return
+ # ----------------------------------------------------------------------
+ # ๐ PARALLEL / BATCH PROCESSING MODE (max_concurrent > 1)
+ # ----------------------------------------------------------------------
+ if max_concurrent > 1:
+
+ st.info(f"โก Running in parallel mode with max_concurrent = {max_concurrent}")
+
+ total_files = len(selected_files)
+
+ # ----------------------- UI: QUEUED FILE LIST -----------------------
+ st.markdown("### ๐ Invoice Queue")
+ file_rows = []
+
+ for f in selected_files:
+ row = st.empty()
+ row.markdown(
+ f"""
+
+ ๐ {os.path.basename(f)}
+ โณ Queued...
+
+ """,
+ unsafe_allow_html=True
+ )
+ file_rows.append(row)
+
+ # Progress bar for file-level parallel completion
+ progress_bar = st.progress(0)
+
+ # ----------------------- RUN PARALLEL PROCESSING -----------------------
+ with st.spinner("๐ค Processing invoices in parallel..."):
+ start_time = datetime.utcnow()
+
+ batch_results = await self.workflow.process_batch(
+ selected_files,
+ workflow_type=workflow_type,
+ max_concurrent=max_concurrent
+ )
+ duration = (datetime.utcnow() - start_time).total_seconds()
+
+ # ----------------------- UPDATE UI AS FILES COMPLETE -----------------------
+ st.markdown("### โ
Processing Results")
+
+ clean_results = []
+ completed_count = 0
+ print("batch_results from main", batch_results)
+ for idx, item in enumerate(batch_results):
+ state = item["state"] # FIXED
+ worked_agents = item["worked_agents"] # FIXED
+
+ print("state from main ---", state)
+ print("worked_agents from main ---", worked_agents)
+
+ file_rows[idx].markdown(
+ f"""
+
+ ๐ {os.path.basename(selected_files[idx])}
+ โ
Completed โ {len(worked_agents)} agents executed
+
+ """,
+ unsafe_allow_html=True
+ )
+
+ completed_count += 1
+ progress_bar.progress(int((completed_count / total_files) * 100))
+ await asyncio.sleep(0.15)
+
+ with st.expander(f"๐ Workflow Details โ {os.path.basename(selected_files[idx])}"):
+ st.markdown("### ๐ง Agent Workflow Replay")
+ for agent in worked_agents:
+ self.show_agent_pipeline(agent, workflow_type, worked_agents)
+ await asyncio.sleep(0.25)
+
+ # clean_results.append({
+ # "state": serialize_state(state),
+ # "worked_agents": worked_agents
+ # })
+ clean_results.append(state)
+ print("cleaned res from main---", clean_results)
+
+ # ----------------------- FINAL BANNER -----------------------
+ st.balloons()
+ st.success(
+ f"๐ Processed {completed_count} invoices in {duration:.2f} seconds (parallel mode)"
+ )
+
+ banner_placeholder.markdown(
+ f"""
+
+ โ
Processed {completed_count} invoice(s) successfully via
+ {workflow_type}
+ workflow in {duration:.2f}s!
+
+ """,
+ unsafe_allow_html=True,
+ )
+
+ # ----------------------- SAVE SESSION -----------------------
+ st.session_state.results = [
+ r.model_dump() if hasattr(r, "model_dump") else r
+ for r in clean_results
+ ]
+ # st.session_state.results = clean_results
+ st.session_state.last_workflow_type = workflow_type
+ return
+
+
+ # SEQUENTIAL MODE (max_concurrent == 1)
+ stage_index = 0
+ total_stages = 0
+
+ for i, file in enumerate(selected_files):
+ st.markdown(f"### ๐ `{os.path.basename(file)}` ({i+1}/{total_files})")
+
+ try:
+ with st.spinner("๐ค Processing Invoice(s) with AI agents..."):
+ start_time = datetime.utcnow()
+
+ # sequential per-file detailed processing
+ state, worked_agents = await self.workflow.process_invoice(
+ file,
+ workflow_type=workflow_type
+ )
+
+ duration += (datetime.utcnow() - start_time).total_seconds()
+ total_stages += len(worked_agents)
+
+ results.append(state)
+
+ # UI pipeline per agent
+ for j, stage in enumerate(worked_agents):
+
+ with pipeline_placeholder:
+ self.show_agent_pipeline(stage, workflow_type, worked_agents)
+
+ status_placeholder.markdown(
+ f""
+ f"๐ง Running {stage}..."
+ f"
",
+ unsafe_allow_html=True
+ )
+
+ # Fake animation to show progress nicely
+ for pct in range(0, 101, 25):
+ await asyncio.sleep(0.15)
+
+ stage_index += 1
+ progress_bar.progress(int((stage_index / total_stages) * 100))
+
+ # Mark completed
+ status_placeholder.markdown(
+ f"""
+
+ โ
All the above Agents have been processed successfully!
+
+ """,
+ unsafe_allow_html=True
+ )
+
+ await asyncio.sleep(0.3)
+
+ except Exception as e:
+ logger.exception(f"Error processing {file}: {e}")
+ st.error(f"โ Failed: {os.path.basename(file)}")
+ continue
+
+ # Final UI completion
+ progress_bar.progress(100)
+ banner_placeholder.empty()
+ st.balloons()
+
+ banner_placeholder.markdown(
+ f"""
+
+ โ
Processed {len(results)} invoice(s) successfully via
+ {workflow_type} workflow in {duration:.2f}s!
+
+ """,
+ unsafe_allow_html=True,
+ )
+
+ st.success(f"๐ All {len(results)} invoices processed in {duration:.2f} seconds")
+
+ st.markdown(
+ f""
+ f"โ
{workflow_type.capitalize()} Workflow Completed"
+ f"
",
+ unsafe_allow_html=True,
+ )
+
+ # Store results
+ st.session_state.results = [
+ r.model_dump() if hasattr(r, "model_dump") else r
+ for r in results
+ ]
+ st.session_state.last_workflow_type = workflow_type
+
+
+ def show_agent_pipeline(self, current_stage: str, workflow_type: str, stages: list[str]):
+ """Display dynamic workflow pipeline based on selected workflow."""
+ colors = {
+ "document_agent": "#00bcd4",
+ "validation_agent": "#fbc02d",
+ "risk_agent": "#e64a19",
+ "payment_agent": "#43a047",
+ "audit_agent": "#1976d2",
+ "escalation_agent": "#ff9800",
+ # "Escalation (if needed)": "#f44336",
+ "human_review_agent": "#8e24aa",
+ # "human_review (if needed)": "#ba68c8",
+ }
+
+ html = ""
+ for stage in stages:
+ glow = "box-shadow:0 0 12px rgba(0,255,0,0.7);" if stage == current_stage else ""
+ html += f"""
+ {stage}
+
"""
+ html += "
+ ๐ Workflow Summary: The AI system routes invoices automatically between extraction, validation, risk, and payment agents.
+
+ """, unsafe_allow_html=True)
+
+ st.markdown("### ๐ค AI Agent Workflow")
+ st.markdown("""
+ 1. ๐งพ **Document Agent** โ Extract invoice data using AI
+ 2. โ
**Validation Agent** โ Validate against purchase orders
+ 3. โ ๏ธ **Risk Agent** โ Assess fraud risk and compliance
+ 4. ๐ณ **Payment Agent** โ Make payment decisions
+ 5. ๐งฎ **Audit Agent** โ Generate compliance records
+ 6. ๐จ **Escalation Agent** โ Handle exceptions
+
+ > ๐ง The workflow uses intelligent routing based on validations, risk scores, and business rules.
+ """)
+
+ self.show_processing_summary(st.session_state.results)
+
+ # INVOICE DETAILS
+ def render_invoice_details_tab(self):
+ st.markdown("""
+
+ ๐ Workflow Summary: The AI system routes invoices automatically between extraction, validation, risk, and payment agents.
+
+ """, unsafe_allow_html=True)
+ st.subheader("Detailed Invoice View")
+ results = st.session_state.results
+ if not results:
+ st.info("No processed invoices yet.")
+ return
+
+ selected_file = st.selectbox("Select invoice for details:", [os.path.basename(r["file_name"]) for r in results])
+ selected = next((r for r in results if os.path.basename(r["file_name"]) == selected_file), None)
+ print("selected....", selected)
+ if selected:
+ self.show_detailed_invoice_view(selected)
+ # --- Add Button for Highlighting Discrepancies ---
+ st.markdown("### ๐ Check PDF for Issues")
+ if st.button("Highlight Discrepancies in Invoice PDF"):
+ invoice_path = next(
+ (f for f in self.get_available_files()
+ if os.path.basename(f) == os.path.basename(selected["file_name"])), None
+ )
+ if invoice_path:
+ with st.spinner("Analyzing invoice for discrepancies..."):
+ discrepancies, output_pdf = self.highlight_invoice_discrepancies(invoice_path)
+ st.session_state.annotated_pdfs[selected_file] = output_pdf
+ if discrepancies:
+ st.error("โ ๏ธ Mismatches found:")
+ df_disc = pd.DataFrame(discrepancies)
+ st.dataframe(df_disc)
+ with open(output_pdf, "rb") as f:
+ st.download_button(
+ label=f"๐ฅ Download Annotated Invoice PDF ({selected_file})",
+ data=f,
+ file_name=os.path.basename(output_pdf),
+ mime="application/pdf"
+ )
+ else:
+ st.success("โ
No discrepancies found. Invoice matches CSV perfectly!")
+
+ # --- Show Dropdown of All Annotated PDFs ---
+ if st.session_state.annotated_pdfs:
+ st.markdown("### ๐ Annotated Invoices Library")
+
+ selected_marked_pdf = st.selectbox(
+ "Select an annotated (highlighted) invoice:",
+ options=list(st.session_state.annotated_pdfs.keys()),
+ key="annotated_pdf_selector"
+ )
+
+ pdf_path = st.session_state.annotated_pdfs[selected_marked_pdf]
+
+ st.info(f"Selected: {selected_marked_pdf}")
+
+ # Download button
+ with open(pdf_path, "rb") as f:
+ st.download_button(
+ label=f"๐ฅ Download {selected_marked_pdf}",
+ data=f,
+ file_name=os.path.basename(pdf_path),
+ mime="application/pdf"
+ )
+
+ def show_detailed_invoice_view(self, result):
+ st.markdown(f"### ๐งพ Invoice Summary")
+ st.markdown(f"**File:** `{os.path.basename(result.get('file_name', 'N/A'))}`")
+ st.markdown(f"**Status:** :{'green_circle:' if result.get('overall_status') == 'completed' else 'orange_circle:'} {result.get('overall_status', 'Unknown')}")
+
+ # --- Invoice Data Section ---
+ st.markdown("### ๐ Invoice Data")
+ invoice_data = result.get("invoice_data", {})
+
+ # Handle datetime formatting
+ for k, v in invoice_data.items():
+ if isinstance(v, datetime):
+ invoice_data[k] = v.strftime("%Y-%m-%d %H:%M:%S")
+ elif isinstance(v, str) and v.startswith("datetime.datetime"):
+ invoice_data[k] = v.replace("datetime.datetime", "").strip("()")
+
+ df_invoice = make_arrow_safe(pd.DataFrame(invoice_data.items(), columns=["Field", "Value"]))
+ st.dataframe(df_invoice, width='stretch')
+
+ # --- Line Items ---
+ items = invoice_data.get("item_details") or invoice_data.get("line_items")
+ if items:
+ st.markdown("#### ๐ฆ Line Items")
+ df_items = make_arrow_safe(pd.DataFrame(items))
+ st.dataframe(df_items, width='stretch')
+
+ # --- Validation Result ---
+ st.markdown("### โ
Validation Result")
+ validation = result.get("validation_result", {})
+ if validation:
+ po_data = validation.pop("po_data", {})
+ df_val = make_arrow_safe(pd.DataFrame(validation.items(), columns=["Check", "Result"]))
+ st.dataframe(df_val, width='stretch')
+
+ if po_data:
+ st.markdown("#### ๐ฆ PO Data")
+ df_po = make_arrow_safe(pd.DataFrame(po_data.items(), columns=["Field", "Value"]))
+ st.dataframe(df_po, width='stretch')
+
+ # --- Risk Assessment ---
+ st.markdown("### โ ๏ธ Risk Assessment")
+ risk = result.get("risk_assessment", {})
+ if risk:
+ df_risk = make_arrow_safe(pd.DataFrame(risk.items(), columns=["Factor", "Value"]))
+ st.dataframe(df_risk, width='stretch')
+
+ # --- Payment Decision ---
+ st.markdown("### ๐ณ Payment Decision")
+ payment = result.get("payment_decision", {})
+ if payment:
+ for k, v in payment.items():
+ if isinstance(v, datetime):
+ payment[k] = v.strftime("%Y-%m-%d %H:%M:%S")
+ elif isinstance(v, str) and v.startswith("datetime.datetime"):
+ payment[k] = v.replace("datetime.datetime", "").strip("()")
+
+ df_payment = make_arrow_safe(pd.DataFrame(payment.items(), columns=["Attribute", "Value"]))
+ st.dataframe(df_payment, width='stretch')
+ else:
+ st.error("There is something mismatch in PO_data and Invoice Details. Please look into the previous data.")
+
+ # --- Audit Trail ---
+ st.markdown("### ๐งฎ Audit Trail")
+ audit = result.get("audit_trail", []) or []
+ if audit:
+ df_audit = make_arrow_safe(pd.DataFrame(audit).head(10))
+ st.dataframe(df_audit, width='stretch')
+ else:
+ st.info("No audit trail entries found.")
+
+ def highlight_invoice_discrepancies(self, invoice_path: str):
+ """Compare invoice PDF with CSV and highlight mismatches visually."""
+ DATA_DIR = os.path.join(os.getcwd(), "data")
+ CSV_PATH = os.path.join(DATA_DIR, "purchase_orders.csv")
+ OUTPUT_PATH = os.path.join(DATA_DIR, "annotated_invoice.pdf")
+
+ FIELD_BOXES = {
+ "invoice_number": (525, 55, 575, 75),
+ "order_id": (45, 470, 230, 490),
+ "customer_name": (40, 135, 100, 155),
+ "quantity": (370, 235, 385, 250),
+ "rate": (450, 235, 500, 250),
+ "expected_amount": (520, 315, 570, 330),
+ }
+
+ pdf = fitz.open(invoice_path)
+ page = pdf[0]
+ pdf_text = page.get_text()
+
+ # === extract fields ===
+ def extract_field(pattern, text, group=1):
+ match = re.search(pattern, text, re.IGNORECASE)
+ return match.group(group).strip() if match else None
+
+ invoice_number_pdf = extract_field(r"#\s*(\d+)", pdf_text)
+ order_id_pdf = extract_field(r"Order ID\s*[:\-]?\s*(\S+)", pdf_text)
+ customer_name_pdf = extract_field(r"Bill To:\s*(.*)", pdf_text)
+
+ po_df = pd.read_csv(CSV_PATH)
+ matched_row = po_df[
+ (po_df['invoice_number'].astype(str) == str(invoice_number_pdf))
+ | (po_df['order_id'] == order_id_pdf)
+ ]
+ if matched_row.empty:
+ st.warning(f"No matching CSV row found for Invoice {invoice_number_pdf} / Order {order_id_pdf}")
+ return None, None
+
+ expected = matched_row.iloc[0].to_dict()
+ expected = {k.lower(): str(v).strip() for k, v in expected.items()}
+
+ invoice_data = {
+ "invoice_number": invoice_number_pdf,
+ "customer_name": customer_name_pdf,
+ "order_id": order_id_pdf,
+ }
+
+ amounts = re.findall(r"\$?([\d,]+\.\d{2})", pdf_text)
+ invoice_data["expected_amount"] = amounts[-3] if amounts else None
+
+ item_lines = re.findall(
+ r"([A-Za-z0-9 ,\-]+)\s+(\d+)\s+\$?([\d,]+\.\d{2})\s+\$?([\d,]+\.\d{2})",
+ pdf_text,
+ )
+ if item_lines:
+ invoice_data["quantity"] = item_lines[0][1]
+ invoice_data["rate"] = item_lines[0][2]
+
+ discrepancies = []
+
+ def add_discrepancy(field, expected_val, found_val):
+ discrepancies.append({"field": field, "expected": expected_val, "found": found_val})
+
+ for field in ["invoice_number", "order_id", "customer_name"]:
+ if str(invoice_data.get(field, "")).strip() != str(expected.get(field, "")).strip():
+ add_discrepancy(field, expected.get(field, ""), invoice_data.get(field, ""))
+
+ for field in ["quantity", "rate", "expected_amount"]:
+ try:
+ found_val = float(str(invoice_data.get(field, 0)).replace(",", "").replace("$", ""))
+ expected_val = float(str(expected.get(field, 0)).replace(",", "").replace("$", ""))
+ if round(found_val, 2) != round(expected_val, 2):
+ add_discrepancy(field, expected_val, found_val)
+ except:
+ if str(invoice_data.get(field, "")) != str(expected.get(field, "")):
+ add_discrepancy(field, expected.get(field, ""), invoice_data.get(field, ""))
+
+ for d in discrepancies:
+ field = d["field"]
+ if field not in FIELD_BOXES:
+ continue
+ rect = fitz.Rect(FIELD_BOXES[field])
+ expected_text = f"{d['expected']}"
+ page.draw_rect(rect, color=(1, 0, 0), width=1.5)
+ page.insert_text((rect.x0, rect.y1 + 10), expected_text, fontsize=9, color=(1, 0, 0))
+
+ pdf.save(OUTPUT_PATH)
+ pdf.close()
+
+ return discrepancies, OUTPUT_PATH
+
+ # AGENT PERFORMANCE TAB
+ def render_agent_performance_tab(self):
+ st.subheader("Agent Performance Metrics")
+ try:
+ metrics = asyncio.run(self.workflow.health_check())["agent"]
+ except Exception as e:
+ print("error in metrics from main.py............", e)
+ st.warning("No live metrics found.")
+ return
+
+ print("metrics from main", metrics)
+ rows = []
+ for agent, health_check_history in metrics.items():
+ rows.append({
+ "Agent": health_check_history.get("Agent"),
+ "Executions": health_check_history.get("Executions", 0),
+ "Success Rate (%)": round(health_check_history.get("Success Rate (%)", 0), 2),
+ "Avg Duration (ms)": health_check_history.get("Avg Duration (ms)", 0),
+ "Total Failures": health_check_history.get("Total Failures", 0),
+ })
+ df = pd.DataFrame(rows)
+ st.dataframe(df, width='stretch')
+
+ # Visualization
+ fig = go.Figure()
+ fig.add_trace(go.Bar(x=df["Agent"], y=df["Success Rate (%)"], name="Success Rate"))
+ fig.add_trace(go.Bar(x=df["Agent"], y=df["Avg Duration (ms)"], name="Duration (ms)", yaxis="y2"))
+ fig.update_layout(
+ title="Agent Success vs Duration",
+ yaxis_title="Success Rate (%)",
+ yaxis2=dict(title="Duration (ms)", overlaying="y", side="right"),
+ barmode="group",
+ )
+ st.plotly_chart(fig, width='stretch')
+
+ # ESCALATIONS TAB
+ def render_escalations_tab(self):
+ st.subheader("Escalations Overview")
+ results = st.session_state.results
+ # rows = [
+ # {
+ # "File": os.path.basename(r.get("file_name", "")),
+ # "Reason": r.get("escalation_reason", "N/A"),
+ # "Assigned To": (r.get("escalation_details") or {}).get("assigned_to", "N/A"),
+ # "SLA Deadline": (r.get("escalation_details") or {}).get("sla_deadline", "N/A"),
+ # }
+ # for r in results if r.get("escalation_required")
+ # ]
+ # if not rows:
+ # st.info("No escalations detected.")
+ # return
+ # st.dataframe(pd.DataFrame(rows), width='stretch')
+ rows = []
+ for r in results:
+ if not r.get("escalation_required"):
+ continue
+
+ esc_dict = parse_escalation_details(r.get("escalation_details"))
+
+ row = {
+ "File": os.path.basename(r.get("file_name", "")),
+ "Reason": r.get("escalation_reason", "N/A"),
+ "Assigned To": esc_dict.get("assigned_to", "N/A"),
+ "SLA Deadline": esc_dict.get("sla_deadline", "N/A"),
+ }
+
+ rows.append(row)
+ if not rows:
+ st.info("No escalations detected")
+ else:
+ st.dataframe(pd.DataFrame(rows))
+
+ # ANALYTICS TAB
+ def render_analytics_tab(self):
+ st.subheader("Processing Analytics")
+ results = st.session_state.results
+ if not results:
+ st.info("No analytics available.")
+ return
+
+ df = pd.DataFrame([
+ {
+ "File": os.path.basename(r["file_name"]),
+ "Amount": (r.get("invoice_data") or {}).get("total", 0),
+ "Risk Score": (r.get("risk_assessment") or {}).get("risk_score", 0),
+ "Status": r.get("overall_status", "unknown"),
+ }
+ for r in results
+ ])
+
+ if df.empty:
+ st.info("No numeric analytics.")
+ return
+
+ col1, col2 = st.columns(2)
+ fig1 = px.bar(df, x="File", y="Amount", color="Status", title="Total Amount by Invoice")
+ fig2 = px.scatter(df, x="Amount", y="Risk Score", color="Status", title="Risk vs Amount")
+ col1.plotly_chart(fig1, width='stretch')
+ col2.plotly_chart(fig2, width='stretch')
+
+
+ def render_smart_insights_tab(self):
+ st.subheader("Smart Insights Assistant")
+
+ results = st.session_state.get("results", [])
+ if not results:
+ st.info("No processed invoices available for insights.")
+ return
+
+ # ------------------ SMART EXPLAINER ------------------
+ st.markdown("### Smart Explainer")
+
+ try:
+ genai.configure(api_key=self.gemini_api_key)
+ except Exception as e:
+ st.warning(f"โ ๏ธ Gemini API not configured properly: {e}")
+
+ selected_file = st.selectbox(
+ "Select an invoice to explain:",
+ [os.path.basename(r.get("file_name", "Unknown")) for r in results],
+ )
+
+ selected = next(
+ (r for r in results if os.path.basename(r.get("file_name", "")) == selected_file),
+ None,
+ )
+
+ if selected:
+ # โ
Sanitize before parsing
+ if "human_review_required" in selected and not isinstance(selected["human_review_required"], bool):
+ selected["human_review_required"] = False
+ if "escalation_details" in selected and not isinstance(selected["escalation_details"], (str, type(None))):
+ selected["escalation_details"] = str(selected["escalation_details"])
+
+ try:
+ parsed_state = InvoiceProcessingState(**selected)
+ except Exception:
+ parsed_state = None
+
+ # ------------------ Header Summary (Enhanced UI) ------------------
+ st.markdown("### Invoice Summary")
+
+ risk = ((selected.get("risk_assessment") or {}).get("risk_level", "Unknown")).capitalize()
+ validation = ((selected.get("validation_result") or {}).get("validation_status", "Unknown")).capitalize()
+ payment = (
+ (selected.get("payment_decision") or {}).get("status", "Pending")
+ if isinstance(selected.get("payment_decision"), dict)
+ else "Pending"
+ )
+
+ # ๐น Professional color palette
+ risk_colors = {
+ "Critical": "#F44336", # Bright red
+ "Medium": "#FF9800", # Deep amber
+ "Low": "#4CAF50", # Balanced green
+ "Unknown": "#9E9E9E", # Neutral gray
+ }
+ validation_colors = {
+ "Passed": "#4CAF50",
+ "Failed": "#F44336",
+ "Missing_po": "#FFC107",
+ "Unknown": "#9E9E9E",
+ }
+ payment_colors = {
+ "Paid": "#4CAF50",
+ "Pending": "#FFC107",
+ "Overdue": "#E91E63",
+ "Unknown": "#9E9E9E",
+ }
+
+ def badge_html(label, value, color):
+ return f"""
+
+ {label}: {value}
+
+ """
+
+ col1, col2, col3 = st.columns(3)
+ with col1:
+ st.markdown(badge_html("Risk", risk, risk_colors.get(risk, "#9E9E9E")), unsafe_allow_html=True)
+ with col2:
+ st.markdown(badge_html("Validation", validation, validation_colors.get(validation, "#9E9E9E")), unsafe_allow_html=True)
+ with col3:
+ st.markdown(badge_html("Payment", payment, payment_colors.get(payment, "#9E9E9E")), unsafe_allow_html=True)
+
+ # ------------------ Gemini Explanation ------------------
+
+ if parsed_state:
+ try:
+ # Safe fallback handling for missing invoice_data fields
+ inv = getattr(parsed_state, "invoice_data", None)
+ vendor = getattr(inv, "customer_name", None) or getattr(inv, "vendor_name", "Unknown Vendor")
+ amount = getattr(inv, "total", "Unknown Amount")
+ due_date = getattr(inv, "due_date", "Unknown Due Date")
+
+ prompt = (
+ f"Provide a clear, professional summary of the following invoice:\n\n"
+ f"Vendor: {vendor}\n"
+ f"Amount: {amount}\n"
+ f"Due Date: {due_date}\n"
+ f"Risk Level: {risk}\n"
+ f"Validation: {validation}\n"
+ f"Payment Status: {payment}\n\n"
+ f"Generate a concise, professional explanation with sections:\n"
+ f"1. Overview\n2. Risk and Validation Summary\n3. Recommended Actions"
+ )
+
+ model = genai.GenerativeModel("gemini-2.0-flash")
+ response = model.generate_content(prompt)
+ st.markdown(response.text)
+
+ except Exception as e:
+ st.warning(f"Gemini explanation failed: {e}")
+ st.json(selected)
+ else:
+ st.warning("Could not parse invoice or Gemini unavailable.")
+ st.json(selected)
+
+ st.markdown("---")
+
+
+ # ------------------ CONSOLIDATED SPEND INSIGHTS ------------------
+ st.markdown("### Consolidated Spend Insights")
+
+ try:
+ df = pd.DataFrame([
+ {
+ "File": os.path.basename(r.get("file_name", "Unknown")),
+ "Vendor": (r.get("invoice_data") or {}).get("customer_name", "Unknown"),
+ "Amount": float((r.get("invoice_data") or {}).get("total", 0)),
+ "Risk": ((r.get("risk_assessment") or {}).get("risk_level", "Unknown")).capitalize(),
+ "Validation": ((r.get("validation_result") or {}).get("validation_status", "Unknown")).capitalize(),
+ }
+ for r in results
+ ])
+
+ if df.empty:
+ st.info("No invoice data available for visualization.")
+ else:
+ total_spend = df["Amount"].sum()
+ st.metric("Total Spend (USD)", f"${total_spend:,.2f}")
+ st.metric("Invoices Processed", len(df))
+
+ col1, col2 = st.columns(2)
+ with col1:
+ fig1 = px.pie(
+ df,
+ names="Risk",
+ values="Amount",
+ title="Spend by Risk Level",
+ color="Risk",
+ color_discrete_map={
+ "Critical": "#F44336",
+ "Medium": "#FF9800",
+ "Low": "#4CAF50",
+ "Unknown": "#9E9E9E",
+ },
+ )
+ st.plotly_chart(fig1, use_container_width=True)
+
+ with col2:
+ fig2 = px.bar(
+ df,
+ x="Vendor",
+ y="Amount",
+ color="Validation",
+ title="Spend by Vendor",
+ color_discrete_map={
+ "Passed": "#4CAF50",
+ "Failed": "#F44336",
+ "Missing_po": "#FFC107",
+ "Unknown": "#9E9E9E",
+ },
+ )
+ st.plotly_chart(fig2, use_container_width=True)
+ except Exception as e:
+ st.error(f"Error generating spend insights: {e}")
+
+ st.markdown("---")
+
+ # ------------------ FORECAST & ANOMALY SECTION ------------------
+ st.markdown("### ๐ Forecast & Anomaly Insights")
+
+ try:
+ clean_states = []
+ for r in results:
+ if isinstance(r, dict):
+ try:
+ clean_states.append(InvoiceProcessingState(**r))
+ except Exception:
+ r_fixed = dict(r)
+ if isinstance(r_fixed.get("human_review_required"), str):
+ val = r_fixed["human_review_required"].strip().lower()
+ r_fixed["human_review_required"] = val in ("true", "yes", "1", "required")
+ if isinstance(r_fixed.get("escalation_details"), dict):
+ r_fixed["escalation_details"] = str(r_fixed["escalation_details"])
+ try:
+ clean_states.append(InvoiceProcessingState(**r_fixed))
+ except Exception:
+ continue
+
+ forecast_data = self.forecast.predict_cashflow(clean_states, months=6)
+ if forecast_data.get("chart"):
+ st.plotly_chart(forecast_data["chart"], use_container_width=True)
+ st.metric("Average Monthly Spend (USD)", f"${forecast_data['average_monthly_spend']:,.2f}")
+ st.metric("Total Forecast (next 6 months)", f"${forecast_data['total_forecast']:,.2f}")
+
+
+ if not forecast_data or not forecast_data.get("chart"):
+ st.info("No sufficient data for forecast.")
+ else:
+ st.plotly_chart(forecast_data["chart"], use_container_width=True)
+ st.success(
+ f"**Average Monthly Spend:** ${forecast_data['average_monthly_spend']:,} \n"
+ f"**Total Forecast (Next {len(forecast_data['forecast_values'])} months):** ${forecast_data['total_forecast']:,}"
+ )
+
+ anomalies = self.forecast.detect_anomalies(clean_states)
+ if anomalies is not None and not anomalies.empty:
+ st.markdown("### โ ๏ธ Anomalies Detected")
+ st.dataframe(
+ anomalies[["invoice_date", "vendor", "total", "risk_score", "anomaly_reason"]],
+ use_container_width=True,
+ )
+ st.warning(f"{len(anomalies)} anomalies found (high spend or high risk).")
+ else:
+ st.info("No anomalies detected. Spend trends look stable โ
")
+
+ except Exception as e:
+ st.error(f"Forecast section failed: {e}")
+
+ def show_workflow_diagram(self):
+ # pass
+ st.subheader("Workflow Diagram (Conceptual)")
+ st.image(os.path.join("assets", "workflow_diagram.png")) if os.path.exists(os.path.join("assets", "workflow_diagram.png")) else st.text("Diagram not provided.")
+
+
+ # HEALTH CHECK TAB
+ def show_health_check(self):
+ st.subheader("System Health Check")
+ try:
+ health = asyncio.run(self.workflow.health_check())
+ # st.json(health)
+ agents_data = health.get("agent", {})
+ if not agents_data:
+ st.warning("No Agents Data Found")
+ return
+ df_health = pd.DataFrame(agents_data).T
+ df_health = make_arrow_safe(df_health)
+ orchestrator_status = health.get("orchestrator", "unknown")
+ st.markdown(f"**Orchestrator Status:** `{orchestrator_status}`")
+ st.dataframe(df_health, width='stretch')
+ except Exception as e:
+ st.error(f"Health check failed: {e}")
+
+ # RUN APP
+ def run(self):
+ self.render_header()
+ if self.workflow:
+ st.success("โ
All agents and workflow initialized successfully!")
+ else:
+ st.error("โ ๏ธ Workflow not initialized. Please check logs.")
+
+ # if "last_pipeline_stage" in st.session_state and "last_workflow_type" in st.session_state:
+ # stages = self._get_stages_for_workflow(st.session_state["last_workflow_type"])
+ # self.show_agent_pipeline(st.session_state["last_pipeline_stage"], st.session_state["last_workflow_type"], stages)
+
+ st.info("๐ Select invoice files from the sidebar and click **Process Invoices** to get started.")
+
+ self.render_sidebar()
+ self.render_main_dashboard()
+
+ # if st.session_state.last_run:
+ # st.sidebar.markdown("---")
+ # st.sidebar.info(f"Last Run: {st.session_state.last_run}")
+
+
+if __name__ == "__main__":
+ app = InvoiceProcessingApp()
+ app.run()
\ No newline at end of file
diff --git a/Project/payment_api.py b/Project/payment_api.py
new file mode 100644
index 0000000000000000000000000000000000000000..df777c8b18c9579baea83c4d6f864e89d3e71dc5
--- /dev/null
+++ b/Project/payment_api.py
@@ -0,0 +1,154 @@
+"""Payment API service"""
+# TODO: Implement payment endpoints
+
+from fastapi import FastAPI, Request, HTTPException
+from pydantic import BaseModel
+from datetime import datetime
+import random
+import uvicorn
+import logging
+from typing import Dict, Any
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+app = FastAPI(
+ title="Invoice Payment API",
+ description="Payment processing simulation for invoice automation",
+ version="1.0.0"
+)
+
+TRANSACTIONS: Dict[str, Dict[str, Any]] = {}
+PAYMENT_METRICS = {
+ 'total_transactions': 0,
+ 'successful_payments': 0,
+ 'failed_payments': 0,
+ 'cancelled_payments': 0,
+ 'last_transaction_time': None
+}
+
+
+class PaymentRequest(BaseModel):
+ order_id: str
+ customer_name: str
+ amount: float
+ currency: str = 'USD'
+ method: str = 'bank_transfer'
+ recipient_account: str = 'default_account'
+ due_date: str
+
+
+class PaymentResponse(BaseModel):
+ transaction_id: str
+ status: str
+ message: str
+ processed_at: str
+
+
+@app.get("/")
+async def root():
+ return {'message': 'Welcome to the Invoice Payment API', 'version': '1.0.0'}
+
+
+@app.get("/health")
+async def health_check():
+ return {
+ 'status': 'healthy',
+ 'timestamp': datetime.utcnow().isoformat(),
+ 'uptime': 'OK'
+ }
+
+
+@app.post("/initiate_payment")
+async def initiate_payment(payment_request: PaymentRequest):
+ transaction_id = f"txn_{int(datetime.utcnow().timestamp())}_{random.randint(1000,9999)}"
+ PAYMENT_METRICS['total_transactions'] += 1
+ PAYMENT_METRICS['last_transaction_time'] = datetime.utcnow().isoformat()
+
+ success_chance = 0.85
+ if random.random() < success_chance:
+ status = 'SUCCESS'
+ message = f"Payment of {payment_request.amount} {payment_request.currency} successful."
+ PAYMENT_METRICS['successful_payments'] += 1
+ else:
+ status = 'FAILED'
+ message = 'Payment failed due to simulated gateway error.'
+ PAYMENT_METRICS['failed_payments'] += 1
+
+ transaction_data = {
+ 'transaction_id': transaction_id,
+ 'order_id': payment_request.order_id,
+ 'customer_name': payment_request.customer_name,
+ 'amount': payment_request.amount,
+ 'currency': payment_request.currency,
+ 'method': payment_request.method,
+ 'recipient_account': payment_request.recipient_account,
+ 'status': status,
+ 'message': message,
+ 'created_at': datetime.utcnow().isoformat(),
+ 'updated_at': datetime.utcnow().isoformat()
+ }
+
+ TRANSACTIONS[transaction_id] = transaction_data
+ logger.info(f"Processed payment transaction {transaction_id} with status: {status}")
+
+ return PaymentResponse(
+ transaction_id=transaction_id,
+ status=status,
+ message=message,
+ processed_at=transaction_data['updated_at']
+ )
+
+
+@app.get("/transaction/{transaction_id}")
+async def get_transaction_status(transaction_id: str):
+ transaction = TRANSACTIONS.get(transaction_id)
+ if not transaction:
+ raise HTTPException(status_code=404, detail="Transaction not found")
+ return transaction
+
+
+@app.post("/cancel_payment/{transaction_id}")
+async def cancel_payment(transaction_id: str):
+ transaction = TRANSACTIONS.get(transaction_id)
+ if not transaction:
+ raise HTTPException(status_code=400, detail="Cannot cancel a non-existent payment")
+
+ if transaction['status'] in ['SUCCESS', 'FAILED']:
+ raise HTTPException(status_code=400, detail="Cannot cancel a completed payment")
+
+ transaction['status'] = 'CANCELLED'
+ transaction['updated_at'] = datetime.utcnow().isoformat()
+ PAYMENT_METRICS['cancelled_payments'] += 1
+ logger.info(f"Cancelled payment transaction: {transaction_id}")
+ return {
+ 'transaction_id': transaction_id,
+ 'status': 'CANCELLED',
+ 'message': 'Payment has been cancelled successfully'
+ }
+
+
+@app.get("/payment_methods")
+async def get_payment_methods():
+ return {
+ 'available_methods': [
+ 'bank_transfer',
+ 'credit_card',
+ 'paypal',
+ 'upi',
+ 'crypto'
+ ],
+ 'default_method': 'bank_transfer'
+ }
+
+
+@app.get("/metrics")
+async def get_metrics():
+ return {
+ 'metrics': PAYMENT_METRICS,
+ 'active_transactions': len(TRANSACTIONS)
+ }
+
+
+if __name__ == "__main__":
+ uvicorn.run(app, host="0.0.0.0", port=8001)
\ No newline at end of file
diff --git a/Project/requirements.txt b/Project/requirements.txt
new file mode 100644
index 0000000000000000000000000000000000000000..6029905ea516b387412ad11ce24cb674d66b42a6
--- /dev/null
+++ b/Project/requirements.txt
@@ -0,0 +1,36 @@
+# Core data processing
+pandas
+numpy
+
+# PDF processing
+pymupdf==1.23.22
+pdfplumber
+
+# Web framework and API
+streamlit
+fastapi
+uvicorn
+
+# AI/ML
+google-generativeai
+
+# LangGraph and LangChain dependencies
+langgraph==0.2.50
+langchain==0.3.7
+langchain-core==0.3.15
+
+# Data validation and typing
+pydantic==2.9.2
+typing-extensions==4.12.2
+
+# Fuzzy string matching for validation
+fuzzywuzzy==0.18.0
+python-Levenshtein==0.25.0
+
+# Utilities
+python-dotenv==1.0.1
+requests==2.32.3
+plotly
+
+# Testing
+pytest
\ No newline at end of file
diff --git a/Project/state.py b/Project/state.py
new file mode 100644
index 0000000000000000000000000000000000000000..8b1a48b1b17e113a48fb98285d3936fb93aa07a2
--- /dev/null
+++ b/Project/state.py
@@ -0,0 +1,281 @@
+
+"""State models and enumerations"""
+# TODO: Define state models
+from __future__ import annotations
+import uuid # extra import
+
+from typing import Dict, List, Optional, Any, Literal
+from pydantic import BaseModel, Field, root_validator
+from datetime import datetime
+from enum import Enum
+
+
+class ProcessingStatus(str, Enum):
+ PENDING = "pending"
+ IN_PROGRESS = "in_progress"
+ COMPLETED = "completed"
+ FAILED = "failed"
+ CANCELLED = "cancelled"
+
+
+
+class RiskLevel(str, Enum):
+ LOW = "low"
+ MEDIUM = "medium"
+ HIGH = "high"
+ CRITICAL = "critical"
+
+
+class PaymentStatus(str, Enum):
+ NOT_STARTED = "not_started"
+ PENDING_APPROVAL = "pending_approval"
+ APPROVED = "approved"
+ SCHEDULED = "scheduled"
+ PAID = "paid"
+ FAILED = "failed"
+
+
+
+class ItemDetail(BaseModel):
+ item_id: Optional[str] = None
+ item_name: Optional[str] = None
+ # description: Optional[str] = None
+ quantity: int = Field(..., ge=0)
+ rate: float = Field(..., ge=0.0)
+ # total: Optional[float] = None
+ # unit: Optional[str] = None
+ amount: float = Field(..., ge=0.0)
+ category: Optional[str] = None
+
+
+class InvoiceData(BaseModel):
+ invoice_number: Optional[str] = None
+ order_id: Optional[str] = None
+ file_name: Optional[str] = None
+ customer_name: Optional[str] = None
+ invoice_date: Optional[datetime] = None
+ due_date: Optional[datetime] = None
+ currency: Optional[str] = "USD"
+ total: Optional[float] = None
+ # line_items: List[ItemDetail] = Field(default_factory=list)
+ raw_text: Optional[str] = None
+ item_details: Optional[list] = None
+ # confidence_scores: Dict[str, float] = Field(default_factory=dict)
+ extraction_confidence: Optional[float] = None
+
+class ValidationStatus(str, Enum):
+ NOT_STARTED = "not_started"
+ VALID = "valid"
+ INVALID = "invalid"
+ PARTIAL_MATCH = "partial_match"
+ MISSING_PO = "missing_po"
+
+class ValidationResult(BaseModel):
+ po_found: bool = False
+ quantity_match: bool = False
+ rate_match: bool = False
+ amount_match: bool = False
+ validation_status: ValidationStatus = ValidationStatus.NOT_STARTED
+ validation_result: Optional[str] = None
+ discrepencies: List[str] = Field(default_factory=list)
+ confidence_score: Optional[float] = None
+ # expected_amount: Optional[float] = None
+ po_data: Optional[Dict[str, Any]] = None
+
+
+class RiskAssessment(BaseModel):
+ risk_score: float = Field(0.0, ge=0.0, le=1.0)
+ risk_level: RiskLevel = RiskLevel.LOW
+ signals: List[str] = Field(default_factory=list)
+ vendor_status: Optional[str] = None
+ compliance_violations: List[str] = Field(default_factory=list)
+
+
+class PaymentDecision(BaseModel):
+ decision: Optional[Literal["auto_pay", "manual_approval", "hold", "reject"]]
+ status: PaymentStatus = PaymentStatus.NOT_STARTED
+ scheduled_date: Optional[datetime] = None
+ transaction_id: Optional[str] = None
+ attempts: int = 0
+ reason: Optional[str] = None
+
+
+class AuditTrail(BaseModel):
+ process_id: Optional[str] = None
+ timestamp: datetime = Field(default_factory=datetime.utcnow)
+ agent_name: str
+ action: str
+ details: Dict[str, Any] = Field(default_factory=dict)
+
+ class Config:
+ arbitrary_types_allowed = True
+
+
+class AgentMetrics(BaseModel):
+ processed_count: int = 0
+ avg_latency_ms: Optional[float] = None
+ last_run_at: Optional[datetime] = None
+ errors: int = 0
+ success_rate: Optional[float] = None
+
+
+class InvoiceProcessingState(BaseModel):
+ # Core identifiers
+ process_id: str = Field(
+ default_factory=lambda: f"proc_{datetime.utcnow().strftime('%Y%m%d_%H%M%S')}_{uuid.uuid4().hex[:6]}"
+ )
+ file_name: Optional[str] = None
+
+ # Processing status
+ overall_status: ProcessingStatus = ProcessingStatus.PENDING
+ current_agent: Optional[str] = None
+ workflow_type: str = "Standard"
+
+ # Agent Outputs
+ invoice_data: Optional[InvoiceData] = None
+ validation_result: Optional[ValidationResult] = None
+ validation_status: Optional[str] = None
+ risk_assessment: Optional[RiskAssessment] = None
+ payment_decision: Optional[PaymentDecision] = None
+ approval_chain: Optional[List[Dict[str, Any]]] = None
+
+ # Audit and Tracking
+ agent_name: Optional[str] = None
+ audit_trail: List[AuditTrail] = Field(default_factory=list)
+ agent_metrics: Dict[str, AgentMetrics] = Field(default_factory=dict)
+ compliance_report: Optional[Dict[str, Any]] = None
+ audit_summary: Optional[Dict[str, Any]] = None
+ reportable_events: Optional[List[Dict[str, Any]]] = None
+
+ # Escalation
+ escalation_required: bool = False
+ human_review_required: bool = False
+ escalation_details: Optional[str] = None
+ escalation_reason: Optional[str] = None
+
+ # Workflow Control
+ retry_count: int = 0
+ completed_agents: List[str] = Field(default_factory=list)
+
+ # Timestamps
+ created_at: datetime = Field(default_factory=datetime.utcnow)
+ updated_at: datetime = Field(default_factory=datetime.utcnow)
+
+ # Convenience Methods
+ def add_audit_entry(self, agent_name: str, action: str, status: Optional[ProcessingStatus] = None, details: Optional[Dict[str, Any]] = None, process_id: Optional[str] = None) -> None:
+ entry = AuditTrail(agent_name=agent_name, action=action, status = status or self.overall_status, details=details or {}, process_id=process_id)
+ print("entry.....", entry)
+ print("self.audit_trail...", self.audit_trail)
+ self.audit_trail.append(entry)
+ self.updated_at = datetime.utcnow()
+
+ def add_agent_metric(self, agent: str, processed: int = 0, latency_ms: Optional[float] = None, errors: int = 0) -> None:
+ metrics = self.agent_metrics.get(agent) or AgentMetrics()
+ metrics.processed_count += processed
+ metrics.errors += errors
+
+ if latency_ms is not None:
+ if metrics.avg_latency_ms is None:
+ metrics.avg_latency_ms = latency_ms
+ else:
+ metrics.avg_latency_ms = (metrics.avg_latency_ms + latency_ms) / 2.0
+ metrics.last_run_at = datetime.utcnow()
+
+ if metrics.processed_count > 0:
+ metrics.success_rate = max(0.0, 1.0 - (metrics.errors / max(1, metrics.processed_count)))
+ self.agent_metrics[agent] = metrics
+ self.updated_at = datetime.utcnow()
+
+ def update_agent_metrics(self, agent_name: str, success: bool, duration_ms: float):
+ """
+ Update or create performance metrics for an agent.
+ Expected structure aligns with test_9_agent_metrics: attributes like executions, success_count, failure_count.
+ """
+ # Ensure agent_metrics dict exists
+ if self.agent_metrics is None:
+ self.agent_metrics = {}
+
+ # Get or initialize metrics object
+ metrics = self.agent_metrics.get(agent_name)
+
+ # If existing metrics is a dict or None, replace with a new AgentMetrics-like object
+ if isinstance(metrics, dict) or metrics is None:
+ metrics = type("DynamicMetrics", (), {})()
+ metrics.executions = 0
+ metrics.successes = 0
+ metrics.failure_count = 0
+ metrics.total_duration_ms = 0.0
+ metrics.avg_duration_ms = 0.0
+
+ # Update fields
+ metrics.executions += 1
+ if success:
+ metrics.successes += 1
+ else:
+ metrics.failure_count += 1
+
+ metrics.total_duration_ms += duration_ms
+ metrics.avg_duration_ms = round(metrics.total_duration_ms / metrics.executions, 2)
+
+ # Save back
+ self.agent_metrics[agent_name] = metrics
+ self.updated_at = datetime.utcnow()
+
+
+ def mark_agent_completed(self, agent: str) -> None:
+ if agent not in self.completed_agents:
+ self.completed_agents.append(agent)
+ self.updated_at = datetime.utcnow()
+
+ def requires_escalation(self, risk_threshold: float = 0.6, confidence_threshold: float = 0.7) -> bool:
+ if self.validation_result and self.validation_result.validation_status == ValidationStatus.INVALID:
+ return True
+ if self.risk_assessment and self.risk_assessment.risk_score >= risk_threshold:
+ return True
+ if self.validation_result and self.validation_result.confidence_score is not None and self.validation_result.confidence_score < confidence_threshold:
+ return True
+ return False
+
+ def to_dict(self) -> Dict[str, Any]:
+ return self.model_dump()
+
+ @root_validator(pre=True)
+ def ensure_timestamps(cls, values: Dict[str, Any]) -> Dict[str, Any]:
+ if "created_at" not in values or values.get("created_at") is None:
+ values["created_at"] = datetime.utcnow()
+ if "updated_at" not in values or values.get("updated_at") is None:
+ values["updated_at"] = datetime.utcnow()
+ return values
+
+
+class WorkflowConfig(BaseModel):
+ name: str
+ auto_approve_threshold: float = 0.3
+ auto_approve_amount_limit: Optional[float] = None
+ tolerance_percent: float = 5.0
+ escalation_rules: Dict[str, Any] = Field(default_factory=dict)
+
+
+WORKFLOW_CONFIGS: Dict[str, WorkflowConfig] = {
+ "standard": WorkflowConfig(
+ name="standard",
+ auto_approve_threshold=0.3,
+ auto_approve_amount_limit=10000.0,
+ tolerance_percent=5.0,
+ escalation_rules={"sla_hours": 24},
+ ),
+ "high_value": WorkflowConfig(
+ name="high_value",
+ auto_approve_threshold=0.1,
+ auto_approve_amount_limit=5000.0,
+ tolerance_percent=2.0,
+ escalation_rules={"require_cfo": True, "sla_hours": 12},
+ ),
+ "expedited": WorkflowConfig(
+ name="expedited",
+ auto_approve_threshold=0.5,
+ auto_approve_amount_limit=5000.0,
+ tolerance_percent=10.0,
+ escalation_rules={"skip_manual_for_low_risk": True},
+ ),
+}
diff --git a/Project/utils/__init__.py b/Project/utils/__init__.py
new file mode 100644
index 0000000000000000000000000000000000000000..3374942f29932bb1e22800526db92f1d9c411dec
--- /dev/null
+++ b/Project/utils/__init__.py
@@ -0,0 +1,8 @@
+"""
+Utility Modules for Invoice Processing Workflow
+
+Contains:
+- logger.py: Structured logging utilities
+"""
+
+__all__ = []
\ No newline at end of file
diff --git a/Project/utils/logger.py b/Project/utils/logger.py
new file mode 100644
index 0000000000000000000000000000000000000000..7a0178c7ab87ace1f1dafa300fdf8247ee6d1d27
--- /dev/null
+++ b/Project/utils/logger.py
@@ -0,0 +1,213 @@
+"""
+Centralized Logging Utility for Invoice Processing System
+Provides structured logging with different levels and formatters
+"""
+
+import logging
+import sys
+from datetime import datetime
+from typing import Optional
+from pathlib import Path
+
+
+class ColoredFormatter(logging.Formatter):
+ """Custom formatter with colors for console output"""
+
+ # Color codes
+ COLORS = {
+ 'DEBUG': '\033[36m', # Cyan
+ 'INFO': '\033[32m', # Green
+ 'WARNING': '\033[33m', # Yellow
+ 'ERROR': '\033[31m', # Red
+ 'CRITICAL': '\033[35m', # Magenta
+ 'RESET': '\033[0m' # Reset
+ }
+
+ def format(self, record):
+ # Add color to levelname
+ if record.levelname in self.COLORS:
+ record.levelname = (
+ f"{self.COLORS[record.levelname]}{record.levelname}"
+ f"{self.COLORS['RESET']}"
+ )
+
+ return super().format(record)
+
+
+def setup_logging(log_level: str = "INFO", log_file: Optional[str] = None):
+ """
+ Setup centralized logging configuration
+
+ Args:
+ log_level: Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL)
+ log_file: Optional log file path
+ """
+
+ # Create logs directory if it doesn't exist
+ if log_file:
+ log_path = Path(log_file)
+ log_path.parent.mkdir(parents=True, exist_ok=True)
+
+ # Configure root logger
+ root_logger = logging.getLogger()
+ root_logger.setLevel(getattr(logging, log_level.upper()))
+
+ # Clear existing handlers
+ root_logger.handlers.clear()
+
+ # Console handler with colors
+ console_handler = logging.StreamHandler(sys.stdout)
+ console_formatter = ColoredFormatter(
+ fmt='%(asctime)s | %(levelname)-8s | %(name)-20s | %(message)s',
+ datefmt='%Y-%m-%d %H:%M:%S'
+ )
+ console_handler.setFormatter(console_formatter)
+ root_logger.addHandler(console_handler)
+
+ # File handler (if specified)
+ if log_file:
+ file_handler = logging.FileHandler(log_file)
+ file_formatter = logging.Formatter(
+ fmt='%(asctime)s | %(levelname)-8s | %(name)-20s | %(funcName)-15s | %(message)s',
+ datefmt='%Y-%m-%d %H:%M:%S'
+ )
+ file_handler.setFormatter(file_formatter)
+ root_logger.addHandler(file_handler)
+
+ # Set specific logger levels
+ logging.getLogger("httpx").setLevel(logging.WARNING)
+ logging.getLogger("urllib3").setLevel(logging.WARNING)
+ logging.getLogger("requests").setLevel(logging.WARNING)
+
+
+def get_logger(name: str) -> logging.Logger:
+ """
+ Get a logger instance with the specified name
+
+ Args:
+ name: Logger name (typically module or class name)
+
+ Returns:
+ Configured logger instance
+ """
+ return logging.getLogger(name)
+
+
+class StructuredLogger:
+ """
+ Structured logger for agent operations with consistent formatting
+ """
+
+ def __init__(self, name: str):
+ self.logger = get_logger(name)
+ self.name = name
+
+ def log_agent_start(self, agent_name: str, process_id: str, **kwargs):
+ """Log agent execution start"""
+ self.logger.info(
+ f"[START] Agent {agent_name} started for process {process_id}",
+ extra={"agent": agent_name, "process_id": process_id, **kwargs}
+ )
+
+ def log_agent_complete(self, agent_name: str, process_id: str,
+ duration_ms: int, **kwargs):
+ """Log agent execution completion"""
+ self.logger.info(
+ f"[COMPLETE] Agent {agent_name} completed for process {process_id} "
+ f"in {duration_ms}ms",
+ extra={
+ "agent": agent_name,
+ "process_id": process_id,
+ "duration_ms": duration_ms,
+ **kwargs
+ }
+ )
+
+ def log_agent_error(self, agent_name: str, process_id: str,
+ error: Exception, **kwargs):
+ """Log agent execution error"""
+ self.logger.error(
+ f"[ERROR] Agent {agent_name} failed for process {process_id}: {str(error)}",
+ extra={
+ "agent": agent_name,
+ "process_id": process_id,
+ "error_type": type(error).__name__,
+ "error_message": str(error),
+ **kwargs
+ },
+ exc_info=True
+ )
+
+ def log_decision(self, agent_name: str, process_id: str,
+ decision: str, reasoning: str, confidence: float = None):
+ """Log agent decision"""
+ message = f"[DECISION] Agent {agent_name} decided: {decision} - {reasoning}"
+ if confidence:
+ message += f" (confidence: {confidence:.2f})"
+
+ self.logger.info(
+ message,
+ extra={
+ "agent": agent_name,
+ "process_id": process_id,
+ "decision": decision,
+ "reasoning": reasoning,
+ "confidence": confidence
+ }
+ )
+
+ def log_escalation(self, agent_name: str, process_id: str,
+ reason: str, **kwargs):
+ """Log escalation event"""
+ self.logger.warning(
+ f"[ESCALATION] Agent {agent_name} escalating process {process_id}: {reason}",
+ extra={
+ "agent": agent_name,
+ "process_id": process_id,
+ "escalation_reason": reason,
+ **kwargs
+ }
+ )
+
+ def log_workflow_start(self, workflow_type: str, process_id: str, **kwargs):
+ """Log workflow start"""
+ self.logger.info(
+ f"[WORKFLOW] Starting {workflow_type} workflow for process {process_id}",
+ extra={
+ "workflow_type": workflow_type,
+ "process_id": process_id,
+ **kwargs
+ }
+ )
+
+ def log_workflow_complete(self, workflow_type: str, process_id: str,
+ duration_ms: int, **kwargs):
+ """Log workflow completion"""
+ self.logger.info(
+ f"[WORKFLOW_COMPLETE] Completed {workflow_type} workflow for process {process_id} "
+ f"in {duration_ms}ms",
+ extra={
+ "workflow_type": workflow_type,
+ "process_id": process_id,
+ "duration_ms": duration_ms,
+ **kwargs
+ }
+ )
+
+ def log_metric(self, metric_name: str, value: float, **kwargs):
+ """Log metric value"""
+ self.logger.info(
+ f"[METRIC] {metric_name}: {value}",
+ extra={
+ "metric_name": metric_name,
+ "metric_value": value,
+ **kwargs
+ }
+ )
+
+
+# Initialize default logging
+setup_logging()
+
+# Create default structured logger
+default_logger = StructuredLogger("invoice_system")
\ No newline at end of file
diff --git a/Project/workflow/__init__.py b/Project/workflow/__init__.py
new file mode 100644
index 0000000000000000000000000000000000000000..5ca9c0f139b5b6d16c8f228f853a622d0536d353
--- /dev/null
+++ b/Project/workflow/__init__.py
@@ -0,0 +1,5 @@
+"""Workflow module - exports the main workflow."""
+
+from workflow.workflow import InvoiceProcessingGraph
+
+__all__ = ["InvoiceProcessingGraph"]
\ No newline at end of file
diff --git a/Project/workflow/workflow.py b/Project/workflow/workflow.py
new file mode 100644
index 0000000000000000000000000000000000000000..170e77ed8a46c3ca88266c4caa1db5eca1d87e35
--- /dev/null
+++ b/Project/workflow/workflow.py
@@ -0,0 +1,4 @@
+"""Workflow module"""
+# TODO: Setup workflow exports
+from graph import InvoiceProcessingGraph, get_workflow
+__all__ = ["InvoiceProcessingGraph", "get_workflow"]
\ No newline at end of file