PixeltablePixeltable Logo

How to

How do I build an AI data pipeline in Python?

An AI data pipeline ingests files, runs models, and writes derived columns that later steps can query. How to build an AI pipeline is a schema question: typed inputs, transforms that run on insert, and indexes that stay current. AI pipeline tools and a data ingestion pipeline are the same job when storage, orchestration, and serving share one Python package.

Updated · Part of How do I ... in Python?

Before you start

About 20 min.

  • Python 3.11+
  • pip install -U 'pixeltable[serve]'
  • pxt init in the project root
  • An OpenAI API key for gpt-4o-mini

1. Ingest with typed columns, not a drop folder

A data ingestion pipeline that only copies bytes into a bucket still needs a table. Declare Image, Document, or String columns so later transforms know the type.

import pixeltable as pxt
from pixeltable.functions.document import document_splitter
from pixeltable.functions.huggingface import sentence_transformer
from pixeltable.functions.openai import chat_completions


TableModel = pxt.model_base()
text_embed = sentence_transformer.using(model_id='all-MiniLM-L6-v2')


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

2. Declare transforms as columns and views

AI pipeline tools are usually a queue, a worker, and a store. Here the worker is the column: an LLM call, a splitter view, an embedding index. Insert runs the affected expressions. Unchanged rows stay cached.

class Docs(TableModel, name='docs'):
    source: pxt.String
    notes: pxt.String
    document: pxt.Document
    summary = chat_completions(
        messages=[{'role': 'user', 'content': notes}],
        model='gpt-4o-mini',
    ).choices[0].message.content


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

3. Query and serve from the same file

How to build an AI pipeline ends at a query you can call, not a folder of derived files. @pxt.query ranks chunks. FastAPIRouter can expose insert and search. Apply with pxt schema update -f; serve with pxt service update -f. FastAPIRouter needs the serve extra, which the install line includes.

from pixeltable.serving import FastAPIRouter


@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, sim)


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


# pxt init
# pxt schema update app.py pipeline -f
# pxt service update app.py pipeline -f

AI pipeline tools versus one schema

AI pipeline tools versus one schema
JobTypical glue stackOne Python package
IngestA loader into object storage plus a metadata DBTyped columns on the table you own
TransformAirflow or a queue of workersComputed columns and iterator views
IndexA vector database you keep alignedEmbeddingIndex on the derived rows
ServeA FastAPI app that joins those copiesFastAPIRouter beside the tables

Where Pixeltable fits

Pixeltable is the unified multimodal backend those AI pipeline tools try to assemble: ingest, orchestration, and serving in one Python package (pip install pixeltable[serve]). Compare Pixeltable vs Airflow when you already run a fleet of unrelated DAGs, and Pixeltable vs Daft when the job is a batch dataframe, not an application table.

import pixeltable as pxt
from pixeltable.functions.document import document_splitter
from pixeltable.functions.huggingface import sentence_transformer
from pixeltable.functions.openai import chat_completions


TableModel = pxt.model_base()
text_embed = sentence_transformer.using(model_id='all-MiniLM-L6-v2')


class Docs(TableModel, name='docs'):
    notes: pxt.String
    document: pxt.Document
    summary = chat_completions(
        messages=[{'role': 'user', 'content': notes}],
        model='gpt-4o-mini',
    ).choices[0].message.content


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

Questions

Is this a replacement for Airflow?
For media, models, and retrieval on one table, yes: the DAG is the schema. For a fleet of unrelated batch jobs, keep Airflow. Pixeltable vs Airflow is that split. Daft is the batch-dataframe aisle.
What does a data ingestion pipeline look like here?
Insert rows into typed columns (local paths, URLs, or object storage). Computed columns and views run for those rows. You do not stage a second copy for the worker to find.
Where do AI pipeline tools fit?
Most stacks split ingest, orchestration, vectors, and HTTP. One Python package holds those layers when the unit of work is a row. Hosted ANN or a labeling UI can still sit beside it.