Your Backend for Multimodal AI Applications

%pip install -qU pixeltable google-genai 'fastapi[standard]'
%pip install -qU torch torchvision transformers  # optional, for object detection

import getpass
import logging
import os
import warnings

warnings.filterwarnings('ignore')
logging.getLogger('asyncio').setLevel(logging.CRITICAL)
logging.getLogger('huggingface_hub').setLevel(logging.CRITICAL)

if (
    'GEMINI_API_KEY' not in os.environ
    and 'GOOGLE_API_KEY' not in os.environ
):
    os.environ['GEMINI_API_KEY'] = getpass.getpass('Gemini API Key: ')

import pixeltable as pxt
from pixeltable.functions import gemini

BASE_URL = 'https://raw.githubusercontent.com/pixeltable/pixeltable/release/docs/resources'

from pixeltable.functions.uuid import uuid7

pxt.drop_dir('demo', force=True, if_not_exists='ignore')
pxt.create_dir('demo')

videos = pxt.create_table(
    'demo/videos',
    {'id': uuid7(), 'video': pxt.Video, 'title': pxt.String},
    primary_key='id',
)
videos

videos.add_computed_column(
    response=gemini.generate_content(
        [videos.video, 'Describe this video in detail.'],
        model='gemini-3-flash-preview',
    )
)

videos.add_computed_column(
    description=videos.response.candidates[0]
    .content.parts[0]
    .text.astype(pxt.String)
)

videos.add_embedding_index(
    'description',
    embedding=gemini.embed_content.using(
        model='gemini-embedding-2-preview'
    ),
)

videos.insert(
    [
        {
            'video': f'{BASE_URL}/bangkok.mp4',
            'title': 'Bangkok Street Tour',
        },
        {
            'video': f'{BASE_URL}/The-Pursuit-of-Happiness-Video-Extract.mp4',
            'title': 'The Pursuit of Happiness',
        },
    ]
)

videos = pxt.get_table('demo/videos')
videos.select(videos.title, videos.description).collect()

sim = videos.description.similarity(string='street food')

# Filter + rank in one expression
videos.where(videos.description != None).order_by(sim, asc=False).limit(
    5
).select(videos.title, videos.description, sim).collect()

from pixeltable.functions import huggingface
from pixeltable.functions.vision import bboxes_draw

frame = videos.video.extract_frame(timestamp=2.0)
detections = huggingface.detr_for_object_detection(
    frame, model_id='facebook/detr-resnet-50'
)

videos.select(
    videos.title,
    annotated=bboxes_draw(
        frame,
        boxes=detections.boxes,
        labels=detections.label_text,
        fill=True,
        fill_alpha=0.15,
        width=2,
        font_size=14,
    ),
).collect()

videos.describe()

import fastapi
from fastapi.testclient import TestClient
from pixeltable.serving import FastAPIRouter


@pxt.query
def search_videos(query_text: str, limit: int = 5):
    sim = videos.description.similarity(string=query_text)
    return (
        videos.order_by(sim, asc=False)
        .limit(limit)
        .select(videos.title, videos.description, sim)
    )


app = fastapi.FastAPI()
router = FastAPIRouter()
router.add_query_route(path='/search', query=search_videos)
router.add_insert_route(
    videos,
    path='/ingest',
    inputs=['video', 'title'],
    outputs=['id', 'title', 'description'],
)
router.add_delete_route(videos, path='/delete')
app.include_router(router)

client = TestClient(app)

resp = client.post(
    '/search', json={'query_text': 'street food', 'limit': 2}
)
resp.json()

videos.history()

from pixeltable.functions.anthropic import invoke_tools, messages

# 1. Define tools from existing @pxt.query functions
tools = pxt.tools(search_videos)

# 2. Memory: chat history with embedding index
chat = pxt.create_table(
    'demo/chat',
    {
        'role': pxt.String,
        'content': pxt.String,
        'conversation_id': pxt.String,
    },
    if_exists='ignore',
)
chat.add_embedding_index(
    'content',
    string_embed=gemini.embed_content.using(
        model='gemini-embedding-2-preview'
    ),
    if_exists='ignore',
)

# 3. Agent pipeline: each step is a computed column
agent = pxt.create_table(
    'demo/agent', {'prompt': pxt.String}, if_exists='ignore'
)

# Step 1: LLM decides which tools to call
agent.add_computed_column(
    response=messages(
        model='claude-sonnet-4-20250514',
        messages=[{'role': 'user', 'content': agent.prompt}],
        tools=tools,
        tool_choice=tools.choice(required=True),
        max_tokens=4096,
    ),
    if_exists='ignore',
)

# Step 2: Pixeltable executes the tool calls (runs search_videos)
agent.add_computed_column(
    tool_output=invoke_tools(tools, agent.response), if_exists='ignore'
)

# In production, add more steps: assemble context, call LLM again with results.
# See the Starter Kit for the full multi-step agent pipeline.
agent.describe()

# ~/.pixeltable/config.toml
[pixeltable]
api_key = "your-pixeltable-cloud-api-key"
output_media_dest = "pxt://yourorg:yourdb/buckets/home"