bumpy-student-61140
02/18/2026, 10:29 AMtall-airline-16756
02/18/2026, 10:29 AMtall-airline-16756
02/18/2026, 10:29 AMcontext.session.generate_reply or session.say) after a small delay.
• Meanwhile the function continues the long‑running operation.
• If the operation finishes quickly, cancel the scheduled status message so you don’t speak unnecessarily.
• This decouples speech from waiting on the tool result and gives users a proactive “I’m checking that now…” message.
Example pattern (Python)
1. Create a short background task that waits a bit (e.g., 0.5s) and then triggers a spoken update.
2. Start that task, perform the long operation, cancel the task if the operation completes quickly, then return the result.
import asyncio
from livekit.agents import function_tool, RunContext
@function_tool()
async def search_knowledge_base(self, context: RunContext, query: str) -> str:
async def _speak_status_update(delay: float = 0.5):
await asyncio.sleep(delay)
await context.session.generate_reply(instructions=f"""
You are searching the knowledge base for "{query}" but it is taking a little while.
Update the user on your progress, but be very brief.
""")
status_update_task = asyncio.create_task(_speak_status_update(0.5))
result = await _perform_search(query) # your long-running call
status_update_task.cancel()
return result
Notes and best practices
• Use a short delay (0.3–0.7s) so the status message only triggers on noticeably long operations.
• Cancel the background task when the result is ready to avoid duplicate or unnecessary speech.
• The same concept is supported in Node.js using equivalent async/task utilities.
• Optionally combine verbal updates with short background “thinking” audio for a more natural UX.
If you’d like, I can:
• Provide a Node.js example that uses Task.from() or equivalent, or
• Help adapt this pattern to a specific function you have (share the function signature or a brief snippet). Sources: External data and RAG | LiveKit Documentationbumpy-student-61140
02/18/2026, 10:41 AMtall-airline-16756
02/18/2026, 10:41 AM@function_tool implementation:
import logging
from livekit.agents import function_tool, RunContext
# configure logging to write to a file
logging.basicConfig(filename='function_calls.txt', level=<http://logging.INFO|logging.INFO>, format='%(asctime)s %(message)s')
logger = logging.getLogger(__name__)
@function_tool()
async def search_knowledge_base(self, context: RunContext, query: str) -> str:
<http://logger.info|logger.info>(f'Function called: search_knowledge_base(query={query!r})') # capture invocation
# Optionally send an early spoken update to users (see "Speak while waiting" below)
# await context.session.generate_reply(instructions="Please wait while I look that up.")
# or await context.session.say("Please wait while I look that up.")
result = await _perform_search(query) # your long-running call
<http://logger.info|logger.info>(f'Function response: {result!r}') # capture output
return result
1. The logging.basicConfig(filename=...) line writes entries into function_calls.txt. Each invocation will add a timestamped line for the call and a line for the response.
Node.js approach (concept)
• Use the same idea: log the function name, args, and response to a file (e.g., with fs.appendFile) inside your tool implementation.
• If using an async task helper (like Task.from()), log around the awaited call so you capture both the start and completion.
Testing & programmatic inspection
• Use test utilities like mock_tools() and assertions (if you are unit-testing agents) to assert and inspect that a function was called with certain arguments and check returned values programmatically. This is useful for automated validation before relying on the text logs.
Make the agent speak while the function runs
• Send a brief spoken message early in the function (or schedule a background verbal update) so the user hears a status update while the tool runs.
• Pattern: schedule a short delayed announcement (0.3–0.7s) and cancel it if the function finishes quickly. That prevents unnecessary spoken messages for fast calls.
Example: schedule-and-cancel pattern (Python)
import asyncio
async def _speak_status_update():
await asyncio.sleep(0.5)
await context.session.generate_reply(instructions=f'Looking up "{query}". I will be right back with the result.')
status_task = asyncio.create_task(_speak_status_update())
result = await _perform_search(query)
status_task.cancel()
Tips for using the logs to tune speech behavior
• Inspect the timestamps in your log file to measure how long calls actually take; use that to choose your delay threshold for the status message.
• Log both start and end times or include durations to identify which functions are frequently slow.
• If you log the generated spoken messages as well, you can correlate what users heard with the function timing and output.
• For privacy/PII, avoid logging sensitive user data or redact it before writing to disk.
If you want, I can:
• Provide a full Node.js example that appends calls/results to a .txt file.
• Help adapt the Python pattern to a specific function you’re using (share the function signature or a small snippet).
• Suggest a format for the log file (e.g., JSON lines) that’s easier to parse for analytics. Which option would be most helpful? Sources: SIP Lifecycle Management Agent | LiveKit Documentation | Testing and evaluation | LiveKit Documentation | Playing Audio | LiveKit Documentation