AI SystemsSeptember 14, 2026 · 11 min read

Database Schema Design for AI Applications

Q: Should I use PostgreSQL or a specialized vector database?

Start with PostgreSQL + pgvector for <1M vectors - it's simpler, cheaper, and handles relational data alongside vectors. Switch to Pinecone/Weaviate/Qdrant when you need: (1) 1M vectors, (2) <10ms p99 latency, (3) advanced vector features (hybrid search, filtering), or (4) horizontal scaling. Most production RAG systems use both - Postgres for metadata/conversations, vector DB for embeddings.

Q: How do I handle embedding model changes (1536 dims → 3072 dims)?

Add versioned embedding columns: embeddingv1 vector(1536), embeddingv2 vector(3072). Backfill new embeddings in batches, update application to use new column, drop old column after migration complete. Or use separate tables: documentchunksv1, documentchunksv2 with routing logic. Never modify embedding dimension in place.

Q: What's the best way to store multi-turn conversations?

Messages table with sequencenumber (not timestamps for ordering). Use conversationid foreign key to group messages. For branching, add parentmessageid. Store system prompt in conversations table. Denormalize token counts for fast quota checking. See conversation history section for complete schema.

Q: How do I prevent vector index from slowing down writes?

Batch inserts (1000+ vectors at once), create index after initial bulk load, use IVFFlat for write-heavy workloads (HNSW is slower to build). For real-time systems, consider async indexing: write to table immediately, update vector index in background job. Or use separate "hot" table (no index, for recent vectors) and "cold" table (indexed, for older vectors).

Q: Should I store full document content in the same table as embeddings?

No - store content in object storage (S3) or separate table. Vector tables should be lean: chunkid, embedding, minimal metadata. Large content (10KB) hurts cache efficiency and index performance. Store contenturl or contenthash in vector table, fetch full text only when needed.

Q: How do I implement multi-tenant isolation in vector search?

Denormalize organizationid into vector table, add to WHERE clause: WHERE organizationid = $1 AND similarity 0.7. Index on (organizationid, similarity) for performance. Or use metadata filtering: metadata @ '{"organizationid": "org123"}'. Pinecone supports namespaces for tenant isolation. Never rely on application filtering alone - enforce at database level.

Q: What's the right chunk size for vector storage?

200-500 tokens (roughly 150-400 words) balances context and granularity. Smaller chunks (100 tokens) give precise matches but lose context. Larger chunks (1000+ tokens) retain context but dilute relevance. Test with your data - technical docs work well with 300 tokens, narrative content with 500+. See our RAG chunking guide for benchmarks.

Q: How do I handle schema changes in production without downtime?

Multi-step deployments: (1) Add new column (nullable), (2) Deploy code that writes to both old and new, (3) Backfill existing rows in batches, (4) Deploy code that reads from new column, (5) Drop old column. Use feature flags to control rollout. For critical tables, use blue-green database deployment or read replicas during migration.

Database Schema Design for AI Applications guidance for engineers: compare architecture choices, avoid failure modes, and ship a measurable, reliable.

Muhammad Abdul Sami

September 14, 2026 · 11 min read

PostgreSQL
Architecture
Performance
Data Pipelines

AI applications need specialized database schemas - vector embeddings, conversation history, and tool execution logs don't fit traditional normalized designs. This guide covers schema patterns from systems storing 10 million+ AI interactions.

What you'll learn:

Database schema for vector embeddings and RAG
Conversation history and multi-turn storage
Tool execution and agent state management
Performance optimization for AI workloads
Migration patterns for existing systems

Reading time: 17 minutes

Key Takeaways:

Treat Database Schema Design for AI Applications as a system with an explicit input and output contract.

Benchmark a representative baseline before choosing an optimization.

Bound retries, queues, concurrency, and total request deadlines.

Roll out through offline replay, shadow traffic, and a measurable canary.

Keep rollback simple and attach version identifiers to every decision.

Table of Contents:

Why AI Databases Are Different
Vector Storage Schema
Conversation History Design
Agent State Management
Document & Metadata Schema
Performance Optimization
Schema Migration Strategies
Production Examples
Frequently Asked Questions

Why AI Databases Are Different

Traditional RDBMS design principles don't apply to AI data - embeddings are high-dimensional vectors, conversations are tree structures, and agent executions are append-only event logs.

Key Differences

Traditional Apps	AI Applications
Normalized schemas	Denormalized for performance
Relational queries	Vector similarity search
ACID transactions	Eventually consistent
Fixed schema	Semi-structured JSON
KB-MB per row	MB per vector array
Index on columns	Index on vector distance

Real example: A typical RAG system stores:

Documents: Original text, metadata, chunking strategy
Chunks: Text segments with positional info
Embeddings: 1536-dimensional vectors (6KB each for OpenAI)
Conversations: Multi-turn dialogue with tool calls
Agent state: Execution logs, decisions, tool results

Traditional normalized design would require 8+ joins for a single RAG query. Production systems denormalize aggressively.

For our RAG system implementations, we design schemas optimized for AI workloads.

Vector Storage Schema

Vector embeddings are the core of modern AI systems - used for semantic search, RAG, and similarity matching.

PostgreSQL with pgvector

sql

-- Enable pgvector extension
CREATE EXTENSION IF NOT EXISTS vector;

-- Documents table
CREATE TABLE documents (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    title TEXT NOT NULL,
    content TEXT NOT NULL,
    mime_type VARCHAR(100),
    url TEXT,
    source VARCHAR(50), -- 'upload', 'crawl', 'api'
    user_id UUID REFERENCES users(id),
    metadata JSONB DEFAULT '{}',
    created_at TIMESTAMPTZ DEFAULT NOW(),
    updated_at TIMESTAMPTZ DEFAULT NOW(),
    
    -- Indexes
    INDEX idx_documents_user_id (user_id),
    INDEX idx_documents_source (source),
    INDEX idx_documents_created_at (created_at)
);

-- Add GiST index for metadata queries
CREATE INDEX idx_documents_metadata ON documents USING GIN (metadata);

-- Chunks table (denormalized for performance)
CREATE TABLE document_chunks (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    document_id UUID REFERENCES documents(id) ON DELETE CASCADE,
    
    -- Content
    content TEXT NOT NULL,
    chunk_index INTEGER NOT NULL,
    token_count INTEGER,
    
    -- Vector embedding (1536 dims for OpenAI text-embedding-3-small)
    embedding vector(1536),
    
    -- Metadata (denormalized from parent document)
    document_title TEXT,
    document_url TEXT,
    metadata JSONB DEFAULT '{}',
    
    -- Timestamps
    created_at TIMESTAMPTZ DEFAULT NOW(),
    
    -- Indexes
    INDEX idx_chunks_document_id (document_id),
    INDEX idx_chunks_chunk_index (document_id, chunk_index)
);

-- Vector similarity index (CRITICAL for performance)
CREATE INDEX idx_chunks_embedding ON document_chunks 
USING ivfflat (embedding vector_cosine_ops)
WITH (lists = 100);

-- For large datasets (>1M vectors), use HNSW
CREATE INDEX idx_chunks_embedding_hnsw ON document_chunks
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);

-- Vector search function
CREATE OR REPLACE FUNCTION search_chunks(
    query_embedding vector(1536),
    match_threshold float DEFAULT 0.7,
    match_count int DEFAULT 10,
    filter_metadata jsonb DEFAULT NULL
)
RETURNS TABLE (
    chunk_id UUID,
    document_id UUID,
    content TEXT,
    similarity float,
    metadata JSONB
) AS $$
BEGIN
    RETURN QUERY
    SELECT 
        dc.id,
        dc.document_id,
        dc.content,
        1 - (dc.embedding <=> query_embedding) AS similarity,
        dc.metadata
    FROM document_chunks dc
    WHERE 
        (filter_metadata IS NULL OR dc.metadata @> filter_metadata)
        AND (1 - (dc.embedding <=> query_embedding)) > match_threshold
    ORDER BY dc.embedding <=> query_embedding
    LIMIT match_count;
END;
$$ LANGUAGE plpgsql;

-- Usage
SELECT * FROM search_chunks(
    '[0.1, 0.2, ...]'::vector(1536),
    match_threshold := 0.7,
    match_count := 5,
    filter_metadata := '{"category": "documentation"}'::jsonb
);

Pinecone Schema (Vector-Native DB)

python

from pinecone import Pinecone, ServerlessSpec
pc = Pinecone(api_key="your-key")

# Create index
index = pc.create_index(
    name="ai-docs",
    dimension=1536,
    metric="cosine",
    spec=ServerlessSpec(
        cloud="aws",
        region="us-east-1"
    )
)

# Schema via metadata
vector_record = {
    "id": "chunk_123",
    "values": [0.1, 0.2, ...],  # 1536-dim embedding
    "metadata": {
        # Document info
        "document_id": "doc_abc",
        "document_title": "AI Architecture Guide",
        "document_url": "https://...",
        
        # Chunk info
        "chunk_index": 0,
        "content": "Full chunk text here...",
        "token_count": 250,
        
        # Search facets
        "category": "documentation",
        "tags": ["ai", "architecture", "rag"],
        "created_at": "2026-09-14T10:00:00Z",
        
        # User context
        "user_id": "user_xyz",
        "organization_id": "org_123"
    }
}

# Upsert
index.upsert(vectors=[vector_record])

# Query with metadata filtering
results = index.query(
    vector=[0.1, 0.2, ...],
    top_k=5,
    include_metadata=True,
    filter={
        "category": {"$eq": "documentation"},
        "user_id": {"$eq": "user_xyz"}
    }
)

Vector storage principles:

Denormalize metadata into vector records for filtering
Choose index type based on size (IVFFlat <1M, HNSW >1M)
Tune index parameters - more lists/m = better recall, slower writes
Separate embeddings from full content (store content in object storage for large docs)

See our RAG chunking strategies guide for chunk schema optimization.

Conversation History Design

Multi-turn conversations require careful schema design to support context windows, branching, and tool calls.

Conversation Schema

sql

-- Conversations (sessions)
CREATE TABLE conversations (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    user_id UUID REFERENCES users(id) ON DELETE CASCADE,
    title TEXT,
    
    -- Configuration
    model VARCHAR(50) DEFAULT 'gpt-4',
    system_prompt TEXT,
    temperature DECIMAL(3,2) DEFAULT 0.7,
    
    -- State
    message_count INTEGER DEFAULT 0,
    total_tokens INTEGER DEFAULT 0,
    total_cost DECIMAL(10,4) DEFAULT 0,
    
    -- Metadata
    metadata JSONB DEFAULT '{}',
    created_at TIMESTAMPTZ DEFAULT NOW(),
    updated_at TIMESTAMPTZ DEFAULT NOW(),
    archived_at TIMESTAMPTZ,
    
    INDEX idx_conversations_user_id (user_id),
    INDEX idx_conversations_created_at (created_at)
);

-- Messages (append-only)
CREATE TABLE messages (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    conversation_id UUID REFERENCES conversations(id) ON DELETE CASCADE,
    
    -- Message ordering
    sequence_number INTEGER NOT NULL,
    parent_message_id UUID REFERENCES messages(id), -- For branching
    
    -- Content
    role VARCHAR(20) NOT NULL CHECK (role IN ('system', 'user', 'assistant', 'tool')),
    content TEXT,
    
    -- Tool calls (for assistant messages)
    tool_calls JSONB, -- [{"id": "call_123", "function": {"name": "...", "arguments": "{}"}}]
    
    -- Tool results (for tool messages)
    tool_call_id VARCHAR(100), -- Links to tool_calls[].id
    
    -- Tokens and cost
    input_tokens INTEGER,
    output_tokens INTEGER,
    total_tokens INTEGER,
    cost_dollars DECIMAL(10,6),
    
    -- Metadata
    model VARCHAR(50),
    finish_reason VARCHAR(20), -- 'stop', 'length', 'tool_calls'
    metadata JSONB DEFAULT '{}',
    
    created_at TIMESTAMPTZ DEFAULT NOW(),
    
    -- Indexes
    INDEX idx_messages_conversation_id (conversation_id),
    INDEX idx_messages_sequence (conversation_id, sequence_number),
    INDEX idx_messages_parent (parent_message_id),
    
    -- Ensure sequence uniqueness per conversation
    UNIQUE (conversation_id, sequence_number)
);

-- Get conversation history
CREATE OR REPLACE FUNCTION get_conversation_history(
    conv_id UUID,
    max_messages INTEGER DEFAULT 100
)
RETURNS TABLE (
    message_id UUID,
    role VARCHAR(20),
    content TEXT,
    tool_calls JSONB,
    sequence_number INTEGER
) AS $$
BEGIN
    RETURN QUERY
    SELECT 
        m.id,
        m.role,
        m.content,
        m.tool_calls,
        m.sequence_number
    FROM messages m
    WHERE m.conversation_id = conv_id
    ORDER BY m.sequence_number ASC
    LIMIT max_messages;
END;
$$ LANGUAGE plpgsql;

-- Get context window (last N messages within token limit)
CREATE OR REPLACE FUNCTION get_context_window(
    conv_id UUID,
    max_tokens INTEGER DEFAULT 8000
)
RETURNS TABLE (
    message_id UUID,
    role VARCHAR(20),
    content TEXT,
    tool_calls JSONB,
    cumulative_tokens INTEGER
) AS $$
DECLARE
    running_total INTEGER := 0;
BEGIN
    RETURN QUERY
    SELECT 
        m.id,
        m.role,
        m.content,
        m.tool_calls,
        (running_total := running_total + COALESCE(m.total_tokens, 0))
    FROM messages m
    WHERE m.conversation_id = conv_id
    ORDER BY m.sequence_number DESC
    LIMIT (
        SELECT COUNT(*)
        FROM (
            SELECT 
                SUM(total_tokens) OVER (ORDER BY sequence_number DESC) as cumulative
            FROM messages
            WHERE conversation_id = conv_id
        ) sub
        WHERE sub.cumulative <= max_tokens
    );
END;
$$ LANGUAGE plpgsql;

Conversation Branching

sql

-- Support branching conversations (regenerate, edit)
CREATE TABLE conversation_branches (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    conversation_id UUID REFERENCES conversations(id) ON DELETE CASCADE,
    branch_name VARCHAR(100) NOT NULL,
    parent_branch_id UUID REFERENCES conversation_branches(id),
    
    -- Branch point
    branch_from_message_id UUID REFERENCES messages(id),
    
    -- Active branch tracking
    is_active BOOLEAN DEFAULT false,
    
    created_at TIMESTAMPTZ DEFAULT NOW(),
    
    INDEX idx_branches_conversation_id (conversation_id)
);

-- Link messages to branches
ALTER TABLE messages ADD COLUMN branch_id UUID REFERENCES conversation_branches(id);
CREATE INDEX idx_messages_branch_id ON messages(branch_id);

-- Get active conversation path
CREATE OR REPLACE FUNCTION get_active_conversation_path(conv_id UUID)
RETURNS TABLE (
    message_id UUID,
    role VARCHAR(20),
    content TEXT,
    sequence_number INTEGER
) AS $$
BEGIN
    RETURN QUERY
    WITH RECURSIVE active_branch AS (
        -- Start with active branch
        SELECT id, parent_branch_id
        FROM conversation_branches
        WHERE conversation_id = conv_id AND is_active = true
        
        UNION ALL
        
        -- Follow branch ancestry
        SELECT cb.id, cb.parent_branch_id
        FROM conversation_branches cb
        JOIN active_branch ab ON cb.id = ab.parent_branch_id
    )
    SELECT m.id, m.role, m.content, m.sequence_number
    FROM messages m
    WHERE m.conversation_id = conv_id
      AND (m.branch_id IN (SELECT id FROM active_branch) OR m.branch_id IS NULL)
    ORDER BY m.sequence_number ASC;
END;
$$ LANGUAGE plpgsql;

Conversation schema principles:

Append-only messages - never update, always insert
Sequence numbers for ordering (not timestamps - can be concurrent)
Parent pointers for branching support
Denormalize tokens/cost for fast analytics
Context window functions to stay within LLM limits

Check our multi-turn conversation evaluation guide for conversation analysis patterns.

Agent State Management

AI agents need to track execution state, tool calls, and decision history.

Agent Execution Schema

sql

-- Agent executions (top-level)
CREATE TABLE agent_executions (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    user_id UUID REFERENCES users(id),
    conversation_id UUID REFERENCES conversations(id),
    
    -- Agent config
    agent_type VARCHAR(50), -- 'rag', 'code_interpreter', 'research', etc.
    model VARCHAR(50),
    
    -- Execution state
    status VARCHAR(20) CHECK (status IN (
        'pending', 'running', 'completed', 'failed', 'cancelled'
    )),
    
    -- Input/output
    input_data JSONB NOT NULL,
    output_data JSONB,
    error_message TEXT,
    
    -- Metrics
    total_steps INTEGER DEFAULT 0,
    tool_calls_count INTEGER DEFAULT 0,
    total_tokens INTEGER DEFAULT 0,
    total_cost DECIMAL(10,4) DEFAULT 0,
    duration_ms INTEGER,
    
    -- Timestamps
    started_at TIMESTAMPTZ DEFAULT NOW(),
    completed_at TIMESTAMPTZ,
    
    INDEX idx_executions_user_id (user_id),
    INDEX idx_executions_status (status),
    INDEX idx_executions_started_at (started_at)
);

-- Execution steps (agent reasoning)
CREATE TABLE execution_steps (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    execution_id UUID REFERENCES agent_executions(id) ON DELETE CASCADE,
    
    -- Step ordering
    step_number INTEGER NOT NULL,
    parent_step_id UUID REFERENCES execution_steps(id),
    
    -- Step type
    step_type VARCHAR(50), -- 'think', 'tool_call', 'observation', 'answer'
    
    -- Content
    thought TEXT, -- Agent's reasoning
    action TEXT, -- Tool name or action
    action_input JSONB, -- Tool parameters
    observation TEXT, -- Tool result
    
    -- Status
    status VARCHAR(20) DEFAULT 'completed',
    error TEXT,
    
    -- Metrics
    tokens INTEGER,
    duration_ms INTEGER,
    
    created_at TIMESTAMPTZ DEFAULT NOW(),
    
    INDEX idx_steps_execution_id (execution_id),
    INDEX idx_steps_step_number (execution_id, step_number),
    UNIQUE (execution_id, step_number)
);

-- Tool invocations (detailed tracking)
CREATE TABLE tool_invocations (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    execution_id UUID REFERENCES agent_executions(id) ON DELETE CASCADE,
    step_id UUID REFERENCES execution_steps(id) ON DELETE CASCADE,
    
    -- Tool info
    tool_name VARCHAR(100) NOT NULL,
    tool_version VARCHAR(20),
    
    -- Invocation
    parameters JSONB NOT NULL,
    result JSONB,
    
    -- Status
    status VARCHAR(20) CHECK (status IN ('success', 'error', 'timeout')),
    error_message TEXT,
    
    -- Metrics
    duration_ms INTEGER,
    retry_count INTEGER DEFAULT 0,
    
    -- Timestamps
    invoked_at TIMESTAMPTZ DEFAULT NOW(),
    completed_at TIMESTAMPTZ,
    
    INDEX idx_tool_invocations_execution_id (execution_id),
    INDEX idx_tool_invocations_tool_name (tool_name),
    INDEX idx_tool_invocations_status (status)
);

-- Get execution trace
CREATE OR REPLACE FUNCTION get_execution_trace(exec_id UUID)
RETURNS TABLE (
    step_number INTEGER,
    step_type VARCHAR(50),
    thought TEXT,
    action TEXT,
    observation TEXT,
    duration_ms INTEGER
) AS $$
BEGIN
    RETURN QUERY
    SELECT 
        es.step_number,
        es.step_type,
        es.thought,
        es.action,
        es.observation,
        es.duration_ms
    FROM execution_steps es
    WHERE es.execution_id = exec_id
    ORDER BY es.step_number ASC;
END;
$$ LANGUAGE plpgsql;

-- Get tool usage stats
CREATE OR REPLACE FUNCTION get_tool_usage_stats(
    start_date TIMESTAMPTZ,
    end_date TIMESTAMPTZ
)
RETURNS TABLE (
    tool_name VARCHAR(100),
    total_calls BIGINT,
    success_count BIGINT,
    error_count BIGINT,
    avg_duration_ms NUMERIC,
    total_duration_ms BIGINT
) AS $$
BEGIN
    RETURN QUERY
    SELECT 
        ti.tool_name,
        COUNT(*)::BIGINT,
        COUNT(*) FILTER (WHERE ti.status = 'success')::BIGINT,
        COUNT(*) FILTER (WHERE ti.status = 'error')::BIGINT,
        AVG(ti.duration_ms),
        SUM(ti.duration_ms)::BIGINT
    FROM tool_invocations ti
    WHERE ti.invoked_at BETWEEN start_date AND end_date
    GROUP BY ti.tool_name
    ORDER BY COUNT(*) DESC;
END;
$$ LANGUAGE plpgsql;

Agent state principles:

Hierarchical structure - execution → steps → tool calls
Append-only logs - immutable audit trail
Status tracking at every level
Performance metrics for optimization
Rich metadata for debugging and analysis

Our AI agent development services use these patterns for production agent systems.

Document & Metadata Schema

RAG systems need flexible metadata for filtering and faceted search.

Document Metadata Design

sql

-- Flexible metadata using JSONB
CREATE TABLE documents_with_metadata (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    title TEXT NOT NULL,
    content TEXT NOT NULL,
    
    -- Structured metadata
    metadata JSONB DEFAULT '{}'::jsonb,
    
    created_at TIMESTAMPTZ DEFAULT NOW(),
    updated_at TIMESTAMPTZ DEFAULT NOW()
);

-- Metadata indexes (critical for performance)
-- Index specific metadata fields you query frequently
CREATE INDEX idx_documents_metadata_category 
ON documents_with_metadata ((metadata->>'category'));

CREATE INDEX idx_documents_metadata_tags 
ON documents_with_metadata USING GIN ((metadata->'tags'));

CREATE INDEX idx_documents_metadata_date 
ON documents_with_metadata ((metadata->>'created_date'));

-- Full-text search on content
CREATE INDEX idx_documents_content_fts 
ON documents_with_metadata USING GIN (to_tsvector('english', content));

-- Example metadata structure
INSERT INTO documents_with_metadata (title, content, metadata) VALUES (
    'RAG System Architecture',
    'Full content here...',
    '{
        "category": "documentation",
        "tags": ["ai", "rag", "architecture"],
        "author": "John Doe",
        "department": "engineering",
        "created_date": "2026-09-14",
        "visibility": "public",
        "version": "1.0",
        "source": {
            "type": "upload",
            "url": "https://...",
            "uploaded_by": "user_123"
        },
        "custom_fields": {
            "priority": "high",
            "project_id": "proj_abc"
        }
    }'::jsonb
);

-- Query with metadata filtering
SELECT id, title, content
FROM documents_with_metadata
WHERE 
    metadata->>'category' = 'documentation'
    AND metadata->'tags' ? 'ai'
    AND metadata->>'visibility' = 'public'
ORDER BY (metadata->>'created_date') DESC;

-- Full-text + metadata combination
SELECT 
    id,
    title,
    ts_rank(to_tsvector('english', content), query) AS rank
FROM documents_with_metadata,
     to_tsquery('english', 'architecture & rag') AS query
WHERE 
    to_tsvector('english', content) @@ query
    AND metadata->>'category' = 'documentation'
ORDER BY rank DESC;

Metadata Validation

python

from pydantic import BaseModel, Field
from typing import Optional, List
from datetime import datetime

class DocumentMetadata(BaseModel):
    """Validated metadata schema."""
    
    # Required fields
    category: str = Field(..., description="Document category")
    tags: List[str] = Field(default_factory=list, description="Search tags")
    
    # Optional fields
    author: Optional[str] = None
    department: Optional[str] = None
    created_date: Optional[str] = None
    visibility: str = Field(default="private", pattern="^(public|private|restricted)$")
    version: str = "1.0"
    
    # Source tracking
    source: dict = Field(default_factory=dict)
    
    # Custom fields (flexible)
    custom_fields: dict = Field(default_factory=dict)
    
    class Config:
        extra = "allow"  # Allow additional fields

# Usage
metadata = DocumentMetadata(
    category="documentation",
    tags=["ai", "rag"],
    author="John Doe",
    custom_fields={"priority": "high"}
)

# Insert with validated metadata
INSERT INTO documents_with_metadata (title, content, metadata)
VALUES ('...', '...', %s::jsonb)
ON CONFLICT DO NOTHING
""", (metadata.json(),))

Metadata schema principles:

JSONB for flexibility - evolve schema without migrations
Index queried fields - extract and index specific JSON keys
Validate at application layer - use Pydantic for consistency
Namespaced custom fields - prevent collision with system fields
Audit trail - track source, author, timestamps

Performance Optimization

AI workloads have unique performance characteristics:

Partitioning Strategy

sql

-- Partition conversations by month
CREATE TABLE conversations_partitioned (
    id UUID NOT NULL,
    user_id UUID NOT NULL,
    created_at TIMESTAMPTZ NOT NULL,
    -- ... other fields
) PARTITION BY RANGE (created_at);

-- Create monthly partitions
CREATE TABLE conversations_2026_09 PARTITION OF conversations_partitioned
FOR VALUES FROM ('2026-09-01') TO ('2026-10-01');

CREATE TABLE conversations_2026_10 PARTITION OF conversations_partitioned
FOR VALUES FROM ('2026-10-01') TO ('2026-11-01');

-- Automatic partition creation
CREATE OR REPLACE FUNCTION create_monthly_partition(table_name text, date_col text)
RETURNS void AS $$
DECLARE
    partition_name text;
    start_date date;
    end_date date;
BEGIN
    start_date := date_trunc('month', now());
    end_date := start_date + interval '1 month';
    partition_name := table_name || '_' || to_char(start_date, 'YYYY_MM');
    
    EXECUTE format(
        'CREATE TABLE IF NOT EXISTS %I PARTITION OF %I FOR VALUES FROM (%L) TO (%L)',
        partition_name, table_name, start_date, end_date
    );
END;
$$ LANGUAGE plpgsql;

Connection Pooling

python

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

# Production connection pool
engine = create_async_engine(
    "postgresql+asyncpg://user:pass@localhost/aidb",
    pool_size=20,  # Concurrent connections
    max_overflow=10,  # Burst capacity
    pool_pre_ping=True,  # Verify connections
    pool_recycle=3600,  # Recycle after 1 hour
    echo=False
)

# Session factory
async_session = sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False
)

# Usage
async with async_session() as session:
    result = await session.execute(query)

Caching Layer

python

import redis.asyncio as redis
from functools import wraps
import json

class EmbeddingCache:
    """Cache embeddings to avoid recomputation."""
    
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client
        self.ttl = 86400 * 30  # 30 days
    
    async def get_embedding(self, text: str) -> Optional[List[float]]:
        """Get cached embedding."""
        key = f"emb:{hash(text)}"
        cached = await self.redis.get(key)
        
        if cached:
            return json.loads(cached)
        return None
    
    async def set_embedding(self, text: str, embedding: List[float]):
        """Cache embedding."""
        key = f"emb:{hash(text)}"
        await self.redis.setex(key, self.ttl, json.dumps(embedding))

cache = EmbeddingCache(redis_client)

async def get_or_compute_embedding(text: str) -> List[float]:
    """Get embedding from cache or compute."""
    
    # Check cache
    if cached := await cache.get_embedding(text):
        return cached
    
    # Compute
    embedding = await openai_client.embeddings.create(
        input=text,
        model="text-embedding-3-small"
    )
    
    vector = embedding.data[0].embedding
    
    # Cache for future
    await cache.set_embedding(text, vector)
    
    return vector

Performance optimization principles:

Partition large tables by time - improves query performance dramatically
Connection pooling - reuse connections, avoid overhead
Cache embeddings - generating embeddings is expensive
Batch operations - insert/update in batches, not row-by-row
Async all the way - use async DB drivers (asyncpg, motor)

See our LLM inference optimization guide for end-to-end performance.

Schema Migration Strategies

Production AI systems need careful migration strategies:

Alembic Migrations

python

# migrations/versions/001_add_vector_support.py
"""Add vector support

Revision ID: 001
Create Date: 2026-09-14
"""

from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql

def upgrade():
    # Enable pgvector
    op.execute("CREATE EXTENSION IF NOT EXISTS vector")
    
    # Add embedding column
    op.add_column(
        'document_chunks',
        sa.Column('embedding', postgresql.ARRAY(sa.Float), nullable=True)
    )
    
    # Create vector index (after embeddings populated)
    # Note: Create index separately after backfill
    pass

def downgrade():
    op.drop_column('document_chunks', 'embedding')

Zero-Downtime Migrations

python

# Step 1: Add new column (nullable)
op.add_column('messages', sa.Column('tool_calls', sa.JSON, nullable=True))

# Step 2: Backfill in batches
async def backfill_tool_calls():
    """Backfill tool_calls from old format."""
    
    batch_size = 1000
    offset = 0
    
    while True:
        # Fetch batch
        query = """
            SELECT id, metadata
            FROM messages
            WHERE tool_calls IS NULL
              AND metadata->>'tool_calls' IS NOT NULL
            LIMIT %s OFFSET %s
        """
        
        rows = await db.fetch_all(query, (batch_size, offset))
        
        if not rows:
            break
        
        # Update batch
        updates = [
            (row['metadata']['tool_calls'], row['id'])
            for row in rows
        ]
        
        await db.execute_many(
            "UPDATE messages SET tool_calls = %s WHERE id = %s",
            updates
        )
        
        offset += batch_size
        
        # Throttle to avoid overload
        await asyncio.sleep(1)

# Step 3: Make column non-nullable (after backfill complete)
op.alter_column('messages', 'tool_calls', nullable=False)

Migration principles:

Additive changes - add columns, don't remove (yet)
Backfill separately - don't block deployment
Batched operations - avoid locking entire table
Rollback plan - test downgrade migrations
Monitor performance - track migration progress

Production Examples

Complete RAG System Schema

sql

-- Production-ready schema for RAG system
BEGIN;

-- Users and auth (simplified)
CREATE TABLE users (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    email VARCHAR(255) UNIQUE NOT NULL,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

-- Organizations (multi-tenant)
CREATE TABLE organizations (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    name VARCHAR(255) NOT NULL,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

CREATE TABLE organization_members (
    organization_id UUID REFERENCES organizations(id),
    user_id UUID REFERENCES users(id),
    role VARCHAR(50) DEFAULT 'member',
    PRIMARY KEY (organization_id, user_id)
);

-- Documents
CREATE TABLE documents (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    organization_id UUID REFERENCES organizations(id),
    title TEXT NOT NULL,
    content TEXT NOT NULL,
    url TEXT,
    metadata JSONB DEFAULT '{}',
    created_by UUID REFERENCES users(id),
    created_at TIMESTAMPTZ DEFAULT NOW(),
    updated_at TIMESTAMPTZ DEFAULT NOW(),
    
    INDEX idx_documents_org_id (organization_id),
    INDEX idx_documents_created_at (created_at)
);

-- Chunks with embeddings
CREATE EXTENSION IF NOT EXISTS vector;

CREATE TABLE document_chunks (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    document_id UUID REFERENCES documents(id) ON DELETE CASCADE,
    organization_id UUID REFERENCES organizations(id), -- Denormalized for filtering
    
    content TEXT NOT NULL,
    chunk_index INTEGER NOT NULL,
    embedding vector(1536),
    
    metadata JSONB DEFAULT '{}',
    created_at TIMESTAMPTZ DEFAULT NOW(),
    
    INDEX idx_chunks_document_id (document_id),
    INDEX idx_chunks_org_id (organization_id)
);

CREATE INDEX idx_chunks_embedding ON document_chunks 
USING hnsw (embedding vector_cosine_ops);

-- Conversations
CREATE TABLE conversations (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    organization_id UUID REFERENCES organizations(id),
    user_id UUID REFERENCES users(id),
    
    title TEXT,
    model VARCHAR(50) DEFAULT 'gpt-4',
    message_count INTEGER DEFAULT 0,
    total_tokens INTEGER DEFAULT 0,
    
    created_at TIMESTAMPTZ DEFAULT NOW(),
    updated_at TIMESTAMPTZ DEFAULT NOW(),
    
    INDEX idx_conversations_org_id (organization_id),
    INDEX idx_conversations_user_id (user_id)
);

-- Messages
CREATE TABLE messages (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    conversation_id UUID REFERENCES conversations(id) ON DELETE CASCADE,
    
    sequence_number INTEGER NOT NULL,
    role VARCHAR(20) NOT NULL,
    content TEXT,
    tool_calls JSONB,
    
    total_tokens INTEGER,
    
    created_at TIMESTAMPTZ DEFAULT NOW(),
    
    INDEX idx_messages_conversation_id (conversation_id),
    UNIQUE (conversation_id, sequence_number)
);

-- RAG queries (analytics)
CREATE TABLE rag_queries (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    conversation_id UUID REFERENCES conversations(id),
    message_id UUID REFERENCES messages(id),
    
    query_text TEXT NOT NULL,
    query_embedding vector(1536),
    
    -- Retrieved chunks
    chunks_retrieved INTEGER,
    chunks_used INTEGER,
    
    -- Performance
    embedding_time_ms INTEGER,
    search_time_ms INTEGER,
    total_time_ms INTEGER,
    
    created_at TIMESTAMPTZ DEFAULT NOW(),
    
    INDEX idx_rag_queries_conversation_id (conversation_id)
);

CREATE TABLE rag_query_chunks (
    query_id UUID REFERENCES rag_queries(id) ON DELETE CASCADE,
    chunk_id UUID REFERENCES document_chunks(id),
    
    rank INTEGER,
    similarity_score DECIMAL(5,4),
    used_in_context BOOLEAN DEFAULT false,
    
    PRIMARY KEY (query_id, chunk_id)
);

COMMIT;

Check our RAG system development services for production implementations.

Primary references: official documentation, official documentation, official documentation, official documentation.

Operating Database Schema Design for AI Applications as a System

The implementation is only one part of Database Schema Design for AI Applications. A production design also needs an explicit contract for inputs, outputs, ownership, and failure behavior. Write that contract before selecting a library. It should identify which component validates input, where state lives, what may be retried, and which result is authoritative when two components disagree. This prevents a convenient prototype boundary from silently becoming the long-term architecture.

Start with a representative baseline. Capture request shape, traffic distribution, dependency latency, error classes, and the quality signal users actually care about. Averages hide the cases that cause incidents, so keep percentiles and segment measurements by workload type. Record the configuration and dataset version beside every result. Without that context, a faster or more accurate run cannot be reproduced and should not be used to approve a rollout.

Define the failure model

List failures by where they originate: invalid input, capacity exhaustion, dependency timeout, partial state change, malformed output, and semantically wrong output. Each class needs a different response. Validation errors should fail immediately. Transient dependency failures may be retried with a budget and jitter. An operation that may have committed must use an idempotency key or reconciliation step before retrying. A syntactically valid but incorrect result belongs in evaluation and review, not a blind retry loop.

Set a deadline for the complete operation and derive smaller budgets for each dependency. Local timeouts that add up to more than the caller's deadline merely create abandoned work. Propagate cancellation where the protocol supports it. Bound every queue, retry loop, context buffer, and concurrency pool; an unbounded safety mechanism becomes a second outage during overload.

Design a degraded mode before it is needed. Depending on the workload, that can mean returning a cached answer, selecting a simpler path, placing work in a durable queue, or asking for human review. The degraded response must be visible in telemetry and, where it changes meaning, visible to the caller. Silent fallback makes quality regressions almost impossible to diagnose.

Measure the decision, not just the component

Use three layers of signals. System metrics cover latency, throughput, saturation, and errors. Correctness metrics measure whether the result satisfies its contract. Business or user metrics show whether the system solved the intended problem. Improving only one layer can move the others backward, so release criteria should name acceptable movement for all three.

Attach a reason code to every route, rejection, fallback, and retry. Include version identifiers for configuration, code, model, schema, and data when relevant. Logs should let an engineer reconstruct a decision without storing secrets or raw personal data. Traces should cross process boundaries, while metrics should remain low-cardinality enough to operate reliably.

Alert on symptoms that require action, not every internal anomaly. A useful alert names the affected service objective, links to a runbook, and distinguishes a customer-visible incident from exhausted headroom. Dashboards serve a different purpose: they support diagnosis and capacity planning. Treating a dashboard as an alerting strategy leaves failures undiscovered until someone happens to look.

Roll out with reversible steps

Ship Database Schema Design for AI Applications behind a versioned interface and a kill switch. Begin with offline replay using production-shaped, privacy-safe samples. Then use shadow execution when duplicate work has acceptable cost and side effects can be suppressed. A small canary should exercise the real dependency graph before traffic expands. Compare the canary with the baseline by cohort rather than mixing both populations into one aggregate.

Promotion gates should be written before the rollout. Include a minimum sample size or observation window, maximum regression in tail latency and error rate, and a correctness threshold. Roll back automatically when a hard safety boundary is crossed; use manual review for ambiguous quality movement. Preserve enough evidence from both paths to explain why the gate passed or failed.

Configuration deserves the same discipline as code. Review changes, validate them before activation, keep an immutable history, and make rollback a single operation. If a deployment changes code and configuration together, record both versions. Otherwise an incident responder may roll back the binary while leaving the triggering configuration active.

Capacity and cost controls

Model capacity in units the bottleneck understands: concurrent connections, tokens, queue jobs, database transactions, GPU memory, or bytes in flight. Convert the expected traffic distribution into those units and include burst behavior. Then load-test the first constrained dependency, not merely the public endpoint. A system that accepts more work than it can finish within its deadline is overloaded even if CPU utilization looks comfortable.

Cost is also a reliability limit. Add per-request attribution, tenant or workflow budgets, and a global circuit breaker for unexpectedly expensive paths. Review unit economics at the same granularity as performance; a cheap median can conceal a small class of requests responsible for most spend. Optimize only after measuring, because reducing context, replicas, validation, or redundancy can trade visible cost for less visible risk.

Production readiness review

Before launch, ask an engineer who did not build the feature to follow the runbook through one simulated failure. Verify backups or checkpoints by restoring them, not by checking that a job reported success. Exercise credential rotation, dependency unavailability, bad configuration, and rollback. Assign an owner for each alarm and a date for reviewing thresholds after real traffic arrives.

The final architecture document should be short enough to remain current. Keep the decision, rejected alternatives, invariants, dependency contracts, dashboards, and rollback procedure. Link detailed experiments rather than pasting them into the document. Teams that need help turning this review into an operable service can use our Database Schema Design for AI Applications engineering support.

Frequently Asked Questions

Should I use PostgreSQL or a specialized vector database?

Start with PostgreSQL + pgvector for <1M vectors - it's simpler, cheaper, and handles relational data alongside vectors. Switch to Pinecone/Weaviate/Qdrant when you need: (1) >1M vectors, (2) <10ms p99 latency, (3) advanced vector features (hybrid search, filtering), or (4) horizontal scaling. Most production RAG systems use both - Postgres for metadata/conversations, vector DB for embeddings.

How do I handle embedding model changes (1536 dims → 3072 dims)?

Add versioned embedding columns: embedding_v1 vector(1536), embedding_v2 vector(3072). Backfill new embeddings in batches, update application to use new column, drop old column after migration complete. Or use separate tables: document_chunks_v1, document_chunks_v2 with routing logic. Never modify embedding dimension in place.

What's the best way to store multi-turn conversations?

Messages table with sequence_number (not timestamps for ordering). Use conversation_id foreign key to group messages. For branching, add parent_message_id. Store system prompt in conversations table. Denormalize token counts for fast quota checking. See conversation history section for complete schema.

How do I prevent vector index from slowing down writes?

Batch inserts (1000+ vectors at once), create index after initial bulk load, use IVFFlat for write-heavy workloads (HNSW is slower to build). For real-time systems, consider async indexing: write to table immediately, update vector index in background job. Or use separate "hot" table (no index, for recent vectors) and "cold" table (indexed, for older vectors).

Should I store full document content in the same table as embeddings?

No - store content in object storage (S3) or separate table. Vector tables should be lean: chunk_id, embedding, minimal metadata. Large content (>10KB) hurts cache efficiency and index performance. Store content_url or content_hash in vector table, fetch full text only when needed.

How do I implement multi-tenant isolation in vector search?

Denormalize organization_id into vector table, add to WHERE clause: WHERE organization_id = $1 AND similarity > 0.7. Index on (organization_id, similarity) for performance. Or use metadata filtering: metadata @> '{"organization_id": "org_123"}'. Pinecone supports namespaces for tenant isolation. Never rely on application filtering alone - enforce at database level.

What's the right chunk size for vector storage?

200-500 tokens (roughly 150-400 words) balances context and granularity. Smaller chunks (100 tokens) give precise matches but lose context. Larger chunks (1000+ tokens) retain context but dilute relevance. Test with your data - technical docs work well with 300 tokens, narrative content with 500+. See our RAG chunking guide for benchmarks.

How do I handle schema changes in production without downtime?

Multi-step deployments: (1) Add new column (nullable), (2) Deploy code that writes to both old and new, (3) Backfill existing rows in batches, (4) Deploy code that reads from new column, (5) Drop old column. Use feature flags to control rollout. For critical tables, use blue-green database deployment or read replicas during migration.

Conclusion

AI application databases require specialized schemas optimized for vector similarity search, conversation history, and agent execution logs. Traditional normalized designs don't work - denormalize aggressively for performance.

Key takeaways:

Vector storage needs careful indexing - HNSW for reads, IVFFlat for writes
Denormalize metadata into vector records for filtering performance
Conversations are append-only - sequence numbers, not timestamps
Agent state is hierarchical - execution → steps → tool calls
Partition large tables by time - dramatically improves query performance
Cache embeddings - generating is expensive, storage is cheap
Multi-tenant isolation - denormalize org_id, never trust application filtering

The schemas in this guide power production AI systems handling 10M+ interactions monthly. Start with Postgres + pgvector, add vector database when you hit scale limits, optimize with caching and partitioning.

Need help designing your AI database? Our backend API engineering team specializes in high-performance schema design for AI agent systems, RAG applications, and multi-tenant AI platforms.

Related guides:

Free consultation

Book a free consultation call on database design for AI applications

30-minute call with the HinterBuild team. Discuss your project, architecture questions, or next steps — no obligation.

Book a meeting

Keep reading

Backend SystemsSeptember 11, 2026 · 11 min read

Database Schema Migrations Without Downtime

Database Schema Migrations Without Downtime guidance for engineers: compare architecture choices, avoid failure modes, and ship a measurable, reliable.

Read post

Backend SystemsSeptember 11, 2026 · 10 min read

Multi-Tenancy Patterns: Database Strategies for SaaS

Learn multi-tenancy patterns through concrete architecture trade-offs, failure modes, rollout controls, and production measurement practices.

Read post

Backend SystemsSeptember 11, 2026 · 11 min read

SQLite in Production: When It Beats PostgreSQL (Guide)

SQLite in Production guidance for engineers: compare architecture choices, avoid failure modes, and ship a measurable, reliable production implementation.

Read post

Backend SystemsSeptember 11, 2026 · 9 min read

Redis vs Valkey What Changed After the Fork (Complete Guide)

Redis vs Valkey What Changed After the Fork (Complete Guide) guidance for engineers: compare architecture choices, avoid failure modes, and ship a.

Read post

Why AI Databases Are Different

Key Differences

Vector Storage Schema

PostgreSQL with pgvector

Pinecone Schema (Vector-Native DB)

Conversation History Design

Conversation Schema

Conversation Branching

Agent State Management

Agent Execution Schema

Document & Metadata Schema

Document Metadata Design

Metadata Validation

Performance Optimization

Partitioning Strategy

Connection Pooling

Caching Layer

Schema Migration Strategies

Alembic Migrations

Zero-Downtime Migrations

Production Examples

Complete RAG System Schema

Operating Database Schema Design for AI Applications as a System

Define the failure model

Measure the decision, not just the component

Roll out with reversible steps

Capacity and cost controls

Production readiness review

Frequently Asked Questions

Should I use PostgreSQL or a specialized vector database?

How do I handle embedding model changes (1536 dims → 3072 dims)?

What's the best way to store multi-turn conversations?

How do I prevent vector index from slowing down writes?

Should I store full document content in the same table as embeddings?

How do I implement multi-tenant isolation in vector search?

What's the right chunk size for vector storage?

How do I handle schema changes in production without downtime?

Conclusion

Related articles

Database Schema Migrations Without Downtime

Multi-Tenancy Patterns: Database Strategies for SaaS

SQLite in Production: When It Beats PostgreSQL (Guide)

Redis vs Valkey What Changed After the Fork (Complete Guide)