Streaming et outils

Streaming avec function calling

Le streaming permet de recevoir la réponse du modèle progressivement, token par token. C’est essentiel pour les interfaces utilisateur réactives. Mais comment le streaming interagit-il avec le function calling ?

La règle fondamentale

Les appels de fonctions sont retournés en un seul chunk, pas streamés progressivement. C’est une différence majeure avec le texte normal, qui arrive token par token.

Concrètement :

• Le texte de la réponse est streamé progressivement
• Les tool_calls arrivent d’un bloc dans un seul événement

Cette contrainte existe pour une bonne raison : un tool_call doit être complet (nom + arguments JSON valides) pour que vous puissiez l’exécuter. Un JSON partiel serait inutilisable.

Implémenter le streaming avec outils

stream = client.chat.completions.create(
    model="grok-3",
    messages=messages,
    tools=tools,
    stream=True
)

collected_tool_calls = []
collected_content = ""

for chunk in stream:
    delta = chunk.choices[0].delta

    # Texte streamé
    if delta.content:
        collected_content += delta.content
        print(delta.content, end="", flush=True)

    # Tool calls (arrivent en un bloc)
    if delta.tool_calls:
        for tc in delta.tool_calls:
            collected_tool_calls.append(tc)

    # Vérifier la fin du stream
    if chunk.choices[0].finish_reason == "tool_calls":
        # Le modèle attend des résultats d'outils
        break
    elif chunk.choices[0].finish_reason == "stop":
        # Réponse terminée, pas d'outil appelé
        break

Reconstruction des tool_calls en streaming

Dans certains cas, les arguments d’un tool_call peuvent arriver en plusieurs morceaux dans le stream. Vous devez les reconstruire :

tool_calls_buffer = {}

for chunk in stream:
    delta = chunk.choices[0].delta

    if delta.tool_calls:
        for tc_chunk in delta.tool_calls:
            idx = tc_chunk.index

            if idx not in tool_calls_buffer:
                tool_calls_buffer[idx] = {
                    "id": tc_chunk.id,
                    "function": {
                        "name": tc_chunk.function.name or "",
                        "arguments": ""
                    }
                }

            if tc_chunk.function.name:
                tool_calls_buffer[idx]["function"]["name"] = tc_chunk.function.name
            if tc_chunk.function.arguments:
                tool_calls_buffer[idx]["function"]["arguments"] += tc_chunk.function.arguments

Cette approche de buffer garantit que vous reconstruisez correctement chaque tool_call, même si les données arrivent fragmentées.

Architecture pour le streaming avec outils

Dans une application réelle, vous devez gérer deux flux simultanés :

1. Affichage en temps réel : montrer le texte au fur et à mesure
2. Accumulation des tool_calls : collecter les appels pour exécution

async def handle_streaming_response(stream):
    content_parts = []
    tool_calls = []
    current_tool = None

    async for chunk in stream:
        delta = chunk.choices[0].delta

        if delta.content:
            content_parts.append(delta.content)
            yield {"type": "text", "data": delta.content}

        if delta.tool_calls:
            for tc in delta.tool_calls:
                tool_calls.append(tc)

        if chunk.choices[0].finish_reason == "tool_calls":
            # Exécuter les outils puis relancer
            results = await execute_tools(tool_calls)
            yield {"type": "tool_results", "data": results}

Latence et expérience utilisateur

Le streaming avec outils implique des pauses naturelles dans le flux :

1. L’utilisateur pose sa question
2. Le modèle commence à réfléchir (pas de token visible)
3. Le modèle décide d’appeler un outil (vous recevez le tool_call)
4. Pause — vous exécutez la fonction
5. Vous renvoyez le résultat
6. Le modèle génère sa réponse (streamée progressivement)

Pour l’expérience utilisateur, affichez un indicateur pendant l’étape 4 :

🔍 Recherche de la météo à Paris...

Cela évite que l’utilisateur pense que l’application est bloquée.

Server-Sent Events (SSE) pour les applications web

Si vous construisez une API web, utilisez les SSE pour relayer le stream :

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

@app.get("/chat")
async def chat(message: str):
    async def event_stream():
        stream = client.chat.completions.create(
            model="grok-3",
            messages=[{"role": "user", "content": message}],
            tools=tools,
            stream=True
        )
        for chunk in stream:
            if chunk.choices[0].delta.content:
                yield f"data: {chunk.choices[0].delta.content}\n\n"

    return StreamingResponse(event_stream(), media_type="text/event-stream")

Points clés à retenir

• Les tool_calls arrivent en un seul chunk, pas streamés progressivement
• Le texte normal continue d’être streamé token par token
• Reconstruisez les tool_calls fragmentés avec un buffer indexé par index
• Gérez deux flux en parallèle : affichage texte + collecte des tool_calls
• Affichez un indicateur pendant l’exécution des outils pour l’UX
• Utilisez les SSE pour les applications web