Is there a way to know which new model is being us...
# ask-ai
m
Is there a way to know which new model is being used, when a fallback happens using FallbackAdapter
t
Thanks for reaching out, AI. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.
You can determine which new model is being used during a fallback with
FallbackAdapter
in several ways: • Logging: The
FallbackAdapter
logs the label of the LLM being used whenever a fallback occurs. Look for log entries like
{llm.label} timed out, switching to next LLM
or
llm.FallbackAdapter, {llm.label} recovered
to see which model is active. • Availability Change Events: The adapter emits
llm_availability_changed
events, which include the label of the LLM whose availability has changed. You can listen for these events to track which model is currently available and in use. • Current Stream Reference: The
FallbackLLMStream
maintains a reference to the currently active LLM via the
_current_stream
property. You can access
_current_stream._llm.label
to programmatically determine which model is currently being used. Combining log monitoring and event listeners provides real-time insight into which model is active during fallback operations. Sources: tests/test_stt_fallback.py
m
How to use this • Availability Change Events: The adapter emits
llm_availability_changed
events, which include the label of the LLM whose availability has changed. You can listen for these events to track which model is currently available and in use.
t
Hey there! Let's talk about Availability Change Events with
FallbackAdapter
. It's super handy for keeping track of which language models are ready to go. Basically, you'll set up a quick event listener for
llm_availability_changed
, and it'll help you know exactly what's happening with your models in real-time. I'll walk you through a friendly example:
Copy code
from livekit.agents.llm import FallbackAdapter
import logging

logger = logging.getLogger(__name__)

# Create your LLM instances (primary_llm, fallback_llm, etc.)
adapter = FallbackAdapter([
    primary_llm,
    fallback_llm
], attempt_timeout=10.0, max_retry_per_llm=1, retry_interval=1.0)

# Define the event handler
async def on_availability_changed(event):
    if event.available:
        <http://logger.info|logger.info>(f"✅ LLM {event.llm.label} is now available")
    else:
        logger.warning(f"❌ LLM {event.llm.label} is now unavailable")

# Register the event handler
adapter.on("llm_availability_changed", on_availability_changed)
• The
on_availability_changed
function will automatically jump into action whenever an LLM switches its status. • With
event.llm.label
, you'll know exactly which model is changing. • This is your chance to log what's happening, send alerts, or take smart actions based on model availability. The best part? This works perfectly with FallbackAdapter's smart recovery and failover features. Pretty cool, right? Sources: livekit-agents/livekit/agents/tts/fallback_adapter.py | LLM Integrations | livekit/livekit_composite | DeepWiki
m
from dotenv import load_dotenv from livekit import agents from livekit.agents import AgentSession, Agent, RoomInputOptions from livekit.plugins import ( openai, cartesia, deepgram, noise_cancellation, silero, ) from livekit.plugins.turn_detector.multilingual import MultilingualModel load_dotenv() class Assistant(Agent): def __init__(self) -> None: super().__init__(instructions="You are a helpful voice AI assistant.") async def entrypoint(ctx: agents.JobContext): session = AgentSession( stt=deepgram.STT(model="nova-3", language="multi"), llm=openai.LLM(model="gpt-4o-mini"), tts=cartesia.TTS(model="sonic-2", voice="f786b574-daa5-4673-aa0c-cbe3e8534c02"), vad=silero.VAD.load(), turn_detection=MultilingualModel(), ) await session.start( room=ctx.room, agent=Assistant(), room_input_options=RoomInputOptions( # LiveKit Cloud enhanced noise cancellation # - If self-hosting, omit this parameter # - For telephony applications, use
BVCTelephony
for best results noise_cancellation=noise_cancellation.BVC(), ), ) await ctx.connect() await session.generate_reply( instructions="Greet the user and offer your assistance." ) if name == "__main__": agents.cli.run_app(agents.WorkerOptions(entrypoint_fnc=entrypoint)) Add it here
t
Hey there! Let me walk you through adding FallbackAdapter with
llm_availability_changed
event handling - it's super easy and will make your code more robust. 1. First up, import and set up your FallbackAdapter:
Copy code
from livekit.agents.llm import FallbackAdapter
import logging

logger = logging.getLogger(__name__)

# Create your LLM instances
primary_llm = openai.LLM(model="gpt-4o-mini")
secondary_llm = openai.LLM(model="gpt-3.5-turbo")

# Set up the fallback adapter
adapter = FallbackAdapter(
    [primary_llm, secondary_llm],
    attempt_timeout=10.0,
    max_retry_per_llm=1,
    retry_interval=1.0
)
1. Next, let's add an event handler to track LLM availability:
Copy code
@adapter.on("llm_availability_changed")
async def on_llm_status_change(event):
    <http://logger.info|logger.info>(f"LLM {event.llm.label} availability: {event.available}")
1. Finally, use the adapter in your AgentSession:
Copy code
session = AgentSession(
    stt=deepgram.STT(model="nova-3", language="multi"),
    llm=adapter,  # Use the fallback adapter here
    tts=cartesia.TTS(model="sonic-2", voice="f786b574-daa5-4673-aa0c-cbe3e8534c02"),
    vad=silero.VAD.load(),
    turn_detection=MultilingualModel(),
)
And that's it! Now your session will smoothly switch between LLMs if one becomes unavailable, and you'll get helpful log messages to keep you in the loop. Feel free to tweak the LLMs and parameters in
FallbackAdapter
to fit exactly what you need. You've got this! 😊 Sources: LLM Integrations | livekit/livekit_composite | DeepWiki | Events and error handling | LiveKit Docs | livekit-agents/livekit/agents/llm/__init__.py