https://livekit.io logo
Join Slack
Powered by
session.generate_reply doesn't work inside a funct...
# ask-ai
c
session.generate_reply doesn't work inside a function_tool, why?
t
Thanks for reaching out, Subham. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.
Thanks — I’m responding about why 
session.generate_reply
doesn’t work when called inside a 
function_tool
. Short answer: tool execution pauses the agent’s normal reply flow, so calling 
session.generate_reply
from inside a tool will conflict with the agent’s execution model and usually won’t produce a reply unless you explicitly manage session/context and control flow. What to try next: • Confirm you have access to the same 
session
and 
RunContext
inside the tool (pass 
context
into the tool). • Use the async call pattern and await it from the tool: 
await context.session.generate_reply()
so it runs in the correct async flow. • If the tool is long-running or must not block the agent, spawn a background task (outside the blocking path) that calls 
generate_reply
after the tool finishes. • Ensure the session is active and there are no concurrency conflicts (no other code is awaiting generation at the same time). Could you share: 1. A minimal code snippet of the tool and how you call it (how you pass `context`/`session`), and 2. Any error messages or logs you see when you call 
session.generate_reply
? With that I can suggest an exact code fix. Sources: Tool definition and use | LiveKit docs | Agent speech and audio | LiveKit docs | External data and RAG | LiveKit docs
c
Copy code
from __future__ import annotations

import re
from dataclasses import dataclass
from typing import TYPE_CHECKING

from ... import llm, stt, tts, vad
from ...llm.tool_context import ToolError, function_tool
from ...types import NOT_GIVEN, NotGivenOr
from ...voice.agent import AgentTask
from ...voice.events import RunContext
from ...voice.speech_handle import SpeechHandle

if TYPE_CHECKING:
    from ...voice.agent_session import TurnDetectionMode

EMAIL_REGEX = (
    r"^[A-Za-z0-9][A-Za-z0-9._%+\-]*@(?:[A-Za-z0-9](?:[A-Za-z0-9\-]*[A-Za-z0-9])?\.)+[A-Za-z]{2,}$"
)


@dataclass
class GetEmailResult:
    email_address: str


class GetEmailTask(AgentTask[GetEmailResult]):
    def __init__(
        self,
        chat_ctx: NotGivenOr[llm.ChatContext] = NOT_GIVEN,
        turn_detection: NotGivenOr[TurnDetectionMode | None] = NOT_GIVEN,
        stt: NotGivenOr[stt.STT | None] = NOT_GIVEN,
        vad: NotGivenOr[vad.VAD | None] = NOT_GIVEN,
        llm: NotGivenOr[llm.LLM | llm.RealtimeModel | None] = NOT_GIVEN,
        tts: NotGivenOr[tts.TTS | None] = NOT_GIVEN,
        allow_interruptions: NotGivenOr[bool] = NOT_GIVEN,
    ) -> None:
        super().__init__(
            instructions=(
                "You are only a single step in a broader system, responsible solely for capturing an email address.\n"
                "Handle input as noisy voice transcription. Expect that users will say emails aloud with formats like:\n"
                "- 'john dot doe at gmail dot com'\n"
                "- 'susan underscore smith at yahoo dot co dot uk'\n"
                "- 'dave dash b at protonmail dot com'\n"
                "- 'jane at example' (partial—prompt for the domain)\n"
                "- 'theo t h e o at livekit dot io' (name followed by spelling)\n"
                "Normalize common spoken patterns silently:\n"
                "- Convert words like 'dot', 'underscore', 'dash', 'plus' into symbols: `.`, `_`, `-`, `+`.\n"
                "- Convert 'at' to `@`.\n"
                "- Recognize patterns where users speak their name or a word, followed by spelling: e.g., 'john j o h n'.\n"
                "- Filter out filler words or hesitations.\n"
                "- Assume some spelling if contextually obvious (e.g. 'mike b two two' → mikeb22).\n"
                "Don't mention corrections. Treat inputs as possibly imperfect but fix them silently.\n"
                "Call `update_email_address` at the first opportunity whenever you form a new hypothesis about the email. "
                "(before asking any questions or providing any answers.) \n"
                "Don't invent new email addresses, stick strictly to what the user said. \n"
                "Call `confirm_email_address` after the user confirmed the email address is correct. \n"
                "If the email is unclear or invalid, or it takes too much back-and-forth, prompt for it in parts: first the part before the '@', then the domain—only if needed. \n"
                "Ignore unrelated input and avoid going off-topic. Do not generate markdown, greetings, or unnecessary commentary. \n"
                "Always explicitly invoke a tool when applicable. Do not simulate tool usage, no real action is taken unless the tool is explicitly called."
            ),
            chat_ctx=chat_ctx,
            turn_detection=turn_detection,
            stt=stt,
            vad=vad,
            llm=llm,
            tts=tts,
            allow_interruptions=allow_interruptions,
        )

        self._current_email = ""

        # speech_handle/turn used to update the email address.
        # used to ignore the call to confirm_email_address in case the LLM is hallucinating and not asking for user confirmation
        self._email_update_speech_handle: SpeechHandle | None = None

    async def on_enter(self) -> None:
        self.session.generate_reply(
            instructions=(
                "Ask the user to provide an email address. If you already have it, ask for confirmation.\n"
                "Do not call `decline_email_capture`"
            )
        )

    @function_tool
    async def update_email_address(self, email: str, ctx: RunContext) -> str:
        """Update the email address provided by the user.

        Args:
            email: The email address provided by the user
        """
        self._email_update_speech_handle = ctx.speech_handle
        email = email.strip()

        if not re.match(EMAIL_REGEX, email):
            raise ToolError(f"Invalid email address provided: {email}")

        self._current_email = email
        separated_email = " ".join(email)

        return (
            f"The email has been updated to {email}\n"
            f"Repeat the email character by character: {separated_email} if needed\n"
            f"Prompt the user for confirmation, do not call `confirm_email_address` directly"
        )

    @function_tool
    async def confirm_email_address(self, ctx: RunContext) -> None:
        """Validates/confirms the email address provided by the user."""
        await ctx.wait_for_playout()

        if ctx.speech_handle == self._email_update_speech_handle:
            raise ToolError("error: the user must confirm the email address explicitly")

        if not self._current_email.strip():
            raise ToolError(
                "error: no email address were provided, `update_email_address` must be called before"
            )

        if not self.done():
            self.complete(GetEmailResult(email_address=self._current_email))

    @function_tool
    async def decline_email_capture(self, reason: str) -> None:
        """Handles the case when the user explicitly declines to provide an email address.

        Args:
            reason: A short explanation of why the user declined to provide the email address
        """
        if not self.done():
            self.complete(ToolError(f"couldn't get the email address: {reason}"))
t
Thanks — I’m responding about why 
session.generate_reply
doesn’t work when called inside a 
function_tool
, and how to fix it in your GetEmailTask. Why it fails • Tool execution pauses the agent’s normal reply flow. While a 
function_tool
is running, the agent’s main message loop is waiting for the tool to return. Calling 
session.generate_reply
inside that paused flow conflicts with the agent execution model and typically won’t emit a reply. • The agent won’t automatically resume or create new replies from inside a tool unless you explicitly run 
generate_reply
on the active session/context in the correct async flow or spawn a separate task that runs after the tool returns. What to change in your code (high level) • Don’t call 
session.generate_reply
directly from inside a 
function_tool
synchronous path. Instead: ◦ If you need to trigger a response immediately after the tool finishes, return a message from the tool and let the agent handle calling 
generate_reply
after the tool returns. ◦ If you must trigger a reply from inside the tool, call 
await ctx.session.generate_reply(...)
using the RunContext passed into the tool and ensure you 
await
it in the tool’s async context. That uses the correct async flow attached to the current run context. ◦ For long-running or non-blocking updates (e.g., incremental feedback while processing), spawn a background task that calls 
generate_reply
so the tool’s primary logic can finish and return normally. Targeted suggestions for your GetEmailTask • update_email_address: This should not try to cause immediate agent speech via 
session.generate_reply
. Return the update text (which you already do). Let the agent produce a reply after the tool returns so the agent’s reply flow remains consistent. • confirm_email_address: You already 
await ctx.wait_for_playout()
which is correct for waiting on playout. Do not call 
session.generate_reply
here either; complete the task and let the agent respond. • If you need the agent to speak immediately when 
update_email_address
occurs, change the tool to: a. Accept 
ctx: RunContext
. b. After updating state, call 
await ctx.session.generate_reply(instructions=...)
(awaiting ensures correct async flow). c. Or better: return the text and let the outer agent generate the reply — simpler and less error-prone. Example patterns (pseudo) • Preferred (return from tool): ◦ Tool returns a string describing update. ◦ Agent receives tool result and calls 
session.generate_reply(...)
after tool completes. • If calling from inside tool (use RunContext): ◦ 
await ctx.session.generate_reply(instructions="...")
— ensure you await and that no other concurrent generate is running. • Non-blocking feedback: ◦ Spawn background task that does 
await session.generate_reply(...)
after the tool returns. One short checklist to debug now • Are you passing 
ctx: RunContext
into the tool and using 
ctx.session
(not 
self.session
)? • Are you awaiting 
ctx.session.generate_reply()
inside the tool? • Do logs show any concurrency errors when 
generate_reply
is invoked? • If you need immediate speech, can you instead return the message and let the agent call 
generate_reply
after the tool returns? If you want, share: • the code where you call the tool (how RunContext / session is passed), and • any error logs you see when you try to call 
generate_reply
. I’ll give a precise code patch based on that.
c
Copy code
@function_tool
    async def get_users_email(self, context: RunContext, reason: str | None = None):
        "Get the user's email from the user's input."

        email_result = await beta.workflows.GetEmailTask()
        email_address = email_result.email_address

        <http://logger.info|logger.info>(f"User's email address: {email_address}")

        return f"The user's email address is {email_address}. "
give me the details of all the required changs
t
Thanks — I’m responding about the changes required to make your 
get_users_email
tool work correctly with 
GetEmailTask
and with reply generation. I’ll list concrete code changes and why, plus a minimal example patch you can apply. Summary of required changes • Fix tool signature: put 
ctx: RunContext
first and name it 
ctx
(tools receive a RunContext). • Await task creation correctly: call and await the task runner instead of directly awaiting the class. • Use the task’s API to run it (create the task instance and call its runner, or call the workflow runner you use across your codebase). • Don’t call 
self.session.generate_reply
inside a tool; if you need immediate agent speech from inside the tool, call 
await ctx.session.generate_reply(...)
and await it. • Add error handling for ToolError and validation failures from 
GetEmailTask
. • Return the tool result string (or call 
ctx.session.generate_reply
then return) — let the agent handle normal reply flow unless you intentionally call 
ctx.session.generate_reply
. • Ensure imports and typing match your project (RunContext, ToolError, GetEmailTask). Concrete changes and rationale 1) Signature and param order • Change 
async def get_users_email(self, context: RunContext, reason: str | None = None)
to: ◦ 
async def get_users_email(self, ctx: RunContext, reason: str | None = None)
◦ Tool frameworks expect the RunContext as the first parameter; naming it 
ctx
matches earlier code and examples. 2) Run the GetEmailTask correctly • Current: 
email_result = await beta.workflows.GetEmailTask()
◦ Problem: you’re awaiting the class or not invoking the task runner. You must instantiate the task and run it via your agent/session API. • Correct pattern (two common options): ◦ If your codebase provides a helper to run tasks (preferred), use it: ▪︎ 
email_task = beta.workflows.GetEmailTask(...)
▪︎ 
email_result = await self.run_task(email_task, ctx)
# replace with your task-runner call ◦ Or instantiate and run via session/workflow API: ▪︎ 
email_task = beta.workflows.GetEmailTask(...)
▪︎ 
email_result = await ctx.session.run_task(email_task)
# example — replace with your actual runner 3) Await properly and extract result • After running the task, access 
email_result.email_address
. • Validate before using. 4) Error handling • Wrap the run in try/except catching 
ToolError
and other exceptions: ◦ On ToolError: return a helpful message string or re-raise as a ToolError so the agent can handle it. ◦ On unexpected exceptions: log and return a generic failure string. 5) Reply generation choices • Preferred (simpler, safer): return a descriptive string and let the agent generate the reply after the tool returns. ◦ Example: 
return f"The user's email address is {email_address}."
• If you need the agent to speak immediately from inside the tool: ◦ Call 
await ctx.session.generate_reply(instructions="...")
and await it; then return or finish. ◦ Ensure no concurrent generate is running. 6) Logging and privacy • Keep logging minimal and avoid logging full email addresses in production. For debugging, you can log masked emails. Minimal patched example (apply replacements where your task-runner API differs)
Copy code
@function_tool
async def get_users_email(self, ctx: RunContext, reason: str | None = None):
    "Get the user's email from the user's input."

    try:
        # instantiate the task
        email_task = beta.workflows.GetEmailTask()
        # run the task via your agent/session runner — replace `run_task` with your project's runner API
        email_result = await self.run_task(email_task, ctx)
        # OR, if your API exposes session.run_task:
        # email_result = await ctx.session.run_task(email_task)

        email_address = email_result.email_address

        # optional: immediate speech from inside the tool (only if needed)
        # await ctx.session.generate_reply(instructions=f"The user's email is {email_address}. Is this correct?")

        # Prefer returning a string so the agent handles reply generation
        return f"The user's email address is {email_address}."
    except ToolError as te:
        # propagate or return a clear message so agent can handle
        return f"Could not capture email: {str(te)}"
    except Exception as e:
        logger.exception("Unexpected error capturing email")
        return "Sorry, I couldn't capture the email due to an internal error."
Checklist to verify after changes • RunContext is the first parameter and named 
ctx
. • You instantiate the task and run it with the proper runner API (don’t await the class). • You await the task run and extract 
email_address
. • You handle ToolError and other exceptions. • You either return a string (preferred) or use 
await ctx.session.generate_reply(...)
if immediate speech is required. • No direct calls to 
self.session.generate_reply
inside the tool—use 
ctx.session
if inside a tool. If you share how tasks are normally run in your codebase (the exact function you use to run AgentTask instances — e.g., 
self.run_task(...)
or 
ctx.session.run_task(...)
), I’ll give the exact one-line replacement and a ready-to-drop patch.
2 Views
Open in Slack
PreviousNext
Powered by