If the user has interrupted the agent while speaking, is there a way to get the interrupted: true/false field to my next agent request similar to what I see in the metrics?

Thanks for reaching out, Bharath. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.

I'm using python livekit sdk

Great — thanks for confirming. Below is a concise Python example showing how to capture

interrupted

from the

SpeechHandle

and pass it into your next agent request (via

instructions

or any other field you use for context).

# Start speech and get the handle
handle = session.say("Hello, how can I help you?")

# Wait for playback to finish
await handle  # or: await handle.wait_for_playout() if using that API

# Check interrupted flag
was_interrupted = handle.interrupted  # True/False

# Use the flag in your next request (example: include in instructions)
if was_interrupted:
    next_instructions = "The user interrupted you. Give a very brief summary and ask if they want more."
else:
    next_instructions = "The user listened fully. Continue with the detailed explanation."

# Send next agent request
session.generate_reply(instructions=next_instructions)

If you prefer to store the flag in metadata or session context instead of

instructions

, replace the

instructions

argument with your preferred field. Tell me which approach you want and I’ll give the exact snippet.

I'm using AgentSession and VoicePipelineagent how does it work?

Thanks — here’s exactly how interruption works with AgentSession and VoicePipelineAgent in the Python SDK and a minimal example you can drop into your flow. Key points • Both AgentSession and VoicePipelineAgent return a SpeechHandle from

session.say()

and

session.generate_reply()

. • After playback finishes (await the handle), check

handle.interrupted

(True/False). • Use that boolean in your next agent request (e.g., put it in

instructions

, session context, or metadata) to change behavior. Minimal Python example (AgentSession / VoicePipelineAgent)

# start speaking (AgentSession or VoicePipelineAgent exposes session.say/generate_reply)
handle = session.say("Hello, how can I help you?")

# wait for playout to finish
await handle  # or: await handle.wait_for_playout()

# read interrupted flag
was_interrupted = handle.interrupted  # True or False

# include flag in next agent request (instructions, context, or metadata)
if was_interrupted:
    next_instructions = "User interrupted — give a very brief summary and ask a clarifying question."
else:
    next_instructions = "User listened — continue with full details."

await session.generate_reply(instructions=next_instructions)

Notes and options • You can store the flag in session context or metadata instead of

instructions

if you prefer. • For global handling, listen to the

playback_finished

event;

ev.interrupted

is available there. • To prevent interruptions during critical operations, use

allow_interruptions=False

on the speech call or

run_ctx.disallow_interruptions()

while running long tool calls. If you share a short snippet of how you currently call

say()

/

generate_reply()

I’ll adapt this exact example to your code. Sources: Agent speech and audio | LiveKit docs | Tool definition and use | LiveKit docs | Uninterruptable Agent | LiveKit docs | Improving voice AI's turn detection with transformers | Voice AI quickstart | LiveKit docs

But I just bind STT/LLM/TTS node into my AgentSession

AgentSession(
        vad=vadConfig,
        # any combination of STT, LLM, TTS, or realtime API can be used
        llm=openai.LLM(model="gpt-4o-mini"),
        stt=sttConfig,
        tts=tts_config,
        # use LiveKit's turn detection model
        turn_detection=MultilingualModel(),
        allow_interruptions=True,
        discard_audio_if_uninterruptible = False,
        min_endpointing_delay= config["agent_session.min_endpointing_delay"],
        max_endpointing_delay= config["agent_session.max_endpointing_delay_initial"],
        min_interruption_duration=config["agent_session.min_interruption_duration"], 
        min_interruption_words=config["agent_session.min_interruption_words"] 

    )

and there are no places where I actually use session.say methods

Thanks — since you bind STT/LLM/TTS into AgentSession and don’t call

session.say()

yourself, the interrupted flag is produced internally but not exposed as a SpeechHandle you can read directly. What you can do instead: • Listen for agent state transitions to infer interruptions:

@session.on("agent_state_changed")
def on_agent_state_change(ev):
    if ev.previous_state == "speaking" and ev.new_state == "listening":
        # interruption detected

• Use

on_user_turn_started

/

on_user_turn_completed

in your Agent subclass to detect user speech that begins during agent playback:

class MyAgent(Agent):
    async def on_user_turn_started(self):
        if self.session.current_speech and not self.session.current_speech.finished:
            # interruption detected

• For global handling, subscribe to

playback_finished

/

playback_finished

-style events (they include

ev.interrupted

) if available in your runtime. If you need an explicit

interrupted

boolean per response (to log or pass into the next LLM call), you have two options: • Keep automatic pipeline but infer interruptions via the above events and set a session/context flag you read in the next turn. • Switch the agent to manual reply generation: stop auto-generation and call

session.generate_reply()

(or

session.say()

), await the returned

SpeechHandle

, then read

handle.interrupted

and act on it. Which would you prefer — a short code example to add event listeners to your current AgentSession pipeline, or a snippet showing how to switch to manual

generate_reply()

flow? Sources: Turns overview | LiveKit docs | examples/voice_agents | livekit/livekit_composite | DeepWiki | Agent speech and audio | LiveKit docs

I prefer the first version based on events

Great — since you prefer the event-based approach, here’s a concise, drop-in set of patterns and code you can add to your existing AgentSession / VoicePipelineAgent setup to detect interruptions and surface that state for the next agent turn. 1) Detect interruption via agent state change

from livekit.agents import AgentStateChangedEvent

@session.on("agent_state_changed")
def on_agent_state_changed(ev: AgentStateChangedEvent):
    # interruption: agent was speaking and switched to listening
    if ev.from_state == "speaking" and ev.to_state == "listening":
        # mark interruption in session context / metadata so next turn can read it
        session.context["last_turn_interrupted"] = True
        <http://logger.info|logger.info>("Agent was interrupted by user.")

2) Detect user speech start (early signal)

from livekit.agents import UserStateChangedEvent

@session.on("user_state_changed")
def on_user_state_changed(ev: UserStateChangedEvent):
    if ev.state == "speaking":
        # early indication user began speaking — may imply interruption
        session.context["user_began_speaking_at"] = ev.timestamp

3) Clear or read the interruption flag in the next agent turn

# inside your agent logic that runs when processing the next user turn:
was_interrupted = session.context.pop("last_turn_interrupted", False)

if was_interrupted:
    instructions = "The user interrupted the previous response. Give a brief summary and ask a clarifying question."
else:
    instructions = "Continue with the normal detailed response."

await session.generate_reply(instructions=instructions)

4) Optional: observe current speech object for an active play

# check if agent is currently speaking (useful inside on_user_turn_started)
current = getattr(session, "current_speech", None)
if current and not current.finished:
    # Treat this as an interruption
    session.context["last_turn_interrupted"] = True

5) Tuning and notes • Keep using your existing AgentSession options (min_interruption_duration, min_interruption_words, allow_interruptions) to tune sensitivity. • Use

user_state_changed

for early detection, and

agent_state_changed

for definitive interruption detection. • Store the flag in

session.context

or metadata so it’s available to the next LLM call or tool run. If you’d like, I can adapt these snippets to match the exact class names and event payload shapes in your codebase (share a short snippet of your Agent subclass or how you access

session

) and provide a ready-to-drop patch. Sources: Turns overview | LiveKit docs | Agent speech and audio | LiveKit docs | Tool definition and use | LiveKit docs | Agent session | LiveKit docs

Whenever i collect metrics using:

@session.on("metrics_collected")
    def _on_metrics_collected(ev: MetricsCollectedEvent):
        metrics.log_metrics(ev.metrics)
        usage_collector.collect(ev.metrics)

I'm unable to get ctx details. Is there a way to capture the chat_ctx data here?

You're right —

metrics_collected

does not include

chat_ctx

. That event is telemetry-only and is decoupled from conversation state. You can still correlate metrics with chat context using one of these lightweight patterns: • Store relevant context in session-scoped storage before generation

# save before LLM/TTS runs
session.userdata["last_request_id"] = current_request_id
session.userdata["last_user_message"] = last_user_message_text

Then read those fields inside your metrics handler:

@session.on("metrics_collected")
def _on_metrics_collected(ev):
    req_id = session.userdata.get("last_request_id")
    user_msg = session.userdata.get("last_user_message")
    metrics.log_metrics(ev.metrics)
    usage_collector.collect(ev.metrics, request_id=req_id, user_message=user_msg)

• Capture context in lifecycle hooks (recommended) Use

on_user_turn_completed

or similar to snapshot the chat context right before generation:

async def on_user_turn_completed(self, turn_ctx, new_message):
    self.session.userdata["ctx_snapshot"] = [m.model_dump() for m in turn_ctx.messages]
    self.session.userdata["current_request_id"] = generate_request_id()

Then correlate that snapshot in

metrics_collected

. • Correlate by request_id If metrics include a

request_id

, store a mapping from

request_id

→ chat context when you start the request and look it up in the handler. • Emit enriched/custom events Emit your own event that bundles metrics + serialized chat context if you need an atomic record:

session.emit("enriched_metrics", {"metrics": ev.metrics, "chat_ctx": session.userdata["ctx_snapshot"]})

Which option works best for you? If you want, share where you currently record the user turn (a snippet of your Agent methods) and I’ll give a concrete patch to wire one of these patterns into your

metrics_collected

handler. Sources: Agents (Node.js) | livekit/livekit_composite | DeepWiki | LLM Metrics | LiveKit docs | TTS Metrics | LiveKit docs | External data and RAG | LiveKit docs | https://blog.livekit.io/streamline-troubleshooting-with-agent-observability/#/portal/signup|