Airflow + vector DB + FastAPI vs Pixeltable

import pixeltable as pxt
from pixeltable.functions.document import document_splitter
from pixeltable.functions.huggingface import sentence_transformer
from pixeltable.serving import FastAPIRouter

TableModel = pxt.model_base()
embed = sentence_transformer.using(
    model_id='sentence-transformers/all-MiniLM-L6-v2'
)

class Docs(TableModel, name='docs'):
    document: pxt.Document
    title: pxt.String

class Chunks(
    TableModel,
    name='chunks',
    base=Docs,
    iterator=document_splitter(
        document=Docs.document, separators='sentence,token_limit', limit=512
    ),
):
    __indexes__ = [pxt.EmbeddingIndex(text, string_embed=embed)]

@pxt.query
def search_docs(query: str, limit: int = 5):
    sim = Chunks.text.similarity(string=query)
    return (
        Chunks.order_by(sim, asc=False)
        .limit(limit)
        .select(Chunks.text, Chunks.title, sim)
    )

api = FastAPIRouter(name='docs')
api.add_insert_route(Docs, path='/docs', inputs=[Docs.document, Docs.title])
api.add_query_route(path='/search', query=search_docs)

from datetime import datetime
from airflow.sdk import dag, task

@dag(schedule=None, start_date=datetime(2026, 1, 1), catchup=False)
def rag_stack():
    @task
    def chunk_pdfs():
        # Files still live in object storage you operate.
        return [{'id': '1', 'text': 'Refund policy...'}]

    @task
    def embed(chunks):
        return [{'id': c['id'], 'values': [0.1, 0.2]} for c in chunks]

    @task
    def upsert(vectors):
        # index.upsert(vectors)  # Pinecone, Weaviate, or pgvector
        return len(vectors)

    upsert(embed(chunk_pdfs()))

rag_stack()
# HTTP is a second FastAPI app that queries the vector DB and reads S3.