Phase 4: LLM Orchestration for Voice · 50 min · Python · OpenAI SDK · NeMo Guardrails
The Concept
RAG for Voice Agents
| RAG Component | Latency | Purpose |
|---|
| Query embedding | 20-50ms | Convert question to vector |
| Vector search | 10-30ms | Find relevant documents |
| Context assembly | 5-10ms | Add docs to LLM prompt |
| Total RAG overhead | 35-90ms | Added to LLM latency |
Voice-Specific RAG Considerations
| Factor | Text Chatbot | Voice Agent |
|---|
| Latency budget | 1-2s | <90ms for RAG |
| Documents retrieved | 5-10 | 2-3 (fewer = faster) |
| Chunk size | 500-1000 tokens | 200-400 tokens (shorter) |
| Response style | Can cite sources | Must speak naturally |
| Fallback | "I don't know" | "Let me connect you with someone" |
Why should voice RAG retrieve fewer documents (2-3) than text chatbot RAG (5-10)?
Voice agents have a strict 500ms latency budget. RAG adds 35-90ms overhead (embedding + search + assembly). Each retrieved document adds tokens to the LLM input, increasing both cost and TTFT. With 2-3 chunks of 200-400 tokens, the LLM gets enough context (600-1200 tokens) to answer accurately. With 5-10 chunks (3000+ tokens), the LLM takes longer to process, increasing TTFT beyond the budget. Voice RAG optimizes for speed: fewer, smaller, more relevant chunks.
Guardrails for Voice Agents
| Guardrail Type | What It Prevents | Implementation |
|---|
| Content filter | Profanity, harmful content | LLM moderation API or keyword filter |
| Topic restriction | Off-topic conversations | System prompt + classification |
| PII protection | Sharing sensitive data | Regex filter on LLM output |
| Hallucination prevention | Making up facts | RAG + confidence threshold |
| Price/commitment guard | Promising discounts | Tool validation before confirming |
| Escalation trigger | Frustrated callers | Sentiment detection + transfer |
Guardrail Implementation
class VoiceGuardrails:
def __init__(self):
self.blocked_topics = ["politics", "religion", "competitors"]
self.pii_patterns = [
r'\b\d{3}-\d{2}-\d{4}\b', # SSN
r'\b\d{16}\b', # Credit card
r'\b\d{3}-\d{3}-\d{4}\b', # Phone
]
def check_response(self, response):
"""Check LLM response before sending to TTS."""
# Check for PII
for pattern in self.pii_patterns:
if re.search(pattern, response):
return {"blocked": True, "reason": "PII detected",
"replacement": "I can't share that information."}
# Check for blocked topics
for topic in self.blocked_topics:
if topic in response.lower():
return {"blocked": True, "reason": f"Topic: {topic}",
"replacement": "I can't discuss that topic."}
return {"blocked": False, "response": response}
Cost Optimization Strategies
| Strategy | Savings | Impact on Quality |
|---|
| Use GPT-4o-mini instead of GPT-4o | 17x | Minimal for short responses |
| Context window management | 65% | None (summarization preserves info) |
| Cache common responses | 50-80% | None (same response) |
| Batch tool calls | 30% | None (parallel execution) |
| Semantic caching | 40-60% | Minimal (similar questions) |
| Reduce RAG chunks | 20-30% | Slight (fewer docs) |
| Self-host Llama 3.1 8B | 100% | Moderate (lower quality) |
Semantic Caching
class SemanticCache:
"""Cache LLM responses for semantically similar questions."""
def __init__(self, similarity_threshold=0.95):
self.cache = {} # embedding -> response
self.threshold = similarity_threshold
async def get_or_generate(self, query, generate_fn):
# Check cache for similar query
query_embedding = await embed(query)
for cached_embedding, cached_response in self.cache.items():
similarity = cosine_similarity(query_embedding, cached_embedding)
if similarity > self.threshold:
return cached_response # Cache hit!
# Cache miss — generate new response
response = await generate_fn(query)
self.cache[query_embedding] = response
return response
Unlock the full lesson
You've read the first 2 sections. The rest of this lesson covers Build, Check Yourself, Key Terms & Next — plus a hands-on lab, quiz, and project artifact.
Create a free account — the opening phases of 29 of 35 courses are free, no credit card.