Real-Time AI Voice Agent with RAG

Build a voice AI agent that routes speech to RAG or CRM using LangGraph.

Introduction

30 Second Summary

When you call a company's support line and get routed to the right department instantly, that experience is increasingly powered by AI voice agents. These systems listen, understand intent, and respond in real time without a human in the loop.

In this project, you will build a real-time AI voice agent that transcribes speech, routes requests to the correct backend system using a multi-agent state machine, and responds with synthesized audio.

What You'll Build

A production-grade voice AI system for a fictional enterprise support line, using WebSocket streaming, multi-agent routing with LangGraph, and RAG-powered knowledge retrieval to handle spoken requests in real time.

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

  • A FastAPI WebSocket server that accepts audio input, processes it through the full voice pipeline, and streams synthesized speech back to the client.
  • A LangGraph multi-agent state machine that autonomously classifies user intent and routes queries to either a RAG knowledge base or a simulated CRM system.
  • A ChromaDB-backed RAG pipeline with OpenAI embeddings, plus streaming Text-to-Speech via ElevenLabs and Speech-to-Text via Deepgram, all containerized with Docker.
  • Secret Mission: Add conversation memory so the agent remembers previous exchanges within a session, enabling multi-turn contextual interactions.

Are there any prerequisites?

This is an advanced project. You should have Python and Docker installed, along with active accounts (and API keys) for Deepgram, ElevenLabs, and OpenAI.

Familiarity with Python async programming and basic API concepts will help you move through the steps smoothly.

Set Up the Environment and Dependencies

A real-time voice agent depends on multiple libraries working together: audio transcription, language model orchestration, vector search, and speech synthesis. Each library version must be pinned so the stack stays stable.

In this step, you will create the project folder, set up a Python virtual environment, install all pinned dependencies, and configure your API keys.

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

A virtual environment keeps this project's dependencies isolated from your system Python. This prevents version conflicts 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 cmd (Windows), and press Enter.
  • 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
  • Confirm the folder exists by running:
ls

You should see an empty listing (the folder was just created).

  • Create a Python virtual environment inside the project by running:
python3 -m venv venv

What does this command do?

python3 -m venv venv tells Python to create a self-contained environment inside a folder called venv. All packages you install after activation go here instead of your global Python.

  • Activate the virtual environment.

macOS

  • Run this activation command:
source venv/bin/activate

Windows

  • Run this activation command:
venv\Scripts\activate

You should see (venv) appear at the start of your terminal prompt.

Don't see (venv) in your prompt?

Make sure you ran the activation command from inside the voice-agent folder. On Windows, if you get a permissions error, run Set-ExecutionPolicy -Scope CurrentUser RemoteSigned in PowerShell first, then retry activation.

help me activate my Python virtual environment

Now create the requirements file that pins every dependency to an exact version.

  • Create a file called requirements.txt in the voice-agent folder and paste the following content:
fastapi[standard]==0.137.2
langgraph==1.2.5
langchain-openai==1.3.2
chromadb==1.5.9
deepgram-sdk==7.3.1
elevenlabs==2.53.0
openai==2.43.0
python-dotenv==1.1.0

What does this file do?

Pinning exact versions (e.g. langgraph==1.2.5) guarantees every learner installs the same package versions. This prevents subtle bugs caused by breaking changes in newer releases.

  • Save requirements.txt.
  • Install all dependencies by running:
pip install -r requirements.txt

This will take a minute or two as pip downloads and installs all eight packages and their sub-dependencies.

  • Confirm the installation succeeded by running:
pip show langgraph

You should see version 1.2.5 printed in the output.

Seeing installation errors?

Make sure (venv) is visible in your prompt before running pip install. If you see compiler errors related to chromadb, confirm you are running Python 3.11 or 3.12 (not 3.13+).

help me fix pip install errors for the voice agent project

✔️ Awesome, I've got everything!

Great. Double check you saved requirements.txt and that the install completed without errors.

ⓧ I'd like to double check the full code

fastapi[standard]==0.137.2
langgraph==1.2.5
langchain-openai==1.3.2
chromadb==1.5.9
deepgram-sdk==7.3.1
elevenlabs==2.53.0
openai==2.43.0
python-dotenv==1.1.0
Configure environment variables

Your voice agent calls three external APIs: OpenAI for language model inference, Deepgram for speech-to-text, and ElevenLabs for text-to-speech. Each requires an API key stored in a .env file that is never committed to version control.

  • Create a file called .env in the voice-agent folder and paste the following structure, replacing each placeholder with your real key:
OPENAI_API_KEY=your-openai-api-key-here
DEEPGRAM_API_KEY=your-deepgram-api-key-here
ELEVENLABS_API_KEY=your-elevenlabs-api-key-here

Where do I find my API keys?

  • OpenAI: Visit your API Keys page at platform.openai.com and copy your secret key.
  • Deepgram: Visit your Deepgram console dashboard and create or copy your API key.
  • ElevenLabs: Visit your ElevenLabs profile settings to find your API key.
  • Save .env.
  • Create a file called .env.example in the same folder (this is a safe-to-commit reference showing the required variables without real values):
OPENAI_API_KEY=your-openai-api-key-here
DEEPGRAM_API_KEY=your-deepgram-api-key-here
ELEVENLABS_API_KEY=your-elevenlabs-api-key-here
  • Save .env.example.
  • Create the app/ directory and the configuration module by running:
mkdir app
  • Create a file called app/config.py and paste this code:
import os
from dotenv import load_dotenv

load_dotenv()

OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
DEEPGRAM_API_KEY = os.getenv("DEEPGRAM_API_KEY")
ELEVENLABS_API_KEY = os.getenv("ELEVENLABS_API_KEY")

What does this code do?

load_dotenv() reads the .env file and loads its key-value pairs into environment variables. Then os.getenv() pulls each key into a Python constant that other modules can import.

  • Save app/config.py.
  • Create an empty app/__init__.py file so Python treats app/ as a package:
touch app/__init__.py
  • Verify your keys load correctly by running this quick check:
python3 -c "from app.config import OPENAI_API_KEY, DEEPGRAM_API_KEY, ELEVENLABS_API_KEY; print('OpenAI:', bool(OPENAI_API_KEY)); print('Deepgram:', bool(DEEPGRAM_API_KEY)); print('ElevenLabs:', bool(ELEVENLABS_API_KEY))"

You should see three lines each showing True. If any line shows False, open your .env file and confirm the key is pasted correctly with no extra spaces.

Seeing False for one of the keys?

Double-check that your .env file uses exact variable names: OPENAI_API_KEY, DEEPGRAM_API_KEY, ELEVENLABS_API_KEY. There should be no quotes around the values and no trailing whitespace.

help me debug my .env file loading

✔️ Awesome, I've got everything!

Great. Make sure both app/config.py and app/__init__.py are saved.

ⓧ I'd like to double check the full code

import os
from dotenv import load_dotenv

load_dotenv()

OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
DEEPGRAM_API_KEY = os.getenv("DEEPGRAM_API_KEY")
ELEVENLABS_API_KEY = os.getenv("ELEVENLABS_API_KEY")
Create the project file structure

The voice agent is split into separate modules: one for the RAG knowledge base, one for the CRM simulator, one for the multi-agent router, one for the voice pipeline, and one for the FastAPI server.

  • Create the remaining stub files by running these commands:
touch app/rag.py app/crm.py app/agent.py app/voice.py app/main.py
touch test_client.py
  • Confirm all files exist by running:
ls app/

You should see: __init__.py, config.py, rag.py, crm.py, agent.py, voice.py, and main.py.

Why separate modules?

Each file handles one responsibility. This keeps the codebase navigable and lets you test components in isolation. You will fill in each module over the next steps.

Command not found: touch?

On Windows, touch is not available by default. Use New-Item instead, or create the files manually in your editor.

help me create empty files on Windows

Your environment is locked and loaded. Next up, you will populate the RAG knowledge base with product documentation and build the retrieval pipeline.

Build the RAG Knowledge Base

Your voice agent needs a knowledge source to answer customer questions intelligently. Without structured product documentation, the AI would hallucinate answers or give vague, unhelpful responses.

In this step, you'll build a RAG (Retrieval-Augmented Generation) pipeline. You'll store product documentation as vector embeddings in ChromaDB, then create a retrieval function that finds the most relevant information for any user query.

In this step, get ready to:
  • Store product documentation as vector embeddings in ChromaDB.
  • Build a retrieval function for semantic similarity search.
  • Test the complete RAG pipeline with sample queries.
Create the product knowledge base

A RAG system works by converting text documents into numerical vectors (embeddings) and storing them in a vector database. When a user asks a question, the system converts the question into a vector too, then finds the stored documents that are closest in meaning.

  • Open app/rag.py in your editor and add the module imports and client setup:
import chromadb
from openai import OpenAI
from app.config import OPENAI_API_KEY

# Initialize the OpenAI client for generating embeddings
client = OpenAI(api_key=OPENAI_API_KEY)
# Create an in-memory ChromaDB client and collection
chroma_client = chromadb.Client()
collection = chroma_client.get_or_create_collection(name="product_docs")

What does this code do?

You're initializing two clients that work together to power the RAG system.

  • The OpenAI client generates vector embeddings from text using the text-embedding-3-small model.
  • The chromadb.Client() creates an in-memory vector database. No server or account needed.
  • The get_or_create_collection method safely creates the collection if it doesn't exist, or returns the existing one.

Now add the product documentation data below the collection setup. This simulates the knowledge base for a fictional SaaS product called StreamLine.

  • Add this PRODUCT_DOCS list below the collection line in app/rag.py:
PRODUCT_DOCS = [
    {
        "id": "pricing-001",
        "text": "StreamLine SaaS offers three pricing tiers: Starter at $29/month for up to 5 users, Professional at $79/month for up to 25 users with advanced analytics, and Enterprise at $199/month for unlimited users with dedicated support and custom integrations.",
        "metadata": {"source": "pricing.md", "category": "pricing"},
    },
    {
        "id": "features-001",
        "text": "StreamLine includes real-time collaboration, automated workflow builder, API access on Professional and above, custom dashboards, and 99.9% uptime SLA on Enterprise tier.",
        "metadata": {"source": "features.md", "category": "features"},
    },
    {
        "id": "troubleshooting-001",
        "text": "If you cannot log in, first check that your email is verified. Reset your password via the login page. If issues persist, clear browser cache or try incognito mode. Contact support if the problem continues.",
        "metadata": {"source": "troubleshooting.md", "category": "troubleshooting"},
    },
    {
        "id": "features-002",
        "text": "The automated workflow builder supports conditional logic, scheduled triggers, webhook integrations, and connects to over 200 third-party applications including Slack, Salesforce, and HubSpot.",
        "metadata": {"source": "features.md", "category": "features"},
    },
    {
        "id": "troubleshooting-002",
        "text": "For slow dashboard loading, ensure your browser is updated to the latest version. Disable browser extensions that may interfere. If using a VPN, try connecting without it. Enterprise users can contact their dedicated support engineer.",
        "metadata": {"source": "troubleshooting.md", "category": "troubleshooting"},
    },
]

What does this data represent?

Each document has three fields: an id for unique identification, text containing the actual content, and metadata tracking which source file and category it belongs to.

In a production system, you'd load hundreds of documents from real files. These 5 documents cover pricing, features, and troubleshooting to demonstrate multi-category retrieval.

  • Save app/rag.py.

You won't see output yet. The next chunk adds the function that embeds and stores this data.

Seeing an import error?

  • Confirm your virtual environment is activated. You should see the environment name in your terminal prompt.
  • Check that chromadb and openai are installed by running pip list | grep -i chromadb.

Help me debug an import error in my rag.py file.

Now you need a function to embed all the documents and store them in ChromaDB. This is the function that populates your knowledge base on startup.

  • Add the initialize_knowledge_base function below the PRODUCT_DOCS list:
def initialize_knowledge_base():
    # Skip if documents are already embedded
    existing = collection.get()
    if len(existing["ids"]) > 0:
        return

    # Extract text, IDs, and metadata from the product docs
    documents = [doc["text"] for doc in PRODUCT_DOCS]
    ids = [doc["id"] for doc in PRODUCT_DOCS]
    metadatas = [doc["metadata"] for doc in PRODUCT_DOCS]

    # Generate embeddings for all documents in one API call
    response = client.embeddings.create(
        input=documents, model="text-embedding-3-small"
    )
    embeddings = [item.embedding for item in response.data]

    # Store everything in ChromaDB
    collection.add(documents=documents, embeddings=embeddings, ids=ids, metadatas=metadatas)

What does this function do?

  • The collection.get() check prevents re-embedding documents that are already stored. This makes the function safe to call multiple times.
  • The client.embeddings.create() call sends all 5 documents to OpenAI in a single batch request. Each document becomes a high-dimensional vector.
  • The collection.add() stores the vectors alongside their original text, IDs, and metadata. ChromaDB uses these vectors for similarity search later.
  • Save app/rag.py.

You won't see output yet. The retrieval function in the next substep completes the pipeline for testing.

Worried about the OpenAI API call?

  • The text-embedding-3-small model costs fractions of a cent for 5 documents. You won't see a meaningful charge.
  • If your OPENAI_API_KEY is invalid, you'll get an authentication error when you test. Double-check the value in your .env file.

Help me fix an OpenAI API authentication error in my RAG knowledge base setup.

Build the retrieval function

The retrieval function is the query-time half of your RAG pipeline. It converts a user question into a vector and finds the stored documents closest in meaning.

  • Add the retrieve function below initialize_knowledge_base in app/rag.py:
def retrieve(query: str, n_results: int = 3) -> list[dict]:
    # Embed the user's query using the same model
    response = client.embeddings.create(input=[query], model="text-embedding-3-small")
    query_embedding = response.data[0].embedding

    # Find the most similar documents in ChromaDB
    results = collection.query(query_embeddings=[query_embedding], n_results=n_results)

    # Format results with text and source attribution
    retrieved = []
    for i in range(len(results["ids"][0])):
        retrieved.append(
            {
                "text": results["documents"][0][i],
                "source": results["metadatas"][0][i]["source"],
            }
        )
    return retrieved

What does this function do?

  • The query is embedded using the same text-embedding-3-small model as the documents. Using the same model ensures vectors are comparable.
  • The collection.query() performs a similarity search and returns the top 3 closest matches by default.
  • Each result includes the original text and source file. This source attribution lets the voice agent cite where its information comes from.
  • Save app/rag.py.

The file is now complete. The next substep runs the full pipeline to confirm everything works.

Not sure about the nested list indexing?

ChromaDB's query() returns results nested inside lists (one list per query batch). Since you're querying with a single embedding, the results are at index [0]. For example, results["documents"][0][i] accesses the i-th document from the first (and only) query's results.

Help me understand ChromaDB's query result format and the nested list indexing.

✔️ Awesome, I've got everything!

Great. Double check you've saved app/rag.py before moving on to the test.

ⓧ I'd like to double check the full code

import chromadb
from openai import OpenAI
from app.config import OPENAI_API_KEY

client = OpenAI(api_key=OPENAI_API_KEY)
chroma_client = chromadb.Client()
collection = chroma_client.get_or_create_collection(name="product_docs")

PRODUCT_DOCS = [
    {
        "id": "pricing-001",
        "text": "StreamLine SaaS offers three pricing tiers: Starter at $29/month for up to 5 users, Professional at $79/month for up to 25 users with advanced analytics, and Enterprise at $199/month for unlimited users with dedicated support and custom integrations.",
        "metadata": {"source": "pricing.md", "category": "pricing"},
    },
    {
        "id": "features-001",
        "text": "StreamLine includes real-time collaboration, automated workflow builder, API access on Professional and above, custom dashboards, and 99.9% uptime SLA on Enterprise tier.",
        "metadata": {"source": "features.md", "category": "features"},
    },
    {
        "id": "troubleshooting-001",
        "text": "If you cannot log in, first check that your email is verified. Reset your password via the login page. If issues persist, clear browser cache or try incognito mode. Contact support if the problem continues.",
        "metadata": {"source": "troubleshooting.md", "category": "troubleshooting"},
    },
    {
        "id": "features-002",
        "text": "The automated workflow builder supports conditional logic, scheduled triggers, webhook integrations, and connects to over 200 third-party applications including Slack, Salesforce, and HubSpot.",
        "metadata": {"source": "features.md", "category": "features"},
    },
    {
        "id": "troubleshooting-002",
        "text": "For slow dashboard loading, ensure your browser is updated to the latest version. Disable browser extensions that may interfere. If using a VPN, try connecting without it. Enterprise users can contact their dedicated support engineer.",
        "metadata": {"source": "troubleshooting.md", "category": "troubleshooting"},
    },
]


def initialize_knowledge_base():
    existing = collection.get()
    if len(existing["ids"]) > 0:
        return

    documents = [doc["text"] for doc in PRODUCT_DOCS]
    ids = [doc["id"] for doc in PRODUCT_DOCS]
    metadatas = [doc["metadata"] for doc in PRODUCT_DOCS]

    response = client.embeddings.create(
        input=documents, model="text-embedding-3-small"
    )
    embeddings = [item.embedding for item in response.data]

    collection.add(documents=documents, embeddings=embeddings, ids=ids, metadatas=metadatas)


def retrieve(query: str, n_results: int = 3) -> list[dict]:
    response = client.embeddings.create(input=[query], model="text-embedding-3-small")
    query_embedding = response.data[0].embedding

    results = collection.query(query_embeddings=[query_embedding], n_results=n_results)

    retrieved = []
    for i in range(len(results["ids"][0])):
        retrieved.append(
            {
                "text": results["documents"][0][i],
                "source": results["metadatas"][0][i]["source"],
            }
        )
    return retrieved
Test the RAG pipeline end-to-end

Time to verify that your knowledge base actually returns relevant results for a given query. This test initializes the embeddings and runs a similarity search.

  • Test the RAG pipeline by running this command in your terminal:
python -c "from app.rag import initialize_knowledge_base, retrieve; initialize_knowledge_base(); results = retrieve('What are the pricing tiers?'); print(f'Found {len(results)} results'); [print(f'  Source: {r[\"source\"]}, Text: {r[\"text\"][:80]}...') for r in results]"
  • You should see 3 results printed to the terminal, with pricing.md as the top source since your query is about pricing tiers.

What should I see?

The output shows 3 results ordered by relevance. The first result should come from pricing.md because it directly mentions pricing tiers. The other results may come from features.md since features relate to what each tier offers.

This proves your RAG pipeline correctly embeds documents, indexes them in ChromaDB, and retrieves semantically relevant results based on meaning rather than exact keyword matches.

Seeing an error instead of results?

  • If you see ModuleNotFoundError, make sure your virtual environment is activated and you installed all dependencies from requirements.txt.
  • If you see an authentication error, double-check that your OPENAI_API_KEY in .env is valid and has billing enabled.
  • If you see 0 results, confirm your PRODUCT_DOCS list has all 5 documents and initialize_knowledge_base() is called before retrieve().

Help me debug why my RAG pipeline test is failing.

Your knowledge base is live and returning relevant results. Next up, you'll build the multi-agent router that decides whether a user's question should hit this RAG pipeline or go to a CRM system instead.

Build the Multi-Agent Router with LangGraph

Your RAG knowledge base can answer product questions. But a real support line handles more than that. Customers also want to check their account, create tickets, or look up order status.

In this step, you will build a multi-agent router using LangGraph. The router inspects each user message, classifies the intent, and sends it down the correct path. Product questions go to the RAG agent. Account actions go to a CRM agent.

In this step, get ready to:
  • Define the LangGraph state schema and intent classification node.
  • Create the RAG and CRM agent nodes with conditional routing.
  • Compile and test the graph with sample queries.
Define the state schema and intent classifier

Every LangGraph graph needs a state schema. The schema defines what data flows between nodes. Your agent will pass around the user message, the classified intent, retrieved context, CRM results, and a final response.

  • Open app/agent.py and add the imports, LLM initialization, and state schema by pasting this code:
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from app.config import OPENAI_API_KEY
from app.rag import retrieve
from app.crm import lookup_customer, create_ticket, get_ticket_status

# Initialize the LLM with deterministic output
llm = ChatOpenAI(model="gpt-4o-mini", api_key=OPENAI_API_KEY, temperature=0)


# Define the state that flows between all nodes in the graph
class AgentState(TypedDict):
    user_message: str
    intent: str
    rag_context: list[dict]
    crm_result: dict
    response: str

What does this code do?

The AgentState TypedDict defines the shape of data that every node in your graph can read and write. Each node receives the full state, does its work, and returns an updated copy.

The ChatOpenAI instance uses gpt-4o-mini with temperature=0 for deterministic classification. The CRM imports will resolve once you create that module shortly.

  • Below the AgentState class, add the intent classification node and routing function:
def classify_intent(state: AgentState) -> AgentState:
    # Use the LLM to determine if the query is about products or account actions
    classification_prompt = f"""Classify the following user message into one of two categories:
- "rag" if the user is asking about product features, pricing, troubleshooting, or general knowledge
- "crm" if the user wants to look up their account, create a support ticket, or check ticket status

User message: {state["user_message"]}

Respond with ONLY "rag" or "crm"."""

    result = llm.invoke(classification_prompt)
    intent = result.content.strip().lower()
    # Default to RAG if classification is unclear
    if intent not in ("rag", "crm"):
        intent = "rag"
    return {**state, "intent": intent}


def route_by_intent(state: AgentState) -> Literal["rag_agent", "crm_agent"]:
    # Conditional edge function that directs flow based on classified intent
    return "rag_agent" if state["intent"] == "rag" else "crm_agent"

What does this code do?

The classify_intent function is a graph node. It sends the user's message to GPT-4o-mini with a prompt that forces a binary classification. The result is stored in the intent field of the state.

The route_by_intent function is not a node. It is a conditional edge function that LangGraph calls after classify_intent finishes. It reads the intent field and returns which node should execute next.

  • Save app/agent.py.

Seeing an import error for app.crm?

That is expected. The app/crm.py module is still empty. You will create it in the next substep. The import will resolve once that file has content.

Ask for help: Why am I seeing an import error for app.crm in my agent module?

Create the CRM simulator and agent nodes

The CRM agent needs mock customer data to simulate real account lookups and ticket operations. You will create a standalone CRM module, then build both the RAG agent node and the CRM agent node.

  • Open app/crm.py and add the mock customer database:
# Mock customer database simulating a real CRM system
MOCK_CUSTOMERS = {
    "john@example.com": {
        "name": "John Smith",
        "plan": "Professional",
        "tickets": [
            {"id": "TKT-001", "subject": "API rate limiting", "status": "resolved"}
        ],
    },
    "sarah@example.com": {
        "name": "Sarah Johnson",
        "plan": "Enterprise",
        "tickets": [
            {"id": "TKT-002", "subject": "SSO configuration", "status": "open"}
        ],
    },
}

# Counter for generating new ticket IDs
TICKET_COUNTER = 3

What does this code do?

The MOCK_CUSTOMERS dictionary acts as a stand-in for a real CRM database. Each customer has a name, subscription plan, and list of support tickets. The TICKET_COUNTER tracks the next ID to assign when creating new tickets.

  • Below the TICKET_COUNTER, add the three CRM operation functions:
def lookup_customer(email: str) -> dict:
    # Search for a customer by email address
    customer = MOCK_CUSTOMERS.get(email)
    if customer:
        return {"found": True, **customer}
    return {"found": False, "message": f"No customer found with email {email}"}


def create_ticket(email: str, subject: str) -> dict:
    # Create a new support ticket and return its details
    global TICKET_COUNTER
    TICKET_COUNTER += 1
    ticket_id = f"TKT-{TICKET_COUNTER:03d}"
    return {
        "ticket_id": ticket_id,
        "email": email,
        "subject": subject,
        "status": "open",
        "message": f"Ticket {ticket_id} created successfully for {subject}",
    }


def get_ticket_status(ticket_id: str) -> dict:
    # Look up an existing ticket by ID across all customers
    for customer in MOCK_CUSTOMERS.values():
        for ticket in customer["tickets"]:
            if ticket["id"] == ticket_id:
                return ticket
    return {"found": False, "message": f"Ticket {ticket_id} not found"}

What does this code do?

  • lookup_customer searches the mock database by email and returns the customer record or a not-found message.
  • create_ticket generates a new ticket ID (incrementing the counter) and returns the ticket details.
  • get_ticket_status iterates through all customers' tickets to find a match by ticket ID.
  • Save app/crm.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved app/crm.py before continuing.

ⓧ I'd like to double check the full code

MOCK_CUSTOMERS = {
    "john@example.com": {
        "name": "John Smith",
        "plan": "Professional",
        "tickets": [
            {"id": "TKT-001", "subject": "API rate limiting", "status": "resolved"}
        ],
    },
    "sarah@example.com": {
        "name": "Sarah Johnson",
        "plan": "Enterprise",
        "tickets": [
            {"id": "TKT-002", "subject": "SSO configuration", "status": "open"}
        ],
    },
}

TICKET_COUNTER = 3


def lookup_customer(email: str) -> dict:
    customer = MOCK_CUSTOMERS.get(email)
    if customer:
        return {"found": True, **customer}
    return {"found": False, "message": f"No customer found with email {email}"}


def create_ticket(email: str, subject: str) -> dict:
    global TICKET_COUNTER
    TICKET_COUNTER += 1
    ticket_id = f"TKT-{TICKET_COUNTER:03d}"
    return {
        "ticket_id": ticket_id,
        "email": email,
        "subject": subject,
        "status": "open",
        "message": f"Ticket {ticket_id} created successfully for {subject}",
    }


def get_ticket_status(ticket_id: str) -> dict:
    for customer in MOCK_CUSTOMERS.values():
        for ticket in customer["tickets"]:
            if ticket["id"] == ticket_id:
                return ticket
    return {"found": False, "message": f"Ticket {ticket_id} not found"}

Now switch back to app/agent.py to build the RAG agent node.

  • Below the route_by_intent function, add the RAG agent node:
def rag_agent(state: AgentState) -> AgentState:
    # Retrieve relevant documents from the knowledge base
    context_docs = retrieve(state["user_message"])
    context_text = "\n\n".join(
        [f"[{doc['source']}]: {doc['text']}" for doc in context_docs]
    )

    # Generate a grounded answer using only the retrieved context
    generation_prompt = f"""You are a helpful product support agent for StreamLine SaaS.
Answer the user's question using ONLY the provided context. Be concise and professional.
If the context doesn't contain the answer, say so honestly.

Context:
{context_text}

User question: {state["user_message"]}

Answer:"""

    result = llm.invoke(generation_prompt)
    return {**state, "rag_context": context_docs, "response": result.content}

What does this code do?

The rag_agent node calls the retrieve function you built earlier to pull the top matching documents from ChromaDB. It formats them with source attribution, then asks the LLM to answer using only that context. This is the retrieval-augmented generation pattern in action.

  • Below the rag_agent function, add the CRM agent node:
def crm_agent(state: AgentState) -> AgentState:
    # Ask the LLM to determine which CRM action to take
    crm_prompt = f"""You are a CRM assistant. Based on the user message, determine the action:
- If they want to look up an account, extract the email and call lookup
- If they want to create a ticket, extract the subject
- If they want to check a ticket, extract the ticket ID

User message: {state["user_message"]}

Respond with a JSON action: {{"action": "lookup"|"create_ticket"|"check_status", "email": "...", "subject": "...", "ticket_id": "..."}}
Include only relevant fields."""

    result = llm.invoke(crm_prompt)
    content = result.content.strip()

    try:
        import json
        action_data = json.loads(content)
        action = action_data.get("action", "")

        # Route to the appropriate CRM function based on the LLM's decision
        if action == "lookup":
            crm_result = lookup_customer(action_data.get("email", ""))
        elif action == "create_ticket":
            crm_result = create_ticket(
                action_data.get("email", "unknown@example.com"),
                action_data.get("subject", "General inquiry"),
            )
        elif action == "check_status":
            crm_result = get_ticket_status(action_data.get("ticket_id", ""))
        else:
            crm_result = {"message": "I can help you look up your account, create a support ticket, or check ticket status. What would you like to do?"}
    except (json.JSONDecodeError, KeyError):
        crm_result = {"message": "I'd be happy to help with your account. Could you provide your email address or ticket ID?"}

    # Convert the raw CRM result into a conversational response
    response_prompt = f"""Summarize this CRM result in a helpful, conversational way for the customer:
{crm_result}"""

    response = llm.invoke(response_prompt)
    return {**state, "crm_result": crm_result, "response": response.content}

What does this code do?

The crm_agent node uses a two-step pattern. First, it asks the LLM to extract structured data (action type, email, subject, or ticket ID) from the user's natural language request. Then it calls the matching CRM function and generates a friendly response.

The try/except block handles cases where the LLM output is not valid JSON. Instead of crashing, the agent asks the user for clarification.

  • Below the crm_agent function, add the graph builder and compile it:
def build_agent_graph() -> StateGraph:
    # Construct the state machine graph
    graph = StateGraph(AgentState)

    # Register each function as a named node
    graph.add_node("classify_intent", classify_intent)
    graph.add_node("rag_agent", rag_agent)
    graph.add_node("crm_agent", crm_agent)

    # Set the entry point and define conditional routing
    graph.set_entry_point("classify_intent")
    graph.add_conditional_edges("classify_intent", route_by_intent)
    graph.add_edge("rag_agent", END)
    graph.add_edge("crm_agent", END)

    return graph.compile()


# Create the compiled agent instance
agent = build_agent_graph()

How does the routing work?

The StateGraph defines a state machine. Every invocation starts at classify_intent. After classification, add_conditional_edges calls route_by_intent to decide the next node. If the intent is rag, the graph executes rag_agent. If crm, it executes crm_agent. Both paths terminate at END.

The final line compiles and instantiates the graph so other modules can import and invoke it directly.

  • Save app/agent.py.

Seeing a ModuleNotFoundError?

Make sure your virtual environment is activated. If you see an error about langgraph or langchain_openai not being found, re-run your install command from Step 1.

Ask for help: Why am I getting a ModuleNotFoundError for langgraph or langchain_openai?

✔️ Awesome, I've got everything!

Great. Double check you have saved app/agent.py before continuing.

ⓧ I'd like to double check the full code

from typing import TypedDict, Literal
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from app.config import OPENAI_API_KEY
from app.rag import retrieve
from app.crm import lookup_customer, create_ticket, get_ticket_status

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


class AgentState(TypedDict):
    user_message: str
    intent: str
    rag_context: list[dict]
    crm_result: dict
    response: str


def classify_intent(state: AgentState) -> AgentState:
    classification_prompt = f"""Classify the following user message into one of two categories:
- "rag" if the user is asking about product features, pricing, troubleshooting, or general knowledge
- "crm" if the user wants to look up their account, create a support ticket, or check ticket status

User message: {state["user_message"]}

Respond with ONLY "rag" or "crm"."""

    result = llm.invoke(classification_prompt)
    intent = result.content.strip().lower()
    if intent not in ("rag", "crm"):
        intent = "rag"
    return {**state, "intent": intent}


def route_by_intent(state: AgentState) -> Literal["rag_agent", "crm_agent"]:
    return "rag_agent" if state["intent"] == "rag" else "crm_agent"


def rag_agent(state: AgentState) -> AgentState:
    context_docs = retrieve(state["user_message"])
    context_text = "\n\n".join(
        [f"[{doc['source']}]: {doc['text']}" for doc in context_docs]
    )

    generation_prompt = f"""You are a helpful product support agent for StreamLine SaaS.
Answer the user's question using ONLY the provided context. Be concise and professional.
If the context doesn't contain the answer, say so honestly.

Context:
{context_text}

User question: {state["user_message"]}

Answer:"""

    result = llm.invoke(generation_prompt)
    return {**state, "rag_context": context_docs, "response": result.content}


def crm_agent(state: AgentState) -> AgentState:
    crm_prompt = f"""You are a CRM assistant. Based on the user message, determine the action:
- If they want to look up an account, extract the email and call lookup
- If they want to create a ticket, extract the subject
- If they want to check a ticket, extract the ticket ID

User message: {state["user_message"]}

Respond with a JSON action: {{"action": "lookup"|"create_ticket"|"check_status", "email": "...", "subject": "...", "ticket_id": "..."}}
Include only relevant fields."""

    result = llm.invoke(crm_prompt)
    content = result.content.strip()

    try:
        import json
        action_data = json.loads(content)
        action = action_data.get("action", "")

        if action == "lookup":
            crm_result = lookup_customer(action_data.get("email", ""))
        elif action == "create_ticket":
            crm_result = create_ticket(
                action_data.get("email", "unknown@example.com"),
                action_data.get("subject", "General inquiry"),
            )
        elif action == "check_status":
            crm_result = get_ticket_status(action_data.get("ticket_id", ""))
        else:
            crm_result = {"message": "I can help you look up your account, create a support ticket, or check ticket status. What would you like to do?"}
    except (json.JSONDecodeError, KeyError):
        crm_result = {"message": "I'd be happy to help with your account. Could you provide your email address or ticket ID?"}

    response_prompt = f"""Summarize this CRM result in a helpful, conversational way for the customer:
{crm_result}"""

    response = llm.invoke(response_prompt)
    return {**state, "crm_result": crm_result, "response": response.content}


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

    graph.add_node("classify_intent", classify_intent)
    graph.add_node("rag_agent", rag_agent)
    graph.add_node("crm_agent", crm_agent)

    graph.set_entry_point("classify_intent")
    graph.add_conditional_edges("classify_intent", route_by_intent)
    graph.add_edge("rag_agent", END)
    graph.add_edge("crm_agent", END)

    return graph.compile()


agent = build_agent_graph()
Test the multi-agent router

Time to verify that the router correctly classifies intents and sends queries down the right path. You will invoke the compiled graph with a product question and a CRM action to confirm both routes work.

  • Test the agent by running this command in your terminal:
python -c "
from app.agent import agent
from app.rag import initialize_knowledge_base

initialize_knowledge_base()

# Test RAG routing
state = {'user_message': 'What are the pricing tiers?', 'intent': '', 'rag_context': [], 'crm_result': {}, 'response': ''}
result = agent.invoke(state)
print(f'Query: What are the pricing tiers?')
print(f'Intent: {result[\"intent\"]}')
print(f'Response: {result[\"response\"][:150]}...')
print('---')

# Test CRM routing
state = {'user_message': 'Create a support ticket for login issues', 'intent': '', 'rag_context': [], 'crm_result': {}, 'response': ''}
result = agent.invoke(state)
print(f'Query: Create a support ticket for login issues')
print(f'Intent: {result[\"intent\"]}')
print(f'Response: {result[\"response\"][:150]}...')
"

What should I see?

You should see two test results. The first query about pricing tiers should show Intent: rag and a response mentioning the Starter, Professional, and Enterprise tiers. The second query about creating a ticket should show Intent: crm and a response confirming ticket creation.

Seeing an API error or timeout?

Check that your .env file has a valid OPENAI_API_KEY. The test makes several API calls to OpenAI (embedding + LLM classification + LLM response generation), so it may take a few seconds.

If you see a rate limit error, wait 30 seconds and try again. Free-tier OpenAI accounts have lower rate limits.

Ask for help: Why is my agent test failing with an API error?

Your multi-agent router is working. Product questions route to the RAG knowledge base, and account actions route to the CRM agent. Next up, you will wire this intelligence into a real-time voice pipeline with speech-to-text and text-to-speech.

Integrate the Real-Time Voice Pipeline

Your multi-agent router can classify intent and generate intelligent responses. The missing piece is the human interface. Right now, the agent only accepts text input and returns text output.

In this step, you will wire up real-time audio I/O so users can speak to the agent and hear its response. Deepgram handles speech-to-text transcription, and ElevenLabs handles text-to-speech synthesis. A FastAPI WebSocket endpoint ties everything together into a single real-time loop.

In this step, get ready to:
  • Build the Deepgram STT integration for real-time transcription.
  • Build the ElevenLabs TTS integration for streaming speech synthesis.
  • Create the FastAPI WebSocket endpoint orchestrating the full voice pipeline.
Build the Deepgram STT integration

Speech-to-text (STT) converts raw audio bytes into a text transcript the agent can process. You will use Deepgram's nova-3 model via the REST transcription method for simplicity.

  • Open app/voice.py and add the imports, client initialization, and the transcription function by pasting the following code:
from deepgram import DeepgramClient, LiveOptions, LiveTranscriptionEvents
from elevenlabs import stream as play_stream
from elevenlabs.client import ElevenLabs
from app.config import DEEPGRAM_API_KEY, ELEVENLABS_API_KEY

# Initialize API clients for both speech services
deepgram_client = DeepgramClient(api_key=DEEPGRAM_API_KEY)
elevenlabs_client = ElevenLabs(api_key=ELEVENLABS_API_KEY)


async def transcribe_audio(audio_bytes: bytes) -> str:
    # Send audio bytes to Deepgram's REST API for transcription
    response = deepgram_client.listen.v1.rest.transcribe_file(
        request=audio_bytes,
        model="nova-3",
    )
    # Extract the transcript from the first channel's best alternative
    return response.results.channels[0].alternatives[0].transcript

What does this code do?

The imports pull in both the Deepgram and ElevenLabs SDKs. You initialize both clients at module level so they are ready when any function in this file is called.

The transcribe_audio function sends raw audio bytes to Deepgram's Listen v1 REST endpoint using the nova-3 model. It returns the best transcript from the response.

  • Save app/voice.py.
  • You won't see a change yet. The next chunk adds the TTS function that completes this module.

Seeing an import error?

Make sure your virtual environment is activated and all dependencies from requirements.txt are installed. The deepgram-sdk and elevenlabs packages must both be present.

Help me debug a Deepgram or ElevenLabs import error in my voice.py file.

Build the ElevenLabs TTS integration

Text-to-speech (TTS) converts the agent's text response into natural-sounding audio. The eleven_flash_v2_5 model provides ultra-low latency (75ms) ideal for real-time conversational voice.

  • Below the transcribe_audio function in app/voice.py, add the speech synthesis function:
def synthesize_speech(text: str) -> bytes:
    # Stream TTS audio using ElevenLabs Flash v2.5 for low latency
    audio_stream = elevenlabs_client.text_to_speech.stream(
        text=text,
        voice_id="JBFqnCBsd6RMkjVDRZzb",
        model_id="eleven_flash_v2_5",
        output_format="mp3_44100_128",
    )

    # Collect all audio chunks into a single bytes object
    audio_chunks = []
    for chunk in audio_stream:
        if isinstance(chunk, bytes):
            audio_chunks.append(chunk)
    return b"".join(audio_chunks)

What does this code do?

The synthesize_speech function calls ElevenLabs' streaming TTS API. It uses voice ID JBFqnCBsd6RMkjVDRZzb (the "George" voice) and the eleven_flash_v2_5 model optimized for conversational speed.

The stream returns audio in chunks. The function collects them into one bytes object that can be sent over the WebSocket to the client.

  • Save app/voice.py.
  • You won't see a change yet. The next substep builds the FastAPI server that calls both functions.

Unsure about your ElevenLabs voice ID?

The voice ID JBFqnCBsd6RMkjVDRZzb is a pre-made ElevenLabs voice. If your account does not have access to it, you can replace it with any voice ID from your ElevenLabs dashboard.

Help me find a working voice ID for my ElevenLabs TTS integration.

✔️ Awesome, I've got everything!

Great. Double check you have saved app/voice.py with both functions before moving on.

ⓧ I'd like to double check the full code

from deepgram import DeepgramClient, LiveOptions, LiveTranscriptionEvents
from elevenlabs import stream as play_stream
from elevenlabs.client import ElevenLabs
from app.config import DEEPGRAM_API_KEY, ELEVENLABS_API_KEY

deepgram_client = DeepgramClient(api_key=DEEPGRAM_API_KEY)
elevenlabs_client = ElevenLabs(api_key=ELEVENLABS_API_KEY)


async def transcribe_audio(audio_bytes: bytes) -> str:
    response = deepgram_client.listen.v1.rest.transcribe_file(
        request=audio_bytes,
        model="nova-3",
    )
    return response.results.channels[0].alternatives[0].transcript


def synthesize_speech(text: str) -> bytes:
    audio_stream = elevenlabs_client.text_to_speech.stream(
        text=text,
        voice_id="JBFqnCBsd6RMkjVDRZzb",
        model_id="eleven_flash_v2_5",
        output_format="mp3_44100_128",
    )

    audio_chunks = []
    for chunk in audio_stream:
        if isinstance(chunk, bytes):
            audio_chunks.append(chunk)
    return b"".join(audio_chunks)
Create the FastAPI WebSocket endpoint

The server needs a single WebSocket endpoint that ties together audio input, the agent brain, and audio output into one real-time loop. You will also add a health check endpoint and a startup event that pre-loads the RAG knowledge base.

  • Open app/main.py and add the FastAPI app setup, startup event, and health endpoint:
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from app.agent import agent, AgentState
from app.voice import transcribe_audio, synthesize_speech
from app.rag import initialize_knowledge_base

# Create the FastAPI application
app = FastAPI(title="Enterprise Voice Agent")


@app.on_event("startup")
async def startup():
    # Pre-load the RAG knowledge base when the server starts
    initialize_knowledge_base()


@app.get("/health")
async def health():
    # Simple health check endpoint for monitoring
    return {"status": "healthy", "service": "voice-agent"}

What does this code do?

The startup event runs initialize_knowledge_base() once when the server boots. This ensures the ChromaDB collection is populated with product docs before any voice queries arrive.

The /health endpoint returns a simple JSON response, useful for Docker health checks and load balancer probes.

  • You won't see a change yet. The next chunk adds the WebSocket endpoint that handles the full voice loop.
  • Below the health endpoint in app/main.py, add the WebSocket voice endpoint:
@app.websocket("/ws/voice")
async def voice_endpoint(websocket: WebSocket):
    await websocket.accept()

    try:
        while True:
            # Receive raw audio bytes from the client
            audio_data = await websocket.receive_bytes()

            # Transcribe the audio to text using Deepgram
            transcript = await transcribe_audio(audio_data)
            if not transcript.strip():
                continue

            # Send the transcript back to the client for display
            await websocket.send_json({"type": "transcript", "text": transcript})

            # Route through the LangGraph multi-agent system
            state: AgentState = {
                "user_message": transcript,
                "intent": "",
                "rag_context": [],
                "crm_result": {},
                "response": "",
            }
            result = agent.invoke(state)

            # Send the text response and detected intent
            await websocket.send_json(
                {
                    "type": "response",
                    "text": result["response"],
                    "intent": result["intent"],
                }
            )

            # Synthesize the response as speech and stream audio back
            audio_response = synthesize_speech(result["response"])
            await websocket.send_bytes(audio_response)

    except WebSocketDisconnect:
        pass

What does the WebSocket loop do?

Each iteration of the loop is one full voice exchange. The client sends audio bytes, the server transcribes them, routes the text through the LangGraph agent, synthesizes the response into speech, and streams the audio bytes back.

The send_json calls send metadata (transcript text, response text, detected intent) alongside the binary audio. This lets a frontend display the conversation in text form while playing audio.

  • Save app/main.py.

Seeing a circular import error?

Make sure app/__init__.py exists and is empty. Also verify that app/voice.py is saved with the complete code from the previous substep before running the server.

Help me fix a circular import error in my FastAPI voice agent project.

✔️ Awesome, I've got everything!

Great. Double check you have saved app/main.py with both the startup/health code and the WebSocket endpoint.

ⓧ I'd like to double check the full code

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from app.agent import agent, AgentState
from app.voice import transcribe_audio, synthesize_speech
from app.rag import initialize_knowledge_base

app = FastAPI(title="Enterprise Voice Agent")


@app.on_event("startup")
async def startup():
    initialize_knowledge_base()


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


@app.websocket("/ws/voice")
async def voice_endpoint(websocket: WebSocket):
    await websocket.accept()

    try:
        while True:
            audio_data = await websocket.receive_bytes()

            transcript = await transcribe_audio(audio_data)
            if not transcript.strip():
                continue

            await websocket.send_json({"type": "transcript", "text": transcript})

            state: AgentState = {
                "user_message": transcript,
                "intent": "",
                "rag_context": [],
                "crm_result": {},
                "response": "",
            }
            result = agent.invoke(state)

            await websocket.send_json(
                {
                    "type": "response",
                    "text": result["response"],
                    "intent": result["intent"],
                }
            )

            audio_response = synthesize_speech(result["response"])
            await websocket.send_bytes(audio_response)

    except WebSocketDisconnect:
        pass

Now create a test client that exercises the agent pipeline without needing a live microphone.

  • Open test_client.py in your project root and paste the following test script:
import asyncio
import websockets
import json


async def test_text_mode():
    """Test the agent pipeline with a text query (simulates post-transcription)."""
    from app.agent import agent, AgentState
    from app.rag import initialize_knowledge_base

    # Ensure the knowledge base is loaded before querying
    initialize_knowledge_base()

    test_queries = [
        "What are the pricing tiers for StreamLine?",
        "Create a support ticket for login issues",
        "How do I fix slow dashboard loading?",
    ]

    for query in test_queries:
        state: AgentState = {
            "user_message": query,
            "intent": "",
            "rag_context": [],
            "crm_result": {},
            "response": "",
        }
        result = agent.invoke(state)
        print(f"\nQuery: {query}")
        print(f"Intent: {result['intent']}")
        print(f"Response: {result['response'][:200]}...")
        print("-" * 60)


if __name__ == "__main__":
    asyncio.run(test_text_mode())

What does this test do?

This script bypasses the audio layer and sends three text queries directly to the LangGraph agent. It verifies that intent classification and routing work correctly. You should see two queries routed to RAG (pricing and troubleshooting) and one to CRM (ticket creation).

  • Save test_client.py.
  • Run the test script by executing this command:
python test_client.py
  • You should see three query results printed in your terminal. Each shows the query text, the detected intent (rag or crm), and the first 200 characters of the agent's response.

Test script fails with a connection or API error?

Verify that your .env file contains valid values for OPENAI_API_KEY. The test script calls OpenAI for intent classification and response generation. Your Deepgram and ElevenLabs keys are not needed for this text-only test.

Help me debug why my test_client.py script is failing when calling the LangGraph agent.

✔️ Awesome, I've got everything!

Great. Make sure you saved test_client.py and confirmed the output shows correct routing.

ⓧ I'd like to double check the full code

import asyncio
import websockets
import json


async def test_text_mode():
    """Test the agent pipeline with a text query (simulates post-transcription)."""
    from app.agent import agent, AgentState
    from app.rag import initialize_knowledge_base

    initialize_knowledge_base()

    test_queries = [
        "What are the pricing tiers for StreamLine?",
        "Create a support ticket for login issues",
        "How do I fix slow dashboard loading?",
    ]

    for query in test_queries:
        state: AgentState = {
            "user_message": query,
            "intent": "",
            "rag_context": [],
            "crm_result": {},
            "response": "",
        }
        result = agent.invoke(state)
        print(f"\nQuery: {query}")
        print(f"Intent: {result['intent']}")
        print(f"Response: {result['response'][:200]}...")
        print("-" * 60)


if __name__ == "__main__":
    asyncio.run(test_text_mode())

Your voice pipeline is fully wired. Next up, you will containerize the entire application with Docker for portable deployment.

Containerize with Docker

Your voice agent works locally, but sharing it with others or deploying it to a server means packaging all the moving parts into a single portable unit.

In this step, you will containerize the entire application using Docker. A Docker image bundles your Python runtime, installed packages, and application code into one artifact that runs identically on any machine.

In this step, get ready to:
  • Write a production Dockerfile that builds a self-contained image of the voice agent.
  • Create a docker-compose.yml file for single-command local development.
  • Build, run, and test the containerized pipeline end-to-end.
Write the production Dockerfile

A Dockerfile is a recipe that tells Docker how to build your image layer by layer. You start from a base image, install system-level tools, copy your code in, and define the command that runs when the container starts.

  • Create a new file called Dockerfile in the root of your project folder (the same level as requirements.txt).
  • Paste in the following Dockerfile:
FROM python:3.11-slim

WORKDIR /app

# Install system-level build dependencies
RUN apt-get update && apt-get install -y --no-install-recommends \
    gcc \
    && rm -rf /var/lib/apt/lists/*

# Install Python dependencies first for layer caching
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy application code
COPY app/ ./app/

# Expose the WebSocket port
EXPOSE 8000

# Run the FastAPI server
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

What does this Dockerfile do?

Each instruction creates a layer in the final image:

  • FROM python:3.11-slim starts from a minimal Python 3.11 image, keeping the image small.
  • RUN apt-get install ... gcc adds the C compiler needed by some Python packages during installation.
  • COPY requirements.txt and RUN pip install are separated from the app code so Docker can cache installed packages when only your code changes.
  • CMD defines what runs when the container starts. Here it launches uvicorn on port 8000.
  • Save Dockerfile.
  • Confirm the file exists by running:
ls Dockerfile

You should see Dockerfile listed in the output.

Don't see Dockerfile?

Make sure you created it in the project root, not inside the app/ directory. It should sit alongside requirements.txt.

Also confirm the filename has no extension. It should be exactly Dockerfile with a capital D, not Dockerfile.txt.

Help me troubleshoot my Dockerfile creation

✔️ Awesome, I've got everything!

Great. Double check you have saved the file before moving on.

ⓧ I'd like to double check the full code

FROM python:3.11-slim

WORKDIR /app

RUN apt-get update && apt-get install -y --no-install-recommends \
    gcc \
    && rm -rf /var/lib/apt/lists/*

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

COPY app/ ./app/

EXPOSE 8000

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]
Create docker-compose.yml and run the container

Docker Compose lets you define how to build, configure, and run your container in a single YAML file. For development, you can mount your local code as a volume so changes reflect without rebuilding.

  • Create a new file called docker-compose.yml in the project root (same level as Dockerfile).
  • Paste the following configuration:
services:
  voice-agent:
    build: .
    ports:
      - "8000:8000"
    env_file:
      - .env
    volumes:
      - ./app:/app/app

What does this compose file do?

  • build: . tells Compose to build the image from the Dockerfile in the current directory.
  • ports: "8000:8000" maps port 8000 inside the container to port 8000 on your machine, making the WebSocket endpoint reachable.
  • env_file: .env loads your API keys into the container without baking them into the image.
  • volumes: ./app:/app/app mounts your local app/ folder into the container. Any code edits you make locally appear inside the running container immediately.
  • Save docker-compose.yml.

You won't see a change yet. The next command uses this file to build and start your container.

✔️ Awesome, I've got everything!

Great. Make sure you have saved the file before continuing.

ⓧ I'd like to double check the full code

services:
  voice-agent:
    build: .
    ports:
      - "8000:8000"
    env_file:
      - .env
    volumes:
      - ./app:/app/app
  • Build and start the container in detached mode by running:
docker compose up --build -d

What does this command do?

docker compose up creates and starts the container defined in your compose file. The --build flag forces Docker to rebuild the image from scratch. The -d flag runs it in the background so you get your terminal back.

Docker will pull the python:3.11-slim base image, install system packages, install Python dependencies, and start the server. This may take a few minutes on the first run.

  • Once the build finishes, verify the container is running by hitting the health endpoint:
curl http://localhost:8000/health

You should see this JSON response:

{"status":"healthy","service":"voice-agent"}

Seeing connection refused?

The container may still be starting up. Wait 10-15 seconds and try the curl command again.

If it still fails, check the container logs for errors:

docker compose logs voice-agent

Common causes include a missing .env file (the compose file expects it to exist) or a typo in your Dockerfile. Verify that .env sits in the same directory as docker-compose.yml.

Help me debug my Docker container not starting

  • Test the full agent pipeline inside the container by running the test client locally:
python test_client.py

You should see output for each test query showing the intent classification and a response snippet. The RAG queries route to the knowledge base, and the CRM query routes to the ticket system.

Why does this work outside the container?

The test_client.py script imports from app.agent and app.rag directly, testing the Python logic without going through the WebSocket endpoint. It proves the same code running inside the container produces correct routing decisions. In production, you would connect to the WebSocket at ws://localhost:8000/ws/voice with real audio data.

Seeing import errors from test_client.py?

Make sure your virtual environment is still activated. The test client runs on your local machine and needs the installed packages from your venv.

If you see an API key error, confirm your .env file is in the project root and contains valid keys.

Help me fix errors when running test_client.py

Secret mission

Add Conversation Memory

Your voice agent currently treats every utterance as a brand new conversation. It has no memory of what was said 10 seconds ago. In this mission, you'll add multi-turn conversation history so the agent can handle follow-up questions like a real support agent would.

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. This project runs locally with Docker, but API calls to Deepgram, ElevenLabs, and OpenAI incur usage-based charges whenever the voice agent processes requests.

Cost warning

Every voice interaction sends API calls to Deepgram (STT), OpenAI (LLM routing), and ElevenLabs (TTS). If you leave the Docker container running and connect clients to it, you will accumulate charges on all three services.

Stop the container when you are not actively testing. This halts all API calls immediately.

Resources you used:

  • Docker container and image (local, no ongoing charge when stopped).
  • Python virtual environment and project files on your Desktop.
  • API usage on Deepgram, ElevenLabs, and OpenAI (usage-based billing, charges stop when no requests are made).

Keep everything running

No action needed. Choose this if you are still actively building or experimenting with the voice agent.

  • Your Docker container continues running and is ready to accept WebSocket connections on port 8000.
  • Your Python virtual environment and project files remain on your Desktop.
  • API charges only accumulate when you actively send audio to the agent.

Pause - I'll come back to this later

Shut down the running container to free up system resources and stop any possibility of API charges. Your project files and images stay intact for next time.

  • Stop the Docker container by running this command from your project folder:
docker compose down
  • If your virtual environment is still active in your terminal, deactivate it:
deactivate

When you are ready to come back, navigate to your project folder and run docker compose up --build to start everything again.

Delete - I don't want to use this again

Remove all project resources and start fresh. This deletes your Docker container, images, virtual environment, and all project files.

  • Stop the container and remove the Docker image by running:
docker compose down --rmi all
  • Deactivate the virtual environment if it is still active:
deactivate
  • Navigate to your Desktop and remove the project folder entirely:
cd ~/Desktop
rm -rf voice-agent

What about my API accounts?

Your Deepgram, ElevenLabs, and OpenAI accounts existed before this project. They do not incur charges unless you make API calls. No further cleanup is needed for them.

If you want to revoke the API keys you created, visit each provider's dashboard and delete the keys from their respective API key management pages.

Nice Work!

Nice Work!

Nice work! You just built a production-grade real-time AI voice agent with autonomous multi-agent routing.

You've learned how to:

  • Build a real-time voice pipeline that transcribes speech with Deepgram nova-3, routes through a LangGraph state machine, and responds with synthesized audio from ElevenLabs.
  • Implement a multi-agent routing system with conditional edges that autonomously decides between a RAG knowledge base (ChromaDB with OpenAI embeddings) and a simulated CRM based on user intent.
  • Containerize the entire application with Docker for single-command portable deployment using a Dockerfile and docker-compose.yml.
  • Secret Mission: Add conversation memory for multi-turn contextual interactions, so the agent remembers previous exchanges within a session.

Ready to quiz yourself?