Skip to content

📥 Extraction Pipelines

pipelines/extraction/ implements the hexa runtime: an extensible, stateful pipeline that fetches a source document and reduces it to normalized expense records, flagging anything ambiguous.

🏗 Common Runtime

Concept Where Purpose
BaseExtractionPipeline shared Template method: fetch → stage → emit; tracks ambiguous state
Fetcher shared Loads raw content (file bytes, emails, llama docs)
ExtractionStage shared A single reduction step (headers, rows, reconciliation)
PauseForResolution shared Raised when numbers can't be reconciled; workflow pauses
get_families_to_enrich shared Families that should continue to enrichment

Every family registers a factory so pipelines/enrichment/ can build nodes from config at runtime.

📂 Extraction Families

Family Payload Stages (typical)
pdf Bank statement PDFs page split → text extraction → balance reconciliation
csv CSV exports header detect → row mapping → dtype normalization
xlsx Excel exports (openpyxl) sheet select → header detect → row mapping (Axis/ICICI first)
email IMAP mailbox + attachments traversal paradigm → attachment dispatch to family
llama LlamaParse documents page distribution → markdown parse → amount extraction

Cross-cutting utilities (pipelines/extraction/shared/)

  • tabular.py — generic tabular parsing helpers (used by CSV/XLSX families)
  • files.py, utils.py — content sniffing, date/amount normalization
  • ambiguity.py — PauseForResolution + AmbiguityCandidate helpers
  • extractor.py — the BaseExtractionPipeline template and registry

⚠️ Ambiguity & Reconciliation

Amount/balance reconciliation is the safety net: if extracted amounts don't balance against the statement, the pipeline raises PauseForResolution, the fetch request flips to paused, and the candidate is stored. A human (or later the LLM resolver) chooses the correct tuple; the worker resets the retry counter on pause.

🔄 Lifecycle Inside a Run

  1. Worker polls pending fetch requests (status == "pending").
  2. PipelineOrchestrator.run_pipeline(fr_id):
  3. picks the family from fr.source.type + bank,
  4. runs extraction stages to raw Expense candidates,
  5. keeps unresolved candidates as ambiguities,
  6. forwards the clean set to enrichment,
  7. Enrichment assigns entity/tag; expenses are persisted; SSE progress events stream the whole way.

Components · Fetch Request Workflow · Development