interrupt()

# 1) Create session with higher turn detection sensitivity
session = AgentSession(
    llm=aws.realtime.RealtimeModel.with_nova_sonic_2(
        turn_detection="HIGH"   # more responsive to interruptions
    ),
)

# 2) Start TTS and keep handle so it can be interrupted
speech_handle = session.say("Hello, I will explain how to use the new feature...")

# 3) Example: an event listener that triggers when user speech starts
def on_user_speech_start(event):
    # Event fired when VAD/ASR detects user starting to speak
    # Immediately interrupt the current speech playback
    try:
        speech_handle.interrupt()   # stops current TTS right away
    except Exception:
        # Fallback: interrupt the whole session if handle isn't available
        session.interrupt()

    # Optionally: capture user audio or route to ASR for transcription
    user_audio = event.audio_segment
    transcript = asr.recognize(user_audio)
    # Continue conversation with user's intent
    session.respond_to_transcript(transcript)

# 4) Hook the event listener into your audio pipeline / VAD
audio_input.on("speech_start", on_user_speech_start)

turn_detection="HIGH"

SpeechHandle

interrupt()

turn_detection="HIGH"

MEDIUM

agent.py

turn_detection="HIGH"

SpeechHandle

"""
agent.py

Template example showing Nova Sonic 2 barge-in handling:
- Uses turn_detection="HIGH"
- Keeps a SpeechHandle to call .interrupt() when user speech starts
- Routes user audio to ASR after interruption and responds

Adapt the placeholders (AudioInput, AudioOutput, ASRClient, LiveKit/AgentSession) to your stack.
"""

import threading
import time

# --- Placeholder SDK imports (replace with your actual SDKs) ---
# from livekit_agent import AgentSession, SpeechHandle  # example
# from asr_client import ASRClient
# from audio_pipeline import AudioInput, AudioOutput

# For this template we'll mock the minimal interfaces used below:
class SpeechHandle:
    def __init__(self, text):
        self.text = text
        self._interrupted = False

    def interrupt(self):
        # Immediately stop playback in a real implementation
        self._interrupted = True
        print("[SpeechHandle] interrupted")

    def is_interrupted(self):
        return self._interrupted

class AgentSession:
    def __init__(self, turn_detection="MEDIUM"):
        self.turn_detection = turn_detection
        self._current_handle = None
        print(f"[AgentSession] created with turn_detection={turn_detection}")

    def say(self, text):
        # Start TTS playback and return a handle
        handle = SpeechHandle(text)
        self._current_handle = handle
        # Simulate asynchronous playback on a separate thread
        def playback():
            print(f"[TTS] start playing: {text}")
            # simulate chunked playback, check for interrupt between chunks
            for i in range(10):
                if handle.is_interrupted():
                    print("[TTS] playback stopped mid-message")
                    return
                time.sleep(0.05)  # simulate short chunk playback
            print("[TTS] playback finished")
        t = threading.Thread(target=playback, daemon=True)
        t.start()
        return handle

    def interrupt(self):
        # Stop all playback
        if self._current_handle:
            self._current_handle.interrupt()

    def respond_to_transcript(self, transcript):
        # Generate a response to the transcript. Replace with real LLM/agent call.
        response = f"Understood: {transcript}"
        print(f"[Agent] responding: {response}")
        self.say(response)

class ASRClient:
    def recognize(self, audio_segment):
        # Replace with real ASR call
        # For template, return a mocked transcript
        return "mocked user transcript"

# --- End placeholders ---


# Simple event emitter to simulate VAD/speech-start events
class AudioInputMock:
    def __init__(self):
        self._listeners = []

    def on_speech_start(self, callback):
        self._listeners.append(callback)

    def simulate_user_starts_speaking(self, audio_segment=None):
        event = {"audio_segment": audio_segment}
        for cb in self._listeners:
            try:
                cb(event)
            except Exception as e:
                print(f"[AudioInputMock] listener error: {e}")


# Realistic interruption manager
class BargeInManager:
    def __init__(self, session: AgentSession, asr: ASRClient):
        self.session = session
        self.asr = asr
        self.current_handle = None
        self.lock = threading.Lock()

    def speak(self, text):
        with self.lock:
            self.current_handle = self.session.say(text)

    def on_user_speech_start(self, event):
        # Called when VAD detects user started speaking.
        # Immediately interrupt current TTS and hand off to ASR.
        with self.lock:
            print("[BargeInManager] user started speaking -> interrupting TTS")
            try:
                if self.current_handle:
                    self.current_handle.interrupt()
                else:
                    # fallback: interrupt whole session
                    self.session.interrupt()
            except Exception as e:
                print(f"[BargeInManager] interrupt call failed: {e}")
                self.session.interrupt()

            # Optional small smoothing delay to avoid jarring audio artifacts
            time.sleep(0.04)  # 40ms fade/gap — tune as needed

        # Route audio to ASR and continue flow
        audio = event.get("audio_segment")
        transcript = self.asr.recognize(audio)
        print(f"[ASR] transcript: {transcript}")

        # Let the session/agent respond to the transcript
        self.session.respond_to_transcript(transcript)

        # Clear current handle after handing off
        with self.lock:
            self.current_handle = None


def main():
    # Create session with HIGH turn detection for faster barge-in responsiveness
    session = AgentSession(turn_detection="HIGH")
    asr = ASRClient()

    barge_manager = BargeInManager(session, asr)

    # Hook into your real audio input's speech-start event
    audio_input = AudioInputMock()
    audio_input.on_speech_start(barge_manager.on_user_speech_start)

    # Start by having the agent speak something longer that could be interrupted
    barge_manager.speak("Hello — I will explain the new onboarding steps in a few parts. If you interrupt, I'll stop and listen.")

    # Simulate the user interrupting after 0.25 seconds
    time.sleep(0.25)
    audio_input.simulate_user_starts_speaking(audio_segment=b"<audio-bytes>")

    # Let threads run briefly to show flow in this mock
    time.sleep(1.0)

if __name__ == "__main__":
    main()

AgentSession

turn_detection="HIGH"

SpeechHandle

session.say(...)

.interrupt()

on_user_speech_start

ASRClient.recognize()

agent.py

turn_detection

.env

turn_detection

AWS_ACCESS_KEY_ID=your_aws_access_key_id
AWS_SECRET_ACCESS_KEY=your_aws_secret_access_key
AWS_DEFAULT_REGION=us-east-1
LIVEKIT_URL=your_livekit_url
LIVEKIT_API_KEY=your_livekit_api_key
LIVEKIT_API_SECRET=your_livekit_api_secret

pip install "livekit-plugins-aws[realtime]" "livekit-agents[silero]" python-dotenv

"""
agent.py

Complete example showing:
- Nova Sonic 2.0 via LiveKit AWS realtime plugin
- turn_detection tuned for responsive barge-in
- prewarming Silero VAD
- interruption handling (interrupting current TTS playback)
Adapt placeholders and imports as needed for your environment.
"""

import logging
from dotenv import load_dotenv
from typing import Optional

# LiveKit agents & plugins
from livekit.agents import Agent, AgentSession, AgentServer, JobContext, JobProcess, cli
from livekit.plugins import aws, silero

load_dotenv()

# Configure logging
logging.basicConfig(level=<http://logging.INFO|logging.INFO>)
logger = logging.getLogger("nova-sonic-agent")

# Create the agent server instance
server = AgentServer()

# Prewarm VAD once per process for faster interruption response
def prewarm(proc: JobProcess):
    """
    Load Silero VAD model into proc.userdata for reuse across sessions.
    This avoids loading on each connection and reduces latency.
    """
    proc.userdata["vad"] = silero.VAD.load()
    <http://logger.info|logger.info>("Silero VAD prewarmed for process")

server.setup_fnc = prewarm

class NovaSonicAgent(Agent):
    def __init__(self) -> None:
        super().__init__(
            instructions="""
            You are a friendly voice assistant using Nova Sonic 2.0. Speak naturally,
            be concise, and allow users to interrupt. If interrupted, stop speaking
            and handle the user's input promptly.
            """.strip()
        )

    async def on_enter(self):
        # Agent speaks first when the session starts
        await self.session.generate_reply(
            instructions="Hello — I am your assistant. How can I help you today?"
        )

@server.rtc_session(agent_name="nova-sonic-agent")
async def entrypoint(ctx: JobContext):
    """
    RTC session entrypoint. Configures Nova Sonic 2.0 and starts AgentSession.
    """
    ctx.log_context_fields = {"room": ctx.room.name}
    <http://logger.info|logger.info>("Starting Nova Sonic 2.0 session for room=%s", ctx.room.name)

    # Configure the Nova Sonic 2 realtime model:
    # - voice: pick a voice available in your account/region (e.g., "tiffany")
    # - turn_detection: HIGH/MEDIUM/LOW (HIGH -> more responsive, more false interrupts)
    # - modalities="mixed" enables both audio and programmatic text input
    llm = aws.realtime.RealtimeModel.with_nova_sonic_2(
        voice="tiffany",
        modalities="mixed",
        turn_detection="MEDIUM",  # tune to HIGH for faster detection if needed
        generate_reply_timeout=15.0,
    )

    # Build AgentSession with prewarmed VAD for better barge-in coordination
    session = AgentSession(
        llm=llm,
        vad=ctx.proc.userdata.get("vad"),  # Silero VAD prewarmed in server.setup_fnc
        # Optional tuning parameters (if available in your SDK):
        # min_interruption_duration=0.15,
        # allow_interruptions=True,
    )

    # Start the session with our agent and connect to the LiveKit room
    agent = NovaSonicAgent()
    await session.start(agent=agent, room=ctx.room)
    await ctx.connect()

    # The session will now handle incoming audio, interruption events, and replies.
    # If you need programmatic interruption control (stop currently playing TTS),
    # you can call session.interrupt() or use the SpeechHandle returned from say().
    #
    # Example (conceptual):
    # handle = session.say("This message can be interrupted")
    # ...on user speech detected: handle.interrupt() or session.interrupt()

if __name__ == "__main__":
    cli.run_app(server)

turn_detection="HIGH"

SpeechHandle

session.say()

handle.interrupt()

session.interrupt()

interrupt()

SpeechHandle

handle.interrupt()