what is the latest replacement for agent_speech_committed, etc in order to get the transcript for what the agent says in the new version of livekit

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

but I feel this does not consider the agent_speech_interrupted conversation which we had earlier

Thanks — you're asking how to handle the old

agent_speech_interrupted

behavior (in addition to

agent_speech_committed

) in the new LiveKit version. There is no direct

agent_speech_interrupted

event. To detect/handle interruptions you can: • Listen for

conversation_item_added

for committed transcript pieces (as before) and check

role == "assistant"

. • Check the

SpeechHandle.interrupted

flag returned by

session.say()

/

session.generate_reply()

or available as

run_ctx.speech_handle.interrupted

to know if that speech was interrupted. • Optionally watch

session.on("agent_state_changed")

for transitions (e.g.,

speaking

→ other) which can indicate interruption. If you want, tell me which SDK/language you’re using and whether you need the partial (pre-interruption) text, final truncated text, or interruption timestamps — I’ll provide a short code snippet for that environment. Sources: Turns overview | LiveKit docs | Agent speech and audio | LiveKit docs | Tool definition and use | LiveKit docs

Give a detailed code to implmet smthg likt this :

@session.on("conversation_item_added")
    def on_conversation_item_added(event):
        try:
            timestamp = get_elapsed_time()

            # NEW: extract real conversation item
            ci = getattr(event, "item", None)
            if not ci:
                logger.warning("conversation_item_added: no conversation item in event")
                return

            # speaker
            if ci.role == "user":
                identity = "CANDIDATE"
                if getattr(ci, "participant", None):
                    identity = ci.participant.identity
                    if "listener" in identity.lower():
                        identity = "Admin"

            elif ci.role == "assistant":
                identity = "AGENT"
            else:
                return

            # text extraction
            text = None
            content = getattr(ci, "content", None)

            if isinstance(content, str):
                text = content

            elif isinstance(content, list):
                for block in content:
                    if hasattr(block, "text") and block.text:
                        text = block.text
                        break

            if not text and hasattr(ci, "text"):
                text = ci.text

            if text:
                clean = clean_text(text)
                if clean.strip():
                    fnc_ctx.transcription[timestamp] = [identity, clean]
                    <http://logger.info|logger.info>(f"[TRANSCRIPT] {identity}: {clean[:100]}...")

        except Exception as e:
            logger.error(f"[conversation_item_added ERROR] {e}", exc_info=True)

Thanks — you want a complete, detailed Python implementation of a

conversation_item_added

handler that extracts speaker identity, pulls the text (including different content shapes), logs/transcribes entries with timestamps, and accounts for interruptions via

SpeechHandle.interrupted

. Below is a self-contained example you can adapt to your environment. Notes: • This uses the

AgentSession

-style event model shown earlier. • It extracts text from

item.content

(string or list of blocks) or

item.text

. • It determines identity from

item.role

and

item.participant.identity

when present. • It records a timestamped transcript dict and logs entries. • It shows how to check a

SpeechHandle.interrupted

flag when available (you must attach/check the handle where you call

session.say()

/

session.generate_reply()

). Code example:

import logging
import time
from typing import Any, Dict

logger = logging.getLogger("transcription")
logger.setLevel(<http://logging.INFO|logging.INFO>)
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(message)s"))
logger.addHandler(handler)

# Global transcription storage (timestamp -> [identity, text, interrupted_flag?])
transcription: Dict[float, Dict[str, Any]] = {}

def get_elapsed_time() -> float:
    """Return a monotonic timestamp suitable as a key for transcripts."""
    return time.time()

def clean_text(s: str) -> str:
    """Basic cleaning: normalize whitespace and strip control chars. Extend as needed."""
    if not s:
        return ""
    cleaned = " ".join(s.split())
    return cleaned.strip()

def extract_text_from_content(content) -> str:
    """
    Extract first available text from content that can be:
    - a plain string
    - a list of block objects with .text attributes or dicts with 'text'
    - None
    """
    if not content:
        return None

    # plain string
    if isinstance(content, str):
        return content

    # list of blocks
    if isinstance(content, list):
        for block in content:
            # object with .text
            text = getattr(block, "text", None)
            if text:
                return text
            # dict-like with 'text' key
            if isinstance(block, dict) and block.get("text"):
                return block.get("text")

    # fallback
    return None

def determine_identity(ci) -> str:
    """
    Determine speaker identity label from ConversationItem:
    - user -> "CANDIDATE" or participant.identity (special-case 'listener' -> 'Admin')
    - assistant -> "AGENT"
    - else -> None
    """
    role = getattr(ci, "role", None)
    if role == "user":
        identity = "CANDIDATE"
        participant = getattr(ci, "participant", None)
        if participant and getattr(participant, "identity", None):
            identity = participant.identity
            if "listener" in identity.lower():
                identity = "Admin"
        return identity
    elif role == "assistant":
        return "AGENT"
    return None

def get_speech_interrupted_flag(ci) -> bool:
    """
    Attempt to read an interruption flag from the ConversationItem.
    Common places:
    - ci.speech_handle.interrupted
    - ci.interrupted
    - ci.meta or other container depending on SDK; adapt as needed.
    """
    # direct flag
    if hasattr(ci, "interrupted"):
        return bool(getattr(ci, "interrupted"))
    # nested speech_handle
    speech_handle = getattr(ci, "speech_handle", None)
    if speech_handle and hasattr(speech_handle, "interrupted"):
        return bool(getattr(speech_handle, "interrupted"))
    # run_ctx style (if available)
    run_ctx = getattr(ci, "run_ctx", None)
    if run_ctx:
        sh = getattr(run_ctx, "speech_handle", None)
        if sh and hasattr(sh, "interrupted"):
            return bool(getattr(sh, "interrupted"))
    return False

# Example event handler registration (depends on your SDK)
# @session.on("conversation_item_added")
def on_conversation_item_added(event) -> None:
    """
    Event handler that:
    - validates event.item
    - extracts identity and text
    - checks interruption
    - stores transcription and logs a short preview
    """
    try:
        timestamp = get_elapsed_time()

        # NEW: extract real conversation item
        ci = getattr(event, "item", None)
        if not ci:
            logger.warning("conversation_item_added: no conversation item in event")
            return

        # speaker identity
        identity = determine_identity(ci)
        if not identity:
            # unknown role - ignore
            return

        # text extraction
        text = extract_text_from_content(getattr(ci, "content", None))

        # fallback to ci.text
        if not text and hasattr(ci, "text"):
            text = getattr(ci, "text")

        if not text:
            # nothing to save
            return

        # clean text
        clean = clean_text(text)
        if not clean:
            return

        # interruption detection
        interrupted = get_speech_interrupted_flag(ci)

        # store transcript entry (you can adapt structure)
        transcription[timestamp] = {
            "identity": identity,
            "text": clean,
            "interrupted": interrupted,
        }

        # log a preview
        preview = clean if len(clean) <= 200 else clean[:200] + "..."
        if interrupted:
            <http://logger.info|logger.info>(f"[TRANSCRIPT] {identity} (INTERRUPTED): {preview}")
        else:
            <http://logger.info|logger.info>(f"[TRANSCRIPT] {identity}: {preview}")

    except Exception as e:
        logger.exception(f"[conversation_item_added ERROR] {e}")

# Example of attaching speech handles when you initiate speech (pseudo-code)
# If you call session.say(...) or session.generate_reply(...), capture the returned SpeechHandle
# and store mapping from some ID to that handle so you can later check `.interrupted`.
#
# Example (pseudo):
# sh = session.say("Hello, how can I help?")
# # later, when you get a conversation_item_added referencing this speech,
# # the item may include a speech_handle reference you can inspect for `.interrupted`.

# To wire this up: register `on_conversation_item_added` with whatever event API your SDK provides.
# If you're using the Python AgentSession from earlier messages, use:
# @session.on("conversation_item_added")
# def on_conversation_item_added(event): ...
#
# Adapt the helper functions if your ConversationItem shapes differ (e.g. dicts vs objects).

If you want, tell me: • which SDK and version you’re using (Python, Node, or another), and whether ConversationItems arrive as dicts or objects; and • whether you prefer transcripts stored in a list (chronological) instead of keyed by timestamp. I’ll adapt the snippet to your environment and include how to associate `SpeechHandle`s with conversation items for guaranteed interruption detection.

I am using python and I basically want all the users an agent transcription stored together and added to this :fnc_ctx.transcription[timestamp] = [identity, clean]

Thanks — I’ll give a focused, ready-to-use Python handler that: • Works with the Python AgentSession event model • Extracts speaker identity for both users and agents • Pulls text from the different content shapes (

str

, list of blocks,

ci.text

) • Detects interruptions via

speech_handle.interrupted

/

ci.interrupted

• Stores entries in your existing

fnc_ctx.transcription[timestamp] = [identity, clean]

• Keeps entries chronological (uses monotonic timestamp) and logs a short preview Paste/adapt this into your code and register the handler with your

session.on("conversation_item_added")

.

import logging
import time
from typing import Any, Dict, Optional

logger = logging.getLogger("transcription")
logger.setLevel(<http://logging.INFO|logging.INFO>)
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(message)s"))
logger.addHandler(handler)

class FnCtx:
    def __init__(self):
        # keyed by timestamp -> [identity, clean, interrupted_flag(optional)]
        self.transcription: Dict[float, list] = {}

fnc_ctx = FnCtx()

def get_elapsed_time() -> float:
    # Use time.time() for wall-clock; replace with time.monotonic() if preferred
    return time.time()

def clean_text(s: str) -> str:
    if not s:
        return ""
    # normalize whitespace and strip
    return " ".join(s.split()).strip()

def extract_text_from_content(content: Any) -> Optional[str]:
    # plain string
    if isinstance(content, str):
        return content

    # list of blocks (objects or dicts)
    if isinstance(content, list):
        for block in content:
            # object with .text
            text = getattr(block, "text", None)
            if text:
                return text
            # dict-like with 'text'
            if isinstance(block, dict) and block.get("text"):
                return block.get("text")

    return None

def determine_identity(ci: Any) -> Optional[str]:
    role = getattr(ci, "role", None)
    if role == "user":
        identity = "CANDIDATE"
        participant = getattr(ci, "participant", None)
        if participant and getattr(participant, "identity", None):
            identity = participant.identity
            if "listener" in identity.lower():
                identity = "Admin"
        return identity
    if role == "assistant":
        return "AGENT"
    return None

def get_speech_interrupted_flag(ci: Any) -> bool:
    # direct flag
    if hasattr(ci, "interrupted"):
        return bool(getattr(ci, "interrupted"))
    # nested speech_handle
    speech_handle = getattr(ci, "speech_handle", None)
    if speech_handle and hasattr(speech_handle, "interrupted"):
        return bool(getattr(speech_handle, "interrupted"))
    # run_ctx style
    run_ctx = getattr(ci, "run_ctx", None)
    if run_ctx:
        sh = getattr(run_ctx, "speech_handle", None)
        if sh and hasattr(sh, "interrupted"):
            return bool(getattr(sh, "interrupted"))
    return False

# Register this with your AgentSession: @session.on("conversation_item_added")
def on_conversation_item_added(event) -> None:
    try:
        timestamp = get_elapsed_time()

        ci = getattr(event, "item", None)
        if not ci:
            logger.warning("conversation_item_added: no conversation item in event")
            return

        identity = determine_identity(ci)
        if not identity:
            # ignore system/other items
            return

        # Try content shapes first
        content = getattr(ci, "content", None)
        text = extract_text_from_content(content)

        # fallback to ci.text
        if not text and hasattr(ci, "text"):
            text = getattr(ci, "text")

        if not text:
            return

        clean = clean_text(text)
        if not clean:
            return

        interrupted = get_speech_interrupted_flag(ci)

        # Store per your requested shape: fnc_ctx.transcription[timestamp] = [identity, clean]
        # If you want to store interruption info, append it as a third element.
        entry = [identity, clean]
        if interrupted:
            entry.append(True)

        fnc_ctx.transcription[timestamp] = entry

        preview = clean if len(clean) <= 200 else clean[:200] + "..."
        if interrupted:
            <http://logger.info|logger.info>(f"[TRANSCRIPT] {identity} (INTERRUPTED): {preview}")
        else:
            <http://logger.info|logger.info>(f"[TRANSCRIPT] {identity}: {preview}")

    except Exception as e:
        logger.exception(f"[conversation_item_added ERROR] {e}")

# Example note:
# If you initiate agent speech yourself (session.say() or session.generate_reply()),
# capture the returned SpeechHandle and keep a mapping to verify interruption later
# so the conversation item can be correlated to that handle if needed.
#
# For example:
#   sh = session.say("Hello")
#   speech_handles_map[sh.id] = sh
# Later you can check sh.interrupted or correlate with the incoming ci if it references the handle.

If you want, I can: • Convert this into a chronological list (append to a list) instead of using timestamps as keys. • Add logic to deduplicate near-identical items or merge partial segments produced when interruptions occur. • Provide an example showing how to capture and correlate

SpeechHandle

objects returned by

session.say()

/

session.generate_reply()

to guarantee interruption detection. Which of those would you like next?

what is the new liveit version of this : @billions-ice-17644.on("user_started_speaking") def user_speaking(): nonlocal speech_flag, warned speech_flag = True warned = False cancel_silence_timer() @billions-ice-17644.on("user_stopped_speaking") def user_stopped(): nonlocal speech_flag speech_flag = False start_silence_timer() @billions-ice-17644.on("agent_started_speaking") def agent_speaking(): nonlocal speech_flag speech_flag = True cancel_silence_timer() @billions-ice-17644.on("agent_stopped_speaking") def agent_stopped(): nonlocal speech_flag speech_flag = False start_silence_timer() def pause_silence_timer(): shared_state.silence_timer_paused = True pause_event.clear() cancel_silence_timer()

You're using Python and want the new LiveKit equivalents for those start/stop speaking handlers plus silence-timer logic. Use the

user_state_changed

and

agent_state_changed

events and check

new_state == "speaking"

/

new_state == "listening"

(or

"thinking"

for agent stop). Below is a concise, ready-to-use Python example that mirrors your original handlers and pause/resume silence-timer behavior.

import threading
import time
from typing import Optional

# Replace these with your session and event imports
# from livekit.agents import AgentSession, UserStateChangedEvent, AgentStateChangedEvent

# Shared state
class SharedState:
    def __init__(self):
        self.silence_timer_paused = False
        self.silence_timer: Optional[threading.Timer] = None

shared_state = SharedState()

# Local flags used in your original snippet
speech_flag = False
warned = False
pause_event = threading.Event()
pause_event.set()  # set = not paused; clear = paused

SILENCE_TIMEOUT = 5.0  # seconds, adjust as needed

def cancel_silence_timer():
    if shared_state.silence_timer is not None:
        shared_state.silence_timer.cancel()
        shared_state.silence_timer = None

def start_silence_timer():
    if shared_state.silence_timer_paused:
        return
    cancel_silence_timer()
    shared_state.silence_timer = threading.Timer(SILENCE_TIMEOUT, on_silence_timeout)
    shared_state.silence_timer.start()

def on_silence_timeout():
    global speech_flag, warned
    # handle silence timeout (user or agent stopped speaking long enough)
    speech_flag = False
    warned = False
    # Put your silence handling logic here
    print("Silence timeout fired")

def pause_silence_timer():
    shared_state.silence_timer_paused = True
    pause_event.clear()
    cancel_silence_timer()

def resume_silence_timer():
    shared_state.silence_timer_paused = False
    pause_event.set()
    # Optionally restart the timer immediately or leave it to next stop event
    start_silence_timer()

# Event handlers using new LiveKit events
# Register these with your AgentSession: @session.on("user_state_changed") / @session.on("agent_state_changed")
def on_user_state_changed(ev):
    """
    ev.new_state can be "speaking", "listening", "away", etc.
    """
    global speech_flag, warned
    new_state = getattr(ev, "new_state", None)

    if new_state == "speaking":
        speech_flag = True
        warned = False
        cancel_silence_timer()
        # optional: any other logic for user started speaking
        print("User started speaking")
    elif new_state == "listening":
        speech_flag = False
        start_silence_timer()
        print("User stopped speaking")

def on_agent_state_changed(ev):
    """
    ev.new_state can be "speaking", "listening", "thinking", etc.
    Treat "thinking" and "listening" as stops for speaking.
    """
    global speech_flag
    new_state = getattr(ev, "new_state", None)

    if new_state == "speaking":
        speech_flag = True
        cancel_silence_timer()
        print("Agent started speaking")
    elif new_state in ("listening", "thinking"):
        speech_flag = False
        start_silence_timer()
        print("Agent stopped speaking")

# Example: wiring to session (pseudo-code - adapt to your SDK)
# session.on("user_state_changed")(on_user_state_changed)
# session.on("agent_state_changed")(on_agent_state_changed)

# Example usage: pause/resume from other logic
# pause_silence_timer()
# resume_silence_timer()

If you want, I can: • Convert this to use asyncio instead of threading (if your app is async). • Add debounce logic to ignore rapid start/stop flapping. • Show the exact decorator registration for your LiveKit Python SDK version. Which would you prefer? Sources: Turns overview | LiveKit docs | Agent session | LiveKit docs

is this still supported in new livkeit : @billions-ice-17644.on("user_started_speaking") def _user_started(): nonlocal last_user_activity, warned last_user_activity = time.monotonic() warned = False cancel_silence_timer() @billions-ice-17644.on("user_speech_committed") def _user_committed(msg: llm.ChatMessage): nonlocal last_user_activity, warned last_user_activity = time.monotonic() warned = False cancel_silence_timer() @billions-ice-17644.on("user_stopped_speaking") def _user_stopped(): start_silence_timer()

Short answer: those legacy events are deprecated. Use

user_state_changed

plus

user_input_transcribed

(or

conversation_item_added

) instead. What to use: • Replace

user_started_speaking

/

user_stopped_speaking

with

user_state_changed

and check

ev.new_state == "speaking"

/

ev.new_state == "listening"

. • Replace

user_speech_committed

with

user_input_transcribed

and check

transcript.is_final == True

, or use

conversation_item_added

for committed chat items. Concise Python example that mirrors your original logic (silence timer, last_user_activity, warned, and cancel/start behavior):

import time
import asyncio

last_user_activity = time.monotonic()
warned = False
silence_task = None
SILENCE_TIMEOUT = 5.0

def cancel_silence_timer():
    global silence_task
    if silence_task and not silence_task.done():
        silence_task.cancel()
    silence_task = None

def start_silence_timer():
    global silence_task
    cancel_silence_timer()
    async def _wait():
        try:
            await asyncio.sleep(SILENCE_TIMEOUT)
            # silence timeout action
            print("Silence timeout fired")
        except asyncio.CancelledError:
            return
    silence_task = asyncio.create_task(_wait())

# Register these on your AgentSession:
# @session.on("user_state_changed")
def _on_user_state_changed(ev):
    global last_user_activity, warned
    new_state = getattr(ev, "new_state", None)
    if new_state == "speaking":
        last_user_activity = time.monotonic()
        warned = False
        cancel_silence_timer()
    elif new_state == "listening":
        start_silence_timer()

# @session.on("user_input_transcribed")
def _on_user_input_transcribed(transcript):
    # transcript.is_final indicates commit
    global last_user_activity, warned
    if getattr(transcript, "is_final", False):
        last_user_activity = time.monotonic()
        warned = False
        cancel_silence_timer()
        # transcript.transcript contains final text

# Optional: also listen to conversation_item_added for committed user messages
# @session.on("conversation_item_added")
def _on_conversation_item_added(event):
    item = getattr(event, "item", None)
    if not item:
        return
    if getattr(item, "role", None) == "user":
        # handle committed user chat item
        start_silence_timer()

If you want, I can: • Convert this to a threaded version (non-async) or adapt it to your existing async app structure. • Show the exact decorator/registration lines for your LiveKit Python SDK version. Which do you prefer? Sources: Turns overview | LiveKit docs | Agent session | LiveKit docs | Transcriber | LiveKit docs | Interrupt User | LiveKit docs