Erstellen Sie einen vollständigen Sprach-Agenten

Aktualisiert am 2. November 2026 — Audio-Modelle. Mehrere Modelle in diesem Kurs sind veraltet: tts-1, tts-1-hd und gpt-4o-mini-tts werden am 6. Januar 2027 eingestellt (Nachfolger: gpt-realtime-2.1-mini, der laut OpenAI-TTS-Leitfaden nicht durch /v1/audio/speech unterstützt wird: Die Migration der Sprachsynthese erfolgt über die Realtime API); gpt-realtime, gpt-4o-realtime und veraltete Audio-Modelle am 20. Januar 2027 (gpt-realtime-2.1, gpt-realtime-2.1-mini, gpt-audio-1.5); whisper-1, gpt-4o-transcribe und /v1/audio/translations am 26. Februar 2027 (gpt-transcribe, gpt-live-transcribe, v1/realtime/translations). Für die Voll-Duplex-Stimme ist GPT-Live 1 seit dem 10. September 2026 in der API verfügbar.

Korrektur vom 3. November 2026. Die Eingabe-Transkription verwendete gpt-live-transcribe, die nur in einer Transkriptions-Sitzung funktioniert und weder server_vad noch semantic_vad unterstützt. In einer Gesprächssitzung wie dieser, in der VAD jeden Turn validiert, legt die Dokumentation gpt-transcribe fest, die validierte Turns transkribiert und vorherige Turns als Kontext verwendet. Quellen: Realtime-Transkription · GPT-Transcribe · GPT-Live-Transcribe.

In dieser Lektion werden alle bisher behandelten Konzepte zusammengeführt, um einen voll funktionsfähigen End-to-End-Sprach-Agenten zu erstellen. Sie erstellen einen Assistenten, der den Benutzer hört, dessen Anfrage versteht, Aktionen ausführt und per Stimme antwortet – alles in Echtzeit über die Realtime API.

Agenten-Architektur

Vier Module und vor allem eine Einschränkung, die die gesamte Architektur antreibt: alles muss gleichzeitig geschehen. Das Mikrofon erfasst, während das Modell antwortet; Ereignisse eintreffen, während Audio abgespielt wird; und der Benutzer kann jederzeit unterbrechen. Ein Sprach-Agent ist kein Frage-Antwort-Loop; es ist ein System, in dem mehrere Ströme parallel voranschreiten – weshalb wir im Folgenden stark auf asyncio setzen.

Audio

Erfassung und Wiedergabe

Sitzung

WebSocket und Konfiguration

Werkzeuge

Geschäftliche Aktionen

Zustand

Gesprächsmanagement

Die VoiceAgent-Klasse

Die Klasse sammelt den Zustand, den diese parallelen Ströme teilen müssen: die Verbindung, die aktuelle Sitzung und die aktuell abgespielte Antwort. Genau weil mehrere Aufgaben diesen Zustand gleichzeitig zugreifen, ist ein Objekt vorzuziehen gegenüber einer Reihe globaler Variablen – wenn Sie eine laufende Antwort abbrechen müssen, müssen Sie wissen, dass eine existiert.

import asyncio
import json
import base64
import websockets
import sounddevice as sd
import numpy as np
import pyaudio

class VoiceAgent:
    """Vollständiger Sprach-Agent basierend auf der Realtime API."""

    def __init__(self, api_key: str, instructions: str, tools: list = None):
        self.api_key = api_key
        self.instructions = instructions
        self.tools = tools or []
        self.ws = None
        self.is_playing = False
        self.sample_rate = 24000
        self.chunk_size = 2400  # 100ms

    async def connect(self):
        """Etabliert die Verbindung und konfiguriert die Sitzung."""
        url = "wss://api.openai.com/v1/realtime?model=gpt-realtime-2.1"
        headers = {
            "Authorization": f"Bearer {self.api_key}"
        }

        self.ws = await websockets.connect(url, additional_headers=headers)

        # Warten auf session.created
        event = json.loads(await self.ws.recv())
        print(f"Sitzung: {event['session']['id']}")

        # Konfigurieren
        await self.ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "type": "realtime",
                "output_modalities": ["audio"],
                "instructions": self.instructions,
                "audio": {
                    "input": {
                        "format": {"type": "audio/pcm", "rate": 24000},
                        "transcription": {"model": "gpt-transcribe", "languages": ["fr"]},
                        "turn_detection": {
                            "type": "server_vad",
                            "threshold": 0.5,
                            "silence_duration_ms": 600
                        }
                    },
                    "output": {
                        "format": {"type": "audio/pcm", "rate": 24000},
                        "voice": "marin"
                    }
                },
                "tools": self.tools,
                "tool_choice": "auto"
            }
        }))

        await self.ws.recv()  # session.updated
        print("Sprach-Agent bereit.")

    async def start(self):
        """Startet den Sprach-Agenten."""
        await self.connect()
        await asyncio.gather(
            self._capture_audio(),
            self._process_events()
        )

Audio-Erfassungs-Modul

    async def _capture_audio(self):
        """Erfasst Mikrofon-Eingabe und sendet sie an die Realtime API."""
        loop = asyncio.get_event_loop()
        queue = asyncio.Queue()

        def callback(indata, frames, time_info, status):
            pcm = (indata[:, 0] * 32767).astype(np.int16)
            loop.call_soon_threadsafe(queue.put_nowait, pcm)

        stream = sd.InputStream(
            samplerate=self.sample_rate,
            channels=1,
            dtype="float32",
            blocksize=self.chunk_size,
            callback=callback
        )

        with stream:
            while True:
                pcm_data = await queue.get()
                audio_b64 = base64.b64encode(pcm_data.tobytes()).decode()
                await self.ws.send(json.dumps({
                    "type": "input_audio_buffer.append",
                    "audio": audio_b64
                }))

Ereignisverarbeitungs-Modul

    async def _process_events(self):
        """Verarbeitet alle Ereignisse vom Server."""
        pa = pyaudio.PyAudio()
        player = pa.open(
            format=pyaudio.paInt16, channels=1,
            rate=self.sample_rate, output=True
        )

        try:
            async for raw in self.ws:
                event = json.loads(raw)
                t = event["type"]

                if t == "response.output_audio.delta":
                    audio = base64.b64decode(event["delta"])
                    player.write(audio)
                    self.is_playing = True

                elif t == "response.output_audio.done":
                    self.is_playing = False

                elif t == "response.output_audio_transcript.delta":
                    print(event["delta"], end="", flush=True)

                elif t == "input_audio_buffer.speech_started":
                    if self.is_playing:
                        await self.ws.send(json.dumps({
                            "type": "response.cancel"
                        }))
                        self.is_playing = False

                elif t == "response.done":
                    await self._handle_response_done(event)

                elif t == "error":
                    print(f"\nFehler: {event['error']['message']}")

        finally:
            player.close()
            pa.terminate()

Werkzeug-Management-Modul

Dieses Modul verwandelt den Agenten von einem einfachen Sprach-Chatbot in einen Assistenten, der Aktionen ausführen kann. Der Unterschied zum klassischen Funktionsaufruf liegt in der zeitlichen Einschränkung: Der Benutzer ist am Telefon ohne visuellen Fortschrittsindikator. Wenn die Ausführung länger als eine oder zwei Sekunden dauert, sollte der Agent sprechen, während er arbeitet – „Ich überprüfe das gerade für Sie” – denn Stille in einem Sprachgespräch klingt nach einem Ausfall.

    async def _handle_response_done(self, event):
        """Verarbeitet Funktionsaufrufe in der Antwort."""
        response = event["response"]
        function_calls = [
            o for o in response.get("output", [])
            if o["type"] == "function_call"
        ]

        if not function_calls:
            print()  # Neue Zeile nach der Transkription
            return

        for call in function_calls:
            name = call["name"]
            args = json.loads(call["arguments"])
            call_id = call["call_id"]

            print(f"\n→ {name}({args})")

            # Führen Sie die Funktion aus (an Ihre Geschäftslogik anpassen)
            result = await self._execute_function(name, args)

            await self.ws.send(json.dumps({
                "type": "conversation.item.create",
                "item": {
                    "type": "function_call_output",
                    "call_id": call_id,
                    "output": json.dumps(result)
                }
            }))

        # Bitten Sie das Modell, fortzufahren
        await self.ws.send(json.dumps({"type": "response.create"}))

    async def _execute_function(self, name: str, args: dict) -> dict:
        """Leitet den Funktionsaufruf weiter."""
        # Implementieren Sie dies entsprechend Ihrer Geschäftslogik
        return {"error": f"Funktion {name} nicht implementiert"}

Starten Sie den Agenten

async def main():
    agent = VoiceAgent(
        api_key="sk-...",
        instructions=(
            "Sie sind ein Sprach-Assistent für Corsen AI. "
            "Sie helfen Benutzern bei der Verwaltung ihrer Termine "
            "und erhalten Informationen über unsere Dienstleistungen. "
            "Antworten Sie auf Französisch, prägnant und professionell."
        ),
        tools=[
            {
                "type": "function",
                "name": "check_availability",
                "description": "Überprüft die Verfügbarkeit für einen Zeitslot",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "date": {"type": "string"},
                        "service": {"type": "string"}
                    },
                    "required": ["date"]
                }
            }
        ]
    )

    await agent.start()

asyncio.run(main())

Wichtige Erkenntnisse

Ein vollständiger Sprach-Agent kombiniert vier Module: Audio, Sitzung, Werkzeuge und Zustandsmanagement
Die VoiceAgent-Klasse kapselt die gesamte Logik hinter einer einfachen Schnittstelle
asyncio.gather ermöglicht die parallele Mikrofon-Erfassung und Ereignisverarbeitung
Das Behandeln von Unterbrechungen (Abbrechen einer Antwort, wenn der Benutzer spricht) ist essenziell
Werkzeuge verwandeln den Agenten von einem einfachen Sprach-Chatbot in einen Assistenten, der Aktionen ausführen kann