Building a Voice Assistant Backend with LangGraph: Complete Guide
Voice assistants have evolved from simple command-response systems into sophisticated conversational agents capable of handling complex, multi-turn interactions. Building the backend for such a system requires careful orchestration of speech recognition, language understanding, tool execution, state management, and response generation. LangGraph, an extension of LangChain, provides a powerful framework for building stateful, multi-actor applications that are perfect for voice assistant backends.
In this guide, you'll learn how to build a production-ready voice assistant backend using LangGraph. We'll cover everything from basic concepts to advanced patterns, including state management, tool integration, streaming responses, and deployment considerations.
What Is LangGraph?
LangGraph is a library built on top of LangChain that enables developers to create stateful, multi-actor applications using graph-based workflows. Unlike traditional LangChain chains that follow a linear execution path, LangGraph allows you to define complex workflows as directed graphs where nodes represent computational units and edges represent the flow of data and control.
For voice assistants, this graph-based approach is invaluable. A voice interaction often involves multiple steps: transcribing audio, understanding intent, retrieving information, executing actions, and generating spoken responses. LangGraph lets you model each of these steps as nodes in a graph, with conditional edges that adapt based on the conversation's state.
Why LangGraph for Voice Assistants?
Building a voice assistant backend presents unique challenges that LangGraph addresses effectively:
- Stateful conversations: Voice assistants must maintain context across multiple turns. LangGraph's built-in state management handles this naturally.
- Complex routing: Different user intents require different processing paths. LangGraph's conditional edges make intent-based routing straightforward.
- Tool integration: Voice assistants often need to call external APIs or databases. LangGraph seamlessly integrates with LangChain's tool ecosystem.
- Human-in-the-loop: Some actions require user confirmation. LangGraph supports pausing and resuming execution for approval workflows.
- Streaming support: Voice assistants benefit from streaming responses for lower latency. LangGraph supports streaming at multiple levels.
- Persistence: Long-running conversations need durable state storage. LangGraph provides checkpointers for persistence.
Prerequisites and Setup
Before we begin building, let's set up the development environment. You'll need Python 3.10 or later and several key packages.
# Create a virtual environment
python -m venv voice-assistant-env
source voice-assistant-env/bin/activate # On Windows: voice-assistant-env\Scripts\activate
# Install required packages
pip install langgraph langchain langchain-openai langchain-core
pip install fastapi uvicorn python-dotenv
pip install pydub SpeechRecognition
pip install langgraph-checkpoint-sqlite
# Create project structure
mkdir voice-assistant-backend
cd voice-assistant-backend
mkdir app app/nodes app/tools app/models
touch app/__init__.py app/nodes/__init__.py app/tools/__init__.py
Create a .env file in your project root with the necessary API keys:
OPENAI_API_KEY=your-openai-api-key-here
TAVILY_API_KEY=your-tavily-api-key-here
DATABASE_URL=sqlite:///./voice_assistant.db
Defining the Assistant State
The foundation of any LangGraph application is the state schema. For a voice assistant, the state needs to track the conversation history, current user input, retrieved context, and any pending actions.
# app/models/state.py
from typing import Annotated, Optional, Literal
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage
from pydantic import BaseModel, Field
class VoiceInput(BaseModel):
"""Represents processed voice input."""
transcript: str = Field(description="Transcribed text from audio input")
language: str = Field(default="en", description="Detected language code")
confidence: float = Field(default=1.0, description="Transcription confidence score")
intent: Optional[str] = Field(default=None, description="Detected user intent")
class AssistantAction(BaseModel):
"""Represents an action the assistant wants to take."""
action_type: Literal["search", "call_api", "send_message", "create_reminder"]
parameters: dict
requires_confirmation: bool = True
class AssistantState(TypedDict):
"""Main state schema for the voice assistant."""
messages: Annotated[list[BaseMessage], add_messages]
voice_input: Optional[VoiceInput]
context: Optional[str]
pending_action: Optional[AssistantAction]
response_text: Optional[str]
needs_confirmation: bool
user_id: str
session_id: str
error: Optional[str]
This state schema captures everything the assistant needs to function. The messages field uses LangGraph's add_messages reducer, which automatically appends new messages rather than replacing them. This is critical for maintaining conversation history across turns.
Building the Core Nodes
With the state defined, we can now build the individual nodes that will form our graph. Each node is a function that takes the current state and returns a partial state update.
1. Input Processing Node
The first node handles incoming voice input, transcribes it, and performs initial intent detection.
# app/nodes/input_processor.py
from app.models.state import AssistantState, VoiceInput
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
import json
async def process_input(state: AssistantState) -> dict:
"""Process raw voice input and detect intent."""
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
transcript = state["voice_input"].transcript if state.get("voice_input") else ""
if not transcript:
# Extract from latest human message if no voice_input
messages = state.get("messages", [])
if messages:
transcript = messages[-1].content
system_prompt = SystemMessage(content="""You are an intent detection system for a voice assistant.
Analyze the user's input and classify it into one of these intents:
- "search": User wants to find information
- "action": User wants to perform an action (call, message, reminder)
- "conversation": General conversation or follow-up
- "confirmation": User is confirming or denying a previous action
- "clarification": User is asking for clarification
Respond with JSON: {"intent": "intent_name", "confidence": 0.0-1.0}
""")
response = await llm.ainvoke([system_prompt, HumanMessage(content=transcript)])
try:
result = json.loads(response.content)
intent = result.get("intent", "conversation")
except json.JSONDecodeError:
intent = "conversation"
voice_input = VoiceInput(
transcript=transcript,
intent=intent,
confidence=1.0
)
return {
"voice_input": voice_input,
"messages": [HumanMessage(content=transcript)]
}
2. Context Retrieval Node
This node retrieves relevant context based on the user's query, which could include previous conversation history, user preferences, or external knowledge.
# app/nodes/context_retriever.py
from app.models.state import AssistantState
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage
async def retrieve_context(state: AssistantState) -> dict:
"""Retrieve relevant context for the current query."""
voice_input = state.get("voice_input")
if not voice_input or voice_input.intent == "conversation":
return {"context": None}
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# Summarize conversation history for context
messages = state.get("messages", [])
history_text = "\n".join([f"{m.type}: {m.content}" for m in messages[-6:]])
system_prompt = SystemMessage(content="""Extract key context from the conversation
that would be relevant for answering the user's latest query.
Focus on entities, preferences, and prior actions discussed.
Keep it concise - maximum 3 sentences.""")
response = await llm.ainvoke([
system_prompt,
HumanMessage(content=f"History:\n{history_text}\n\nCurrent query: {voice_input.transcript}")
])
return {"context": response.content}
3. Tool Execution Node
For voice assistants that can perform actions, we need a node that executes tools and APIs based on the detected intent.
# app/nodes/tool_executor.py
from app.models.state import AssistantState, AssistantAction
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langchain_core.messages import AIMessage, ToolMessage
import json
@tool
def web_search(query: str) -> str:
"""Search the web for current information."""
# In production, integrate with Tavily, SerpAPI, or similar
return f"Search results for: {query}"
@tool
def send_message(recipient: str, message: str) -> str:
"""Send a message to a contact."""
# Integrate with your messaging service
return f"Message sent to {recipient}: {message}"
@tool
def create_reminder(time: str, description: str) -> str:
"""Create a reminder for the user."""
# Integrate with calendar or reminder service
return f"Reminder set for {time}: {description}"
@tool
def get_weather(location: str) -> str:
"""Get current weather for a location."""
# Integrate with weather API
return f"Weather in {location}: Sunny, 72°F"
tools = [web_search, send_message, create_reminder, get_weather]
tools_by_name = {t.name: t for t in tools}
async def execute_tools(state: AssistantState) -> dict:
"""Execute tools based on the assistant's decision."""
messages = state.get("messages", [])
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0).bind_tools(tools)
system_prompt = SystemMessage(content="""You are a helpful voice assistant.
Use available tools when the user requests specific actions or information.
Always explain what you're doing before calling a tool.
Keep responses conversational and suitable for voice output.""")
context = state.get("context", "")
context_msg = f"\n\nRelevant context: {context}" if context else ""
response = await llm.ainvoke([system_prompt] + messages)
# If the model wants to call tools
if response.tool_calls:
tool_messages = []
for tool_call in response.tool_calls:
tool_name = tool_call["name"]
tool_args = tool_call["args"]
if tool_name in tools_by_name:
result = tools_by_name[tool_name].invoke(tool_args)
tool_messages.append(ToolMessage(
content=str(result),
tool_call_id=tool_call["id"]
))
# Get final response after tool execution
final_response = await llm.ainvoke(
[system_prompt] + messages + [response] + tool_messages
)
return {
"messages": [response] + tool_messages + [final_response],
"response_text": final_response.content
}
return {
"messages": [response],
"response_text": response.content
}
4. Response Formatter Node
Voice output requires concise, natural-sounding responses. This node formats the assistant's response for speech synthesis.
# app/nodes/response_formatter.py
from app.models.state import AssistantState
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
async def format_response(state: AssistantState) -> dict:
"""Format the response for voice output."""
response_text = state.get("response_text", "")
if not response_text:
messages = state.get("messages", [])
if messages and isinstance(messages[-1], AIMessage):
response_text = messages[-1].content
else:
response_text = "I'm sorry, I didn't understand that."
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.3)
system_prompt = SystemMessage(content="""You are formatting a response for a voice assistant.
Rewrite the input to be:
- Concise (under 3 sentences for simple queries, under 5 for complex ones)
- Natural for spoken output (no markdown, no bullet points, no URLs)
- Conversational and friendly
- Clear and unambiguous when heard aloud
Return only the formatted text, nothing else.""")
formatted = await llm.ainvoke([
system_prompt,
HumanMessage(content=response_text)
])
return {
"response_text": formatted.content,
"messages": [AIMessage(content=formatted.content)]
}
5. Confirmation Handler Node
For actions that require user confirmation, this node manages the confirmation flow.
# app/nodes/confirmation_handler.py
from app.models.state import AssistantState
from langchain_core.messages import AIMessage
async def handle_confirmation(state: AssistantState) -> dict:
"""Handle confirmation requests for sensitive actions."""
voice_input = state.get("voice_input")
if not voice_input or voice_input.intent != "confirmation":
return {}
transcript = voice_input.transcript.lower()
# Simple confirmation detection
positive_words = ["yes", "yeah", "sure", "ok", "okay", "confirm", "do it", "go ahead"]
negative_words = ["no", "nope", "cancel", "don't", "stop", "never mind"]
is_confirmed = any(word in transcript for word in positive_words)
is_denied = any(word in transcript for word in negative_words)
if is_confirmed:
return {
"needs_confirmation": False,
"response_text": "Great, proceeding with your request.",
"messages": [AIMessage(content="Great, proceeding with your request.")]
}
elif is_denied:
return {
"needs_confirmation": False,
"pending_action": None,
"response_text": "Okay, I've cancelled that action. Is there anything else I can help with?",
"messages": [AIMessage(content="Okay, I've cancelled that action. Is there anything else I can help with?")]
}
return {
"response_text": "I didn't catch that. Did you want me to proceed? Please say yes or no.",
"messages": [AIMessage(content="I didn't catch that. Did you want me to proceed? Please say yes or no.")]
}
Assembling the Graph
Now that we have all the nodes, let's wire them together into a LangGraph workflow. The graph will define how data flows between nodes and under what conditions.
# app/graph.py
from langgraph.graph import StateGraph, END
from app.models.state import AssistantState
from app.nodes.input_processor import process_input
from app.nodes.context_retriever import retrieve_context
from app.nodes.tool_executor import execute_tools
from app.nodes.response_formatter import format_response
from app.nodes.confirmation_handler import handle_confirmation
def should_handle_confirmation(state: AssistantState) -> str:
"""Determine if we need to handle a confirmation flow."""
voice_input = state.get("voice_input")
if voice_input and voice_input.intent == "confirmation":
return "confirmation"
return "process"
def route_after_processing(state: AssistantState) -> str:
"""Route based on detected intent."""
voice_input = state.get("voice_input")
if not voice_input:
return "format"
intent = voice_input.intent
if intent in ["search", "action"]:
return "tools"
elif intent == "conversation":
return "format"
elif intent == "clarification":
return "context"
else:
return "tools"
def build_assistant_graph():
"""Build and compile the voice assistant graph."""
# Create the graph
workflow = StateGraph(AssistantState)
# Add nodes
workflow.add_node("process_input", process_input)
workflow.add_node("retrieve_context", retrieve_context)
workflow.add_node("execute_tools", execute_tools)
workflow.add_node("format_response", format_response)
workflow.add_node("handle_confirmation", handle_confirmation)
# Set entry point
workflow.set_entry_point("process_input")
# Add conditional edges from input processing
workflow.add_conditional_edges(
"process_input",
should_handle_confirmation,
{
"confirmation": "handle_confirmation",
"process": "retrieve_context"
}
)
# Add conditional edges from context retrieval
workflow.add_conditional_edges(
"retrieve_context",
route_after_processing,
{
"tools": "execute_tools",
"format": "format_response",
"context": "retrieve_context"
}
)
# Tools always go to formatting after execution
workflow.add_edge("execute_tools", "format_response")
# Confirmation handler goes to formatting
workflow.add_edge("handle_confirmation", "format_response")
# Format response is the terminal node
workflow.add_edge("format_response", END)
# Compile the graph
return workflow.compile()
Adding Persistence with Memory
For a voice assistant to maintain context across sessions, we need persistent state storage. LangGraph provides checkpointers for this purpose.
# app/persistence.py
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.checkpoint.memory import MemorySaver
import sqlite3
import os
from contextlib import contextmanager
def get_checkpointer(use_sqlite: bool = True):
"""Get the appropriate checkpointer for state persistence."""
if use_sqlite:
db_path = os.getenv("DATABASE_URL", "sqlite:///./voice_assistant.db")
db_path = db_path.replace("sqlite:///", "")
conn = sqlite3.connect(db_path, check_same_thread=False)
return SqliteSaver(conn)
else:
return MemorySaver()
def build_persistent_graph():
"""Build the assistant graph with persistence."""
from app.graph import build_assistant_graph
checkpointer = get_checkpointer(use_sqlite=True)
workflow = build_assistant_graph()
return workflow.compile(checkpointer=checkpointer)
Creating the API Layer
To make our voice assistant accessible, we'll wrap it in a FastAPI application that handles HTTP requests for voice interactions.
# app/main.py
from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from langchain_core.messages import HumanMessage, AIMessage
from app.persistence import build_persistent_graph
from app.models.state import AssistantState, VoiceInput
import uuid
import json
import asyncio
app = FastAPI(title="Voice Assistant Backend", version="1.0.0")
# Build the graph once at startup
assistant_graph = build_persistent_graph()
class VoiceRequest(BaseModel):
"""Request model for voice assistant queries."""
transcript: str
user_id: str = "default_user"
session_id: str = None
language: str = "en"
class VoiceResponse(BaseModel):
"""Response model for voice assistant queries."""
response_text: str
session_id: str
intent: str = None
needs_confirmation: bool = False
@app.post("/chat", response_model=VoiceResponse)
async def chat(request: VoiceRequest):
"""Handle a voice assistant query."""
session_id = request.session_id or str(uuid.uuid4())
thread_id = f"{request.user_id}_{session_id}"
config = {"configurable": {"thread_id": thread_id}}
initial_state = {
"voice_input": VoiceInput(
transcript=request.transcript,
language=request.language
),
"user_id": request.user_id,
"session_id": session_id,
"needs_confirmation": False,
}
try:
result = await assistant_graph.ainvoke(initial_state, config=config)
return VoiceResponse(
response_text=result.get("response_text", "I'm sorry, I didn't understand that."),
session_id=session_id,
intent=result.get("voice_input").intent if result.get("voice_input") else None,
needs_confirmation=result.get("needs_confirmation", False)
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/chat/stream")
async def chat_stream(request: VoiceRequest):
"""Stream response tokens for lower latency voice output."""
session_id = request.session_id or str(uuid.uuid4())
thread_id = f"{request.user_id}_{session_id}"
config = {"configurable": {"thread_id": thread_id}}
initial_state = {
"voice_input": VoiceInput(
transcript=request.transcript,
language=request.language
),
"user_id": request.user_id,
"session_id": session_id,
"needs_confirmation": False,
}
async def event_stream():
try:
async for event in assistant_graph.astream(
initial_state,
config=config,
stream_mode="values"
):
if event.get("response_text"):
data = {
"response_text": event["response_text"],
"session_id": session_id
}
yield f"data: {json.dumps(data)}\n\n"
except Exception as e:
yield f"data: {json.dumps({'error': str(e)})}\n\n"
return StreamingResponse(
event_stream(),
media_type="text/event-stream"
)
@app.get("/health")
async def health_check():
"""Health check endpoint."""
return {"status": "healthy", "service": "voice-assistant-backend"}
@app.get("/history/{user_id}/{session_id}")
async def get_history(user_id: str, session_id: str):
"""Retrieve conversation history for a session."""
thread_id = f"{user_id}_{session_id}"
config = {"configurable": {"thread_id": thread_id}}
try:
state = await assistant_graph.aget_state(config)
if state and state.values.get("messages"):
messages = []
for msg in state.values["messages"]:
messages.append({
"role": msg.type,
"content": msg.content,
"timestamp": msg.additional_kwargs.get("timestamp")
})
return {"messages": messages, "session_id": session_id}
return {"messages": [], "session_id": session_id}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
Adding Human-in-the-Loop Confirmation
For sensitive actions like sending messages or making purchases, we want to pause execution and ask for user confirmation. LangGraph supports this through its interrupt mechanism.
# app/nodes/action_planner.py
from app.models.state import AssistantState, AssistantAction
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
from langgraph.types import interrupt, Command
import json
SENSITIVE_ACTIONS = ["send_message", "create_reminder"]
async def plan_action(state: AssistantState) -> dict:
"""Plan an action and request confirmation if needed."""
voice_input = state.get("voice_input")
if not voice_input or voice_input.intent != "action":
return {}
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
system_prompt = SystemMessage(content="""Analyze the user's request and determine
if an action needs to be taken. If so, create an action plan.
Respond with JSON:
{
"action_type": "send_message|create_reminder|call_api|search",
"parameters": {...},
"description": "Human-readable description of the action"
}
If no action is needed, respond with: {"action_type": null}
""")
response = await llm.ainvoke([
system_prompt,
HumanMessage(content=voice_input.transcript)
])
try:
plan = json.loads(response.content)
if not plan.get("action_type"):
return {}
action = AssistantAction(
action_type=plan["action_type"],
parameters=plan.get("parameters", {}),
requires_confirmation=plan["action_type"] in SENSITIVE_ACTIONS
)
if action.requires_confirmation:
# Interrupt execution and ask for confirmation
confirmation_prompt = f"I'm about to {plan.get('description', 'perform an action')}. Would you like me to proceed?"
user_response = interrupt({
"prompt": confirmation_prompt,
"action": action.dict()
})
# Process the user's confirmation response
if isinstance(user_response, dict) and user_response.get("confirmed"):
return {
"pending_action": action,
"needs_confirmation": False,
"messages": [AIMessage(content=f"Proceeding with the action.")]
}
else:
return {
"pending_action": None,
"needs_confirmation": False,
"messages": [AIMessage(content="Okay, I've cancelled that action.")]
}
return {"pending_action": action}
except json.JSONDecodeError:
return {}
To use the interrupt mechanism, you need to compile the graph with a checkpointer and handle the interrupt in your API layer:
# app/main.py (updated chat endpoint with interrupt support)
from langgraph.types import Command
@app.post("/chat", response_model=VoiceResponse)
async def chat(request: VoiceRequest):
"""Handle a voice assistant query with interrupt support."""
session_id = request.session_id or str(uuid.uuid4())
thread_id = f"{request.user_id}_{session_id}"
config = {"configurable": {"thread_id": thread_id}}
# Check if there's a pending interrupt
state = await assistant_graph.aget_state(config)
if state and state.next:
# Resume from interrupt with user's response
confirmed = "yes" in request.transcript.lower() or "confirm" in request.transcript.lower()
result = await assistant_graph.ainvoke(
Command(resume={"confirmed": confirmed}),
config=config
)
else:
# Start new conversation
initial_state = {
"voice_input": VoiceInput(
transcript=request.transcript,
language=request.language
),
"user_id": request.user_id,
"session_id": session_id,
"needs_confirmation": False,
}
result = await assistant_graph.ainvoke(initial_state, config=config)
# Check if execution was interrupted
state_after = await assistant_graph.aget_state(config)
if state_after and state_after.next:
# Execution was interrupted, return the prompt
interrupt_data = state_after.tasks[0].interrupt
return VoiceResponse(
response_text=interrupt_data.get("prompt", "Please confirm."),
session_id=session_id,
needs_confirmation=True
)
return VoiceResponse(
response_text=result.get("response_text", "I'm sorry, I didn't understand that."),
session_id=session_id,
intent=result.get("voice_input").intent if result.get("voice_input") else None,
needs_confirmation=False
)
Integrating Speech-to-Text and Text-to-Speech
While the core logic handles text, a complete voice assistant needs STT and TTS integration. Here's how to add audio processing endpoints.
# app/audio.py
from fastapi import UploadFile, File, APIRouter
from fastapi.responses import StreamingResponse
import io
import base64
audio_router = APIRouter()
@audio_router.post("/transcribe")
async def transcribe_audio(file: UploadFile = File(...)):
"""Convert audio to text using speech-to-text."""
import openai
audio_data = await file.read()
audio_buffer = io.BytesIO(audio_data)
audio_buffer.name = "input.wav"
try:
result = openai.audio.transcriptions.create(
model="whisper-1",
file=audio_buffer
)
return {"transcript": result.text, "success": True}
except Exception as e:
return {"transcript": "", "success": False, "error": str(e)}
@audio_router.post("/synthesize")
async def synthesize_speech(text: str):
"""Convert text to audio using text-to-speech."""
import openai
try:
response = openai.audio.speech.create(
model="tts-1",
voice="alloy",
input=text
)
audio_bytes = response.content
return StreamingResponse(
io.BytesIO(audio_bytes),
media_type="audio/mpeg",
headers={"Content-Disposition": "attachment; filename=speech.mp3"}
)
except Exception as e:
return {"success": False, "error": str(e)}
Add the audio router to your main application:
# In app/main.py, add:
from app.audio import audio_router
app.include_router(audio_router, prefix="/audio", tags=["audio"])
Testing the Voice Assistant
Let's write tests to verify our assistant works correctly across different scenarios.
# tests/test_assistant.py
import pytest
from app.graph import build_assistant_graph
from app.models.state import AssistantState, VoiceInput
@pytest.fixture
def graph():
return build_assistant_graph()
@pytest.fixture
def config():
return {"configurable": {"thread_id": "test_session_1"}}
@pytest.mark.asyncio
async def test_simple_conversation(graph, config):
"""Test a simple conversational query."""
state = {
"voice_input": VoiceInput(transcript="Hello, how are you?"),
"user_id": "test_user",
"session_id": "test_session_1",
"needs_confirmation": False,
}
result = await graph.ainvoke(state, config=config)
assert result.get("response_text") is not None
assert len(result["response_text"]) > 0
assert "messages" in result
@pytest.mark.asyncio
async def test_search_intent(graph, config):
"""Test that search queries route correctly."""
state = {
"voice_input": VoiceInput(transcript="What's the weather in San Francisco?"),
"user_id": "test_user",
"session_id": "test_session_2",
"needs_confirmation": False,
}
result = await graph.ainvoke(state, config=config)
assert result.get("response_text") is not None
assert "messages" in result
@pytest.mark.asyncio
async def test_context_preservation(graph, config):
"""Test that context is preserved across turns."""
# First turn
state1 = {
"voice_input": VoiceInput(transcript="My name is Alice."),
"user_id": "test_user",
"session_id": "test_session_3",
"needs_confirmation": False,
}
await graph.ainvoke(state1, config=config)
# Second turn - should remember the name
state2 = {
"voice_input": VoiceInput(transcript="What's my name?"),
"user_id": "test_user",
"session_id": "test_session_3",
"needs_confirmation": False,
}
result = await graph.ainvoke(state2, config=config)
assert "Alice" in result.get("response_text", "")
@pytest.mark.asyncio
async def test_error_handling(graph, config):
"""Test graceful error handling."""
state = {
"voice_input": VoiceInput(transcript=""),
"user_id": "test_user",
"session_id": "test_session_4",
"needs_confirmation": False,
}
result = await graph.ainvoke(state, config=config)
assert result.get("response_text") is not None
Best Practices
As you build and deploy your voice assistant backend, keep these best practices in mind:
- Keep responses concise: Voice interactions demand brevity. Users cannot scan text—they must listen to every word. Aim for responses under three sentences for simple queries.
- Design for ambiguity: Voice input is inherently noisy. Always have fallback responses and clarification prompts ready for when transcription confidence is low or intent is unclear.
- Use streaming aggressively: Stream both the graph execution and the LLM token generation. Even a 200ms reduction in time-to-first-audio significantly improves perceived responsiveness.
- Implement proper session management: Use unique thread IDs per user session and clean up old sessions periodically. LangGraph's checkpointing can accumulate significant data over time.
- Handle interruptions gracefully: Users may interrupt the assistant mid-response. Design your frontend to cancel ongoing TTS playback and start processing new input immediately.
- Log conversation flows: Log the full graph execution path, not just inputs and outputs. This helps debug routing issues and identify where conversations break down.
- Rate limit aggressively: Voice interfaces can generate rapid-fire inputs. Implement rate limiting at the API level to prevent abuse and control costs.
- Test with real audio: Don't just test with text input. Real audio introduces transcription errors, background noise, and partial words that your pipeline must handle.
- Version your graphs: As you modify the graph structure, maintain version compatibility. Old sessions may have state schemas that don't match new graph definitions.
- Monitor token usage: Voice assistants can consume tokens quickly due to context accumulation. Monitor usage and implement context windowing or summarization for long conversations.
Deployment Considerations
When deploying your voice assistant backend to production, consider these additional factors:
# Dockerfile
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 8000
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]
# requirements.txt
langgraph>=0.2.0
langchain>=0.3.0
langchain-openai>=0.2.0
langchain-core>=0.3.0
langgraph-checkpoint-sqlite>=2.0.0
fastapi>=0.115.0
uvicorn>=0.30.0
python-dotenv>=1.0.0
pydantic>=2.0.0
openai>=1.50.0
pytest>=8.0.0
pytest-asyncio>=0.24.0
# docker-compose.yml
version: '3.8'
services:
assistant:
build: .
ports:
- "8000:8000"
environment:
- OPENAI_API_KEY=${OPENAI_API_KEY}
- DATABASE_URL=sqlite:///./data/voice_assistant.db
volumes:
- ./data:/app/data
restart: unless-stopped
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
interval: 30s
timeout: 10s
retries: 3
Conclusion
Building a voice assistant backend with LangGraph gives you a powerful, flexible foundation for creating sophisticated conversational experiences. The graph-based architecture naturally models the complex, multi-step workflows that voice interactions demand, while built-in state management, persistence, and human-in-the-loop capabilities handle the challenges that make voice assistants difficult to build with traditional frameworks. By following the patterns outlined in this guide—thoughtful state design, modular node construction, conditional routing, and careful attention to voice-specific concerns like response brevity and streaming latency—you can build a voice assistant that feels natural, responsive, and genuinely helpful to your users. As you iterate on your assistant, leverage LangGraph's observability features to trace execution paths, identify bottlenecks, and continuously refine the conversation flow for the best possible user experience.