https://livekit.io logo
Join Slack
Powered by
I want to periodically send my agent an image to h...
# ask-ai
b
I want to periodically send my agent an image to help it understand the user's experience. But the way I'm doing it right now is interrupting the AI's speech. How do I send the image without causing that interruption? The Python code right now looks like this: data_uri = f"data:image/{image_format};base64,{image_b64}" chat_ctx = runningAgent.chat_ctx.copy() chat_ctx.add_message( role="user", content=[ text, ImageContent(image=data_uri), ], ) await runningAgent.update_chat_ctx(chat_ctx)
✅ 1
t
Thanks for reaching out, Mat. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.
Thanks — I understand you want to periodically send images to the agent without interrupting its ongoing speech. You're currently calling 
update_chat_ctx()
with an ImageContent mid-turn, which can trigger interruption behavior. Two safe approaches you can use: • Use a ByteStream / data topic to send the image (no immediate chat context update). a. Send the image over a dedicated topic (e.g., "user_images") or file/byte stream from the client. b. In the agent, receive and buffer/process the stream in a background task. c. Convert to a 
data:image/...;base64,...
URI and add it to chat context only when the agent is not currently speaking (or schedule it to be added after the current speech finishes). • Debounce / queue 
update_chat_ctx()
calls so updates occur only when safe. a. If 
runningAgent.session.current_speech
(or equivalent) is active, enqueue the image update instead of calling 
update_chat_ctx()
immediately. b. When the speech finishes (or on a periodic checkpoint), flush queued updates by calling 
update_chat_ctx()
once with the batched image messages. Example pattern (conceptual Python): 1. Send image via byte stream/topic from client. 2. Agent handler: ◦ Read image bytes asynchronously. ◦ Create 
data_uri = f"data:image/{fmt};base64,{b64}"
◦ If not speaking: call 
await runningAgent.update_chat_ctx(...)
◦ Else: add to a 
pending_image_updates
queue and schedule a task to flush when speech completes. Why this works: • Sending images over a separate data stream avoids changing the active turn state immediately. • Deferring or batching 
update_chat_ctx()
prevents mid-speech context changes that trigger interruption handling. If you want, share: • how you detect "currently speaking" in your runtime (field names like 
current_speech
), or • a short snippet of the agent-side handler you use to receive data — and I’ll give a concrete code example adapted to your setup. Sources: Adaptive interruption handling | LiveKit Documentation | Turns overview | LiveKit Documentation | Pipeline nodes and hooks | LiveKit Documentation | Images | LiveKit Documentation, with interactive elements
b
The Python method in which I send the image is already async. I'd like Python code which makes that method wait until the agent is not speaking. It should minimize CPU instead of polling in a tight loop.
t
Thanks — I’ll show a minimal, non-polling Python pattern that waits for the agent to finish speaking before sending the image. Choose one of these depending on where your code runs: 1) If you’re inside a tool / RunContext (recommended) • Use 
await ctx.wait_for_playout()
to yield until current speech finishes (no tight loop). Example:
Copy code
async def send_image_when_idle(ctx, image_b64, image_format, send_image_fn):
    # wait_for_playout returns when the agent finishes current audio playout
    await ctx.wait_for_playout()

    data_uri = f"data:image/{image_format};base64,{image_b64}"
    # send_image_fn should perform the actual byte-stream or chat-context update
    await send_image_fn(data_uri)
2) If you called 
session.say()
or 
session.generate_reply()
and have a SpeechHandle • Await the handle directly. Example:
Copy code
handle = session.say("I'll describe the app now.")
await handle  # yields until speech completes
data_uri = f"data:image/{image_format};base64,{image_b64}"
await send_image_fn(data_uri)
3) If you prefer event-driven callbacks • Listen for 
playback_finished
and schedule the image send when it fires. Example:
Copy code
@session.output.audio.on("playback_finished")
def on_playback_finished(ev):
    if not ev.interrupted:
        asyncio.create_task(send_image_fn(data_uri))
Notes: • Prefer sending images via a ByteStream/topic and only update chat context after speech ends. • Replace 
send_image_fn
with your async routine that sends the bytes or calls 
update_chat_ctx()
. If you share whether your code runs inside a tool (has 
ctx
) or uses 
session.say()
, I’ll adapt one of the examples into a drop-in snippet for your code. Sources: Agent speech and audio | LiveKit Documentation | Function tool definition | LiveKit Documentation | How to detect when an agent has finished speaking, with interactive elements
b
My code is not in a tool and does not have a SpeechHandle. I tried this, but the await never returned, even after my AI stopped speaking: current_speech = session.current_speech if current_speech: await current_speech
t
Thanks — I hear you: your code runs outside a tool, you don't have a SpeechHandle to await, and awaiting 
session.current_speech
hung. Below is a minimal, low-CPU pattern that works outside tool contexts by using the speech handle's callback and an asyncio.Event to wait non‑pollingly. • It checks 
session.current_speech
. • If present, it attaches a 
done
callback to set an 
asyncio.Event
. • It awaits the Event with no tight loop. • If no speech is active, it proceeds immediately. Example (drop-in):
Copy code
import asyncio

async def wait_until_agent_idle(session, timeout=None):
    """
    Wait until session.current_speech is finished.
    Returns True if finished (or none was active), False if timed out.
    """
    current = session.current_speech
    if current is None:
        return True

    ev = asyncio.Event()

    def _on_done(_):
        # mark completion from the callback
        ev.set()

    try:
        current.add_done_callback(_on_done)
    except Exception:
        # if handle doesn't support add_done_callback, fallback to short-poll
        while session.current_speech is not None:
            await asyncio.sleep(0.1)
        return True

    try:
        await asyncio.wait_for(ev.wait(), timeout=timeout)
        return True
    except asyncio.TimeoutError:
        return False

# Usage example: send image only after agent finished speaking
async def send_image_when_idle(session, image_b64, image_format, send_image_fn, timeout=30):
    ok = await wait_until_agent_idle(session, timeout=timeout)
    if not ok:
        # timed out; decide whether to send or skip
        return False

    data_uri = f"data:image/{image_format};base64,{image_b64}"
    await send_image_fn(data_uri)
    return True
Notes: • This avoids tight polling and is efficient. • If 
current.add_done_callback
is unsupported or throws, the code falls back to a mild sleep loop (0.1s) as a safe alternative. • Use 
timeout
to avoid waiting forever if needed. If you share the agent SDK version or what 
session.current_speech
type exposes (methods/attrs), I can tailor the callback handling to your exact API. Sources: Agent speech and audio | LiveKit Documentation, with interactive elements
2 Views
Open in Slack
PreviousNext
Powered by