Amazon Nova Sonic

1
import asyncio
2
import base64
3
import json
4
import logging
5
import os
6
import uuid
7
from typing import Optional
8
9
from aws_sdk_bedrock_runtime.client import (
10
    BedrockRuntimeClient,
11
    InvokeModelWithBidirectionalStreamOperationInput,
12
)
13
from aws_sdk_bedrock_runtime.config import Config
14
from aws_sdk_bedrock_runtime.models import (
15
    BidirectionalInputPayloadPart,
16
    InvokeModelWithBidirectionalStreamInputChunk,
17
)
18
from dotenv import load_dotenv
19
from smithy_aws_core.identity.environment import EnvironmentCredentialsResolver
20
from websockets.exceptions import ConnectionClosed
21
22
from agentduet import (
23
    BufferFullError,
24
    Call,
25
    CallAudioConfig,
26
    CallClosedError,
27
    IncomingCallNotification,
28
    SessionManager,
29
    SessionManagerConfig,
30
    new_session_id,
31
)
32
33
load_dotenv()
34
35
logging.basicConfig(
36
    level=logging.INFO,
37
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
38
)
39
logger = logging.getLogger(__name__)
40
41
MODEL_ID = "amazon.nova-2-sonic-v1:0"
42
REGION = os.getenv("AWS_REGION", "us-east-1")
43
44
SYSTEM_PROMPT = (
45
    "You are a warm, professional AI assistant on a phone call. Give accurate answers "
46
    "that sound natural and direct. Answer clearly in 1-2 sentences, then expand only "
47
    "enough to stay understandable (3-5 short sentences total)."
48
)
49
50
51
class NovaSonicIntegration:
52
    def __init__(self, call: Call, client: BedrockRuntimeClient):
53
        self._call = call
54
        self._client = client
55
        self._stream = None
56
        self.prompt_name = str(uuid.uuid4())
57
        self.content_name = str(uuid.uuid4())
58
        self.audio_content_name = str(uuid.uuid4())
59
        self._send_to_nova_task: Optional[asyncio.Task] = None
60
        self._recv_from_nova_task: Optional[asyncio.Task] = None
61
        self._is_active = False
62
63
    async def _on_hangup(self, evt):
64
        if not self._is_active:
65
            return
66
        logger.info("Call terminated - cleaning up Nova session")
67
        self._is_active = False
68
        if self._send_to_nova_task:
69
            self._send_to_nova_task.cancel()
70
        if self._recv_from_nova_task:
71
            self._recv_from_nova_task.cancel()
72
        tasks = [t for t in (self._send_to_nova_task, self._recv_from_nova_task) if t]
73
        if tasks:
74
            await asyncio.gather(*tasks, return_exceptions=True)
75
        try:
76
            await self.end_session()
77
        except Exception:
78
            pass
79
80
    async def send_event(self, event_json: str):
81
        event = InvokeModelWithBidirectionalStreamInputChunk(
82
            value=BidirectionalInputPayloadPart(bytes_=event_json.encode("utf-8"))
83
        )
84
        await self._stream.input_stream.send(event)
85
86
    async def start_session(self):
87
        self._stream = await self._client.invoke_model_with_bidirectional_stream(
88
            InvokeModelWithBidirectionalStreamOperationInput(model_id=MODEL_ID)
89
        )
90
        self._is_active = True
91
92
        await self.send_event(
93
            """
94
            {
95
              "event": {
96
                "sessionStart": {
97
                  "inferenceConfiguration": {
98
                    "maxTokens": 1024,
99
                    "topP": 0.9,
100
                    "temperature": 0.7
101
                  }
102
                }
103
              }
104
            }
105
            """
106
        )
107
108
        await self.send_event(
109
            f"""
110
            {{
111
              "event": {{
112
                "promptStart": {{
113
                  "promptName": "{self.prompt_name}",
114
                  "textOutputConfiguration": {{ "mediaType": "text/plain" }},
115
                  "audioOutputConfiguration": {{
116
                    "mediaType": "audio/lpcm",
117
                    "sampleRateHertz": 24000,
118
                    "sampleSizeBits": 16,
119
                    "channelCount": 1,
120
                    "voiceId": "matthew",
121
                    "encoding": "base64",
122
                    "audioType": "SPEECH"
123
                  }}
124
                }}
125
              }}
126
            }}
127
            """
128
        )
129
130
        await self.send_event(
131
            f"""
132
            {{
133
              "event": {{
134
                "contentStart": {{
135
                  "promptName": "{self.prompt_name}",
136
                  "contentName": "{self.content_name}",
137
                  "type": "TEXT",
138
                  "interactive": false,
139
                  "role": "SYSTEM",
140
                  "textInputConfiguration": {{ "mediaType": "text/plain" }}
141
                }}
142
              }}
143
            }}
144
            """
145
        )
146
        await self.send_event(
147
            f"""
148
            {{
149
              "event": {{
150
                "textInput": {{
151
                  "promptName": "{self.prompt_name}",
152
                  "contentName": "{self.content_name}",
153
                  "content": "{SYSTEM_PROMPT}"
154
                }}
155
              }}
156
            }}
157
            """
158
        )
159
        await self.send_event(
160
            f"""
161
            {{
162
              "event": {{
163
                "contentEnd": {{
164
                  "promptName": "{self.prompt_name}",
165
                  "contentName": "{self.content_name}"
166
                }}
167
              }}
168
            }}
169
            """
170
        )
171
        await self.send_event(
172
            f"""
173
            {{
174
              "event": {{
175
                "contentStart": {{
176
                  "promptName": "{self.prompt_name}",
177
                  "contentName": "{self.audio_content_name}",
178
                  "type": "AUDIO",
179
                  "interactive": true,
180
                  "role": "USER",
181
                  "audioInputConfiguration": {{
182
                    "mediaType": "audio/lpcm",
183
                    "sampleRateHertz": 24000,
184
                    "sampleSizeBits": 16,
185
                    "channelCount": 1,
186
                    "audioType": "SPEECH",
187
                    "encoding": "base64"
188
                  }}
189
                }}
190
              }}
191
            }}
192
            """
193
        )
194
195
    async def end_session(self):
196
        if not self._stream:
197
            return
198
        try:
199
            await self.send_event(
200
                f"""
201
                {{
202
                  "event": {{
203
                    "contentEnd": {{
204
                      "promptName": "{self.prompt_name}",
205
                      "contentName": "{self.audio_content_name}"
206
                    }}
207
                  }}
208
                }}
209
                """
210
            )
211
            await self.send_event(
212
                f"""
213
                {{
214
                  "event": {{
215
                    "promptEnd": {{ "promptName": "{self.prompt_name}" }}
216
                  }}
217
                }}
218
                """
219
            )
220
            await self.send_event("""{ "event": { "sessionEnd": {} } }""")
221
        except Exception:
222
            pass
223
        finally:
224
            if self._stream:
225
                await self._stream.input_stream.close()
226
                self._stream = None
227
228
    async def run(self):
229
        self._call.on_hangup(self._on_hangup)
230
        await self.start_session()
231
        self._send_to_nova_task = asyncio.create_task(self.stream_to_nova())
232
        self._recv_from_nova_task = asyncio.create_task(self.receive_audio_from_nova())
233
        await asyncio.gather(
234
            self._send_to_nova_task,
235
            self._recv_from_nova_task,
236
            return_exceptions=True,
237
        )
238
239
    async def stream_to_nova(self):
240
        try:
241
            async for audio_chunk in self._call.caller.audio_stream():
242
                if not self._is_active:
243
                    break
244
                blob = base64.b64encode(audio_chunk).decode("utf-8")
245
                await self.send_event(
246
                    f"""
247
                    {{
248
                      "event": {{
249
                        "audioInput": {{
250
                          "promptName": "{self.prompt_name}",
251
                          "contentName": "{self.audio_content_name}",
252
                          "content": "{blob}"
253
                        }}
254
                      }}
255
                    }}
256
                    """
257
                )
258
        except (ConnectionClosed, CallClosedError):
259
            pass
260
        except asyncio.CancelledError:
261
            raise
262
        except Exception:
263
            logger.exception("Error in stream to Nova")
264
            raise
265
266
    async def receive_audio_from_nova(self):
267
        try:
268
            while self._is_active:
269
                if not self._stream:
270
                    await asyncio.sleep(0.1)
271
                    continue
272
273
                output = await self._stream.await_output()
274
                result = await output[1].receive()
275
276
                if result.value and result.value.bytes_:
277
                    json_data = json.loads(result.value.bytes_.decode("utf-8"))
278
                    if "event" not in json_data:
279
                        continue
280
                    evt = json_data["event"]
281
282
                    if "textOutput" in evt:
283
                        text_content = evt["textOutput"]["content"]
284
                        if '{ "interrupted" : true }' in text_content:
285
                            await self._call.clear_send_audio_buffer()
286
                            logger.debug("Nova interrupted - cleared send buffer")
287
288
                    elif "audioOutput" in evt:
289
                        audio_bytes = base64.b64decode(evt["audioOutput"]["content"])
290
                        try:
291
                            await self._call.send_audio(audio_bytes)
292
                        except BufferFullError:
293
                            logger.warning(
294
                                "Send buffer full - drop chunk or raise buffer_size"
295
                            )
296
        except (ConnectionClosed, asyncio.CancelledError):
297
            pass
298
        except CallClosedError:
299
            logger.debug("Call closed; stopping stream from Nova")
300
        except Exception:
301
            if self._is_active:
302
                logger.exception("Error in stream from Nova")
303
            raise
304
305
306
async def main():
307
    aws_config = Config(
308
        endpoint_uri=f"https://bedrock-runtime.{REGION}.amazonaws.com",
309
        region=REGION,
310
        aws_credentials_identity_resolver=EnvironmentCredentialsResolver(),
311
    )
312
    bedrock_client = BedrockRuntimeClient(config=aws_config)
313
314
    config = SessionManagerConfig.create(
315
        api_key=os.getenv("AGENTDUET_API_KEY"),
316
        connector_uuid=os.getenv("AGENTDUET_CONNECTOR_UUID"),
317
        call_audio=CallAudioConfig(
318
            sample_rate=24000,
319
            buffer_size=1024 * 1024,
320
        ),
321
    )
322
323
    async with SessionManager(config) as sm:
324
        logger.info("Connected. Waiting for calls...")
325
326
        @sm.on_incoming_call
327
        async def on_call(noti: IncomingCallNotification):
328
            session = await sm.open_session(new_session_id(), noti.subscriber)
329
            call = await session.process_call(noti)
330
            logger.info("Incoming call %s", call.id)
331
            try:
332
                result = await call.answer()
333
                if not result:
334
                    logger.error(
335
                        "Answer failed %s: %s (%s)",
336
                        call.id,
337
                        result.error_message,
338
                        result.error_code,
339
                    )
340
                    return
341
                await NovaSonicIntegration(call, bedrock_client).run()
342
            except Exception:
343
                logger.exception("Error in Nova Sonic bridge")
344
                await call.close()
345
                raise
346
347
        await sm.run_forever()
348
349
350
if __name__ == "__main__":
351
    asyncio.run(main())