We have a custom pipeline to stream the agents output as text from the

transcription_node

as chunks to our client where it is rendered. And once the message is final, which we take from

@session.on("conversation_item_added")

we send it as complete message. However, the text from the transcription node is much faster the the agent's audio response. So if the agent is interrupted we already streamed much more (or even the whole) response. In the worstcase, we stream a response but the agent is interrupted before it starts responsing, meaning we streamed an agent response that afterwards doesn't exist at all. How can I intercept and stream only the text that agent is actually saying?

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

we use elevenlabs for TTS, I already enabled

use_tts_aligned_transcript

but it didn't really help. here is the transcription node:

async def transcription_node(
        self, text: AsyncIterable[str], model_settings: ModelSettings
    ) -> AsyncIterable[str]:
        """
        Intercepts agent transcription to stream chunks via Redis.

        The final message is handled separately by conversation_item_added event,
        which calls session_recorder.log_and_stream_conversation_item.
        Message ID continuity is maintained via transcription_drafts.
        """
        # Get session recorder from session userdata
        session_recorder = getattr(self.session.userdata, "session_recorder", None)

        # Start a new transcription stream (generates message_id for this turn)
        if session_recorder:
            try:
                await session_recorder.start_agent_transcription_stream()
            except Exception as e:
                logger.warning(f"Failed to start transcription stream: {e}")

        async for delta in Agent.default.transcription_node(self, text, model_settings):
            # Publish chunk to Redis for real-time streaming (non-blocking)
            if session_recorder and delta:
                try:
                    await session_recorder.publish_agent_transcription_chunk(delta)
                except Exception as e:
                    logger.warning(f"Failed to publish transcription chunk: {e}")

            yield delta

Thanks — helpful details. You're already on the right track with ElevenLabs and

use_tts_aligned_transcript

. Since that alone didn't fix it, the transcription node needs to consume the timed transcript objects and only publish words when their timing aligns with actual audio playback. Key changes to make: 1. Ensure

sync_transcription=True

in

TextOutputOptions

when starting the session. 2. Update your transcription node to accept

TimedString

(not plain

str

) and only publish words when their `start_time`/`end_time` indicate they have been played. 3. Track audio playhead progress (server-side or client-side) and gate publishing of each word/chunk until that word’s start_time <= playhead. Suggested pattern (high level): • Change the node signature to expect

AsyncIterable[str | TimedString]

. • Buffer incoming

TimedString

words and, for each word: ◦ If its

start_time

<= current_played_time: publish it. ◦ If not yet reached: wait (or schedule) until playhead reaches that time, then publish. • On interruption/cancel: stop publishing future buffered words and discard them. Rely on

conversation_item_added

for final committed text. Minimal sketch (conceptual, adapt to your async environment):

async for chunk in upstream_transcription:
    if isinstance(chunk, TimedString):
        for word in chunk.words:  # each has start_time, end_time, text
            await wait_until_played(word.start_time)  # map to current TTS playhead
            await session_recorder.publish_agent_transcription_chunk(word.text)
    else:
        # fallback: treat as provisional, mark as in-progress or skip

Questions to narrow exact code: • Are you driving audio playback timing on the server (emitting audio packets) or on the client (receiving and playing TTS audio)? • Can you access the TTS playhead time where the session runs (or can the client report playhead back to server)? • Do your

TimedString

objects include per-word `start_time`/`end_time` fields (word-level timing)? If you tell me which of the above applies (server vs client playback and whether

TimedString.words

with times is present), I’ll produce a concrete async code snippet you can drop into your

transcription_node

and a simple

wait_until_played

implementation.

We create audio on the server and send it via livekit. I will try the sync_transcription. My biggest concern: how can I make sure I don't stream a response that is interrupted before it starts (meaning the final conversation item is empty as the agent never said a word of that reponse).

Good question — here’s a concise, reliable approach you can implement given server-side TTS + LiveKit and ElevenLabs aligned transcripts. Summary (what to do) • Enable

sync_transcription=True

. • Only start publishing transcription chunks to clients after TTS audio actually begins playing. • Stop/cancel publishing immediately if the speech is interrupted before or during playback. • Rely on

conversation_item_added

to commit the final spoken text; treat all streamed transcription as provisional until then. Concrete steps to adopt 1. Start the transcription stream when audio play begins ◦ Call

session_recorder.start_agent_transcription_stream()

only when you have confirmed audio playback has started (e.g., when you begin sending audio to LiveKit or when LiveKit reports the first audio packets sent). This prevents creating/assigning a message_id for a turn that never starts. 1. Gate word publishing by TTS playhead using

TimedString

◦ Consume

TimedString

objects from the transcription node. Each word has `start_time`/`end_time`. Map those times to your server-side audio playhead (the timestamp offset you start sending audio at). Only publish a word when its

start_time

<= current_played_time. Buffer future words until their time arrives. 1. Handle interruptions immediately ◦ On an interrupt event (agent canceled/stop): ▪︎ Cancel or discard buffered/unplayed words. ▪︎ Close the transcription stream for that turn so no further provisional chunks are sent. ▪︎ Don’t create a final committed message from these provisional chunks. Wait for

conversation_item_added

which will reflect the actual spoken portion (possibly empty or truncated). 1. Use

conversation_item_added

as the single source of truth ◦ When

conversation_item_added

arrives, use it to: ▪︎ Replace the provisional/in-progress transcript on clients with the final text, or ▪︎ If it’s empty (interrupted before speaking), remove the provisional transcript entirely. ◦ Ensure message_id continuity by mapping the

conversation_item_added

message_id to the same draft used for transcription chunks. Minimal behavior flow (server-side) 1. Start producing TTS audio -> begin sending audio to LiveKit -> set playhead_start = now -> call

start_agent_transcription_stream()

. 2. transcription_node receives

TimedString

chunks -> for each word: ◦ if word.start_time <= now - playhead_start: publish ◦ else: buffer until the playhead reaches word.start_time 1. If interrupt occurs before any word was played: ◦ cancel stream, discard buffered words ◦ wait for

conversation_item_added

(will be empty) -> do not emit final message 1. If interrupt occurs after some words were played: ◦ stop publishing further words, discard remainder ◦

conversation_item_added

will contain the truncated final spoken text — replace provisional text with that. Answer to your specific worry (streaming a response that never starts) • Do not start the transcription stream (and do not publish any provisional chunks) until audio has actually started playing. If you delay

start_agent_transcription_stream()

until the first audio packet is sent (or until the server-side playhead begins), then if an interrupt cancels the response before audio starts, you will never have streamed any transcription for that response. If you want, I can provide a concise async code snippet that: • updates your transcription_node to accept

TimedString

, • gates publishing by server playhead, • cancels on interrupt, and • integrates with

conversation_item_added

message handling. Which would you prefer?

can you create me a snippet for: If you want, I can provide a concise async code snippet that: • updates your transcription_node to accept

TimedString

, • gates publishing by server playhead, • cancels on interrupt,

Below is a concise async snippet you can drop into your agent to: • accept

TimedString

(and fallback to

str

), • gate publishing by server-side audio playhead, • start the transcription stream only when audio starts, • cancel/discard buffered words if interrupted. Adjust names to match your session/speech handle and

session_recorder

API.

import asyncio
from typing import AsyncIterable, AsyncGenerator

async def transcription_node(
    self,
    text: AsyncIterable[str | TimedString],
    model_settings: ModelSettings,
) -> AsyncGenerator[str | TimedString, None]:
    """
    Stream transcription words only when spoken. Uses:
    - self.session.current_speech_handle (has .playhead(), .start_time, .interrupted, .wait_until(t))
    - self.session.userdata.session_recorder with start_agent_transcription_stream(), publish_agent_transcription_chunk(), end_agent_transcription_stream()
    """

    # Helpers / handles (adapt to your actual session API)
    speech = getattr(self.session, "current_speech_handle", None)
    session_recorder = getattr(self.session.userdata, "session_recorder", None)

    # If no active speech, passthrough (fallback)
    if not speech:
        async for chunk in text:
            if session_recorder and chunk:
                try:
                    await session_recorder.publish_agent_transcription_chunk(chunk)
                except Exception:
                    pass
            yield chunk
        return

    # Only start transcription stream once audio playback has actually started
    # (ensure you call this when you begin sending audio packets)
    started_stream = False

    async def start_stream_once():
        nonlocal started_stream
        if started_stream:
            return
        started_stream = True
        if session_recorder:
            try:
                await session_recorder.start_agent_transcription_stream()
            except Exception:
                pass

    try:
        async for chunk in text:
            # Stop immediately if interrupted
            if getattr(speech, "interrupted", False):
                # ensure recorder is closed
                if session_recorder and started_stream:
                    try:
                        await session_recorder.end_agent_transcription_stream()
                    except Exception:
                        pass
                break

            # If TimedString: gate by word timings
            if isinstance(chunk, TimedString):
                # Ensure stream only starts when audio playback begins
                # wait until playhead >= first word start (or until playback has started)
                first_start = chunk.start_time if hasattr(chunk, "start_time") else 0.0
                # Wait for audio playhead to reach first_start (maps to server playhead)
                # speech.wait_until(t) should await until playhead >= t; adapt if your API differs
                await start_stream_once()
                try:
                    await speech.wait_until(first_start)
                except asyncio.CancelledError:
                    break

                # After waiting, re-check interruption
                if getattr(speech, "interrupted", False):
                    break

                # Publish the TimedString chunk (or break it into words if needed)
                if session_recorder:
                    try:
                        await session_recorder.publish_agent_transcription_chunk(chunk)
                    except Exception:
                        pass
                yield chunk

            else:
                # Plain string fallback: start stream on first plain chunk if audio has started
                await start_stream_once()
                if session_recorder:
                    try:
                        await session_recorder.publish_agent_transcription_chunk(chunk)
                    except Exception:
                        pass
                yield chunk

    finally:
        # Ensure recorder closed at end of turn
        if session_recorder and started_stream:
            try:
                await session_recorder.end_agent_transcription_stream()
            except Exception:
                pass

If you want, I can: • convert the above to yield per-word text (extracting

chunk.words

), • add example

speech.wait_until

and playhead mapping logic, or • show how to discard partial buffers on interrupt and map to

conversation_item_added

. Which would help most? Sources: Text and transcriptions | LiveKit docs | Agent speech and audio | LiveKit docs | Cartesia TTS plugin guide | LiveKit docs | ElevenLabs TTS plugin guide | LiveKit Docs | How to capture real-time agent transcriptions during conversations

whats the right import for TimedString ?

Use this import:

from livekit.agents.llm import TimedString

If you want, I can add a short example showing how to check

isinstance(chunk, TimedString)

and access

chunk.start_time

,

chunk.end_time

, or

chunk.words

. Would that help? Sources: Text and transcriptions | LiveKit docs

in which livekit agents version was TimedString added?

I don't have the answer you're looking for. You could also try asking your question: • in one of the other Slack channels or • to https://deepwiki.com/livekit/livekit_composite which is trained on all LiveKit source code If you find the answer, please post it here to help others!

ok. an other than TimedString - is there any other way how I can intercept only the text that the agent already voiced? Alternatively can I know if the agent already started speaking its answer?