Hey, I'm building a moderated group discussion system using LiveKit Agents where multiple AI agents take turns speaking (similar to a debate or panel discussion). I'm running into some challenging timing issues with agent handoffs and would love some guidance. What I'm trying to build: • Multiple

Agent

classes with distinct personalities, voices, and tools • A moderator agent (Ray) that manages turn-taking and invites speakers • Participant agents (Leo, Riya, etc.) that speak when invited and return control to the moderator • Clean handoffs where each agent finishes speaking before the next one takes over

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

class ModeratorRay(Agent):
    async def on_enter(self):
        self.session.generate_reply(
            instructions="Welcome participants and invite Leo to speak",
            allow_interruptions=False,
        )
    
    @function_tool
    async def invite_leo(self, ctx: RunContext):
        await ctx.wait_for_playout()
        return ParticipantLeo(chat_ctx=self.chat_ctx)

class ParticipantLeo(Agent):
    async def on_enter(self):
        self.session.generate_reply(
            instructions="Share your perspective, then return control to Ray",
            allow_interruptions=False,
        )
    
    @function_tool
    async def return_to_ray(self, ctx: RunContext):
        await ctx.wait_for_playout()
        return ModeratorRay(chat_ctx=self.chat_ctx)

Issues I'm encountering: *"RuntimeError: cannot schedule new speech, the speech scheduling is draining/pausing"*Happens in

on_enter()

when trying to call

generate_reply()

after an agent handoff 1. Seems like the new agent's

on_enter()

is being called before the previous agent is fully cleaned up *"RuntimeError: no activity context found, the agent is not running"*The previous agent's TTS/LLM tasks continue running after the handoff 1. Causes errors in

_tts_inference_task

and

_llm_inference_task

2. *Tools being called before speech completes*Even with

await ctx.wait_for_playout()

in the tool, sometimes the handoff happens too quickly 3. Results in agents rapidly bouncing between each other without actually speaking 4. *User input being ignored during transitions*Log shows: "skipping user input, speech scheduling is paused" Questions: Is this multi-agent turn-taking pattern supported in LiveKit Agents? Are there better architectural patterns I should be using? How should I properly synchronize speech completion with agent handoffs? Is

ctx.wait_for_playout()

the right approach? Should I be awaiting

wait_for_playout()

in

on_enter()

or only in the function tools? 1. Is there a way to ensure the previous agent is fully stopped before the new agent's

on_enter()

executes? 2. Are there any existing examples of multi-agent turn-based conversations I could reference?

Thanks — this is very helpful detail. I’m responding to your errors and the code you pasted (the RuntimeErrors in

on_enter()

, lingering LLM/TTS tasks after handoff, and tools firing before speech completes). Short answers up front: • Yes — multi-agent turn-taking is supported, but you must explicitly coordinate handoffs; LiveKit does not provide an automatic turn coordinator. • Do NOT call

ctx.wait_for_playout()

in the incoming agent’s

on_enter()

. Use

ctx.wait_for_playout()

in the outgoing agent’s tool that returns the new agent. • Avoid calling

generate_reply()

immediately in `on_enter()`; wait until the session has fully drained/started to prevent the “draining/pausing” scheduling error. Concrete fixes and patterns to apply 1) Handoff tool: wait for playout, then return the next agent • Keep the wait-for-playout inside the tool that performs the handoff. That ensures the outgoing agent finishes speaking before control transfers.

@function_tool()
async def return_to_ray(self, ctx: RunContext):
    await ctx.wait_for_playout()   # wait for outgoing speech to finish
    return ModeratorRay(chat_ctx=self.chat_ctx)

2) Delay scheduling speech in on_enter() • Don’t call

self.session.generate_reply()

immediately. Small async delay (or explicit readiness check) prevents scheduling during draining:

import asyncio

async def on_enter(self):
    await asyncio.sleep(0.4)   # give session time to finish draining
    self.session.generate_reply(
        instructions="Share your perspective, then return control to Ray",
        allow_interruptions=False,
    )

• You can tune the sleep; the goal is to avoid scheduling while the previous agent’s tasks are still shutting down. 3) Use the handoff tool as the single canonical place to end-turn • Make the outgoing speaker explicitly call the tool that ends their turn (and awaits playout) rather than implicitly returning from LLM flows. This avoids race conditions where tools or LLM tasks keep running after a handoff begins. 4) Prevent rapid bouncing with shared state • Track turn ownership in

session.userdata

(or similar). Only allow a handoff if the current speaker is allowed to yield. Example:

ctx.userdata.current_speaker = "leo"
# check before return:
if ctx.userdata.current_speaker == "leo":
    await ctx.wait_for_playout()
    ctx.userdata.current_speaker = "ray"
    return ModeratorRay(...)

• This prevents agents from immediately handing back and forth without real speech. 5) Ensure background tasks drain before scheduling new work • The session shutdown/drain is asynchronous. The incoming agent’s

on_enter()

may run before internal

_tts_inference_task

or

_llm_inference_task

of the previous agent have finished. The combined approach above (wait_for_playout in tool + slight on_enter delay + turn state) avoids scheduling while those tasks are still draining. 6) Don’t call wait_for_playout in on_enter() •

ctx.wait_for_playout()

belongs in the outgoing agent’s tool. Calling it in on_enter can be unreliable because that context is for the new agent, and the previous agent’s tasks may still be in flight. When to ask for more detail • If you want, paste minimal logs showing timestamps for: ◦ when previous agent’s last TTS chunk finished, ◦ when its `_tts_inference_task`/`_llm_inference_task` errors occur, ◦ when the new agent’s

on_enter()

starts. That lets me recommend an exact sleep length or whether you need stronger readiness signalling. References and next steps I can help with • I can: ◦ Show a short, complete example of a moderator-driven loop implementing these patterns. ◦ Help you design a simple state machine (userdata) for strict turn-taking and show code for enqueueing invites and acknowledgements. ◦ Tune exact timing settings (how long to delay

on_enter()

or whether to use manual turn detection). Which help would you like next — a full minimal example of the moderator + participant loop, or assistance adding userdata-based turn state to your current code? Sources: Agents and handoffs | LiveKit docs | Agent speech and audio | LiveKit docs | Job lifecycle | LiveKit docs | Turns overview | LiveKit docs

it is still jumping to function call rather than speaking it's turn on the second agent 142846.619 DEBUG livekit.agents tools execution completed {"speech_id": "speech_2a1c7de09382"} 142847.999 DEBUG livekit.agents executing tool {"function": "invite_leo", "arguments": "{}", "speech_id": "speech_2376b1ca2331"} 142848.061 DEBUG livekit.agents tools execution completed {"speech_id": "speech_2376b1ca2331"} 142849.334 DEBUG livekit.agents executing tool {"function": "return_to_ray", "arguments": "{}", "speech_id": "speech_2e8b11d25894"} 142849.409 DEBUG livekit.agents tools execution completed {"speech_id": "speech_2e8b11d25894"} 142850.899 DEBUG livekit.agents executing tool {"function": "invite_leo", "arguments": "{}", "speech_id": "speech_6aad08bf7792"} 142850.978 DEBUG livekit.agents tools execution completed {"speech_id": "speech_6aad08bf7792"} 142852.405 DEBUG livekit.agents executing tool {"function": "return_to_ray", "arguments": "{}", "speech_id": "speech_9050b47b2898"} 142852.474 DEBUG livekit.agents tools execution completed {"speech_id": "speech_9050b47b2898"} 142852.595 INFO livekit.agents shutting down worker {"id": "unregistered"} 142852.877 ERROR livekit.agents Error in _traceable_on_enter Traceback (most recent call last): File "/Users/lazrus/Documents/Python WorkSpace/Aurator/Agent-Test-Room/.venv/lib/python3.11/site-packages/livekit/agents/utils/log.py", line 16, in async_fn_logs return await fn(*args, **kwargs) ^^^^^^^^^^^^^^^^^^^^^^^^^ File "/Users/lazrus/Documents/Python WorkSpace/Aurator/Agent-Test-Room/.venv/lib/python3.11/site-packages/livekit/agents/voice/agent_activity.py", line 477, in _traceable_on_enter await self._agent.on_enter() File "/Users/lazrus/Documents/Python WorkSpace/Aurator/Agent-Test-Room/AgentOrchestration/agent.py", line 81, in on_enter self.session.generate_reply( File "/Users/lazrus/Documents/Python WorkSpace/Aurator/Agent-Test-Room/.venv/lib/python3.11/site-packages/livekit/agents/voice/agent_session.py", line 945, in generate_reply raise RuntimeError("AgentSession is closing, cannot use generate_reply()") RuntimeError: AgentSession is closing, cannot use generate_reply() 142852.885 ERROR livekit.agents Error in _update_activity_task Traceback (most recent call last): File "/Users/lazrus/Documents/Python WorkSpace/Aurator/Agent-Test-Room/.venv/lib/python3.11/site-packages/livekit/agents/utils/log.py", line 16, in async_fn_logs return await fn(*args, **kwargs) ^^^^^^^^^^^^^^^^^^^^^^^^^ File "/Users/lazrus/Documents/Python WorkSpace/Aurator/Agent-Test-Room/.venv/lib/python3.11/site-packages/livekit/agents/voice/agent_session.py", line 1094, in _update_activity_task await self._update_activity(agent) File "/Users/lazrus/Documents/Python WorkSpace/Aurator/Agent-Test-Room/.venv/lib/python3.11/site-packages/livekit/agents/voice/agent_session.py", line 1085, in _update_activity await asyncio.shield(self._activity._on_enter_task) File "/Users/lazrus/Documents/Python WorkSpace/Aurator/Agent-Test-Room/.venv/lib/python3.11/site-packages/livekit/agents/voice/agent_activity.py", line 411, in _context_aware_coro return await coro ^^^^^^^^^^ File "/Users/lazrus/Documents/Python WorkSpace/Aurator/Agent-Test-Room/.venv/lib/python3.11/site-packages/opentelemetry/util/_decorator.py", line 71, in async_wrapper return await func(*args, **kwargs) # type: ignore ^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/Users/lazrus/Documents/Python WorkSpace/Aurator/Agent-Test-Room/.venv/lib/python3.11/site-packages/livekit/agents/utils/log.py", line 16, in async_fn_logs return await fn(*args, **kwargs) ^^^^^^^^^^^^^^^^^^^^^^^^^ File "/Users/lazrus/Documents/Python WorkSpace/Aurator/Agent-Test-Room/.venv/lib/python3.11/site-packages/livekit/agents/voice/agent_activity.py", line 477, in _traceable_on_enter await self._agent.on_enter() File "/Users/lazrus/Documents/Python WorkSpace/Aurator/Agent-Test-Room/AgentOrchestration/agent.py", line 81, in on_enter self.session.generate_reply( File "/Users/lazrus/Documents/Python WorkSpace/Aurator/Agent-Test-Room/.venv/lib/python3.11/site-packages/livekit/agents/voice/agent_session.py", line 945, in generate_reply raise RuntimeError("AgentSession is closing, cannot use generate_reply()") RuntimeError: AgentSession is closing, cannot use generate_reply() 142854.893 DEBUG livekit.agents session closed {"reason": "user_initiated", "error": null} 142854.897 DEBUG livekit.agents shutting down job task {"reason": "", "user_initiated": false} 142854.900 DEBUG livekit.agents job exiting {"reason": "", "tid": 3103129, "job_id": "fake-job-363fc721dc11", "room_id": "FAKE_RM_4cc166d20c1e"} 142854.902 DEBUG livekit.agents http_session(): closing the httpclient ctx 142854.906 INFO livekit.agents process exiting {"reason": "", "pid": 89374, "inference": true} /Users/lazrus/.pyenv/versions/3.11.9/lib/python3.11/asyncio/base_events.py679 RuntimeWarning: coroutine 'AgentServer.aclose' was never awaited self._ready.clear()

Thanks — I see the logs and the error: the incoming agent’s

on_enter()

is running while the session is closing/draining, so

generate_reply()

raises "AgentSession is closing, cannot use generate_reply()". That explains why the second agent immediately goes to tool calls instead of speaking. Quick checklist and targeted fixes to stop the jump-to-tool behavior 1) Keep handoff tools as the single place to await playout and return the next agent • Outgoing agent tool must await playout and then return the next Agent. Example:

@function_tool()
async def return_to_ray(self, ctx: RunContext):
    await ctx.wait_for_playout()
    return ModeratorRay(chat_ctx=self.chat_ctx)

This ensures the outgoing TTS finishes before starting the next activity. 2) Prevent the new agent from scheduling speech while session is closing/draining • Don’t call

generate_reply()

immediately in

on_enter()

. Instead, poll for readiness or add a small retry loop that checks session state before scheduling:

import asyncio, time

async def on_enter(self):
    # wait until session is not closing; timeout/fail-safe to avoid infinite wait
    deadline = time.time() + 3.0
    while time.time() < deadline:
        if not self.session._is_closing:   # read-only check; adapt if API differs
            break
        await asyncio.sleep(0.05)
    await asyncio.sleep(0.2)  # extra buffer
    self.session.generate_reply(...)

If

_is_closing

is not accessible, use a short fixed delay (0.3–0.6s) plus a small retry on generate_reply exceptions. 3) Add guarded generate_reply with retry/backoff • If

generate_reply()

raises "closing" or "draining" errors, catch and retry a few times with small delays instead of failing the agent on first attempt. 4) Stop tools from auto-firing in

on_enter()

• From your logs tools fire repeatedly. Ensure

on_enter()

doesn’t call tools directly or cause immediate tool decisions. Let the agent produce a reply (TTS) first; only after successful playout should a tool be invoked to hand off. 5) Use userdata turn guard to avoid immediate re-invocation • Set

ctx.userdata.current_speaker

before returning the next agent and check it on enter to refuse immediate handback. That prevents rapid back-and-forth tool calls shown in the logs. 6) Increase visibility in logs to verify timing • Log timestamps when: ◦ outgoing agent schedules final TTS chunk, ◦

ctx.wait_for_playout()

returns, ◦ handoff tool returns the new agent, ◦ new agent

on_enter()

starts, ◦ new agent schedules TTS. Compare to ensure ordering is correct. If you want, I can: • Provide a minimal, end-to-end example that uses: (a) outgoing tool awaiting playout, (b) userdata turn guard, (c) guarded/delayed

on_enter()

with retry — and show expected log ordering. Which would help more: a short minimal example, or the guarded-retry snippet tuned to your current code?

can you implement these points and give me an example I can try

Thanks — below is a compact, runnable example that implements the fixes we discussed: (1) outgoing tool awaits playout before returning handoff, (2) incoming agent delays and guards

generate_reply()

so it doesn’t schedule while session is draining, (3) retry/backoff for

generate_reply()

, and (4) shared

userdata

turn state to prevent rapid bouncing. Copy into your project and adapt provider/LLM/TTS imports as needed.

import asyncio
import logging
import time
from dataclasses import dataclass

from livekit.agents import Agent, AgentSession, ChatContext, ChatMessage, RunContext, function_tool
from livekit.agents.llm import inference  # adapt to your LLM provider
from livekit.plugins import silero  # optional VAD example

logger = logging.getLogger("moderated_multi_agent")
logger.setLevel(logging.DEBUG)


@dataclass
class TurnState:
    current_speaker: str | None = None
    turn_active: bool = False
    last_handoff_id: str | None = None


class ModeratorRay(Agent):
    def __init__(self, name="ray"):
        super().__init__(instructions="You are Ray, the moderator.")
        self.name = name
        # session.userdata will be set by orchestrator

    async def on_enter(self):
        # wait a short time and check turn state before scheduling speech
        await _wait_until_session_ready(self.session)
        ts: TurnState = self.session.userdata
        if ts.turn_active and ts.current_speaker != self.name:
            logger.debug("Moderator skipping speak: another turn active")
            return

        ts.current_speaker = self.name
        ts.turn_active = True

        # retry-safe generate_reply
        await _generate_reply_retry(
            session=self.session,
            instructions="Welcome participants and invite Leo to speak.",
            allow_interruptions=False,
        )

    @function_tool()
    async def invite_leo(self, ctx: RunContext):
        # ensure moderator's speech is finished before handing off
        await ctx.wait_for_playout()
        # set turn state and return the next agent
        ts: TurnState = ctx.userdata
        ts.current_speaker = "leo"
        ts.last_handoff_id = str(time.time())
        return ParticipantLeo(chat_ctx=self.chat_ctx)


class ParticipantLeo(Agent):
    def __init__(self, name="leo"):
        super().__init__(instructions="You are Leo, share your perspective briefly.")
        self.name = name

    async def on_enter(self):
        await _wait_until_session_ready(self.session)
        ts: TurnState = self.session.userdata

        # Prevent immediate speaking if turn already active for another agent
        if ts.turn_active and ts.current_speaker != self.name:
            logger.debug("Leo skipping speak: another turn active")
            return

        # Register as active speaker
        ts.current_speaker = self.name
        ts.turn_active = True

        await _generate_reply_retry(
            session=self.session,
            instructions="Share your perspective, then return control to Ray.",
            allow_interruptions=False,
        )

    @function_tool()
    async def return_to_ray(self, ctx: RunContext):
        # ensure Leo's speech finished
        await ctx.wait_for_playout()
        ts: TurnState = ctx.userdata
        ts.current_speaker = "ray"
        ts.last_handoff_id = str(time.time())
        ts.turn_active = False
        return ModeratorRay(chat_ctx=self.chat_ctx)


# Helper utilities ---------------------------------------------------------

async def _wait_until_session_ready(session: AgentSession, timeout: float = 3.0):
    """
    Wait a short while for the session to stop draining/closing before scheduling speech.
    If session exposes an is_closing flag use it; otherwise rely on a short delay.
    """
    deadline = time.time() + timeout
    # Try a best-effort non-private attribute check; if not present, fallback to delay
    is_closing = getattr(session, "_is_closing", None)
    if is_closing is None:
        # no direct indicator -> conservative fixed delay
        await asyncio.sleep(0.35)
        return

    while time.time() < deadline:
        if not session._is_closing:
            # small buffer so internal tasks complete
            await asyncio.sleep(0.12)
            return
        await asyncio.sleep(0.05)
    # fallback buffer
    await asyncio.sleep(0.2)


async def _generate_reply_retry(session: AgentSession, instructions: str, allow_interruptions: bool, max_retries: int = 4):
    """
    Try generate_reply with small retries/backoff if session is draining/closing.
    """
    backoff = 0.1
    for attempt in range(max_retries):
        try:
            handle = session.generate_reply(instructions=instructions, allow_interruptions=allow_interruptions)
            # `generate_reply` may return an awaitable/handle; await it to ensure scheduling completes
            await handle
            return
        except RuntimeError as e:
            msg = str(e).lower()
            # Retry only for session closing/draining errors
            if "closing" in msg or "draining" in msg or "paused" in msg:
                logger.debug(f"generate_reply attempt {attempt+1} failed: {e}; retrying in {backoff}s")
                await asyncio.sleep(backoff)
                backoff *= 1.5
                continue
            raise
    # If we exhausted retries, raise the last observed error
    raise RuntimeError("generate_reply failed after retries")


# Orchestrator / Entrypoint ------------------------------------------------

async def entrypoint(ctx):
    """
    Replace ctx.room / provider setup with your environment.
    This entrypoint shows session creation and starting with ModeratorRay.
    """
    # optional VAD preloading
    try:
        vad = silero.VAD.load()
    except Exception:
        vad = None

    chat_ctx = ChatContext(messages=[ChatMessage(type="system", text="Multi-agent moderated discussion")])

    session = AgentSession(
        stt=inference.STT(),
        llm=inference.LLM(model="openai/gpt-4o-mini"),
        tts=inference.TTS(),
        vad=vad,
        chat_ctx=chat_ctx,
        # tune endpointing as desired
        turn_detection="vad",
        min_endpointing_delay=0.25,
        max_endpointing_delay=4.0,
    )

    # initialize shared userdata turn state
    session.userdata = TurnState()

    # start with moderator
    moderator = ModeratorRay()
    await session.start(agent=moderator, room=ctx.room)

    # keep job alive (depends on your host framework)
    await ctx.connect()


# If your environment needs a direct run, wire entrypoint into your job runner.

Notes and tuning suggestions • The tool that performs the handoff must

await ctx.wait_for_playout()

before returning the next agent. That guarantees outgoing TTS finishes. • The incoming agent uses

_wait_until_session_ready()

to avoid scheduling speech while AgentSession is still closing/draining; if your session exposes a non-private flag like

_is_closing

, the helper checks it — otherwise it uses a conservative delay. •

_generate_reply_retry()

catches transient RuntimeError messages about closing/draining/paused and retries with backoff; tune

max_retries

and backoff durations for your infra. •

session.userdata

holds

TurnState

to prevent immediate handbacks; adapt fields if you want stricter sequencing (e.g., require explicit ack tool from new agent before allowing others). • Add logging where helpful to verify the chronological order: final TTS chunk -> ctx.wait_for_playout returns -> handoff tool returns -> new agent on_enter -> new agent schedules TTS -> TTS playout begins. If you want, I can: • Produce a smaller minimal example focusing only on the moderator->single participant->moderator loop, or • Adjust the timing/retry values to match your logs if you paste timestamps showing when playout completes vs on_enter starts. Which would you prefer? Sources: examples/voice_agents | livekit/livekit_composite | DeepWiki | Workflows | LiveKit docs

from dotenv import load_dotenv import asyncio from livekit import agents, rtc from livekit.agents import ( Agent, AgentSession, ChatContext, RoomInputOptions, RunContext, function_tool, ) from livekit.agents import inference from livekit.plugins import noise_cancellation, silero from livekit.plugins.turn_detector.multilingual import MultilingualModel # Prompts from AgentOrchestration.prompts.ModeratorRay import RAY_INSTRUCTIONS from AgentOrchestration.prompts.ParticipantLeo import Leo_INSTRUCTIONS load_dotenv(".env.local") # ========================================================= # MODERATOR — RAY (WISE OLD OWL) # ========================================================= class ModeratorRay(Agent): def __init__(self, chat_ctx: ChatContext) -> None: super().__init__( instructions=RAY_INSTRUCTIONS, chat_ctx=chat_ctx, stt="deepgram/nova-3", llm=inference.LLM( model="openai/gpt-4.1-mini", provider="openai", extra_kwargs={ "temperature": 0.6, "max_tokens": 150, }, ), tts=inference.TTS( model="elevenlabs/eleven_turbo_v2_5", voice="iP95p4xoKVk53GoZ742B", language="en", ), vad=silero.VAD.load(), turn_detection=MultilingualModel(), ) async def on_enter(self) -> None: # FIX 1: Delay scheduling to allow session to finish draining await asyncio.sleep(0.4) try: store = self.session.userdata except ValueError: self.session.userdata = {} store = self.session.userdata # FIX 2: Track current speaker to prevent rapid bouncing store["current_speaker"] = "ray" already_greeted = bool(store.get("moderator_greeted")) if not already_greeted: store["moderator_greeted"] = True _# Single generate_reply: speak THEN call tool_ self.session.generate_reply( instructions=( "Welcome everyone. State the topic and rules briefly. " "Confirm you'll invite one speaker at a time. " "Introduce Leo and say you'll now invite him to speak. " "After you finish speaking, call the invite_leo tool. " "Do not say out loud that you're calling the function." ), allow_interruptions=False, ) else: _# Single generate_reply: speak THEN call tool_ self.session.generate_reply( instructions=( "Thank or acknowledge the previous speaker succinctly. " "Give a one-line recap if helpful. " "Then call the invite_leo tool to invite the next speaker." ), allow_interruptions=False, ) # ===== TOOLS (STRICT TURN-CONTROL HANDOFFS) ===== @function_tool async def invite_leo(self, ctx: RunContext): """ Invite Leo to speak next. This tool will be called by Ray after he finishes speaking. """ # FIX 3: Check turn ownership to prevent rapid bouncing store = self.session.userdata if store.get("current_speaker") != "ray": # Not Ray's turn to hand off return None # FIX 4: Wait for Ray's speech to complete before handoff await ctx.wait_for_playout() # FIX 5: Update turn ownership store["current_speaker"] = "leo" return ParticipantLeo(chat_ctx=self.chat_ctx) @function_tool async def invite_human(self, ctx: RunContext): """ Invite the human participant to speak next. Ray remains active and listens. """ store = self.session.userdata if store.get("current_speaker") != "ray": return None await ctx.wait_for_playout() store["current_speaker"] = "human" return None # ========================================================= # SPEAKER 1 — LEO (EXTROVERTED LION) # ========================================================= class ParticipantLeo(Agent): def __init__(self, chat_ctx: ChatContext) -> None: super().__init__( instructions=Leo_INSTRUCTIONS, chat_ctx=chat_ctx, stt="deepgram/nova-3", llm=inference.LLM( model="openai/gpt-4.1-mini", provider="openai", extra_kwargs={ "temperature": 0.7, "max_tokens": 120, }, ), tts=inference.TTS( model="cartesia/sonic-3", voice="9626c31c-bec5-4cca-baa8-f8ba9e84c8bc", language="en", ), vad=silero.VAD.load(), turn_detection=MultilingualModel(), ) async def on_enter(self): # FIX 1: Delay scheduling to allow session to finish draining await asyncio.sleep(0.4) # FIX 2: Track current speaker store = self.session.userdata store["current_speaker"] = "leo" _# Single generate_reply: speak THEN call tool_ self.session.generate_reply( instructions=( "Acknowledge Ray briefly, share your concise perspective on the topic. " "After you finish speaking, call the return_to_ray tool to hand control back. " "Do not say out loud that you're calling the function." ), allow_interruptions=False, ) @function_tool async def return_to_ray(self, ctx: RunContext): """ Hand control back to Ray. This tool will be called by Leo after he finishes speaking. """ # FIX 3: Check turn ownership to prevent rapid bouncing store = self.session.userdata if store.get("current_speaker") != "leo": # Not Leo's turn to hand off return None # FIX 4: Wait for Leo's speech to complete before handoff await ctx.wait_for_playout() # FIX 5: Update turn ownership store["current_speaker"] = "ray" return ModeratorRay(chat_ctx=self.chat_ctx) # ========================================================= # ENTRYPOINT # ========================================================= async def entrypoint(ctx: agents.JobContext): session = AgentSession( preemptive_generation=False, userdata={ "current_speaker": None, # FIX 6: Initialize turn state "moderator_greeted": False, }, min_interruption_duration=0.5, min_interruption_words=1, ) await session.start( agent=ModeratorRay(chat_ctx=ChatContext()), room=ctx.room, room_input_options=RoomInputOptions( noise_cancellation=lambda params: noise_cancellation.BVCTelephony() if params.participant.kind == rtc.ParticipantKind.PARTICIPANT_KIND_SIP else noise_cancellation.BVC(), ), ) if name == "__main__": agents.cli.run_app( agents.WorkerOptions( entrypoint_fnc=entrypoint, agent_name="gd-moderated-discussion-agent", ) ) this is my pipeline can you update this with all of these changes form your script?

Thanks — I updated your pipeline with the fixes discussed: outgoing tools await playout, incoming agents delay/guard

generate_reply()

, a retry/backoff wrapper around

generate_reply()

, and a shared

session.userdata

turn state to prevent rapid bouncing. I preserved your providers and structure; adapt any import names if your environment differs. Copy this into your project (replace existing ModeratorRay/ParticipantLeo/entrypoint). Keep your prompt imports at top.

from dotenv import load_dotenv
import asyncio
import time
import logging

from livekit import agents, rtc
from livekit.agents import (
    Agent,
    AgentSession,
    ChatContext,
    RoomInputOptions,
    RunContext,
    function_tool,
)
from livekit.agents import inference
from livekit.plugins import noise_cancellation, silero
from livekit.plugins.turn_detector.multilingual import MultilingualModel

# Prompts (keep your existing prompt imports)
from AgentOrchestration.prompts.ModeratorRay import RAY_INSTRUCTIONS
from AgentOrchestration.prompts.ParticipantLeo import Leo_INSTRUCTIONS

load_dotenv(".env.local")
logger = logging.getLogger("gd-moderated-discussion")
logger.setLevel(logging.DEBUG)

# -------------------------
# Shared TurnState helpers
# -------------------------
class TurnState(dict):
    """
    Simple dict-like turn state stored on session.userdata.
    Keys used: 'current_speaker', 'moderator_greeted', 'last_handoff_ts'
    """

def _ensure_userdata(session: AgentSession):
    if session.userdata is None:
        session.userdata = TurnState()
    elif not isinstance(session.userdata, TurnState):
        # preserve if user provided dict
        session.userdata = TurnState(session.userdata)

# -------------------------
# Retry + readiness helpers
# -------------------------
async def _wait_until_session_ready(session: AgentSession, timeout: float = 3.0):
    """
    Best-effort wait for session to finish draining/closing before scheduling speech.
    If session exposes `_is_closing`, prefer that; otherwise use conservative delay.
    """
    deadline = time.time() + timeout
    is_closing = getattr(session, "_is_closing", None)
    if is_closing is None:
        await asyncio.sleep(0.4)
        return
    while time.time() < deadline:
        if not session._is_closing:
            # small buffer for internal tasks to settle
            await asyncio.sleep(0.12)
            return
        await asyncio.sleep(0.05)
    await asyncio.sleep(0.2)


async def _generate_reply_retry(session: AgentSession, *args, max_retries: int = 4, **kwargs):
    """
    Call session.generate_reply with retries/backoff for transient "closing/draining" errors.
    Await the returned handle (if awaitable) so scheduling completes before returning.
    """
    backoff = 0.12
    last_exc = None
    for attempt in range(max_retries):
        try:
            handle = session.generate_reply(*args, **kwargs)
            # generate_reply may return an awaitable/handle on which we should wait
            if hasattr(handle, "__await__"):
                await handle
            return
        except RuntimeError as e:
            last_exc = e
            msg = str(e).lower()
            if any(token in msg for token in ("closing", "draining", "paused")):
                logger.debug(f"generate_reply transient error (attempt {attempt+1}): {e}; retrying in {backoff}s")
                await asyncio.sleep(backoff)
                backoff *= 1.5
                continue
            raise
    # exhausted retries
    raise last_exc or RuntimeError("generate_reply failed after retries")

# -------------------------
# Moderator — Ray
# -------------------------
class ModeratorRay(Agent):
    def __init__(self, chat_ctx: ChatContext) -> None:
        super().__init__(
            instructions=RAY_INSTRUCTIONS,
            chat_ctx=chat_ctx,
            stt="deepgram/nova-3",
            llm=inference.LLM(
                model="openai/gpt-4.1-mini",
                provider="openai",
                extra_kwargs={"temperature": 0.6, "max_tokens": 150},
            ),
            tts=inference.TTS(
                model="elevenlabs/eleven_turbo_v2_5",
                voice="iP95p4xoKVk53GoZ742B",
                language="en",
            ),
            vad=silero.VAD.load(),
            turn_detection=MultilingualModel(),
        )

    async def on_enter(self) -> None:
        _ensure_userdata(self.session)
        # FIX: small delay / readiness check so session isn't draining
        await _wait_until_session_ready(self.session)

        store: TurnState = self.session.userdata
        # Ensure current speaker set to ray while Ray is about to speak
        store["current_speaker"] = "ray"

        already_greeted = bool(store.get("moderator_greeted"))
        if not already_greeted:
            store["moderator_greeted"] = True
            instructions = (
                "Welcome everyone. State the topic and rules briefly. "
                "Confirm you'll invite one speaker at a time. "
                "Introduce Leo and say you'll now invite him to speak. "
                "After you finish speaking, call the invite_leo tool. "
                "Do not say out loud that you're calling the function."
            )
        else:
            instructions = (
                "Thank or acknowledge the previous speaker succinctly. "
                "Give a one-line recap if helpful. "
                "Then call the invite_leo tool to invite the next speaker."
            )

        await _generate_reply_retry(
            session=self.session,
            instructions=instructions,
            allow_interruptions=False,
        )

    @function_tool
    async def invite_leo(self, ctx: RunContext):
        """
        Invite Leo to speak next. Called by Ray after he finishes speaking.
        """
        # Ensure userdata present
        _ensure_userdata(self.session)
        store: TurnState = self.session.userdata
        # Guard: only Ray can hand off now
        if store.get("current_speaker") != "ray":
            logger.debug("invite_leo refused: not Ray's turn")
            return None

        # Wait for Ray's speech playout to complete before handing off
        await ctx.wait_for_playout()

        # Update turn ownership and timestamp
        store["current_speaker"] = "leo"
        store["last_handoff_ts"] = time.time()
        return ParticipantLeo(chat_ctx=self.chat_ctx)

    @function_tool
    async def invite_human(self, ctx: RunContext):
        _ensure_userdata(self.session)
        store: TurnState = self.session.userdata
        if store.get("current_speaker") != "ray":
            return None
        await ctx.wait_for_playout()
        store["current_speaker"] = "human"
        store["last_handoff_ts"] = time.time()
        return None

# -------------------------
# Participant — Leo
# -------------------------
class ParticipantLeo(Agent):
    def __init__(self, chat_ctx: ChatContext) -> None:
        super().__init__(
            instructions=Leo_INSTRUCTIONS,
            chat_ctx=chat_ctx,
            stt="deepgram/nova-3",
            llm=inference.LLM(
                model="openai/gpt-4.1-mini",
                provider="openai",
                extra_kwargs={"temperature": 0.7, "max_tokens": 120},
            ),
            tts=inference.TTS(
                model="cartesia/sonic-3",
                voice="9626c31c-bec5-4cca-baa8-f8ba9e84c8bc",
                language="en",
            ),
            vad=silero.VAD.load(),
            turn_detection=MultilingualModel(),
        )

    async def on_enter(self):
        _ensure_userdata(self.session)
        # Delay / readiness check to avoid scheduling during draining
        await _wait_until_session_ready(self.session)

        store: TurnState = self.session.userdata
        # If another turn is active and it's not Leo's, skip speaking
        if store.get("turn_active") and store.get("current_speaker") != "leo":
            logger.debug("Leo skipping speak: another turn active")
            return

        # Register Leo as active speaker
        store["current_speaker"] = "leo"
        store["turn_active"] = True

        await _generate_reply_retry(
            session=self.session,
            instructions=(
                "Acknowledge Ray briefly, share your concise perspective on the topic. "
                "After you finish speaking, call the return_to_ray tool to hand control back. "
                "Do not say out loud that you're calling the function."
            ),
            allow_interruptions=False,
        )

    @function_tool
    async def return_to_ray(self, ctx: RunContext):
        _ensure_userdata(self.session)
        store: TurnState = self.session.userdata
        if store.get("current_speaker") != "leo":
            logger.debug("return_to_ray refused: not Leo's turn")
            return None

        # Ensure Leo's speech finished
        await ctx.wait_for_playout()

        store["current_speaker"] = "ray"
        store["turn_active"] = False
        store["last_handoff_ts"] = time.time()
        return ModeratorRay(chat_ctx=self.chat_ctx)

# -------------------------
# Entrypoint
# -------------------------
async def entrypoint(ctx: agents.JobContext):
    # optional VAD preloading (silero)
    try:
        vad = silero.VAD.load()
    except Exception:
        vad = None

    # create chat context
    chat_ctx = ChatContext(messages=[{"type": "system", "text": "Multi-agent moderated discussion"}])

    session = AgentSession(
        preemptive_generation=False,
        userdata=TurnState({"current_speaker": None, "moderator_greeted": False}),
        min_interruption_duration=0.5,
        min_interruption_words=1,
        stt=inference.STT(),  # adapt if you set STT on agents individually
        llm=inference.LLM(model="openai/gpt-4o-mini"),
        tts=inference.TTS(),
        vad=vad,
        turn_detection="vad",
        min_endpointing_delay=0.25,
        max_endpointing_delay=4.0,
    )

    await session.start(
        agent=ModeratorRay(chat_ctx=chat_ctx),
        room=ctx.room,
        room_input_options=RoomInputOptions(
            noise_cancellation=lambda params: noise_cancellation.BVCTelephony()
            if params.participant.kind == rtc.ParticipantKind.PARTICIPANT_KIND_SIP
            else noise_cancellation.BVC(),
        ),
    )

    await ctx.connect()

# Run via agents CLI (unchanged)
if __name__ == "__main__":
    agents.cli.run_app(
        agents.WorkerOptions(entrypoint_fnc=entrypoint, agent_name="gd-moderated-discussion-agent")
    )

Notes and next steps • I left your per-agent STT/LLM/TTS settings intact. If you already set STT/LLM/ TTS on each Agent (as you do), you can remove them from session-level kwargs. • If your environment does not expose

session._is_closing

, the readiness helper falls back to fixed delays. • If you still see the behavior where tools fire immediately, paste a short log snippet with timestamps around a handoff and I’ll tune the delays/backoff values for your observed timing. • Run with DEBUG logging enabled to verify ordering: ◦ final TTS chunk -> ctx.wait_for_playout returns -> tool returns new agent -> new agent on_enter starts -> generate_reply schedules TTS. If you'd like, I can produce a minimal reduced example (no external providers) you can run locally for quicker iteration. Which would you prefer?

Traceback (most recent call last): File "/Users/lazrus/Documents/Python WorkSpace/Aurator/Agent-Test-Room/.venv/lib/python3.11/site-packages/opentelemetry/util/_decorator.py", line 71, in async_wrapper return await func(*args, **kwargs) # type: ignore ^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/Users/lazrus/Documents/Python WorkSpace/Aurator/Agent-Test-Room/.venv/lib/python3.11/site-packages/livekit/agents/ipc/job_proc_lazy_main.py", line 286, in _traceable_entrypoint await self._job_entrypoint_fnc(job_ctx) File "/Users/lazrus/Documents/Python WorkSpace/Aurator/Agent-Test-Room/AgentOrchestration/test-agent.py", line 254, in entrypoint chat_ctx = ChatContext(messages=[{"type": "system", "text": "Multi-agent moderated discussion"}]) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ TypeError: ChatContext.__init__() got an unexpected keyword argument 'messages'

Thanks — the traceback shows ChatContext was constructed incorrectly. Fix: create an empty ChatContext and add the system message with add_message(). Update your entrypoint like this:

# replace the incorrect line:
# chat_ctx = ChatContext(messages=[{"type": "system", "text": "Multi-agent moderated discussion"}])

# with:
chat_ctx = ChatContext()
chat_ctx.add_message(role="system", content="Multi-agent moderated discussion")

Make that change where you build the initial chat context before creating the AgentSession. After this the session.start call should accept the chat_ctx you pass into ModeratorRay. If you want, paste the current entrypoint function and I’ll return the full function updated with that exact change plus any small imports needed. Sources: Vision | LiveKit docs | Agents and handoffs | LiveKit docs | Pipeline nodes and hooks | LiveKit docs

152413.847 ERROR livekit.agents unhandled exception while running the job task Traceback (most recent call last): File "/Users/lazrus/Documents/Python WorkSpace/Aurator/Agent-Test-Room/.venv/lib/python3.11/site-packages/opentelemetry/util/_decorator.py", line 71, in async_wrapper return await func(*args, **kwargs) # type: ignore ^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/Users/lazrus/Documents/Python WorkSpace/Aurator/Agent-Test-Room/.venv/lib/python3.11/site-packages/livekit/agents/ipc/job_proc_lazy_main.py", line 286, in _traceable_entrypoint await self._job_entrypoint_fnc(job_ctx) File "/Users/lazrus/Documents/Python WorkSpace/Aurator/Agent-Test-Room/AgentOrchestration/test-agent.py", line 264, in entrypoint tts=inference.TTS(), ^^^^^^^^^^^^^^^ TypeError: TTS.__init__() missing 1 required positional argument: 'model'

Thanks — the traceback shows your TTS constructor call is missing the required

model

parameter. What to change • Replace

tts=inference.TTS()

with a TTS construction that includes a model string, for example: ◦

tts=inference.TTS(model="cartesia/sonic-3")

◦ or use your existing per-agent TTS settings (if you supply TTS per agent you can omit the session-level

tts

). Exact quick fix in your entrypoint (one-line)

# old
tts=inference.TTS(),

# new — pick the model you use for agents
tts=inference.TTS(model="cartesia/sonic-3"),

Notes • If you already set

tts

individually on each Agent (ModeratorRay / ParticipantLeo), you can remove the session-level

tts=

argument entirely. • If you want, paste the few lines around your AgentSession creation and I’ll return the corrected block with a recommended model value matched to your agent TTS choices.

Same Behaviour they are bouncing of the function calls ool {"function": "invite_leo", "arguments": "{}", "speech_id": "speech_5f8a10d3e65c"} 152615.747 DEBUG livekit.agents tools execution completed {"speech_id": "speech_5f8a10d3e65c"} 152616.804 DEBUG livekit.agents executing tool {"function": "return_to_ray", "arguments": "{}", "speech_id": "speech_086c6d5c883f"} 152616.882 DEBUG livekit.agents tools execution completed {"speech_id": "speech_086c6d5c883f"} 152618.011 DEBUG livekit.agents executing tool {"function": "invite_leo", "arguments": "{}", "speech_id": "speech_4ff6529ce023"} 152618.086 DEBUG livekit.agents tools execution completed {"speech_id": "speech_4ff6529ce023"} 152619.224 DEBUG livekit.agents executing tool {"function": "return_to_ray", "arguments": "{}", "speech_id": "speech_862855e734f4"} 152619.303 DEBUG livekit.agents tools execution completed {"speech_id": "speech_862855e734f4"} 152620.556 DEBUG livekit.agents executing tool {"function": "invite_leo", "arguments": "{}", "speech_id": "speech_0936ea90556f"} 152620.633 DEBUG livekit.agents tools execution completed {"speech_id": "speech_0936ea90556f"}

Thanks — I reviewed your logs: agents are still handing off immediately (tools execute) before any speech is scheduled/played. That causes the bounce loop. To stop it, try this minimal, targeted checklist and the small code changes below. Apply them in order and re-run with DEBUG logging; if it still bounces, paste the new log snippet and I’ll tune timings. Checklist (priority) • Ensure each agent actually schedules speech in on_enter() (call generate_reply) before any tool handoff is returned. • Make every handoff tool: 1) speak a short transition phrase (via generate_reply) if you need the agent to announce handoff, then 2) await playout with ctx.wait_for_playout(), then 3) return the new agent as a tuple: return NextAgent(...), "reason" • Prevent on_enter() from calling tools or returning immediately — on_enter should schedule speech (with a small readiness delay) and only tools should perform returns. • Use session.userdata turn guard so only the current speaker can hand off. • Wrap generate_reply in retry/backoff to handle transient "closing/draining" errors. Small concrete changes to your pipeline 1) Make sure on_enter actually awaits scheduling (use retry helper)

# in on_enter()
await _wait_until_session_ready(self.session)
# mark turn ownership, guard duplicates
store = self.session.userdata
if store.get("turn_active") and store.get("current_speaker") != "ray":
    return
store["current_speaker"] = "ray"
store["turn_active"] = True

# schedule speech and await the handle (use retry helper)
await _generate_reply_retry(
    session=self.session,
    instructions="Welcome... then invite Leo to speak. After you finish, call invite_leo.",
    allow_interruptions=False,
)
# do NOT call invite_leo() here

2) Make handoff tools speak (optional) then wait_for_playout then return agent

@function_tool
async def invite_leo(self, ctx: RunContext):
    store = ctx.userdata
    if store.get("current_speaker") != "ray":
        return None

    # optional: say a one-liner about handing off
    await _generate_reply_retry(session=ctx.session,
                                instructions="Inviting Leo now.",
                                allow_interruptions=False)

    await ctx.wait_for_playout()                # ensure audio finished
    store["current_speaker"] = "leo"
    store["turn_active"] = False
    return ParticipantLeo(chat_ctx=self.chat_ctx), "invite leo"

3) Participant on_enter should delay, check turn, then speak

async def on_enter(self):
    await _wait_until_session_ready(self.session)
    store = self.session.userdata
    if store.get("turn_active") and store.get("current_speaker") != "leo":
        return
    store["current_speaker"] = "leo"
    store["turn_active"] = True
    await _generate_reply_retry(session=self.session,
                                instructions="Share perspective, then call return_to_ray.",
                                allow_interruptions=False)

4) return_to_ray tool must wait_for_playout and return ModeratorRay

@function_tool
async def return_to_ray(self, ctx: RunContext):
    store = ctx.userdata
    if store.get("current_speaker") != "leo":
        return None
    await ctx.wait_for_playout()
    store["current_speaker"] = "ray"
    store["turn_active"] = False
    return ModeratorRay(chat_ctx=self.chat_ctx), "return to ray"

Helpful defaults to add • _wait_until_session_ready: small sleep 0.3–0.5s fallback if no session flag. • _generate_reply_retry: catch RuntimeError messages containing "closing", "draining", or "paused"; retry 3 times with backoff 0.1→0.3s. • Initialize session.userdata = {"current_speaker": None, "turn_active": False, "moderator_greeted": False}. What to log (add these to verify order) • When you call generate_reply (agent name + timestamp) • When generate_reply handle resolves / playout starts/ends • When ctx.wait_for_playout returns • When handoff tool returns If you want, I’ll paste a minimal runnable file that replaces only your ModeratorRay / ParticipantLeo / tools with these helpers so you can drop it into your project and test quickly. Which do you prefer: the single-file minimal example, or I should patch your existing file and return the full updated version? Sources: Agents and handoffs | LiveKit docs | Tool definition and use | LiveKit docs | Agent speech and audio | LiveKit docs | External data and RAG | LiveKit docs

first give me the single minimal example I can run quickly

Below is a minimal, runnable example that demonstrates a moderator (Ray) inviting a single participant (Leo), proper speech scheduling, guarded handoffs, and waiting for playout so agents don’t immediately bounce function calls. Adapt provider imports (LLM/TT S/STT) as needed for your environment. Save as quick_moderated.py and run inside your agent worker environment.

import asyncio
import time
import logging

from livekit import agents, rtc
from livekit.agents import Agent, AgentSession, ChatContext, RunContext, function_tool
from livekit.agents import inference

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger("quick-moderated")

# -------------------------
# Small helpers
# -------------------------
async def _wait_until_session_ready(session, timeout=2.0):
    # Best-effort wait for session to finish draining. Use small delay if no flag present.
    is_closing = getattr(session, "_is_closing", None)
    if is_closing is None:
        await asyncio.sleep(0.35)
        return
    deadline = time.time() + timeout
    while time.time() < deadline:
        if not session._is_closing:
            await asyncio.sleep(0.12)
            return
        await asyncio.sleep(0.05)
    await asyncio.sleep(0.2)


async def _generate_reply_retry(session, *args, max_retries=4, **kwargs):
    backoff = 0.12
    for attempt in range(max_retries):
        try:
            handle = session.generate_reply(*args, **kwargs)
            # Await handle if awaitable to ensure scheduling completes
            if hasattr(handle, "__await__"):
                await handle
            return
        except RuntimeError as e:
            msg = str(e).lower()
            if any(tok in msg for tok in ("closing", "draining", "paused")):
                logger.debug(f"generate_reply transient error: {e}; retry {attempt+1}")
                await asyncio.sleep(backoff)
                backoff *= 1.5
                continue
            raise
    raise RuntimeError("generate_reply failed after retries")


# -------------------------
# Agents
# -------------------------
class ModeratorRay(Agent):
    def __init__(self, chat_ctx=None):
        super().__init__(instructions="You are Ray, the moderator.", chat_ctx=chat_ctx)
    async def on_enter(self):
        # ensure session userdata exists
        if self.session.userdata is None:
            self.session.userdata = {"current_speaker": None, "turn_active": False}

        await _wait_until_session_ready(self.session)

        store = self.session.userdata
        # If another turn active, skip
        if store.get("turn_active") and store.get("current_speaker") != "ray":
            logger.debug("Ray skipping speak: another turn active")
            return

        store["current_speaker"] = "ray"
        store["turn_active"] = True

        await _generate_reply_retry(
            self.session,
            instructions=(
                "Welcome everyone. I'll invite Leo to speak next. "
                "After I finish, I'll call the invite_leo tool."
            ),
            allow_interruptions=False,
        )

    @function_tool
    async def invite_leo(self, ctx: RunContext):
        store = ctx.userdata
        if store.get("current_speaker") != "ray":
            logger.debug("invite_leo refused: not Ray's turn")
            return None

        # optionally confirm handoff with a short phrase (demonstrated in moderator's TTS)
        # ensure TTS finished
        await ctx.wait_for_playout()

        store["current_speaker"] = "leo"
        store["turn_active"] = False
        return ParticipantLeo(chat_ctx=self.chat_ctx), "invite leo"


class ParticipantLeo(Agent):
    def __init__(self, chat_ctx=None):
        super().__init__(instructions="You are Leo. Speak concisely.", chat_ctx=chat_ctx)

    async def on_enter(self):
        if self.session.userdata is None:
            self.session.userdata = {"current_speaker": None, "turn_active": False}

        await _wait_until_session_ready(self.session)

        store = self.session.userdata
        if store.get("turn_active") and store.get("current_speaker") != "leo":
            logger.debug("Leo skipping speak: another turn active")
            return

        store["current_speaker"] = "leo"
        store["turn_active"] = True

        await _generate_reply_retry(
            self.session,
            instructions=(
                "Thanks Ray. I will share a brief perspective now, then call return_to_ray."
            ),
            allow_interruptions=False,
        )

    @function_tool
    async def return_to_ray(self, ctx: RunContext):
        store = ctx.userdata
        if store.get("current_speaker") != "leo":
            logger.debug("return_to_ray refused: not Leo's turn")
            return None

        await ctx.wait_for_playout()

        store["current_speaker"] = "ray"
        store["turn_active"] = False
        return ModeratorRay(chat_ctx=self.chat_ctx), "return to ray"


# -------------------------
# Entrypoint (minimal)
# -------------------------
async def entrypoint(ctx: agents.JobContext):
    # minimal chat context
    chat_ctx = ChatContext()
    chat_ctx.add_message(role="system", content="Minimal moderated multi-agent demo")

    # NOTE: set providers appropriate to your environment. If you set per-agent providers
    # you can omit session-level stt/llm/tts here.
    session = AgentSession(
        stt=inference.STT(),           # replace or remove if using per-agent STT
        llm=inference.LLM(model="openai/gpt-4o-mini"),  # adapt
        tts=inference.TTS(model="cartesia/sonic-3"),    # adapt
        chat_ctx=chat_ctx,
        preemptive_generation=False,
        userdata={"current_speaker": None, "turn_active": False},
    )

    # start with moderator Ray
    await session.start(agent=ModeratorRay(chat_ctx=chat_ctx), room=ctx.room)
    await ctx.connect()


if __name__ == "__main__":
    agents.cli.run_app(agents.WorkerOptions(entrypoint_fnc=entrypoint, agent_name="quick-moderated"))

How to test quickly • Run in your usual agent worker environment. • Watch logs: you should see ModeratorRay schedule speech, wait_for_playout, then the invite_leo tool returns ParticipantLeo; ParticipantLeo should schedule speech and then call return_to_ray only after playout. • If bouncing persists, increase

_wait_until_session_ready

fallback delay from 0.35 to 0.5s and enable DEBUG logs to paste the relevant timestamped lines. If you want, paste the short log snippet after running this and I’ll suggest exact delay/backoff tuning.

call last): File "/Users/lazrus/Documents/Python WorkSpace/Aurator/Agent-Test-Room/.venv/lib/python3.11/site-packages/opentelemetry/util/_decorator.py", line 71, in async_wrapper return await func(*args, **kwargs) # type: ignore ^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/Users/lazrus/Documents/Python WorkSpace/Aurator/Agent-Test-Room/.venv/lib/python3.11/site-packages/livekit/agents/ipc/job_proc_lazy_main.py", line 286, in _traceable_entrypoint await self._job_entrypoint_fnc(job_ctx) File "/Users/lazrus/Documents/Python WorkSpace/Aurator/Agent-Test-Room/AgentOrchestration/test-agent.py", line 148, in entrypoint session = AgentSession( ^^^^^^^^^^^^^ TypeError: AgentSession.__init__() got an unexpected keyword argument 'chat_ctx'

Thanks — I’ll give you a small patch to fix the AgentSession error and stop the bouncing. You asked for a single minimal example earlier; below is a corrected, runnable minimal file that: • passes ChatContext to Agents (not AgentSession) • ensures agents speak in on_enter before any handoff tools run • handoff tools await playout and return the next Agent • includes readiness delay + retry around generate_reply to avoid scheduling during draining Save as quick_moderated_fixed.py and run in your agent worker environment.

import asyncio
import time
import logging

from livekit import agents
from livekit.agents import Agent, AgentSession, ChatContext, RunContext, function_tool
from livekit.agents import inference

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger("quick-moderated-fixed")

async def _wait_until_session_ready(session, timeout=2.0):
    is_closing = getattr(session, "_is_closing", None)
    if is_closing is None:
        await asyncio.sleep(0.35)
        return
    deadline = time.time() + timeout
    while time.time() < deadline:
        if not session._is_closing:
            await asyncio.sleep(0.12)
            return
        await asyncio.sleep(0.05)
    await asyncio.sleep(0.2)

async def _generate_reply_retry(session, *args, max_retries=4, **kwargs):
    backoff = 0.12
    for attempt in range(max_retries):
        try:
            handle = session.generate_reply(*args, **kwargs)
            if hasattr(handle, "__await__"):
                await handle
            return
        except RuntimeError as e:
            msg = str(e).lower()
            if any(tok in msg for tok in ("closing", "draining", "paused")):
                logger.debug(f"generate_reply transient error: {e}; retry {attempt+1}")
                await asyncio.sleep(backoff)
                backoff *= 1.5
                continue
            raise
    raise RuntimeError("generate_reply failed after retries")

class ModeratorRay(Agent):
    def __init__(self, chat_ctx=None):
        super().__init__(instructions="You are Ray, the moderator.", chat_ctx=chat_ctx)

    async def on_enter(self):
        if self.session.userdata is None:
            self.session.userdata = {"current_speaker": None, "turn_active": False}

        await _wait_until_session_ready(self.session)
        store = self.session.userdata
        if store.get("turn_active") and store.get("current_speaker") != "ray":
            logger.debug("Ray skipping speak: another turn active")
            return

        store["current_speaker"] = "ray"
        store["turn_active"] = True

        await _generate_reply_retry(
            self.session,
            instructions=(
                "Welcome everyone. I'll invite Leo to speak next. "
                "After I finish, I'll call the invite_leo tool."
            ),
            allow_interruptions=False,
        )

    @function_tool
    async def invite_leo(self, ctx: RunContext):
        store = ctx.userdata
        if store.get("current_speaker") != "ray":
            logger.debug("invite_leo refused: not Ray's turn")
            return None

        await ctx.wait_for_playout()
        store["current_speaker"] = "leo"
        store["turn_active"] = False
        return ParticipantLeo(chat_ctx=self.chat_ctx), "invite leo"

class ParticipantLeo(Agent):
    def __init__(self, chat_ctx=None):
        super().__init__(instructions="You are Leo. Speak concisely.", chat_ctx=chat_ctx)

    async def on_enter(self):
        if self.session.userdata is None:
            self.session.userdata = {"current_speaker": None, "turn_active": False}

        await _wait_until_session_ready(self.session)
        store = self.session.userdata
        if store.get("turn_active") and store.get("current_speaker") != "leo":
            logger.debug("Leo skipping speak: another turn active")
            return

        store["current_speaker"] = "leo"
        store["turn_active"] = True

        await _generate_reply_retry(
            self.session,
            instructions=(
                "Thanks Ray. I will share a brief perspective now, then call return_to_ray."
            ),
            allow_interruptions=False,
        )

    @function_tool
    async def return_to_ray(self, ctx: RunContext):
        store = ctx.userdata
        if store.get("current_speaker") != "leo":
            logger.debug("return_to_ray refused: not Leo's turn")
            return None

        await ctx.wait_for_playout()
        store["current_speaker"] = "ray"
        store["turn_active"] = False
        return ModeratorRay(chat_ctx=self.chat_ctx), "return to ray"

async def entrypoint(ctx: agents.JobContext):
    # correct: ChatContext is passed to agents, not AgentSession
    chat_ctx = ChatContext()
    chat_ctx.add_message(role="system", content="Minimal moderated multi-agent demo")

    session = AgentSession(
        stt=inference.STT(),                # adapt/remove if setting per-agent
        llm=inference.LLM(model="openai/gpt-4o-mini"),
        tts=inference.TTS(model="cartesia/sonic-3"),
        preemptive_generation=False,
        userdata={"current_speaker": None, "turn_active": False},
    )

    await session.start(agent=ModeratorRay(chat_ctx=chat_ctx), room=ctx.room)
    await ctx.connect()

if __name__ == "__main__":
    agents.cli.run_app(agents.WorkerOptions(entrypoint_fnc=entrypoint, agent_name="quick-moderated-fixed"))

Run it and watch logs. If bouncing still appears, paste the next ~20 lines of DEBUG logs around a handoff (showing timestamps) and I’ll tune the delay/backoff values. Sources: Agents and handoffs | LiveKit docs | Gemini Realtime Agent with Live Vision | LiveKit docs

leo does speak but he only says returning to ray logs: DEBUGlivekit.agentsusing transcript io:

AgentSession

-> (none) DEBUG livekit.agents using transcript io:

AgentSession

-> (none) MacBook Pro Microphone ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ <Ctrl+T> Text ModeDEBUGlivekit.agentsexecuting tool 153413.861 DEBUG livekit.agents executing tool {"function": "invite_leo", "arguments": "{}", "speech_id": "speech_c4c861970d9b"} MacBook Pro Microphone ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ <Ctrl+T> Text ModeDEBUGlivekit.agentstools execution completed 153413.884 DEBUG livekit.agents tools execution completed {"speech_id": "speech_c4c861970d9b"} MacBook Pro Microphone ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ <Ctrl+T> Text ModeDEBUGlivekit.agentsexecuting tool 153422.351 DEBUG livekit.agents executing tool {"function": "return_to_ray", "arguments": "{}", "speech_id": "speech_03ed042d4f5c"} MacBook Pro Microphone ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ <Ctrl+T> Text ModeDEBUGlivekit.agentstools execution completed 153422.374 DEBUG livekit.agents tools execution completed {"speech_id": "speech_03ed042d4f5c"} MacBook Pro Microphone ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ <Ctrl+T> Text ModeDEBUGlivekit.agentsexecuting tool 153426.068 DEBUG livekit.agents executing tool {"function": "invite_leo", "arguments": "{}", "speech_id": "speech_a1056f857f54"} MacBook Pro Microphone ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ <Ctrl+T> Text ModeDEBUGlivekit.agentstools execution completed 153426.086 DEBUG livekit.agents tools execution completed {"speech_id": "speech_a1056f857f54"} MacBook Pro Microphone ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ <Ctrl+T> Text ModeDEBUGlivekit.agentsexecuting tool 153434.133 DEBUG livekit.agents executing tool {"function": "return_to_ray", "arguments": "{}", "speech_id": "speech_92237f9d7f82"} MacBook Pro Microphone ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ <Ctrl+T> Text ModeDEBUGlivekit.agentstools execution completed 153434.150 DEBUG livekit.agents tools execution completed {"speech_id": "speech_92237f9d7f82"} MacBook Pro Microphone ▁ ▁ ▂ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ <Ctrl+T> Text ModeDEBUGlivekit.agentsexecuting tool 153437.914 DEBUG livekit.agents executing tool {"function": "invite_leo", "arguments": "{}", "speech_id": "speech_8784bd351c75"} MacBook Pro Microphone ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ <Ctrl+T> Text ModeDEBUGlivekit.agentstools execution completed 153437.934 DEBUG livekit.agents tools execution completed {"speech_id": "speech_8784bd351c75"} INFOlivekit.agentsshutting down worker 153441.233 INFO livekit.agents shutting down worker {"id": "unregistered"} DEBUGlivekit.agentssession closed 153443.239 DEBUG livekit.agents session closed {"reason": "user_initiated", "error": null} DEBUGlivekit.agentsshutting down job task 153443.242 DEBUG livekit.agents shutting down job task {"reason": "", "user_initiated": false} DEBUGlivekit.agentsjob exiting 153443.243 DEBUG livekit.agents job exiting {"reason": "", "tid": 3152704, "job_id": "fake-job-167f8e601be4", "room_id": "FAKE_RM_6123ac9ee2b4"} DEBUGlivekit.agentshttp_session(): closing the httpclient ctx 153443.245 DEBUG livekit.agents http_session(): closing the httpclient ctx /Users/lazrus/.pyenv/versions/3.11.9/lib/python3.11/asyncio/base_events.py679 RuntimeWarning: coroutine 'AgentServer.aclose' was never awaited self._ready.clear() RuntimeWarning: Enable tracemalloc to get the object allocation traceback

Thanks — I see why Leo only says “returning to Ray.” The LLM is invoking the handoff tool immediately instead of letting Leo generate and play his speech first. To fix this, make Leo speak in on_enter() and hand back control only after speech completes — either by returning the next Agent from the tool that performs the handoff, or by calling update_agent() after awaiting the generate_reply handle. Quick checklist and minimal changes (apply to your current code): 1) Remove or stop auto-calling return tools from on_enter() • on_enter should schedule speech only, not call handoff tools. 2) Let Leo speak in on_enter(), await the generated handle, then hand back Option A — agent-initiated handoff (recommended for simple flow)

async def on_enter(self):
    # readiness guard...
    handle = self.session.generate_reply(
        instructions="Thanks Ray. I will share a brief perspective now."
    )
    await handle        # ensures the reply is scheduled/completed
    # after speaking, hand back to Ray
    await self.session.update_agent(ModeratorRay(chat_ctx=self.chat_ctx))

Option B — tool-based handoff (if LLM chooses to call a tool)

@function_tool
async def return_to_ray(self, ctx: RunContext):
    # ensure Leo's speech finished
    await ctx.wait_for_playout()
    return ModeratorRay(chat_ctx=self.chat_ctx), "return to ray"

Important: If you use Option B, Leo must still speak first in on_enter(). The tool should only be responsible for waiting for playout and returning the new agent. 3) Ensure invite tool returns the new agent (moderator side)

@function_tool
async def invite_leo(self, ctx: RunContext):
    await ctx.wait_for_playout()
    return ParticipantLeo(chat_ctx=self.chat_ctx), "invite leo"

4) Add turn guards so only the current speaker can hand off • Use session.userdata like:

store = self.session.userdata
store["current_speaker"] = "leo"
# tools check store["current_speaker"] before proceeding

5) Add short readiness delays / retry on generate_reply to avoid scheduling during draining • Use small sleeps (0.2–0.5s) or a retry wrapper catching RuntimeError messages containing "closing", "draining", or "paused". What to inspect in logs if problems continue • Timestamp when generate_reply handle resolves (scheduling) • When ctx.wait_for_playout returns • When tool executes and returns the new agent If you paste a 20–30 line log snippet around one handoff (with timestamps), I’ll tune exact delays/backoff and give a tiny patch to your file. Would you like: • I) I patch your current file (you paste the current ModeratorRay/ParticipantLeo code), or • II) I produce a minimal runnable snippet that uses Option A (agent-initiated update_agent) so you can test a single agent roundtrip quickly? Which do you prefer? Sources: Tool definition and use | LiveKit docs | Agents and handoffs | LiveKit docs

to test it quickly,

Got it — here's a very small runnable example you can drop into your agent worker to test quickly. This uses agent-initiated handoff (Leo speaks in on_enter, awaits the reply handle, then calls

update_agent()

to return to Ray). That prevents the LLM from immediately invoking handoff tools. Save as quick_roundtrip.py and run in your LiveKit agent environment.

import asyncio
import logging
import time

from livekit import agents
from livekit.agents import Agent, AgentSession, ChatContext
from livekit.agents import RunContext, function_tool
from livekit.agents import inference

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger("quick-roundtrip")

class ModeratorRay(Agent):
    def __init__(self, chat_ctx=None):
        super().__init__(instructions="You are Ray, the moderator.", chat_ctx=chat_ctx)

    async def on_enter(self):
        if self.session.userdata is None:
            self.session.userdata = {"current_speaker": None, "turn_active": False}

        # small readiness delay to avoid scheduling during draining
        await asyncio.sleep(0.3)

        store = self.session.userdata
        store["current_speaker"] = "ray"
        store["turn_active"] = True

        # Ray speaks and *does not* immediately call a handoff tool here
        handle = self.session.generate_reply(
            instructions="Welcome everyone. I will now invite Leo to speak."
        )
        await handle  # ensure scheduling completes and LLM/TTS started

        # After Ray's speech is scheduled, we hand off by returning Leo from a tool call.
        # Simulate the moderator calling a tool (LLM may call this tool).
        # For quick testing we directly return Leo via update_agent below, or you can implement a tool.
        # We'll wait a moment then programmatically hand off:
        await asyncio.sleep(0.2)
        await self.session.update_agent(ParticipantLeo(chat_ctx=self.chat_ctx))


class ParticipantLeo(Agent):
    def __init__(self, chat_ctx=None):
        super().__init__(instructions="You are Leo. Give a concise opinion.", chat_ctx=chat_ctx)

    async def on_enter(self):
        if self.session.userdata is None:
            self.session.userdata = {"current_speaker": None, "turn_active": False}

        # small readiness delay
        await asyncio.sleep(0.35)

        store = self.session.userdata
        store["current_speaker"] = "leo"
        store["turn_active"] = True

        # Leo generates his full reply and we await the handle so speech scheduling completes
        handle = self.session.generate_reply(
            instructions="Thanks Ray. I will share a brief perspective now."
        )
        await handle

        # Ensure playout finishes before handing back (best-effort)
        # If running in a setup with ctx.wait_for_playout available inside tools,
        # that is preferable; here we add a short sleep as a simple test.
        await asyncio.sleep(0.5)

        # Hand back to moderator after speaking
        await self.session.update_agent(ModeratorRay(chat_ctx=self.chat_ctx))


async def entrypoint(ctx: agents.JobContext):
    chat_ctx = ChatContext()
    chat_ctx.add_message(role="system", content="Quick roundtrip demo")

    session = AgentSession(
        stt=inference.STT(),
        llm=inference.LLM(model="openai/gpt-4o-mini"),
        tts=inference.TTS(model="cartesia/sonic-3"),
        preemptive_generation=False,
        userdata={"current_speaker": None, "turn_active": False},
    )

    await session.start(agent=ModeratorRay(chat_ctx=chat_ctx), room=ctx.room)
    await ctx.connect()

if __name__ == "__main__":
    agents.cli.run_app(agents.WorkerOptions(entrypoint_fnc=entrypoint, agent_name="quick-roundtrip"))

How to verify • Run and watch DEBUG logs. • Expect sequence: Ray schedules speech → session.update_agent to Leo → Leo schedules speech (full reply) → Leo calls update_agent to Ray. • If Leo still only says the short handoff message, increase the readiness delays: Ray sleep 0.3→0.5s, Leo sleep 0.35→0.6s, and post-reply sleep 0.5→1.0s. If you want, I can convert the programmatic handoff to an explicit tool-based handoff (tool returns new Agent) — tell me which you prefer.

^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/Users/lazrus/Documents/Python WorkSpace/Aurator/Agent-Test-Room/.venv/lib/python3.11/site-packages/livekit/agents/utils/log.py", line 16, in async_fn_logs return await fn(*args, **kwargs) ^^^^^^^^^^^^^^^^^^^^^^^^^ File "/Users/lazrus/Documents/Python WorkSpace/Aurator/Agent-Test-Room/.venv/lib/python3.11/site-packages/livekit/agents/voice/agent_activity.py", line 477, in _traceable_on_enter await self._agent.on_enter() File "/Users/lazrus/Documents/Python WorkSpace/Aurator/Agent-Test-Room/AgentOrchestration/test-agent.py", line 69, in on_enter await self.session.update_agent(ModeratorRay(chat_ctx=self.chat_ctx)) TypeError: object NoneType can't be used in 'await' expression DEBUGlivekit.agentssession closed 153855.148 DEBUG livekit.agents session closed {"reason": "user_initiated", "error": null} DEBUGlivekit.agentsshutting down job task 153855.152 DEBUG livekit.agents shutting down job task {"reason": "", "user_initiated": false} DEBUGlivekit.agentsjob exiting 153855.154 DEBUG livekit.agents job exiting {"reason": "", "tid": 3156758, "job_id": "fake-job-41c7997ee800", "room_id": "FAKE_RM_73631273b746"} DEBUGlivekit.agentshttp_session(): closing the httpclient ctx 153855.156 DEBUG livekit.agents http_session(): closing the httpclient ctx /Users/lazrus/.pyenv/versions/3.11.9/lib/python3.11/asyncio/base_events.py679 RuntimeWarning: coroutine 'AgentServer.aclose' was never awaited self._ready.clear() RuntimeWarning: Enable tracemalloc to get the object allocation traceback this time the TTS started with leo directly

Thanks — the error shows you awaited a function that returned None.

session.update_agent(...)

is not awaitable (it returns None), so

await self.session.update_agent(...)

raises TypeError. That’s why the session closed right after Leo’s TTS started. What to change (very short) • Remove

await

when calling `update_agent()`: ◦ Wrong:

await self.session.update_agent(ModeratorRay(...))

◦ Right:

self.session.update_agent(ModeratorRay(...))

• Or use a handoff tool that RETURNS the Agent instance (tool return is awaited by the framework), not `update_agent()`: ◦ Tool pattern:

@function_tool
    async def return_to_ray(self, ctx: RunContext):
        await ctx.wait_for_playout()
        return ModeratorRay(chat_ctx=self.chat_ctx), "return to ray"

• Then the LLM calling that tool triggers the handoff. Minimal recommended pattern (agent-initiated handoff — quick & robust) • Let Leo speak in on_enter and await the generate_reply handle. • After speech/playout, call

self.session.update_agent(ModeratorRay(...))

(no await). Example (fix inside Leo.on_enter):

handle = self.session.generate_reply(instructions="Thanks Ray. I will share a brief perspective now.")
await handle              # ensure scheduling completes
# optionally wait for playout if available in your environment
await asyncio.sleep(0.5)  # simple fallback
self.session.update_agent(ModeratorRay(chat_ctx=self.chat_ctx))  # NO await

If you prefer tool-based handoff (LLM decides), implement the tool to wait_for_playout and return the Agent as shown above. Next step • Apply the no-await fix for

update_agent()

in your Leo code and re-run. If you still see quick handoffs or Leo only says the handoff phrase, paste the next 20–30 log lines and I’ll propose exact timing/backoff values.

now try and implement all what we have found to work in me script to do the final test from dotenv import load_dotenv import asyncio from livekit import agents, rtc from livekit.agents import ( Agent, AgentSession, ChatContext, RoomInputOptions, RunContext, function_tool, ) from livekit.agents import inference from livekit.plugins import noise_cancellation, silero from livekit.plugins.turn_detector.multilingual import MultilingualModel # Prompts from AgentOrchestration.prompts.ModeratorRay import RAY_INSTRUCTIONS from AgentOrchestration.prompts.ParticipantLeo import Leo_INSTRUCTIONS load_dotenv(".env.local") # ========================================================= # MODERATOR — RAY (WISE OLD OWL) # ========================================================= class ModeratorRay(Agent): def __init__(self, chat_ctx: ChatContext) -> None: super().__init__( instructions=RAY_INSTRUCTIONS, chat_ctx=chat_ctx, stt="deepgram/nova-3", llm=inference.LLM( model="openai/gpt-4.1-mini", provider="openai", extra_kwargs={ "temperature": 0.6, "max_tokens": 150, }, ), tts=inference.TTS( model="elevenlabs/eleven_turbo_v2_5", voice="iP95p4xoKVk53GoZ742B", language="en", ), vad=silero.VAD.load(), turn_detection=MultilingualModel(), ) async def on_enter(self) -> None: # FIX 1: Delay scheduling to allow session to finish draining await asyncio.sleep(0.4) try: store = self.session.userdata except ValueError: self.session.userdata = {} store = self.session.userdata # FIX 2: Track current speaker to prevent rapid bouncing store["current_speaker"] = "ray" already_greeted = bool(store.get("moderator_greeted")) if not already_greeted: store["moderator_greeted"] = True _# Single generate_reply: speak THEN call tool_ self.session.generate_reply( instructions=( "Welcome everyone. State the topic and rules briefly. " "Confirm you'll invite one speaker at a time. " "Introduce Leo and say you'll now invite him to speak. " "After you finish speaking, call the invite_leo tool. " "Do not say out loud that you're calling the function." ), allow_interruptions=False, ) else: _# Single generate_reply: speak THEN call tool_ self.session.generate_reply( instructions=( "Thank or acknowledge the previous speaker succinctly. " "Give a one-line recap if helpful. " "Then call the invite_leo tool to invite the next speaker." ), allow_interruptions=False, ) # ===== TOOLS (STRICT TURN-CONTROL HANDOFFS) ===== @function_tool async def invite_leo(self, ctx: RunContext): """ Invite Leo to speak next. This tool will be called by Ray after he finishes speaking. """ # FIX 3: Check turn ownership to prevent rapid bouncing store = self.session.userdata if store.get("current_speaker") != "ray": # Not Ray's turn to hand off return None # FIX 4: Wait for Ray's speech to complete before handoff await ctx.wait_for_playout() # FIX 5: Update turn ownership store["current_speaker"] = "leo" return ParticipantLeo(chat_ctx=self.chat_ctx) @function_tool async def invite_human(self, ctx: RunContext): """ Invite the human participant to speak next. Ray remains active and listens. """ store = self.session.userdata if store.get("current_speaker") != "ray": return None await ctx.wait_for_playout() store["current_speaker"] = "human" return None # ========================================================= # SPEAKER 1 — LEO (EXTROVERTED LION) # ========================================================= class ParticipantLeo(Agent): def __init__(self, chat_ctx: ChatContext) -> None: super().__init__( instructions=Leo_INSTRUCTIONS, chat_ctx=chat_ctx, stt="deepgram/nova-3", llm=inference.LLM( model="openai/gpt-4.1-mini", provider="openai", extra_kwargs={ "temperature": 0.7, "max_tokens": 120, }, ), tts=inference.TTS( model="cartesia/sonic-3", voice="9626c31c-bec5-4cca-baa8-f8ba9e84c8bc", language="en", ), vad=silero.VAD.load(), turn_detection=MultilingualModel(), ) async def on_enter(self): # FIX 1: Delay scheduling to allow session to finish draining await asyncio.sleep(0.4) # FIX 2: Track current speaker store = self.session.userdata store["current_speaker"] = "leo" _# Single generate_reply: speak THEN call tool_ self.session.generate_reply( instructions=( "Acknowledge Ray briefly, share your concise perspective on the topic. " "After you finish speaking, call the return_to_ray tool to hand control back. " "Do not say out loud that you're calling the function." ), allow_interruptions=False, ) @function_tool async def return_to_ray(self, ctx: RunContext): """ Hand control back to Ray. This tool will be called by Leo after he finishes speaking. """ # FIX 3: Check turn ownership to prevent rapid bouncing store = self.session.userdata if store.get("current_speaker") != "leo": # Not Leo's turn to hand off return None # FIX 4: Wait for Leo's speech to complete before handoff await ctx.wait_for_playout() # FIX 5: Update turn ownership store["current_speaker"] = "ray" return ModeratorRay(chat_ctx=self.chat_ctx) # ========================================================= # ENTRYPOINT # ========================================================= async def entrypoint(ctx: agents.JobContext): session = AgentSession( preemptive_generation=False, userdata={ "current_speaker": None, # FIX 6: Initialize turn state "moderator_greeted": False, }, min_interruption_duration=0.5, min_interruption_words=1, ) await session.start( agent=ModeratorRay(chat_ctx=ChatContext()), room=ctx.room, room_input_options=RoomInputOptions( noise_cancellation=lambda params: noise_cancellation.BVCTelephony() if params.participant.kind == rtc.ParticipantKind.PARTICIPANT_KIND_SIP else noise_cancellation.BVC(), ), ) if name == "__main__": agents.cli.run_app( agents.WorkerOptions( entrypoint_fnc=entrypoint, agent_name="gd-moderated-discussion-agent", ) )

Thanks — I’ve integrated everything we tested into your script: readiness delay and retry for

generate_reply()

,

session.userdata

turn guards, tools that await playout and return agent instances, no

await

on

update_agent()

, and correct

ChatContext

handling. Replace your file with the code below and run the job. Notes: • Keep your prompt imports at top (I preserved RAY_INSTRUCTIONS and Leo_INSTRUCTIONS). • Adjust model strings / provider args if your environment differs. • Run with DEBUG logs and paste the 20–30 lines around any failing handoff if it still misbehaves. Final integrated script:

from dotenv import load_dotenv
import asyncio
import time
import logging

from livekit import agents, rtc
from livekit.agents import (
    Agent,
    AgentSession,
    ChatContext,
    RoomInputOptions,
    RunContext,
    function_tool,
)
from livekit.agents import inference
from livekit.plugins import noise_cancellation, silero
from livekit.plugins.turn_detector.multilingual import MultilingualModel

# Prompts (keep these as you had them)
from AgentOrchestration.prompts.ModeratorRay import RAY_INSTRUCTIONS
from AgentOrchestration.prompts.ParticipantLeo import Leo_INSTRUCTIONS

load_dotenv(".env.local")
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger("gd-moderated-discussion")

# -------------------------
# Helpers: readiness + retry
# -------------------------
async def _wait_until_session_ready(session, timeout: float = 3.0):
    """Best-effort wait for session to stop draining/closing before scheduling speech."""
    is_closing = getattr(session, "_is_closing", None)
    if is_closing is None:
        await asyncio.sleep(0.35)
        return
    deadline = time.time() + timeout
    while time.time() < deadline:
        if not session._is_closing:
            await asyncio.sleep(0.12)
            return
        await asyncio.sleep(0.05)
    await asyncio.sleep(0.2)

async def _generate_reply_retry(session, *args, max_retries: int = 4, **kwargs):
    """
    Call session.generate_reply with small retries/backoff for transient session errors.
    Await the returned handle so scheduling completes.
    """
    backoff = 0.12
    last_exc = None
    for attempt in range(max_retries):
        try:
            handle = session.generate_reply(*args, **kwargs)
            if hasattr(handle, "__await__"):
                await handle
            return
        except RuntimeError as e:
            last_exc = e
            msg = str(e).lower()
            if any(tok in msg for tok in ("closing", "draining", "paused")):
                logger.debug(f"generate_reply transient error (attempt {attempt+1}): {e}; retrying in {backoff}s")
                await asyncio.sleep(backoff)
                backoff *= 1.5
                continue
            raise
    raise last_exc or RuntimeError("generate_reply failed after retries")

def _ensure_userdata(session):
    if session.userdata is None:
        session.userdata = {"current_speaker": None, "turn_active": False, "moderator_greeted": False}
    else:
        # ensure keys exist
        session.userdata.setdefault("current_speaker", None)
        session.userdata.setdefault("turn_active", False)
        session.userdata.setdefault("moderator_greeted", False)

# -------------------------
# Moderator — Ray
# -------------------------
class ModeratorRay(Agent):
    def __init__(self, chat_ctx: ChatContext) -> None:
        super().__init__(
            instructions=RAY_INSTRUCTIONS,
            chat_ctx=chat_ctx,
            stt="deepgram/nova-3",
            llm=inference.LLM(
                model="openai/gpt-4.1-mini",
                provider="openai",
                extra_kwargs={"temperature": 0.6, "max_tokens": 150},
            ),
            tts=inference.TTS(
                model="elevenlabs/eleven_turbo_v2_5",
                voice="iP95p4xoKVk53GoZ742B",
                language="en",
            ),
            vad=silero.VAD.load(),
            turn_detection=MultilingualModel(),
        )

    async def on_enter(self) -> None:
        _ensure_userdata(self.session)
        # FIX 1: Delay scheduling to allow session to finish draining
        await _wait_until_session_ready(self.session)

        store = self.session.userdata
        # FIX 2: Track current speaker to prevent rapid bouncing
        store["current_speaker"] = "ray"

        already_greeted = bool(store.get("moderator_greeted", False))
        if not already_greeted:
            store["moderator_greeted"] = True
            instructions = (
                "Welcome everyone. State the topic and rules briefly. "
                "Confirm you'll invite one speaker at a time. "
                "Introduce Leo and say you'll now invite him to speak. "
                "After you finish speaking, call the invite_leo tool. "
                "Do not say out loud that you're calling the function."
            )
        else:
            instructions = (
                "Thank or acknowledge the previous speaker succinctly. "
                "Give a one-line recap if helpful. "
                "Then call the invite_leo tool to invite the next speaker."
            )

        # Use retry wrapper so transient session states don't fail scheduling
        await _generate_reply_retry(
            session=self.session,
            instructions=instructions,
            allow_interruptions=False,
        )
        # Do not call the tool here — the LLM may call invite_leo; tool will await playout and return agent.

    @function_tool
    async def invite_leo(self, ctx: RunContext):
        """
        Invite Leo to speak next. This tool will be called by Ray after he finishes speaking.
        """
        _ensure_userdata(self.session)
        store = self.session.userdata
        # FIX 3: Check turn ownership to prevent rapid bouncing
        if store.get("current_speaker") != "ray":
            logger.debug("invite_leo refused: not Ray's turn")
            return None

        # FIX 4: Wait for Ray's speech to complete before handoff
        await ctx.wait_for_playout()

        # FIX 5: Update turn ownership
        store["current_speaker"] = "leo"
        store["turn_active"] = False
        store["last_handoff_ts"] = time.time()

        # Return the next agent instance (tool return triggers the handoff)
        return ParticipantLeo(chat_ctx=self.chat_ctx), "invite leo"

    @function_tool
    async def invite_human(self, ctx: RunContext):
        _ensure_userdata(self.session)
        store = self.session.userdata
        if store.get("current_speaker") != "ray":
            return None
        await ctx.wait_for_playout()
        store["current_speaker"] = "human"
        store["turn_active"] = False
        store["last_handoff_ts"] = time.time()
        return None

# -------------------------
# Participant — Leo
# -------------------------
class ParticipantLeo(Agent):
    def __init__(self, chat_ctx: ChatContext) -> None:
        super().__init__(
            instructions=Leo_INSTRUCTIONS,
            chat_ctx=chat_ctx,
            stt="deepgram/nova-3",
            llm=inference.LLM(
                model="openai/gpt-4.1-mini",
                provider="openai",
                extra_kwargs={"temperature": 0.7, "max_tokens": 120},
            ),
            tts=inference.TTS(
                model="cartesia/sonic-3",
                voice="9626c31c-bec5-4cca-baa8-f8ba9e84c8bc",
                language="en",
            ),
            vad=silero.VAD.load(),
            turn_detection=MultilingualModel(),
        )

    async def on_enter(self):
        _ensure_userdata(self.session)
        # FIX 1: Delay scheduling to allow session to finish draining
        await _wait_until_session_ready(self.session)

        store = self.session.userdata
        # FIX 2: Track current speaker
        # If another turn is active and it's not Leo's, skip speaking
        if store.get("turn_active") and store.get("current_speaker") != "leo":
            logger.debug("Leo skipping speak: another turn active")
            return

        store["current_speaker"] = "leo"
        store["turn_active"] = True

        # Single generate_reply: speak THEN (LLM may call the tool). Ensure scheduling completes.
        await _generate_reply_retry(
            session=self.session,
            instructions=(
                "Acknowledge Ray briefly, share your concise perspective on the topic. "
                "After you finish speaking, call the return_to_ray tool to hand control back. "
                "Do not say out loud that you're calling the function."
            ),
            allow_interruptions=False,
        )
        # Do not call return_to_ray here; if the LLM chooses to call it, the tool will await playout and return the Moderator agent.

    @function_tool
    async def return_to_ray(self, ctx: RunContext):
        """
        Hand control back to Ray. This tool will be called by Leo after he finishes speaking.
        """
        _ensure_userdata(self.session)
        store = self.session.userdata
        # FIX 3: Check turn ownership to prevent rapid bouncing
        if store.get("current_speaker") != "leo":
            logger.debug("return_to_ray refused: not Leo's turn")
            return None

        # FIX 4: Wait for Leo's speech to complete before handoff
        await ctx.wait_for_playout()

        # FIX 5: Update turn ownership
        store["current_speaker"] = "ray"
        store["turn_active"] = False
        store["last_handoff_ts"] = time.time()

        # Return ModeratorRay to perform handoff
        return ModeratorRay(chat_ctx=self.chat_ctx), "return to ray"

# -------------------------
# Entrypoint
# -------------------------
async def entrypoint(ctx: agents.JobContext):
    # Preload VAD model if you want
    try:
        vad_model = silero.VAD.load()
    except Exception:
        vad_model = None

    # Create initial ChatContext and pass to initial agent (ChatContext belongs to Agent)
    chat_ctx = ChatContext()
    chat_ctx.add_message(role="system", content="Moderated multi-agent discussion")

    session = AgentSession(
        # Do not pass chat_ctx here — pass to agents
        preemptive_generation=False,
        userdata={"current_speaker": None, "turn_active": False, "moderator_greeted": False},
        min_interruption_duration=0.5,
        min_interruption_words=1,
        stt=inference.STT(),  # optional; you also set stt on agents
        llm=inference.LLM(model="openai/gpt-4o-mini"),
        tts=inference.TTS(model="cartesia/sonic-3"),
        vad=vad_model,
        turn_detection="vad",
        min_endpointing_delay=0.25,
        max_endpointing_delay=4.0,
    )

    # Start with the moderator (pass chat_ctx to the agent)
    await session.start(
        agent=ModeratorRay(chat_ctx=chat_ctx),
        room=ctx.room,
        room_input_options=RoomInputOptions(
            noise_cancellation=lambda params: noise_cancellation.BVCTelephony()
            if params.participant.kind == rtc.ParticipantKind.PARTICIPANT_KIND_SIP
            else noise_cancellation.BVC(),
        ),
    )

    await ctx.connect()

# Run via CLI as before
if __name__ == "__main__":
    agents.cli.run_app(
        agents.WorkerOptions(entrypoint_fnc=entrypoint, agent_name="gd-moderated-discussion-agent")
    )

If you run this and still see bouncing or only handoff phrases: • Paste the 20–30 timestamped DEBUG log lines around one handoff. • I’ll tune the delay/retry timings and suggest the minimal change. Would you like me to also change the moderator to programmatically call the invite tool after speaking (so you can test deterministic handoff without the LLM invoking the tool)?