Real-Time AI Voice Agent with RAG

Build a real-time voice AI system with multi-agent routing and RAG pipelines.

Introduction

30 Second Summary

Every time you call a customer service line, an AI voice understands your question, decides where to route it, and responds in natural speech. Behind the scenes, a real-time pipeline is making those split-second decisions.

In this project, you will build a production-grade real-time AI voice agent from scratch. Your system will accept streaming audio, transcribe it, route queries through an intelligent multi-agent pipeline, and stream synthesized speech back to the caller.

What You'll Build

A real-time voice agent built on FastAPI WebSockets that chains Deepgram speech-to-text, LangGraph multi-agent routing, ChromaDB knowledge retrieval, and ElevenLabs text-to-speech into a single low-latency pipeline.

By the end of this project, you'll have:

  • A real-time voice pipeline with Deepgram STT and ElevenLabs TTS, served through a FastAPI WebSocket server.
  • A ChromaDB-backed RAG knowledge base with OpenAI embeddings for enterprise document retrieval.
  • A LangGraph StateGraph with conditional routing between specialized agents, containerized with Docker for deployment.
  • Secret Mission: Extend the agent graph with a CRM lookup agent using structured tool calls.

Are there any prerequisites?

This project assumes you have experience with Python backend development and async programming. You will also need active API keys for Deepgram, ElevenLabs, and OpenAI.

Before We Start

You are building a real-time AI voice agent that demonstrates the core architecture behind every modern enterprise voice AI system. The pattern is simple: audio in, intelligent processing, voice response out.

The specific architecture you are implementing (speech-to-text to multi-agent router to text-to-speech) is the same pattern used by customer service automation platforms processing millions of calls. Your implementation uses LangGraph for explicit, deterministic routing rather than a black-box agent. This is what regulated industries require for auditability.

In this step, get ready to:
  • Create the project directory structure and a Python virtual environment.
  • Install all project dependencies from a pinned requirements file.
  • Configure API keys and verify connectivity to Deepgram, OpenAI, and ElevenLabs.
Create the project structure and virtual environment

Every Python project needs its own isolated environment so that package versions don't conflict with other projects on your machine.

  • Open your terminal. Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Terminal (macOS) or PowerShell (Windows) and press Enter to open it.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir voice-agent
cd voice-agent
  • Create a Python virtual environment by running:
python3 -m venv .venv

What is a virtual environment?

A virtual environment is an isolated Python installation inside your project folder. Packages you install here stay separate from your system Python, preventing version conflicts between projects.

  • Activate the virtual environment by running:

macOS

source .venv/bin/activate

Windows

.venv\Scripts\activate

You should see (.venv) appear at the beginning of your terminal prompt. This confirms the virtual environment is active.

Don't see (.venv) in your prompt?

On Windows, you may need to allow script execution first. Run Set-ExecutionPolicy -Scope CurrentUser -ExecutionPolicy RemoteSigned in PowerShell, then try the activate command again.

On macOS, make sure you typed source before the path. Without it, the shell won't load the activate script.

Help me activate my Python virtual environment.

Install all dependencies from the pinned requirements file

This project uses eleven pinned packages that handle everything from the web server to speech processing to the AI agent graph.

  • Create the requirements.txt file inside your voice-agent folder. In VS Code, press Cmd+Space (macOS) or the Windows key (Windows), type VS Code, and press Enter to open it.
  • Click File in the top menu bar.
  • Click Open Folder.
  • Select the voice-agent folder on your Desktop, then click Open.
  • Create a new file by clicking File then New File.
  • Name the file requirements.txt and press Enter.
  • Paste the following dependency list into requirements.txt:
fastapi==0.137.2
uvicorn[standard]==0.34.0
websockets==15.0.1
python-dotenv==1.1.0
deepgram-sdk==7.3.1
elevenlabs==2.53.0
langgraph==1.2.5
langchain-openai==0.4.3
langchain-core==1.4.0
chromadb==1.5.9
pydantic==2.11.4

What does this file do?

  • fastapi and uvicorn handle the WebSocket server that streams audio in and out.
  • deepgram-sdk transcribes incoming audio to text using Deepgram's Nova-3 model.
  • elevenlabs converts the agent's text response back into spoken audio.
  • langgraph and langchain-openai power the multi-agent routing graph that decides how to handle each query.
  • chromadb stores and retrieves enterprise knowledge documents using vector embeddings.
  • Save requirements.txt by pressing Cmd+S (macOS) or Ctrl+S (Windows).
  • Install all the dependencies by running this command in your terminal:
pip install -r requirements.txt

This will take a minute or two. ChromaDB is a large package, so don't worry if it seems slow.

✔️ I see Successfully installed...

All packages are installed. You're ready to configure your API keys.

ⓧ I see an error

Check that your virtual environment is active. You should see (.venv) in your terminal prompt.

If you see a permissions error, make sure you activated the virtual environment before running pip install. Run the activate command from the previous substep, then try again.

Help me fix my pip install error.

Configure environment variables and verify connectivity

The voice agent needs API keys for three external services: Deepgram for speech-to-text, OpenAI for embeddings and chat, and ElevenLabs for text-to-speech.

  • In VS Code, create a new file called .env in the root of your voice-agent folder.
  • Paste the following template into your .env file:
DEEPGRAM_API_KEY=your-deepgram-api-key-here
OPENAI_API_KEY=your-openai-api-key-here
ELEVENLABS_API_KEY=your-elevenlabs-api-key-here
  • Replace each placeholder value with your actual API key from the respective service dashboard.

Where do I find my API keys?

  • Deepgram: Go to console.deepgram.com, click API Keys in the sidebar, and create a new key.
  • OpenAI: Go to platform.openai.com/api-keys and create a new secret key.
  • ElevenLabs: Go to elevenlabs.io, click your profile icon, then Profile + API key.
  • Save .env by pressing Cmd+S (macOS) or Ctrl+S (Windows).
  • Verify your API keys are set correctly by running this command in your terminal:
python3 -c "
import os
from dotenv import load_dotenv
load_dotenv()
for key in ['DEEPGRAM_API_KEY', 'OPENAI_API_KEY', 'ELEVENLABS_API_KEY']:
    value = os.getenv(key, '')
    if value and 'your-' not in value:
        print(f'  {key}: Ready')
    else:
        print(f'  {key}: NOT SET - update your .env file')
"

You should see all three keys reported as Ready.

Seeing NOT SET for one or more keys?

Double check that your .env file is saved and located in the voice-agent folder (not inside a subfolder). The file name must start with a dot.

Make sure you replaced the placeholder text (your-deepgram-api-key-here, etc.) with your actual API key string. The key should not contain any spaces or quotes.

Help me fix my .env configuration.

✔️ Awesome, I've got everything!

Great. Double check you've saved both files.

ⓧ I'd like to double check the full code

Here is what requirements.txt should look like:

fastapi==0.137.2
uvicorn[standard]==0.34.0
websockets==15.0.1
python-dotenv==1.1.0
deepgram-sdk==7.3.1
elevenlabs==2.53.0
langgraph==1.2.5
langchain-openai==0.4.3
langchain-core==1.4.0
chromadb==1.5.9
pydantic==2.11.4

Here is what .env should look like (with your actual keys replacing the placeholders):

DEEPGRAM_API_KEY=your-deepgram-api-key-here
OPENAI_API_KEY=your-openai-api-key-here
ELEVENLABS_API_KEY=your-elevenlabs-api-key-here

Your development environment is ready. Next up, you'll build a ChromaDB knowledge base that stores enterprise documents as vector embeddings for your RAG agent to query.

Set Up the Project Environment

Every real-time voice agent connects multiple cloud services, from speech recognition to language models to speech synthesis. Before diving into building, you need a solid project foundation with pinned dependencies and secure credential management.

In this step, you'll confirm your project environment is correctly configured and create a reference file that keeps your team aligned on required credentials without exposing real API keys.

In this step, get ready to:
  • Confirm your project structure and virtual environment are active.
  • Verify all pinned dependencies are installed.
  • Create a .env.example file for safe credential documentation.
Confirm your project structure

Your project directory should already contain the essential configuration files. A quick directory listing confirms everything is in place before you start writing application code.

  • In your terminal (with your virtual environment active), confirm your project files by running:
ls

You should see .env and requirements.txt listed in the output.

What should I see?

Your project directory contains two files at this point. The requirements.txt file pins every dependency to an exact version. The .env file holds your real API credentials for Deepgram, OpenAI, and ElevenLabs.

Verify your installed dependencies

This project uses eleven pinned packages that handle everything from the web server to audio processing. Confirming they're installed at the correct versions prevents subtle compatibility issues later.

  • Verify your installed packages by running:
pip list

You should see fastapi, langgraph, deepgram-sdk, elevenlabs, chromadb, and other packages listed with their version numbers.

Why pin exact versions?

The requirements.txt file uses exact version pins (the == syntax) instead of flexible ranges. This guarantees every developer on the team installs identical packages, eliminating "works on my machine" problems.

For a voice pipeline that depends on precise SDK behavior (like Deepgram streaming events and ElevenLabs audio formats), a minor version bump can silently break your application.

Missing packages in the list?

Make sure your virtual environment is active. Look for the (venv) or similar prefix in your terminal prompt. If it's missing, activate the environment and try pip list again.

If packages are still missing, re-run pip install -r requirements.txt from inside your project directory.

Still stuck? Help me troubleshoot missing packages in my virtual environment.

Create the .env.example reference file

Your .env file contains real API keys that should never be committed to version control. A .env.example file documents which credentials are required without exposing actual values.

  • Create a new file called .env.example in your project directory.
  • Add the following content to .env.example:
DEEPGRAM_API_KEY=your-deepgram-api-key-here
OPENAI_API_KEY=your-openai-api-key-here
ELEVENLABS_API_KEY=your-elevenlabs-api-key-here

Why keep .env.example separate from .env?

The .env.example file is safe to commit to Git because it only contains placeholder values. When a new team member clones the repo, they copy .env.example to .env and fill in their own real keys.

This pattern prevents leaked credentials while still documenting exactly which environment variables the application needs.

  • Save .env.example.
  • Confirm the file exists by running:
ls -a

You should see both .env and .env.example in the output.

Don't see .env.example?

Check that you saved the file in the project root directory (the same folder as requirements.txt). Files starting with a dot are hidden by default. The -a flag in ls -a reveals hidden files.

Need more help? Help me find my .env.example file.

✔️ Awesome, I've got everything!

Great. Double check you've saved .env.example and you're ready to move on.

ⓧ I'd like to double check the full code

Here is what each file in your project should look like at this point.

DEEPGRAM_API_KEY=your-deepgram-api-key-here
OPENAI_API_KEY=your-openai-api-key-here
ELEVENLABS_API_KEY=your-elevenlabs-api-key-here
DEEPGRAM_API_KEY=your-deepgram-api-key-here
OPENAI_API_KEY=your-openai-api-key-here
ELEVENLABS_API_KEY=your-elevenlabs-api-key-here
fastapi==0.137.2
uvicorn[standard]==0.34.0
websockets==15.0.1
python-dotenv==1.1.0
deepgram-sdk==7.3.1
elevenlabs==2.53.0
langgraph==1.2.5
langchain-openai==0.4.3
langchain-core==1.4.0
chromadb==1.5.9
pydantic==2.11.4

Your environment is locked down and documented. Next up, you'll build the ChromaDB-backed knowledge base that gives your voice agent domain expertise.

Build the RAG Knowledge Base

Your voice agent needs access to company knowledge so it can answer questions about pricing, policies, and technical details.

In this step, you will build a RAG (Retrieval-Augmented Generation) knowledge base using ChromaDB as the vector store and OpenAI embeddings to convert text into searchable vectors.

In this step, get ready to:
  • Create sample enterprise documents and ingest them into ChromaDB with OpenAI embeddings.
  • Build a retrieval function that performs similarity search and returns relevant document chunks.
  • Test the knowledge base by querying it and verifying results.
Create the knowledge documents and ingestion logic

A RAG knowledge base stores documents as vector embeddings. When a user asks a question, the system converts the question into an embedding and finds the most similar documents.

  • Create a new file called knowledge_base.py in your project root by running:
touch knowledge_base.py
  • Confirm the file exists by running:
ls

You should see knowledge_base.py listed alongside your requirements.txt and .env files.

  • Open knowledge_base.py in VS Code and add the imports and document definitions by pasting this code:
import chromadb
from langchain_openai import OpenAIEmbeddings
import os
from dotenv import load_dotenv

load_dotenv()

DOCUMENTS = [
    {
        "id": "doc_1",
        "text": "Our enterprise platform offers three pricing tiers: Starter at $49/month for up to 1000 API calls, Professional at $199/month for up to 10000 API calls, and Enterprise with custom pricing for unlimited usage. All plans include 24/7 support.",
        "metadata": {"category": "pricing", "source": "product_catalog"},
    },
    {
        "id": "doc_2",
        "text": "The refund policy allows full refunds within 30 days of purchase. After 30 days, a prorated refund is available for annual subscriptions only. To request a refund, contact support@company.com or call our support line.",
        "metadata": {"category": "policy", "source": "company_policies"},
    },
    {
        "id": "doc_3",
        "text": "Our API rate limits are enforced per API key. Starter plans have a rate limit of 10 requests per second. Professional plans allow 50 requests per second. Enterprise plans have configurable rate limits based on the service agreement.",
        "metadata": {"category": "technical", "source": "api_documentation"},
    },
    {
        "id": "doc_4",
        "text": "To integrate our SDK, install the package using pip install company-sdk. Initialize the client with your API key using client = CompanySDK(api_key='your_key'). The SDK supports Python 3.10 and above.",
        "metadata": {"category": "technical", "source": "sdk_guide"},
    },
    {
        "id": "doc_5",
        "text": "Our uptime SLA guarantees 99.9% availability for Professional and Enterprise plans. Starter plans have a 99.5% availability target. Downtime credits are issued automatically when SLA thresholds are breached.",
        "metadata": {"category": "sla", "source": "service_agreement"},
    },
]

What does this code do?

The DOCUMENTS list simulates an enterprise knowledge base with five documents covering pricing, refund policies, rate limits, SDK integration, and SLA guarantees.

Each document has a unique id, the full text content, and metadata tags so you can filter or trace results back to their source later.

  • Below the DOCUMENTS list, add the embedding helper and the ingestion function:
def get_embedding_function():
    return OpenAIEmbeddings(
        model="text-embedding-3-small",
        api_key=os.getenv("OPENAI_API_KEY"),
    )


def initialize_knowledge_base():
    client = chromadb.PersistentClient(path="./chroma_data")
    embedding_fn = get_embedding_function()

    collection = client.get_or_create_collection(
        name="enterprise_knowledge",
        metadata={"hnsw:space": "cosine"},
    )

    existing_ids = collection.get()["ids"]
    new_docs = [doc for doc in DOCUMENTS if doc["id"] not in existing_ids]

    if new_docs:
        texts = [doc["text"] for doc in new_docs]
        embeddings = embedding_fn.embed_documents(texts)
        collection.add(
            ids=[doc["id"] for doc in new_docs],
            documents=texts,
            embeddings=embeddings,
            metadatas=[doc["metadata"] for doc in new_docs],
        )
        print(f"Added {len(new_docs)} documents to knowledge base.")
    else:
        print("Knowledge base already populated.")

    return collection, embedding_fn

What does this code do?

The get_embedding_function() helper returns an OpenAI embeddings client using the text-embedding-3-small model. This model converts text into 1536-dimensional vectors that capture semantic meaning.

The initialize_knowledge_base() function creates a persistent ChromaDB collection at ./chroma_data using cosine similarity for distance measurement. It checks which documents are already stored and only adds new ones, making it safe to call multiple times.

  • Save knowledge_base.py.

Seeing an import error?

Make sure your virtual environment is activated. You should see the environment name in your terminal prompt.

If chromadb or langchain_openai cannot be found, re-run pip install -r requirements.txt from your project folder.

help me debug an import error in knowledge_base.py

Build the retrieval function

The ingestion function stores documents. Now you need a way to query them. The retrieval function accepts a natural language query, converts it into an embedding, and finds the most similar documents in the collection.

  • In knowledge_base.py, add the query_knowledge_base function below initialize_knowledge_base():
def query_knowledge_base(collection, embedding_fn, query: str, top_k: int = 3):
    query_embedding = embedding_fn.embed_query(query)
    results = collection.query(
        query_embeddings=[query_embedding],
        n_results=top_k,
        include=["documents", "metadatas", "distances"],
    )
    return results

What does this code do?

The query_knowledge_base() function takes a plain-English query, generates its embedding with embed_query(), and performs a nearest-neighbor search against the ChromaDB collection.

The top_k parameter controls how many results to return. The function includes document text, metadata, and cosine distances in the results so the agent can see both the content and how relevant each match is.

  • Save knowledge_base.py.
Test the knowledge base

Before wiring this into the voice agent, you should verify that the knowledge base returns relevant documents for sample queries.

  • Add the following test block at the bottom of knowledge_base.py:
if __name__ == "__main__":
    collection, embedding_fn = initialize_knowledge_base()
    results = query_knowledge_base(collection, embedding_fn, "What are the pricing plans?")
    print("Query results:")
    for doc, meta, dist in zip(
        results["documents"][0], results["metadatas"][0], results["distances"][0]
    ):
        print(f"  [{dist:.4f}] ({meta['category']}) {doc[:80]}...")

What does this code do?

The if __name__ == "__main__" guard means this code only runs when you execute the file directly. It initializes the knowledge base, queries it with a pricing question, and prints the top results with their cosine distances and categories.

  • Save knowledge_base.py and run it by executing:
python knowledge_base.py

You should see output showing that 5 documents were added, followed by query results where the pricing document appears first with the lowest distance score (closest match).

What should I see?

The first line should read Added 5 documents to knowledge base. followed by three results. The pricing document should have the lowest distance value (closest to 0), confirming it is the most relevant match for the query.

If you run the script a second time, it should print Knowledge base already populated. because the documents are persisted to disk.

Seeing an authentication error?

If you see an error mentioning an invalid API key, double check that your .env file has a valid OPENAI_API_KEY value. The embedding function calls the OpenAI API and requires a working key.

help me fix an OpenAI authentication error when running knowledge_base.py

✔️ Awesome, I've got everything!

Great. Double check you have saved knowledge_base.py and confirmed the query results look correct.

ⓧ I'd like to double check the full code

import chromadb
from langchain_openai import OpenAIEmbeddings
import os
from dotenv import load_dotenv

load_dotenv()

DOCUMENTS = [
    {
        "id": "doc_1",
        "text": "Our enterprise platform offers three pricing tiers: Starter at $49/month for up to 1000 API calls, Professional at $199/month for up to 10000 API calls, and Enterprise with custom pricing for unlimited usage. All plans include 24/7 support.",
        "metadata": {"category": "pricing", "source": "product_catalog"},
    },
    {
        "id": "doc_2",
        "text": "The refund policy allows full refunds within 30 days of purchase. After 30 days, a prorated refund is available for annual subscriptions only. To request a refund, contact support@company.com or call our support line.",
        "metadata": {"category": "policy", "source": "company_policies"},
    },
    {
        "id": "doc_3",
        "text": "Our API rate limits are enforced per API key. Starter plans have a rate limit of 10 requests per second. Professional plans allow 50 requests per second. Enterprise plans have configurable rate limits based on the service agreement.",
        "metadata": {"category": "technical", "source": "api_documentation"},
    },
    {
        "id": "doc_4",
        "text": "To integrate our SDK, install the package using pip install company-sdk. Initialize the client with your API key using client = CompanySDK(api_key='your_key'). The SDK supports Python 3.10 and above.",
        "metadata": {"category": "technical", "source": "sdk_guide"},
    },
    {
        "id": "doc_5",
        "text": "Our uptime SLA guarantees 99.9% availability for Professional and Enterprise plans. Starter plans have a 99.5% availability target. Downtime credits are issued automatically when SLA thresholds are breached.",
        "metadata": {"category": "sla", "source": "service_agreement"},
    },
]


def get_embedding_function():
    return OpenAIEmbeddings(
        model="text-embedding-3-small",
        api_key=os.getenv("OPENAI_API_KEY"),
    )


def initialize_knowledge_base():
    client = chromadb.PersistentClient(path="./chroma_data")
    embedding_fn = get_embedding_function()

    collection = client.get_or_create_collection(
        name="enterprise_knowledge",
        metadata={"hnsw:space": "cosine"},
    )

    existing_ids = collection.get()["ids"]
    new_docs = [doc for doc in DOCUMENTS if doc["id"] not in existing_ids]

    if new_docs:
        texts = [doc["text"] for doc in new_docs]
        embeddings = embedding_fn.embed_documents(texts)
        collection.add(
            ids=[doc["id"] for doc in new_docs],
            documents=texts,
            embeddings=embeddings,
            metadatas=[doc["metadata"] for doc in new_docs],
        )
        print(f"Added {len(new_docs)} documents to knowledge base.")
    else:
        print("Knowledge base already populated.")

    return collection, embedding_fn


def query_knowledge_base(collection, embedding_fn, query: str, top_k: int = 3):
    query_embedding = embedding_fn.embed_query(query)
    results = collection.query(
        query_embeddings=[query_embedding],
        n_results=top_k,
        include=["documents", "metadatas", "distances"],
    )
    return results


if __name__ == "__main__":
    collection, embedding_fn = initialize_knowledge_base()
    results = query_knowledge_base(collection, embedding_fn, "What are the pricing plans?")
    print("Query results:")
    for doc, meta, dist in zip(
        results["documents"][0], results["metadatas"][0], results["distances"][0]
    ):
        print(f"  [{dist:.4f}] ({meta['category']}) {doc[:80]}...")

Why cosine similarity?

Cosine similarity measures the angle between two vectors regardless of their magnitude. This makes it ideal for comparing text embeddings because it focuses on semantic direction rather than raw vector length.

A distance of 0 means the texts are semantically identical. The closer to 0, the more relevant the document is to your query.

Your RAG knowledge base is live and returning relevant results. Next up, you will build the multi-agent router that decides whether a user's question should be answered from this knowledge base or handled as general conversation.

Build the Multi-Agent Router with LangGraph

Your knowledge base is loaded and returning relevant documents. Now you need a brain that decides what to do with an incoming query.

In this step, you will build a multi-agent state machine using LangGraph. A classifier node reads the user's input and decides whether it is a knowledge question or general conversation. Based on that classification, the graph routes the query to either a RAG agent (which retrieves context from your ChromaDB knowledge base) or a general conversation agent.

In this step, get ready to:
  • Define an agent state schema and build individual agent node functions.
  • Construct a LangGraph StateGraph with conditional routing edges.
  • Test the graph independently with sample text inputs.
Define the agent state and build the agent nodes

Every LangGraph graph shares a single state object that flows between nodes. Each node reads from the state, does its work, and writes its results back.

  • Create a new file called agent_graph.py in your project folder.
  • Add the imports, LLM initialization, and state schema by pasting this code:
import os
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage
from dotenv import load_dotenv
from knowledge_base import initialize_knowledge_base, query_knowledge_base

load_dotenv()

# Initialize the LLM with temperature=0 for consistent, deterministic routing
llm = ChatOpenAI(
    model="gpt-4o-mini",
    api_key=os.getenv("OPENAI_API_KEY"),
    temperature=0,
)


# Shared state that flows between all nodes in the graph
class AgentState(TypedDict):
    user_input: str
    intent: str
    context: str
    response: str

What does this code do?

The imports pull in LangGraph's graph primitives (StateGraph, START, END) and LangChain's chat message types. The knowledge_base import gives the RAG agent access to the retrieval function you built in the previous step.

Setting temperature=0 on the LLM ensures the intent classifier produces consistent outputs. AgentState is a TypedDict that defines every field the graph nodes can read or write.

  • Save agent_graph.py.

No visible output yet. The next chunk adds the first node function.

  • Below the AgentState class, add the classifier node function:
# Classifier node - uses the LLM to determine query intent
def classify_intent(state: AgentState) -> AgentState:
    messages = [
        SystemMessage(
            content=(
                "You are an intent classifier. Classify the user query into exactly one category.\n"
                "Categories:\n"
                "- 'knowledge': Questions about company products, pricing, policies, technical docs, or SLAs\n"
                "- 'general': General conversation, greetings, or questions unrelated to company knowledge\n\n"
                "Respond with ONLY the category name, nothing else."
            )
        ),
        HumanMessage(content=state["user_input"]),
    ]
    response = llm.invoke(messages)
    intent = response.content.strip().lower()
    # Fall back to general if the LLM returns an unexpected category
    if intent not in ("knowledge", "general"):
        intent = "general"
    return {"intent": intent}

What does this code do?

The classify_intent function is the first node in the graph. It sends the user's input to the LLM with a strict system prompt that forces a one-word classification response.

The fallback logic (if intent not in ("knowledge", "general")) ensures the graph never crashes from an unexpected LLM output. Any unrecognized response defaults to the general conversation path.

  • Save agent_graph.py.

No visible output yet. The next chunks add the routing function and the two agent nodes.

  • Below classify_intent, add the routing function, knowledge base initialization, and RAG agent node:
# Routing function - returns the next node name based on classified intent
def route_by_intent(state: AgentState) -> Literal["rag_agent", "general_agent"]:
    if state["intent"] == "knowledge":
        return "rag_agent"
    return "general_agent"


# Initialize the knowledge base so the RAG agent can query it
collection, embedding_fn = initialize_knowledge_base()


# RAG agent node - retrieves context from ChromaDB and generates a grounded answer
def rag_agent(state: AgentState) -> AgentState:
    results = query_knowledge_base(collection, embedding_fn, state["user_input"], top_k=3)
    context_docs = results["documents"][0] if results["documents"] else []
    context = "\n\n".join(context_docs)

    messages = [
        SystemMessage(
            content=(
                "You are a knowledgeable enterprise assistant. Answer the user's question "
                "based ONLY on the provided context. If the context does not contain the answer, "
                "say you do not have that information. Be concise and professional.\n\n"
                f"Context:\n{context}"
            )
        ),
        HumanMessage(content=state["user_input"]),
    ]
    response = llm.invoke(messages)
    return {"context": context, "response": response.content}

What does this code do?

The route_by_intent function is the conditional router. LangGraph calls it after the classifier runs and uses the returned string to pick the next node.

The collection, embedding_fn = initialize_knowledge_base() line loads your ChromaDB collection at module level so the RAG agent can query it without re-initializing on every call.

The rag_agent function retrieves the top 3 matching documents, injects them into a system prompt as context, and instructs the LLM to answer only from that context.

  • Save agent_graph.py.

You should see Knowledge base already populated. print briefly as the module initializes the ChromaDB collection.

Seeing an import error?

Make sure knowledge_base.py is in the same folder as agent_graph.py. The import from knowledge_base import ... looks for the file in the current directory.

help me fix this import error in agent_graph.py

  • Below rag_agent, add the general conversation agent node:
# General conversation node - handles greetings and non-knowledge queries
def general_agent(state: AgentState) -> AgentState:
    messages = [
        SystemMessage(
            content=(
                "You are a friendly, professional voice assistant. Provide helpful, concise responses "
                "suitable for spoken conversation. Keep responses under 3 sentences."
            )
        ),
        HumanMessage(content=state["user_input"]),
    ]
    response = llm.invoke(messages)
    return {"context": "", "response": response.content}

What does this code do?

The general_agent handles any query that is not about company knowledge. Its system prompt enforces concise, spoken-friendly responses (under 3 sentences) since this will ultimately feed into text-to-speech synthesis.

It returns an empty string for context because no documents were retrieved for this path.

  • Save agent_graph.py.

No visible output yet. You now have all four node functions ready to wire into the graph.

Construct the StateGraph with conditional edges

With the nodes defined, you can now assemble them into a directed graph. LangGraph's add_conditional_edges method connects the classifier output to the correct downstream agent at runtime.

  • Below general_agent, add the graph construction function and compile the agent:
def build_agent_graph() -> StateGraph:
    graph = StateGraph(AgentState)

    # Register each node function with a name
    graph.add_node("classifier", classify_intent)
    graph.add_node("rag_agent", rag_agent)
    graph.add_node("general_agent", general_agent)

    # Wire the graph: START -> classifier -> conditional routing -> END
    graph.add_edge(START, "classifier")
    graph.add_conditional_edges("classifier", route_by_intent)
    graph.add_edge("rag_agent", END)
    graph.add_edge("general_agent", END)

    return graph.compile()


# Compile the graph into a runnable agent
agent = build_agent_graph()

What does this code do?

The build_agent_graph function creates a StateGraph parameterized by AgentState. It registers three nodes, then wires the edges that define execution flow.

The add_conditional_edges("classifier", route_by_intent) call tells LangGraph to run route_by_intent after the classifier finishes. The returned string ("rag_agent" or "general_agent") picks the next node.

Calling .compile() freezes the graph into a runnable object. The module-level agent variable holds the compiled graph so other modules can import and invoke it directly.

Why explicit graph routing instead of a single agent?

A single LLM agent with tools can technically handle both knowledge retrieval and general conversation. But explicit graph routing gives you deterministic control over which code path executes.

In regulated industries (finance, healthcare), you need to prove that certain queries always go through the RAG pipeline and never hallucinate from the base model. A LangGraph StateGraph makes this auditable because the routing logic is inspectable Python code, not a black-box tool-selection decision inside the LLM.

  • Save agent_graph.py.

No visible output yet. The next substep runs the graph with test inputs.

Test the graph with sample queries

Before connecting the graph to voice I/O, you should verify that routing works correctly. A knowledge question should reach the RAG agent, and a greeting should reach the general agent.

  • At the bottom of agent_graph.py, add the test block:
if __name__ == "__main__":
    # Test with a knowledge query - should route to rag_agent
    result = agent.invoke(
        {"user_input": "What pricing plans do you offer?", "intent": "", "context": "", "response": ""}
    )
    print(f"Intent: {result['intent']}")
    print(f"Response: {result['response']}")

    # Test with a general query - should route to general_agent
    result = agent.invoke(
        {"user_input": "Hello, how are you today?", "intent": "", "context": "", "response": ""}
    )
    print(f"Intent: {result['intent']}")
    print(f"Response: {result['response']}")

What does this code do?

The if __name__ == "__main__" block only runs when you execute the file directly. It invokes the compiled graph twice with different inputs.

The first query ("What pricing plans do you offer?") should classify as knowledge and route to the RAG agent, which retrieves pricing documents from ChromaDB. The second query ("Hello, how are you today?") should classify as general and route to the general agent.

  • Save agent_graph.py.
  • Run the test by executing this command:
python agent_graph.py

You should see two sets of output. The first shows Intent: knowledge followed by a response about pricing tiers sourced from your knowledge base. The second shows Intent: general followed by a friendly conversational response.

Seeing an API error or timeout?

Check that your .env file has a valid OPENAI_API_KEY with active billing. The graph makes LLM calls to gpt-4o-mini for both the classifier and the agent responses.

If you see Knowledge base already populated. but then an error, the issue is with the OpenAI key, not ChromaDB.

help me debug this OpenAI API error in my LangGraph agent

✔️ Awesome, I've got everything!

Great. Double check you have saved agent_graph.py and that both test queries returned the correct intent classification.

ⓧ I'd like to double check the full code

import os
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage
from dotenv import load_dotenv
from knowledge_base import initialize_knowledge_base, query_knowledge_base

load_dotenv()

llm = ChatOpenAI(
    model="gpt-4o-mini",
    api_key=os.getenv("OPENAI_API_KEY"),
    temperature=0,
)


class AgentState(TypedDict):
    user_input: str
    intent: str
    context: str
    response: str


def classify_intent(state: AgentState) -> AgentState:
    messages = [
        SystemMessage(
            content=(
                "You are an intent classifier. Classify the user query into exactly one category.\n"
                "Categories:\n"
                "- 'knowledge': Questions about company products, pricing, policies, technical docs, or SLAs\n"
                "- 'general': General conversation, greetings, or questions unrelated to company knowledge\n\n"
                "Respond with ONLY the category name, nothing else."
            )
        ),
        HumanMessage(content=state["user_input"]),
    ]
    response = llm.invoke(messages)
    intent = response.content.strip().lower()
    if intent not in ("knowledge", "general"):
        intent = "general"
    return {"intent": intent}


def route_by_intent(state: AgentState) -> Literal["rag_agent", "general_agent"]:
    if state["intent"] == "knowledge":
        return "rag_agent"
    return "general_agent"


collection, embedding_fn = initialize_knowledge_base()


def rag_agent(state: AgentState) -> AgentState:
    results = query_knowledge_base(collection, embedding_fn, state["user_input"], top_k=3)
    context_docs = results["documents"][0] if results["documents"] else []
    context = "\n\n".join(context_docs)

    messages = [
        SystemMessage(
            content=(
                "You are a knowledgeable enterprise assistant. Answer the user's question "
                "based ONLY on the provided context. If the context does not contain the answer, "
                "say you do not have that information. Be concise and professional.\n\n"
                f"Context:\n{context}"
            )
        ),
        HumanMessage(content=state["user_input"]),
    ]
    response = llm.invoke(messages)
    return {"context": context, "response": response.content}


def general_agent(state: AgentState) -> AgentState:
    messages = [
        SystemMessage(
            content=(
                "You are a friendly, professional voice assistant. Provide helpful, concise responses "
                "suitable for spoken conversation. Keep responses under 3 sentences."
            )
        ),
        HumanMessage(content=state["user_input"]),
    ]
    response = llm.invoke(messages)
    return {"context": "", "response": response.content}


def build_agent_graph() -> StateGraph:
    graph = StateGraph(AgentState)

    graph.add_node("classifier", classify_intent)
    graph.add_node("rag_agent", rag_agent)
    graph.add_node("general_agent", general_agent)

    graph.add_edge(START, "classifier")
    graph.add_conditional_edges("classifier", route_by_intent)
    graph.add_edge("rag_agent", END)
    graph.add_edge("general_agent", END)

    return graph.compile()


agent = build_agent_graph()


if __name__ == "__main__":
    result = agent.invoke(
        {"user_input": "What pricing plans do you offer?", "intent": "", "context": "", "response": ""}
    )
    print(f"Intent: {result['intent']}")
    print(f"Response: {result['response']}")

    result = agent.invoke(
        {"user_input": "Hello, how are you today?", "intent": "", "context": "", "response": ""}
    )
    print(f"Intent: {result['intent']}")
    print(f"Response: {result['response']}")

Your multi-agent router is working. The graph correctly classifies incoming text and routes it to the right specialist. Next up, you will connect this graph to real-time audio by wiring Deepgram speech-to-text and ElevenLabs text-to-speech into a live WebSocket pipeline.

Build the Real-Time Voice Pipeline

Your multi-agent router can classify intent and generate text responses. Now you need to give it a voice.

In this step, you will build the real-time audio pipeline that connects speech recognition to your LangGraph agent and streams synthesized speech back to the client. This is the core integration layer that turns text-based AI into a live voice experience.

In this step, get ready to:
  • Build a Deepgram STT streaming client using the Nova-3 model.
  • Build an ElevenLabs TTS streaming client for speech synthesis.
  • Wire the full pipeline into a FastAPI WebSocket endpoint.
Implement the Deepgram STT streaming client

The speech-to-text layer is what converts raw audio bytes into text your agent can process. You will use Deepgram's Nova-3 model with WebSocket streaming, configured with endpointing to detect natural pauses in conversation.

  • Create a new file called voice_pipeline.py in your project folder.
  • Add the following code to set up the Deepgram and ElevenLabs clients, plus the transcription function:
import os
import asyncio
import json
import base64
from dotenv import load_dotenv
from deepgram import DeepgramClient, LiveOptions
from elevenlabs import ElevenLabs

load_dotenv()

deepgram_client = DeepgramClient(api_key=os.getenv("DEEPGRAM_API_KEY"))
elevenlabs_client = ElevenLabs(api_key=os.getenv("ELEVENLABS_API_KEY"))


async def transcribe_audio(audio_chunks: asyncio.Queue, transcript_callback):
    connection = deepgram_client.listen.live.v("1")

    async def on_message(self, result, **kwargs):
        transcript = result.channel.alternatives[0].transcript
        if transcript and result.is_final:
            await transcript_callback(transcript)

    connection.on("Results", on_message)

    options = LiveOptions(
        model="nova-3",
        language="en-US",
        encoding="linear16",
        sample_rate=16000,
        channels=1,
        interim_results=False,
        endpointing=300,
        smart_format=True,
    )

    await connection.start(options)

    try:
        while True:
            chunk = await audio_chunks.get()
            if chunk is None:
                break
            await connection.send(chunk)
    finally:
        await connection.finish()

What does this code do?

The module sets up two API clients at the top level so they are ready the moment this file is imported. The transcribe_audio function opens a live WebSocket connection to Deepgram's Nova-3 speech recognition engine.

  • LiveOptions configures the audio format (16-bit linear PCM at 16 kHz) and sets endpointing=300 to wait 300ms of silence before finalizing a transcript. This gives natural turn-taking in conversation.
  • The on_message callback fires every time Deepgram returns a result. It only passes along finalized transcripts (not interim guesses) to the callback function.
  • The while True loop pulls audio chunks from an async queue and streams them to Deepgram until a None sentinel signals the end.
  • You won't see a change yet. The next chunk adds the text-to-speech function that completes this file.
Implement the ElevenLabs TTS streaming client

The text-to-speech layer converts your agent's text response into audible speech. You will use ElevenLabs' eleven_flash_v2_5 model, which is optimized for ultra-low latency streaming.

  • In voice_pipeline.py, add the following function below the transcribe_audio function:
def generate_speech(text: str):
    audio_stream = elevenlabs_client.text_to_speech.convert(
        text=text,
        voice_id="JBFqnCBsd6RMkjVDRZzb",
        model_id="eleven_flash_v2_5",
        output_format="pcm_16000",
    )
    for chunk in audio_stream:
        yield chunk

What does this code do?

The generate_speech function is a Python generator. It calls the ElevenLabs text_to_speech.convert method, which returns an iterable of audio chunks.

  • voice_id="JBFqnCBsd6RMkjVDRZzb" selects a pre-made voice (George). You can swap this for any voice in your ElevenLabs account.
  • output_format="pcm_16000" outputs raw PCM at 16 kHz to match the same sample rate used for STT input. This keeps the audio pipeline consistent.
  • Using yield means audio chunks flow to the caller as they arrive from ElevenLabs, rather than waiting for the entire response to generate.
  • Save voice_pipeline.py.
  • Verify the file has no syntax errors by running:
python -c "import voice_pipeline; print('voice_pipeline loaded successfully')"
  • You should see voice_pipeline loaded successfully printed in the terminal.

Seeing an import error?

  • Make sure your virtual environment is activated. The deepgram-sdk and elevenlabs packages must be installed in the active environment.
  • Check that your .env file is in the same directory as voice_pipeline.py. The load_dotenv() call loads environment variables from this file.

Why is my voice_pipeline import failing?

✔️ Awesome, I've got everything!

Great. Double check you have saved voice_pipeline.py before continuing.

ⓧ I'd like to double check the full code

import os
import asyncio
import json
import base64
from dotenv import load_dotenv
from deepgram import DeepgramClient, LiveOptions
from elevenlabs import ElevenLabs

load_dotenv()

deepgram_client = DeepgramClient(api_key=os.getenv("DEEPGRAM_API_KEY"))
elevenlabs_client = ElevenLabs(api_key=os.getenv("ELEVENLABS_API_KEY"))


async def transcribe_audio(audio_chunks: asyncio.Queue, transcript_callback):
    connection = deepgram_client.listen.live.v("1")

    async def on_message(self, result, **kwargs):
        transcript = result.channel.alternatives[0].transcript
        if transcript and result.is_final:
            await transcript_callback(transcript)

    connection.on("Results", on_message)

    options = LiveOptions(
        model="nova-3",
        language="en-US",
        encoding="linear16",
        sample_rate=16000,
        channels=1,
        interim_results=False,
        endpointing=300,
        smart_format=True,
    )

    await connection.start(options)

    try:
        while True:
            chunk = await audio_chunks.get()
            if chunk is None:
                break
            await connection.send(chunk)
    finally:
        await connection.finish()


def generate_speech(text: str):
    audio_stream = elevenlabs_client.text_to_speech.convert(
        text=text,
        voice_id="JBFqnCBsd6RMkjVDRZzb",
        model_id="eleven_flash_v2_5",
        output_format="pcm_16000",
    )
    for chunk in audio_stream:
        yield chunk
Wire the complete pipeline into a FastAPI WebSocket endpoint

Now you will build the FastAPI server that ties everything together. A single WebSocket endpoint will accept audio from a client, transcribe it, route the text through your LangGraph agent, and stream synthesized speech back.

  • Create a new file called main.py in your project folder.
  • Add the application setup and health check endpoint by pasting this code:
import os
import asyncio
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from dotenv import load_dotenv
from agent_graph import agent
from voice_pipeline import deepgram_client, elevenlabs_client, generate_speech
from deepgram import LiveOptions

load_dotenv()

app = FastAPI(title="Real-Time AI Voice Agent")


@app.get("/health")
async def health_check():
    return {"status": "healthy", "service": "voice-agent"}

What does this code do?

  • The imports pull in the compiled agent graph from agent_graph.py and the STT/TTS clients from voice_pipeline.py.
  • The /health endpoint returns a simple JSON response. You will use this to confirm the server is running before testing WebSocket connections.
  • Below the health check endpoint, add the WebSocket handler that wires the full audio pipeline:
@app.websocket("/ws/audio")
async def audio_websocket(websocket: WebSocket):
    await websocket.accept()
    print("Client connected to /ws/audio")

    transcript_queue = asyncio.Queue()
    audio_in_queue = asyncio.Queue()

    try:
        connection = deepgram_client.listen.live.v("1")

        async def on_transcript(self, result, **kwargs):
            transcript = result.channel.alternatives[0].transcript
            if transcript and result.is_final:
                await transcript_queue.put(transcript)

        connection.on("Results", on_transcript)

        options = LiveOptions(
            model="nova-3",
            language="en-US",
            encoding="linear16",
            sample_rate=16000,
            channels=1,
            interim_results=False,
            endpointing=300,
            smart_format=True,
        )

        await connection.start(options)

        async def receive_audio():
            try:
                while True:
                    data = await websocket.receive_bytes()
                    await connection.send(data)
            except WebSocketDisconnect:
                await connection.finish()

        async def process_and_respond():
            try:
                while True:
                    transcript = await transcript_queue.get()
                    print(f"Transcribed: {transcript}")

                    result = await asyncio.to_thread(
                        agent.invoke,
                        {
                            "user_input": transcript,
                            "intent": "",
                            "context": "",
                            "response": "",
                        },
                    )

                    response_text = result["response"]
                    print(f"Agent ({result['intent']}): {response_text}")

                    for audio_chunk in generate_speech(response_text):
                        await websocket.send_bytes(audio_chunk)

            except WebSocketDisconnect:
                pass

        await asyncio.gather(receive_audio(), process_and_respond())

    except WebSocketDisconnect:
        print("Client disconnected")
    except Exception as e:
        print(f"Error: {e}")
    finally:
        print("Connection closed")


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="0.0.0.0", port=8000)

What does the WebSocket handler do?

This is where the entire voice pipeline comes together. Two concurrent tasks run in parallel using asyncio.gather:

  • receive_audio() listens for incoming audio bytes from the WebSocket client and forwards them directly to the Deepgram streaming connection.
  • process_and_respond() waits for finalized transcripts to appear in the queue, invokes the LangGraph agent (wrapped in asyncio.to_thread to avoid blocking the event loop), then streams the synthesized audio response back to the client.
  • The agent invocation uses asyncio.to_thread because LangGraph executes synchronously. This keeps the event loop responsive while the LLM generates a response.
  • Save main.py.
  • Start the server by running:
python main.py
  • You should see output indicating the server is running on http://0.0.0.0:8000.
  • Open a new terminal tab and verify the health endpoint by running:
curl http://localhost:8000/health
  • You should see {"status":"healthy","service":"voice-agent"} returned.

Server not starting?

  • If you see an error about port 8000 already in use, stop the other process or change the port in main.py.
  • If you see an import error for agent_graph, make sure agent_graph.py and knowledge_base.py are in the same directory as main.py.
  • If the knowledge base initialization fails with an OpenAI error, confirm your OPENAI_API_KEY in .env is valid and has credits available.

Why is my FastAPI voice agent server failing to start?

✔️ Awesome, I've got everything!

Great. Double check you have saved main.py and the server is running.

ⓧ I'd like to double check the full code

import os
import asyncio
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from dotenv import load_dotenv
from agent_graph import agent
from voice_pipeline import deepgram_client, elevenlabs_client, generate_speech
from deepgram import LiveOptions

load_dotenv()

app = FastAPI(title="Real-Time AI Voice Agent")


@app.get("/health")
async def health_check():
    return {"status": "healthy", "service": "voice-agent"}


@app.websocket("/ws/audio")
async def audio_websocket(websocket: WebSocket):
    await websocket.accept()
    print("Client connected to /ws/audio")

    transcript_queue = asyncio.Queue()
    audio_in_queue = asyncio.Queue()

    try:
        connection = deepgram_client.listen.live.v("1")

        async def on_transcript(self, result, **kwargs):
            transcript = result.channel.alternatives[0].transcript
            if transcript and result.is_final:
                await transcript_queue.put(transcript)

        connection.on("Results", on_transcript)

        options = LiveOptions(
            model="nova-3",
            language="en-US",
            encoding="linear16",
            sample_rate=16000,
            channels=1,
            interim_results=False,
            endpointing=300,
            smart_format=True,
        )

        await connection.start(options)

        async def receive_audio():
            try:
                while True:
                    data = await websocket.receive_bytes()
                    await connection.send(data)
            except WebSocketDisconnect:
                await connection.finish()

        async def process_and_respond():
            try:
                while True:
                    transcript = await transcript_queue.get()
                    print(f"Transcribed: {transcript}")

                    result = await asyncio.to_thread(
                        agent.invoke,
                        {
                            "user_input": transcript,
                            "intent": "",
                            "context": "",
                            "response": "",
                        },
                    )

                    response_text = result["response"]
                    print(f"Agent ({result['intent']}): {response_text}")

                    for audio_chunk in generate_speech(response_text):
                        await websocket.send_bytes(audio_chunk)

            except WebSocketDisconnect:
                pass

        await asyncio.gather(receive_audio(), process_and_respond())

    except WebSocketDisconnect:
        print("Client disconnected")
    except Exception as e:
        print(f"Error: {e}")
    finally:
        print("Connection closed")


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="0.0.0.0", port=8000)

Your real-time voice pipeline is live. Next up, you will containerize the entire application with Docker so it is ready for production deployment.

Containerize with Docker

Your voice agent is running locally and processing audio end-to-end. The final step is packaging it into a container so it can be deployed to any environment with a single command.

Containerization ensures every dependency, model, and config travels with your application. No more "works on my machine" problems.

In this step, get ready to:
  • Write a multi-stage Dockerfile optimized for production.
  • Create a Docker Compose configuration with health checks and volume mounts.
  • Build and run the containerized voice agent.
Write the Dockerfile

A multi-stage build keeps the final image small by separating the dependency installation step from the runtime image. Only the installed packages carry over to the final stage, leaving behind build caches and temporary files.

  • Create a new file called Dockerfile in the root of your project folder (the same directory as main.py).
  • Add the following multi-stage build configuration:
FROM python:3.12-slim AS builder

WORKDIR /app

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

FROM python:3.12-slim

WORKDIR /app

COPY --from=builder /install /usr/local
COPY . .

EXPOSE 8000

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

What does this Dockerfile do?

The build uses two stages to keep the final image lean.

  • Stage 1 (builder): Installs all Python packages from requirements.txt into a /install prefix. The --no-cache-dir flag prevents pip from storing download caches in the image.
  • Stage 2 (runtime): Starts fresh from the same python:3.12-slim base, copies only the installed packages from the builder, then copies your application code.
  • EXPOSE 8000 documents the port. CMD launches uvicorn pointed at your FastAPI app.
  • Save Dockerfile.

Why a multi-stage build?

ChromaDB alone pulls in hundreds of megabytes of build-time dependencies. A single-stage image would carry all of that dead weight into production.

The multi-stage pattern installs everything in a throwaway stage, then copies only the final installed packages. This can cut your image size by 50% or more.

Create the Docker Compose configuration

Docker Compose lets you define your service, environment, volumes, and health checks in a single declarative file. This is how you will start the container in practice.

  • Create a new file called docker-compose.yml in the same project folder as your Dockerfile.
  • Add the following service definition:
services:
  voice-agent:
    build: .
    ports:
      - "8000:8000"
    env_file:
      - .env
    volumes:
      - ./chroma_data:/app/chroma_data
    healthcheck:
      test: ["CMD", "python", "-c", "import httpx; r = httpx.get('http://localhost:8000/health'); assert r.status_code == 200"]
      interval: 30s
      timeout: 10s
      retries: 3
    restart: unless-stopped

What does this configuration do?

  • build: . tells Compose to build the image from the Dockerfile in the current directory.
  • ports maps container port 8000 to your machine's port 8000 so WebSocket clients can connect.
  • env_file passes your .env API keys into the container without baking them into the image.
  • volumes mounts your local ./chroma_data directory into the container so the knowledge base persists across restarts.
  • healthcheck pings the /health endpoint every 30 seconds. If 3 checks fail, Docker marks the container unhealthy.
  • restart: unless-stopped automatically restarts the container if it crashes.
  • Save docker-compose.yml.
Build and run the container

Before starting the container, you need to stop the local uvicorn process that is currently using port 8000. Then you will use Docker to build and run the application.

  • Switch back to your terminal and press Ctrl+C to stop the running uvicorn process.

Now confirm Docker is installed and running on your machine. Check by running:

docker --version

✔️ I see a version number

Docker is ready. Continue to the next instruction.

ⓧ Command not found

Docker Desktop is not installed on your machine. Follow the official installation guide for your platform:

After installation, open Docker Desktop and wait for the engine to start. Then re-run docker --version to confirm.

  • Make sure Docker Desktop is running (the whale icon in your system tray should be visible).
  • Build and start the container by running:
docker compose up --build

What does this command do?

docker compose up --build does two things. First, it builds the Docker image from your Dockerfile. Then it starts a container from that image with all the port mappings, volumes, and environment variables defined in docker-compose.yml.

The build will take a few minutes the first time as it downloads the python:3.12-slim base image and installs all dependencies. Subsequent builds use Docker's layer cache and are much faster.

  • Wait until you see output similar to:
voice-agent-1  | INFO:     Uvicorn running on http://0.0.0.0:8000
  • Open a new terminal tab and verify the health endpoint by running:
curl http://localhost:8000/health
  • You should see the following JSON response:
{"status":"healthy","service":"voice-agent"}

Container not starting?

  • If you see a port conflict error, make sure you stopped the local uvicorn process with Ctrl+C before running docker compose up.
  • If the build fails on pip install, check that your requirements.txt file is in the same directory as the Dockerfile.
  • If you see API key errors in the container logs, verify your .env file is in the project root and contains valid keys.

Help me debug my Docker container startup issue.

✔️ Awesome, I've got everything!

Great. Make sure Dockerfile and docker-compose.yml are both saved.

ⓧ I'd like to double check the full code

Here is the complete Dockerfile:

FROM python:3.12-slim AS builder

WORKDIR /app

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

FROM python:3.12-slim

WORKDIR /app

COPY --from=builder /install /usr/local
COPY . .

EXPOSE 8000

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

And the complete docker-compose.yml:

services:
  voice-agent:
    build: .
    ports:
      - "8000:8000"
    env_file:
      - .env
    volumes:
      - ./chroma_data:/app/chroma_data
    healthcheck:
      test: ["CMD", "python", "-c", "import httpx; r = httpx.get('http://localhost:8000/health'); assert r.status_code == 200"]
      interval: 30s
      timeout: 10s
      retries: 3
    restart: unless-stopped

Your voice agent is now containerized and running in Docker. Anyone with Docker installed can clone your project, add their API keys, and run docker compose up --build to have the full voice pipeline running in minutes. Next up is the Secret Mission, where you will extend the agent graph with a CRM lookup agent.

Secret mission

Add a CRM Lookup Agent

Your voice agent currently handles knowledge queries and general conversation. Real enterprise voice systems also need to pull live customer data. Extend your LangGraph to a 3-path router with a CRM lookup agent that uses structured tool calls to retrieve customer information.

Clean Up Your Resources

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely. All three APIs (Deepgram, OpenAI, ElevenLabs) are pay-per-use, so there are no ongoing costs once you stop sending requests.

Resources you used:

  • Docker container and image running on port 8000 (local, no cost).
  • ChromaDB persistent storage directory ./chroma_data (local, no cost).
  • Project directory with Python virtual environment and all source files (local, no cost).
  • API usage during testing: Deepgram, ElevenLabs, and OpenAI (usage-based billing, no ongoing charges).

Keep everything running

No action needed. Choose this if you are still actively building or testing your voice agent.

  • Your Docker container continues running on port 8000.
  • The ChromaDB data persists in ./chroma_data for future queries.
  • API costs only accrue when you actively send audio or invoke the agent.

Pause - I'll come back to this later

Shut down the running container to free up system resources. Your project files and ChromaDB data stay intact for when you return.

  • Stop the Docker container by running this command from your project directory:
docker compose down
  • No ongoing costs apply. All three APIs are pay-per-use, so stopping the container means zero charges.
  • To restart later, run docker compose up --build from the same directory.

Delete - I don't want to use this again

Remove all project resources and start fresh. This removes the Docker image, local data, and all source files.

  • Stop and remove the Docker container by running:
docker compose down
  • Remove the Docker image by running:
docker rmi voice-agent
  • Navigate out of the project directory and delete it by running:
cd ..
rm -rf voice-agent
  • On Windows, use this command instead:
cd ..
rmdir /s /q voice-agent
  • Deactivate and remove the Python virtual environment (if it was outside the project folder) by running:
deactivate

What about my API accounts?

Your Deepgram, OpenAI, and ElevenLabs accounts remain active but incur no charges when idle. All three services bill only for usage. You do not need to delete these accounts unless you want to.

Nice Work!

Nice Work!

Nice work! You just built a production-grade real-time AI voice agent from scratch.

Your system accepts streaming audio over a WebSocket, transcribes it, routes it through an intelligent multi-agent pipeline, and streams synthesized speech back to the client in real time.

You've learned how to:

  • Build a FastAPI WebSocket server with bidirectional audio streaming, integrating Deepgram Nova-3 for speech-to-text and ElevenLabs for ultra-low-latency text-to-speech.
  • Create a ChromaDB-backed RAG knowledge base with OpenAI embeddings for grounded enterprise document retrieval.
  • Implement a LangGraph StateGraph with conditional routing between specialized agents and containerize the entire application with Docker for production deployment.
  • Secret Mission: Extended the agent graph with a CRM lookup agent using structured tool calls and a Pydantic schema for typed data retrieval.

Ready to quiz yourself?