From pixeltable
Build multimodal AI applications with Pixeltable — declarative tables replace LangChain + pandas + vector DB with one system. Automates chunking, embedding, retrieval, tool-calling agents, and 25+ AI provider integrations (OpenAI, Anthropic, Gemini, etc.) via computed columns that run on insert. Use when building RAG pipelines, processing images/video/audio/documents, orchestrating LLM inference, or deploying agents with persistent memory. Covers incremental computation, version control, similarity search, FastAPI serving, and production patterns. Do NOT use for general Python or direct PostgreSQL administration.
How this skill is triggered — by the user, by Claude, or both
Slash command
/pixeltable:pixeltable-skillThe summary Claude sees in its skill listing — used to decide when to auto-load this skill
Open-source Python library for **declarative multimodal AI data infrastructure** — tables, computed columns, embedding indexes, and 25+ AI providers in one interface.
Open-source Python library for declarative multimodal AI data infrastructure — tables, computed columns, embedding indexes, and 25+ AI providers in one interface.
pip install pixeltable (Python >= 3.10) | Docs | GitHub
Pixeltable replaces the entire imperative stack you're about to reach for. If you find yourself importing any of these, you are off-path:
document_splitter), embedding indexes (add_embedding_index), retrieval (.similarity()), and tool-calling (pxt.tools() + invoke_tools()). There is zero reason to add a framework on top..collect().to_pandas() only for final export/visualization, never as an intermediate.for row in ...: loops calling AI models — wrap the call in a computed column. Pixeltable handles batching, rate limiting, caching, and retry automatically.t.add_embedding_index(col, embedding=fn) and search with .similarity(). The index lives inside the table.while not done: agent loops — model the agent as a table where inserting a row triggers the entire computed-column chain (LLM → tool selection → tool execution → final answer) declaratively.See anti-patterns.md for the full 15-bias reference with wrong/right code examples.
Jump to the right section based on what you're building:
| If the user wants to... | Read |
|---|---|
| Create tables, insert data, query | Core Concepts (below) and core-api.md |
| Add AI-powered columns (summarize, classify, embed) | Computed Columns (below) and providers.md |
| Chunk documents, extract video frames, split audio | Views and Iterators (below) and core-api.md → Views |
| Build semantic search / embedding indexes | Embedding Indexes (below) and core-api.md → Embedding Indexes |
| Build a RAG pipeline | workflows.md → RAG Pipeline |
| Build a tool-calling agent | Tool-Calling Agent Pipeline (below) and workflows.md → Tool-Calling Agent |
| Build an agent with persistent memory | agents-memory-mcp.md — chat history, knowledge bank, user scoping |
| Use MCP tools with an agent | agents-memory-mcp.md → Adding MCP Tools |
Use invoke_tools() with OpenAI, Groq, Gemini, Bedrock | agents-memory-mcp.md → Multi-Provider |
| Build a video RAG agent (video + search + agent) | video-rag-agents.md — dedicated combined recipe |
| Process video (frames, transcription, visual search) | workflows.md → Video Analysis Pipeline |
| Process images (classify, tag, search) | workflows.md → Image Classification and Search |
| Process audio (transcribe, summarize) | workflows.md → Audio Transcription |
| Wrangle data for ML training (label, version, export) | ml-data-pipeline.md — ingest, enrich, snapshot, PyTorch export |
| Export to PyTorch, Parquet, or pandas | ml-data-pipeline.md → Export for Training |
Look up structured data with retrieval_udf | ml-data-pipeline.md → Retrieval UDFs |
| Retry failed computed columns | Error Handling (below) — recompute_columns() |
| Use agentic patterns (chaining, routing, parallelization, eval-optimize) | agentic-patterns.md — 6 patterns + 2 reasoning strategies |
| Run batch processing (ingest, compute, export, exit) | workflows.md → Batch Processing |
| Configure rate limits, media storage, API keys | core-api.md → Configuration |
| Export to CSV, JSON, Parquet, LanceDB | core-api.md → Export |
| Export to SQL databases (Postgres, Snowflake, SQLite) | core-api.md → Export to SQL |
Share tables across teams (publish, replicate) | core-api.md → Data Sharing |
| Compare multiple AI providers | workflows.md → Multi-Provider Comparison |
| Build a FastAPI web app (hand-written endpoints) | workflows.md → FastAPI App Pattern |
| Serve tables/queries via FastAPIRouter (v0.6+) | workflows.md → FastAPIRouter and core-api.md → Serving |
Serve via CLI (pxt serve + TOML config) | core-api.md → pxt serve |
Store media in Pixeltable Cloud (pxtfs://) | core-api.md → Media Destinations |
| Write UDFs or query functions | UDFs / Query Functions (below) and core-api.md → UDFs |
Use pxt.tools() and invoke_tools() for agents | Tool-Calling Agent Pipeline (below) and core-api.md → Tools and Agents |
| Avoid common mistakes (wrong imports, broken schemas, serialization) | Common Pitfalls (below) and core-api.md → Common Pitfalls |
| Understand what NOT to use with Pixeltable (LangChain, pandas, vector DBs) | anti-patterns.md — 15 training-distribution biases with wrong/right code |
| Look up a specific provider's import and output shape | providers.md → Quick Reference |
openai.vision does not exist — use openai.chat_completions with image_url content blockspxt.String before embedding — use .text.astype(pxt.String) on AI function outputs before add_embedding_indexif_exists='ignore' won't fix bugs — if a computed column has wrong logic, you must drop_column() then recreate; re-running is a silent no-opframe_iterator as a function — from pixeltable.functions.video import frame_iterator, NOT from pixeltable.iterators import FrameIteratorstring= keyword in similarity — always t.col.similarity(string=query), not positionalSee Common Pitfalls below for full details and code examples.
Scaffold a complete project from the Starter Kit with uvx pixeltable-new. Run uvx pixeltable-new --list first to see the options available on your installed version, then pick one.
Application templates — a full app (schema + API/UI) for a use case:
uvx pixeltable-new --template knowledge-base my-kb # serving + backend: docs/images/video/audio upload, unified search + RAG Q&A
uvx pixeltable-new --template chat-agent my-agent # serving + backend: persistent agent, durable memory, tool calling, MCP
uvx pixeltable-new --template audio-transcription my-pod # serving + backend: transcription, summarization, semantic search
uvx pixeltable-new --template video-search my-video # serving: frames, transcription, detection, temporal search
uvx pixeltable-new --template media-indexing my-pipe # batch: ingest from S3, process all modalities, export
uvx pixeltable-new --template image-dataset my-dataset # batch: auto-annotate, curate, version, export to PyTorch
uvx pixeltable-new --template full-stack-showcase my-app # serving + backend: complete reference app (Gemini + DETR + Whisper, React UI)
Structural patterns — bare API/pipeline scaffolds (each template builds on one of these):
uvx pixeltable-new myapp # serving (default): declarative API from schema.py
uvx pixeltable-new myapp --backend # FastAPI scaffold (headless): setup_pixeltable.py + main.py
uvx pixeltable-new myapp --batch # batch processing script with export_sql
Pick a fresh directory name (the generator refuses to overwrite an existing one) and follow the Next steps it prints. Template slugs are descriptive use-case names; if --list on your machine shows different names or a --template fetch fails (network, or a version skew between your installed pixeltable-new and the starter kit), use a name from --list or fall back to the structural pattern the template builds on — do NOT retry guessed template names.
import pixeltable as pxt
pxt.create_dir('my_project', if_exists='ignore')
t = pxt.create_table('my_project.documents', {
'title': pxt.String,
'content': pxt.String,
'image': pxt.Image,
'video': pxt.Video,
'audio': pxt.Audio,
'doc': pxt.Document,
'metadata': pxt.Json,
'score': pxt.Float,
'count': pxt.Int,
'is_active': pxt.Bool,
'created_at': pxt.Timestamp,
}, if_exists='ignore')
Available types: String, Int, Float, Bool, Image, Video, Audio, Document, Json, Array, Timestamp, Date, UUID, Binary. Use pxt.Required[pxt.String] for non-nullable.
Use uuid7() for auto-generated primary keys (recommended for production):
from pixeltable.functions.uuid import uuid7
t = pxt.create_table('my_project.items', {
'content': pxt.String,
'uuid': uuid7(), # auto-generated on insert
'timestamp': pxt.Timestamp,
}, primary_key=['uuid'], if_exists='ignore')
t.insert([{'title': 'Doc 1', 'content': 'Hello world', 'score': 0.95}]) # list of dicts
t.insert(title='Doc 2', content='Single row', score=0.75) # keyword syntax
t.insert(source='path/to/data.csv') # from file
Auto-run on insert. Chain AI providers, UDFs, or expressions:
from pixeltable.functions.openai import chat_completions
t.add_computed_column(
summary=chat_completions(
messages=[{'role': 'user', 'content': t.content}],
model='gpt-4o-mini'
).choices[0].message.content,
if_exists='ignore'
)
t.add_computed_column(upper_title=t.title.upper(), if_exists='ignore')
results = t.select(t.title, t.score).collect()
results = t.where(t.score > 0.8).select(t.title, t.content).collect()
results = t.order_by(t.score, asc=False).limit(10).select(t.title).collect()
count = t.count()
df = t.select(t.title, t.score).collect().to_pandas()
items = list(t.select(title=t.title, score=t.score).collect().to_pydantic(MyModel))
Split rows into sub-rows (chunking, frame extraction, audio splitting):
from pixeltable.functions.document import document_splitter
from pixeltable.functions.video import frame_iterator
from pixeltable.functions.string import string_splitter
from pixeltable.functions.audio import audio_splitter
# Chunk documents into 300-token pieces (requires: pip install tiktoken)
chunks = pxt.create_view(
'my_project.doc_chunks', t,
iterator=document_splitter(t.doc, separators='token_limit', limit=300),
if_exists='ignore'
)
# Extract video frames at 1 fps
frames = pxt.create_view(
'my_project.video_frames', t,
iterator=frame_iterator(t.video, fps=1.0),
if_exists='ignore'
)
# Split text into sentences
sentences = pxt.create_view(
'my_project.sentences', t,
iterator=string_splitter(t.content, separators='sentence'),
if_exists='ignore'
)
# Split audio into 30-second chunks
audio_chunks = pxt.create_view(
'my_project.audio_chunks', t,
iterator=audio_splitter(audio=t.audio, duration=30.0),
if_exists='ignore'
)
# Filtered view (no iterator needed)
active = pxt.create_view(
'my_project.active', t.where(t.is_active == True),
if_exists='ignore'
)
from pixeltable.functions.huggingface import clip, sentence_transformer
embed_fn = clip.using(model_id='openai/clip-vit-base-patch32')
t.add_embedding_index('content', embedding=embed_fn, if_exists='ignore')
# Search
sim = t.content.similarity(string='search query')
results = t.order_by(sim, asc=False).limit(5).select(t.title, t.content, sim).collect()
# Image search with text (multimodal CLIP)
sim = t.image.similarity(string='a photo of a cat')
results = t.order_by(sim, asc=False).limit(5).select(t.image, sim).collect()
from pixeltable.functions import image as pxt_image
from pixeltable.functions.video import extract_audio
# Image thumbnails and encoding
t.add_computed_column(
thumbnail=pxt_image.b64_encode(
pxt_image.thumbnail(t.image, size=(320, 320))
),
if_exists='ignore'
)
# Extract audio from video
t.add_computed_column(
audio=extract_audio(t.video, format='mp3'),
if_exists='ignore'
)
@pxt.udf
def clean_text(text: str) -> str:
return text.strip().lower()
@pxt.udf
def safe_length(text: str | None) -> str:
return 0 if text is None else len(text)
t.add_computed_column(cleaned=clean_text(t.content), if_exists='ignore')
@pxt.query
def search_documents(query_text: str, limit: int = 10):
sim = t.content.similarity(string=query_text)
return t.order_by(sim, asc=False).limit(limit).select(t.title, t.content, sim)
results = search_documents('machine learning').collect()
Inserting a row triggers the entire computed column chain automatically.
import pixeltable as pxt
from pixeltable.functions.anthropic import messages, invoke_tools
from datetime import datetime
tools = pxt.tools(web_search, search_documents) # @pxt.udf + @pxt.query
@pxt.udf
def assemble_context(question: str, tool_outputs: list | None, doc_context: list | None) -> str:
tool_str = str(tool_outputs) if tool_outputs else 'N/A'
doc_str = '\n'.join(
f"- {item.get('text', '')}" for item in (doc_context or []) if isinstance(item, dict)
) or 'N/A'
return (f"QUESTION: {question}\n\n"
f"<tool_results>\n{tool_str}\n</tool_results>\n\n"
f"<retrieved_documents>\n{doc_str}\n</retrieved_documents>")
agent = pxt.create_table('my_project.agent', {
'prompt': pxt.String, 'timestamp': pxt.Timestamp,
'system_prompt': pxt.String, 'max_tokens': pxt.Int, 'temperature': pxt.Float,
}, if_exists='ignore')
# LLM selects tools → execute tools → RAG retrieval → assemble → final answer
agent.add_computed_column(initial_response=messages(
model='claude-sonnet-4-20250514',
messages=[{'role': 'user', 'content': [{'type': 'text', 'text': agent.prompt}]}],
tools=tools, tool_choice=tools.choice(required=True),
max_tokens=agent.max_tokens,
model_kwargs={'system': agent.system_prompt, 'temperature': agent.temperature},
), if_exists='ignore')
agent.add_computed_column(tool_output=invoke_tools(tools, agent.initial_response), if_exists='ignore')
agent.add_computed_column(doc_context=search_documents(agent.prompt), if_exists='ignore')
agent.add_computed_column(context=assemble_context(agent.prompt, agent.tool_output, agent.doc_context), if_exists='ignore')
agent.add_computed_column(final_response=messages(
model='claude-sonnet-4-20250514',
messages=[{'role': 'user', 'content': [{'type': 'text', 'text': agent.context}]}],
max_tokens=agent.max_tokens,
model_kwargs={'system': agent.system_prompt, 'temperature': agent.temperature},
), if_exists='ignore')
agent.add_computed_column(answer=agent.final_response.content[0].text, if_exists='ignore')
# Usage
agent.insert([{'prompt': 'What is quantum computing?', 'timestamp': datetime.now(),
'system_prompt': 'You are a helpful assistant.', 'max_tokens': 1024}])
result = agent.where(agent.prompt == 'What is quantum computing?').select(agent.answer).collect()
25+ providers in pixeltable.functions.* — see providers.md → Quick Reference for the full table and examples.
See core-api.md → Import and core-api.md → Export.
CRITICAL: Always use if_exists='ignore' on every create_* and add_* call.
# Fault-tolerant inserts
status = t.insert(rows, on_error='ignore')
# Inspect errors
t.where(t.summary.errortype != None).select(t.title, t.summary.errormsg).collect()
# Retry failed columns
t.recompute_columns(columns=['summary'], where=t.summary.errortype != None)
| # | Wrong | Correct |
|---|---|---|
| 1 | openai.vision(prompt=..., image=t.image) | openai.chat_completions(messages=[{'role':'user','content':[{'type':'text','text':'...'}, {'type':'image_url','image_url':{'url':t.image}}]}], model='gpt-4o-mini').choices[0].message.content |
| 2 | from pixeltable.iterators import FrameIterator | from pixeltable.functions.video import frame_iterator |
| 3 | t.add_embedding_index('transcript', ...) on Json col | Extract .text.astype(pxt.String) first, then index |
| 4 | Fix code + re-run with if_exists='ignore' | Must t.drop_column('col') then recreate — re-run is a no-op |
| 5 | {'type':'image', 'data': t.image} in messages | Use {'type':'image_url', 'image_url':{'url': t.image}} |
| 6 | t.content.similarity(query) (positional) | t.content.similarity(string=query) (keyword) |
| 7 | Schema corruption (IntegrityError) | Try pxt.drop_dir('my_project', force=True) first; last resort (dev only, manual, with backup): upgrade pixeltable, then delete only the ~/.pixeltable directory — never in production |
| 8 | .collect() or pxt.get_table() inside @pxt.query | @pxt.query compiles the body at decoration time with expression placeholders — don't call .collect(), insert(), or reference tables that may not exist. Use a plain def for imperative logic |
| 9 | 'id': pxt.String as primary key | PK columns must be non-nullable. Use pxt.Required[pxt.String] or uuid7() as a computed default |
| 10 | Module-level Table object used in FastAPI endpoint | Table objects are thread-bound. Call pxt.get_table() inside each endpoint function, not at module level |
Full examples in core-api.md → Common Pitfalls.
pxt.list_tables()
t = pxt.get_table('my_project.my_table')
pxt.drop_table('my_project.my_table')
pxt.drop_dir('my_project', force=True)
t.describe()
t.columns()
# Snapshots (point-in-time copy)
snap = pxt.create_snapshot('my_project.snapshot_v1', t, if_exists='ignore')
# Update and delete
t.update({'score': 1.0}, where=t.category == 'important')
t.delete(where=t.is_active == False)
FastAPIRouter (v0.6+) over hand-written endpoints — add_insert_route, add_query_route, add_delete_route generate endpoints from tables and @pxt.query functionsbackground=True on add_insert_route for long-running inserts (returns a job handle, client polls for completion)def not async def (Pixeltable is synchronous)@pxt.udf / @pxt.query, not in endpoint handlersfrom pixeltable.serving import FastAPIRouter
import pixeltable as pxt
router = FastAPIRouter(prefix="/api/data", tags=["data"])
docs = pxt.get_table("app.documents")
router.add_insert_route(docs, path="/upload", uploadfile_inputs=["document"],
inputs=["timestamp"], outputs=["uuid"], background=True)
router.add_delete_route(docs, path="/delete")
@pxt.query
def list_docs():
return docs.select(uuid=docs.uuid, name=docs.document).order_by(docs.timestamp, asc=False)
router.add_query_route(path="/list", query=list_docs, method="get")
Reference: Pixeltable Starter Kit | workflows.md → FastAPIRouter | core-api.md → Serving
serving, backend, batch) + 7 templates (knowledge-base, chat-agent, audio-transcription, video-search, media-indexing, image-dataset, full-stack-showcase); scaffold with uvx pixeltable-new --template <name> my-app| File | Coverage |
|---|---|
| core-api.md | Tables, querying, views, embeddings, UDFs, tools, serving (FastAPIRouter), B-tree indexes, recompute, config, data sharing, SQL export |
| providers.md | Quick-reference table + full examples for all 25+ AI providers |
| workflows.md | RAG, video analysis, image classification, audio, multi-provider, agent, batch processing, FastAPI, FastAPIRouter, export |
| video-rag-agents.md | Video + transcript/frame retrieval + tool-calling agent |
| agents-memory-mcp.md | Agent with persistent memory, MCP integration, multi-provider invoke_tools |
| ml-data-pipeline.md | Ingest, enrich, version, export to PyTorch/Parquet/pandas |
| agentic-patterns.md | 6 architectural patterns + 2 reasoning strategies |
| anti-patterns.md | 15 training-distribution biases LLMs bring; wrong/right code for each |
npx claudepluginhub pixeltable/pixeltable-skill --plugin pixeltableCreates, edits, and optimizes skills for Claude Code, including drafting, evaluating with test prompts, iterating on performance, and improving skill descriptions for better triggering accuracy.