Function calling en temps réel

Function calling en temps réel

Le function calling dans la Realtime API permet à votre agent vocal d’exécuter des actions concrètes pendant la conversation : consulter une base de données, passer une commande, vérifier un statut. La mécanique est similaire à celle de la Responses API, mais adaptée au flux événementiel WebSocket.

Déclarer des outils dans la session

Les outils sont déclarés dans session.update, exactement comme les instructions et la voix. Chaque outil est un schéma JSON décrivant ses paramètres :

tools = [
    {
        "type": "function",
        "name": "get_weather",
        "description": "Obtenir la météo actuelle pour une ville donnée",
        "parameters": {
            "type": "object",
            "properties": {
                "city": {
                    "type": "string",
                    "description": "Nom de la ville"
                },
                "unit": {
                    "type": "string",
                    "enum": ["celsius", "fahrenheit"],
                    "description": "Unité de température"
                }
            },
            "required": ["city"]
        }
    },
    {
        "type": "function",
        "name": "book_appointment",
        "description": "Réserver un rendez-vous",
        "parameters": {
            "type": "object",
            "properties": {
                "date": {"type": "string", "description": "Date au format YYYY-MM-DD"},
                "time": {"type": "string", "description": "Heure au format HH:MM"},
                "service": {"type": "string", "description": "Type de service"}
            },
            "required": ["date", "time", "service"]
        }
    }
]

await ws.send(json.dumps({
    "type": "session.update",
    "session": {
        "modalities": ["text", "audio"],
        "instructions": "Vous êtes un assistant vocal. Utilisez les outils disponibles.",
        "voice": "alloy",
        "tools": tools,
        "tool_choice": "auto"
    }
}))

Le flux d’un appel de fonction

Quand le modèle décide d’appeler un outil, la séquence événementielle est la suivante :

Implémenter le gestionnaire de fonctions

Voici un exemple complet de gestion des appels de fonctions :

import json

# Registre des fonctions disponibles
async def get_weather(city: str, unit: str = "celsius") -> dict:
    # Appel à une API météo réelle
    return {"city": city, "temperature": 22, "unit": unit, "condition": "ensoleillé"}

async def book_appointment(date: str, time: str, service: str) -> dict:
    # Logique de réservation
    return {"confirmed": True, "date": date, "time": time, "reference": "RDV-2026-0042"}

FUNCTION_REGISTRY = {
    "get_weather": get_weather,
    "book_appointment": book_appointment,
}

async def handle_function_call(ws, event):
    """Traite un appel de fonction et renvoie le résultat."""
    response = event["response"]

    for output in response.get("output", []):
        if output["type"] != "function_call":
            continue

        func_name = output["name"]
        call_id = output["call_id"]
        arguments = json.loads(output["arguments"])

        print(f"Appel fonction : {func_name}({arguments})")

        # Exécuter la fonction
        func = FUNCTION_REGISTRY.get(func_name)
        if func:
            result = await func(**arguments)
        else:
            result = {"error": f"Fonction inconnue : {func_name}"}

        # Renvoyer le résultat au modèle
        await ws.send(json.dumps({
            "type": "conversation.item.create",
            "item": {
                "type": "function_call_output",
                "call_id": call_id,
                "output": json.dumps(result)
            }
        }))

    # Demander au modèle de continuer avec le résultat
    await ws.send(json.dumps({
        "type": "response.create"
    }))

Intégrer dans la boucle d’événements

Le gestionnaire de fonctions s’intègre dans votre boucle principale :

async def event_loop(ws):
    """Boucle principale de traitement des événements."""
    async for raw_message in ws:
        event = json.loads(raw_message)
        event_type = event["type"]

        if event_type == "response.done":
            response = event["response"]
            # Vérifier si le modèle a appelé des fonctions
            has_function_calls = any(
                o["type"] == "function_call"
                for o in response.get("output", [])
            )
            if has_function_calls:
                await handle_function_call(ws, event)

        elif event_type == "response.audio.delta":
            # Jouer l'audio de réponse
            audio_bytes = base64.b64decode(event["delta"])
            player.write(audio_bytes)

        elif event_type == "error":
            print(f"Erreur : {event['error']['message']}")

Appels de fonctions multiples

Le modèle peut décider d’appeler plusieurs fonctions dans une même réponse. Le champ output de response.done peut contenir plusieurs entrées de type function_call. Traitez-les toutes avant d’appeler response.create.

Points clés à retenir

• Les outils sont déclarés dans session.update avec un schéma JSON standard
• Les arguments arrivent en streaming via response.function_call_arguments.delta
• Après exécution, renvoyez le résultat via conversation.item.create avec le call_id
• Appelez response.create pour que le modèle formule sa réponse vocale avec le résultat
• Gérez les appels multiples : le modèle peut appeler plusieurs fonctions à la fois