Introduction to Building a Voice Assistant Backend with AutoGen
Voice assistants have evolved from simple command-response systems into sophisticated conversational agents capable of handling complex, multi-step tasks. Microsoft's AutoGen framework provides a powerful foundation for building these systems by enabling multiple AI agents to collaborate, reason, and execute tasks autonomously. In this tutorial, you'll learn how to build a production-ready voice assistant backend that combines speech-to-text, multi-agent orchestration, and text-to-speech into a cohesive system.
What Is AutoGen and Why It Matters for Voice Assistants
AutoGen is an open-source framework developed by Microsoft Research for building multi-agent conversational AI systems. Unlike traditional single-model approaches, AutoGen allows you to define specialized agentsâeach with distinct roles, tools, and personalitiesâthat communicate with each other to solve complex problems.
Why AutoGen Is Ideal for Voice Assistants
- Multi-agent orchestration: Route user queries to specialized agents (e.g., a scheduler agent, a knowledge agent, a weather agent) for more accurate responses.
- Tool integration: Agents can call external APIs, databases, and functions seamlessly, enabling real-world actions like booking appointments or querying live data.
- Conversational memory: Built-in support for maintaining context across turns, which is essential for natural voice interactions.
- Human-in-the-loop: AutoGen supports human proxy agents, allowing for confirmation steps before executing sensitive actions.
- Scalability: The agent-based architecture makes it easy to add new capabilities without rewriting core logic.
For voice assistants specifically, AutoGen solves a critical problem: voice interactions demand fast, accurate, and context-aware responses. By distributing responsibilities across specialized agents, you reduce the cognitive load on any single model and improve both latency and response quality.
Architecture Overview
Before diving into code, let's outline the architecture of our voice assistant backend:
- Speech-to-Text (STT) layer: Converts incoming audio to text using a service like Whisper or Azure Speech.
- AutoGen orchestration layer: Processes the transcribed text through a group of specialized agents.
- Text-to-Speech (TTS) layer: Converts the final agent response back to audio for the user.
- API gateway: Exposes WebSocket endpoints for real-time streaming audio communication.
- Session manager: Maintains conversation state and memory per user session.
Prerequisites and Setup
You'll need Python 3.10 or higher, an OpenAI API key, and optionally an Azure Speech resource for STT/TTS. Start by creating a project directory and installing the required dependencies.
# Create project directory
mkdir voice-assistant-autogen
cd voice-assistant-autogen
# Create virtual environment
python -m venv venv
source venv/bin/activate # On Windows: venv\Scripts\activate
# Install dependencies
pip install pyautogen openai fastapi uvicorn websockets python-dotenv
pip install openai-whisper # For local STT (optional)
pip install azure-cognitiveservices-speech # For Azure TTS (optional)
Create a .env file to manage your configuration:
# .env
OPENAI_API_KEY=sk-your-openai-api-key-here
AZURE_SPEECH_KEY=your-azure-speech-key
AZURE_SPEECH_REGION=eastus
MODEL_NAME=gpt-4o
Step 1: Defining Your Agents
The core of any AutoGen application is the agent definitions. For our voice assistant, we'll create three specialized agents: a coordinator that routes requests, a knowledge agent that answers factual questions, and an action agent that executes tasks like setting reminders or checking calendars.
# agents.py
import os
from autogen import AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager
from dotenv import load_dotenv
load_dotenv()
config_list = [{
"model": os.getenv("MODEL_NAME", "gpt-4o"),
"api_key": os.getenv("OPENAI_API_KEY"),
}]
llm_config = {
"config_list": config_list,
"temperature": 0.7,
"timeout": 30,
}
def create_coordinator_agent():
"""Routes user requests to the appropriate specialist agent."""
return AssistantAgent(
name="Coordinator",
system_message=(
"You are the coordinator of a voice assistant team. "
"Your job is to analyze the user's request and determine which "
"specialist agent should handle it. "
"For factual questions or general knowledge, forward to 'KnowledgeAgent'. "
"For task execution (reminders, scheduling, actions), forward to 'ActionAgent'. "
"If you can answer directly for simple greetings or confirmations, do so. "
"Keep responses concise and conversational since they will be spoken aloud."
),
llm_config=llm_config,
)
def create_knowledge_agent():
"""Handles factual questions and general knowledge queries."""
return AssistantAgent(
name="KnowledgeAgent",
system_message=(
"You are a knowledgeable assistant specializing in answering "
"factual questions clearly and concisely. "
"Since your responses will be converted to speech, avoid markdown, "
"code blocks, or special characters. "
"Keep answers under 3 sentences unless the user asks for detail. "
"Use a natural, conversational tone."
),
llm_config=llm_config,
)
def create_action_agent():
"""Executes tasks like reminders and scheduling."""
return AssistantAgent(
name="ActionAgent",
system_message=(
"You are an action-oriented assistant that helps users with tasks "
"like setting reminders, checking schedules, and performing actions. "
"Always confirm important details before executing. "
"Use the provided tools when available. "
"Respond in a concise, spoken-friendly format."
),
llm_config=llm_config,
)
def create_user_proxy():
"""Represents the human user in the conversation."""
return UserProxyAgent(
name="User",
human_input_mode="NEVER",
max_consecutive_auto_reply=0,
is_termination_msg=lambda msg: msg.get("content") is not None and "TERMINATE" in msg["content"],
)
Step 2: Building the Group Chat Manager
AutoGen's GroupChat enables multiple agents to participate in a single conversation. The GroupChatManager coordinates turn-taking and message routing between agents.
# chat_manager.py
from agents import (
create_coordinator_agent,
create_knowledge_agent,
create_action_agent,
create_user_proxy,
)
from autogen import GroupChat, GroupChatManager
from agents import llm_config
def create_group_chat():
"""Create and configure the multi-agent group chat."""
coordinator = create_coordinator_agent()
knowledge = create_knowledge_agent()
action = create_action_agent()
user_proxy = create_user_proxy()
groupchat = GroupChat(
agents=[user_proxy, coordinator, knowledge, action],
messages=[],
max_round=8,
speaker_selection_method="auto",
)
manager = GroupChatManager(
groupchat=groupchat,
llm_config=llm_config,
)
return user_proxy, manager
def process_message(user_proxy, manager, message: str) -> str:
"""Process a user message through the agent group chat."""
user_proxy.initiate_chat(
manager,
message=message,
clear_history=False,
)
# Extract the last assistant response
last_message = None
for msg in reversed(manager.groupchat.messages):
if msg.get("role") == "assistant" and msg.get("name") != "User":
last_message = msg["content"]
break
return last_message or "I'm sorry, I didn't catch that."
Step 3: Adding Tools for the Action Agent
To make the voice assistant genuinely useful, the action agent needs tools to interact with the real world. Let's add a reminder tool and a weather lookup tool.
# tools.py
import json
from datetime import datetime
from typing import Annotated
# In-memory storage for demo purposes.
# In production, use a database like Redis or PostgreSQL.
reminders_db = []
def set_reminder(
task: Annotated[str, "The task or event to remind about"],
time: Annotated[str, "When to remind, in ISO format (e.g., 2024-01-15T14:30:00)"]
) -> str:
"""Set a reminder for the user at a specified time."""
try:
reminder_time = datetime.fromisoformat(time)
except ValueError:
return "I couldn't understand that time format. Please try again."
reminder = {
"task": task,
"time": reminder_time.isoformat(),
"created_at": datetime.now().isoformat(),
}
reminders_db.append(reminder)
return f"Reminder set: {task} at {reminder_time.strftime('%I:%M %p on %B %d')}."
def get_reminders() -> str:
"""Retrieve all pending reminders."""
if not reminders_db:
return "You have no pending reminders."
result = "Here are your reminders: "
for i, r in enumerate(reminders_db, 1):
reminder_time = datetime.fromisoformat(r["time"])
result += f"{i}. {r['task']} at {reminder_time.strftime('%I:%M %p on %B %d')}. "
return result
def get_weather(
location: Annotated[str, "City name to get weather for"]
) -> str:
"""Get current weather for a location (mock implementation)."""
# In production, call a real weather API like OpenWeatherMap
mock_weather = {
"New York": "72 degrees and sunny",
"London": "55 degrees and rainy",
"Tokyo": "68 degrees and cloudy",
}
weather = mock_weather.get(location, "Weather data not available for that location.")
return f"The current weather in {location} is {weather}."
Now register these tools with the action agent:
# register_tools.py
from agents import create_action_agent
from tools import set_reminder, get_reminders, get_weather
def get_action_agent_with_tools():
action_agent = create_action_agent()
# Register tools with the agent
action_agent.register_for_llm(
name="set_reminder",
description="Set a reminder for the user at a specific time."
)(set_reminder)
action_agent.register_for_llm(
name="get_reminders",
description="Retrieve all pending reminders for the user."
)(get_reminders)
action_agent.register_for_llm(
name="get_weather",
description="Get the current weather for a given city."
)(get_weather)
return action_agent
Step 4: Implementing Speech-to-Text and Text-to-Speech
The voice layer bridges the gap between audio input/output and the text-based agent system. We'll use OpenAI's Whisper for STT and a simple TTS solution.
# voice.py
import os
import io
import wave
import tempfile
from dotenv import load_dotenv
from openai import OpenAI
load_dotenv()
openai_client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
def transcribe_audio(audio_bytes: bytes, sample_rate: int = 16000) -> str:
"""Convert audio bytes to text using OpenAI Whisper."""
# Write audio bytes to a temporary WAV file
with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmp:
with wave.open(tmp.name, "wb") as wav_file:
wav_file.setnchannels(1)
wav_file.setsampwidth(2)
wav_file.setframerate(sample_rate)
wav_file.writeframes(audio_bytes)
tmp_path = tmp.name
try:
with open(tmp_path, "rb") as audio_file:
transcript = openai_client.audio.transcriptions.create(
model="whisper-1",
file=audio_file,
language="en",
)
return transcript.text.strip()
finally:
os.unlink(tmp_path)
def synthesize_speech(text: str, output_path: str = None) -> bytes:
"""Convert text to speech using OpenAI TTS."""
response = openai_client.audio.speech.create(
model="tts-1",
voice="nova",
input=text,
)
audio_bytes = response.content
if output_path:
with open(output_path, "wb") as f:
f.write(audio_bytes)
return audio_bytes
Step 5: Building the Session Manager
Each user needs their own conversation context. The session manager maintains separate agent instances and conversation history per user.
# session_manager.py
import uuid
from typing import Dict, Optional
from chat_manager import create_group_chat, process_message
class SessionManager:
"""Manages per-user conversation sessions."""
def __init__(self):
self._sessions: Dict[str, dict] = {}
def create_session(self) -> str:
"""Create a new user session and return the session ID."""
session_id = str(uuid.uuid4())
user_proxy, manager = create_group_chat()
self._sessions[session_id] = {
"user_proxy": user_proxy,
"manager": manager,
"message_count": 0,
}
return session_id
def get_session(self, session_id: str) -> Optional[dict]:
"""Retrieve an existing session by ID."""
return self._sessions.get(session_id)
def process_user_input(self, session_id: str, text: str) -> str:
"""Process user text input and return the assistant's response."""
session = self._sessions.get(session_id)
if not session:
# Auto-create session if it doesn't exist
session_id = self.create_session()
session = self._sessions[session_id]
response = process_message(
session["user_proxy"],
session["manager"],
text,
)
session["message_count"] += 1
return response
def end_session(self, session_id: str):
"""Clean up a user session."""
if session_id in self._sessions:
del self._sessions[session_id]
# Global session manager instance
session_manager = SessionManager()
Step 6: Creating the WebSocket API Server
For real-time voice interaction, we'll use WebSockets to stream audio between the client and server. FastAPI provides excellent WebSocket support.
# server.py
import json
import asyncio
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.middleware.cors import CORSMiddleware
from session_manager import session_manager
from voice import transcribe_audio, synthesize_speech
app = FastAPI(title="Voice Assistant Backend")
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["*"],
allow_headers=["*"],
)
@app.websocket("/ws/voice")
async def voice_websocket(websocket: WebSocket):
"""
WebSocket endpoint for real-time voice communication.
Protocol:
1. Client sends JSON: {"type": "start_session"}
2. Server responds: {"type": "session_started", "session_id": "..."}
3. Client sends binary audio data (PCM 16-bit, 16kHz, mono)
4. Client sends JSON: {"type": "end_audio"}
5. Server responds: {"type": "audio_response", "audio": ""}
"""
await websocket.accept()
try:
# Start session
init_msg = await websocket.receive_text()
init_data = json.loads(init_msg)
if init_data.get("type") != "start_session":
await websocket.send_json({"type": "error", "message": "Expected start_session"})
return
session_id = session_manager.create_session()
await websocket.send_json({
"type": "session_started",
"session_id": session_id,
})
while True:
# Receive audio data
audio_chunks = []
while True:
msg = await websocket.receive()
if "text" in msg:
data = json.loads(msg["text"])
if data.get("type") == "end_audio":
break
elif data.get("type") == "end_session":
session_manager.end_session(session_id)
await websocket.send_json({"type": "session_ended"})
return
elif "bytes" in msg:
audio_chunks.append(msg["bytes"])
if not audio_chunks:
continue
# Combine audio chunks
audio_data = b"".join(audio_chunks)
# Transcribe audio to text
try:
transcribed_text = transcribe_audio(audio_data)
except Exception as e:
await websocket.send_json({
"type": "error",
"message": f"Transcription failed: {str(e)}",
})
continue
if not transcribed_text:
await websocket.send_json({
"type": "error",
"message": "Could not understand audio.",
})
continue
await websocket.send_json({
"type": "transcription",
"text": transcribed_text,
})
# Process through AutoGen agents
try:
response_text = await asyncio.get_event_loop().run_in_executor(
None,
session_manager.process_user_input,
session_id,
transcribed_text,
)
except Exception as e:
response_text = f"I encountered an error: {str(e)}"
await websocket.send_json({
"type": "response_text",
"text": response_text,
})
# Synthesize speech from response
try:
audio_response = synthesize_speech(response_text)
import base64
audio_b64 = base64.b64encode(audio_response).decode("utf-8")
await websocket.send_json({
"type": "audio_response",
"audio": audio_b64,
"format": "mp3",
})
except Exception as e:
await websocket.send_json({
"type": "error",
"message": f"Speech synthesis failed: {str(e)}",
})
except WebSocketDisconnect:
print(f"Client disconnected from session {session_id}")
session_manager.end_session(session_id)
except Exception as e:
print(f"WebSocket error: {e}")
session_manager.end_session(session_id)
@app.get("/health")
async def health_check():
"""Health check endpoint."""
return {"status": "healthy", "active_sessions": len(session_manager._sessions)}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
Step 7: Running and Testing the Server
Start the server with the following command:
python server.py
You should see output indicating the server is running on port 8000. You can test the health endpoint:
curl http://localhost:8000/health
# Expected: {"status":"healthy","active_sessions":0}
For a quick text-based test without audio, create a test script:
# test_assistant.py
from session_manager import session_manager
def main():
session_id = session_manager.create_session()
print(f"Session created: {session_id}\n")
test_inputs = [
"Hello, what can you help me with?",
"What's the weather like in New York?",
"Set a reminder to call mom at 2024-12-25T10:00:00",
"What reminders do I have?",
]
for user_input in test_inputs:
print(f"User: {user_input}")
response = session_manager.process_user_input(session_id, user_input)
print(f"Assistant: {response}\n")
print("-" * 60)
session_manager.end_session(session_id)
if __name__ == "__main__":
main()
Best Practices for Production Voice Assistants
1. Optimize for Latency
Voice interactions require sub-second response times to feel natural. Consider these optimizations:
- Use streaming STT and TTS instead of waiting for complete audio chunks.
- Cache frequent responses for common queries like greetings and confirmations.
- Use a faster model (e.g., GPT-4o-mini) for the coordinator agent and reserve larger models for complex reasoning.
- Implement parallel agent execution when queries don't depend on each other.
2. Handle Errors Gracefully
Voice assistants must handle failures without leaving users in silence. Always implement fallback responses:
# error_handling.py
from functools import wraps
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def with_fallback(fallback_message="I'm sorry, I didn't catch that. Could you repeat?"):
"""Decorator that catches exceptions and returns a spoken-friendly fallback."""
def decorator(func):
@wraps(func)
async def wrapper(*args, **kwargs):
try:
return await func(*args, **kwargs)
except Exception as e:
logger.error(f"Error in {func.__name__}: {e}", exc_info=True)
return fallback_message
return wrapper
return decorator
3. Implement Conversation Memory Persistence
In production, store conversation history in a database so users can resume sessions across restarts:
# persistence.py
import json
from datetime import datetime
from typing import List, Dict
class ConversationStore:
"""Persist conversation history to a database (mock implementation)."""
def __init__(self):
# In production, use Redis, PostgreSQL, or MongoDB
self._store: Dict[str, List[dict]] = {}
def save(self, session_id: str, messages: List[dict]):
self._store[session_id] = messages
def load(self, session_id: str) -> List[dict]:
return self._store.get(session_id, [])
def get_recent_sessions(self, user_id: str, limit: int = 10):
"""Retrieve recent sessions for a user."""
return list(self._store.keys())[-limit:]
4. Add Rate Limiting and Authentication
Protect your backend from abuse with proper authentication and rate limiting:
# middleware.py
from fastapi import Request, HTTPException
from collections import defaultdict, deque
from time import time
class RateLimiter:
"""Simple in-memory rate limiter."""
def __init__(self, max_requests: int = 30, window_seconds: int = 60):
self.max_requests = max_requests
self.window = window_seconds
self.requests = defaultdict(deque)
def check(self, client_id: str):
now = time()
reqs = self.requests[client_id]
# Remove old entries
while reqs and reqs[0] < now - self.window:
reqs.popleft()
if len(reqs) >= self.max_requests:
raise HTTPException(
status_code=429,
detail="Too many requests. Please slow down."
)
reqs.append(now)
rate_limiter = RateLimiter(max_requests=30, window_seconds=60)
5. Monitor and Log Agent Interactions
Understanding how agents interact helps debug issues and improve response quality. Log every agent message with metadata:
# monitoring.py
import logging
import json
from datetime import datetime
logger = logging.getLogger("voice_assistant")
def log_agent_interaction(session_id: str, agent_name: str, message: str,
metadata: dict = None):
"""Log agent interactions for monitoring and debugging."""
log_entry = {
"timestamp": datetime.now().isoformat(),
"session_id": session_id,
"agent": agent_name,
"message": message,
"metadata": metadata or {},
}
logger.info(json.dumps(log_entry))
6. Design for Spoken Output
Text that reads well on screen doesn't always sound natural when spoken. Follow these guidelines in your agent system prompts:
- Forbid markdown, code blocks, and special characters in responses.
- Keep responses under 3 sentences for most interactions.
- Use contractions ("I'll" instead of "I will") for natural speech.
- Spell out symbols and abbreviations ("dollar sign" instead of "$").
- Avoid numbered lists in spoken responses; use transitional phrases instead.
Extending the Assistant
Once you have the core system running, you can extend it with additional capabilities:
- Add a RAG agent: Connect a retrieval-augmented generation agent that queries your internal knowledge base or documents.
- Add a code execution agent: Use AutoGen's code execution capabilities to let the assistant run calculations or data analysis.
- Add multi-language support: Configure Whisper for auto-detection and add language-specific agents.
- Add voice activity detection (VAD): Use libraries like Silero VAD to detect when the user starts and stops speaking, eliminating the need for manual "end_audio" signals.
- Add emotion detection: Analyze the user's tone to adjust the assistant's response style accordingly.
Conclusion
Building a voice assistant backend with AutoGen gives you a flexible, multi-agent architecture that can handle complex user requests far more effectively than a single-model approach. By separating concerns into specialized agentsâcoordinator, knowledge, and actionâyou create a system that is easier to debug, extend, and maintain. The combination of AutoGen's orchestration with real-time WebSocket communication, speech-to-text, and text-to-speech creates a complete pipeline from spoken input to spoken output. As you move toward production, focus on latency optimization, error handling, conversation persistence, and monitoring to ensure a smooth and reliable user experience. The architecture presented here provides a solid foundation that you can iteratively enhance with additional agents, tools, and integrations as your assistant's capabilities grow.