LangChain for Voice Apps: Combining STT, LLM and TTS

Why voice apps need more than "just an LLM"

Everyone wants to build the next voice assistant until they realize a chatbot demo and a real-time voice app are two completely different engineering problems. A text chatbot only has to worry about one thing: generating a good response. A voice app has to listen, transcribe, reason, decide, generate, and speak — all while a human is sitting there waiting, in real time, for something that sounds natural rather than robotic.

This is where most tutorials fall short. They show you how to call an OpenAI Whisper endpoint, paste the transcript into a prompt, and play back a text-to-speech response. That works for a demo. It falls apart the moment you need memory across turns, tool calls mid-conversation, interruption handling, or streaming responses so your user isn't sitting in dead silence for four seconds.

LangChain earns its keep here not because it does speech-to-text or text-to-speech itself — it doesn't — but because it gives you the orchestration layer that sits between STT and TTS: chains, memory, agents, streaming callbacks, and a consistent interface across LLM providers. In this article we're going to build a real STT-to-LLM-to-TTS pipeline, talk about where the actual engineering difficulty lives (hint: it's not the LLM call), and cover the patterns you need for latency, interruption, and state management. If you're building anything from a voice-based customer support bot to an in-car assistant, this is the architecture you'll end up converging on anyway — so let's get there directly.

The three-stage pipeline, conceptually

Before writing code, it helps to be precise about what each stage owns:

• STT (speech-to-text): converts raw audio (a stream of bytes) into text. Options include OpenAI's Whisper API, Deepgram, AssemblyAI, or a locally hosted faster-whisper model. STT quality determines how much garbage your LLM has to deal with — bad transcription cascades into bad answers.
• LLM orchestration (LangChain): takes the transcript, combines it with conversation memory, retrieved context (if you're doing RAG), and any tools the model needs (order lookup, calendar, database queries), and produces a response — ideally as a token stream, not a single blocking call.
• TTS (text-to-speech): converts the LLM's text output back into audio. Options include ElevenLabs, Azure Neural TTS, Google Cloud TTS, or OpenAI's TTS endpoint. The key requirement here is that TTS must be able to consume a *stream* of text, not wait for the full response, or your users will experience multi-second dead air.

The mistake engineers make is treating these as three sequential, blocking function calls: transcribe() -> generate() -> speak(). In a synchronous world that's a 3-8 second round trip before the user hears anything. The fix is to stream at every boundary: stream audio into STT as it's captured, stream tokens out of the LLM as they're generated, and stream those token chunks into TTS as sentences complete. LangChain's streaming callback system is precisely the piece that makes the middle step possible without you hand-rolling a token buffer from scratch.

Setting up your environment

Let's get the scaffolding in place. You'll need LangChain, an LLM provider SDK, and clients for your chosen STT/TTS providers.

pip install langchain langchain-openai langchain-community
pip install openai sounddevice numpy websockets
pip install elevenlabs

A minimal .env file:

OPENAI_API_KEY=sk-your-key-here
ELEVENLABS_API_KEY=your-elevenlabs-key

We'll use OpenAI for both STT (Whisper) and the LLM, and ElevenLabs for TTS, but the architecture doesn't care which vendor sits in each slot — that's the entire point of putting LangChain in the middle.

Stage 1: Speech-to-text with streaming capture

The naive approach records a fixed-length audio clip, saves it to disk, and sends the whole file to Whisper. That works for a voice memo app. For conversational voice apps, you need voice activity detection (VAD) so you know when the user has *stopped* talking, and you want to start transcribing as early as possible.

Here's a practical capture-and-transcribe function using sounddevice for audio capture and OpenAI's Whisper endpoint for transcription:

import sounddevice as sd
import numpy as np
import queue
import tempfile
import wave
from openai import OpenAI

client = OpenAI()

def record_until_silence(threshold=500, silence_duration=1.2, samplerate=16000):
    """Records audio until the user stops speaking for `silence_duration` seconds."""
    q = queue.Queue()
    frames = []
    silent_chunks = 0
    chunk_duration = 0.1
    max_silent_chunks = int(silence_duration / chunk_duration)

    def callback(indata, frame_count, time_info, status):
        q.put(indata.copy())

    with sd.InputStream(samplerate=samplerate, channels=1, dtype="int16", callback=callback):
        while True:
            chunk = q.get()
            frames.append(chunk)
            volume = np.abs(chunk).mean()
            if volume < threshold:
                silent_chunks += 1
            else:
                silent_chunks = 0
            if silent_chunks > max_silent_chunks and len(frames) > 5:
                break

    audio = np.concatenate(frames, axis=0)
    with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f:
        with wave.open(f.name, "wb") as wf:
            wf.setnchannels(1)
            wf.setsampwidth(2)
            wf.setframerate(samplerate)
            wf.writeframes(audio.tobytes())
        return f.name

def transcribe(audio_path: str) -> str:
    with open(audio_path, "rb") as f:
        result = client.audio.transcriptions.create(
            model="whisper-1",
            file=f,
            language="en"
        )
    return result.text

This is deliberately simple — a volume-threshold VAD rather than a neural VAD model — because most teams over-engineer this step first. Get the pipeline working end-to-end with a crude silence detector, then swap in webrtcvad or Silero VAD once you know the rest of the system is solid. Optimizing the wrong stage first is the number one reason voice app side-projects stall out.

Stage 2: LangChain orchestration with memory

This is where LangChain actually contributes value. A voice conversation needs to remember what was said three turns ago, potentially call tools (check an order status, look up a knowledge base), and produce output as a stream of tokens so downstream TTS can start speaking before the full response is ready.

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

llm = ChatOpenAI(model="gpt-4o", temperature=0.4, streaming=True)

prompt = ChatPromptTemplate.from_messages([
    SystemMessage(content=(
        "You are a helpful voice assistant. Keep responses conversational, "
        "under three sentences, and avoid markdown or lists since this will "
        "be spoken aloud."
    )),
    MessagesPlaceholder(variable_name="history"),
    HumanMessage(content="{input}")
])

chain = prompt | llm

class VoiceSession:
    def __init__(self):
        self.history = []

    def respond_stream(self, user_text: str):
        """Yields text chunks as they arrive from the LLM."""
        response_chunks = []
        for chunk in chain.stream({"history": self.history, "input": user_text}):
            if chunk.content:
                response_chunks.append(chunk.content)
                yield chunk.content

        full_response = "".join(response_chunks)
        self.history.append(HumanMessage(content=user_text))
        self.history.append(AIMessage(content=full_response))

        # Trim history to keep latency down on long conversations
        if len(self.history) > 20:
            self.history = self.history[-20:]

Notice the system prompt explicitly tells the model to avoid markdown and lists — a detail that trips up almost everyone building their first voice app. LLMs default to bulleted lists and bold text because that's what they're trained to produce for chat UIs. Your TTS engine will read **important** as literal asterisks or garble numbered lists. Constrain the output format in the prompt itself; don't try to strip markdown with regex after the fact, since regex stripping breaks on edge cases and adds latency for no benefit.

Stage 3: Streaming text to speech

The critical design decision here is *sentence-level chunking*. You don't want to wait for the entire LLM response before calling TTS — that reintroduces the latency you just eliminated. Instead, buffer LLM tokens until you hit a sentence boundary, then fire that sentence off to TTS while the LLM keeps generating the next one.

import re
from elevenlabs import generate, stream

SENTENCE_END = re.compile(r"[.!?]\s")

def stream_to_speech(text_stream):
    buffer = ""
    for chunk in text_stream:
        buffer += chunk
        match = SENTENCE_END.search(buffer)
        while match:
            sentence = buffer[:match.end()].strip()
            buffer = buffer[match.end():]
            if sentence:
                speak(sentence)
            match = SENTENCE_END.search(buffer)

    if buffer.strip():
        speak(buffer.strip())

def speak(sentence: str):
    audio = generate(
        text=sentence,
        voice="Rachel",
        model="eleven_turbo_v2"
    )
    stream(audio)

Wiring the full pipeline together:

session = VoiceSession()

def run_turn():
    audio_path = record_until_silence()
    user_text = transcribe(audio_path)
    print(f"User said: {user_text}")

    text_stream = session.respond_stream(user_text)
    stream_to_speech(text_stream)

while True:
    run_turn()

This loop is intentionally synchronous and blocking for clarity — you record, transcribe, generate, and speak, then loop. In production you'd run capture and playback on separate threads so you can support barge-in (the user interrupting the assistant mid-sentence), which we'll cover next.

Handling interruptions and barge-in

Nothing makes a voice app feel more broken than an assistant that keeps talking over the user. Real conversational systems need to detect when the user starts speaking again and immediately stop TTS playback. This requires running your VAD continuously, even while audio is playing back, and having a mechanism to cancel in-flight TTS.

A simplified pattern using a cancellation flag:

import threading

class InterruptibleSpeaker:
    def __init__(self):
        self.cancel_event = threading.Event()

    def speak(self, sentence: str):
        if self.cancel_event.is_set():
            return
        audio = generate(text=sentence, voice="Rachel", model="eleven_turbo_v2")
        for audio_chunk in audio:
            if self.cancel_event.is_set():
                break
            play_chunk(audio_chunk)

    def interrupt(self):
        self.cancel_event.set()

    def reset(self):
        self.cancel_event.clear()

You'd pair this with a background thread that monitors the microphone input level even during assistant playback. The moment volume crosses your VAD threshold, call interrupt(), discard the rest of the queued sentences, and immediately start a new record_until_silence() cycle. This is genuinely one of the harder engineering problems in voice apps — not because the code is complex, but because you're coordinating two audio streams (mic input and speaker output) and need to avoid the speaker's own output triggering false-positive interruptions through echo. Acoustic echo cancellation (AEC) either needs to be handled by your audio hardware/driver or a library like webrtc-audio-processing — don't try to solve echo cancellation by hand in Python.

Adding tools: letting the assistant take action

A voice assistant that can only chat is a toy. Real voice apps need to check order status, book appointments, or query a database — and LangChain's tool-calling abstraction is what makes this clean rather than a pile of if/else string matching on the transcript.

from langchain_core.tools import tool
from langchain.agents import create_tool_calling_agent, AgentExecutor

@tool
def check_order_status(order_id: str) -> str:
    """Look up the current status of a customer order by its ID."""
    # In production this hits your order management system
    fake_db = {"1001": "shipped", "1002": "processing"}
    return fake_db.get(order_id, "Order not found")

@tool
def get_store_hours(location: str) -> str:
    """Return store hours for a given location name."""
    return "Monday to Saturday, 9 AM to 8 PM. Closed Sundays."

tools = [check_order_status, get_store_hours]

agent_prompt = ChatPromptTemplate.from_messages([
    SystemMessage(content=(
        "You are a voice assistant for a retail store. Use tools when the "
        "customer asks about orders or store hours. Keep spoken responses short."
    )),
    MessagesPlaceholder(variable_name="chat_history"),
    HumanMessage(content="{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad"),
])

agent = create_tool_calling_agent(llm, tools, agent_prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=False)

result = agent_executor.invoke({
    "input": "What's the status of order 1001?",
    "chat_history": []
})
print(result["output"])

One important caveat for voice specifically: tool calls introduce latency, and unlike a chat UI where a spinner is acceptable, dead silence during a tool call feels broken in a voice conversation. A common pattern is to have the assistant say a short filler phrase — "Let me check that for you" — the moment a tool call is triggered, generated either from a small fixed set of phrases or a fast, separate LLM call, so the user isn't staring into silence for the 1-3 seconds a database lookup takes.

Latency budgeting: where your milliseconds actually go

If you only take one lesson from this article, make it this: voice app latency is a budget you have to actively manage across every stage, not something you fix once at the end. A rough breakdown for a well-optimized pipeline:

• VAD + silence detection: 200-500ms of unavoidable "waiting to be sure they're done talking"
• STT transcription: 200-800ms depending on provider and audio length
• First LLM token: 300-800ms time-to-first-token, heavily dependent on model and prompt length
• First TTS audio chunk: 200-500ms depending on provider

Add those up and you're looking at 1-2.5 seconds minimum before the user hears anything, even in a well-tuned system. That's acceptable for a customer support bot; it's noticeably sluggish for something meant to feel like a real conversation. The levers you actually have:

• Use a smaller, faster model for simple queries and reserve larger models for complex reasoning — a routing layer in LangChain can decide this per-turn.
• Keep system prompts and conversation history lean; every token in the prompt adds to time-to-first-token.
• Choose STT and TTS providers benchmarked for low latency specifically (not just accuracy) — Deepgram and ElevenLabs' turbo models exist precisely because standard models are too slow for real-time voice.
• Pre-warm connections. Opening a fresh HTTPS connection to your TTS provider on every turn adds meaningfully to latency; keep a persistent client/session alive across turns.

Testing your voice pipeline without a human in the loop

You can't manually talk to your assistant for every regression test. Once the pipeline works, build an automated harness that feeds pre-recorded audio files (or synthetic ones from TTS) through the whole pipeline and asserts on the transcript and response content, decoupled from actually verifying audio quality by ear.

import unittest

class VoicePipelineTest(unittest.TestCase):
    def setUp(self):
        self.session = VoiceSession()

    def test_greeting_flow(self):
        response_chunks = list(self.session.respond_stream("Hello there"))
        full_response = "".join(response_chunks)
        self.assertTrue(len(full_response) > 0)
        self.assertNotIn("**", full_response)  # no stray markdown

    def test_memory_persists_across_turns(self):
        list(self.session.respond_stream("My name is Alex"))
        response_chunks = list(self.session.respond_stream("What's my name?"))
        full_response = "".join(response_chunks).lower()
        self.assertIn("alex", full_response)

if __name__ == "__main__":
    unittest.main()