Talk to the agent

Prerequisites

• An API key allowed Realtime sessions (prerequisites).
• Python 3.10 or newer, a microphone and headphones.

1. 1
   Install the OpenAI SDK and an audio library
   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.

2. 2
   Save the script
   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.

3. 3
   Talk to it
   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.

Next steps