Streaming d'outils à granularité fine

Le « fine-grained tool streaming » (streaming d'outils à granularité fine) est disponible sur tous les modèles et toutes les plateformes. Il permet le streaming des valeurs de paramètres d'utilisation d'outils sans mise en mémoire tampon ni validation JSON, réduisant ainsi la latence avant de commencer à recevoir des paramètres volumineux.



Comment utiliser le streaming d'outils à granularité fine

Le streaming d'outils à granularité fine est pris en charge sur l'API Claude, Claude Platform sur AWS, Amazon Bedrock, Vertex AI et Microsoft Foundry. Pour l'utiliser, définissez eager_input_streaming sur true pour tout outil défini par l'utilisateur sur lequel vous souhaitez activer le streaming à granularité fine, et activez le streaming sur votre requête.

Voici un exemple d'utilisation du streaming d'outils à granularité fine avec l'API :

client = anthropic.Anthropic()

with client.messages.stream(
    max_tokens=65536,
    model="claude-opus-4-8",
    tools=[
        {
            "name": "make_file",
            "description": "Write text to a file",
            "eager_input_streaming": True,
            "input_schema": {
                "type": "object",
                "properties": {
                    "filename": {
                        "type": "string",
                        "description": "The filename to write text to",
                    },
                    "lines_of_text": {
                        "type": "array",
                        "description": "An array of lines of text to write to the file",
                    },
                },
                "required": ["filename", "lines_of_text"],
            },
        }
    ],
    messages=[
        {
            "role": "user",
            "content": "Can you write a long poem and make a file called poem.txt?",
        }
    ],
) as stream:
    final_message = stream.get_final_message()

print(f"Input tokens: {final_message.usage.input_tokens}")
print(f"Output tokens: {final_message.usage.output_tokens}")

Dans cet exemple, le streaming d'outils à granularité fine permet à Claude de diffuser les lignes d'un long poème dans l'appel d'outil make_file sans mise en mémoire tampon pour valider si le paramètre lines_of_text est un JSON valide. Cela signifie que vous pouvez voir le paramètre arriver en streaming au fur et à mesure, sans avoir à attendre que l'intégralité du paramètre soit mise en mémoire tampon et validée.



Accumulation des deltas d'entrée d'outil

Lorsqu'un bloc de contenu tool_use est diffusé en streaming, l'événement initial content_block_start contient input: {} (un objet vide). Il s'agit d'un espace réservé. L'entrée réelle arrive sous forme d'une série d'événements input_json_delta, chacun transportant un fragment de chaîne partial_json. Pour assembler l'entrée complète, concaténez ces fragments et analysez le résultat lorsque le bloc se ferme.

Lorsque votre SDK fournit un utilitaire d'accumulation (comme utilisé dans le premier exemple de cette page), celui-ci gère cela pour vous. Le modèle manuel est destiné aux SDK sans utilitaire, ou lorsque vous devez réagir à une entrée partielle avant la fermeture du bloc.

Le contrat d'accumulation :

1. Sur content_block_start avec type: "tool_use", initialisez une chaîne vide : input_json = ""
2. Pour chaque content_block_delta avec type: "input_json_delta", ajoutez : input_json += event.delta.partial_json
3. Sur content_block_stop, analysez la chaîne accumulée : json.loads(input_json)

L'incompatibilité de type entre le input: {} initial (objet) et partial_json (chaîne) est intentionnelle. L'objet vide marque l'emplacement dans le tableau de contenu ; les chaînes delta construisent la valeur réelle.

client = anthropic.Anthropic()

tool_inputs: dict[int, str] = {}  # index -> accumulated JSON string

with client.messages.stream(
    model="claude-opus-4-8",
    max_tokens=1024,
    tools=[
        {
            "name": "get_weather",
            "description": "Get current weather for a city",
            "eager_input_streaming": True,
            "input_schema": {
                "type": "object",
                "properties": {"city": {"type": "string"}},
                "required": ["city"],
            },
        }
    ],
    messages=[{"role": "user", "content": "Weather in Paris?"}],
) as stream:
    for event in stream:
        match event.type:
            case "content_block_start" if event.content_block.type == "tool_use":
                tool_inputs[event.index] = ""
            case "content_block_delta" if event.delta.type == "input_json_delta":
                tool_inputs[event.index] += event.delta.partial_json
            case "content_block_stop" if event.index in tool_inputs:
                parsed = json.loads(tool_inputs[event.index])
                print(f"Tool input: {parsed}")



Gestion du JSON invalide dans les réponses d'outils

Lorsque vous utilisez le streaming d'outils à granularité fine, vous pouvez recevoir du JSON invalide ou incomplet de la part du modèle. Si vous devez renvoyer ce JSON invalide au modèle dans un bloc de réponse d'erreur, vous pouvez l'encapsuler dans un objet JSON pour garantir un traitement correct (avec une clé appropriée). Par exemple :

{
  "INVALID_JSON": "<your invalid json string>"
}

Cette approche aide le modèle à comprendre que le contenu est un JSON invalide tout en préservant les données malformées d'origine à des fins de débogage.



Étapes suivantes