---
title: "How do I build an AI data pipeline in Python?"
description: "Build an AI data pipeline in Python: ingest typed media, declare computed columns, and keep indexes current on insert — one package, not a glue DAG."
url: "https://www.pixeltable.com/learn/how-to/build-an-ai-data-pipeline"
updated: "2026-10-14"
doc: "https://docs.pixeltable.com/datastore/computed-columns"
---

# 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: 2026-10-14
Part of [How do I ... in Python?](https://www.pixeltable.com/learn/how-to).

## 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

## On this page

- [Before you start](https://www.pixeltable.com/learn/how-to/build-an-ai-data-pipeline#before-you-start)
- [Ingest with typed columns, not a drop folder](https://www.pixeltable.com/learn/how-to/build-an-ai-data-pipeline#step-1)
- [Declare transforms as columns and views](https://www.pixeltable.com/learn/how-to/build-an-ai-data-pipeline#step-2)
- [Query and serve from the same file](https://www.pixeltable.com/learn/how-to/build-an-ai-data-pipeline#step-3)
- [Comparison](https://www.pixeltable.com/learn/how-to/build-an-ai-data-pipeline#comparison)
- [Where Pixeltable fits](https://www.pixeltable.com/learn/how-to/build-an-ai-data-pipeline#where-pixeltable-fits)
- [Questions](https://www.pixeltable.com/learn/how-to/build-an-ai-data-pipeline#questions)

## 1. Ingest with typed columns, not a drop folder {#step-1}

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.

```python
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 {#step-2}

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.

```python
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 {#step-3}

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.

```python
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 {#comparison}

| Job | Typical glue stack | One Python package |
| --- | --- | --- |
| Ingest | A loader into object storage plus a metadata DB | Typed columns on the table you own |
| Transform | Airflow or a queue of workers | Computed columns and iterator views |
| Index | A vector database you keep aligned | EmbeddingIndex on the derived rows |
| Serve | A FastAPI app that joins those copies | FastAPIRouter beside the tables |

## Where Pixeltable fits {#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.

```python
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),
    ]
```

## Concepts

- [What is a declarative pipeline?](https://www.pixeltable.com/learn/what-is-a-declarative-pipeline)
- [What is a computed column?](https://www.pixeltable.com/learn/what-is-a-computed-column)
- [Best multimodal data platforms for AI apps (2026)](https://www.pixeltable.com/learn/best-multimodal-data-platforms)

## Use case

- [Declarative AI Infrastructure: Define Pipelines, Not Plumbing](https://www.pixeltable.com/use-cases/declarative-ai-data-infrastructure)

## Questions {#questions}

### Is this a replacement for Airflow? {#faq-1}

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? {#faq-2}

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? {#faq-3}

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.

## Related

- [What is a declarative pipeline?](https://www.pixeltable.com/learn/what-is-a-declarative-pipeline)
- [What is a computed column?](https://www.pixeltable.com/learn/what-is-a-computed-column)
- [Best multimodal data platforms (2026)](https://www.pixeltable.com/learn/best-multimodal-data-platforms)
- [Declarative data infrastructure](https://www.pixeltable.com/use-cases/declarative-ai-data-infrastructure)
- [How do I add computed columns with LLM calls in Python?](https://www.pixeltable.com/learn/how-to/add-computed-columns-with-llm-calls)
- [Pixeltable vs Airflow](https://www.pixeltable.com/compare/pixeltable-vs-airflow)
- [Pixeltable vs Daft](https://www.pixeltable.com/compare/pixeltable-vs-daft)
- [Pixeltable vs LangChain](https://www.pixeltable.com/compare/pixeltable-vs-langchain)
- [Computed columns](https://docs.pixeltable.com/datastore/computed-columns)