Streaming granulaire des appels d'outils

En streaming, les arguments d’un appel d’outil arrivent par fragments, comme le texte. Par défaut, l’API retient chaque argument jusqu’à l’avoir validé, ce qui produit des pauses suivies de rafales. Le streaming granulaire (fine-grained tool streaming) supprime cette attente, au prix de la validation. Cette leçon explique le compromis, et la manière d’en profiter sans risque.

Les fragments d’arguments

Un bloc tool_use diffusé en streaming commence par un événement content_block_start dont input vaut {} : un simple emplacement. Les arguments arrivent ensuite dans des événements input_json_delta, chacun portant un fragment de texte JSON dans partial_json. On concatène les fragments, et on analyse le tout à la fermeture du bloc (content_block_stop).

L’assistant de streaming du SDK Python fait ce travail pour vous. Pour chaque fragment, il produit aussi un événement simplifié de type input_json, avec deux champs :

• partial_json : le fragment reçu, sous forme de chaîne ;
• snapshot : l’objet reconstitué jusque-là, déjà analysé par le SDK.

with client.messages.stream(
    model=model,
    max_tokens=16000,
    messages=messages,
    tools=[save_article_schema],
) as stream:
    for event in stream:
        if event.type == "input_json":
            print(event.partial_json, end="", flush=True)
    final_message = stream.get_final_message()

snapshot est un objet Python, pas une chaîne : il ne faut pas le repasser à json.loads, comme le faisait l’ancienne version de cette leçon.

Par défaut : valider, puis envoyer

Par défaut, l’API met en mémoire tampon chaque valeur de paramètre et la valide avant de la transmettre. Pour un petit argument comme {"location": "Lyon"}, la différence ne se voit pas. Pour un argument long, un document ou un bloc de code, rien n’arrive tant que la valeur entière n’est pas générée : le flux semble figé, puis tout arrive d’un coup. Ce n’est pas votre code, c’est la validation de l’API.

Le streaming granulaire : envoyer tout de suite

On l’active outil par outil, avec "eager_input_streaming": True dans la définition, sur une requête en streaming :

save_article_schema = ToolParam({
    "name": "save_article",
    "description": (
        "Enregistre un article : son résumé et ses métadonnées. "
        "Utilisez cet outil quand l'utilisateur demande d'archiver un article qu'il vient de fournir. "
        "Le résumé tient en un paragraphe ; la relecture signale les points faibles de l'article."
    ),
    "eager_input_streaming": True,
    "input_schema": {
        "type": "object",
        "properties": {
            "abstract": {"type": "string", "description": "Résumé de l'article, en un paragraphe."},
            "meta": {
                "type": "object",
                "properties": {
                    "word_count": {"type": "integer", "description": "Nombre de mots de l'article."},
                    "review": {"type": "string", "description": "Relecture critique en quelques phrases."},
                },
                "required": ["word_count", "review"],
            },
        },
        "required": ["abstract", "meta"],
    },
})

Les fragments arrivent alors dès leur génération, sans attente entre les paramètres, et le premier arrive presque tout de suite. Ce champ par outil remplace l’en-tête bêta fine-grained-tool-streaming-2025-05-14 des premières versions. La fonctionnalité est stable sur la Claude API pour tous les modèles actuels ; sur une plateforme cloud, vérifiez sa disponibilité pour votre déploiement.

Ce que vous perdez : la validation

Sans tampon, l’API ne valide plus rien. Le JSON accumulé peut être invalide, par exemple "word_count": undefined au lieu d’un nombre, ou coupé net si la réponse atteint max_tokens au milieu d’un argument. En mode par défaut, la même valeur fautive vous serait arrivée sous forme de chaîne de caractères : le mode granulaire ne crée pas l’erreur, il la déplace chez vous. Trois précautions :

1. Rattraper l’erreur d’analyse. Quand le JSON est trop abîmé pour être analysé, le SDK Python lève une ValueError depuis l’itération du flux. Entourez le bloc with d’un try, ne rattrapez que cette erreur, et relancez un nombre limité de fois. Les erreurs d’API, comme un dépassement de quota, ne sont pas des ValueError et doivent remonter.
2. Vérifier stop_reason. S’il vaut max_tokens alors qu’un bloc tool_use est présent, l’argument est probablement tronqué : n’exécutez pas l’outil, relancez avec un max_tokens plus grand. S’il vaut refusal, n’exécutez rien.
3. Valider les arguments avant d’exécuter l’outil. Le SDK reconstitue les arguments avec un analyseur tolérant : une entrée mal formée peut revenir sous forme d’objet tronqué, sans erreur. Comparez-la au schéma, ici avec Pydantic 2.

import json
from pydantic import BaseModel, ValidationError

class Meta(BaseModel):
    word_count: int
    review: str

class SaveArticle(BaseModel):
    abstract: str
    meta: Meta

def save_article(abstract, meta):
    ...  # enregistrement en base
    return "Article enregistré."

def stream_with_retry(messages, tools, retries=2):
    for _ in range(retries + 1):
        try:
            with client.messages.stream(model=model, max_tokens=16000, messages=messages, tools=tools) as stream:
                for event in stream:
                    if event.type == "input_json":
                        print(event.partial_json, end="", flush=True)
                return stream.get_final_message()
        except ValueError as e:  # arguments impossibles à analyser
            print(f"\nArguments illisibles, nouvel essai : {e}")
    raise RuntimeError("arguments d'outil illisibles après plusieurs essais")

response = stream_with_retry(messages, [save_article_schema])
if response.stop_reason in ("max_tokens", "refusal"):
    raise RuntimeError(f"réponse interrompue ({response.stop_reason}) : aucun outil exécuté")

tool_results = []
for block in response.content:
    if block.type != "tool_use":
        continue
    try:
        args = SaveArticle.model_validate(block.input)
    except ValidationError:
        tool_results.append({
            "type": "tool_result",
            "tool_use_id": block.id,
            "is_error": True,
            "content": json.dumps({"INVALID_JSON": json.dumps(block.input, ensure_ascii=False)}),
        })
        continue
    tool_results.append({
        "type": "tool_result",
        "tool_use_id": block.id,
        "content": save_article(**args.model_dump()),
    })

Quand les arguments ne passent pas la validation, on n’exécute pas l’outil : on renvoie l’entrée reçue à Claude dans un tool_result marqué en erreur, sous une clé unique INVALID_JSON, comme le propose la documentation. Claude comprend que son appel était mal formé et le reformule. Construisez cette enveloppe avec json.dumps : une concaténation de chaînes n’échapperait pas les guillemets.

Quand l’activer

Activez-le pour un outil dont un argument est long et que vous voulez afficher ou traiter au fil de l’eau : le contenu d’un fichier, un document, du code. Pour des arguments courts, le gain est invisible, et la validation de l’API vaut mieux que la vôtre.

À vous de jouer

Voir la différence. Avec l’outil save_article, demandez en streaming la fiche d’un long article, d’abord sans eager_input_streaming, puis avec. Notez le délai avant le premier fragment et la régularité du flux. Provoquez ensuite une coupure en abaissant max_tokens à 300, et vérifiez que votre code refuse d’exécuter l’outil.

Sources officielles : Fine-grained tool calling (Claude Academy) · Fine-grained tool streaming (documentation Claude : activation, accumulation, JSON invalide) · Streaming messages (événements input_json_delta)

Points clés à retenir

• Par défaut, l’API valide chaque valeur de paramètre avant de l’envoyer : des pauses, puis des rafales.
• eager_input_streaming: true, outil par outil, envoie les fragments dès leur génération.
• Sans validation côté API : rattrapez la ValueError du SDK, vérifiez stop_reason, validez les arguments.
• Arguments invalides : pas d’exécution, un tool_result en erreur avec l’entrée sous INVALID_JSON.
• snapshot est déjà un objet analysé ; ne le repassez pas à json.loads.