Claude 流式输出与 Tool Use 笔记

1. Claude 一次完整回复是什么？

非流式情况下，可以把 Claude 的一次 assistant 回复理解成：

1
2
3
4
5
6
7
8
9
Message(
    role="assistant",
    content=[
        TextBlock(...),
        ToolUseBlock(...),
        ToolUseBlock(...),
    ],
    stop_reason="tool_use",
)

重点是：

1
message.content

是一个 ContentBlock 数组，并不保证只有一个 block。

因此一次回复完全可能是：

1
2
3
4
5
6
Message
│
├── TextBlock
├── ToolUseBlock A
├── ToolUseBlock B
└── ToolUseBlock C

Claude 官方明确支持一次 assistant turn 返回多个 tool_use block；当存在多个独立工具调用时，可以并发或串行执行，由应用自己决定。(Claude Platform)

2. 流式模式不是直接返回完整 Block

Streaming 时，一个 ContentBlock 会经历：

1
2
3
4
5
6
7
8
9
content_block_start
        ↓
content_block_delta
        ↓
content_block_delta
        ↓
content_block_delta
        ↓
content_block_stop

整条 Message 则大致是：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
message_start

    content_block_start
    content_block_delta
    ...
    content_block_stop

    content_block_start
    content_block_delta
    ...
    content_block_stop

message_delta
message_stop

所以一定要区分：

1
content_block_stop

和：

1
message_stop

content_block_stop 不是“结束时间”。

它是一个事件，意思是：

当前第 N 个 ContentBlock 已经生成完整。

而：

1
message_stop

才代表：

Claude 这整条 assistant message 输出完了。

Anthropic 官方 streaming 协议就是按照 content block 的 start / delta / stop 生命周期组织的。(Claude Platform)

3. TextBlock 流式时确实是碎的

比如最终文本是：

1
宝宝你好呀

流里不一定一次给你：

1
"宝宝你好呀"

而可能：

1
2
3
4
"宝"
"宝你"
"好"
"呀"

也可能：

1
2
"宝宝"
"你好呀"

所以：

不要依赖 delta 的切割边界。

对于文本，这完全没问题，因为我们通常就是：

1
2
if event.type == "text":
    print(event.text, end="")

然后：

1
2
3
text delta
→ 立即推给 UI
→ 打字机效果

4. ToolUseBlock 也是一点一点生成的

假设最终工具调用是：

1
2
3
4
get_weather(
    city="Tokyo",
    unit="celsius",
)

刚开始可能先出现：

1
content_block_start

此时大概已经知道：

1
2
3
id   = toolu_xxx
name = get_weather
input = {}

然后参数通过 input_json_delta 一点一点过来，例如：

1
2
3
4
{"city"
:"Tok
yo","unit":
"celsius"}

这些中间碎片不是完整 JSON。

所以不能：

1
json.loads(event.delta.partial_json)

看到一个 delta 就调用工具也不行。

官方建议就是累积这些 JSON delta，并在 content_block_stop 后再解析完整工具输入；SDK helper 也提供了对累计值的支持。(Claude Platform)

核心原则：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
Text delta
    ↓
可以立即消费


Tool input delta
    ↓
不能直接执行
    ↓
等待完整参数
    ↓
content_block_stop
    ↓
解析 / 校验
    ↓
执行工具

5. content_block_stop 为什么这么重要？

假设 Claude 一次调用两个工具：

1
查东京和大阪的天气

可能产生：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
message_start

TextBlock
├── text_delta
└── content_block_stop


ToolUseBlock A
├── name = get_weather
├── input_json_delta
├── input_json_delta
└── content_block_stop
        ↑
        A 已经完整


ToolUseBlock B
├── name = get_weather
├── input_json_delta
├── input_json_delta
└── content_block_stop
        ↑
        B 已经完整


message_stop

所以理论上：

1
2
3
4
5
6
7
8
9
A content_block_stop
        ↓
A 参数已经完整
        ↓
A 可以开始执行

与此同时
        ↓
Claude 还可以继续生成 B

这就是我们前面说的提前执行工具。

不是必须这么做，只是一种 latency 优化。

6. 多个 ToolUseBlock 是完全正常的

例如 Claude 最终 Message：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
content = [
    TextBlock(
        text="我帮你查一下。"
    ),

    ToolUseBlock(
        id="toolu_01",
        name="get_weather",
        input={"city": "Tokyo"},
    ),

    ToolUseBlock(
        id="toolu_02",
        name="get_weather",
        input={"city": "Osaka"},
    ),
]

此时：

1
response.stop_reason == "tool_use"

Claude 官方说明，一次响应可以包含一个或多个tool_use block；多个调用的执行顺序由你的程序决定。(Claude Platform)

如果工具之间没有依赖：

1
2
read_file(a.py)
read_file(b.py)

完全可以：

1
2
A ───────────→
B ───────────→

并行执行。

但如果 B 依赖 A：

1
2
3
4
5
读取 config
↓
从 config 得到 URL
↓
clone URL

那么通常要：

1
2
3
4
5
6
7
8
9
10
11
12
13
Claude round 1
↓
tool_use read_config

你的程序执行
↓
tool_result

Claude round 2
↓
看到 config 内容
↓
tool_use git_clone

7. ToolUse 和 ToolResult 是一一对应的

每一个：

1
2
3
4
ToolUseBlock(
    id="toolu_123",
    ...
)

都需要对应：

1
2
3
4
5
{
    "type": "tool_result",
    "tool_use_id": "toolu_123",
    "content": result,
}

多个工具则：

1
2
3
4
5
6
7
8
9
10
11
12
[
    {
        "type": "tool_result",
        "tool_use_id": "toolu_A",
        "content": result_a,
    },
    {
        "type": "tool_result",
        "tool_use_id": "toolu_B",
        "content": result_b,
    },
]

官方建议多个调用的结果一起放进下一条 user message，并通过 tool_use_id 对应原调用。(Claude Platform)

于是 Agent Loop 本质上就是：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
User
 ↓
Claude
 ↓
tool_use A
tool_use B
 ↓
你的程序执行 A / B
 ↓
tool_result A
tool_result B
 ↓
Claude
 ↓
继续推理 / 最终回答

注意：

Claude 只是“请求调用工具”。真正执行工具的是你的代码。 (Claude Platform)

对照你现在的代码

你现在 llm.py 用的是：

1
2
3
4
5
6
7
8
async with _get_client().messages.stream(...) as stream:
    async for event in stream:
        if event.type == "thinking":
            yield "thinking", event.thinking
        elif event.type == "text":
            yield "text", event.text

    yield "message_end", stream.current_message_snapshot

也就是说你现在只向 Agent 暴露：

1
2
3
thinking
text
message_end

并没有把 content_block_stop 往上层传。

所以当前流程是：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
Claude streaming
      ↓
text
      ↓
tool_use A 生成完
      ↓
你没处理 block_stop
      ↓
tool_use B 生成完
      ↓
你没处理 block_stop
      ↓
message_stop
      ↓
stream 结束
      ↓
message_end
      ↓
Agent 得到完整 final

然后你的 Agent：

1
2
3
4
5
for block in final.content:
    if getattr(block, "type", None) != "tool_use":
        continue

    arguments = cast(dict, block.input)

把所有 ToolUseBlock 找出来。

接着：

1
output = await session.tool_registry.execute(...)

执行工具，再组装：

1
2
3
4
5
{
    "type": "tool_result",
    "tool_use_id": block.id,
    "content": output,
}

回填 Claude。

所以你当前实现可以概括成：

1
2
3
4
5
6
7
8
9
10
11
等待整个 Message 完成
        ↓
拿 final.content
        ↓
遍历全部 ToolUseBlock
        ↓
一个一个执行
        ↓
收集全部 ToolResult
        ↓
下一轮 Claude

这是完全正确的 Agent Loop。

以后可以做的优化

你现在：

1
2
3
4
5
6
7
8
Claude:
生成 Tool A
生成 Tool B
message_stop
        ↓
执行 A
        ↓
执行 B

以后性能优化可以变成：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
Claude:
生成 Tool A
    ↓
A content_block_stop
    └────────→ A 开始执行

Claude 继续生成 Tool B
    ↓
B content_block_stop
    └────────→ B 开始执行

Claude message_stop
    ↓
await A + B
    ↓
统一回填 tool_result

甚至多个独立工具：

1
2
3
4
5
await asyncio.gather(
    run_tool_a(),
    run_tool_b(),
    run_tool_c(),
)

Anthropic 官方也明确指出，对于相互独立的调用，可以使用 asyncio.gather 等方式并发执行；有副作用、共享状态或顺序要求的工具则更适合串行。(Claude Platform)

不过这属于：

1
性能优化

而不是：

1
Agent 正确运行的必要条件

所以你现在完全不用急着改。

最后记住这一张图就够了

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
                 Claude Message
                       │
        ┌──────────────┼──────────────┐
        ↓              ↓              ↓
    TextBlock     ToolUseBlock A  ToolUseBlock B
        │              │              │
     delta           delta          delta
     delta           delta          delta
     delta              │              │
        │               ↓              ↓
        │       content_block_stop content_block_stop
        │               │              │
        │               └──────┬───────┘
        │                      ↓
        │                  执行工具
        │                      ↓
        ↓                  ToolResult
实时推给 UI                  ↓
                       下一轮 Claude

然后再牢牢记住这三句话：

text delta：碎一点没事，直接流给 UI。
content_block_stop：当前 block 完整了，ToolUse 参数此时可安全取得。
message_stop：Claude 整个 assistant turn 完整结束。

以及你目前的代码是：

暂时不利用 content_block_stop，而是等 message_end 后统一处理所有 ToolUseBlock。这个方案正确、简单、稳。

等你后面开始专门压 Agent 响应延迟，再把 content_block_stop + asyncio task + 多工具并行 这颗“小涡轮”装上去就行 😼