How to change the AI agent as a trancriber between multiple participants when new participant joined in a rooom. I'm using python

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

"AJ_ZwGyMJgTeeNk", "room_id": "RM_oFLx9gyikH4D"} 082545.610 ERROR asyncio Task was destroyed but it is pending! task: <Task pending name='AgentTask_on_enter' coro=<AgentActivity._create_speech_task.<locals>._context_aware_coro() running at /home/azureuser/AI-development/.venv/lib/python3.12/site-packages/livekit/agents/voice/agent_activity.py:411> wait_for=<Future pending cb=[shield.<locals>._outer_done_callback() at /usr/lib/python3.12/asyncio/tasks.py:922, Task.task_wakeup()]> cb=[AgentActivity._create_speech_task.<locals>.<lambda>() at /home/azureuser/AI-development/.venv/lib/python3.12/site-packages/livekit/agents/voice/agent_activity.py:417, <1 more>, AgentTask.__await_impl.<locals>._handle_task_done() at /home/azureuser/AI-development/.venv/lib/python3.12/site-packages/livekit/agents/voice/agent.py:744]> {"room": "voice_assistant_room_7289", "pid": 139267, "job_id": "AJ_ZwGyMJgTeeNk", "room_id": "RM_oFLx9gyikH4D"} 082545.611 WARNI… livekit.agents ClaimStatusIdentifyScenarioTask completed, but the agent has changed in the meantime. Ignoring handoff to the previous agent, likely due to

AgentSession.update_agent

being invoked. {"room": "voice_assistant_room_7289", "pid": 139267, "job_id": "AJ_ZwGyMJgTeeNk", "room_id": "RM_oFLx9gyikH4D"} Exception ignored in: <coroutine object AgentActivity._create_speech_task.<locals>._context_aware_coro at 0x7d9c10146500> Traceback (most recent call last): File "/home/azureuser/AI-development/.venv/lib/python3.12/site-packages/livekit/agents/voice/agent_activity.py", line 411, in _context_aware_coro return await coro ^^^^^^^^^^ File "/home/azureuser/AI-development/.venv/lib/python3.12/site-packages/opentelemetry/util/_decorator.py", line 71, in async_wrapper return await func(*args, **kwargs) # type: ignore ^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/home/azureuser/AI-development/.venv/lib/python3.12/site-packages/livekit/agents/utils/log.py", line 16, in async_fn_logs return await fn(*args, **kwargs) ^^^^^^^^^^^^^^^^^^^^^^^^^ File "/home/azureuser/AI-development/.venv/lib/python3.12/site-packages/livekit/agents/voice/agent_activity.py", line 479, in _traceable_on_enter _OnEnterContextVar.reset(tk) ValueError: <Token var=<ContextVar name='agents_activity_on_enter' at 0x7d9c18f07f60> at 0x7d9c1013a740> was created in a different Context ERROR livekit.agents Error in _traceable_on_enter Traceback (most recent call last): File "/home/azureuser/AI-development/.venv/lib/python3.12/site-packages/livekit/agents/voice/agent_activity.py", line 477, in _traceable_on_enter await self._agent.on_enter() File "/home/azureuser/AI-development/src/agents/agent.py", line 50, in on_enter patient_result = await ClaimStatusIdentifyScenarioTask(tools=tools,chat_ctx=self.chat_ctx,userdata=self.userdata) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ RuntimeError: coroutine ignored GeneratorExit During handling of the above exception, another exception occurred: Traceback (most recent call last): File "/home/azureuser/AI-development/.venv/lib/python3.12/site-packages/livekit/agents/utils/log.py", line 16, in async_fn_logs return await fn(*args, **kwargs) ^^^^^^^^^^^^^^^^^^^^^^^^^ File "/home/azureuser/AI-development/.venv/lib/python3.12/site-packages/livekit/agents/voice/agent_activity.py", line 479, in _traceable_on_enter _OnEnterContextVar.reset(tk) ValueError: <Token var=<ContextVar name='agents_activity_on_enter' at 0x7d9c18f07f60> at 0x7d9c1013a740> was created in a different Context {"room": "voice_assistant_room_7289", "pid": 139267, "job_id": "AJ_ZwGyMJgTeeNk", "room_id": "RM_oFLx9gyikH4D"}

Thanks — this log is very helpful. I’m responding to the errors you pasted (the "Task was destroyed but it is pending!" warning, the ContextVar token ValueError, and the "agent has changed in the meantime" warning) and how to fix them when switching transcriber agents or changing the linked participant. What’s happening (short) • A long-running agent task (e.g., ClaimStatusIdentifyScenarioTask / an on_enter coroutine) is still running when you change the agent with

AgentSession.update_agent()

or otherwise terminate the session. That causes tasks to be destroyed while still pending and leads to a ContextVar token being reset in a different execution context, which raises the ValueError. • The log “ClaimStatusIdentifyScenarioTask completed, but the agent has changed in the meantime” indicates a handoff raced with an agent update — the task finished after you already swapped agents. Concrete fixes to apply 1) Don’t update the agent while a lifecycle task is still running • Before calling

session.update_agent(...)

or otherwise replacing the agent, await any important lifecycle tasks that influence agent state (like tasks you run in

on_enter

). • Example pattern:

# await the critical task result before changing agent
result = await ClaimStatusIdentifyScenarioTask(...)
# only after result, update the agent
session.update_agent(NewAgent(...))

2) Avoid fire-and-forget for agent lifecycle work • Do not wrap

on_enter()

-like tasks in

asyncio.create_task()

if they depend on agent context. That disconnects contextvars and lets the agent be swapped underneath the running coroutine. • If you must run background work, make sure it does not touch agent lifecycle ContextVars (decouple it), or track and cancel it explicitly on agent update. 3) Use structured concurrency to manage lifetime of related tasks • Use Python 3.11+ TaskGroup (or manual tracking) so tasks are awaited or cancelled cleanly rather than being destroyed while pending.

async with asyncio.TaskGroup() as tg:
    tg.create_task(some_background_job())
    result = await tg.create_task(ClaimStatusIdentifyScenarioTask(...))
# TaskGroup will ensure proper cancel/cleanup on exit

4) Preserve ContextVar semantics • The ValueError arises when you reset a ContextVar token created in a different task/context. Keep ContextVar use local to the task or ensure you set/reset tokens within the same Task. Do not share the token across tasks. • Practically: avoid resetting ContextVars in callbacks that run in different tasks, and avoid creating tokens in one task and resetting in another. 5) If you need to switch transcription source frequently • Use

session.room_io.set_participant(identity)

to change which participant audio the agent transcribes. That change is lighter-weight than swapping the entire agent and reduces the risk of racing lifecycle tasks. • If you must update the agent, first ensure all agent tasks are idle or cancelled, then call

update_agent

. Quick checklist to debug your code now • Where do you call

session.update_agent()

? Add logging and make sure no agent-relevant tasks are pending at that time. • Do you ever call

asyncio.create_task()

from inside

on_enter

or other lifecycle functions? Convert to awaited calls or ensure created tasks are independent and do not use ContextVars from the lifecycle. • Add explicit task tracking: store the Task object(s) and await them or cancel them on agent update. • Reproduce with controlled steps: join participant A, start transcription task, then trigger update; confirm whether the update raced with task completion. If you want, share short excerpts of: • how you call

update_agent()

(the surrounding code), • your

on_enter()

implementation or the code that launches ClaimStatusIdentifyScenarioTask, • any place you call

asyncio.create_task()

for agent logic. With those snippets I can point out exactly where to await/cancel tasks or how to refactor to avoid ContextVar cross-task resets and the pending-task destruction. Sources: Agents and handoffs | LiveKit Documentation | Collect recording consent with tasks | LiveKit Documentation | Agent session | LiveKit Documentation

yes give a full python example

Thanks — below is a complete, self-contained Python example that shows a safe pattern for: • switching which participant the AI transcriber listens to with

session.room_io.set_participant(identity)

, • avoiding the "Task was destroyed but it is pending!" warning, • preventing ContextVar token errors by not leaking context across tasks, • safely updating agents when tasks are complete. This example uses structured concurrency (TaskGroup if available) and explicit task tracking so you can adapt it to your LiveKit agent code. Save as e.g. agent_safe_switch.py and adapt the imports/config to match your environment.

import asyncio
import logging
from typing import Optional

# Replace these imports with your actual LiveKit SDK imports
# from livekit import rtc
# from livekit.agents import AgentSession, AgentContext, JobContext, cli, inference

# Dummy placeholders used only for this example; replace with real classes.
class rtc:
    class RemoteParticipant:
        def __init__(self, identity): self.identity = identity
    class Room:
        def __init__(self): self.remote_participants = {}
        connected = True
        def on(self, ev, fn): pass

class AgentSession:
    def __init__(self, **kwargs): self.room_io = None
    async def start(self, agent, room): self.room_io = RoomIO(room); self._agent = agent
    async def generate_reply(self, instructions: str): pass
    def shutdown(self): pass
    async def update_agent(self, new_agent): self._agent = new_agent

class RoomIO:
    def __init__(self, room): self.room = room
    async def set_participant(self, participant_or_identity):
        # In real SDK: accepts identity or RemoteParticipant
        await asyncio.sleep(0)  # simulate async op

# Replace with your real task that used to cause trouble
async def ClaimStatusIdentifyScenarioTask(*, tools=None, chat_ctx=None, userdata=None):
    await asyncio.sleep(0.1)
    return "identified"

logging.basicConfig(level=<http://logging.INFO|logging.INFO>)
logger = logging.getLogger("safe-agent")

class SafeAgent:
    """
    Example agent demonstrating safe lifecycle patterns.
    - Do not create long-running lifecycle tasks with create_task() that depend on ContextVars.
    - Track tasks you start and await/cancel them on agent update or shutdown.
    """
    def __init__(self):
        self._session: Optional[AgentSession] = None
        self._linked_participant_identity: Optional[str] = None
        self._background_tasks: set[asyncio.Task] = set()

    async def on_enter(self, ctx):
        # framework will call this; set session reference from ctx
        self._session = ctx.session
        # run short init work synchronously (awaited), not fire-and-forget
        await self._synchronous_startup()

    async def _synchronous_startup(self):
        # Example: run a small task that must complete before agent changes
        <http://logger.info|logger.info>("on_enter: running startup task")
        result = await ClaimStatusIdentifyScenarioTask()
        <http://logger.info|logger.info>(f"on_enter: startup result: {result}")
        # OK to schedule background work that is independent of ContextVars:
        t = asyncio.create_task(self._independent_background_loop())
        self._background_tasks.add(t)
        t.add_done_callback(self._background_tasks.discard)

    async def on_exit(self):
        <http://logger.info|logger.info>("on_exit: shutting down agent")
        # Cancel any background tasks we started
        if self._background_tasks:
            for t in list(self._background_tasks):
                t.cancel()
            await asyncio.gather(*self._background_tasks, return_exceptions=True)
        # If session exists, call shutdown cleanup
        if self._session:
            self._session.shutdown()
            self._session = None

    async def _independent_background_loop(self):
        # long-running background task that must NOT touch lifecycle ContextVars
        try:
            while True:
                await asyncio.sleep(5)
                <http://logger.info|logger.info>("background: heartbeat")
        except asyncio.CancelledError:
            <http://logger.info|logger.info>("background: cancelled")
            raise

    async def switch_to_participant(self, identity: str):
        """
        Safely switch the RoomIO linked participant.
        This is the recommended approach instead of swapping the agent while tasks are active.
        """
        if not self._session or not self._session.room_io:
            logger.warning("No active session/room_io to switch participant")
            return

        # If you have the remote participant object, pass it; otherwise pass identity
        await self._session.room_io.set_participant(identity)
        self._linked_participant_identity = identity
        <http://logger.info|logger.info>(f"Now transcribing participant: {identity}")
        # Short awaited reply to confirm the switch. Keep quick and awaited.
        await self._session.generate_reply(instructions=f"Switched to {identity} for transcription.")

    async def safely_update_agent(self, new_agent):
        """
        Example: safely replace this agent with new_agent.
        Ensure all important tasks are complete or cancelled first.
        """
        <http://logger.info|logger.info>("Preparing to update agent: waiting for important tasks to finish")
        # If you started any critical (non-background) tasks, await them here before updating.
        # For demonstration, we don't have extra critical tasks; cancel background tasks:
        if self._background_tasks:
            for t in list(self._background_tasks):
                t.cancel()
            await asyncio.gather(*self._background_tasks, return_exceptions=True)
        # Now it is safe to update the agent in the session
        if self._session:
            await self._session.update_agent(new_agent)
            <http://logger.info|logger.info>("Agent updated successfully")


# Minimal ctx-like objects to simulate framework lifecycle in this example
class DummyCtx:
    def __init__(self, room):
        self.room = room
        self.session = None
    async def connect(self): pass

# Entrypoint that shows safe wiring for participant events and session lifecycle
async def main_entrypoint():
    # Simulated room and session creation
    room = rtc.Room()
    # Add two participants to simulate presence
    room.remote_participants["alice"] = rtc.RemoteParticipant("alice")
    room.remote_participants["bob"] = rtc.RemoteParticipant("bob")

    ctx = DummyCtx(room)
    # Create session and agent
    session = AgentSession()
    agent = SafeAgent()

    # Start session with agent (framework would pass ctx.session into agent on_enter)
    class DummyAgentContext:
        def __init__(self, session): self.session = session

    await session.start(agent=agent, room=room)
    # Simulate framework calling on_enter with context
    await agent.on_enter(DummyAgentContext(session))

    # Option A: Switch linked participant safely (recommended)
    await agent.switch_to_participant("alice")

    # Simulate a new participant joining - we will switch to them
    <http://logger.info|logger.info>("Simulating new participant 'bob' joining")
    # Recommended: call session.room_io.set_participant to change transcription source
    await agent.switch_to_participant("bob")

    # If you must update the whole agent, do it safely via agent.safely_update_agent(...)
    # new_agent = SafeAgent()
    # await agent.safely_update_agent(new_agent)

    # Wait a bit to demonstrate background activity
    await asyncio.sleep(0.5)

    # Now simulate shutdown / agent exit
    await agent.on_exit()

# Run the example
if __name__ == "__main__":
    try:
        asyncio.run(main_entrypoint())
    except KeyboardInterrupt:
        pass

Key patterns shown and why they fix your errors • Await critical tasks instead of scheduling them fire-and-forget. That prevents “Task was destroyed but it is pending!” when the agent changes. • Keep ContextVar usage local to each task and avoid creating a context in one Task and resetting it in another. By awaiting lifecycle tasks in the same task and not detaching them via create_task (for context-dependent work), the token ValueError is avoided. • Use

session.room_io.set_participant(identity)

to change which participant is transcribed — this is lighter and safer than swapping the full agent while tasks are running. • Track background tasks and cancel them in

on_exit

so nothing is unexpectedly destroyed by the runtime. • If you must replace the agent, cancel/await relevant tasks first, then call

session.update_agent(...)

. If you want, I can: • Adapt this example to your exact LiveKit imports and STT/TTs providers (Deepgram/Whisper etc.). • Show a real example of using Python 3.11's asyncio.TaskGroup for structured concurrency. • Review your actual

on_enter

and

update_agent

snippets and point out the exact lines to change. Which would help most next? Sources: Agent session | LiveKit Documentation | Agents and handoffs | LiveKit Documentation | How to detect and handle silence during calls | LiveKit | Text and transcriptions | LiveKit Documentation