how to implement retry logic on LLM node for google.LLM when the LLM response is empty?

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

I need a mechanism to handle when the response from llm is empty. how do I do that when the LLM used is google.LLM?

I'm here to help you gracefully handle empty responses from

google.LLM

. Here are some friendly strategies you can try: • Log Warnings: It’s a good idea to log a warning whenever the response doesn’t have valid candidates or content parts. This way, you stay informed but the process keeps moving smoothly. For example:

if not response.candidates or not response.candidates[0].content:
    logger.warning(f"No candidates in the response: {response}")
    continue

• Skip Empty Chunks: When you’re working through chunks, simply filter out any empty content to prevent errors from popping up:

if not delta.content:
    return None

• Retry Logic: Implement a custom retry loop that handles empty responses by raising a retryable error. This can really help keep things running smoothly. You can refer back to the earlier example for how to set this up. • Fallback Mechanism: Consider using the

FallbackAdapter

to switch over to another LLM if one fails or times out. This adds a nice layer of resilience if you encounter repeated empty responses. • Inject Dummy Messages: Sometimes, to meet API requirements, it helps to inject a dummy user message—like a simple period `'.'`—especially if the last role in the conversation isn’t "user." These tips should help you handle empty or malformed responses more robustly. If anything feels unclear or if you’d like some extra guidance with implementation, just let me know—I’m here to support you! Sources: livekit-agents/livekit/agents/llm/_provider_format/google.py | Modifying LLM output before TTS | LiveKit Docs | agents/src/llm/utils.test.ts

should this be in entrypoint or inside the agent? please give me the full code

Hey there! Just a quick tip: the retry logic for empty responses in

google.LLM

works best when it's handled inside the agent itself, rather than in the entrypoint. Keeping the entrypoint simple helps it focus on initializing and starting the agent, while the agent takes care of the more specific tasks like interacting with the LLM and managing retries. To make things clearer, here’s a complete code example that shows how you can implement retry logic for empty responses:

from livekit import agents
from livekit.agents import llm, JobContext, WorkerOptions
from livekit.plugins.google import LLM as GoogleLLM
from livekit.agents.llm import ChatContext
import asyncio

class RetryingGoogleLLM:
    """Wrapper that retries on empty responses from Google LLM."""
    
    def __init__(self, model="gemini-1.5-flash", max_retries=3):
        self.llm = GoogleLLM(model=model)
        self.max_retries = max_retries

    async def chat(self, chat_ctx: ChatContext) -> llm.LLMStream:
        for attempt in range(self.max_retries + 1):
            stream = self.llm.chat(chat_ctx=chat_ctx)
            empty_response = True
            
            try:
                # Consume the stream to detect if any content is produced
                async for chunk in stream:
                    if chunk.delta and chunk.delta.content:
                        empty_response = False
                    yield chunk
                    
                # If no content was yielded, treat as failure
                if not empty_response:
                    return  # Successful response
                
                # Empty response — retry if attempts remain
                if attempt < self.max_retries:
                    print(f"Empty response received, retrying... (attempt {attempt + 1})")
                    await asyncio.sleep(1)  # Simple backoff
                else:
                    print("Max retries reached with empty responses")
                    raise Exception("Google LLM returned empty response after retries")

            except Exception as e:
                if attempt < self.max_retries:
                    print(f"Error occurred: {e}, retrying... (attempt {attempt + 1})")
                    await asyncio.sleep(1)
                else:
                    raise

async def entrypoint(ctx: JobContext):
    # Minimal entrypoint — just initializes agent logic
    llm_client = RetryingGoogleLLM(model="gemini-1.5-flash")
    
    chat_ctx = ChatContext()
    chat_ctx.add(
        message="Explain quantum entanglement in simple terms",
        role="user"
    )
    
    try:
        async for chunk in await llm_client.chat(chat_ctx):
            if chunk.delta and chunk.delta.content:
                print("LLM:", chunk.delta.content)
    except Exception as e:
        print(f"Failed to get valid response: {e}")

if __name__ == "__main__":
    WorkerOptions(entrypoint_fnc=entrypoint).run()

Here are some key points to keep in mind: • Retry Logic: This wrapper carefully retries when it encounters empty responses by consuming the stream and checking if any content was produced. • Backoff: There’s a gentle pause (

asyncio.sleep

) between retries to avoid putting too much pressure on the service. • Encapsulation: By keeping retry logic inside the agent, you get better control and a cleaner separation of responsibilities. If anything’s unclear or you’d like me to walk you through any part, just let me know—I’m here to help! Sources: tests/fake_llm.py | plugins/openai/src/llm.ts | meditation_assistant/assistant_functions.py

I have a voice ai pipeline with STT -> LLM -> TTS. my entry point initialises the AgentSession and handsover the control of the conversation to Agent. the Agent then continues with the conversation. How do I explicitly write the retry logic in Agent for LLM interaction only. I would like to be able to handle situation like the one given below (extract from my log) - 2025-09-16 112254,555 - WARNING livekit.plugins.google - no candidates in the response: sdk_http_response=HttpResponse( headers=<dict len=11> ) candidates=[Candidate( content=Content( role='model'

I'm happy to help you with writing retry logic for LLM interactions in your voice AI pipeline (STT → LLM → TTS). You can implement this directly inside the Agent to gracefully handle cases like the one you saw in your log, where the LLM response has no candidates. Let’s walk through it together: 1. Use a Custom Wrapper for LLM Retry Logic A great approach is to encapsulate the retry logic in a wrapper class that detects empty responses and retries as needed. Here’s an example you can build on:

from livekit.agents import llm, JobContext
from livekit.plugins.google import LLM as GoogleLLM
from livekit.agents.llm import ChatContext
import asyncio

class RetryingGoogleLLM:
    """Wrapper that retries on empty responses from Google LLM."""
    def __init__(self, model="gemini-1.5-flash", max_retries=3):
        self.llm = GoogleLLM(model=model)
        self.max_retries = max_retries

    async def chat(self, chat_ctx: ChatContext) -> llm.LLMStream:
        for attempt in range(self.max_retries + 1):
            stream = self.llm.chat(chat_ctx=chat_ctx)
            empty_response = True

            try:
                # Consume the stream to detect if any content is produced
                async for chunk in stream:
                    if chunk.delta and chunk.delta.content:
                        empty_response = False
                    yield chunk

                # If no content was yielded, treat as failure
                if not empty_response:
                    return  # Successful response

                # Empty response — retry if attempts remain
                if attempt < self.max_retries:
                    print(f"Empty response received, retrying... (attempt {attempt + 1})")
                    await asyncio.sleep(1)  # Simple backoff
                else:
                    print("Max retries reached with empty responses")
                    raise Exception("Google LLM returned empty response after retries")

            except Exception as e:
                if attempt < self.max_retries:
                    print(f"Error occurred: {e}, retrying... (attempt {attempt + 1})")
                    await asyncio.sleep(1)
                else:
                    raise

async def entrypoint(ctx: JobContext):
    # Initialize the retrying LLM wrapper
    llm_client = RetryingGoogleLLM(model="gemini-1.5-flash")

    chat_ctx = ChatContext()
    chat_ctx.add(
        message="Explain quantum entanglement in simple terms",
        role="user"
    )

    try:
        async for chunk in await llm_client.chat(chat_ctx):
            if chunk.delta and chunk.delta.content:
                print("LLM:", chunk.delta.content)
    except Exception as e:
        print(f"Failed to get valid response: {e}")

2. Integrate with Your Agent Within your Agent, you can replace the default LLM interaction with this retrying wrapper. This way, retries happen smoothly behind the scenes without extra hassle on your end. 3. Use FallbackAdapter for Multi-Provider Resilience If you’d like to add even more robustness by failing over to another LLM provider when retries are exhausted, the

FallbackAdapter

is a handy tool:

from livekit.agents.llm import FallbackAdapter
from livekit.plugins.google import LLM as GoogleLLM
from livekit.plugins.openai import LLM as OpenAILLM

llm = FallbackAdapter([
    GoogleLLM(model="gemini-1.5-flash"),
    OpenAILLM(model="gpt-4o")
], max_retry_per_llm=2, retry_on_chunk_sent=True)

This setup will try retries within each LLM and then switch to the next provider if needed, giving your pipeline extra resilience. If you have any questions or want me to guide you through integrating this into your pipeline, just let me know—I’m here to help! Sources: LLM Integrations | livekit/livekit_composite | DeepWiki | agents/src/voice/agent.ts | Modifying LLM output before TTS | LiveKit Docs | livekit-agents/livekit/agents/llm/fallback_adapter.py | Events and error handling | LiveKit Docs

async def entrypoint(ctx: JobContext): session = AgentSession( stt=deepgram.STT(), llm=google.LLM(model="gemini-2.5-flash-lite"), tts=cartesia.TTS(), vad=silero.VAD.load(), ) await session.start(agent=MyAgent(), room=ctx.room)

To help you implement retry logic for the LLM in your

AgentSession

, you can create a custom retrying wrapper around the

google.LLM

instance. This approach makes sure retries happen smoothly and transparently when interacting with the LLM. Here’s a friendly guide on how you can update your code:

from livekit.agents import AgentSession, JobContext
from livekit.plugins.google import LLM as GoogleLLM
from livekit.plugins.deepgram import STT
from livekit.plugins.cartesia import TTS
from livekit.plugins.silero import VAD
import asyncio

class RetryingGoogleLLM:
    """Wrapper that retries on empty responses from Google LLM."""
    def __init__(self, model="gemini-2.5-flash-lite", max_retries=3):
        self.llm = GoogleLLM(model=model)
        self.max_retries = max_retries

    async def chat(self, chat_ctx):
        for attempt in range(self.max_retries + 1):
            stream = self.llm.chat(chat_ctx=chat_ctx)
            empty_response = True

            try:
                async for chunk in stream:
                    if chunk.delta and chunk.delta.content:
                        empty_response = False
                    yield chunk

                if not empty_response:
                    return

                if attempt < self.max_retries:
                    print(f"Empty response received, retrying... (attempt {attempt + 1})")
                    await asyncio.sleep(1)
                else:
                    raise Exception("Google LLM returned empty response after retries")

            except Exception as e:
                if attempt < self.max_retries:
                    print(f"Error occurred: {e}, retrying... (attempt {attempt + 1})")
                    await asyncio.sleep(1)
                else:
                    raise

async def entrypoint(ctx: JobContext):
    session = AgentSession(
        stt=STT(),
        llm=RetryingGoogleLLM(model="gemini-2.5-flash-lite"),
        tts=TTS(),
        vad=VAD.load(),
    )

    await session.start(agent=MyAgent(), room=ctx.room)

Key Changes: • We replaced the

google.LLM

instance with

RetryingGoogleLLM

, which adds retry logic to handle empty responses gracefully. • The retry logic is nicely wrapped inside the custom class, making it easy to integrate with your

AgentSession

without extra hassle. If you have any questions or need a hand with anything else, please don’t hesitate to ask—I’m here to help!

AttributeError: 'RetryingGoogleLLM' object has no attribute 'on' since I am listening to @session.llm.on("metrics_collected"). how to fix this?

It looks like the AttributeError is happening because your custom

RetryingGoogleLLM

wrapper doesn’t implement the

on

method, which is part of the

EventEmitter

interface used by LiveKit’s LLM components. No worries though! To fix this, you just need to make sure your wrapper properly forwards event handling to the underlying

google.LLM

instance. Here’s a friendly guide on how you can update your

RetryingGoogleLLM

class to support event handling:

from livekit.agents.llm import LLM
from livekit.plugins.google import LLM as GoogleLLM
from livekit.rtc import EventEmitter
import asyncio

class RetryingGoogleLLM(LLM, EventEmitter):
    """Wrapper that retries on empty responses from Google LLM."""
    def __init__(self, model="gemini-2.5-flash-lite", max_retries=3):
        super().__init__()
        self.llm = GoogleLLM(model=model)
        self.max_retries = max_retries

        # Forward events from the underlying LLM
        self.llm.on("metrics_collected", self.emit)
        self.llm.on("error", self.emit)

    async def chat(self, chat_ctx):
        for attempt in range(self.max_retries + 1):
            stream = self.llm.chat(chat_ctx=chat_ctx)
            empty_response = True

            try:
                async for chunk in stream:
                    if chunk.delta and chunk.delta.content:
                        empty_response = False
                    yield chunk

                if not empty_response:
                    return

                if attempt < self.max_retries:
                    print(f"Empty response received, retrying... (attempt {attempt + 1})")
                    await asyncio.sleep(1)
                else:
                    raise Exception("Google LLM returned empty response after retries")

            except Exception as e:
                if attempt < self.max_retries:
                    print(f"Error occurred: {e}, retrying... (attempt {attempt + 1})")
                    await asyncio.sleep(1)
                else:
                    raise

Here are the key updates to keep in mind: • Your

RetryingGoogleLLM

class now inherits from

EventEmitter

and forwards events from the underlying

google.LLM

instance using

self.llm.on

. • This means any event listeners attached to

session.llm

will work just as you expect. With this change, you’ll be able to use

@session.llm.on("metrics_collected")

without running into that AttributeError. If you have any more questions or need a hand with this, please don’t hesitate to ask — I’m here to help! Sources: livekit-plugins/livekit-plugins-openai/livekit/plugins/openai/llm.py | Events and error handling | LiveKit Docs | Events and error handling | LiveKit Docs

there is no attribute called emit self.llm.on("metrics_collected", self.emit)

I don't have the answer you're looking for. You could also try asking your question: • in one of the other Slack channels or • to https://deepwiki.com/livekit/livekit_composite which is trained on all LiveKit source code If you find the answer, please post it here to help others!

is there a way I can override llm_node method to handle responses?