https://livekit.io logo
Join Slack
Powered by
Issue: Agent's spoken transcript is not saved to c...
# ask-ai
c
Issue: Agent's spoken transcript is not saved to chat context and 
conversation_item_added
event does not fire when user disconnects during speech In my voice agent, I call session.say() with an intro message in the on_enter method. When the agent is actively speaking this intro message and the user disconnects (hangs up the call), the following happens: The assistant message that was being spoken is never added to the chat context (chat_ctx) 1. The session.on("conversation_item_added") event never fires 2. No error or warning is logged — the transcript is silently lost This happens even though the agent had already started speaking and the user heard part (or all) of the message before disconnecting. Setting allow_interruptions=False does not prevent this — the disconnect still causes the transcript to be dropped. What I expect: When the user disconnects mid-speech, the portion of the message that was spoken (or the full message if it completed) should still be added to the chat context, and the 
conversation_item_added
event should still fire. What actually happens: The transcript is completely lost. Downstream logic that depends on 
conversation_item_added
(logging, analytics, etc.) never executes. How to reproduce: In Agent.on_enter(), call 
await self.session.say("Hello, welcome to our service...", allow_interruptions=False)
Listen for session.on("conversation_item_added", handler) Have the user disconnect while the agent is speaking 1. Observe that the handler is never called and the message is not in chat_ctx This affects both say() and generate_reply() — any agent speech that is interrupted by a participant disconnect loses its transcript.
t
Thanks for reaching out, Kushagra. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.
Thanks — I see you’re reporting that when your voice agent is speaking (via 
session.say
or 
session.generate_reply
) and the participant disconnects mid-speech, the assistant message is never added to chat_ctx and the 
conversation_item_added
handler does not fire. You expect the spoken portion (or the full message if it finished) to persist and trigger downstream logic; instead the transcript is silently dropped. What’s happening • When a participant disconnects, the AgentSession is closed and any in-progress speech is canceled. In-progress speech that hasn’t completed playout is not treated as a finished conversation item, so it is not added to the chat context and 
conversation_item_added
does not fire. • 
allow_interruptions=False
prevents user audio from interrupting agent speech but does not prevent session shutdown caused by a disconnect. The disconnect causes termination of active speech and event handling without emitting an error, which is why the transcript is silently lost. • The system only updates chat context for speech that finishes playout (see how speech completion is detected via 
SpeechHandle.wait_for_playout()
). Workarounds you can use now • Immediately persist the message to your chat context when you initiate speech, rather than waiting for the playout completion event. That guarantees the message exists for downstream logging and analytics even if the session terminates mid-speech. ◦ Example flow: after 
await self.session.say("...")
(or directly before playing), call your function that writes the assistant message into 
chat_ctx
(mark it as “pending” if you want to differentiate completed vs. interrupted). • Use 
SpeechHandle.add_done_callback()
to be notified when the speech either completes or is interrupted, and update your systems accordingly. Note: if the disconnect is abrupt, callbacks may not always run reliably, so this is supplementary rather than a full solution. • If you need to mark partial playout specifically, you can: ◦ Save the full intended text immediately, and when/if a playout-complete callback arrives, update that saved entry to set status = finished and include any timestamps/metadata. Questions to help me give a tighter implementation • Do you want the chat context entry to reflect the full original message text (even if it was only partially heard), or do you want to record the actual amount that was played? • Are you using any downstream consumers that depend on a playout timestamp or an explicit “finished” flag from 
conversation_item_added
? I can suggest a minimal schema for “pending → finished” entries so downstream logic can be robust. • Which SDK / language are you using for the agent (so I can show an idiomatic snippet)? If you want, I can provide: • A short code snippet that saves the assistant message into chat_ctx immediately and then patches it on speech completion/interrupt. • An example pattern using 
add_done_callback()
plus safe persistence for abrupt disconnects. Thanks for the clear repro steps — once you tell me how you prefer to represent interrupted vs finished messages (full text vs partial), I’ll give a concise code example you can drop into Agent.on_enter. Sources: How to detect when an agent has finished speaking | LiveKit
c
i want to store only for the actual amount that was played. I am using python. provide me the code ot an example
t
Thanks — that’s clear. Below is a Python example pattern you can adapt which: • Saves the assistant message text immediately as a “pending” conversation item. • Uses the 
SpeechHandle
callbacks and playout timing to determine how much of the message actually played. • Updates the saved conversation item with the actual played portion when speech finishes or is interrupted. • Handles abrupt disconnects as best-effort (writes pending state immediately so you won't lose the intent text). Notes: • Replace the persistence functions (
save_chat_item
, 
update_chat_item
) with your own database/analytics calls. • This example assumes the session API returns a 
SpeechHandle
with 
add_done_callback(callback)
and provides either 
played_ms
in the callback args or you can compute played duration from timestamps. If your SDK exposes 
wait_for_playout()
or a playout-complete event that includes played duration, use that instead. Example code (Python):
Copy code
import time
import threading

# --- Replace these with your real persistence functions ---
def save_chat_item(item):
    # Persist the item and return its id
    # Example item: {"role":"assistant","text":text,"status":"pending","played_text":"", "created_at":ts}
    item_id = ...  # insert into DB and return id
    return item_id

def update_chat_item(item_id, fields):
    # Update persisted item with fields dict
    pass
# --------------------------------------------------------

def get_played_text(full_text, played_ms, total_ms):
    if played_ms >= total_ms:
        return full_text
    # simple proportional truncation by characters
    ratio = played_ms / total_ms if total_ms > 0 else 0
    char_count = max(1, int(len(full_text) * ratio))
    return full_text[:char_count]

def play_and_persist(session, text, chat_ctx, sample_rate_ms=100):
    """
    session: your AgentSession
    text: full message the agent will speak
    chat_ctx: local in-memory context object (optional)
    """
    ts_start = int(time.time() * 1000)

    # 1) Persist immediately as a pending assistant message so you never lose the intent.
    item = {
        "role": "assistant",
        "text": text,
        "status": "pending",
        "played_text": "",
        "started_at": ts_start,
        "finished_at": None,
    }
    item_id = save_chat_item(item)

    # 2) Start speech and get SpeechHandle
    # This call should return a handle with add_done_callback; adjust per your SDK
    speech_handle = session.say(text, allow_interruptions=False)

    # Helper callback when speech completes/interrupts
    def on_speech_done(speech_result):
        """
        speech_result: the object provided by SDK callback.
        Ideally contains:
            - reason: 'finished' or 'interrupted' or 'cancelled'
            - played_ms: milliseconds of playout performed
            - total_ms: total milliseconds if available (or estimated)
        If your SDK doesn't provide these fields directly, you'll need to infer
        played_ms from timestamps or a playout API call.
        """
        try:
            played_ms = getattr(speech_result, "played_ms", None)
            total_ms = getattr(speech_result, "total_ms", None)

            # Fallback: if SDK doesn't provide durations, try to compute elapsed time
            if played_ms is None:
                played_ms = int(time.time() * 1000) - ts_start
            if total_ms is None:
                # Rough estimate: approximate spoken duration from text length.
                # Tune words-per-minute or char->ms factor for better accuracy.
                words = len(text.split())
                # assume ~150 wpm => 400 ms per word
                total_ms = words * 400

            played_text = get_played_text(text, played_ms, total_ms)

            update_fields = {
                "status": "finished" if played_ms >= total_ms else "interrupted",
                "played_text": played_text,
                "played_ms": played_ms,
                "total_ms": total_ms,
                "finished_at": int(time.time() * 1000),
            }
            update_chat_item(item_id, update_fields)

            # Optionally emit your own conversation_item_added event here,
            # so downstream systems behave the same regardless of SDK event firing.
            emit_conversation_item_added(item_id, update_fields)

        except Exception as e:
            # Best-effort update on error
            update_chat_item(item_id, {"status": "error", "error": str(e)})
            raise

    # 3) Attach done callback. If SDK gives callback args differently, adapt accordingly.
    try:
        speech_handle.add_done_callback(on_speech_done)
    except Exception:
        # If add_done_callback isn't available, spawn a watcher thread that waits
        # for the handle's playout or checks session state periodically.
        def watcher():
            try:
                # If SDK exposes wait_for_playout(), use it:
                if hasattr(speech_handle, "wait_for_playout"):
                    result = speech_handle.wait_for_playout()
                    on_speech_done(result)
                    return

                # Otherwise, poll for session/handle state until it's not playing.
                polling_interval = sample_rate_ms / 1000.0
                while True:
                    state = getattr(speech_handle, "state", None)
                    if state in ("done", "cancelled", "interrupted", None):
                        # Try to fetch played_ms if available
                        result = getattr(speech_handle, "result", None)
                        on_speech_done(result or type("R", (), {})())
                        break
                    time.sleep(polling_interval)
            except Exception as e:
                update_chat_item(item_id, {"status": "error", "error": str(e)})

        t = threading.Thread(target=watcher, daemon=True)
        t.start()

    # Return the persisted item id immediately if caller wants it
    return item_id

# Placeholder: emit an application-level conversation_item_added so downstream code can react.
def emit_conversation_item_added(item_id, fields):
    # Your application event emitter / analytics hook
    pass
Key implementation notes and tradeoffs • Immediate persistence ensures you won't lose the message entirely, but you must store both the full 
text
and the 
played_text
so you can show what actually played. • Getting an accurate 
played_ms
is ideal. If the speech system provides that in the callback, use it. If not, compute it from timestamps or estimate using words-per-minute/characters-per-second heuristics. • Mark pending items clearly so downstream consumers can handle “pending → interrupted/finished” transitions. • Emitting your own 
conversation_item_added
after you update the record ensures downstream logic that depends on that event runs even when the SDK's event doesn't fire. If you tell me: • what exact fields your SDK's speech callback provides (for example: 
played_ms
, 
reason
, 
total_ms
), I’ll tailor the snippet to use those fields directly and remove the estimation fallback. • how you persist chat items (database/schema), I can show a concrete example that writes to that store. Would you like a focused snippet that uses 
wait_for_playout()
if available, or one that relies only on 
add_done_callback()
?
c
you said that on partial text output if agent is interrupted then the chat is not saved. though the text_synchronizer syncs the partial text and stores it in forwareded text variable to be used further
t
Thanks — that helps clarify your point. You’re right that partial text can be synchronized/forwarded in real time, but that does not automatically mean it’s persisted into the chat context when the agent is interrupted or the participant disconnects. I’ll explain the difference and give a concise Python pattern you can use to capture the partial text from your synchronizer and persist only the actual played portion. Short summary • The real‑time synchronizer (what you called text_synchronizer) can forward partial agent output to an in‑memory or transient variable. That partial text is NOT automatically committed to the persistent chat history when speech is interrupted or the session shuts down. • Finalized speech that finishes playout is what the system treats as a conversation item and adds to chat context automatically. Interrupted/incomplete plays are discarded by default. • To preserve only the actually played text, explicitly capture the partial text you receive from the synchronizer (or from the speech handle callback) and persist the played portion when the speech finishes or is interrupted. Recommended pattern (Python) • Immediately persist a pending chat item when you start speaking. • Keep receiving partial updates from your text synchronizer and store the latest partial text in that pending item. • When the speech finishes or is interrupted (via SpeechHandle callback, wait_for_playout, or disconnect event), compute the actual played portion (preferably using a played_ms value from the SDK; otherwise derive from timing) and update the pending item to mark it finished with the played_text only. Concise example • Replace persistence functions with your DB/storage calls. • This example assumes your synchronizer emits partial strings via a callback you can subscribe to, and your speech handle gives a done callback or you can infer played time.
Copy code
import time
import threading

# Replace with your real DB/save functions
def save_chat_item(item): ...
def update_chat_item(item_id, fields): ...
def emit_conversation_item_added(item_id, fields): ...

def truncate_by_ratio(full_text, ratio):
    if ratio >= 1.0: return full_text
    # simple char-based truncation; tune for your language/needs
    n = max(1, int(len(full_text) * ratio))
    return full_text[:n]

def speak_and_capture(session, text, synchronizer, allow_interruptions=False):
    """
    session: AgentSession
    text: full assistant text to speak
    synchronizer: object exposing subscribe_partial(callback) and unsubscribe
    """

    ts_start = int(time.time() * 1000)
    # persist pending item immediately
    item = {
        "role": "assistant",
        "text": text,            # full intended text
        "status": "pending",
        "latest_partial": "",    # will be updated by synchronizer
        "played_text": "",
        "started_at": ts_start,
        "finished_at": None,
    }
    item_id = save_chat_item(item)

    # handler to receive partial updates from your text_synchronizer
    def on_partial(partial_text):
        try:
            # persist latest partial so you never lose it
            update_chat_item(item_id, {"latest_partial": partial_text, "updated_at": int(time.time()*1000)})
        except Exception:
            pass

    synchronizer.subscribe_partial(on_partial)

    # start speech and get handle
    speech_handle = session.say(text, allow_interruptions=allow_interruptions)

    # called when speech completes or is interrupted
    def on_done(result):
        # Try to get accurate played_ms / total_ms from result
        played_ms = getattr(result, "played_ms", None)
        total_ms = getattr(result, "total_ms", None)

        # fallback: compute played_ms by elapsed time
        if played_ms is None:
            played_ms = int(time.time() * 1000) - ts_start

        # fallback: estimate total_ms from word count if SDK doesn't give it
        if total_ms is None:
            words = len(text.split())
            total_ms = words * 400  # ~150 wpm ~= 400 ms/word

        ratio = min(1.0, played_ms / total_ms if total_ms > 0 else 0.0)

        # Prefer the synchronizer's last partial if it exists and matches ratio
        last_partial = get_latest_partial_from_store(item_id)  # implement helper to read saved latest_partial
        if last_partial:
            # If synchronizer already gave a partial equal/shorter than expected played_text, use it.
            # Otherwise, derive from full text by ratio.
            played_text = last_partial if len(last_partial) <= len(text) else truncate_by_ratio(text, ratio)
        else:
            played_text = truncate_by_ratio(text, ratio)

        final_fields = {
            "status": "finished" if ratio >= 1.0 else "interrupted",
            "played_text": played_text,
            "played_ms": played_ms,
            "total_ms": total_ms,
            "finished_at": int(time.time() * 1000),
        }
        try:
            update_chat_item(item_id, final_fields)
            # emit an application-level conversation_item_added so downstream logic runs
            emit_conversation_item_added(item_id, final_fields)
        finally:
            try:
                synchronizer.unsubscribe_partial(on_partial)
            except Exception:
                pass

    # attach done callback or watcher
    try:
        speech_handle.add_done_callback(on_done)
    except Exception:
        # fallback watcher using wait_for_playout or polling
        def watcher():
            try:
                if hasattr(speech_handle, "wait_for_playout"):
                    res = speech_handle.wait_for_playout()
                    on_done(res)
                else:
                    # poll for state change; adapt to your SDK's state field
                    while True:
                        state = getattr(speech_handle, "state", None)
                        if state in ("done", "cancelled", "interrupted", None):
                            res = getattr(speech_handle, "result", None)
                            on_done(res or type("R", (), {})())
                            break
                        time.sleep(0.1)
            except Exception as e:
                update_chat_item(item_id, {"status": "error", "error": str(e)})
        threading.Thread(target=watcher, daemon=True).start()

    return item_id
Notes and tradeoffs • Use the synchronizer’s partials as the authoritative source of what has been produced so far. Persisting the latest partial on every update ensures you don’t lose content on abrupt disconnects. • Accurate played_ms from the speech system is the best way to compute how much actually played. If you rely solely on synchronizer partials, ensure the synchronizer lags minimally behind playout. • Mark items as pending → interrupted/finished so downstream consumers know to wait for a finalization update. • Emit your own application event after you update the DB so analytics / logging run even when the system’s 
conversation_item_added
doesn’t fire. If you want, I can: • Provide a tighter snippet that uses the exact names for your synchronizer subscription methods if you paste how you currently subscribe to partial text. • Show a sample schema for the persisted chat item and example SQL/NoSQL writes. Which would you prefer? Sources: Transcriber | LiveKit Documentation | Text and transcriptions | LiveKit Documentation | How to capture real-time agent transcriptions during conversations
2 Views
Open in Slack
PreviousNext
Powered by