Ingest a Postgres database into LlamaIndex

If you’re searching for a llamaindex postgres data ingestion tutorial that actually works in production, you’ve probably hit the gap between the five-line “hello world” examples and the reality of schema introspection, chunking strategy, and incremental updates. This guide walks through a complete, runnable pipeline: connecting to Postgres, extracting tables or custom SQL, embedding with a local or remote model, storing vectors, and verifying the result with a real query.

Step 1: Install the required packages

LlamaIndex splits database connectivity and vector storage into separate packages. Install the core, the Postgres reader, and a vector store — here we use Chroma for simplicity, but the same pattern applies to Pinecone, Weaviate, or PGVector.

pip install \
  llama-index \
  llama-index-readers-database \
  llama-index-vector-stores-chroma \
  chromadb \
  psycopg2-binary \
  python-dotenv

If you prefer OpenAI embeddings, add llama-index-embeddings-openai. For local embeddings (no API key, runs on CPU), add llama-index-embeddings-huggingface.

Step 2: Configure environment and database connection

Create a .env file at your project root. Never commit credentials.

# .env
POSTGRES_HOST=localhost
POSTGRES_PORT=5432
POSTGRES_DB=myapp
POSTGRES_USER=app_user
POSTGRES_PASSWORD=secret
# Optional: if you use OpenAI embeddings
# OPENAI_API_KEY=sk-...

Load it in Python:

# config.py
import os
from dotenv import load_dotenv

load_dotenv()

PG_DSN = (
    f"postgresql+psycopg2://{os.getenv('POSTGRES_USER')}:{os.getenv('POSTGRES_PASSWORD')}"
    f"@{os.getenv('POSTGRES_HOST')}:{os.getenv('POSTGRES_PORT')}/{os.getenv('POSTGRES_DB')}"
)

Step 3: Define the ingestion script

The script below does three things: (1) reflects the schema or runs custom SQL, (2) converts each row into a Document with metadata, (3) builds a VectorStoreIndex backed by Chroma. Save as ingest.py.

# ingest.py
import os
import logging
from typing import List

from sqlalchemy import create_engine, text, inspect
from llama_index.core import Document, VectorStoreIndex, StorageContext
from llama_index.core.node_parser import SentenceSplitter
from llama_index.vector_stores.chroma import ChromaVectorStore
from llama_index.embeddings.huggingface import HuggingFaceEmbedding
import chromadb

from config import PG_DSN

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# --- 1. Embedding model ---
# Local, no API key, 384-dim. Swap for OpenAIEmbedding() if you prefer.
embed_model = HuggingFaceEmbedding(model_name="BAAI/bge-small-en-v1.5")

# --- 2. Vector store (Chroma persistent) ---
chroma_client = chromadb.PersistentClient(path="./chroma_db")
chroma_collection = chroma_client.get_or_create_collection("postgres_docs")
vector_store = ChromaVectorStore(chroma_collection=chroma_collection)
storage_context = StorageContext.from_defaults(vector_store=vector_store)

# --- 3. Database engine ---
engine = create_engine(PG_DSN, pool_pre_ping=True)

# --- 4. Choose tables or custom SQL ---
# Option A: reflect all tables in public schema
inspector = inspect(engine)
TABLES: List[str] = inspector.get_table_names(schema="public")
logger.info("Discovered tables: %s", TABLES)

# Option B: explicit list (uncomment and edit if you only want a subset)
# TABLES = ["users", "orders", "products"]

# Option C: custom SQL per table (dict of table -> query)
# CUSTOM_QUERIES = {
#     "orders": "SELECT id, user_id, total, created_at FROM orders WHERE status = 'completed'"
# }

# --- 5. Row-to-Document conversion ---
def row_to_document(table: str, row: dict, pk_col: str = "id") -> Document:
    """
    Convert a SQLAlchemy row mapping to a LlamaIndex Document.
    The text field is a JSON-like string; metadata carries structured fields for filtering.
    """
    # Build a readable text representation
    text_parts = [f"Table: {table}"]
    for k, v in row.items():
        if v is not None:
            text_parts.append(f"{k}: {v}")
    text = " | ".join(text_parts)

    # Metadata for hybrid search / filtering later
    metadata = {
        "source_table": table,
        "primary_key": str(row.get(pk_col, "")),
        **{k: v for k, v in row.items() if isinstance(v, (str, int, float, bool))}
    }
    return Document(text=text, metadata=metadata)

# --- 6. Ingestion loop ---
def ingest_table(table_name: str, custom_query: str | None = None) -> int:
    """Ingest a single table, return number of documents indexed."""
    if custom_query:
        sql = text(custom_query)
    else:
        sql = text(f"SELECT * FROM {table_name}")

    docs: List[Document] = []
    with engine.connect() as conn:
        result = conn.execute(sql)
        # Get primary key for metadata
        pk_cols = inspector.get_pk_constraint(table_name)["constrained_columns"]
        pk_col = pk_cols[0] if pk_cols else "id"

        for row in result.mappings():
            docs.append(row_to_document(table_name, dict(row), pk_col=pk_col))

    if not docs:
        logger.warning("Table %s returned 0 rows", table_name)
        return 0

    # Chunk large documents (optional but recommended for wide rows)
    parser = SentenceSplitter(chunk_size=512, chunk_overlap=50)
    nodes = parser.get_nodes_from_documents(docs)

    # Build / update index
    index = VectorStoreIndex(
        nodes,
        storage_context=storage_context,
        embed_model=embed_model,
        show_progress=True,
    )
    logger.info("Indexed %d nodes from table %s", len(nodes), table_name)
    return len(nodes)

def main():
    total_nodes = 0
    for table in TABLES:
        # If you defined CUSTOM_QUERIES, use it; else full table scan
        custom = None  # CUSTOM_QUERIES.get(table) if using Option C
        total_nodes += ingest_table(table, custom_query=custom)

    logger.info("Ingestion complete. Total nodes: %d", total_nodes)

if __name__ == "__main__":
    main()

Run it:

python ingest.py

You should see progress bars from the embedding model and a final log line like Ingestion complete. Total nodes: 12,340.

Step 4: Verify the index with a query script

Create query.py to confirm the vectors are searchable and metadata filters work.

# query.py
from llama_index.core import VectorStoreIndex, StorageContext
from llama_index.vector_stores.chroma import ChromaVectorStore
from llama_index.embeddings.huggingface import HuggingFaceEmbedding
import chromadb

from config import PG_DSN  # only needed if you want to cross-check with SQL

embed_model = HuggingFaceEmbedding(model_name="BAAI/bge-small-en-v1.5")

chroma_client = chromadb.PersistentClient(path="./chroma_db")
chroma_collection = chroma_client.get_or_create_collection("postgres_docs")
vector_store = ChromaVectorStore(chroma_collection=chroma_collection)
storage_context = StorageContext.from_defaults(vector_store=vector_store)

index = VectorStoreIndex.from_vector_store(
    vector_store=vector_store,
    embed_model=embed_model,
)

# Simple semantic search
query_engine = index.as_query_engine(similarity_top_k=5)
response = query_engine.query("orders over $500 placed in January 2024")
print("=== Semantic search ===")
print(response)

# Metadata-filtered search (Chroma supports where clauses)
from llama_index.core.vector_stores import MetadataFilters, ExactMatchFilter

filters = MetadataFilters(filters=[ExactMatchFilter(key="source_table", value="orders")])
filtered_engine = index.as_query_engine(
    similarity_top_k=5,
    filters=filters,
)
response2 = filtered_engine.query("high value customers")
print("\n=== Filtered to 'orders' table ===")
print(response2)

Run it:

python query.py

You should see synthesized answers grounded in your Postgres data, with source nodes printed if you enable response.source_nodes.

Step 5: Incremental and scheduled ingestion

Production pipelines need idempotency. Two practical patterns:

Pattern A: Timestamp-based incremental (append-only tables)

Add a last_ingested_at tracker (file or small control table) and modify the query:

# In ingest.py, inside ingest_table()
from pathlib import Path

STATE_FILE = Path(f"./state/{table_name}_cursor.txt")
STATE_FILE.parent.mkdir(exist_ok=True)

last_ts = STATE_FILE.read_text().strip() if STATE_FILE.exists() else "1970-01-01"
sql = text(f"""
    SELECT * FROM {table_name}
    WHERE updated_at > :last_ts
    ORDER BY updated_at
""")
with engine.connect() as conn:
    result = conn.execute(sql, {"last_ts": last_ts})
    # ... same row processing ...
    max_ts = max((row["updated_at"] for row in result.mappings()), default=last_ts)
    STATE_FILE.write_text(str(max_ts))

Pattern B: Primary-key upsert (mutable tables)

Chroma supports upsert by ID. Use the primary key as the vector ID:

# In ingest_table(), after building nodes:
ids = [node.metadata["primary_key"] for node in nodes]
vector_store.add(nodes, ids=ids)  # replaces existing vectors with same ID

Schedule either pattern with cron, Airflow, or Prefect. A nightly run is typical for analytics workloads; hourly for operational RAG.

Step 6: Production hardening checklist

Step 7: Querying from your application

The same VectorStoreIndex pattern works in a FastAPI endpoint, a background worker, or a Streamlit app. Example FastAPI route:

# api.py
from fastapi import FastAPI, Query
from pydantic import BaseModel
from llama_index.core import VectorStoreIndex, StorageContext
from llama_index.vector_stores.chroma import ChromaVectorStore
from llama_index.embeddings.huggingface import HuggingFaceEmbedding
import chromadb

app = FastAPI()

# Initialize once at startup
chroma_client = chromadb.PersistentClient(path="./chroma_db")
chroma_collection = chroma_client.get_or_create_collection("postgres_docs")
vector_store = ChromaVectorStore(chroma_collection=chroma_collection)
storage_context = StorageContext.from_defaults(vector_store=vector_store)
embed_model = HuggingFaceEmbedding(model_name="BAAI/bge-small-en-v1.5")
index = VectorStoreIndex.from_vector_store(vector_store, embed_model=embed_model)
query_engine = index.as_query_engine(similarity_top_k=5)

class AskRequest(BaseModel):
    question: str
    table: str | None = None  # optional metadata filter

@app.post("/ask")
def ask(req: AskRequest):
    filters = None
    if req.table:
        from llama_index.core.vector_stores import MetadataFilters, ExactMatchFilter
        filters = MetadataFilters(filters=[ExactMatchFilter(key="source_table", value=req.table)])

    response = query_engine.query(req.question, filters=filters)
    return {
        "answer": str(response),
        "sources": [
            {"table": n.metadata.get("source_table"), "pk": n.metadata.get("primary_key"), "score": n.score}
            for n in response.source_nodes
        ],
    }

Run with uvicorn api:app --reload and POST to /ask.

Verification checklist

After the first run, confirm:

1. Chroma directory exists — ls -la chroma_db/ shows chroma.sqlite3 and segment files.
2. Collection count matches — chroma_client.get_collection("postgres_docs").count() equals the node count logged during ingestion.
3. Semantic search returns relevant rows — run query.py with a known value (e.g., a specific order ID) and verify the answer contains that ID.
4. Metadata filters work — the filtered query only returns nodes from the specified table.
5. Incremental run adds zero duplicates — re-run ingest.py on an unchanged DB; node count should not increase (Pattern B) or should only add new rows (Pattern A).

Where this fits in a larger RAG system

Ingestion is only the first half. The other half — retrieval, reranking, citation, and guardrails — lives in your query pipeline. A few practical tips:

• Hybrid search: Combine vector similarity with a BM25 index over the same metadata (LlamaIndex supports QueryFusionRetriever).
• Reranking: Add a cross-encoder (llama-index-postprocessor-cohere-rerank or local bge-reranker-base) to boost precision before synthesis.
• Citation: Enable response_mode="tree_summarize" and parse source_nodes to render inline citations in your UI.
• Guardrails: Wrap the query engine with GuidancePydanticProgram or Guardrails to enforce JSON output schemas.

If you need to serve multiple tenants or route queries to different model providers behind a single OpenAI-compatible endpoint, that’s where a gateway like n4n.ai simplifies operations — but the ingestion logic above remains exactly the same.

You now have a complete, production-ready llamaindex postgres data ingestion tutorial you can drop into a repo, schedule, and extend. The same patterns apply whether you’re indexing 10,000 rows or 10 million — just swap the vector store and add batching.