FastAPI + PostgreSQL + FalkorDB: The Enterprise AI Stack

TL;DR — The FastAPI + PostgreSQL + FalkorDB stack is the open-source enterprise AI architecture for self-hosted RAG. FastAPI provides async API orchestration. PostgreSQL with pgvector handles vector search, BM25 full-text search, metadata, ACLs, and audit logging. FalkorDB provides the knowledge graph for multi-hop reasoning and entity-relationship traversal. Together: keyword (BM25) + semantic (vector) + relational (graph) retrieval in one stack. FalkorDB is Redis-compatible, uses sparse adjacency matrices with GraphBLAS, supports multi-tenancy natively, and is lighter than Neo4j. Docker Compose deploys all three services in minutes. Production-ready with ACID transactions, row-level security, and async request handling. Add Prometheus + Grafana for monitoring, Nango for connectors, and BGE-Reranker for cross-encoder reranking. Handles 1000+ concurrent users. No per-user costs, no vendor lock-in, full data sovereignty.

The FalkorDB GraphRAG SDK demonstrated that "a well-architected retrieval and reasoning layer consistently outperforms a weak pipeline running a more expensive model." The stack matters more than the model.

This post describes the production architecture we use at AI App Lab: FastAPI for API orchestration, PostgreSQL with pgvector for vector and keyword search, and FalkorDB for knowledge graph traversal. Three open-source components covering all three retrieval modalities in a single self-hosted deployment.

Why This Stack

Architecture

Docker Compose Setup

version: "3.9"
services:
  # API Layer
  api:
    build: .
    ports: ["8000:8000"]
    environment:
      DATABASE_URL: postgresql+asyncpg://postgres:${DB_PASSWORD}@postgres/aidb
      FALKORDB_URL: redis://falkordb:6379
      OLLAMA_URL: http://ollama:11434
    depends_on:
      postgres:
        condition: service_healthy
      falkordb:
        condition: service_started
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
      interval: 30s
      timeout: 5s
      retries: 3

  # Data Layer: PostgreSQL + pgvector
  postgres:
    image: pgvector/pgvector:pg16
    ports: ["5432:5432"]
    environment:
      POSTGRES_DB: aidb
      POSTGRES_PASSWORD: ${DB_PASSWORD}
    volumes:
      - pg_data:/var/lib/postgresql/data
      - ./init.sql:/docker-entrypoint-initdb.d/init.sql
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U postgres"]
      interval: 10s
      timeout: 5s
      retries: 5

  # Knowledge Graph: FalkorDB
  falkordb:
    image: falkordb/falkordb:latest
    ports: ["6379:6379"]
    volumes:
      - falkor_data:/data
    command: ["--save", "60", "1", "--loglevel", "warning"]

  # LLM Inference: Ollama
  ollama:
    image: ollama/ollama:latest
    ports: ["11434:11434"]
    volumes:
      - ollama_data:/root/.ollama
    deploy:
      resources:
        reservations:
          devices:
            - driver: nvidia
              count: 1
              capabilities: [gpu]

  # Monitoring
  prometheus:
    image: prom/prometheus:latest
    ports: ["9090:9090"]
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml

volumes:
  pg_data:
  falkor_data:
  ollama_data:

FastAPI Application Structure

from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel
from contextlib import asynccontextmanager
import asyncpg
import redis.asyncio as redis

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Initialize connections
    app.state.pg = await asyncpg.create_pool(
        os.environ["DATABASE_URL"].replace("+asyncpg", ""),
        min_size=5, max_size=20
    )
    app.state.falkor = redis.from_url(os.environ["FALKORDB_URL"])
    app.state.llm = OllamaClient(os.environ["OLLAMA_URL"])
    yield
    # Cleanup
    await app.state.pg.close()
    await app.state.falkor.close()

app = FastAPI(lifespan=lifespan, title="Enterprise AI API")

class QueryRequest(BaseModel):
    query: str
    user_id: str
    conversation_id: str | None = None

class QueryResponse(BaseModel):
    answer: str
    sources: list[dict]
    confidence: float
    cached: bool

@app.post("/query", response_model=QueryResponse)
async def query(req: QueryRequest):
    # 1. Get user ACLs
    user_acls = await get_user_acls(app.state.pg, req.user_id)

    # 2. Compute query embedding
    embedding = await compute_embedding(req.query)

    # 3. Parallel retrieval: PostgreSQL + FalkorDB
    pg_results, graph_results = await asyncio.gather(
        hybrid_search(app.state.pg, req.query, embedding, user_acls),
        graph_search(app.state.falkor, req.query, user_acls),
    )

    # 4. RRF fusion
    fused = rrf_fuse(pg_results, graph_results)

    # 5. Cross-encoder reranking
    reranked = await rerank(req.query, fused[:50])

    # 6. Generate with citations
    answer = await app.state.llm.generate(
        query=req.query,
        context=format_context(reranked[:10]),
    )

    # 7. Audit log
    await audit_log(app.state.pg, req.user_id, req.query, answer)

    return QueryResponse(
        answer=answer,
        sources=reranked[:5],
        confidence=reranked[0]["rerank_score"] if reranked else 0,
        cached=False,
    )

PostgreSQL Schema

-- Enable pgvector
CREATE EXTENSION IF NOT EXISTS vector;

-- Documents table with vector + full-text search
CREATE TABLE documents (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    title TEXT NOT NULL,
    content TEXT NOT NULL,
    source TEXT NOT NULL,
    source_url TEXT,
    acl TEXT[] NOT NULL DEFAULT '{}',
    content_hash TEXT UNIQUE NOT NULL,
    embedding VECTOR(1536),
    tsv TSVECTOR GENERATED ALWAYS AS (
        to_tsvector('english', content)
    ) STORED,
    created_at TIMESTAMPTZ DEFAULT NOW(),
    updated_at TIMESTAMPTZ DEFAULT NOW()
);

-- HNSW index for vector search
CREATE INDEX idx_documents_embedding
    ON documents USING hnsw (embedding vector_cosine_ops)
    WITH (m = 16, ef_construction = 64);

-- GIN index for full-text search
CREATE INDEX idx_documents_tsv ON documents USING gin(tsv);

-- GIN index for ACL array filtering
CREATE INDEX idx_documents_acl ON documents USING gin(acl);

-- Audit log table
CREATE TABLE audit_log (
    id BIGSERIAL PRIMARY KEY,
    user_id TEXT NOT NULL,
    query TEXT NOT NULL,
    answer TEXT,
    sources JSONB,
    latency_ms INTEGER,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

-- Row-level security for ACLs
ALTER TABLE documents ENABLE ROW LEVEL SECURITY;
CREATE POLICY acl_policy ON documents
    USING (acl && current_user_acl());

FalkorDB Graph Schema

# Create knowledge graph in FalkorDB
# Entities: Person, Organization, Policy, Document, Department
# Relationships: REPORTS_TO, BELONGS_TO, REFERENCES, AUTHORED_BY

# Cypher query for multi-hop reasoning
GRAPH_QUERY = """
MATCH (p:Person {name: $person_name})
-[:BELONGS_TO]->(d:Department)
-[:HAS_POLICY]->(pol:Policy)
WHERE pol.effective_date <= date() AND
      (pol.expiry_date IS NULL OR pol.expiry_date >= date())
RETURN pol.title, pol.content, pol.effective_date
ORDER BY pol.effective_date DESC
"""

async def graph_search(falkor_client, query: str, user_acls: list):
    """Search knowledge graph for entity-relationship context."""
    # Extract entities from query using LLM
    entities = await extract_entities(query)

    results = []
    for entity in entities:
        # Query FalkorDB for relationships
        graph_result = await falkor_client.execute_command(
            "GRAPH.QUERY", "knowledge_graph",
            GRAPH_QUERY,
            {"person_name": entity}
        )
        results.extend(parse_graph_result(graph_result))

    return results

Component Comparison

FastAPI PostgreSQL FalkorDB Stack Checklist

• [ ] Set up Docker Compose with FastAPI, PostgreSQL, FalkorDB, and Ollama
• [ ] Enable pgvector extension and create HNSW index on documents
• [ ] Create tsvector column for BM25 full-text search
• [ ] Enable row-level security for document-level ACLs
• [ ] Create audit log table for EU AI Act compliance
• [ ] Initialize FalkorDB knowledge graph with entity and relationship schema
• [ ] Implement hybrid retrieval in PostgreSQL (BM25 + vector)
• [ ] Implement graph traversal in FalkorDB for multi-hop reasoning
• [ ] Use asyncio.gather for parallel PostgreSQL + FalkorDB queries
• [ ] Apply RRF fusion to combine results
• [ ] Add cross-encoder reranking with BGE-Reranker
• [ ] Implement citation tagging in LLM generation
• [ ] Set up semantic answer caching for cost reduction
• [ ] Add content hash deduplication in ingestion
• [ ] Configure Prometheus metrics for latency, cache hit rate, error rate
• [ ] Set up health checks for all services
• [ ] Configure volume persistence for PostgreSQL, FalkorDB, and Ollama
• [ ] Use asyncpg connection pool (min=5, max=20 for medium scale)
• [ ] Implement hallucination prevention: confidence threshold
• [ ] Use query rewriting for complex queries
• [ ] Integrate with self-hosted AI architecture
• [ ] Test with AI company brain use cases
• [ ] Consider temporal knowledge graphs for versioned facts
• [ ] Configure backups: PostgreSQL PITR + FalkorDB RDB snapshots
• [ ] Set up alerting: latency p99 > 500ms, error rate > 1%, cache hit < 20%

FAQ

What is the FastAPI PostgreSQL FalkorDB stack?

The FastAPI + PostgreSQL + FalkorDB stack is an open-source enterprise AI architecture for self-hosted RAG. FastAPI provides the async API layer for query handling and orchestration. PostgreSQL with pgvector handles vector search, full-text search (BM25 via tsvector), metadata, ACLs, and audit logging. FalkorDB provides the knowledge graph for multi-hop reasoning and entity-relationship traversal. Together, they cover all three retrieval modalities: keyword (BM25), semantic (vector), and relational (graph) — in a single self-hosted deployment with no per-user costs.

Why use FalkorDB instead of Neo4j?

FalkorDB is lighter, faster for graph traversal workloads, and Redis-compatible (uses the RESP protocol). It uses sparse adjacency matrices with GraphBLAS for efficient storage and AVX acceleration for performance. FalkorDB supports multi-tenancy natively (multiple graphs in one instance with zero overhead) and is designed for horizontal scalability. Neo4j is more feature-rich (ACID transactions, larger ecosystem) but heavier and more expensive at enterprise scale. For RAG and knowledge graph use cases, FalkorDB's performance and simplicity make it the better choice for self-hosted deployments.

How does FastAPI integrate with PostgreSQL and FalkorDB?

FastAPI serves as the orchestration layer. It receives user queries, computes embeddings, runs hybrid retrieval against PostgreSQL (BM25 + pgvector), queries FalkorDB for graph-based relationships, fuses results with RRF, calls the LLM for generation, and returns source-cited answers. FastAPI's async nature allows parallel queries to PostgreSQL and FalkorDB. The stack uses asyncpg for PostgreSQL connections and redis-py for FalkorDB connections. Docker Compose orchestrates all three services with health checks and volume persistence.

Can I use this stack for production enterprise AI?

Yes. The FastAPI + PostgreSQL + FalkorDB stack is production-ready for enterprise AI. PostgreSQL provides ACID transactions, row-level security for ACLs, and point-in-time recovery. FalkorDB provides fast graph traversal with multi-tenancy support. FastAPI provides async request handling, automatic OpenAPI documentation, and type safety with Pydantic. Add Prometheus + Grafana for monitoring, Nango for source connectors, and BGE-Reranker for cross-encoder reranking for a complete production deployment. The stack handles 1000+ concurrent users with proper configuration.

What are the alternatives to this stack?

Alternatives: Replace FastAPI with Django or Flask (less async, more batteries-included). Replace PostgreSQL with OpenSearch or Elasticsearch (better full-text search, no ACID). Replace FalkorDB with Neo4j (more features, larger ecosystem, higher cost) or Apache AGE (PostgreSQL extension for graphs). Replace the entire stack with LangChain or LlamaIndex (higher abstraction, less control). The FastAPI + PostgreSQL + FalkorDB stack is optimized for: self-hosted deployment, maximum control, ACID compliance, and multi-modal retrieval (keyword + vector + graph) in a single infrastructure.

Want a self-hosted AI company brain that does all of this out of the box?
Book a demo →