> ## Documentation Index
> Fetch the complete documentation index at: https://docs.kugelaudio.com/llms.txt
> Use this file to discover all available pages before exploring further.
# Talk to the agent
> Talk to a voice agent through your computer's microphone and speakers, with the OpenAI Python SDK.
Speech to speech follows the OpenAI Live API, and you use OpenAI's own SDK. For anything these
pages do not answer, read [OpenAI's Live documentation](/speech-to-speech/overview#use-openais-documentation):
it applies as written, except for what [Coming from OpenAI Live](/speech-to-speech/openai-compatibility) lists.
## Prerequisites
- An API key allowed **Realtime sessions** ([prerequisites](/speech-to-speech/overview#prerequisites)).
- Python 3.10 or newer, a microphone and headphones.
```bash
pip install "openai[realtime]>=3.14.1,<4" sounddevice
export KUGELAUDIO_API_KEY=YOUR_API_KEY
```
`sounddevice` opens your microphone and speakers. On Linux, install PortAudio first:
`sudo apt install libportaudio2`.
```python talk.py
import asyncio
import base64
import os
import threading
import sounddevice as sd
from openai import AsyncOpenAI
RATE = 24_000
FRAME = RATE // 50 # 20 ms of audio, in samples
class Speaker:
"""Plays the agent's voice as it arrives."""
def __init__(self) -> None:
self.waiting = bytearray()
self.lock = threading.Lock()
self.stream = sd.RawOutputStream(samplerate=RATE, channels=1, dtype="int16", callback=self.fill)
def play(self, delta: str) -> None:
with self.lock:
self.waiting += base64.b64decode(delta)
def fill(self, out, frames, time, status) -> None:
with self.lock:
audio = bytes(self.waiting[: len(out)])
del self.waiting[: len(out)]
out[:] = audio + bytes(len(out) - len(audio))
async def stream_microphone(connection) -> None:
"""Sends your microphone to the agent, 20 ms at a time."""
loop = asyncio.get_running_loop()
frames: asyncio.Queue[bytes] = asyncio.Queue()
def record(data, count, time, status) -> None:
loop.call_soon_threadsafe(frames.put_nowait, bytes(data))
with sd.RawInputStream(samplerate=RATE, channels=1, dtype="int16", blocksize=FRAME, callback=record):
while True:
frame = await frames.get()
await connection.session.input_audio.append(audio=base64.b64encode(frame).decode("ascii"))
class Transcript:
"""Prints both sides of the call as they speak, one line per turn."""
def __init__(self) -> None:
self.speaking = ""
def add(self, who: str, words: str) -> None:
if who != self.speaking:
self.speaking = who
words = f"\n{who}: {words.lstrip()}"
print(words, end="", flush=True)
def look_up_order(caller_said: str) -> str:
"""Your backend: it reads what the caller said and returns the facts to pass on."""
return "Order 4711 shipped yesterday and arrives tomorrow."
async def answer(connection, delegation_id: str, caller_said: str) -> None:
result = await asyncio.to_thread(look_up_order, caller_said)
await connection.session.commentary.append(delegation_id=delegation_id, content=result)
async def main() -> None:
heard: list[str] = [] # your words, streamed while you speak
tasks: set[asyncio.Task[None]] = set()
speaker = Speaker()
transcript = Transcript()
async with (
AsyncOpenAI(
api_key=os.environ["KUGELAUDIO_API_KEY"],
base_url="https://api.kugelaudio.com/v1",
) as client,
client.live.connect() as connection,
):
await connection.session.start(
session={
"model": "fluid-1",
"instructions": "You answer the phone for a parcel service. Keep replies short.",
"audio": {"format": {"type": "audio/pcm", "rate": RATE}},
"delegation": {"type": "client"},
}
)
with speaker.stream:
async for event in connection:
if event.type == "session.started":
tasks.add(asyncio.create_task(stream_microphone(connection)))
print("Talk now. Press Ctrl+C to hang up.", end="")
elif event.type == "session.output_audio.delta":
speaker.play(event.delta)
elif event.type == "session.input_transcript.delta":
heard.append(event.delta)
transcript.add("You", event.delta)
elif event.type == "session.output_transcript.delta":
transcript.add("Agent", event.delta)
elif event.type == "session.delegation.created":
tasks.add(asyncio.create_task(answer(connection, event.delegation.id, "".join(heard))))
elif event.type == "error":
print("Error:", event.error.code, event.error.message)
elif event.type == "session.closed":
print("\nSession ended:", event.reason)
break
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
print("\nHung up.")
```
Your microphone streams to the agent in 20 ms frames, and its voice plays as it arrives. When the
agent needs your backend, a `session.delegation.created` event starts `answer` in a task of its
own, so the conversation keeps going.
```bash
python talk.py
```
Put on headphones first: the script does not cancel echo, so through speakers the agent hears
itself. On macOS, the first run asks to let your terminal use the microphone.
The agent greets you. Ask "Where is my order? The number is 4711." and it tells you what
`look_up_order` returned. Both sides print as they speak. Press Ctrl+C to hang up.
Keep the API key on your server. A browser or a phone app talks to your backend, and your
backend holds this WebSocket. See [Fit it into your stack](/speech-to-speech/integration).
## Next steps
Replace the fixed answer with a real lookup
Stop playback when the caller interrupts