pixeltable

pxt.Image, pxt.Video, pxt.Audio, pxt.Document, pxt.Json: one table for structured and media data with destination= for S3, GCS, Azure, R2, and more. Not S3 + Postgres + boto3 sync.

t = pxt.create_table(
    'media',
    {
        'img': pxt.Image,
        'video': pxt.Video,
        'audio': pxt.Audio,
        'document': pxt.Document,
        'metadata': pxt.Json,
    },
)

Type system · Tables & data · Cloud storage

create_table(source=...), path/URL insert(), Hugging Face, export_parquet(), PyTorch, COCO, and more. Not per-format ETL scripts.

# Create a table from a file, URL, or Hugging Face dataset
pxt.create_table('app/data', source='data.csv')
pxt.create_table('app/reviews', source=hf_dataset)

# Append rows into an existing table from a path or URL
t.insert('s3://my-bucket/new_rows.parquet')

# Export to analytics/ML formats
pxt.io.export_parquet(t, 'data.parquet')
pytorch_ds = t.to_pytorch_dataset('pt')  # PyTorch DataLoader ready
coco_path = t.to_coco_dataset()  # COCO annotations

CSV import · Hugging Face · PyTorch export · ML data wrangling

create_view() with iterators splits documents into chunks, video into frames, audio into segments, and typed JSON lists into rows. Not FFmpeg/spaCy pipelines with child tables and foreign keys. For custom explode logic, use @pxt.iterator.

from pixeltable.functions.document import document_splitter
from pixeltable.functions.json import list_iterator
from pixeltable.functions.video import frame_iterator

# Document chunking with overlap
chunks = pxt.create_view(
    'chunks',
    docs,
    iterator=document_splitter(
        document=docs.doc,
        separators='sentence,token_limit',
        overlap=50,
        limit=500,
    ),
)

# Video frame extraction
frames = pxt.create_view(
    'frames',
    videos,
    iterator=frame_iterator(video=videos.video, fps=0.5),
)

# JSON list column: one row per element (typed pxt.Json column required)
items = pxt.create_view('items', t, iterator=list_iterator(t.tags))

Views · Iterators · Custom iterators · RAG pipeline

add_computed_column() runs incrementally on new or stale rows only. Built-ins cover media processing, embeddings, and 30+ providers. Not Airflow, full reprocesses, or custom retry glue.

# LLM provider
t.add_computed_column(
    summary=openai.chat_completions(
        messages=[{'role': 'user', 'content': t.text}],
        model='gpt-4o-mini',
    ),
)

# Local model inference
t.add_computed_column(
    classification=huggingface.vit_for_image_classification(t.image),
)

# Multimodal vision
t.add_computed_column(
    description=openai.chat_completions(
        messages=[
            {
                'role': 'user',
                'content': [
                    {'type': 'text', 'text': 'Describe this image'},
                    {'type': 'image_url', 'image_url': t.image},
                ],
            },
        ],
        model='gpt-4o-mini',
    ),
)

Computed columns · Built-ins · AI integrations

@pxt.udf and @pxt.query with parallelize, cache, and retry. Not one-off handlers with no cache or retry.

@pxt.udf
def format_prompt(context: list, question: str) -> str:
    return f'Context: {context}\nQuestion: {question}'


@pxt.query
def search_by_topic(topic: str):
    return t.where(t.category == topic).select(t.title, t.summary)

UDFs · Custom aggregates

add_embedding_index() stays in sync with table data. Combine .similarity() with .where() on metadata in one query — not a separate vector DB plus filter pipeline.

t.add_embedding_index(
    'img',
    embedding=clip.using(model_id='openai/clip-vit-base-patch32'),
)

sim = t.img.similarity(string='cat playing with yarn')
results = (
    t.where(t.category == 'pets')       # metadata filter in the same query
    .order_by(sim, asc=False)
    .select(t.img, t.category, score=sim)
    .limit(10)
    .collect()
)

Embedding indexes · Semantic search · Image search app

.select() and .sample() to test UDFs ephemerally; same expression becomes add_computed_column() when ready. Not notebook experiments rewritten for production.

# Explore: filter, sample, apply UDFs ephemerally
results = (
    t.where(t.score > 0.8)
    .order_by(t.timestamp)
    .select(t.image, score=t.score)
    .limit(10)
    .collect()
)

# Test on a sample (nothing stored, parallelized and cached)
t.sample(5).select(t.text, summary=summarize(t.text)).collect()

# Commit: same expression, full dataset, skips cached rows
t.add_computed_column(summary=summarize(t.text))

Queries & expressions · Iterative workflow

pxt.tools(), invoke_tools(), and MCP: LLMs choose what to invoke and Pixeltable stores results. Not LangChain loops and manual tool wiring.

mcp_tools = pxt.mcp_udfs('http://localhost:8000/mcp')
tools = pxt.tools(get_weather_udf, search_context_query, *mcp_tools)

t.add_computed_column(
    tool_output=invoke_tools(tools, t.llm_tool_choice),
)

Tool calling · Agents & MCP

pxt serve from TOML or FastAPIRouter routes on the same app. Not hand-written FastAPI endpoints for every table operation.

# pyproject.toml
[[tool.pixeltable.service]]
name = "my-service"
modules = ["schema"]

[[tool.pixeltable.service.routes]]
type = "insert"
table = "myapp.docs"
path = "/ingest"
inputs = ["document"]
outputs = ["document", "summary"]

pxt serve my-service

from pixeltable.serving import FastAPIRouter

router = FastAPIRouter(prefix='/api', tags=['data'])
router.add_query_route(path='/search', query=search_documents)
router.add_insert_route(table, path='/upload', uploadfile_inputs=['image'])

CLI serving · Deployment overview

pxt errors and queryable errormsg per cell; pxt dashboard opens a local UI to browse tables, preview media, and trace column lineage. Not log scraping or opaque per-row failures.

pxt errors my_table          # rows where a computed column failed
pxt dashboard                # browse tables, preview media, pipeline graph

Table browser · media lightbox · column lineage · per-column errors · CSV export

CLI · Dashboard

history(), revert(), and snapshot queries for time travel on every insert and schema change. Not DVC, MLflow, and backfill scripts.

t = pxt.get_table('my_table')
t.revert()  # undo last modification
t.history()  # list all versions
snapshot = pxt.get_table('my_table:472')  # query a snapshot

Version control