Building a Voice Assistant Backend with Pydantic AI: Complete Guide

Introduction to Pydantic AI

Pydantic AI is a relatively new agent framework that brings the type-safety and validation power of Pydantic to the world of LLM-powered applications. It allows developers to build production-grade AI agents with strongly typed inputs and outputs, structured tool calling, and dependency injection — all while keeping the codebase clean and maintainable.

When building a voice assistant, you need more than just a chatbot. You need a system that can accept audio input, transcribe it, reason about the user's intent, optionally call external tools, and respond with synthesized speech. Pydantic AI handles the reasoning layer elegantly, while you wire in speech-to-text (STT) and text-to-speech (TTS) services around it.

Why Pydantic AI for Voice Assistants

Voice assistants differ from text chatbots in several important ways. Latency matters more, responses must be concise enough to speak aloud, and the system must handle ambiguous or partial transcriptions gracefully. Pydantic AI addresses these challenges through several key features:

• Type-safe outputs: You define exactly what the agent should return, so downstream TTS systems always receive well-formed data.
• Structured tool calling: Tools like checking the weather, setting reminders, or querying a database are defined as typed Python functions, reducing hallucinated function calls.
• Dependency injection: External services such as databases, APIs, or user session data can be injected cleanly into agents without global state.
• Multi-model support: You can swap between OpenAI, Anthropic, Gemini, or local models with minimal code changes.
• Streaming support: Partial responses can be streamed, which is critical for reducing perceived latency in voice interactions.

Project Setup and Dependencies

Let us start by setting up a new project. Create a virtual environment and install the required packages. We will use FastAPI for the web layer, Pydantic AI for the agent, and OpenAI's APIs for STT, TTS, and the LLM itself.

python -m venv venv
source venv/bin/activate

pip install pydantic-ai fastapi uvicorn openai python-dotenv httpx

Create a .env file to store your API keys:

OPENAI_API_KEY=sk-your-key-here
MODEL_NAME=gpt-4o

Now create the project structure:

voice-assistant/
├── .env
├── main.py
├── agent.py
├── stt.py
├── tts.py
├── tools.py
└── models.py

Defining the Data Models

One of the core strengths of Pydantic AI is the ability to define structured outputs. For a voice assistant, we want the agent to return a response that includes the spoken text and optionally an action summary. Let us define these models in models.py.

from pydantic import BaseModel, Field
from typing import Optional


class AssistantResponse(BaseModel):
    """The structured response returned by the voice assistant agent."""
    spoken_text: str = Field(
        description="The text to be spoken aloud to the user. Keep it concise and conversational."
    )
    intent: str = Field(
        description="The classified intent of the user's request, e.g. 'weather', 'reminder', 'general_chat'."
    )
    action_taken: Optional[str] = Field(
        default=None,
        description="A description of any action taken, if applicable."
    )
    should_end_conversation: bool = Field(
        default=False,
        description="Whether the conversation should be ended after this response."
    )

By defining AssistantResponse, we guarantee that every agent response will contain a spoken_text field that our TTS layer can consume safely. No more parsing unstructured text or worrying about missing fields.

Building the Speech-to-Text Layer

The STT layer converts incoming audio into text. We will use OpenAI's Whisper API, which accepts audio files and returns transcriptions. In stt.py, create a function that handles this conversion.

import openai
from dotenv import load_dotenv
from fastapi import UploadFile

load_dotenv()


async def transcribe_audio(file: UploadFile) -> str:
    """
    Transcribe an uploaded audio file using OpenAI Whisper.
    
    Args:
        file: An uploaded audio file (wav, mp3, etc.)
    
    Returns:
        The transcribed text.
    """
    audio_bytes = await file.read()

    # Whisper API expects a file-like object with a filename
    result = await openai.Audio.atranscribe(
        model="whisper-1",
        file=("audio.webm", audio_bytes),
    )

    return result["text"].strip()

Note that we use the async variant of the OpenAI client. This is important because our FastAPI endpoints will be async, and we want to avoid blocking the event loop while waiting for the transcription.

Building the Text-to-Speech Layer

The TTS layer converts the agent's text response back into audio. We will use OpenAI's TTS API. In tts.py, implement the synthesis function:

import openai
from dotenv import load_dotenv

load_dotenv()


async def synthesize_speech(text: str, voice: str = "alloy") -> bytes:
    """
    Convert text to speech using OpenAI TTS.
    
    Args:
        text: The text to synthesize.
        voice: The voice to use (alloy, echo, fable, onyx, nova, shimmer).
    
    Returns:
        MP3 audio bytes.
    """
    response = await openai.Audio.aspeech(
        model="tts-1",
        voice=voice,
        input=text,
    )

    return response.content

The tts-1 model is optimized for low latency, which is ideal for voice assistants. If you need higher quality and can afford slightly more latency, you can switch to tts-1-hd.

Creating Tools for the Agent

Tools allow the assistant to perform real actions. Let us define a few practical tools in tools.py. Pydantic AI tools are simply decorated functions with type hints, and they support dependency injection.

from pydantic_ai import RunContext
from datetime import datetime
from typing import Dict, List
import httpx


# A simple in-memory store for reminders (in production, use a database)
reminder_store: Dict[str, List[dict]] = {}


async def get_weather(ctx: RunContext, city: str) -> str:
    """
    Get the current weather for a given city.
    
    Args:
        ctx: The run context (provides dependencies).
        city: The name of the city.
    
    Returns:
        A weather summary string.
    """
    # Using a free weather API as an example
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"https://wttr.in/{city}",
            params={"format": "%C+%t+%w+%h"},
        )
        if response.status_code == 200:
            return f"Weather in {city}: {response.text.strip()}"
        return f"Sorry, I could not retrieve weather data for {city}."


async def add_reminder(ctx: RunContext, task: str, when: str) -> str:
    """
    Add a reminder for the user.
    
    Args:
        ctx: The run context.
        task: The task to remind about.
        when: When to remind (natural language or ISO datetime).
    
    Returns:
        A confirmation message.
    """
    user_id = ctx.deps.get("user_id", "default")
    reminder = {
        "task": task,
        "when": when,
        "created_at": datetime.now().isoformat(),
    }
    reminder_store.setdefault(user_id, []).append(reminder)
    return f"Reminder added: '{task}' scheduled for {when}."


async def list_reminders(ctx: RunContext) -> str:
    """
    List all reminders for the current user.
    
    Returns:
        A formatted list of reminders.
    """
    user_id = ctx.deps.get("user_id", "default")
    reminders = reminder_store.get(user_id, [])
    if not reminders:
        return "You have no reminders."
    lines = [f"- {r['task']} (scheduled for {r['when']})" for r in reminders]
    return "Your reminders:\n" + "\n".join(lines)


async def get_current_time(ctx: RunContext) -> str:
    """
    Get the current date and time.
    
    Returns:
        A formatted datetime string.
    """
    now = datetime.now()
    return now.strftime("It is currently %A, %B %d, at %I:%M %p.")

Each tool receives a RunContext as its first argument, which gives access to dependencies injected at runtime. The remaining parameters are extracted from the LLM's function call and validated by Pydantic.

Building the Pydantic AI Agent

Now we bring everything together in agent.py. We create an agent that uses the tools we defined and returns our structured AssistantResponse.

from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIModel
from dotenv import load_dotenv
from models import AssistantResponse
from tools import get_weather, add_reminder, list_reminders, get_current_time

load_dotenv()

# Define the system prompt that shapes the assistant's behavior
SYSTEM_PROMPT = """\
You are a helpful voice assistant. Follow these rules:

1. Keep responses concise and natural for spoken conversation.
2. Use available tools when the user asks about weather, reminders, or time.
3. Do not use markdown, bullet points, or special formatting in spoken_text.
4. If a tool call fails, apologize and suggest an alternative.
5. Always set the 'intent' field based on the user's request.
6. Set 'should_end_conversation' to True only when the user says goodbye.
"""

# Create the model
model = OpenAIModel("gpt-4o")

# Create the agent with structured output
assistant_agent = Agent(
    model=model,
    result_type=AssistantResponse,
    system_prompt=SYSTEM_PROMPT,
    deps_type=dict,
)

# Register tools
assistant_agent.tool(get_weather)
assistant_agent.tool(add_reminder)
assistant_agent.tool(list_reminders)
assistant_agent.tool(get_current_time)


async def run_assistant(transcript: str, user_id: str = "default") -> AssistantResponse:
    """
    Run the voice assistant agent on a transcribed user message.
    
    Args:
        transcript: The transcribed user speech.
        user_id: The identifier for the current user.
    
    Returns:
        A structured AssistantResponse.
    """
    deps = {"user_id": user_id}

    result = await assistant_agent.run(transcript, deps=deps)
    return result.data

The key design decisions here are worth highlighting. First, result_type=AssistantResponse tells Pydantic AI to enforce that the LLM's output conforms to our schema. Second, deps_type=dict allows us to pass user context into every tool call. Third, each tool is registered with assistant_agent.tool(), which makes it available to the LLM for function calling.

Creating the FastAPI Backend

Now we expose everything through a FastAPI application in main.py. We will create two endpoints: one that accepts audio and returns audio, and one that accepts text and returns text (useful for testing).

from fastapi import FastAPI, UploadFile, File, HTTPException
from fastapi.responses import Response, JSONResponse
from pydantic import BaseModel
from dotenv import load_dotenv
from stt import transcribe_audio
from tts import synthesize_speech
from agent import run_assistant

load_dotenv()

app = FastAPI(title="Voice Assistant API", version="1.0.0")


class TextRequest(BaseModel):
    text: str
    user_id: str = "default"


class TextResponse(BaseModel):
    spoken_text: str
    intent: str
    action_taken: str | None = None
    should_end_conversation: bool = False


@app.post("/chat/text", response_model=TextResponse)
async def chat_text(request: TextRequest):
    """
    Text-based endpoint for testing the assistant without audio.
    """
    try:
        result = await run_assistant(request.text, request.user_id)
        return TextResponse(
            spoken_text=result.spoken_text,
            intent=result.intent,
            action_taken=result.action_taken,
            should_end_conversation=result.should_end_conversation,
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.post("/chat/voice")
async def chat_voice(file: UploadFile = File(...), user_id: str = "default"):
    """
    Full voice pipeline: audio in, audio out.
    
    Accepts an audio file, transcribes it, runs the assistant,
    and returns synthesized speech as MP3.
    """
    try:
        # Step 1: Transcribe audio to text
        transcript = await transcribe_audio(file)

        if not transcript:
            raise HTTPException(status_code=400, detail="Could not transcribe audio.")

        # Step 2: Run the assistant
        result = await run_assistant(transcript, user_id)

        # Step 3: Synthesize speech from the response
        audio_bytes = await synthesize_speech(result.spoken_text)

        # Return audio with metadata in headers
        return Response(
            content=audio_bytes,
            media_type="audio/mpeg",
            headers={
                "X-Transcript": transcript,
                "X-Intent": result.intent,
                "X-End-Conversation": str(result.should_end_conversation),
            },
        )
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.get("/health")
async def health():
    return {"status": "ok"}

The /chat/voice endpoint is the heart of the system. It accepts an audio file upload, runs the full pipeline, and returns MP3 audio bytes. Metadata such as the transcript and intent are passed back in custom headers so the client can display them if needed.

Running and Testing the Server

Start the server with uvicorn:

uvicorn main:app --reload --host 0.0.0.0 --port 8000

Test the text endpoint first with curl:

curl -X POST http://localhost:8000/chat/text \
  -H "Content-Type: application/json" \
  -d '{"text": "What is the weather in Tokyo?", "user_id": "user123"}'

You should receive a JSON response like:

{
  "spoken_text": "The weather in Tokyo is currently partly cloudy with a temperature of 22 degrees Celsius and light winds.",
  "intent": "weather",
  "action_taken": "Retrieved weather data for Tokyo.",
  "should_end_conversation": false
}

To test the voice endpoint, record a short audio clip and send it:

curl -X POST http://localhost:8000/chat/voice \
  -F "file=@recording.webm" \
  -F "user_id=user123" \
  -o response.mp3

The response will be an MP3 file you can play back.

Adding Streaming for Lower Latency

One of the biggest challenges in voice assistants is latency. The user speaks, then waits for transcription, reasoning, and synthesis before hearing a response. Pydantic AI supports streaming, which lets you start TTS as soon as the first words are available.

Here is how to modify the agent to support streaming the spoken text:

from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIModel
from dotenv import load_dotenv
from models import AssistantResponse
from tools import get_weather, add_reminder, list_reminders, get_current_time

load_dotenv()

SYSTEM_PROMPT = """\
You are a helpful voice assistant. Keep responses concise and natural.
Use tools when appropriate. Do not use markdown in spoken_text.
"""

model = OpenAIModel("gpt-4o")

assistant_agent = Agent(
    model=model,
    result_type=AssistantResponse,
    system_prompt=SYSTEM_PROMPT,
    deps_type=dict,
)

assistant_agent.tool(get_weather)
assistant_agent.tool(add_reminder)
assistant_agent.tool(list_reminders)
assistant_agent.tool(get_current_time)


async def stream_assistant_response(transcript: str, user_id: str = "default"):
    """
    Stream the assistant's response token by token.
    
    Yields partial text chunks that can be sent to TTS incrementally.
    """
    deps = {"user_id": user_id}

    async with assistant_agent.run_stream(transcript, deps=deps) as result:
        # Stream partial results
        async for partial in result.stream_text(delta=True):
            yield partial

You can then create a streaming endpoint that sends audio chunks as they are generated. This approach, sometimes called "chunked TTS," significantly reduces the time-to-first-audio for the user.

Handling Conversation History

A voice assistant needs to maintain context across turns. Pydantic AI supports message history natively. Let us add a simple session manager that stores conversation history per user.

from pydantic_ai.messages import ModelMessage
from typing import Dict, List
import json

# In-memory session store (use Redis or a database in production)
session_store: Dict[str, List[ModelMessage]] = {}


async def run_assistant_with_history(
    transcript: str,
    user_id: str = "default",
) -> AssistantResponse:
    """
    Run the assistant while maintaining conversation history.
    """
    from agent import assistant_agent
    from models import AssistantResponse

    deps = {"user_id": user_id}
    history = session_store.get(user_id, [])

    result = await assistant_agent.run(transcript, deps=deps, message_history=history)

    # Store updated history
    session_store[user_id] = result.all_messages()

    return result.data


def clear_session(user_id: str):
    """Clear the conversation history for a user."""
    session_store.pop(user_id, None)

With this in place, the assistant remembers previous turns within a session. You can add a POST /session/clear endpoint to let users reset the conversation.

Best Practices

Optimize for Latency

Latency is the most critical UX factor in voice assistants. Use the following strategies to minimize it:

• Use tts-1 instead of tts-1-hd for faster synthesis.
• Stream responses so TTS can begin before the full response is generated.
• Choose a fast LLM model. gpt-4o-mini is often sufficient for voice interactions and is much faster than gpt-4o.
• Cache tool results when possible. For example, weather data does not change every second.
• Use WebSocket connections instead of HTTP for real-time bidirectional communication.

Handle Errors Gracefully

Voice assistants operate in noisy environments. Transcriptions will be imperfect, and tools will occasionally fail. Always provide fallback responses:

async def run_assistant_safe(transcript: str, user_id: str = "default") -> AssistantResponse:
    """Run the assistant with comprehensive error handling."""
    try:
        result = await run_assistant(transcript, user_id)
        return result
    except Exception as e:
        # Return a safe fallback response
        return AssistantResponse(
            spoken_text="I'm sorry, I encountered an error. Could you repeat that?",
            intent="error",
            action_taken=f"Error: {str(e)}",
            should_end_conversation=False,
        )

Keep Responses Concise

Spoken responses should be shorter than written ones. Emphasize this in your system prompt and consider adding a maximum length constraint:

SYSTEM_PROMPT = """\
You are a helpful voice assistant. Rules:
1. Keep spoken_text under 3 sentences unless the user explicitly asks for detail.
2. Use a conversational, friendly tone.
3. Never use markdown, bullet points, or special characters.
4. Use tools when the user asks about weather, reminders, or time.
"""

Validate and Sanitize Inputs

Always validate transcribed text before passing it to the agent. Empty strings, extremely long inputs, or inputs with injection attempts should be handled:

def validate_transcript(text: str) -> str:
    """Sanitize and validate transcribed text."""
    if not text or len(text.strip()) == 0:
        raise ValueError("Empty transcript received.")
    if len(text) > 5000:
        text = text[:5000]
    return text.strip()

Use Dependency Injection for Testability

Pydantic AI's dependency injection system makes it easy to mock external services in tests. Define a dataclass or Pydantic model for your dependencies:

from dataclasses import dataclass
from typing import Optional


@dataclass
class AssistantDeps:
    user_id: str
    db_session: Optional[object] = None
    api_client: Optional[object] = None

Then use AssistantDeps as your deps_type and pass mock instances during testing.

Log Everything

For debugging and improvement, log transcripts, intents, tool calls, and response times:

import logging
import time

logger = logging.getLogger("voice_assistant")


async def run_assistant_with_logging(transcript: str, user_id: str = "default"):
    start = time.time()
    result = await run_assistant(transcript, user_id)
    elapsed = time.time() - start

    logger.info(
        "Assistant call completed",
        extra={
            "user_id": user_id,
            "transcript": transcript,
            "intent": result.intent,
            "elapsed_seconds": elapsed,
            "action_taken": result.action_taken,
        },
    )
    return result

Deploying to Production

When deploying your voice assistant, consider the following infrastructure choices:

• Containerization: Package the application in a Docker container for consistent deployment.
• WebSocket support: For real-time voice, use WebSockets instead of HTTP to enable full-duplex communication.
• Session storage: Replace the in-memory session store with Redis or a database for persistence and scalability.
• Rate limiting: Protect your endpoints with rate limiting to prevent abuse.
• Monitoring: Use tools like Prometheus and Grafana to track latency, error rates, and usage patterns.

Here is a minimal Dockerfile for the application:

FROM python:3.12-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

And the corresponding requirements.txt:

pydantic-ai>=0.0.30
fastapi>=0.110.0
uvicorn>=0.29.0
openai>=1.30.0
python-dotenv>=1.0.0
httpx>=0.27.0

Conclusion

Building a voice assistant backend with Pydantic AI gives you a robust, type-safe foundation that scales from prototype to production. By combining Pydantic AI's structured outputs and tool-calling capabilities with STT and TTS services, you create a system where every component is predictable, testable, and maintainable. The framework's dependency injection and multi-model support mean you can evolve your assistant over time — swapping models, adding tools, or changing providers — without rewriting your core logic. Start with the text endpoint to validate your agent's behavior, then layer in audio processing, streaming, and session management as your requirements grow. With careful attention to latency, error handling, and response conciseness, you can deliver a voice assistant that feels responsive and natural in real-world use.