Invoice AI Backend 🚀

A modern, production-ready, highly performant backend built with FastAPI to process and manage invoice details with local deep learning models (LayoutLMv3) and semantic search (FAISS RAG).

---

🛠️ System Architecture

The following diagram illustrates the complete, decoupled processing pipeline:

graph TD
    Client[Client Browser / API Client] -->|1. POST /extract-invoice| API[FastAPI Gateway]
    API -->|2. Save File| Disk[(Local Uploads Volume)]
    API -->|3. Register Job| DB[(PostgreSQL Database)]
    API -->|4. Push Task| Broker[Redis Message Broker]
    API -->|5. 202 Accepted job_id| Client
    
    Broker -->|6. Consume Task| Worker[Celery Processing Worker]
    Worker -->|7. Load Image| Disk
    Worker -->|8. Run Pipeline| Inference[Inference Pipeline]
    
    subgraph Inference Pipeline
        Inference -->|8a. Preprocess & OCR| OCR[PaddleOCR]
        OCR -->|8b. Spatial Word Tokens| LLM[LayoutLMv3 Token Classification]
        LLM -->|8c. Grounded Entities| Post[Post-Processing Heuristics]
        Post -->|8d. Structured Line Items| Table[Table Parsing Engine]
    end
    
    Inference -->|9. Write Relational Records| DB
    Inference -->|10. Embed Chunks| Embedder[Sentence-Transformers]
    Embedder -->|11. Ingest dense vectors| FAISS[FAISS Vector Store]
    Inference -->|12. Set Job status to COMPLETED| DB
    
    Client -->|13. GET /jobs/job_id| API
    API -->|14. Query Job & Invoice| DB
    API -->|15. Return JSON Result| Client

Loading

---

🗄️ Relational Database Schema (ERD)

The relational schema decouples vector indexing metadata from core business entities:

erDiagram
    jobs ||--o| invoices : "links on completion"
    invoices ||--o{ line_items : "contains"
    invoices ||--o{ embedding_metadata : "maps vectors to"

    invoices {
        uuid invoice_id PK
        string raw_invoice_id
        string seller
        string buyer
        string invoice_date
        string due_date
        string currency
        numeric total_amount
        string processing_status
        float confidence_score
        json extraction_confidence
        json extraction_source
        json extraction_metadata
        float processing_time_ms
        string model_version
        string ocr_engine
        datetime created_at
        datetime updated_at
    }

    line_items {
        uuid line_item_id PK
        uuid invoice_id FK
        string item_name
        string quantity
        string unit_price
        string tax
        string total
        datetime created_at
        datetime updated_at
    }

    embedding_metadata {
        uuid embedding_id PK
        uuid invoice_id FK
        string chunk_type
        integer chunk_index
        integer faiss_index
        datetime created_at
    }

    jobs {
        uuid job_id PK
        string status
        string file_path
        integer retry_count
        datetime last_retry_at
        string failure_reason
        string error_message
        datetime started_at
        datetime completed_at
        string worker_hostname
        uuid invoice_id FK
        datetime created_at
        datetime updated_at
    }

    users {
        uuid id PK
        string username
        string hashed_password
        datetime created_at
    }

Loading

---

📖 API Reference

Endpoint	Method	Description	Request Payload / Params	Response Payload
/	GET	Home / Metadata endpoint.	None	{"message": str, "environment": str}
/health	GET	Liveness probe checking API health.	None	{"status": "healthy", ...}
/ready	GET	Readiness probe checking Postgres, models, and FAISS.	None	{"status": "ready" | "not_ready", "checks": dict}
/auth/register	POST	Registers a new user.	{"username": str, "password": str}	{"id": UUID, "username": str}
/auth/login	POST	User login to retrieve signed JWT access token.	OAuth2 password form (username, password)	{"access_token": str, "token_type": "bearer"}
/extract-invoice	POST	Upload file, create database job, queue worker task (JWT Protected).	Multipart form file (image/png or jpeg)	{"job_id": UUID, "status": "QUEUED"}
/jobs/{job_id}	GET	Job polling endpoint (JWT Protected).	Path Parameter: job_id (UUID)	Complete Job status + nested Invoice JSON
/query	POST	Retrieves matching context segments from FAISS (JWT Protected).	{"query": str, "top_k": int}	{"query": str, "results": list}
/query/ask	POST	QA assistant endpoint over invoices using Groq (JWT Protected).	{"question": str, "top_k": int}	{"answer": str, "sources": list}
/invoices	GET	Paginated metadata list of all processed invoices (JWT Protected).	Query Params: skip (default 0), limit (default 100)	List of Invoice metadata summaries
/invoices/{invoice_uuid}	GET	Detailed metadata and line items of a specific invoice (JWT Protected).	Path Parameter: invoice_uuid (UUID)	Complete Invoice + Line Items JSON
/invoices/review	GET	Lists all invoices requiring manual review (JWT Protected).	Query Params: skip (default 0), limit (default 100)	List of low-confidence Invoice Response objects
/invoices/{invoice_uuid}/correct	POST	Submits manual reviewer corrections for fields/line-items (JWT Protected).	Path Parameter: invoice_uuid, Correction JSON payload	Updated Invoice Response object
/metrics	GET	Resources and application index counters (Prometheus registry).	None	CPU, Memory, Disk, and telemetry metrics
/rag/status	GET	Total chunks and document metrics inside FAISS (JWT Protected).	None	FAISS total index segment metrics
/jobs/cleanup	POST	Housekeeping task enforcing retention policies (JWT Protected).	Query Parameter: retention_days (default 7)	{"message": str}

---

⚡ Quick Start

1. Prerequisites

Ensure you have Python 3.10+ installed on your system.

2. Set Up Virtual Environment

It is highly recommended to use a virtual environment.

On Windows (PowerShell):

# Create virtual environment
python -m venv .venv

# Activate virtual environment
.venv\Scripts\Activate.ps1

On macOS/Linux:

# Create virtual environment
python3 -m venv .venv

# Activate virtual environment
source .venv/bin/activate

3. Install Dependencies

pip install -r requirements.txt

4. Run the Development Server

Launch the application with live reload enabled:

uvicorn app.main:app --reload

The application will be running at http://127.0.0.1:8000. Swagger API docs are accessible at http://127.0.0.1:8000/docs.

5. Run the Test Suite

The backend features an isolated, mocked testing suite. Execute tests with:

python -m pytest -v

---

📁 Project Structure

invoice_ai_backend/
├── .github/
│   └── workflows/
│       └── ci.yml                 # GitHub Actions CI workflow (Ruff + Pytest + Docker Build)
├── app/                           # Core application logic
│   ├── auth.py                    # JWT cryptography & password hashing dependencies
│   ├── auth_routes.py             # User Register / Login endpoints
│   ├── celery_app.py              # Celery config instance
│   ├── config.py                  # Pydantic Settings layer (dotenv files loader)
│   ├── crud.py                    # SQLAlchemy database CRUD operations
│   ├── database.py                # Postgres connection pool and session maker
│   ├── inference.py               # Invoice extraction coordinator (OCR + LayoutLMv3)
│   ├── llm_engine.py              # Groq prompt formatting & QA scaffolding
│   ├── logging_config.py          # Loguru JSON / colorized structured logging
│   ├── main.py                    # FastAPI entrypoint, HTTP middleware, routes
│   ├── metrics_manager.py         # Telemetry manager exposing app & system counters
│   ├── models.py                  # SQLAlchemy relational tables
│   ├── ocr_engine.py              # PaddleOCR image text extraction
│   ├── rag_engine.py              # Embedding generation and retrieval router
│   ├── rate_limiter.py            # Redis-based sliding window rate limiter
│   ├── reranker.py                # Cross-Encoder candidate reranker integration
│   ├── review_routes.py           # Human-in-the-loop review & manual correction router
│   ├── schemas.py                 # Pydantic schemas
│   ├── table_engine.py            # Line-item coordinate parsing
│   ├── tasks.py                   # Celery tasks (idempotency, transient/exp retry)
│   ├── utils.py                   # Heuristic regex helper methods
│   └── vector_db.py               # FAISS vector store wrapper
├── docker/
│   ├── docker-compose.yml         # Dev/Production multi-container composition
│   ├── prometheus.yml             # Prometheus scraping configurations
│   └── grafana/                   # Grafana dashboards configuration volume
├── model/
│   └── layoutlmv3_production_model/ # Fine-tuned LayoutLMv3 weights & configurations
├── outputs/                       # Persisted FAISS binary index & document JSON
├── scratch/                       # Diagnostic and development scripts
├── scripts/
│   └── evaluate.py                # Pipeline evaluation script (Precision, Recall, F1)
└── tests/                         # Pytest suite with import-time ML mocks

---

🚀 Production Capabilities

⚡ 1. Asynchronous Job Architecture (Celery & Redis)

Ingestion workloads are decoupled from the API gateway using Celery and Redis. File uploads register a tracker state in PostgreSQL and push extraction to background workers, immediately returning a 202 Accepted status with a job_id. Workers feature exponential retry backoff (5s $\rightarrow$ 15s $\rightarrow$ 45s) for transient failures, and a clean cron housekeeping scheduler removes stale uploads older than a configured day threshold.

🔍 2. Hybrid RAG, Query Router & Cross-Encoder Reranker

Retrieval utilizes a hybrid search system:

• Segmented Chunking: Invoices are parsed into semantic regions (header, seller, buyer, totals, line_items).
• Query Router: Parses natural language expressions for query filters (e.g. "from Amazon", "above $100", "last month") and executes metadata filtering on PostgreSQL first.
• RRF & Reranking: Combines keyword search (Okapi BM25) and dense similarity matches (FAISS FlatIP) using Reciprocal Rank Fusion, then re-scores the merged candidates with cross-encoder/ms-marco-MiniLM-L-6-v2 (lightweight ~80MB, CPU-optimized) to output the top 5 relevant document nodes.

🔒 3. JWT Access Controls & Redis Rate Limiter

Endpoints are guarded by JWT tokens (signed with HMAC-SHA256, passwords hashed via Bcrypt). API resources are protected against denial of service through a Redis sliding-window rate limiter enforcing a 100 requests/minute limit per user using Redis Sorted Sets (zsets), with path exemptions for monitoring hooks.

📊 4. Telemetry & Observability (Prometheus & Grafana)

Application execution phases (OCR, LayoutLMv3, post-process, tables, FAISS, embeddings) are instrumented using a unified metrics_manager.py telemetry layer. Performance metrics and host resource states are scraped by Prometheus every 5 seconds and visualized on dedicated Grafana dashboards using Docker container compositions backed by persistent volume mounts.

🧪 5. Dynamic SQLite Mock Testing & CI/CD

Our automated test suite features:

• Temporary file-based SQLite database session overlays to isolate integration tests from PostgreSQL database files.
• Lazy loading ML model wrappers and mock overrides to bypass heavy GPU/CPU model loading and online weights download attempts during testing.
• Continuous Integration (CI): Integrated ruff linting, ruff formatting, pytest runs, and Docker multi-stage CPU builds inside a GitHub Actions pipeline.

👥 6. Explainability & Human-in-the-Loop (HITL) Workflow

LayoutLMv3 outputs are mapped to spatial bounding boxes [x1, y1, x2, y2] to explain extraction coordinates. Invoices producing an average field confidence below 0.70 are flagged with "NEEDS_REVIEW" status. Humans can fetch these invoices via GET /invoices/review and post manual field/line-item corrections via POST /invoices/{uuid}/correct, which updates the relational tables, marks status as "COMPLETED", and exports a training JSON pair mapping the original image, OCR text, and corrected fields to corrected_data/ for future model retraining.