OpenAI Realtime

1
import asyncio
2
import logging
3
import os
4
from typing import Optional
5
6
from agents.realtime import RealtimeAgent, RealtimeRunner
7
from agents.realtime.session import RealtimeSession
8
from dotenv import load_dotenv
9
10
from agentduet import (
11
    BufferFullError,
12
    Call,
13
    CallAudioConfig,
14
    CallClosedError,
15
    IncomingCallNotification,
16
    SessionManager,
17
    SessionManagerConfig,
18
    new_session_id,
19
)
20
21
load_dotenv()
22
23
logging.basicConfig(
24
    level=logging.INFO,
25
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
26
)
27
logger = logging.getLogger(__name__)
28
29
AGENT = RealtimeAgent(
30
    name="Assistant",
31
    instructions="You are a helpful and friendly AI assistant.",
32
)
33
34
RUNNER = RealtimeRunner(
35
    starting_agent=AGENT,
36
    config={
37
        "model_settings": {
38
            "model_name": "gpt-realtime-1.5",
39
            "audio": {
40
                "input": {
41
                    "format": "pcm16",
42
                    "transcription": {"model": "gpt-4o-mini-transcribe"},
43
                    "turn_detection": {
44
                        "type": "semantic_vad",
45
                        "interrupt_response": True,
46
                    },
47
                },
48
                "output": {
49
                    "format": "pcm16",
50
                    "voice": "ash",
51
                },
52
            },
53
        }
54
    },
55
)
56
57
58
class OpenAIRealtimeIntegration:
59
    def __init__(self, call: Call, session: RealtimeSession):
60
        self._call = call
61
        self._session = session
62
        self._send_task: Optional[asyncio.Task] = None
63
        self._recv_task: Optional[asyncio.Task] = None
64
        self._terminated = False
65
66
    async def _on_hangup(self, evt):
67
        logger.info("Call terminated - closing OpenAI Realtime session")
68
        self._terminated = True
69
        try:
70
            await self._session.close()
71
        except Exception:
72
            logger.exception("Error closing OpenAI Realtime session")
73
74
        tasks = [t for t in (self._send_task, self._recv_task) if t]
75
        for t in tasks:
76
            t.cancel()
77
        if tasks:
78
            await asyncio.gather(*tasks, return_exceptions=True)
79
80
    async def run(self):
81
        self._call.on_hangup(self._on_hangup)
82
        self._send_task = asyncio.create_task(self.stream_to_openai())
83
        self._recv_task = asyncio.create_task(self.receive_from_openai())
84
        await asyncio.gather(self._send_task, self._recv_task, return_exceptions=True)
85
86
    async def stream_to_openai(self):
87
        try:
88
            async for audio_chunk in self._call.caller.audio_stream():
89
                await self._session.send_audio(audio_chunk)
90
        except asyncio.CancelledError:
91
            raise
92
        except CallClosedError:
93
            logger.debug("Call closed; stopping stream to OpenAI")
94
        except Exception:
95
            logger.exception("Error in stream to OpenAI")
96
            raise
97
98
    async def receive_from_openai(self):
99
        try:
100
            async for event in self._session:
101
                if event.type == "audio":
102
                    audio_bytes = event.audio.data
103
                    if audio_bytes:
104
                        try:
105
                            await self._call.send_audio(audio_bytes)
106
                        except BufferFullError:
107
                            logger.warning(
108
                                "Send buffer full - drop chunk or raise buffer_size"
109
                            )
110
                elif event.type == "audio_interrupted":
111
                    await self._call.clear_send_audio_buffer()
112
                    logger.debug("OpenAI interrupted - cleared send buffer")
113
                elif event.type == "error":
114
                    logger.error("OpenAI Realtime error: %s", event.error)
115
        except asyncio.CancelledError:
116
            raise
117
        except CallClosedError:
118
            logger.debug("Call closed; stopping receive from OpenAI")
119
        except Exception:
120
            logger.exception("Error in receive from OpenAI")
121
            raise
122
123
124
async def main():
125
    config = SessionManagerConfig.create(
126
        api_key=os.getenv("AGENTDUET_API_KEY"),
127
        connector_uuid=os.getenv("AGENTDUET_CONNECTOR_UUID"),
128
        call_audio=CallAudioConfig(
129
            sample_rate=24000,
130
            buffer_size=1024 * 1024,
131
        ),
132
    )
133
134
    async with SessionManager(config) as sm:
135
        logger.info("Connected. Waiting for calls...")
136
137
        @sm.on_incoming_call
138
        async def on_call(noti: IncomingCallNotification):
139
            session = await sm.open_session(new_session_id(), noti.subscriber)
140
            call = await session.process_call(noti)
141
            logger.info("Incoming call %s", call.id)
142
            try:
143
                async with await RUNNER.run() as oai_session:
144
                    result = await call.answer()
145
                    if not result:
146
                        logger.error(
147
                            "Answer failed %s: %s (%s)",
148
                            call.id,
149
                            result.error_message,
150
                            result.error_code,
151
                        )
152
                        return
153
                    await OpenAIRealtimeIntegration(call, oai_session).run()
154
            except Exception:
155
                logger.exception("Error in OpenAI Realtime bridge")
156
                await call.close()
157
                raise
158
159
        await sm.run_forever()
160
161
162
if __name__ == "__main__":
163
    asyncio.run(main())