bedrock_nova_sonic

Manages a single bidirectional streaming session with Nova Sonic.

This class handles the event protocol for sending and receiving audio/text over the InvokeModelWithBidirectionalStream API. Sessions are created via :meth:ChatBedrockNovaSonic.create_session and should be used as an async context manager.

Example::

async with model.create_session(system_prompt="Be helpful.") as session:
    await session.send_audio_chunk(audio_bytes)
    async for event in session.receive_events():
        handle(event)

Chat model for Amazon Nova Sonic bidirectional streaming.

This provides a LangChain integration for Amazon Nova Sonic's bidirectional streaming API (InvokeModelWithBidirectionalStream). Nova Sonic enables real-time speech-to-speech conversations over a persistent, full-duplex streaming connection. Unlike the Converse API used by :class:ChatBedrockConverse, this maintains a persistent full-duplex connection for continuous audio streaming.

For simple text interactions, use :meth:ainvoke or :meth:astream. For full audio streaming, use :meth:create_session to get a :class:NovaSonicSession.

Quick start::

import asyncio
from langchain_aws.chat_models.bedrock_nova_sonic import ChatBedrockNovaSonic

model = ChatBedrockNovaSonic(
    model_id="amazon.nova-sonic-v1:0",
    region_name="us-east-1",
)

# Text-only conversation
response = asyncio.run(
    model.ainvoke("Hello, how are you?")
)
print(response.content)

Audio streaming::

import asyncio
from langchain_aws.chat_models.bedrock_nova_sonic import (
    ChatBedrockNovaSonic,
    NovaSonicSession,
)

async def stream_audio():
    model = ChatBedrockNovaSonic(
        model_id="amazon.nova-sonic-v1:0",
        region_name="us-east-1",
        voice_id="matthew",
    )

    async with model.create_session() as session:
        # Send audio chunks
        await session.send_audio_chunk(audio_bytes)

        # Receive responses
        async for event in session.receive_events():
            if event["type"] == "audio":
                play(event["audio"])
            elif event["type"] == "text":
                print(event["text"])

asyncio.run(stream_audio())