I'm doing well, thanks. Let me check your account. <function=fetch_customer_profile>{"phone_number": "0000000000"}</function>

continue_turn()

llm_node

generate_reply()

llm_node

llm_node

tool_call

import asyncio
from typing import AsyncIterable, Optional
from livekit.agents import Agent, ChatContext, FunctionTool, ModelSettings
from livekit.agents.llm import ChatChunk

class AutoContinueAgent(Agent):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        # tune backoff / guard params as needed
        self._follow_up_cooldown = 1.0  # seconds to avoid immediate retrigger

    async def llm_node(
        self,
        chat_ctx: ChatContext,
        tools: list[FunctionTool],
        model_settings: Optional[ModelSettings] = None,
    ) -> AsyncIterable[ChatChunk]:
        """
        Override that streams LLM output and, when it detects a function-only
        assistant message, triggers a single follow-up generation via
        session.generate_reply(...).
        """
        # Per-turn flag to prevent duplicate follow-ups
        follow_up_triggered = False

        # Helper to schedule follow-up without blocking stream
        def schedule_follow_up():
            async def _do_follow_up():
                await asyncio.sleep(0.05)  # tiny delay to avoid race conditions
                # Guard: ensure we don't re-trigger too fast
                if getattr(self, "_last_follow_up_ts", 0) + self._follow_up_cooldown > asyncio.get_event_loop().time():
                    return
                self._last_follow_up_ts = asyncio.get_event_loop().time()
                await self.session.generate_reply(
                    instructions=(
                        "You just invoked a tool. Now provide a concise, natural follow-up "
                        "to the user that summarizes or uses the tool result."
                    ),
                    allow_interruptions=True,
                )
            asyncio.create_task(_do_follow_up())

        # Stream from default LLM behavior
        async for chunk in super().llm_node(chat_ctx, tools, model_settings):
            # Yield chunk to caller so normal streaming still works
            yield chunk

            # Detect a function-only assistant message once per turn
            if not follow_up_triggered and self._is_function_call_only_chunk(chunk):
                follow_up_triggered = True
                schedule_follow_up()
                # do not immediately return; allow remaining stream chunks to flush

        # end of stream

    def _is_function_call_only_chunk(self, chunk: ChatChunk) -> bool:
        """
        Return True if the chunk represents a function/tool invocation and
        contains no assistant text content in the chunk's delta.
        Adjust this to your LLM plugin's stream structure.
        """
        # If chunk is plain text string -> not a function-only call
        if isinstance(chunk, str):
            return False

        # Safely inspect attributes commonly emitted by LLM streams
        delta = getattr(chunk, "delta", None)
        if not delta:
            return False

        # Example patterns to check:
        # - structured field like delta.tool_call or delta.tool_calls
        # - absence of content / text in delta.content or delta.message
        has_tool_call = False
        if hasattr(delta, "tool_call") and getattr(delta, "tool_call"):
            has_tool_call = True
        elif hasattr(delta, "tool_calls") and getattr(delta, "tool_calls"):
            has_tool_call = True
        elif hasattr(delta, "tool") and getattr(delta, "tool"):
            has_tool_call = True

        # check for presence of any verbal content in this chunk
        content = getattr(delta, "content", None) or getattr(delta, "text", None) or getattr(delta, "message", None)
        has_text = False
        if content:
            # treat empty/whitespace-only as no text
            try:
                text = "".join([c for c in content]) if isinstance(content, (list, tuple)) else str(content)
            except Exception:
                text = str(content)
            if text.strip():
                has_text = True

        return has_tool_call and not has_text

generate_reply()

llm_node

self.session.generate_reply(...)

llm_node

self.session.generate_reply(...)

self.session.generate_reply(...)

llm_node

llm_node

await self.session.generate_reply(...)

asyncio.create_task(...)

generate_reply()

asyncio.create_task()

instructions

generate_reply()

await context.session.generate_reply(...)

_no_followup

llm_node

generate_reply()

context.meta["suppress_followup"] = True

llm_node

await self.session.generate_reply(...)

asyncio.create_task()

llm_node

await self.session.generate_reply(...)

asyncio.create_task(...)

context.meta["suppress_followup"] = True

llm_node

await self.session.generate_reply(...)

asyncio.create_task()

context.meta["suppress_followup"] = True

import asyncio
from typing import AsyncIterable, Optional
from livekit.agents import Agent, ChatContext, FunctionTool, ModelSettings, RunContext
from livekit.agents.llm import ChatChunk

class UniversalFollowupAgent(Agent):
    def __init__(self, *args, cooldown_seconds: float = 2.0, **kwargs):
        super().__init__(*args, **kwargs)
        self._cooldown_seconds = cooldown_seconds
        self._last_followup_ts = 0.0
        self._pending_followup_task: Optional[asyncio.Task] = None

    async def llm_node(
        self,
        chat_ctx: ChatContext,
        tools: list[FunctionTool],
        model_settings: Optional[ModelSettings] = None,
    ) -> AsyncIterable[ChatChunk]:
        """
        Streams the LLM. If the stream indicates a function-only assistant message
        (tool call with no assistant text), schedule a follow-up generation unless
        a tool set context.meta['suppress_followup'] = True or cooldown applies.
        """

        followup_scheduled_for_turn = False
        saw_tool_call_in_turn = False

        # Stream default LLM output
        async for chunk in super().llm_node(chat_ctx, tools, model_settings):
            yield chunk

            # Detect tool/function call in chunk (adjust to your LLM's stream shape):
            if self._chunk_has_tool_call(chunk):
                saw_tool_call_in_turn = True

            # If chunk also contains assistant text, then not function-only for that chunk
            # (we still keep scanning until stream ends)
        
        # After stream completes for this LLM pass:
        if not saw_tool_call_in_turn:
            return  # nothing to do

        # Determine if a tool explicitly suppressed follow-ups.
        # Tools can set during execution:
        #   context.meta["suppress_followup"] = True
        suppress = self._check_suppress_followup_flag()

        if suppress:
            return

        # Cooldown and pending-task checks
        now = asyncio.get_event_loop().time()
        if now - self._last_followup_ts < self._cooldown_seconds:
            return

        if self._pending_followup_task and not self._pending_followup_task.done():
            return

        # Update last-followup timestamp and schedule a follow-up.
        self._last_followup_ts = now
        # Option A: await directly (safe if you want to block until follow-up finishes)
        # await self.session.generate_reply(instructions="You invoked a tool—now provide a concise follow-up to the user.")
        #
        # Option B (recommended inside streaming hooks): schedule non-blocking task
        self._pending_followup_task = asyncio.create_task(self._do_generate_followup())

    def _chunk_has_tool_call(self, chunk: ChatChunk) -> bool:
        """
        Heuristic: return True if the chunk contains a tool/function call and no user-facing text.
        Adjust checks to match your LLM plugin's streaming fields (e.g., delta.tool_call,
        delta.tool_calls, delta.function_call, delta.content).
        """
        if isinstance(chunk, str):
            return False

        delta = getattr(chunk, "delta", None)
        if not delta:
            return False

        # Example checks; adapt to your plugin:
        has_tool = False
        if hasattr(delta, "tool_call") and getattr(delta, "tool_call"):
            has_tool = True
        elif hasattr(delta, "tool_calls") and getattr(delta, "tool_calls"):
            has_tool = True
        elif hasattr(delta, "function_call") and getattr(delta, "function_call"):
            has_tool = True

        # Check for any textual content in this chunk
        content = getattr(delta, "content", None) or getattr(delta, "text", None) or getattr(delta, "message", None)
        has_text = False
        if content:
            try:
                text = "".join(content) if isinstance(content, (list, tuple)) else str(content)
            except Exception:
                text = str(content)
            if text.strip():
                has_text = True

        # Consider it a tool call chunk only if there is a tool marker and no textual content.
        return has_tool and not has_text

    def _check_suppress_followup_flag(self) -> bool:
        """
        Check RunContext or other context storage for suppress flag written by tools.
        Tools should set:
            context.meta['suppress_followup'] = True
        or
            context.userdata.setdefault('meta', {})['suppress_followup'] = True
        depending on your RunContext usage.
        """
        try:
            ctx = RunContext.get_current()
            if not ctx:
                return False
            # prefer context.meta if available
            if hasattr(ctx, "meta") and isinstance(ctx.meta, dict):
                return bool(ctx.meta.get("suppress_followup"))
            # fallback to userdata.meta
            if hasattr(ctx, "userdata") and isinstance(ctx.userdata, dict):
                meta = ctx.userdata.get("meta") or {}
                return bool(meta.get("suppress_followup"))
        except Exception:
            pass
        return False

    async def _do_generate_followup(self):
        """
        The actual follow-up: call generate_reply() directly.
        We include a tiny sleep to reduce race conditions between tool result injection and LLM start.
        """
        try:
            await asyncio.sleep(0.05)
            await self.session.generate_reply(
                instructions=(
                    "You previously invoked a tool. Provide a concise, user-facing follow-up "
                    "that summarizes or uses the tool result."
                ),
                allow_interruptions=True,
            )
        except Exception:
            # log/monitor as appropriate in your environment
            pass
        finally:
            self._pending_followup_task = None

context.meta["suppress_followup"] = True
    # or
    context.userdata.setdefault("meta", {})["suppress_followup"] = True

_chunk_has_tool_call()

delta.function_call

delta.tool_call

asyncio.create_task

await self.session.generate_reply(...)

_cooldown_seconds

_chunk_has_tool_call()

await generate_reply()

session.generate_reply()

generate_reply()

instructions

generate_reply()

instructions = "Continue from where the assistant left off. Don't repeat earlier responses like greetings or acknowledgments. Only add new info based on the tool output."

instructions

session.generate_reply()

generate_reply()

generate_reply()

context.meta["suppress_followup"] = True

generate_reply()

agent.chat_ctx

generate_reply()

await self.session.generate_reply(...)

asyncio.create_task()

generate_reply()

session.generate_reply()

session.generate_reply()

session.generate_reply()

session.generate_reply()

generate_reply()

generate_reply()

generate_reply()

generate_reply()

await agent.update_chat_ctx(modified_chat_ctx)

session.updateChatCtx

await session.generate_reply(...)

generate_reply()

answered: true

generate_reply(..., ignorePrior=true)

generate_reply()

context.meta["suppress_followup"] = True

update_chat_ctx()

generate_reply()

update_chat_ctx()

generate_reply()

update_chat_ctx()

generate_reply()

await self.update_chat_ctx(...)

await self.session.generate_reply(...)

context.meta["suppress_followup"] = True

async def continue_after_tool(self, tool_name: str):
    """
    Call this after a tool has executed and been injected into chat context.
    It annotates/replaces the prior assistant message to avoid repetition,
    updates the chat context, and triggers a focused generate_reply().
    """
    # 1) Quick guard: check if tool opted out
    try:
        ctx = RunContext.get_current()
        if ctx and getattr(ctx, "meta", None) and ctx.meta.get("suppress_followup"):
            return
    except Exception:
        pass

    # 2) Make a copy of the chat context (or operate on a safe mutable copy)
    chat_ctx = self.chat_ctx.copy()

    # 3) Find the last assistant message that may contain the pre-function text.
    #    Adjust index if your flow inserts tool-call messages into the stream.
    #    This example looks back up to 3 messages to find the assistant entry.
    assistant_idx = None
    for i in range(len(chat_ctx.messages) - 1, max(-1, len(chat_ctx.messages) - 4), -1):
        m = chat_ctx.messages[i]
        if getattr(m, "role", None) == "assistant":
            assistant_idx = i
            break

    # 4) Annotate or replace that assistant message to signal it's already spoken.
    if assistant_idx is not None:
        old_msg = chat_ctx.messages[assistant_idx]
        # Option A: Replace with short annotation (safe and easy to debug)
        old_msg.content = ["[PREVIOUSLY SPOKEN] " + (old_msg.content[0] if isinstance(old_msg.content, (list, tuple)) else str(old_msg.content))]
        # Option B: or replace with a brief summary:
        # old_msg.content = ["Previously acknowledged the user; awaiting tool result."]

    # 5) Persist the modified chat context so generate_reply() sees the edited history
    await self.update_chat_ctx(chat_ctx)

    # 6) Trigger a focused continuation. Keep instructions explicit to avoid repetition.
    await self.session.generate_reply(
        instructions=(
            "You previously invoked a tool and the result is available. Continue the assistant's response "
            "from where it left off. Do not repeat greetings, acknowledgments, or earlier text already spoken. "
            "Use only the new tool output to add fresh information."
        ),
        allow_interruptions=True,
    )