Feingranulares Tool-Streaming

Feingranulares Tool-Streaming ist auf allen Modellen und allen Plattformen allgemein verfügbar. Es ermöglicht Streaming von Tool-Nutzungsparameterwerten ohne Pufferung oder JSON-Validierung und reduziert die Latenz beim Empfang großer Parameter.

Verwendung von feingranularem Tool-Streaming

Feingranulares Tool-Streaming ist auf allen Modellen und allen Plattformen verfügbar (Claude API, Amazon Bedrock, Google Vertex AI und Microsoft Foundry). Um es zu verwenden, setzen Sie eager_input_streaming auf true für jedes benutzerdefinierte Tool, für das Sie feingranulares Streaming aktivieren möchten, und aktivieren Sie Streaming für Ihre Anfrage.

Hier ist ein Beispiel für die Verwendung von feingranularem Tool-Streaming mit der 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 diesem Beispiel ermöglicht feingranulares Tool-Streaming Claude, die Zeilen eines langen Gedichts in den Tool-Aufruf make_file zu streamen, ohne zu puffern, um zu validieren, ob der Parameter lines_of_text gültiges JSON ist. Dies bedeutet, dass Sie den Parameter sehen können, während er ankommt, ohne auf die gesamte Pufferung und Validierung des Parameters warten zu müssen.

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'

Akkumulieren von Tool-Eingabe-Deltas

Wenn ein tool_use-Inhaltsblock streamt, enthält das initiale content_block_start-Ereignis input: {} (ein leeres Objekt). Dies ist ein Platzhalter. Die tatsächliche Eingabe kommt als eine Reihe von input_json_delta-Ereignissen an, von denen jedes einen partial_json-String-Fragment trägt. Ihr Code muss diese Fragmente verketten und das Ergebnis analysieren, sobald der Block geschlossen wird.

Der Akkumulierungsvertrag:

1. Bei content_block_start mit type: "tool_use" initialisieren Sie einen leeren String: input_json = ""
2. Für jedes content_block_delta mit type: "input_json_delta" fügen Sie hinzu: input_json += event.delta.partial_json
3. Bei content_block_stop analysieren Sie den akkumulierten String: json.loads(input_json)

Die Typabweichung zwischen dem initialen input: {} (Objekt) und partial_json (String) ist beabsichtigt. Das leere Objekt markiert den Platz im Inhaltsarray; die Delta-Strings erstellen den echten Wert.

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}")

Umgang mit ungültigem JSON in Tool-Antworten

Bei Verwendung von feingranularem Tool-Streaming können Sie ungültiges oder unvollständiges JSON vom Modell erhalten. Wenn Sie dieses ungültige JSON in einem Fehlerantwortblock an das Modell zurückgeben müssen, können Sie es in einem JSON-Objekt einwickeln, um eine ordnungsgemäße Verarbeitung zu gewährleisten (mit einem angemessenen Schlüssel). Zum Beispiel:

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

Dieser Ansatz hilft dem Modell zu verstehen, dass der Inhalt ungültiges JSON ist, während die ursprünglichen fehlerhaften Daten zu Debugging-Zwecken erhalten bleiben.

Nächste Schritte