细粒度工具流式传输

构建工具基础设施

细粒度工具流式传输

逐字符流式传输工具输入，用于对延迟敏感的应用程序。

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（字符串）之间的类型不匹配是设计上的。空对象标记内容数组中的槽位；增量字符串构建实际值。

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（字符串）之间的类型不匹配是设计上的。空对象标记内容数组中的槽位；增量字符串构建实际值。

处理工具响应中的无效 JSON

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

这种方法帮助模型理解内容是无效的 JSON，同时保留原始格式错误的数据以供调试之用。

包装无效 JSON 时，请确保正确转义无效 JSON 字符串中的任何引号或特殊字符，以在包装器对象中维护有效的 JSON 结构。

后续步骤

流式传输消息

服务器发送事件和流事件类型的完整参考。

处理工具调用

执行工具并以所需的消息格式返回结果。

工具参考

Anthropic 架构工具及其版本字符串的完整目录。

Was this page helpful?

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