sql
**Databases for AI Applications**
**SQL Databases**
**PostgreSQL**
The most versatile database for AI applications, with strong extensions for vectors and time-series.
```sql
-- Basic table for RAG application
CREATE TABLE documents (
id SERIAL PRIMARY KEY,
content TEXT NOT NULL,
metadata JSONB,
created_at TIMESTAMP DEFAULT NOW()
);
-- Full-text search
CREATE INDEX idx_content_search ON documents USING GIN(to_tsvector('english', content));
```
**SQLite**
Perfect for local development and embedded applications:
```python
import sqlite3
conn = sqlite3.connect("local_db.sqlite")
```
**Vector Search with PostgreSQL**
**pgvector Extension**
```sql
-- Enable extension
CREATE EXTENSION vector;
-- Create table with embeddings
CREATE TABLE embeddings (
id SERIAL PRIMARY KEY,
content TEXT,
embedding vector(384) -- Match your model dimension
);
-- Create HNSW index for fast search
CREATE INDEX ON embeddings USING hnsw (embedding vector_cosine_ops);
-- Similarity search
SELECT content, embedding <=> '[0.1, 0.2, ...]' AS distance
FROM embeddings
ORDER BY embedding <=> '[0.1, 0.2, ...]'
LIMIT 10;
```
**Time-Series with TimescaleDB**
**For LLM Metrics and Monitoring**
```sql
-- Enable extension
CREATE EXTENSION timescaledb;
-- Create hypertable for metrics
CREATE TABLE llm_requests (
time TIMESTAMPTZ NOT NULL,
model TEXT,
tokens INTEGER,
latency_ms INTEGER,
cost DECIMAL
);
SELECT create_hypertable('llm_requests', 'time');
-- Aggregate by hour
SELECT time_bucket('1 hour', time) AS hour,
model,
avg(latency_ms) AS avg_latency,
sum(tokens) AS total_tokens
FROM llm_requests
GROUP BY hour, model;
```
**Database Comparison for AI**
| Database | Best For | Strengths |
|----------|----------|-----------|
| PostgreSQL + pgvector | Full-stack SQL + vectors | Familiar SQL, rich ecosystem |
| Pinecone | Managed vector DB | Serverless, easy setup |
| Qdrant | Self-hosted vectors | Rust-based, filtering |
| MongoDB | Document storage | Flexible schema |
| Redis | Caching | Sub-ms latency |
| TimescaleDB | Metrics/monitoring | Time-series analysis |
**Connection Patterns**
**Python with psycopg2**
```python
import psycopg2
conn = psycopg2.connect(
host="localhost",
database="mydb",
user="postgres",
password="password"
)
with conn.cursor() as cur:
cur.execute("SELECT * FROM documents WHERE id = %s", (doc_id,))
result = cur.fetchone()
```
**Async with asyncpg**
```python
import asyncpg
async def fetch_documents():
conn = await asyncpg.connect(DATABASE_URL)
rows = await conn.fetch("SELECT * FROM documents")
await conn.close()
return rows
```
**Best Practices**
1. **Connection pooling**: Use pgBouncer or connection pools
2. **Indexing**: Create indexes for common queries
3. **Batching**: Insert/update in batches for embedding storage
4. **Backups**: Automated backups for vector data
5. **Scaling**: Consider read replicas for heavy read loads