How do I transcribe audio uploads and make them searchable?

Before you start

About 30 min.

• Python 3.11+
• pip install -U 'pixeltable[serve]'
• pxt init in the project root
• ffmpeg
• openai-whisper
• sentence-transformers

1. Store uploads as a typed audio column

Put each file in a table as audio, plus any metadata you will filter on. A path string is not enough: splitting and transcription need a typed column.

import pixeltable as pxt
from pixeltable.functions.audio import audio_splitter
from pixeltable.functions.huggingface import sentence_transformer
from pixeltable.functions.whisper import transcribe

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

class Recordings(TableModel, name='recordings'):
    title: pxt.String
    audio: pxt.Audio

2. Split long files into timestamped segments

audio_splitter emits one row per slice with segment_start, segment_end, and audio_segment. Pass exactly one of duration or max_size. duration=30.0 is a starting slice for local Whisper.

class Segments(
    TableModel,
    name='segments',
    base=Recordings,
    iterator=audio_splitter(audio=Recordings.audio, duration=30.0),
):
    pass

3. Transcribe each segment

Whisper on audio_segment writes the text for that slice. A hosted API is an alternative model choice on the same column (openai.transcriptions). The catalog still holds the file and the timestamps.

class Segments(
    TableModel,
    name='segments',
    base=Recordings,
    iterator=audio_splitter(audio=Recordings.audio, duration=30.0),
):
    transcript = transcribe(audio_segment, model='base').text.astype(pxt.String)

4. Embed the text and rank by meaning

Index the transcript with a sentence embedding model. Return the parent title, segment_start, text, and score. The start time is the second to seek to. Sentence splitting with string_splitter needs spaCy; segment rows are enough for a first index.

class Segments(
    TableModel,
    name='segments',
    base=Recordings,
    iterator=audio_splitter(audio=Recordings.audio, duration=30.0),
):
    transcript = transcribe(audio_segment, model='base').text.astype(pxt.String)
    __indexes__ = [
        pxt.EmbeddingIndex(transcript, string_embed=text_embed),
    ]

@pxt.query
def search_audio(query: str, limit: int = 5):
    sim = Segments.transcript.similarity(string=query)
    return (
        Segments.order_by(sim, asc=False)
        .limit(limit)
        .select(
            title=Recordings.title,
            segment_start=Segments.segment_start,
            segment_end=Segments.segment_end,
            text=Segments.transcript,
            score=sim,
        )
    )

5. Serve upload and search over HTTP

Expose an upload route for new files and a search route for the query. background=True on insert returns a job id and a job_url to poll, which fits a long transcription. Apply the schema, then start the service. A new file transcribes only that file.

from pixeltable.serving import FastAPIRouter

api = FastAPIRouter(name='audio_lib')
api.add_insert_route(
    Recordings,
    path='/recordings',
    inputs=[Recordings.title],
    uploadfile_inputs=[Recordings.audio],
    background=True,
)
api.add_query_route(path='/search', query=search_audio)

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

The stack coding agents usually pick vs one schema

Where Pixeltable fits

Pixeltable is where the transcription call runs and the results live, not a speech-to-text model. Deepgram, AssemblyAI, Gladia, and other hosted STT APIs are fine model choices on a computed column; this page does not rank them or quote their prices. The glue stack is FastAPI upload, MinIO or S3, Celery plus Redis, Whisper, and Postgres plus pgvector. Pick that stack when those services are already the contract. Pick this schema when the file, the timestamped segments, the index, and HTTP should live in one app.py. Pixeltable Cloud is Limited Beta. Speakers are a WhisperX cookbook, not this first index.

import pixeltable as pxt
from pixeltable.functions.audio import audio_splitter
from pixeltable.functions.huggingface import sentence_transformer
from pixeltable.functions.whisper import transcribe
from pixeltable.serving import FastAPIRouter

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

class Recordings(TableModel, name='recordings'):
    title: pxt.String
    audio: pxt.Audio

class Segments(
    TableModel,
    name='segments',
    base=Recordings,
    iterator=audio_splitter(audio=Recordings.audio, duration=30.0),
):
    transcript = transcribe(audio_segment, model='base').text.astype(pxt.String)
    __indexes__ = [
        pxt.EmbeddingIndex(transcript, string_embed=text_embed),
    ]

@pxt.query
def search_audio(query: str, limit: int = 5):
    sim = Segments.transcript.similarity(string=query)
    return (
        Segments.order_by(sim, asc=False)
        .limit(limit)
        .select(
            title=Recordings.title,
            segment_start=Segments.segment_start,
            segment_end=Segments.segment_end,
            text=Segments.transcript,
            score=sim,
        )
    )

api = FastAPIRouter(name='audio_lib')
api.add_insert_route(
    Recordings,
    path='/recordings',
    inputs=[Recordings.title],
    uploadfile_inputs=[Recordings.audio],
    background=True,
)
api.add_query_route(path='/search', query=search_audio)

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

Questions

Do I need Celery and Redis?

You need something that runs transcription when a file arrives. Computed columns and the splitter view do that on insert. Celery plus Redis is optional when that queue is already the contract. background=True on the insert route returns a job id so the HTTP caller does not wait on Whisper.

Local Whisper or a transcription API?

Local whisper.transcribe(model=base) needs no API key and runs on CPU or CUDA. openai.transcriptions is the hosted OpenAI API on the same column shape. Other STT vendors are also model choices; this page does not rank them. Pick local when the file must stay on the box; pick an API when you already pay for that model.

How do I get timestamps for a search hit?

Each splitter row keeps segment_start and segment_end in seconds. Return them from the query and seek the parent file to segment_start. A filename alone is not a timestamp.

What about speakers?

This first index is one transcript string per segment, not a speaker label. WhisperX diarization is a separate cookbook: https://docs.pixeltable.com/howto/cookbooks/audio/whisperx-transcription

What happens if a transcription call fails?

Python insert defaults to on_error='abort': the operation fails and the rows are not inserted. on_error='ignore' stores the row with None in the failed cell and records errortype and errormsg (operations docs). Service insert routes do not take on_error; they abort on a computed-column failure.