This will be a two part spectacular, in this part we will cover the core files of the application:
├── seed_db.py → p2p.db seed realistic data + anomalies├── schema_agent.py → schema_context.json semantic layer via Claude├── rag_pipeline.py → interactive Q&A hybrid RAG (SQL + ChromaDB)├── app.py → Streamlit UI chat interface
In the second article we will cover accessory and helper functions:
├── context_packer.py → LLM-ready string token-budget context assembly├── anomaly_agent.py → anomaly_report.json 8 SQL integrity rules├── cache_report.py → logs/cache_usage.log token + cost tracking└── logger.py → logs/<script>.log shared logging
Imagine a small business where invoices and POs (Purchase Orders) live across various Excel sheets, there is no documentation, cryptic column names, and unclear relationships between tabs. Now scale that problem to an enterprise P2P (Purchase-to-Pay) database! It’s the same chaos, but now across:
This disorganized database is huge liability with real financial risk attached.
In this tutorial, I’ll walk through how to build a pipeline that takes an undocumented P2P SQLite database (meaning, there is no explanation on what anything means or how tables connects) and transform it into a human-ready RAG application.
Final product will be a Streamlit App, gif below:
Before we move on, let me briefly touch on this. Schema archaeology matters because most real enterprise databases were built for transactions, not for AI — and GenAI can’t safely reason over data it doesn’t understand.
Schema archaeology turns an undocumented database into something an AI can reason about — without it, you’re asking GenAI to do finance analysis on a system nobody documented, which is where hallucinations and wrong joins start.
Here is a real world example:
“Why was invoice INV-2291 flagged?”
Without schema archaeology, the model might not know that “flagged” could mean missing receipt, duplicate number, or GL imbalance — or which tables to join to find out.
With it, the system knows the P2P workflow, relationship map, and domain terms — so retrieval and synthesis target the right story!
This is the application stack, first we will create our database with data to simulate realistic, disorganized, anomalous data. Then an agent will crawl through the database and learn the PK/FK relationships and data. Finally, a RAG engine will be used so a human can query the data.
Not a raw SQL row — a joined invoice summary (vendor + PO + receipt status + GL). This dramatically improves retrieval precision for AP (Accounts Payable) questions that span multiple tables.
Pure semantic search has high recall but low precision for structured AP queries. SQL pre-filtering narrows the candidate set; vector search ranks by relevance within that set.
The schema prompt and semantic layer are sent as cached system messages (cache_control: ephemeral). The evaluation call reads from cache at 90% cost reduction.
ChromaDB cosine distance > 1.2 signals weak retrieval. Claude is explicitly instructed to surface uncertainty rather than hallucinate — a confident wrong answer is worse than “I don’t know” in financial queries.
Claude scores its own annotation 0–100 and flags accuracy risks. The score is stored in schema_context.json so downstream agents can weight confidence accordingly.
First we have to create our data. Run seed_db.py Creates p2p.db with realistic P2P data and intentional financial anomalies:
code snippet below:
import sqlite3, random, string, datetime, os
DB_PATH = "p2p.db"
CATEGORIES = ["Software", "Hardware", "Consulting", "Logistics", "Office Supplies", "Facilities"]PAYMENT_TERMS = ["NET30", "NET60", "NET90", "NET15", "COD"]STATUSES_PO = ["draft", "approved", "received", "closed", "cancelled"]STATUSES_INV = ["pending", "approved", "paid", "disputed", "overdue"]# 5 account codes seededACCOUNT_CODES = [ ("2000", "AP Control", "liability"), ("5000", "COGS", "expense"), ("6000", "Opex Expense", "expense"), ("6100", "SaaS Expense", "expense"), ("1200", "Prepaid Assets","asset"),]
python
def rand_date(start_days_ago=365, end_days_ago=0): delta = random.randint(end_days_ago, start_days_ago) return (datetime.date.today() - datetime.timedelta(days=delta)).isoformat()
python
def rand_invoice_number(): return "INV-" + "".join(random.choices(string.digits, k=4))
python
def build():.....
Next step — schema_agent.py crawls the schema, infers business meaning via Claude. This script performs “schema archaeology” — given only a database connection, it reverse-engineers the business meaning of every table and column using an LLM, then saves the result as a structured JSON semantic layer.
👉The semantic layer is used downstream by:
rag_pipeline.py — to ground LLM answers in schema context
and
context_packer.py — to assemble invoice context for AI agents
How it works:
💰Both stage 1 and stage 2 outputs are cached to disk so they don’t need to be re-run on every execution.
Output
python3 schema_agent.py --db p2p.db --out schema_context.jsonINFO [schema_agent] 📂 cached crawl_schema from crawl_schema_cache.jsonINFO [schema_agent] 🔍 Crawling schema…INFO [schema_agent] Found 8 tables, 2 implicit FK relationshipsINFO [schema_agent] 💾 Saved crawl_schema output to crawl_schema_cache.jsonINFO [schema_agent] 📂 cached llm_annotate from llm_annotate_cache.jsonINFO [schema_agent] 🧠 Running llm_annotate via Claude…INFO [schema_agent] Annotated 8 tablesINFO [schema_agent] Found 9 relationshipsINFO [schema_agent] 💾 Saved llm_annotate output to llm_annotate_cache.jsonINFO [schema_agent] ✅ Running self-evaluation…INFO [schema_agent] Overall score: 92/100INFO [schema_agent] Coverage score 95/100WARNING [schema_agent] ⚠ The SKU-based implicit FK between po_line_items and receipt_lines is marked as 'medium confidence' - could cause incorrect joins if SKUs are not unique across vendorsWARNING [schema_agent] ⚠ GL account_code relationship is marked as not enforced (enforced: false), which could allow orphaned GL entries if account codes are deletedWARNING [schema_agent] ⚠ No explicit invoice line items table - assumes 1:1 invoice to PO which may not reflect real-world scenarios with partial invoicingWARNING [schema_agent] 🚩 No direct relationship between invoices and goods_receipts for 3-way match validation - requires traversal through purchase_ordersWARNING [schema_agent] 🚩 Missing relationship between gl_entries and payment transactions (no payments table exists in schema)WARNING [schema_agent] 🚩 No audit/approval workflow table linking to invoices or purchase_orders for authorization trackingWARNING [schema_agent] 🔧 Add payment_status or payment_date column to invoices table to track actual payment execution vs approval statusWARNING [schema_agent] 🔧 Include currency field in invoices and purchase_orders tables for multi-currency P2P supportWARNING [schema_agent] 🔧 Add invoice_line_items table to support partial invoicing scenarios common in enterprise P2PWARNING [schema_agent] 🔧 Consider adding a tax_amount column to invoices for VAT/GST compliance queriesWARNING [schema_agent] 🔧 Add created_at timestamp to invoices table for aging analysis beyond due_dateWARNING [schema_agent] 🔧 Include vendor.tax_id or vendor.bank_details as sensitive fields for payment processing contextWARNING [schema_agent] 🔧 Add query pattern for AP aging buckets (0-30, 31-60, 61-90, 90+ days)INFO [schema_agent] ✅ schema_context.json written → schema_context.jsonINFO [schema_agent] ==============================================================
How it’s used in RAG Schema agent is an offline prep step. RAG doesn’t run it during queries, but it depends on schema_context.json to ground answers in business meaning.
Enables querying over 5000+ invoices by combining two complementary retrieval strategies:**1. SQL pre-filtering **— narrows the candidate set using structured filters (invoice number, payment terms, status) extracted from the question. Fast, precise, zero false positives.**2. Semantic search **— ranks the filtered candidates by meaning using ChromaDB vector embeddings. Catches questions that SQL alone can’t answer (e.g. “which invoices look risky?”).
Together these form “hybrid search” — the SQL pass boosts precision,the semantic pass provides recall for unstructured questions.
The Flow:
How it works:
build_invoice_chunks()
Queries SQLite with a 5-table join (invoices, vendors, POs, goods receipts, GL). For each invoice, builds a summary chunk plus metadata. This is the “unit of knowledge” for embedding — not raw SQL rows.
sql_filter_invoice_ids()
Runs a SQL query that returns invoice_id values. This is used as the structured pre-filter before semantic search, to narrow ChromaDB retrieval to a candidate set of invoices.
get_or_build_collection()
Connects to ChromaDB on disk (./chroma_db). If the collection exists, loads it; otherwise creates it and embeds chunks in batches of 500 using the default MiniLM embedding function. Returns the ChromaDB collection.
P2PRAG class
P2PRAG.init()
Loads schema_context.json into:
self.schema_ctx -> creates the Anthropic client (with prompt-caching header) -> Calls build_invoice_chunks() and get_or_build_collection() to prepare retrieval
P2PRAG._detect_filter() parses the user question for structured filters and returns SQL if matched:
Invoice number (INV-1234),
Payment terms (NET60, NET30, etc.),
Invoice status (approved, pending, etc.).
Returns None if no filter is detected.
P2PRAG.query(question, n_results=8)
Main hybrid RAG flow:
Wow! If this seems like a lot, don’t worry, check out the next article for more information!
Sources:
Github: https://github.com/a-rhodes-vcu/schema_archaeology/tree/main
Schema Archaeology: How to Use AI to Reverse-Engineer Business Meaning From an Undocumented… was originally published in Towards AI on Medium, where people are continuing the conversation by highlighting and responding to this story.