細粒度工具串流

建構工具基礎架構

細粒度工具串流

逐字元串流工具輸入，適用於延遲敏感的應用程式。

This feature is eligible for Zero Data Retention (ZDR). When your organization has a ZDR arrangement, data sent through this feature is not stored after the API response is returned.

細粒度工具串流在所有模型和所有平台上普遍可用。它啟用了串流工具使用參數值的功能，無需緩衝或 JSON 驗證，減少了開始接收大型參數的延遲。

使用細粒度工具串流時，您可能會收到無效或不完整的 JSON 輸入。請確保在您的程式碼中考慮這些邊界情況。

如何使用細粒度工具串流

細粒度工具串流在所有模型和所有平台上可用（Claude API、Amazon Bedrock、Google Vertex AI 和 Microsoft Foundry）。要使用它，請在任何您想啟用細粒度串流的使用者定義工具上將 eager_input_streaming 設定為 true，並在您的請求上啟用串流。

以下是如何使用 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)

在此範例中，細粒度工具串流使 Claude 能夠將長詩的行串流到工具呼叫 make_file 中，無需緩衝來驗證 lines_of_text 參數是否為有效的 JSON。這意味著您可以看到參數在到達時的串流，而無需等待整個參數緩衝和驗證。

使用細粒度工具串流時，工具使用區塊開始串流的速度更快，通常更長且包含更少的換行符。這是由於分塊行為的差異。

範例：

沒有細粒度串流（15 秒延遲）：

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'
...

使用細粒度串流（3 秒延遲）：

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

因為細粒度串流在沒有緩衝或 JSON 驗證的情況下發送參數，無法保證產生的串流將以有效的 JSON 字串完成。特別是，如果達到停止原因 max_tokens，串流可能在參數中途結束且可能不完整。您通常必須編寫特定的支援來處理何時達到 max_tokens 的情況。

累積工具輸入增量

當 tool_use 內容區塊串流時，初始 content_block_start 事件包含 input: {}（空物件）。這是一個佔位符。實際輸入以一系列 input_json_delta 事件的形式到達，每個事件都攜帶一個 partial_json 字串片段。您的程式碼必須連接這些片段，並在區塊關閉後解析結果。

累積合約：

在 content_block_start 上，其中 type: "tool_use"，初始化一個空字串：input_json = ""
對於每個 content_block_delta，其中 type: "input_json_delta"，附加：input_json += event.delta.partial_json
在 content_block_stop 上，解析累積的字串：json.loads(input_json)

初始 input: {}（物件）和 partial_json（字串）之間的型別不匹配是設計上的。空物件標記內容陣列中的位置；增量字串構建實際值。

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

Python 和 TypeScript SDK 提供了更高級別的串流幫助程式（stream.get_final_message()、stream.finalMessage()），為您執行此累積。只有在您需要在區塊關閉前對部分輸入做出反應時，才使用上面的手動模式，例如呈現進度指示器或提前啟動下游請求。

處理工具回應中的無效 JSON

使用細粒度工具串流時，您可能會從模型接收無效或不完整的 JSON。如果您需要在錯誤回應區塊中將此無效 JSON 傳回給模型，您可以將其包裝在 JSON 物件中以確保正確處理（使用合理的鍵）。例如：

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

這種方法幫助模型理解內容是無效的 JSON，同時保留原始格式錯誤的資料以供除錯之用。

包裝無效 JSON 時，請確保正確轉義無效 JSON 字串中的任何引號或特殊字元，以在包裝物件中維持有效的 JSON 結構。

後續步驟

串流訊息

伺服器發送事件和串流事件型別的完整參考。

處理工具呼叫

執行工具並以所需的訊息格式返回結果。

工具參考

Anthropic 架構工具及其版本字串的完整目錄。

Was this page helpful?

建構工具基礎架構

細粒度工具串流

逐字元串流工具輸入，適用於延遲敏感的應用程式。

This feature is eligible for Zero Data Retention (ZDR). When your organization has a ZDR arrangement, data sent through this feature is not stored after the API response is returned.

細粒度工具串流在所有模型和所有平台上普遍可用。它啟用了串流工具使用參數值的功能，無需緩衝或 JSON 驗證，減少了開始接收大型參數的延遲。

使用細粒度工具串流時，您可能會收到無效或不完整的 JSON 輸入。請確保在您的程式碼中考慮這些邊界情況。

如何使用細粒度工具串流

以下是如何使用 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)

使用細粒度工具串流時，工具使用區塊開始串流的速度更快，通常更長且包含更少的換行符。這是由於分塊行為的差異。

範例：

沒有細粒度串流（15 秒延遲）：

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'
...

使用細粒度串流（3 秒延遲）：

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

累積工具輸入增量

累積合約：

在 content_block_start 上，其中 type: "tool_use"，初始化一個空字串：input_json = ""
對於每個 content_block_delta，其中 type: "input_json_delta"，附加：input_json += event.delta.partial_json
在 content_block_stop 上，解析累積的字串：json.loads(input_json)

初始 input: {}（物件）和 partial_json（字串）之間的型別不匹配是設計上的。空物件標記內容陣列中的位置；增量字串構建實際值。

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

處理工具回應中的無效 JSON

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

這種方法幫助模型理解內容是無效的 JSON，同時保留原始格式錯誤的資料以供除錯之用。

包裝無效 JSON 時，請確保正確轉義無效 JSON 字串中的任何引號或特殊字元，以在包裝物件中維持有效的 JSON 結構。

後續步驟

串流訊息

伺服器發送事件和串流事件型別的完整參考。

處理工具呼叫

執行工具並以所需的訊息格式返回結果。

工具參考

Anthropic 架構工具及其版本字串的完整目錄。

Was this page helpful?