Heading 1

Heading 2

Heading 3

Heading 4

Heading 5
Heading 6

Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed do eiusmod tempor incididunt ut labore et dolore magna aliqua. Ut enim ad minim veniam, quis nostrud exercitation ullamco laboris nisi ut aliquip ex ea commodo consequat. Duis aute irure dolor in reprehenderit in voluptate velit esse cillum dolore eu fugiat nulla pariatur.

Block quote

Ordered list

  1. Item 1
  2. Item 2
  3. Item 3

Unordered list

Text link

Bold text

Emphasis

Superscript

Subscript

Read more

Speech-To-Text

Building note-taker pipelines in Python: async transcription, LLM integration, and production deployment

Building note-taker pipelines in Python requires async transcription, LLM integration, and production-ready architecture patterns.

Speech-To-Text

Best Google Meet transcription tools and APIs: comparison and selection criteria

Compare Google Meet transcription tools and APIs for product teams. Evaluate WER, latency, pricing at scale, and bot-free capture.

Speech-To-Text

Code-switching detection: how to identify mixed-language speech automatically

Code-switching detection identifies language changes in speech automatically, enabling ASR systems to handle mixed-language audio accurately.

Building note-taker pipelines in Python: async transcription, LLM integration, and production deployment

Published on April 7, 2026
By Ani Ghazaryan
Building note-taker pipelines in Python: async transcription, LLM integration, and production deployment

Building note-taker pipelines in Python requires async transcription, LLM integration, and production-ready architecture patterns.

TL;DR: A production-ready Python note-taker pipeline has four layers: async audio ingestion, managed STT transcription, structured LLM extraction, and validated output storage. Self-hosting Whisper adds GPU infrastructure costs and engineering toil that rarely justify the savings. Gladia async API processes one hour of audio in approximately 60 seconds across 100+ languages, with diarization, code-switching detection, and NER included at the base hourly rate. Pair our transcription output with Pydantic models and OpenAI's structured outputs to get clean, database-ready meeting notes from every recording.

Most Python note-taker tutorials stop at a basic API call. They don't show you how to handle concurrent async requests, LLM context limits, or the reality of accented, multilingual audio in production. This guide covers the complete architecture: audio ingestion, async transcription with Gladia, structured LLM extraction with Pydantic, error handling with dead letter queues (DLQs), and cost modeling that holds up at 1,000 hours a month.

Note-taker pipeline: core architectural flow

Before writing a single line of Python, settle the build-vs-buy question for STT infrastructure, because the answer affects every architectural decision downstream.

Self-hosting Whisper looks free until you price it honestly. A production Whisper deployment requires a strong GPU (an NVIDIA A100 costs $8,000 to $15,000 new), cloud GPU rental at $150 to $400 per month, and the engineering time to manage provisioning, scaling, and model updates. Gladia's own analysis of Whisper infrastructure costs details how these expenses compound. Add two senior software engineers at approximately $143,000 each annually in the US and the total cost of ownership becomes a budget conversation, not an engineering one.

The alternative is a managed STT API where you pay per audio hour and get diarization, translation, and NER as part of that rate. You trade a vendor dependency for freed sprint capacity.

The core pipeline components are:

  1. Audio input layer: File upload, URL fetch, or API ingestion
  2. Async transcription layer: Submit to Gladia, receive structured JSON via webhook or polling
  3. LLM processing layer: Format diarized transcript, extract structured notes
  4. Output validation layer: Pydantic models, JSON schema enforcement
  5. Storage and delivery layer: Database write, CRM sync, or notification

Audio input for note-taker systems

The input method you choose determines your file handling logic and the concurrency model you need, so evaluate these four options before committing:

Input method Pros Cons
CLI (local file path)Simple to test locallyRequires API key auth, no concurrency, manual only
GUI (file picker)Good for end-user toolsRequires frontend, out of scope for API services
API (REST endpoint)Scales to production, supports automationRequires file size validation and rate limiting
Voice (live capture)Real-time use casesRequires WebSocket and real-time architecture

For a meeting note-taker serving multiple users, an API-based input endpoint with async task submission is the right default. Our pre-recorded transcription endpoint accepts WAV, M4A, FLAC, and AAC files up to 1,000 MB and 135 minutes, which covers the vast majority of meeting recordings without file splitting.

Optimizing async transcription latency

Python's asyncio library lets your pipeline handle multiple transcription requests concurrently without blocking on I/O, which matters when our API processes one hour of audio in roughly 60 seconds. Combined with non-blocking async patterns, you can submit dozens of jobs concurrently and process results as they arrive rather than waiting for each one sequentially.

Essential LLM pipeline steps

Transcription gives you raw words. LLMs give you meaning. Keep the two steps distinct in your architecture so you can swap LLM providers or update prompts without re-transcribing audio.

One distinction worth building into your monitoring queries: text-based sentiment inference (NLP classification of transcript polarity based on word choice and syntax) is separate from acoustic emotion detection, which requires analyzing the audio waveform directly. Don't conflate the two when designing pipeline-stage metrics.

What you actually want from the LLM layer:

  • Summary (2 to 3 sentences describing purpose and outcome)
  • Action items with owners and due dates
  • Key decisions that affect product or strategy
  • Speaker attributions where diarization has labeled them

The audio intelligence features (summarization, NER, chapterization) give you a convenience layer on top of the transcript. For CRM syncing and structured database writes, you want your own Pydantic models backed by LLM structured outputs. Summarization quality is ceiling-bounded by transcript quality, so accurate transcription comes first.

Transforming LLM text to structured data

Raw LLM text is not database-ready. A string like "Alice will update the design docs by Friday" can't be queried by due date, assigned to a user record, or synced to a CRM without parsing it into typed fields, and that parsing fails silently when the LLM changes its output format. The pattern is: Gladia returns a diarized transcript as structured JSON, you format that into an LLM prompt, the LLM returns a JSON object defined by your Pydantic schema, and Pydantic validates it before it touches your database. The audio-to-LLM documentation shows how our transcript format maps into this pattern.

Gladia API: fast, scalable audio-to-text

Solaria-1 is evaluated in Gladia’s published benchmark methodology across 8 providers, 7 datasets, and 74+ hours of multilingual audio covering diverse languages, accents, and audio conditions. The benchmark shows where Gladia leads, including up to 3x lower DER and 29% lower WER on conversational speech than alternatives. Solaria-1 also covers 100+ languages, including 42 exclusive to the platform, such as Bengali, Punjabi, Tamil, Urdu, Persian, Marathi, Haitian Creole, and Maori. For meeting note-takers serving global teams, that combination of benchmarked accuracy and broad language coverage is what reduces support tickets from multilingual users.

"We have tested it across many many languages (we work with commentators in pro sports around the world) and have found great accuracy even with custom fields such as team names, player names, etc. We have never come across any sort of hallucination." - Verified user review of Gladia

See the Gladia benchmarks page for the full methodology and dataset breakdown.

Secure Gladia API authentication

Store your credentials in a .env file and load them with python-dotenv:

# .env
GLADIA_API_KEY=your-gladia-key-here
OPENAI_API_KEY=your-openai-key-here
DATABASE_URL=postgresql://localhost/notes
import os
from dotenv import load_dotenv

load_dotenv()

gladia_key = os.getenv("GLADIA_API_KEY")
openai_key = os.getenv("OPENAI_API_KEY")

Install with pip install --no-cache-dir python-dotenv. Add .env to .gitignore before your first commit.

Ingesting audio for transcription

Gladia accepts both local file uploads and remote URLs. The URL path is more efficient for large files already in cloud storage.

import httpx
import os
from dotenv import load_dotenv

load_dotenv()
GLADIA_KEY = os.getenv("GLADIA_API_KEY")
async def upload_audio_file(file_path: str) -> str:
    """Upload a local audio file and return the Gladia audio URL.

    Note: Verify the upload endpoint URL in Gladia's current API documentation.
    """
    async with httpx.AsyncClient() as client:
        with open(file_path, "rb") as f:
            response = await client.post(
                "<https://api.gladia.io/v2/upload>",  # Check docs for current endpoint
                headers={"x-gladia-key": GLADIA_KEY},
                files={"audio": f},
            )
        response.raise_for_status()
        return response.json()["audio_url"]

Creating batch transcription tasks

Once you have an audio URL, submit a transcription task with the features you need. We include diarization, code-switching detection, and NER in the base hourly rate, so you enable them without changing your billing tier.

async def create_transcription_task(audio_url: str) -> str:
    payload = {
        "audio_url": audio_url,
        "diarization": True,
        "code_switching": True,
        "named_entity_recognition": True,
        "sentiment_analysis": True,
    }

    async with httpx.AsyncClient() as client:
        response = await client.post(
            "<https://api.gladia.io/v2/pre-recorded>",
            headers={
                "x-gladia-key": GLADIA_KEY,
                "Content-Type": "application/json",
            },
            json=payload,
        )
        response.raise_for_status()
        task_id = response.json()["id"]
        print(f"Transcription task submitted: {task_id}")
        return task_id

See the diarization documentation for the full parameter reference, including the pyannoteAI Precision-2 model that powers speaker attribution.

Async result delivery: polling or webhooks?

For production pipelines, webhooks are the correct architecture. Polling requires your application to repeatedly query the API on an interval, which wastes API calls on jobs that aren't done yet, adds latency, and creates rate-limiting risk as you scale. Webhooks push the completed result to your endpoint the moment transcription finishes.

Here's a minimal FastAPI webhook receiver. Register its public URL in your Gladia task payload using the webhook_url field.

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()

# Production version should validate payload schema and handle parse errors
@app.post("/webhook/gladia")
async def receive_transcription(request: Request):
    """Receive completed transcription from Gladia and trigger LLM processing."""
    payload = await request.json()

    task_id = payload.get("id")
    status = payload.get("status")

    if status == "done":
        transcript_data = payload.get("result", {}).get("transcription")
        await process_transcript(task_id, transcript_data)

    return JSONResponse(status_code=200, content={"received": True})

The concurrency and rate limits documentation covers concurrent connection limits at each pricing tier, which matters when you're modeling how many simultaneous webhook callbacks your endpoint needs to handle.

Implementing async/await patterns in Python

Python's asyncio library lets you write concurrent I/O code without threads or multiprocessing. For a note-taker pipeline where a 30-minute transcription job spends roughly 30 seconds in Gladia processing and the rest waiting on network I/O, asyncio is the right concurrency model because it lets you handle dozens of jobs concurrently without threading overhead.

Managing concurrent transcription requests

asyncio.gather lets you submit multiple files simultaneously and await all results together. A semaphore bounds how many concurrent requests you send. Check the Gladia documentation for the concurrency limits that apply to your plan before setting this value, as the correct ceiling depends on your tier and is not something you should infer from defaults.

import asyncio

MAX_CONCURRENT = 10  # Adjust based on your plan's concurrency limits
semaphore = asyncio.Semaphore(MAX_CONCURRENT)
async def submit_with_semaphore(file_path: str) -> dict:
"""Submit a single file, respecting the concurrency limit."""
async with semaphore:
audio_url = await upload_audio_file(file_path)
task_id = await create_transcription_task(audio_url)
return {"file": file_path, "task_id": task_id}

async def batch_transcribe(file_paths: list[str]) -> list[dict]:
"""Submit multiple files concurrently with backpressure."""
tasks = [submit_with_semaphore(fp) for fp in file_paths]
return await asyncio.gather(*tasks, return_exceptions=True)

The return_exceptions=True flag means one failed upload doesn't cancel the entire batch. You get exceptions as return values and can route them to your DLQ separately.

Backpressure for stable async pipelines

When you have a backlog of 500 meeting recordings to process, submitting all 500 tasks at once exhausts memory and hits rate limits. The semaphore handles concurrency at the submission layer. For queue-depth control, limit how many unprocessed results sit in memory using asyncio.Queue with a bounded maxsize. The maxsize parameter lets you tune memory footprint based on your transcript size and infrastructure constraints.

async def run_pipeline_with_backpressure(file_paths: list[str]):
    queue = asyncio.Queue(maxsize=50)

    async def producer():
        for fp in file_paths:
            await queue.put(fp)  # Blocks when queue is full
        await queue.put(None)    # Sentinel to stop consumer

    async def consumer():
        while True:
            item = await queue.get()
            if item is None:
                break
            await submit_with_semaphore(item)
            queue.task_done()

    await asyncio.gather(producer(), consumer())

Implementing async retries in Python

Network errors are transient, so use Tenacity for exponential backoff retries before routing to your DLQ.

from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    retry_if_exception_type,
)
import httpx

@retry(
    stop=stop_after_attempt(4),
    wait=wait_exponential(multiplier=1, min=2, max=30),
    retry=retry_if_exception_type(httpx.HTTPStatusError),
)
async def create_transcription_task_with_retry(audio_url: str) -> str:
    return await create_transcription_task(audio_url)

Leveraging LLMs for structured note output

Gladia's transcript JSON gives you word-level timestamps, speaker labels, sentiment scores, and named entities. That's the speech layer. The LLM layer converts the diarized text into a structured meeting notes object.

Refining raw transcription for LLMs

Format our diarized output into a clean speaker-labeled string before passing it to your LLM. This keeps the prompt context tight and avoids sending JSON structure that wastes tokens.

def format_transcript_for_llm(transcription_result: dict) -> str:
    """Convert Gladia's diarized JSON to a speaker-labeled string."""
    utterances = transcription_result.get("utterances", [])
    if not utterances:
        return "(Empty transcript)"
    lines = []
    for u in utterances:
        speaker = u.get("speaker", "Unknown")
        text = u.get("text", "").strip()
        lines.append(f"Speaker {speaker}: {text}")
    return "\\n".join(lines)

Prompt engineering for note summarization

A system prompt that specifies output format, accuracy constraints, and what to do with missing information produces far more consistent results than a vague "summarize this meeting" instruction.

MEETING_NOTES_SYSTEM_PROMPT = """You are an expert meeting assistant tasked with creating clear, actionable meeting notes.

Guidelines:
1. Summary: Provide a concise overview of Include the meeting's stated purpose and outcome or conclusion reached. Extract only information explicitly discussed; exclude inferred or assumed details.
2. Action Items: List specific, measurable tasks. Each must have a clear description, an owner, and a due date.
3. Speakers: List all unique speakers in the transcript.
4. Key Decisions: Extract business decisions that affect product or strategy.
5. Accuracy: Only extract information explicitly stated. Never infer or assume.

Output format: Return ONLY a valid JSON object matching the provided schema, with no additional text.

Important: If the transcript is unclear or insufficient to identify required fields, use null or empty arrays rather than making up information."""

The final instruction matters most. LLMs not told to return null values will invent plausible-sounding data instead.

Choosing LLM latency and cost trade-offs

For a post-meeting async pipeline, latency is less critical than cost and accuracy. Here are the practical trade-offs:

Model Cost per million tokens (in/out) Best for
GPT-4o$2.50 / $10.00Reliable JSON, strong instruction following
Claude 3.5 Sonnet$3.00 / $15.00Long context, comparable accuracy
Llama 3.1 70BVaries (self-hosted)High volume with self-hosting capacity

How to handle LLM context limits

A 2-hour meeting at 150 words per minute produces roughly 18,000 words or around 24,000 tokens. GPT-4o's 128K context window handles this, but you still want a chunking strategy for edge cases.

One approach is to chunk by time segment, extract notes per chunk, then run a final summarization pass over the chunk summaries. This keeps individual LLM calls well within context limits and lets you parallelize chunks with asyncio.gather. Choose chunk size based on your transcript density and target LLM's context window.

def chunk_transcript(utterances: list[dict], window_minutes: int = 15) -> list[list[dict]]:
    """Split utterances into time-based chunks."""
    chunks = []
    current_chunk = []
    chunk_start = 0

    for u in utterances:
        start_seconds = u.get("start", 0)
        if start_seconds - chunk_start > window_minutes * 60 and current_chunk:
            chunks.append(current_chunk)
            current_chunk = []
            chunk_start = start_seconds
        current_chunk.append(u)

    if current_chunk:
        chunks.append(current_chunk)
    return chunks

Building robust pipelines: error detection and monitoring

The most common production failure mode for note-taker pipelines is silent degradation: accuracy drops for a specific language or audio condition, and the engineering team discovers it through support tickets rather than monitoring alerts.

"It's based in EU so it fits our GDPR compliance requirements... The product works great. They're improving the product continuously." - Verified user review of Gladia

Root causes of transcription errors

The three most common sources of transcription errors in production are background noise, heavy accents, and code-switching. Whisper is known to hallucinate on silence or low-signal audio segments, generating text that was never spoken, a failure mode that becomes more frequent as signal quality drops. We address this in Solaria-1 through training on diverse audio conditions, including noisy environments and accented speech. Claap, a meeting intelligence product, achieved 1 to 3% WER in production (97 to 99% word accuracy) and reported that users praised transcription quality directly after switching.

For code-switching, where speakers shift languages mid-sentence, our native code-switching detection handles this across all 100+ supported languages rather than failing silently or returning garbled output.

LLM hallucination detection

Your system prompt instructs the LLM to use null values instead of fabricated data, but you still want a validation layer that catches invented action items. Compare entity names in the LLM output against named entities Gladia extracted from the same transcript. If an action item owner appears in the LLM output but not in Gladia's NER results, flag it for review.

# Note: This flags exact mismatches only. Production version should handle nicknames and partial matches.
def detect_hallucinated_owners(
    meeting_notes: "MeetingNotes",
    ner_entities: list[dict]
) -> list[str]:
    """Flag action item owners not found in the transcript's named entities."""
    known_names = {e["text"].lower() for e in ner_entities if e["type"] == "PERSON"}
    flagged = []
    for item in meeting_notes.action_items:
        if item.owner and item.owner.lower() not in known_names:
            flagged.append(item.owner)
    return flagged

Logs and metrics for error tracing

For async Python pipelines, use structured logging so your log aggregator can query by task ID, pipeline stage, and error type. Pair task IDs from Gladia with your internal job IDs so you can trace a specific recording from submission through LLM output.

import logging
import json

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def log_pipeline_event(stage: str, task_id: str, status: str, details: dict = None):
    logger.info(json.dumps({
        "stage": stage,
        "task_id": task_id,
        "status": status,
        "details": details or {},
    }))

Emit logs to stdout so your container orchestrator (Kubernetes, ECS) can forward them to your logging backend (DataDog, CloudWatch, Splunk) without file management overhead.

Dead letter queues for failed jobs

Any task that exhausts retries goes to a DLQ. Store the original audio URL, the failure reason, and the pipeline stage where it failed so your team can replay it after fixing the root cause. The in-memory asyncio.Queue below is suitable for local development. For production durability, replace it with Redis, RabbitMQ, or a managed queue (AWS SQS, GCP Pub/Sub) so failed jobs survive application restarts.

import asyncio
from dataclasses import dataclass

@dataclass
class FailedJob:
    audio_url: str
    task_id: str | None
    stage: str
    error: str

dlq: asyncio.Queue = asyncio.Queue()

async def send_to_dlq(job: FailedJob):
    await dlq.put(job)
    log_pipeline_event("dlq", job.task_id or "unknown", "failed", {"stage": job.stage, "error": job.error})

Output validation for reliable note generation

Defining JSON schemas for note-taker output

Define your meeting note structure before writing any LLM prompts. The schema drives the prompt, the Pydantic model, and the database migration. This schema handles optional fields by using a type array such as ["string", "null"], which tells validators to accept either the expected type or an explicit null value.

{
  "type": "object",
  "properties": {
    "summary": {"type": "string"},
    "action_items": {
      "type": "array",
      "items": {
        "type": "object",
        "properties": {
          "description": {"type": "string"},
          "owner": {"type": ["string", "null"]},
          "due_date": {"type": ["string", "null"]}
        },
        "required": ["description"]
      }
    },
    "speakers": {"type": "array", "items": {"type": "string"}},
    "key_decisions": {"type": "array", "items": {"type": "string"}},
    "duration_minutes": {"type": "integer"}
  },
  "required": ["summary", "action_items", "speakers", "key_decisions", "duration_minutes"]
}

Ensuring data integrity with Pydantic models

Pydantic v2 validates your LLM output at runtime and raises a clear error if a required field is missing or the wrong type. This is the difference between a silent data corruption bug and an immediately traceable validation failure.

from pydantic import BaseModel, Field
from typing import List, Optional

class ActionItem(BaseModel):
    description: str
    owner: Optional[str] = None
    due_date: Optional[str] = None

class MeetingNotes(BaseModel):
    summary: str = Field(description="Concise 2-3 sentence meeting summary")
    action_items: List[ActionItem] = Field(default_factory=list)
    speakers: List[str]
    key_decisions: List[str] = Field(default_factory=list)
    duration_minutes: int

# Gladia's diarized JSON maps directly into this model after the LLM extraction step
try:
    notes = MeetingNotes.model_validate(llm_json_output)
except ValueError as e:
    await send_to_dlq(FailedJob(
        audio_url=audio_url,
        task_id=task_id,
        stage="pydantic_validation",
        error=str(e)
    ))

Guardrails for LLM output quality

The instructor library wraps the OpenAI and Anthropic clients and enforces structured output against your Pydantic model with automatic retry on validation failure, the right choice if you need cross-provider support or want retry logic handled for you, at the cost of an added dependency. If your pipeline is OpenAI-only and you want to keep the dependency surface minimal, OpenAI's native structured outputs via response_format in the beta.chat.completions.parse endpoint provide the same schema enforcement without a wrapper library.

import instructor
from openai import OpenAI

client = instructor.from_openai(OpenAI())

def extract_meeting_notes(transcript: str) -> MeetingNotes:
    return client.chat.completions.create(
        model="gpt-4o",
        response_model=MeetingNotes,
        messages=[
            {"role": "system", "content": MEETING_NOTES_SYSTEM_PROMPT},
            {"role": "user", "content": f"Extract meeting notes:\\n\\n{transcript}"},
        ],
    )

Scaling your note-taker pipeline efficiently

Containerizing your note-taker pipeline

A minimal Dockerfile for your async Python pipeline:

FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

Your requirements.txt should include httpx, fastapi, uvicorn, pydantic, python-dotenv, instructor, openai, tenacity, and pytest-asyncio.

Managing async worker concurrency

For production workloads with variable queue depth, use Celery or RQ to distribute transcription and LLM tasks across worker processes. This gives you horizontal scaling, task visibility, and built-in retry logic at the infrastructure layer rather than the application layer. A Celery task for the transcription submission step is a standard synchronous function that calls your async code via asyncio.run(), which bridges Celery's sync task model with your async pipeline internals.

Managing AI pipeline costs at scale

The cost difference between all-inclusive pricing and add-on pricing compounds significantly at 1,000 hours a month. Here's an indicative comparison for a pipeline with diarization, NER, and sentiment analysis enabled. Competitor rates are approximate and may not reflect current pricing or your specific usage tier, verify directly with each provider before modeling costs.

Provider Base rate Feature billing Effective rate 1,000 hrs/month
Gladia (Scaling Async)$0.50/hrAll features included$0.50/hr$500
Deepgram$0.26–$0.55/hrAdd-ons per feature~$0.39–$0.46/hr~$390–$462
AssemblyAI$0.15/hrAdd-ons per featureVaries$150+ (base only)

Gladia uses per-hour pricing based on audio duration, which makes total cost straightforward to model at scale. Instead of calculating around block-based billing increments, you can estimate cost directly from the number of audio hours processed and the plan tier you use. See the full Gladia pricing page for tier details, and the Gladia benchmarks for WER comparisons across providers so you can model accuracy-adjusted cost, not just headline rate.

"Gladia AI impresses with its speed and transcription accuracy... you simply upload an audio or video file, and within seconds, you receive a clear, well-organized transcript." - Verified user review of Gladia

Measuring pipeline WER and latency

Don't rely on vendor-reported benchmarks for your specific audio distribution. Build a ground-truth dataset of representative recordings with manually verified transcripts covering at least several hours of audio across the language and accent conditions you serve in production, then compute WER directly.

def compute_wer(reference: str, hypothesis: str) -> float:
    """Compute Word Error Rate between reference and hypothesis strings."""
    ref_words = reference.lower().split()
    hyp_words = hypothesis.lower().split()

    # For datasets over 10,000 words, use python-Levenshtein for 50-100x faster computation
    try:
        import Levenshtein
        distance = Levenshtein.distance(ref_words, hyp_words)
        return distance / len(ref_words)
    except ImportError:
        # Fallback to pure Python implementation
        d = [[0] * (len(hyp_words) + 1) for _ in range(len(ref_words) + 1)]
        for i in range(len(ref_words) + 1):
            d[i][0] = i
        for j in range(len(hyp_words) + 1):
            d[0][j] = j
        for i in range(1, len(ref_words) + 1):
            for j in range(1, len(hyp_words) + 1):
                cost = 0 if ref_words[i - 1] == hyp_words[j - 1] else 1
                d[i][j] = min(d[i - 1][j] + 1, d[i][j - 1] + 1, d[i - 1][j - 1] + cost)
        return d[len(ref_words)][len(hyp_words)] / len(ref_words)

# Install the optimized version with: pip install python-Levenshtein

Setting up local async pipeline tests

Use pytest-asyncio to test your async pipeline functions without needing a live Gladia or OpenAI connection. Mock the external calls so your CI pipeline runs fast and deterministically.

Local testing checklist:

  1. Install pip install pytest pytest-asyncio pytest-mock httpx
  2. Add asyncio_mode = "auto" to your pytest.ini to avoid decorating every test manually
  3. Mock httpx.AsyncClient.post to return a fake Gladia task ID
  4. Mock your LLM client to return a known MeetingNotes fixture
  5. Test Pydantic validation separately with valid and invalid inputs
  6. Run pytest -v and confirm all async tests complete without event loop errors
import pytest
from unittest.mock import MagicMock, patch

`@pytest.mark.asyncio async def test_create_transcription_task(): mock_response = MagicMock() mock_response.json.return_value = {"id": "task-abc-123"} mock_response.raise_for_status = MagicMock()`

with patch("httpx.AsyncClient.post", return_value=mock_response):
        # Replace with your own test audio URL or a mocked value
        task_id = await create_transcription_task("<https://storage.example.com/meeting.wav>")
        assert task_id == "task-abc-123"

The Gladia getting started guide walks through your first API call from zero to a working transcript.

Start with 10 free hours of transcription and run the pipeline code from this guide against your own meeting audio. You'll see how our API handles your specific language distribution, accent profile, and code-switching patterns before committing to a production tier.

FAQs

What is the fastest way to build an AI note taker in Python?

Submit audio to Gladia's async API using httpx, receive the diarized transcript via webhook, then pass it to an LLM with a structured output schema enforced by Pydantic. This gets you from audio file to validated meeting notes JSON in a single pipeline without managing any STT infrastructure.

How do I handle code-switching in multilingual meeting transcriptions?

Enable the code\_switching parameter in your Gladia transcription request and set language\_behaviour to automatic multiple languages. Our Solaria-1 model detects mid-sentence language switches across all 100+ supported languages in both async and real-time modes.

What is the cost of running a note-taker pipeline on 1,000 hours of audio per month?

At Gladia’s current Starter async rate of $0.61/hour, 1,000 hours of transcription costs $610 per month. Growth async pricing starts as low as $0.20/hour, which would put the same 1,000-hour workload at $200 per month. Check the pricing page for the latest tier details and plan-level feature terms before finalizing your cost model.

How do I prevent LLM hallucinations in meeting note generation?

Use the instructor library or OpenAI's native structured outputs to enforce your Pydantic schema at the model level. In your system prompt, explicitly instruct the LLM to return null or empty arrays for missing fields rather than inventing plausible-sounding data, and cross-validate action item owners against named entities from your Gladia transcript.

What are Gladia's file size and duration limits for async transcription?

Our async API accepts audio files up to 1,000 MB and 135 minutes in duration. Supported formats include WAV, M4A, FLAC, and AAC. Files or URLs that exceed these limits need to be split before submission, though the 135-minute limit covers the vast majority of meeting recordings without preprocessing.

How do I test async Python pipeline code without making live API calls?

Use pytest-asyncio with unittest.mock to mock httpx.AsyncClient.post and your LLM client. Set asyncio\_mode = "auto" in pytest.ini, define fixture responses for Gladia task IDs and LLM outputs, and test your Pydantic validation logic independently with both valid and malformed JSON inputs.

Why use webhooks instead of polling for transcription results?

Polling makes repeated API calls that return nothing until the job completes, which wastes rate-limit budget and adds latency. Webhooks push exactly one callback per completed job, which eliminates empty requests entirely and scales linearly with job volume rather than poll frequency times concurrent jobs.

Key terms glossary

WER (word error rate): The percentage of words in a transcript that differ from the reference, calculated as (substitutions + deletions + insertions) / total reference words. Lower is better, and it varies significantly by language and audio condition.

Diarization: The process of segmenting an audio stream by speaker identity, answering "who spoke when." Gladia's implementation uses pyannoteAI's Precision-2 model and runs alongside transcription.

Code-switching: The phenomenon where speakers alternate between two or more languages within a single conversation or sentence. Most STT APIs fail silently on this. Our Solaria-1 detects it natively.

Dead letter queue (DLQ): A message queue that receives tasks that have failed all retry attempts, preserving them for manual inspection and replay rather than dropping them silently.

Backpressure: A flow control mechanism that prevents a fast producer (audio file submissions) from overwhelming a slow consumer (transcription results processing) by bounding the size of the in-flight work queue.

Structured output: An LLM response that conforms to a predefined JSON schema, enforced either by the model's native capability (OpenAI structured outputs) or a wrapper library (instructor). Essential for downstream database writes and CRM syncs.

Asyncio: Python's built-in library for writing concurrent code using the async/await syntax. Ideal for I/O-bound pipelines where the bottleneck is network latency, not CPU computation.

Pydantic: A Python data validation library that enforces type correctness at runtime using standard Python type annotations. The standard choice for validating LLM JSON output before it reaches a database.

Sentiment analysis (text-based): NLP classification of transcript polarity (positive, negative, neutral) based on word choice and syntax, distinct from acoustic emotion detection which analyzes voice characteristics in the raw audio waveform.

Contact us

280
Your request has been registered
A problem occurred while submitting the form.

Read more