How to build live translation into smart glasses and wearables

Smart glasses with a heads-up display have an odd constraint for a translation product: there's no speaker. The wearer can't hear the translation, so it has to be read. That changes almost every design decision, from when you show text to how long a line stays up to what happens when the phone loses signal on a train.

This tutorial walks through how translation glasses work end to end, then builds the part most teams write first: a companion-app relay that takes audio from the glasses, streams it to Universal-3.6 Pro Realtime with real-time translation turned on, and pushes caption events back to the display. It's written for developers building the device or the app, not for shoppers comparing headsets.

If you want the general theory first (cascaded vs end-to-end translation, and why only final turns get translated), start with our guide to real-time speech translation. This post assumes you know the basics of real-time speech-to-text and focuses on what's different on a wearable.

How do translation glasses work?

Translation glasses capture speech with a microphone on the frame or on a paired phone, send the audio to a speech recognition and translation service, and show the translated text on a small display in the lens. Most of them don't do the heavy lifting on the glasses. The glasses handle audio capture and display, a phone app handles the network, and the cloud handles transcription and translation.

So a working system has four parts:

1. Capture. A mic array on the frame, or the phone's own mic when the wearer holds it out. The glasses send audio to the phone over Bluetooth.
2. Relay. A companion app on the phone keeps one WebSocket open to the speech API, forwards audio, and turns server events into display commands.
3. Transcribe and translate. The speech API transcribes the source language and translates each finished turn.
4. Display. The phone sends short caption payloads back over Bluetooth, and the glasses render them.

The relay is where most of the engineering lives. It's also the part you control completely, so it's the part this tutorial builds.

The architecture: glasses, phone relay, one WebSocket

With AssemblyAI, transcription and translation happen on a single streaming connection. You set the llm_gateway connection parameter when you open the WebSocket. When a turn ends, the server sends the final transcript to LLM Gateway with your translation prompt and returns the translation on the same socket as an LLMGatewayResponse message. There's no second API to call from the phone. The real-time translation docs cover the parameter in full.

That matters on a wearable for three reasons:

• One socket to keep alive. Mobile networks drop connections. Managing one connection is far simpler than keeping a transcription socket and a translation client in sync.
• The translation doesn't block the transcript. The docs say the translation runs in the background, so it doesn't delay the final Turn message.
• Zero data retention on the translation step. Translation requests go to LLM Gateway with zero data retention, so the prompt and the translated text aren't stored by the gateway. That's worth knowing when the audio comes from a stranger's conversation.

Here's the flow:

Glasses mic ──BLE──▶ Companion app ──WebSocket──▶ Universal-3.6 Pro Realtime
                         ▲                              │ Turn (partials, final)
                         │                              ▼
Glasses display ◀──BLE── Companion app ◀── LLMGatewayResponse (translated final turn)

The source side is broad. Universal-3.6 Pro Realtime transcribes 32 languages and code-switches natively, so a speaker who mixes English and Spanish mid-sentence still gets a clean transcript. If you don't know what language the person in front of the wearer will speak, the docs suggest pairing translation with multilingual transcription and telling the model to translate "from whatever language is spoken". The target side is whatever the LLM you pick can write. For an overview of everything AssemblyAI does with translation, see the translation feature page.

Partial source text vs final translated text

This is the first design decision that's different on glasses, and it follows from how the API works. Only final turns are translated. Partial Turn messages stay in the spoken language.

So you have two streams of text:

• Partials arrive while the person is still talking. They're fast, but they're in the source language and they change as more audio comes in.
• Translated finals arrive after the turn ends. They're in the wearer's language and they don't change.

A wearer who doesn't speak the source language can't read the partials. But hiding them completely makes the display feel dead for the length of every sentence. The pattern that works on a small display is:

• Show the partial as a single dim preview line. It tells the wearer the system is hearing speech, even if they can't read it. Show only the last line, so a long partial never pushes the previous caption off the screen.
• Replace it with the translated final. When the LLMGatewayResponse arrives, render the translation at full brightness in the main caption area.
• Fall back to the source text if no translation comes. The docs say that if the gateway call fails or times out, no LLMGatewayResponse is sent for that turn. Don't leave a gap. After a short wait, show the final source text, marked as untranslated.
• Match by turn_order, not arrival order. Translations for consecutive short turns can arrive out of order. Key every caption by its turn_order.

If the wearer speaks the source language reasonably well and just wants support, you can flip this: show partials prominently and use the translation as a subtitle line. Make it a setting.

One more lever: translation quality depends heavily on context. A reply like "Yes, that one." translates badly on its own. The prompt template supports {{history_1}} to {{history_10}}, the last final transcripts whose translation has already completed, and the docs call history "the main lever for translation quality". Three turns is the documented default for conversations.

Caption layout for a small heads-up display

A heads-up display in a lens is narrow and low resolution, and the wearer is also looking at the person they're talking to. Captions have to be short, stable and predictable. None of the numbers below are AssemblyAI figures. They're starting points to tune on your hardware.

Do the wrapping and paging on the phone, not the glasses. The phone has the CPU, it knows the text, and it can send each page as one small payload. The glasses then only render lines and run a timer.

The latency budget, stage by stage

The question every wearable team asks is "how long from the end of a sentence to the caption?" The honest answer is that AssemblyAI publishes figures for its own stages, and you have to measure the rest on your hardware. Here's the budget with only documented numbers, each attributed to its source:

Two warnings about this table. First, the figures come from different sources and different conditions, so don't add them up and call the result your end-to-end latency. Pipecat's TTFS is measured "from when the speaker stops to the final transcription segment" on its own dataset, which you can read on the AssemblyAI benchmarks page. Second, the model you choose for translation changes the last AssemblyAI stage. The gateway defaults to gemini-2.5-flash-lite if you don't set a model, and recommend a small, fast model so translation keeps pace with speech. The LLM Gateway models list shows the options and their per-token prices.

The stages you own are usually where the time goes. Instrument each Bluetooth hop with timestamps from the start, and log the gap between the final Turn and its LLMGatewayResponse in production so you know when a model change helps.

Build the companion-app relay in Python

Here's a relay you can run on a laptop today. It mocks the glasses by playing a 16 kHz mono 16-bit WAV file in real time, the same shape of audio a glasses mic would deliver over Bluetooth. It streams that audio to Universal-3.6 Pro Realtime with llm_gateway set, and prints the JSON caption payloads you'd write to the display. In a real app, you'd port the same logic to Swift or Kotlin and swap send() for a Bluetooth characteristic write.

The connection setup, the llm_gateway object and the message handling are adapted from the official raw WebSocket code in the real-time translation docs. What's added is the wearable layer: caption paging, the untranslated fallback, a replay buffer, and reconnect with backoff.

Install the one dependency and set your key:

pip install websocket-client
export ASSEMBLYAI_API_KEY="<YOUR_API_KEY>"

The full code, plus an optional microphone version, is in the companion repo on GitHub. Then save this as relay.py:

# pip install websocket-client
import json
import os
import random
import sys
import textwrap
import threading
import time
import wave
from collections import deque
from urllib.parse import urlencode

import websocket

# Local testing only. A shipped companion app should fetch a short-lived
# temporary token from your backend before each connection instead.
API_KEY = os.environ["ASSEMBLYAI_API_KEY"]

SAMPLE_RATE = 16000
CHUNK_MS = 50
CHUNK_BYTES = SAMPLE_RATE * 2 * CHUNK_MS // 1000  # 16-bit mono PCM: 1,600 bytes

REPLAY_SECONDS = 5        # audio kept while offline; anything older is dropped
CATCH_UP_SPEED = 1.2      # backlog send rate, under the ~1.25x real-time throttle
TRANSLATION_WAIT_S = 2.0  # fall back to source text if no translation arrives
HEARTBEAT_TIMEOUT_S = 15  # no message (Heartbeat arrives every 5 s) for this long = dead link

PROMPT = (
    "Earlier in the conversation: {{history_3}}\n\n"
    "Translate the following English into natural Spanish and reply with only "
    "the translation: {{turn}}"
)
LLM_GATEWAY_CONFIG = {
    "model": "qwen3.5-4b-32k-fast",
    "prompt": PROMPT,
    "max_tokens": 400,
}
CONNECTION_PARAMS = {
    "sample_rate": SAMPLE_RATE,
    "speech_model": "universal-3-6-pro",
    "llm_gateway": json.dumps(LLM_GATEWAY_CONFIG),
    "session_heartbeat": "true",  # server sends a Heartbeat message every 5 s
}
API_ENDPOINT = "wss://streaming.assemblyai.com/v3/ws?" + urlencode(CONNECTION_PARAMS)


class GlassesDisplay:
    """Stands in for the BLE link to the glasses. Replace send() with your
    characteristic write; keep the JSON small so it fits a few packets."""

    MAX_CHARS = 32  # per line on a narrow heads-up display
    MAX_LINES = 2

    def send(self, payload):
        print(json.dumps(payload, ensure_ascii=False))

    def show_partial(self, text):
        # Source-language preview: last line only, so it never pushes captions off.
        tail = textwrap.wrap(text, self.MAX_CHARS)[-1:] if text else []
        self.send({"kind": "partial", "lines": tail})

    def show_caption(self, turn, text, translated):
        lines = textwrap.wrap(text, self.MAX_CHARS)
        pages = [lines[i:i + self.MAX_LINES] for i in range(0, len(lines), self.MAX_LINES)]
        for page in pages:
            hold_s = max(1.5, sum(len(line) for line in page) / 15)  # ~15 chars/s
            self.send({"kind": "caption", "turn": turn, "translated": translated,
                       "lines": page, "hold_s": round(hold_s, 1)})

    def show_status(self, status):
        self.send({"kind": "status", "status": status})


class AudioBuffer:
    """Holds device audio. While offline it keeps only the last REPLAY_SECONDS."""

    def __init__(self):
        self.chunks = deque(maxlen=REPLAY_SECONDS * 1000 // CHUNK_MS)
        self.cond = threading.Condition()
        self.finished = False

    def push(self, chunk):
        with self.cond:
            self.chunks.append(chunk)
            self.cond.notify()

    def pop(self, timeout=0.5):
        with self.cond:
            if not self.chunks:
                self.cond.wait(timeout)
            if not self.chunks:
                return None, 0
            return self.chunks.popleft(), len(self.chunks)

    def requeue(self, chunk):
        with self.cond:
            if len(self.chunks) < self.chunks.maxlen:
                self.chunks.appendleft(chunk)

    def finish(self):
        with self.cond:
            self.finished = True
            self.cond.notify_all()


def mock_device(path, buffer):
    """Plays a 16 kHz mono 16-bit WAV in real time, like a glasses mic over BLE."""
    with wave.open(path, "rb") as wav:
        assert wav.getframerate() == SAMPLE_RATE and wav.getnchannels() == 1
        while True:
            chunk = wav.readframes(CHUNK_BYTES // 2)
            if len(chunk) < CHUNK_BYTES:
                break
            buffer.push(chunk)
            time.sleep(CHUNK_MS / 1000)
    buffer.finish()


class Relay:
    def __init__(self, buffer, display):
        self.buffer = buffer
        self.display = display
        self.session_id = None
        self.pending = {}  # turn_order -> (source_text, deadline)
        self.lock = threading.Lock()
        self.last_error = None
        self.ended = False
        self.last_heard = time.monotonic()

    # --- WebSocket callbacks ---
    def on_open(self, ws):
        self.last_heard = time.monotonic()
        self.display.show_status("live")
        threading.Thread(target=self.send_audio, args=(ws,), daemon=True).start()

    def on_message(self, ws, message):
        data = json.loads(message)
        kind = data.get("type")
        self.last_heard = time.monotonic()  # any message, Heartbeat included

        if kind == "Begin":
            self.session_id = data["id"]
        elif kind == "Turn":
            if data["end_of_turn"]:
                deadline = time.monotonic() + TRANSLATION_WAIT_S
                with self.lock:
                    self.pending[data["turn_order"]] = (data["transcript"], deadline)
            else:
                self.display.show_partial(data["transcript"])
        elif kind == "LLMGatewayResponse":
            turn = data["turn_order"]
            translation = data["data"]["choices"][0]["message"]["content"].strip()
            with self.lock:
                if self.pending.pop(turn, None) is None:
                    return  # already shown untranslated; don't repeat it
            self.display.show_caption(turn, translation, translated=True)
        elif kind == "Error":
            self.last_error = data
        elif kind == "Termination":
            self.ended = True

    def on_error(self, ws, error):
        if isinstance(error, websocket.ABNF) and error.opcode == websocket.ABNF.OPCODE_CLOSE:
            return
        print(f"WebSocket error: {error}", file=sys.stderr)

    def on_close(self, ws, status, msg):
        self.flush_untranslated(force=True)
        if not self.ended:
            self.display.show_status("reconnecting")

    # --- helpers ---
    def flush_untranslated(self, force=False):
        """No LLMGatewayResponse arrives if the gateway call fails, so after a
        short wait show the source-language text rather than nothing."""
        now = time.monotonic()
        with self.lock:
            due = [(t, text) for t, (text, deadline) in sorted(self.pending.items())
                   if force or now >= deadline]
            for turn, _ in due:
                del self.pending[turn]
        for turn, text in due:
            self.display.show_caption(turn, text, translated=False)

    def send_audio(self, ws):
        while ws.sock and ws.sock.connected:
            self.flush_untranslated()
            if time.monotonic() - self.last_heard > HEARTBEAT_TIMEOUT_S:
                ws.close()  # silent link: reconnect instead of waiting on TCP
                return
            chunk, backlog = self.buffer.pop()
            if chunk is None:
                if self.buffer.finished:
                    ws.send(json.dumps({"type": "Terminate"}))
                    return
                continue
            try:
                ws.send(chunk, websocket.ABNF.OPCODE_BINARY)
            except websocket.WebSocketConnectionClosedException:
                self.buffer.requeue(chunk)  # keep it for the next session
                return
            if backlog:
                time.sleep(CHUNK_MS / 1000 / CATCH_UP_SPEED)  # paced catch-up

    def should_retry(self):
        if self.ended:
            return False, 0
        err = self.last_error or {}
        self.last_error = None
        if err.get("error_code") == 1008:
            if "Too many concurrent sessions" in err.get("error", ""):
                return True, 60  # new-session rate limit: try again next minute
            return False, 0      # bad key, token or account: retrying won't help
        return True, None


def main(wav_path):
    buffer = AudioBuffer()
    display = GlassesDisplay()
    relay = Relay(buffer, display)
    threading.Thread(target=mock_device, args=(wav_path, buffer), daemon=True).start()

    attempt = 0
    while True:
        ws = websocket.WebSocketApp(
            API_ENDPOINT,
            header={"Authorization": API_KEY},
            on_open=relay.on_open,
            on_message=relay.on_message,
            on_error=relay.on_error,
            on_close=relay.on_close,
        )
        started = time.monotonic()
        ws.run_forever()

        retry, wait = relay.should_retry()
        if not retry:
            break
        if time.monotonic() - started > 30:
            attempt = 0  # the last session was healthy; start backoff over
        if wait is None:
            wait = min(30, 2 ** attempt) + random.uniform(0, 1)
        attempt += 1
        time.sleep(wait)
    display.show_status("ended")


if __name__ == "__main__":
    main(sys.argv[1] if len(sys.argv) > 1 else "sample_16k_mono.wav")

Run it against any 16 kHz mono WAV of someone speaking English:

python relay.py conversation_16k_mono.wav

The output is one JSON payload per display update. With the docs' own example sentences, it looks like this:

{"kind": "status", "status": "live"}
{"kind": "partial", "lines": ["Hi, I'd like to make a"]}
{"kind": "partial", "lines": ["reservation at a"]}
{"kind": "caption", "turn": 0, "translated": true, "lines": ["¡Hola! Me gustaría hacer una", "reserva en un restaurante."], "hold_s": 3.6}
{"kind": "partial", "lines": ["Saturday around"]}
{"kind": "caption", "turn": 1, "translated": true, "lines": ["Sábado a las 7 de la tarde."], "hold_s": 1.8}

A few details in the code are worth calling out:

• Audio chunks are 50 ms. The docs recommend 50 to 100 ms per chunk and reject anything under 20 ms or over 1,000 ms. Bluetooth audio often arrives in different packet sizes, so re-chunk on the phone before sending.
• The prompt uses {{history_3}}. That's the documented default for conversations. The docs note that a turn joins the history only once its own translation returns, so in fast exchanges the turn right before may be missing.
• The fallback timer runs on the sender thread. It fires even when no messages come in, so an untranslated turn shows up after about 2 seconds rather than waiting for the next partial.
• Change the target language in the prompt. To translate into French instead, change both mentions of Spanish. To accept any source language, say so in the prompt.

Handling flaky connectivity

Wearables move. The wearer walks into an elevator, the phone switches from Wi-Fi to cellular, the train goes into a tunnel. Plan for the connection to drop several times an hour.

Detect a dead link from the heartbeat. A phone that walks out of coverage can leave a socket that looks open but carries nothing. The relay sets session_heartbeat=true, so the server sends a Heartbeat message every 5 seconds, and it closes the socket and reconnects if nothing at all arrives for 15 seconds.

Treat every reconnect as a new session. The docs don't describe resuming a dropped streaming session, so the relay opens a fresh one, with its own Begin message and its own translation history. When the old socket closes, the relay shows any final turns still waiting for a translation as source text, so nothing the wearer was about to read disappears.

Keep a short replay buffer, not a long one. While offline, the relay keeps the last 5 seconds of audio and drops anything older. On reconnect it sends that backlog first, paced at 1.2× real time. The docs say throughput is throttled at about 1.25× real time, and that sending chunks in a tight loop can produce inconsistent Turn messages, so pace the catch-up. Why only 5 seconds? A caption for something said 40 seconds ago is mostly noise to someone in the middle of a conversation, and every second of backlog delays the live captions behind it. The replay buffer exists to rescue the sentence in progress, not to rebuild the past.

Read the Error message before you retry. When a session ends because of an error, the server sends an Error text frame with an error_code just before closing. The common session errors page lists them. The important one for fleets is 1008, which covers both the new-session rate limit and credential problems. Retry only on the "Too many concurrent sessions" text, and wait until the next minute. Any other 1008 (an invalid key, an expired token, low balance) won't fix itself, so stop and surface it.

Back off with jitter. The streaming rate limit counts session starts per minute, not open sessions. A thousand pairs of glasses that lose signal in the same subway station will all try to reconnect at once. Exponential backoff with random jitter spreads them out.

Use a temporary token per connection. Don't ship your API key in the app. Have your backend mint a short-lived temporary token and pass it in place of the key. The docs say to mint a fresh token for each connection and keep expires_in_seconds short, so the reconnect path needs a token fetch too.

Plan for long sessions. A streaming session auto-closes after 3 hours with error 3008. All-day wearables should expect it and reconnect the same way.

What not to promise in your product copy:

• No offline translation. This architecture needs a network connection. When the phone is offline, the display should say so.
• No captions for audio older than the buffer. Anything said during a long outage is gone.
• No fixed end-to-end latency. You can quote documented per-stage figures. You can't quote a total until you've measured your own Bluetooth and network hops.

Battery and bandwidth

Battery life is the constraint that decides whether people wear the glasses for a whole day. These are general engineering guidelines, not measured AssemblyAI figures.

• Compress before the radio. 16 kHz 16-bit mono PCM is 256 kbps. The streaming API also accepts Opus (encoding=opus for raw packets, one per frame, or encoding=ogg_opus), and the docs say Opus uses roughly 8× less bandwidth than 16 kHz PCM. If your glasses already encode Opus for Bluetooth, you may be able to forward packets without decoding them on the phone. See sending audio for the framing rules.
• Close the session when captions are off. Streaming is billed on how long the WebSocket is open, not on how much audio you send. A session left open in a pocket costs money and keeps the radio awake. Send Terminate when the wearer turns captions off, and open a new session when they turn them back on.
• Send fewer display updates. Expect several partial updates before each final turn. Throttle preview-line updates and skip ones that don't change the visible last line. Each Bluetooth write and display refresh costs power on the glasses.
• Keep payloads small. Send pre-wrapped lines, not raw transcripts with word timings. The glasses don't need confidence scores to draw text.
• Do the work on the phone. Wrapping, paging, timers and fallback logic all run in the companion app. The glasses only render.

Optional: speak the translation in earbuds

The glasses have no speaker, but many wearers also have earbuds paired to the same phone. If they want to hear the translation as well as read it, the companion app can send each translated final turn to AssemblyAI text-to-speech and play the audio through the earbuds.

Keep the limits in view:

• Six output languages. AssemblyAI-TTS-1 (assemblyai-tts-1) speaks English, Spanish, French, German, Italian and Portuguese, and every voice speaks all six. If the translation target is outside those six, offer captions only.
• Speak finals, never partials. Partials are in the source language and they change. Only the translated final turn should be voiced.
• Pricing and speed. TTS is $15 per 1M characters, and the AssemblyAI pricing page lists about 90 ms time to first audio.

The TTS WebSocket takes the voice and language as connection parameters. The language value is the full English name, not an ISO code. For Spanish output with the docs' default Spanish voice, the connection URL looks like this:

wss://tts.us.assemblyai.com/v1/ws?voice_id=aai-antonioluis&language=spanish&sample_rate=24000&encoding=pcm_s16le

The TTS voices docs list a default voice for each of the six languages. Note that AssemblyAI doesn't publish a speech-to-speech translation recipe. This STT-to-translation-to-TTS chain is a composition you build and measure yourself.

Where to go from here

The hard part of translation glasses isn't the translation. It's everything around it: deciding what to show before the translation exists, keeping text readable on a tiny display, and failing quietly when the network does. One streaming connection that returns both the transcript and the translation takes a whole client out of that picture, which leaves your team free to work on the parts that are actually specific to your device.

Start with the relay above, measure your own Bluetooth hops, and tune the caption numbers on real hardware with real conversations. Captions that feel fine on a laptop screen often feel slow or cluttered in a lens.

Frequently Asked Questions (FAQs)

How do translation glasses work?

Translation glasses capture speech with a microphone on the frame or a paired phone, send the audio over Bluetooth to a companion app, and the app streams it to a cloud speech API that transcribes and translates it. The translated text is sent back to the glasses and shown on a small heads-up display. Most of the processing happens on the phone and in the cloud, not on the glasses.

Can I build a translation glasses app with a speech API?

Yes. With AssemblyAI, you open one Universal-3.6 Pro Realtime WebSocket from the companion app and set the llm_gateway connection parameter with a translation prompt. Each final turn comes back as a transcript, followed by an LLMGatewayResponse with its translation. Your app handles Bluetooth, caption layout and reconnects.

Why are only final turns translated in real-time translation?

Partial transcripts change as more audio arrives, so translating them would waste tokens and flicker on the display. AssemblyAI translates each final turn and leaves partials in the spoken language. On glasses, a common pattern is to show the partial as a dim preview line and replace it with the translation when it arrives.

How fast is real-time translation on a wearable?

AssemblyAI's docs say the translation typically lands within about 100 ms of the final turn with a small model such as qwen3.5-4b-32k-fast. Pipecat's open STT benchmark reports a 307 ms median time from end of speech to final transcript for Universal-3.6 Pro Realtime. Bluetooth and network hops add time you have to measure on your own hardware, so don't add these figures into one total.

What happens when the phone loses connectivity?

The streaming session closes, and the app should open a new one when the network returns. Keep a short replay buffer of a few seconds so the sentence in progress isn't lost, send it at a paced rate on reconnect, and read the Error message before retrying. Retry 1008 only for the new-session rate limit, and fetch a fresh temporary token for each connection.

Can the translation be spoken through earbuds?

Yes, if the target language is one of the six that AssemblyAI-TTS-1 speaks: English, Spanish, French, German, Italian and Portuguese. Send each translated final turn to the TTS WebSocket and play it through the paired earbuds. For other target languages, show captions only.