Streamen Sie Sprachausgabe mit geringer Latenz von KI-Agenten mit Kokoro TTS auf Amazon SageMaker AI

Ein Kunde stellt einer KI-Voicestimme eine Frage und wartet. Der Agent hat bereits eine Antwort, er ist mitten in der Generierung einer langen Antwort mit mehreren Absätzen, aber der Kunde hört fünf Sekunden lang nichts. Bis die Antwort eintrifft, ist der Kunde frustriert, verwirrt und zahlt nicht mehr weiter auf.

Text-to-Speech-Integrationen erzeugen genau diese Lücke. Eine Antwort mit einem Satz wird möglicherweise in 600 Millisekunden generiert, aber die vollständige Antwort dauert fünf Sekunden, und die Integration wartet auf alles, bevor die Synthese beginnt.

Dieser Beitrag zeigt, wie Sie diese Lücke schließen können. Wir streamen jeden Satz an ein gehostetes Kokoro-82M Text-to-Speech-Modell (TTS) auf Amazon SageMaker AI sobald der Agent ihn erzeugt, sodass die Synthese bereits mit Satz eins beginnt statt erst mit der fertigen Antwort. Diesen Audio-Stream zurück in den Browser zu bekommen erfordert eine eigene Lösung, daher behandeln wir beide Transportmöglichkeiten und was jede davon in der Praxis kostet.

Lösungsübersicht

Diese Lösung schließt die Lücke, indem sie die Synthese parallel zur Generierung streamt, statt auf deren Abschluss zu warten. Während der Agent jeden Satz erzeugt, wird dieser Satz sofort an ein gehostetes Kokoro-82M Text-to-Speech-Modell gestreamt, sodass die Audiowiedergabe beginnen kann, bevor der Agent seine vollständige Antwort ausgegeben hat. Kokoro-82M umfasst nur 82 Millionen Parameter, klein genug, um kostengünstig auf einer einzelnen GPU-Instance der mittleren Klasse zu laufen. Das hält sowohl die Latenz als auch die Inferenzkosten niedrig für einen Voice-Agenten, der echten Traffic verarbeitet.

Kokoro-82M läuft in einem benutzerdefinierten Container auf einem SageMaker-AI-Real-Time-Endpoint auf einer ml.g5.xlarge GPU-Instance. Der Container stellt zwei Routen bereit: POST /invocations, die eine Antwort für eine vollständige Eingabe streamt, und WS /invocations-bidirectional-stream, die eine offene Verbindung über viele Nachrichten in beide Richtungen aufrechterhält.

Das folgende Diagramm zeigt den Anforderungspfad vom Client zum Modell:

Abbildung 1: Anforderungspfad vom Client zum SageMaker-AI-Real-Time-Endpoint über einen bidirektionalen Stream

Voraussetzungen

Bevor Sie beginnen, benötigen Sie:

• Ein AWS-Konto mit Berechtigungen zum Erstellen von SageMaker-AI-Endpoints; AgentCore Runtime, eine Funktion von Amazon Bedrock AgentCore; AWS Identity and Access Management (IAM)-Rollen; und ein Amazon Elastic Container Registry (Amazon ECR)-Repository, einschließlich mindestens der Aktionen sagemaker:CreateEndpoint, sagemaker:CreateEndpointConfig, sagemaker:CreateModel, sagemaker:InvokeEndpointWithBidirectionalStream, und bedrock:InvokeModelWithResponseStream .
• Der Beispielcontainer pinnt ein älteres PyTorch mit einer bekannten Sicherheitshinweis (PYSEC-2026-2286, behoben in torch 2.10.0). Überprüfen und aktualisieren Sie den Pin vor jeder Nutzung, die über einen Proof of Value hinausgeht.
• AWS Cloud Development Kit (AWS CDK) v2, installiert und bootstrapped in Ihrem Zielkonto und Ihrer AWS-Region.
• Ein SageMaker-AI-Real-Time-Endpoint, der aus einem Container bereitgestellt wird, der das com.amazonaws.sagemaker.capabilities.bidirectional-streaming=true Docker-Label deklariert.
• Python 3.12 oder höher, uvund Docker, um das Container-Image zu bauen (der zugrunde liegende HTTP/2-Client wird nur mit vorgebauten Wheels für Python 3.12 und neuer ausgeliefert).
• Das AWS SDK for Python (Boto3) mit Unterstützung für invoke_endpoint_with_bidirectional_stream.
• Vertrautheit mit WebSockets und Starlette oder einem vergleichbaren ASGI-Framework.
• Der Bereitstellungscode aus dem GitHub-Repository auf das in diesem Beitrag verwiesen wird.

Durchgang

Der Durchgang stellt zuerst den Endpunkt und den Agenten bereit und arbeitet dann die clientseitigen Änderungen durch, die inkrementelles Streaming Ende-zu-Ende ermöglichen.

Endpunkt und Agenten bereitstellen

Der Kokoro-82M-Endpunkt und die Agenten-Laufzeit werden über zwei CLI-Tools bereitgestellt, die jeweils ihre eigenen AWS-CDK-Stacks umschließen:

uv run speech-infra deploy kokoro-82m
uv run agent-infra deploy --bedrock-model-id <your-bedrock-model-id>

speech-infra deploy stellt den Kokoro-82M-Endpunkt bereit. Dies umfasst die Foundation- und Model-Cache-Stacks, die die zugrunde liegenden IAM-Rollen erstellen und die Modellgewichte in Amazon Simple Storage Service (Amazon S3) synchronisieren. Der Endpunkt-Stack erstellt dann das Container-Image und erzeugt das SageMaker-AI-Modell, die Endpunkt-Konfiguration und den Endpunkt. agent-infra deploy stellt die Amazon-Bedrock-AgentCore-Laufzeit bereit, die den Strands-Agenten hostet, welcher den soeben bereitgestellten Endpunkt aufruft; stellen Sie daher zuerst den Endpunkt bereit.

Schritt 1: Den WebSocket-Frame-Typ verarbeiten, den SageMaker AI tatsächlich sendet

Die von SageMaker AI verwendete bidirektionale Übertragung liefert jeden Teil der Nachricht eines Clients als binären WebSocket-Frame an Ihren Container. Eine übliche Methode zum Schreiben eines Starlette-WebSocket-Handlers ist der Aufruf von receive_text(), welches message["text"] liest und KeyError auslöst, falls dieser Schlüssel nicht vorhanden ist. Da jede Nachricht, die dieser Client sendet, als Binärdaten ankommt, schlägt dieser Handler bei jeder Anforderung fehl.

Der Fehler wird häufig falsch diagnostiziert. Ein generischer Exception-Handler, der str(e) an den Client weiterleitet, verwandelt KeyError('text') in einen Error-Frame mit dem Text 'text', was wie ein fehlendes Feld aussieht und nicht wie ein Framing-Problem. Der Endpoint selbst meldet durchgehend InService , da er Verbindungen annimmt und auf jede Nachricht antwortet – jeweils mit einem Fehler.

Akzeptieren Sie explizit beide Frame-Typen, statt anzunehmen, dass einer davon auftaucht:

from starlette.websockets import WebSocket, WebSocketDisconnect

async def _receive_message(websocket: WebSocket) -> str:
    message = await websocket.receive()
    if message["type"] == "websocket.disconnect":
        raise WebSocketDisconnect(message.get("code", 1000))
    payload = message.get("text")
    if payload is None:
        payload = message.get("bytes", b"").decode("utf-8")
    return payload

Dies dekodiert Bytes als UTF-8, wenn kein Textfeld vorhanden ist, statt einen bestimmten Frame-Typ anzunehmen. Testen Sie es, indem Sie einen Binary-Frame senden, denn dieser Fehlermodus erzeugt einen wohlgeformten Error-Frame und einen gesunden Endpoint-Status, keinen offensichtlichen Absturz.

Schritt 2: Die Sitzung zwischen den Nachrichten offen halten

Nachdem Ihr Container die Audio-Synthese einer Nachricht abgeschlossen hat, ist der Instinkt, die Sitzung zu schließen, so wie ein HTTP-Zyklus endet. Das ist hier falsch. Nach dem Senden der Audio-Daten einer Nachricht und eines synthesis_complete -Steuerframes sollte die Container-Schleife zurückkehren zum Warten auf receive() für die nächste Nachricht über dieselbe, weiterhin offene Sitzung. Schließen Sie die Verbindung nur, wenn der Client eine explizite {"type": "close"} -Nachricht sendet.

Das folgende Diagramm zeigt eine Sitzung, die zwei Sätze nacheinander trägt und erst am Ende geschlossen wird:

Abbildung 2: Eine bidirektionale Sitzung, die zwei Sätze nacheinander trägt und erst am Ende geschlossen wird

Auf diese Weise können Sie mehrere Sätze, einen nach dem anderen, über eine einzige persistente Sitzung senden, ohne Wiederverbindungskosten dazwischen. Ein voreiliges Schließen nach einer Antwort würde eine neue Verbindung pro Satz erzwingen und damit den Großteil des Latenzvorteils zunichtemachen.

Schritt 3: Sätze streamen, sobald der Agent sie erzeugt

Mit dem Sitzungsverhalten an Ort und Stelle besteht die clientseitige Lösung für das Latenzproblem in einer Klasse, die Text in zwei Erscheinungsformen normalisiert: eine vollständige Zeichenkette, die im Voraus bekannt ist, oder einen Live-Token-Stream. In beiden Fällen erzeugt sie dieselbe Ausgabe: vollständige Sätze, bereit zum Senden, sobald jeder vollständig ist.

import re

_SENTENCE_RE = re.compile(r"(?<=[.!?])\s+")


class IncrementalSentenceChunker:
    def __init__(self) -> None:
        self._buffer = ""

    def feed(self, fragment: str) -> list[str]:
        self._buffer += fragment
        parts = _SENTENCE_RE.split(self._buffer)
        if len(parts) == 1:
            return []
        self._buffer = parts[-1]
        return [s.strip() for s in parts[:-1] if s.strip()]

    def flush(self) -> str | None:
        remainder = self._buffer.strip()
        self._buffer = ""
        return remainder or None

Rufen Sie feed() einmal pro Textfragment auf, sobald es eintrifft. Es gibt alle neu abgeschlossenen Sätze zurück und hält nachlaufenden Text zurück. Rufen Sie flush() einmal, am Ende der Eingabe auf, um den verbleibenden Teiltext freizugeben, auch ohne nachstehende Interpunktion. So wird verhindert, dass ein Stream, der endet, bevor ein Punkt eintrifft, einen letzten Satz verliert.

Dies ist keine gleichzeitige Synthese. Jeder bidirektionale Handler ist eine streng sequenzielle Schleife: eine Nachricht empfangen, ihre Audio-Daten synthetisieren, diese Audio-Daten senden, synthesis_completesenden, dann die nächste Nachricht empfangen. Die Audio-Daten von Satz zwei werden nicht erzeugt, während Satz eins noch übertragen wird. Was das inkrementelle Senden bringt, ist bescheidener: Die Synthese von Satz einen beginnt in dem Moment, in dem er existiert, statt auf die vollständige Antwort zu warten. Bei einer Antwort mit mehreren Sätzen und echter Generierungszeit zwischen dem ersten und dem letzten Satz ist das der gesamte Latenzvorteil.

Wir haben dies an einem Live-Endpoint verifiziert, indem wir eine Eingabe mit drei Sätzen über eine einzige Sitzung gesendet haben. Alle drei Chunks kamen mit Audio an, das zum jeweils gesendeten Text passte:

Schritt 4: Einen Transport wählen, um die Audio-Daten zum Client zurückzubringen

Text inkrementell in das Modell zu streamen, löst die halbe Lösung. Die andere Hälfte besteht darin, die resultierenden Audio-Daten an das Element auszuliefern, das sie rendert, üblicherweise ein Browser. Die meisten Agent- und Chat-Frontends streamen Text bereits über Server-Sent Events (SSE): unidirektional und von den meisten Browsern nativ unterstützt. SSE ist ein Textprotokoll und nicht darauf ausgelegt, eine rohe binäre Nutzlast zu transportieren. Ein Team, das einer bestehenden SSE-Pipeline Sprache hinzufügt, greift natürlich auf das Muster zurück, das es bereits hat: das Audio mit base64 kodieren und es in ein JSON-Event einpacken.

Wir haben die Kosten dieses Ansatzes über 24 Clips hinweg gemessen, mit Amazon Transcribe bestätigt, dass sich die Qualität zwischen den Encodierungen nicht unterschied, und MP3 mit AAC verglichen:

Base64 bläht jede Nutzlast um nahezu die erwarteten 33 % auf. Der subtilere Kostenfaktor ist die Frame-Ausrichtung. SSE-Ereignisse werden durch eine Leerzeile abgegrenzt, aber das Audio darin wird in MP3-Frames unterteilt. Die beiden Schemata müssen nicht unbedingt mit der Art übereinstimmen, wie der Transport die Daten in Blöcke aufteilt. In einer gemessenen Antwort trafen vier logische MP3-Frames als sechs Lieferblöcke ein, von denen zwei mitten im Frame endeten. Ein Consumer, der jeden Lieferblock als eine vollständige Einheit behandelt, scheitert an gewöhnlichem, intaktem Traffic. Die Lösung puffert eingehende Bytes und handelt erst, sobald ein Trennzeichen gefunden wird:

buffer = b""
for event in resp["Body"]:
    if "PayloadPart" not in event:
        continue
    buffer += event["PayloadPart"]["Bytes"]
    while b"\n\n" in buffer:
        frame, buffer = buffer.split(b"\n\n", 1)
        handle(frame)

Die heute für die modellseitige Verbindung verwendete Alternative verzichtet vollständig auf base64 und SSE-Framing. Sie multiplext rohe Binär-Audio- und JSON-Steuerframes über dieselbe bidirektionale Sitzung, die bereits inkrementelle Texteingaben überträgt, und unterscheidet sie mit einer einzigen Regel.

import json


def _decode_control_frame(chunk: bytes) -> dict | None:
    try:
        decoded = json.loads(chunk.decode("utf-8"))
    except (UnicodeDecodeError, json.JSONDecodeError):
        return None
    return decoded if isinstance(decoded, dict) else None

Ein Block, der sich nicht als JSON-Objekt dekodieren lässt, ist Audio und wird unverändert weitergeleitet, ohne dass es irgendwo im Pfad einen Encoding-Schritt gibt. Ein Randfall ist wichtig: Rohes PCM-Audio kann legitim mit dem Byte beginnen 0x7b, dem ASCII-Zeichen {, demselben Zeichen, das ein JSON-Objekt eröffnet. Diese Regel vermeidet eine Fehlklassifizierung dieses Audios, weil sie verlangt, dass der gesamte Block als gültiges JSON geparst wird, nicht allein sein erstes Byte – etwas, was ein PCM-Frame praktisch nie leisten wird.

Hinweis: Wenn Ihr Frontend bereits eine funktionierende SSE-Pipeline für Text hat, ist base64-kodiertes Audio innerhalb eines SSE-Ereignisses die pragmatische Wahl. Es kostet ungefähr 33 Prozent zusätzliche Nutzlast und einen Puffer-und-Zusammensetzungs-Schritt, aber es nutzt Infrastruktur wieder, die Sie bereits betreiben. Wenn Sie den Sprachpfad neu aufbauen oder die Latenz Priorität hat, vermeidet der vereinheitlichte Binär- und JSON-Ansatz beide Kosten und nutzt die Verbindung wieder, die bereits Text überträgt.

Schritt 5: Alles in einem echten Agenten zusammenführen

Der umfassendere End-to-End-Fall ist ein live agierender, generativer Agent, der seine Antwort Token für Token erstellt und erst dann Satz vier kennt, wenn Satz eins existiert. Dieses Beispiel läuft als Strands-Agent, unterstützt von Amazon Bedrock, bereitgestellt in der Amazon Bedrock AgentCore runtime, der den Kokoro-Endpoint mit dem Muster aus Schritt 4 aufruft. Drei Kommunikationsverbindungen sind in diesem Design verbunden: die browserseitige Verbindung, die Server-Sent Events zum Aufrufer zurückträgt, der Streaming-Aufruf des Agenten an Amazon Bedrock zur Textgenerierung und die bidirektionale Sitzung des Agenten mit dem Kokoro-Endpoint zur Synthese.

Das folgende Diagramm zeigt diese drei Kommunikationsverbindungen und wo die Überlappung zwischen Generierung und Synthese stattfindet:

Abbildung 3: Die drei Kommunikationsverbindungen, die den Aufrufer, Amazon Bedrock und den Kokoro-Endpoint miteinander verbinden

Die Anforderungsnutzlast trägt ein mode Feld, entweder "bidi" oder "batch", verteilt auf zwei Codepfade innerhalb derselben Entrypoint-Funktion. Ein Benchmark zwischen ihnen ändert genau eine Variable: wann die Synthese beginnt. In run_bidilaufen zwei Aufgaben gleichzeitig. Eine entleert den Bedrock-Textstrom und speist vollständige Sätze in den Chunker. Die andere entleert die Kokoro-Sitzung und gibt Audio-Chunks aus, sobald sie eintreffen. Beide schreiben in eine gemeinsame Warteschlange, sodass ein Consumer Ereignisse in wahrer Ankunftsreihenfolge erhält: Audio für einen frühen Satz kann den Aufrufer erreichen, während Bedrock noch einen späteren generiert. run_batch ist das bewusste Gegenteil: Es wartet auf die vollständige Antwort, bevor ein einziger Syntheseaufruf erfolgt, und dient als faire Baseline.

In die andere Richtung verwendet die Verbindung vom Agenten-Container zurück zum Aufrufer doch base64, aus einem anderen Grund als in Schritt 4 beschrieben. BedrockAgentCoreApps Entrypoint muss einfache, JSON-serialisierbare Dictionaries zurückgeben, die sein SDK automatisch in data: {json}\n\n Frames umwandelt. Es gibt keine Möglichkeit, ihm einen rohen Binärframe zu übergeben. Das ist eine Folge der JSON-nativen Managed Runtime, nicht der Wahl, die Schritt 4 gemessen und sich davon entfernt hat. Beide, run_bidi und run_batch , zahlen diese identischen Kosten auf dieser Verbindung, sodass der gemessene Unterschied zwischen ihnen rein im Timing liegt: wann der erste Audio-Chunk ausgegeben wird.

Testen und Validierung

Wir isolierten den eigenen Beitrag des TTS-Transports zur Zeit bis zum ersten Audio, indem wir denselben 267 Zeichen langen, viersätzigen Textblock auf zwei Arten sendeten: als eine Nachricht (Batch) und nur als seinen ersten Satz (inkrementell). Wir führten je fünf Versuche gegen den Live-Endpoint durch.

Wie zuvor gemessen, senkte das Senden nur des ersten Satzes die mittlere Zeit bis zum ersten Audio-Byte von 329.3 ms auf 156.7 ms, eine relative Verbesserung von 52.4 Prozent, isoliert auf den TTS-Transport selbst.

Anschließend maßen wir den umfassenderen End-to-End-Fall mit einem echten Sprachmodell, das den Text erzeugt, indem wir denselben Prompt durch run_bidi und run_batch gegen dasselbe Live-Bedrock-Modell und denselben Endpoint laufen ließen, sodass die einzige Variable der Zeitpunkt des Synthesestarts ist.

Mit derselben Methodik war die Zeit bis zum ersten Audio im bidi-Modus 2.4-mal schneller als im Batch-Modus gegen einen lokalen Container und 3.5-mal schneller gegen die bereitgestellte AgentCore runtime.

Das folgende Diagramm zeigt diese Lücke in beiden Umgebungen:

Abbildung 4: Relative Zeit bis zum ersten Audio für Batch- und bidi-Modus, lokal und auf der bereitgestellten Runtime

Die Lücke vergrößert sich beim eingesetzten Fall, sobald ein echter Netzwerk-Hop hinzukommt, statt sich zu verkleinern. run_bidi absorbiert diesen Hop in der Überschneidung, die es bereits zwischen Generierung und Synthese ausnutzt. run_batch hat keine vergleichbare Überschneidung, in der sich diese Kosten verbergen ließe, da es jeweils nur eine Sache gleichzeitig erledigt, sodass jede hinzukommende Millisekunde Netzwerk-Overhead direkt auf seine serielle Wartezeit schlägt.

Um diesen Vergleich gegen Ihre eigene Bereitstellung nachzustellen, führen Sie denselben Bidi-gegen-Batch-Test aus, den der Agent-Container bereitstellt:

uv run agent-infra demo --endpoint-url <your-endpoint-url> --prompt "<your prompt>"

Dies sendet den Prompt durch beide Modi und meldet den Zeitstempel des ersten audio_chunk Ereignisses gegen agent_start für jeden. Falls stattdessen Fehler auftreten, prüfen Sie auf das KeyError: 'text' Symptom aus Schritt 1. Ein Endpoint, der InService meldet, während jede Anfrage fehlschlägt, zeigt diesen Bug an, kein Bereitstellungsproblem.

Aufräumen

Um laufende Kosten zu vermeiden, löschen Sie die von dieser Anleitung erstellten Ressourcen in umgekehrter Bereitstellungsreihenfolge, zuerst den Agent, dann den Endpoint:

uv run agent-infra destroy --bedrock-model-id <your-bedrock-model-id>
uv run speech-infra destroy kokoro-82m

agent-infra destroy entfernt die Amazon Bedrock AgentCore Runtime-Ressource. speech-infra destroy entfernt das SageMaker AI-Modell, die Endpoint-Konfiguration und den Endpoint. Keines der beiden leert den Amazon S3-Bucket, der Modellgewichte zwischenspeichert, oder löscht das Amazon ECR-Repository, das das Container-Image enthält, da beide modellübergreifend geteilt werden können. Wenn Sie sie nicht mehr benötigen, leeren und löschen Sie den Amazon S3-Bucket und entfernen Sie das Amazon ECR-Repository manuell, da ein gespeicherter Modell-Checkpoint und ein Container-Image weiterhin kleine Speicherkosten verursachen, bis sie entfernt werden.

Fazit

Wir haben gezeigt, wie man eine Quelle vermeidbarer Latenz aus einem Voice-Agent entfernt – das Warten auf die vollständige Antwort eines Agents, bevor die Sprachsynthese beginnt – und wie man eine zweite, leisere Kostenkomponente beziffert: Base64-kodiertes Audio in SSE vergrößert die Payload um etwa ein Drittel und kann, wenn man nicht entsprechend puffert, an den MP3-Frame-Grenzen vorbeilaufen. Das Streaming jedes Satzes in einen bidirektionalen SageMaker AI-Endpoint, sobald ein Agent ihn erzeugt, hat die Zeit bis zum ersten Audio isoliert um mehr als die Hälfte reduziert und in einem bereitgestellten Agent um das 2.4- bis 3.5-Fache. Die Multiplexing-Übertragung von binärem Audio und JSON-Kontrollframes über diese Sitzung, anstatt Audio als Text zu kodieren, beseitigt die Größenaufblähung und die Frame-Neuausrichtungsarbeit, die ein naives SSE-Relay erfordert.

Wenn Sie einem Agent, der bereits Text über SSE streamt, Sprache hinzufügen, beginnen Sie mit dem Base64-Ansatz in Schritt 4 und betrachten Sie den einheitlichen Transport erneut, sobald Latenz zur Priorität wird. Weitere Informationen finden Sie in der Dokumentation zu Amazon SageMaker AI Real-Time Inference und in der Amazon Bedrock AgentCore-Dokumentation. Wenn Sie Unterstützung bei der Anwendung dieses Musters auf Ihre eigene Architektur wünschen, wenden Sie sich an Ihr AWS-Kontoteam.

Über den Autor