Function calling with Voice Agents

import aiohttp
from typing import Annotated

from livekit.agents import llm
from livekit.agents.pipeline import VoicePipelineAgent
from livekit.agents.multimodal import MultimodalAgent

# first define a class that inherits from llm.FunctionContext
class AssistantFnc(llm.FunctionContext):
    # the llm.ai_callable decorator marks this function as a tool available to the LLM
    # by default, it'll use the docstring as the function's description
    @llm.ai_callable()
    async def get_weather(
        self,
        # by using the Annotated type, arg description and type are available to the LLM
        location: Annotated[
            str, llm.TypeInfo(description="The location to get the weather for")
        ],
    ):
        """Called when the user asks about the weather. This function will return the weather for the given location."""
        logger.info(f"getting weather for {location}")
        url = f"https://wttr.in/{location}?format=%C+%t"
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as response:
                if response.status == 200:
                    weather_data = await response.text()
                    # response from the function call is returned to the LLM
                    # as a tool response. The LLM's response will include this data
                    return f"The weather in {location} is {weather_data}."
                else:
                    raise f"Failed to get weather data, status code: {response.status}"

fnc_ctx = AssistantFnc()

# pass the function context to the agent
pipeline_agent = VoicePipelineAgent(
    ...
    fnc_ctx=fnc_ctx,
)

multimodal_agent = MultimodalAgent(
    ...
    fnc_ctx=fnc_ctx,
)

import { llm, multimodal } from '@livekit/agents';
import { z } from 'zod';

// first define an llm.FunctionContext object
const fncCtx: llm.FunctionContext = {
    // Each function is listed by name, making it available to the LLM
    weather: {
        // Description is passed to the LLM
        description: 'Get the weather in a location',

        // Use [zod](https://zod.dev/) for parameter typing and automatic schema generation for the LLM
        parameters: z.object({
            location: z.string().describe('The location to get the weather for'),
        }),

        // This function becomes available as a tool for the LLM, and will be called with the parameters defined above
        execute: async ({ location }) => {
            console.info(`getting weather for ${location}`);
            const url = `https://wttr.in/${location}?format=%C+%t`;
            const response = await fetch(url);
            if (!response.ok) {
                throw new Error(`Failed to get weather data, status code: ${response.status}`);
            }

            const weatherData = await response.text();
            // response from the function call is returned to the LLM
            // as a tool response. The LLM's response will include this data
            return `The weather in ${location} is ${weatherData}.`;
        },
    }
};

// pass the function context to the agent
const agent = new multimodal.MultimodalAgent({
    ...
    fncCtx,
});

from livekit.agents import llm
from typing import Annotated

class AssistantFnc(llm.FunctionContext):
    @llm.ai_callable()
    async def get_user_location(self,
        high_accuracy: Annotated[
            bool, llm.TypeInfo(description="Whether to use high accuracy mode, which is slower")
        ] = False
    ):
        """Retrieve the user's current geolocation as lat/lng."""
        try:
            return await ctx.room.local_participant.perform_rpc(
                destination_identity=participant.identity,
                method="getUserLocation",
                payload=json.dumps({
                    "highAccuracy": high_accuracy
                }),
                response_timeout=10.0 if high_accuracy else 5.0,
            )
        except Exception:
            return "Unable to retrieve user location"

import { z } from 'zod';
import type { llm } from '@livekit/agents';

const fncCtx: llm.FunctionContext = {
    getUserLocation: {
        description: "Retrieve the user's current geolocation as lat/lng.",
        parameters: z.object({
            highAccuracy: z
                .boolean()
                .describe("Whether to use high accuracy mode, which is slower"),
        }),
        execute: async (params) => {
            try {
                return await ctx.room.localParticipant!.performRpc({
                    destinationIdentity: participant.identity,
                    method: "getUserLocation",
                    payload: JSON.stringify(params),
                    responseTimeout: params.highAccuracy ? 10000 : 5000,
                });
            } catch (error) {
                return "Unable to retrieve user location";
            }
        }
    }
}

import { RpcError, RpcInvocationData } from 'livekit-client';

localParticipant.registerRpcMethod(
    'getUserLocation',
    async (data: RpcInvocationData) => {
        try {
            let params = JSON.parse(data.payload);
            const position: GeolocationPosition = await new Promise((resolve, reject) => {
                navigator.geolocation.getCurrentPosition(resolve, reject, {
                    enableHighAccuracy: params.highAccuracy ?? false,
                    timeout: data.responseTimeout,
                });
            });

            return JSON.stringify({
                latitude: position.coords.latitude,
                longitude: position.coords.longitude,
            });
        } catch (error) {
            throw new RpcError(1, "Could not retrieve user location");
        }
    }
);