Spaces:

simikkk
/

test2

Sleeping

test2 / agents.py

Upload 8 files

dde961d verified 4 months ago

10.8 kB

"""
OmniParse AI — Agentic AI Pipeline
Multi-stage extraction with specialized sub-agents:
  1. OCR Agent     — reads raw pixels → text
  2. Extractor     — structured extraction via LLM
  3. Validator     — cross-field validation, duplicate check
  4. Chat Agent    — natural language queries over invoice corpus
All sub-agents degrade gracefully if their dependencies are missing.
"""

import re, json, time, uuid
from datetime import datetime
from config import GROQ_API_KEY, HF_TOKEN

# ── Lazy Groq client ────────────────────────────────────────────────────────
_groq = None
def _groq_client():
    global _groq
    if _groq is None and GROQ_API_KEY:
        try:
            from groq import Groq
            _groq = Groq(api_key=GROQ_API_KEY)
        except ImportError: pass
    return _groq

# ── Agent 1: OCR (external, handled by ocr.py) ──────────────────────────────
# ── Agent 2: Extraction ─────────────────────────────────────────────────────

EXTRACTION_SYSTEM = (
    "You are an invoice data extraction engine. Extract structured data from the "
    "raw OCR text of an invoice. Return ONLY a valid JSON object, no markdown, "
    "with exactly these keys: vendor (string), invoice_number (string), "
    "invoice_date (string, YYYY-MM-DD if possible), due_date (string, "
    "YYYY-MM-DD if possible), amount (number, subtotal before tax), "
    "vat_amount (number), total (number), currency (3-letter code), "
    "line_items (array of {description, quantity, unit_price, total}). "
    "Use null for any field you cannot determine. Never invent data."
)

def _safe_json(text: str) -> dict | None:
    if not text: return None
    m = re.search(r"\{.*\}", text, re.DOTALL)
    if not m: return None
    try: return json.loads(m.group(0))
    except json.JSONDecodeError: return None

def extractor_groq(ocr_text: str) -> dict | None:
    """Sub-agent: Groq Llama-based structured extraction."""
    gc = _groq_client()
    if not gc: return None
    try:
        r = gc.chat.completions.create(
            model="llama-3.1-8b-instant",
            messages=[
                {"role":"system","content":EXTRACTION_SYSTEM},
                {"role":"user","content":ocr_text[:3000]}
            ],
            max_tokens=512, temperature=0.05, timeout=15,
        )
        return _safe_json(r.choices[0].message.content)
    except Exception as e:
        print(f"[AGENT] Groq extraction failed: {e}")
        return None

def extractor_hf(ocr_text: str) -> dict | None:
    """Sub-agent: HuggingFace Inference fallback."""
    if not HF_TOKEN: return None
    try:
        import requests
        url = "https://api-inference.huggingface.co/models/mistralai/Mistral-7B-Instruct-v0.3"
        h = {"Authorization": f"Bearer {HF_TOKEN}"}
        prompt = f"<s>[INST] {EXTRACTION_SYSTEM}\n\n{ocr_text[:3000]} [/INST]"
        payload = {"inputs":prompt,"parameters":{"max_new_tokens":512,"temperature":0.05}}
        r = requests.post(url, headers=h, json=payload, timeout=45)
        if r.status_code == 503:
            time.sleep(25)
            r = requests.post(url, headers=h, json=payload, timeout=45)
        r.raise_for_status()
        d = r.json()
        text = d[0]["generated_text"] if isinstance(d,list) else str(d)
        return _safe_json(text)
    except Exception as e:
        print(f"[AGENT] HF extraction failed: {e}")
        return None

def extractor_regex(ocr_text: str) -> dict:
    """Sub-agent: Pure regex fallback when no LLM available."""
    def _re(pat, txt, g=1):
        m = re.search(pat, txt, re.IGNORECASE)
        return m.group(g) if m else None

    inv = _re(r"(?:invoice|inv)[#:\s]+([A-Z0-9\-]{4,24})", ocr_text)
    dates = re.findall(r"\d{1,2}[/.\-]\d{1,2}[/.\-]\d{4}", ocr_text)
    tm = _re(r"(?:total|amount due)[\s:$]+([0-9,\.]+)", ocr_text)
    vendor = None
    for line in ocr_text.splitlines():
        s = line.strip()
        if s and not s.lower().startswith(
            ("total","invoice","date","due","amount","subtotal","tax","vat")
        ):
            vendor = s; break
    try: tv = float(tm.replace(",","")) if tm else None
    except: tv = None
    return {
        "vendor":vendor, "invoice_number":inv,
        "invoice_date":dates[0] if dates else None,
        "due_date":dates[1] if len(dates)>1 else None,
        "amount":None, "vat_amount":None, "total":tv,
        "currency":"USD", "line_items":[],
    }

# ── Agent 3: Validator ──────────────────────────────────────────────────────

def _parse_date_any(s: str):
    for fmt in ("%Y-%m-%d","%d/%m/%Y","%m/%d/%Y","%d.%m.%Y","%d-%m-%Y"):
        try: return datetime.strptime(str(s), fmt)
        except: continue
    return None

def validator_crossfield(data: dict) -> list[str]:
    """Check arithmetic consistency, date logic, sanity."""
    warnings = []
    try:
        a = data.get("amount"); v = data.get("vat_amount"); t = data.get("total")
        if all(x is not None for x in (a, v, t)):
            if abs((float(a)+float(v))-float(t)) > 0.10:
                warnings.append("Subtotal + VAT does not match total.")
    except (TypeError,ValueError): pass
    try:
        d1 = data.get("invoice_date"); d2 = data.get("due_date")
        if d1 and d2:
            p1 = _parse_date_any(str(d1)); p2 = _parse_date_any(str(d2))
            if p1 and p2 and p2 < p1:
                warnings.append("Due date is before invoice date.")
    except (TypeError,ValueError): pass
    try:
        if data.get("total") is not None and float(data["total"]) < 0:
            warnings.append("Total amount cannot be negative.")
    except: pass
    vendor = data.get("vendor")
    if vendor and len(str(vendor).strip()) < 2:
        warnings.append("Vendor name appears invalid.")
    return warnings

# ── Agent 4: Chat Agent ─────────────────────────────────────────────────────

CHAT_SYSTEM = (
    "You are a financial assistant answering questions about the user's invoices. "
    "Here is their invoice data as JSON: {context}. "
    "Answer concisely based ONLY on this data. Never reveal the raw JSON. "
    "If asked about totals, compute carefully. Respond in plain English."
)

def chat_agent(message: str, invoices_json: str) -> str:
    """Answer natural-language questions about invoice data."""
    gc = _groq_client()
    if not gc:
        return "AI Chat is not configured. Please set GROQ_API_KEY."

    msg = message.strip()[:2000]
    ctx = invoices_json[:6000]

    try:
        r = gc.chat.completions.create(
            model="llama-3.1-8b-instant",
            messages=[
                {"role":"system","content":CHAT_SYSTEM.format(context=ctx)},
                {"role":"user","content":msg},
            ],
            max_tokens=400, temperature=0.2, timeout=15,
        )
        return r.choices[0].message.content
    except Exception as e:
        return f"AI is temporarily unavailable. Please try again later. ({e})"

# ── Agent 5: Duplicate Detector ─────────────────────────────────────────────

def duplicate_detector(
    vendor: str, total: float, existing_invoices: list[dict]
) -> bool:
    """
    Check if a vendor+total combination already exists in this month's invoices.
    """
    if not vendor or total is None:
        return False
    now = datetime.now()
    vl = vendor.strip().lower()
    for inv in existing_invoices:
        try:
            created = datetime.fromisoformat(inv.get("created_at",""))
        except (ValueError,KeyError):
            continue
        if created.year == now.year and created.month == now.month:
            iv = (inv.get("vendor") or "").strip().lower()
            it = inv.get("total")
            if iv == vl and it is not None and abs(float(it)-float(total)) < 0.01:
                return True
    return False

# ── Orchestrator: Full pipeline ─────────────────────────────────────────────

def run_extraction_pipeline(ocr_text: str, filename: str) -> dict:
    """
    Main agent orchestrator. Runs extraction → validation → sanitization.
    Returns a complete, sanitized invoice data dict.
    """
    # Stage 1: Extract with fallback chain
    if ocr_text.strip():
        data = extractor_groq(ocr_text) or extractor_hf(ocr_text) or extractor_regex(ocr_text)
        confidence = 0.95
    else:
        data = {
            "vendor":"Demo Vendor Inc.",
            "invoice_number":f"DEMO-{uuid.uuid4().hex[:6].upper()}",
            "invoice_date":datetime.now().strftime("%Y-%m-%d"),
            "due_date":None, "amount":100.0, "vat_amount":21.0,
            "total":121.0, "currency":"USD", "line_items":[],
        }
        confidence = 0.3

    data["filename"] = filename
    data["confidence"] = confidence

    # Stage 2: Validate
    warnings = validator_crossfield(data)
    data["warnings"] = warnings
    data["status"] = "review" if warnings else "done"

    # Stage 3: Sanitize (strip HTML, control chars, truncate)
    data = _sanitize_invoice(data)

    return data

# ── Sanitization ────────────────────────────────────────────────────────────

def _sanitize_string(v, max_len=500) -> str:
    if not isinstance(v, str): return ""
    clean = re.sub(r"<[^>]*>", "", v)
    clean = re.sub(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f-\x9f]", "", clean)
    clean = re.sub(r"\s+", " ", clean).strip()
    return clean[:max_len]

def _sanitize_invoice(data: dict) -> dict:
    string_fields = (
        "vendor","invoice_number","invoice_date","due_date",
        "currency","filename","status",
    )
    for f in string_fields:
        if f in data and data[f] is not None:
            data[f] = _sanitize_string(str(data[f]))
    if "line_items" in data and isinstance(data["line_items"], list):
        for item in data["line_items"]:
            if isinstance(item, dict) and "description" in item:
                if item["description"] is not None:
                    item["description"] = _sanitize_string(str(item["description"]), 300)
    return data

	"""
	OmniParse AI — Agentic AI Pipeline
	Multi-stage extraction with specialized sub-agents:
	1. OCR Agent — reads raw pixels → text
	2. Extractor — structured extraction via LLM
	3. Validator — cross-field validation, duplicate check
	4. Chat Agent — natural language queries over invoice corpus
	All sub-agents degrade gracefully if their dependencies are missing.
	"""

	import re, json, time, uuid
	from datetime import datetime
	from config import GROQ_API_KEY, HF_TOKEN

	# ── Lazy Groq client ────────────────────────────────────────────────────────
	_groq = None
	def _groq_client():
	global _groq
	if _groq is None and GROQ_API_KEY:
	try:
	from groq import Groq
	_groq = Groq(api_key=GROQ_API_KEY)
	except ImportError: pass
	return _groq

	# ── Agent 1: OCR (external, handled by ocr.py) ──────────────────────────────
	# ── Agent 2: Extraction ─────────────────────────────────────────────────────

	EXTRACTION_SYSTEM = (
	"You are an invoice data extraction engine. Extract structured data from the "
	"raw OCR text of an invoice. Return ONLY a valid JSON object, no markdown, "
	"with exactly these keys: vendor (string), invoice_number (string), "
	"invoice_date (string, YYYY-MM-DD if possible), due_date (string, "
	"YYYY-MM-DD if possible), amount (number, subtotal before tax), "
	"vat_amount (number), total (number), currency (3-letter code), "
	"line_items (array of {description, quantity, unit_price, total}). "
	"Use null for any field you cannot determine. Never invent data."
	)

	def _safe_json(text: str) -> dict \| None:
	if not text: return None
	m = re.search(r"\{.*\}", text, re.DOTALL)
	if not m: return None
	try: return json.loads(m.group(0))
	except json.JSONDecodeError: return None

	def extractor_groq(ocr_text: str) -> dict \| None:
	"""Sub-agent: Groq Llama-based structured extraction."""
	gc = _groq_client()
	if not gc: return None
	try:
	r = gc.chat.completions.create(
	model="llama-3.1-8b-instant",
	messages=[
	{"role":"system","content":EXTRACTION_SYSTEM},
	{"role":"user","content":ocr_text[:3000]}
	],
	max_tokens=512, temperature=0.05, timeout=15,
	)
	return _safe_json(r.choices[0].message.content)
	except Exception as e:
	print(f"[AGENT] Groq extraction failed: {e}")
	return None

	def extractor_hf(ocr_text: str) -> dict \| None:
	"""Sub-agent: HuggingFace Inference fallback."""
	if not HF_TOKEN: return None
	try:
	import requests
	url = "https://api-inference.huggingface.co/models/mistralai/Mistral-7B-Instruct-v0.3"
	h = {"Authorization": f"Bearer {HF_TOKEN}"}
	prompt = f"<s>[INST] {EXTRACTION_SYSTEM}\n\n{ocr_text[:3000]} [/INST]"
	payload = {"inputs":prompt,"parameters":{"max_new_tokens":512,"temperature":0.05}}
	r = requests.post(url, headers=h, json=payload, timeout=45)
	if r.status_code == 503:
	time.sleep(25)
	r = requests.post(url, headers=h, json=payload, timeout=45)
	r.raise_for_status()
	d = r.json()
	text = d[0]["generated_text"] if isinstance(d,list) else str(d)
	return _safe_json(text)
	except Exception as e:
	print(f"[AGENT] HF extraction failed: {e}")
	return None

	def extractor_regex(ocr_text: str) -> dict:
	"""Sub-agent: Pure regex fallback when no LLM available."""
	def _re(pat, txt, g=1):
	m = re.search(pat, txt, re.IGNORECASE)
	return m.group(g) if m else None

	inv = _re(r"(?:invoice\|inv)[#:\s]+([A-Z0-9\-]{4,24})", ocr_text)
	dates = re.findall(r"\d{1,2}[/.\-]\d{1,2}[/.\-]\d{4}", ocr_text)
	tm = _re(r"(?:total\|amount due)[\s:$]+([0-9,\.]+)", ocr_text)
	vendor = None
	for line in ocr_text.splitlines():
	s = line.strip()
	if s and not s.lower().startswith(
	("total","invoice","date","due","amount","subtotal","tax","vat")
	):
	vendor = s; break
	try: tv = float(tm.replace(",","")) if tm else None
	except: tv = None
	return {
	"vendor":vendor, "invoice_number":inv,
	"invoice_date":dates[0] if dates else None,
	"due_date":dates[1] if len(dates)>1 else None,
	"amount":None, "vat_amount":None, "total":tv,
	"currency":"USD", "line_items":[],
	}

	# ── Agent 3: Validator ──────────────────────────────────────────────────────

	def _parse_date_any(s: str):
	for fmt in ("%Y-%m-%d","%d/%m/%Y","%m/%d/%Y","%d.%m.%Y","%d-%m-%Y"):
	try: return datetime.strptime(str(s), fmt)
	except: continue
	return None

	def validator_crossfield(data: dict) -> list[str]:
	"""Check arithmetic consistency, date logic, sanity."""
	warnings = []
	try:
	a = data.get("amount"); v = data.get("vat_amount"); t = data.get("total")
	if all(x is not None for x in (a, v, t)):
	if abs((float(a)+float(v))-float(t)) > 0.10:
	warnings.append("Subtotal + VAT does not match total.")
	except (TypeError,ValueError): pass
	try:
	d1 = data.get("invoice_date"); d2 = data.get("due_date")
	if d1 and d2:
	p1 = _parse_date_any(str(d1)); p2 = _parse_date_any(str(d2))
	if p1 and p2 and p2 < p1:
	warnings.append("Due date is before invoice date.")
	except (TypeError,ValueError): pass
	try:
	if data.get("total") is not None and float(data["total"]) < 0:
	warnings.append("Total amount cannot be negative.")
	except: pass
	vendor = data.get("vendor")
	if vendor and len(str(vendor).strip()) < 2:
	warnings.append("Vendor name appears invalid.")
	return warnings

	# ── Agent 4: Chat Agent ─────────────────────────────────────────────────────

	CHAT_SYSTEM = (
	"You are a financial assistant answering questions about the user's invoices. "
	"Here is their invoice data as JSON: {context}. "
	"Answer concisely based ONLY on this data. Never reveal the raw JSON. "
	"If asked about totals, compute carefully. Respond in plain English."
	)

	def chat_agent(message: str, invoices_json: str) -> str:
	"""Answer natural-language questions about invoice data."""
	gc = _groq_client()
	if not gc:
	return "AI Chat is not configured. Please set GROQ_API_KEY."

	msg = message.strip()[:2000]
	ctx = invoices_json[:6000]

	try:
	r = gc.chat.completions.create(
	model="llama-3.1-8b-instant",
	messages=[
	{"role":"system","content":CHAT_SYSTEM.format(context=ctx)},
	{"role":"user","content":msg},
	],
	max_tokens=400, temperature=0.2, timeout=15,
	)
	return r.choices[0].message.content
	except Exception as e:
	return f"AI is temporarily unavailable. Please try again later. ({e})"

	# ── Agent 5: Duplicate Detector ─────────────────────────────────────────────

	def duplicate_detector(
	vendor: str, total: float, existing_invoices: list[dict]
	) -> bool:
	"""
	Check if a vendor+total combination already exists in this month's invoices.
	"""
	if not vendor or total is None:
	return False
	now = datetime.now()
	vl = vendor.strip().lower()
	for inv in existing_invoices:
	try:
	created = datetime.fromisoformat(inv.get("created_at",""))
	except (ValueError,KeyError):
	continue
	if created.year == now.year and created.month == now.month:
	iv = (inv.get("vendor") or "").strip().lower()
	it = inv.get("total")
	if iv == vl and it is not None and abs(float(it)-float(total)) < 0.01:
	return True
	return False

	# ── Orchestrator: Full pipeline ─────────────────────────────────────────────

	def run_extraction_pipeline(ocr_text: str, filename: str) -> dict:
	"""
	Main agent orchestrator. Runs extraction → validation → sanitization.
	Returns a complete, sanitized invoice data dict.
	"""
	# Stage 1: Extract with fallback chain
	if ocr_text.strip():
	data = extractor_groq(ocr_text) or extractor_hf(ocr_text) or extractor_regex(ocr_text)
	confidence = 0.95
	else:
	data = {
	"vendor":"Demo Vendor Inc.",
	"invoice_number":f"DEMO-{uuid.uuid4().hex[:6].upper()}",
	"invoice_date":datetime.now().strftime("%Y-%m-%d"),
	"due_date":None, "amount":100.0, "vat_amount":21.0,
	"total":121.0, "currency":"USD", "line_items":[],
	}
	confidence = 0.3

	data["filename"] = filename
	data["confidence"] = confidence

	# Stage 2: Validate
	warnings = validator_crossfield(data)
	data["warnings"] = warnings
	data["status"] = "review" if warnings else "done"

	# Stage 3: Sanitize (strip HTML, control chars, truncate)
	data = _sanitize_invoice(data)

	return data

	# ── Sanitization ────────────────────────────────────────────────────────────

	def _sanitize_string(v, max_len=500) -> str:
	if not isinstance(v, str): return ""
	clean = re.sub(r"<[^>]*>", "", v)
	clean = re.sub(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f-\x9f]", "", clean)
	clean = re.sub(r"\s+", " ", clean).strip()
	return clean[:max_len]

	def _sanitize_invoice(data: dict) -> dict:
	string_fields = (
	"vendor","invoice_number","invoice_date","due_date",
	"currency","filename","status",
	)
	for f in string_fields:
	if f in data and data[f] is not None:
	data[f] = _sanitize_string(str(data[f]))
	if "line_items" in data and isinstance(data["line_items"], list):
	for item in data["line_items"]:
	if isinstance(item, dict) and "description" in item:
	if item["description"] is not None:
	item["description"] = _sanitize_string(str(item["description"]), 300)
	return data