Streaming granulare degli strumenti

Lo streaming granulare degli strumenti è generalmente disponibile su tutti i modelli e tutte le piattaforme. Abilita lo streaming dei valori dei parametri di utilizzo degli strumenti senza buffering o convalida JSON, riducendo la latenza per iniziare a ricevere parametri di grandi dimensioni.

Come utilizzare lo streaming granulare degli strumenti

Lo streaming granulare degli strumenti è disponibile su tutti i modelli e tutte le piattaforme (Claude API, Amazon Bedrock, Google Vertex AI e Microsoft Foundry). Per utilizzarlo, imposta eager_input_streaming su true su qualsiasi strumento definito dall'utente in cui desideri abilitare lo streaming granulare e abilita lo streaming sulla tua richiesta.

Ecco un esempio di come utilizzare lo streaming granulare degli strumenti con l'API:

client = anthropic.Anthropic()

with client.messages.stream(
    max_tokens=65536,
    model="claude-opus-4-7",
    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:
    for event in stream:
        pass
    final_message = stream.get_final_message()

print(final_message.usage)

In questo esempio, lo streaming granulare degli strumenti consente a Claude di trasmettere i versi di una lunga poesia nella chiamata dello strumento make_file senza buffering per convalidare se il parametro lines_of_text è un JSON valido. Ciò significa che puoi vedere il parametro trasmesso mentre arriva, senza dover aspettare che l'intero parametro sia bufferizzato e convalidato.

Chunk 1: '{"'
Chunk 2: 'query": "Ty'
Chunk 3: 'peScri'
Chunk 4: 'pt 5.0 5.1 '
Chunk 5: '5.2 5'
Chunk 6: '.3'
Chunk 8: ' new f'
Chunk 9: 'eatur'
...

Chunk 1: '{"query": "TypeScript 5.0 5.1 5.2 5.3'
Chunk 2: ' new features comparison'

Accumulo dei delta di input degli strumenti

Quando un blocco di contenuto tool_use trasmette, l'evento iniziale content_block_start contiene input: {} (un oggetto vuoto). Questo è un segnaposto. L'input effettivo arriva come una serie di eventi input_json_delta, ognuno contenente un frammento di stringa partial_json. Il tuo codice deve concatenare questi frammenti e analizzare il risultato una volta che il blocco si chiude.

Il contratto di accumulo:

1. Su content_block_start con type: "tool_use", inizializza una stringa vuota: input_json = ""
2. Per ogni content_block_delta con type: "input_json_delta", aggiungi: input_json += event.delta.partial_json
3. Su content_block_stop, analizza la stringa accumulata: json.loads(input_json)

La mancata corrispondenza di tipo tra l'input: {} iniziale (oggetto) e partial_json (stringa) è intenzionale. L'oggetto vuoto contrassegna lo slot nell'array di contenuto; le stringhe delta costruiscono il valore reale.

import json
import anthropic

client = anthropic.Anthropic()

tool_inputs = {}  # index -> accumulated JSON string

with client.messages.stream(
    model="claude-opus-4-7",
    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:
        if (
            event.type == "content_block_start"
            and event.content_block.type == "tool_use"
        ):
            tool_inputs[event.index] = ""
        elif (
            event.type == "content_block_delta"
            and event.delta.type == "input_json_delta"
        ):
            tool_inputs[event.index] += event.delta.partial_json
        elif event.type == "content_block_stop" and event.index in tool_inputs:
            parsed = json.loads(tool_inputs[event.index])
            print(f"Tool input: {parsed}")

Gestione di JSON non valido nelle risposte degli strumenti

Quando si utilizza lo streaming granulare degli strumenti, potresti ricevere JSON non valido o incompleto dal modello. Se hai bisogno di passare questo JSON non valido al modello in un blocco di risposta di errore, puoi avvolgerlo in un oggetto JSON per garantire una gestione corretta (con una chiave ragionevole). Ad esempio:

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

Questo approccio aiuta il modello a capire che il contenuto è JSON non valido mentre preserva i dati malformati originali per scopi di debug.

Passaggi successivi