fix(model): Anthropic 并行工具结果合并为同一条 user 消息（修真机 400）
根因（agent-193，文件:行）：loop 的 tool_history 按调用逐条产出 RoleTool 消息， 而 anthropic_encode 把每条各自编码成一条独立 user 消息 ⇒ 上游只认紧邻 assistant 的那一条，第 2..N 个 tool_use id 被判「没有 tool_result」⇒ `messages.1: tool_use ids were found without tool_result blocks immediately after`。 单调用回合只有 1 条结果消息，所以长期潜伏；KB1 让 knowledge 真正成功后它才被暴露。 修法：编码器在**线格式**层把相邻的 RoleTool 合并进同一条 user 消息（块序 = 调用序）， 遇到任何非工具结果回合即复位（不同 assistant 回合的结果绝不混入同一条）。 未碰 openai_encode（OpenAI 要求每个 tool_call_id 一条独立 tool 消息）。 测试 3 个新文件：编码器线格式（非流式 + 流式）/ loop 端到端（真适配器打假上游）/ 回填侧契约；含配对断言（每个 tool_use 在下一条消息里有同 id 的 tool_result）。 变异：还原旧编码 ⇒ 4 条红，端到端逐字复现真机 400。 是否本轮引入：否（编码与回填两文件在本轮零改动，属基线潜伏缺陷）。
This commit is contained in:
AskaEth committed
1 parent
fd13b49b24
commit
15181e7063
4 files changed
+652
No files matched your search
@@ -12,12 +12,26 @@ import (
|
||||
// 会被丢弃，因为 Anthropic 拒绝空内容。挂在非用户回合上的
|
||||
// 文档部件在这里就被拒绝（design/24 §2.3-③）：文档本期永不
|
||||
// 抵达编码器，漏网即编程错误，绝不静默丢块。
|
||||
//
|
||||
// 连续的 RoleTool 回合在这里被合并成**一条** user 消息。这是
|
||||
// Anthropic 线格式的硬要求，不是风格取舍：assistant 回合里每个
|
||||
// tool_use 的应答块都必须落在**紧接其后的那一条**消息里，而回填侧
|
||||
// （loop/tool_history.go）是按调用逐条产出 RoleTool 消息的，因此
|
||||
// 一批并行调用的多个结果不合并就会分散在多条 user 消息中，上游
|
||||
// 只在下一条里看到其中一个 id，其余 id 被判「没有 tool_result」，
|
||||
// 整个请求 400（真机 P0；单调用回合因此永远看不出问题）。
|
||||
func buildAnthropicMessages(msgs []Message) (string, []anthropicMessage, error) {
|
||||
if err := ValidateDocumentParts(msgs); err != nil {
|
||||
return "", nil, err
|
||||
}
|
||||
var sys strings.Builder
|
||||
out := make([]anthropicMessage, 0, len(msgs))
|
||||
// pendingToolResult 是 out 中正在累积的工具结果回合下标，-1 表示
|
||||
// 没有可合并的目标。每追加一条非工具结果回合就复位它，因此只有
|
||||
// **真正相邻**的结果才会被并入同一条消息 —— 两批不同 assistant
|
||||
// 回合的结果绝不会被混在一起。RoleSystem 回合被抽到顶层 system
|
||||
// 字段、不进 messages，不打断相邻性，所以不参与复位。
|
||||
pendingToolResult := -1
|
||||
for _, m := range msgs {
|
||||
if m.Role == RoleSystem {
|
||||
if t := messageText(m); t != "" {
|
||||
@@ -35,7 +49,22 @@ func buildAnthropicMessages(msgs []Message) (string, []anthropicMessage, error)
|
||||
if len(blocks) == 0 {
|
||||
continue
|
||||
}
|
||||
if m.Role != RoleTool {
|
||||
pendingToolResult = -1
|
||||
out = append(out, anthropicMessage{Role: anthropicRole(m.Role), Content: blocks})
|
||||
continue
|
||||
}
|
||||
if pendingToolResult >= 0 {
|
||||
// 合并进上一条工具结果回合。类型断言必然成立（上面只写
|
||||
// []any）；不成立时故意**落到下面**再开一条消息，而不是
|
||||
// 静默丢弃已有块 —— 丢块会让请求更坏，且无从察觉。
|
||||
if merged, ok := out[pendingToolResult].Content.([]any); ok {
|
||||
out[pendingToolResult].Content = append(merged, blocks...)
|
||||
continue
|
||||
}
|
||||
}
|
||||
out = append(out, anthropicMessage{Role: anthropicRole(m.Role), Content: blocks})
|
||||
pendingToolResult = len(out) - 1
|
||||
}
|
||||
return sys.String(), out, nil
|
||||
}
|
||||
|
||||
@@ -0,0 +1,299 @@
|
||||
package model
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/http"
|
||||
"reflect"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/config"
|
||||
)
|
||||
|
||||
// 本文件锁定一个真机抓到过的 P0：同一步里模型**并行**发出 ≥2 个 tool_use 时，
|
||||
// 上游对该请求判 400 ——
|
||||
//
|
||||
// messages.1: `tool_use` ids were found without `tool_result` blocks
|
||||
// immediately after: call_01_1RBBKboRTJCl2S80t0FJ3155.
|
||||
// Each `tool_use` block must have a corresponding `tool_result` block in the next message.
|
||||
//
|
||||
// Anthropic 协议要求：assistant 回合里**每一个** tool_use 的应答块，都必须出现在
|
||||
// **紧接其后的那一条**消息里。所以一批并行调用的全部 tool_result 必须落在同一条
|
||||
// user 消息中；分散成多条 user 消息时，校验器只在下一条里看到其中一个 id，
|
||||
// 其余 id 一律被判「没有 tool_result」，整个请求失败（单调用回合因此永远看不出
|
||||
// 问题——回填侧恰好只产出一条 user 回合）。下面三组用例把这条契约钉死在
|
||||
// **线格式**上（回填侧 loop/tool_history.go 的形态即 parallelToolTurnMessages）。
|
||||
|
||||
// parallelToolTurnMessages 复刻 loop 在一个「同一步并行发出 ≥2 个 tool_use」的
|
||||
// 回合里回填的内部消息形态（loop/tool_history.go 的 appendToolCallHistory）：
|
||||
// 一条 assistant 声明全部调用，其后**每个调用一条** RoleTool 消息，顺序与调用
|
||||
// 顺序一致。这正是真机那次请求编码前的输入。
|
||||
func parallelToolTurnMessages(ids []string) []Message {
|
||||
calls := make([]ToolCall, 0, len(ids))
|
||||
for _, id := range ids {
|
||||
calls = append(calls, ToolCall{
|
||||
ID: id, Name: "knowledge", Args: map[string]any{"query": "Cyrene"},
|
||||
})
|
||||
}
|
||||
msgs := []Message{
|
||||
TextMessage(RoleUser, "请调用 knowledge 工具检索关于「Cyrene」的记录"),
|
||||
AssistantToolCallMessage(calls),
|
||||
}
|
||||
for _, id := range ids {
|
||||
msgs = append(msgs, ToolResultMessage(id, `{"count":10}`, false))
|
||||
}
|
||||
return msgs
|
||||
}
|
||||
|
||||
// wireMessages 从抓到的请求体里取出 messages 数组。
|
||||
func wireMessages(t *testing.T, body map[string]any) []map[string]any {
|
||||
t.Helper()
|
||||
raw, ok := body["messages"].([]any)
|
||||
if !ok {
|
||||
t.Fatalf("body.messages = %#v, want an array", body["messages"])
|
||||
}
|
||||
out := make([]map[string]any, 0, len(raw))
|
||||
for _, m := range raw {
|
||||
msg, ok := m.(map[string]any)
|
||||
if !ok {
|
||||
t.Fatalf("message = %#v, want an object", m)
|
||||
}
|
||||
out = append(out, msg)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// wireBlocks 取出一条线格式消息的内容块。
|
||||
func wireBlocks(t *testing.T, msg map[string]any) []map[string]any {
|
||||
t.Helper()
|
||||
raw, ok := msg["content"].([]any)
|
||||
if !ok {
|
||||
t.Fatalf("message content = %#v, want an array of blocks", msg["content"])
|
||||
}
|
||||
out := make([]map[string]any, 0, len(raw))
|
||||
for _, b := range raw {
|
||||
blk, ok := b.(map[string]any)
|
||||
if !ok {
|
||||
t.Fatalf("content block = %#v, want an object", b)
|
||||
}
|
||||
out = append(out, blk)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// blockIDs 按出现顺序取出某类块的 id：tool_use 取 "id"，
|
||||
// tool_result 取 "tool_use_id"（它应答的那个 tool_use id）。
|
||||
func blockIDs(blocks []map[string]any, blockType string) []string {
|
||||
ids := make([]string, 0, len(blocks))
|
||||
for _, b := range blocks {
|
||||
if b["type"] != blockType {
|
||||
continue
|
||||
}
|
||||
key := "id"
|
||||
if blockType == "tool_result" {
|
||||
key = "tool_use_id"
|
||||
}
|
||||
s, _ := b[key].(string)
|
||||
ids = append(ids, s)
|
||||
}
|
||||
return ids
|
||||
}
|
||||
|
||||
// assertToolResultPairing 断言线格式满足 Anthropic 的配对契约：assistant
|
||||
// 回合里的每个 tool_use id，都要在下一条消息里被一条同 id 的 tool_result
|
||||
// 应答，且**数量与顺序完全一致**。这正是上游 400
|
||||
// 「ids were found without `tool_result` blocks immediately after」所判定的东西。
|
||||
func assertToolResultPairing(t *testing.T, body map[string]any) {
|
||||
t.Helper()
|
||||
msgs := wireMessages(t, body)
|
||||
for i, m := range msgs {
|
||||
uses := blockIDs(wireBlocks(t, m), "tool_use")
|
||||
if len(uses) == 0 {
|
||||
continue
|
||||
}
|
||||
if i+1 >= len(msgs) {
|
||||
t.Fatalf("messages.%d 声明了 tool_use %v，但之后没有任何消息应答它", i, uses)
|
||||
}
|
||||
results := blockIDs(wireBlocks(t, msgs[i+1]), "tool_result")
|
||||
if !reflect.DeepEqual(uses, results) {
|
||||
t.Fatalf("messages.%d 的 tool_use %v 在 messages.%d 里没有一一对应（实际 tool_result %v）；"+
|
||||
"上游会判 400（每个 tool_use 都必须在下一条消息里有同 id 的 tool_result）",
|
||||
i, uses, i+1, results)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestAnthropicBuild_ParallelToolResults_OneUserTurn 是三组用例里最直接的一组：
|
||||
// 直接对编码器断言「串行回填的多条 RoleTool 消息被合并成**一条** user 回合，
|
||||
// 块顺序与调用顺序一致」。旧行为（每条 RoleTool 各自一条 user 消息）在这里必红。
|
||||
func TestAnthropicBuild_ParallelToolResults_OneUserTurn(t *testing.T) {
|
||||
t.Parallel()
|
||||
ids := []string{"call_01_a", "call_01_b", "call_01_c"}
|
||||
_, out, err := buildAnthropicMessages(parallelToolTurnMessages(ids))
|
||||
if err != nil {
|
||||
t.Fatalf("buildAnthropicMessages error: %v", err)
|
||||
}
|
||||
wantRoles := []string{"user", "assistant", "user"}
|
||||
if len(out) != len(wantRoles) {
|
||||
t.Fatalf("messages = %d, want %d (并行工具结果必须合并为同一条 user 回合): roles=%v",
|
||||
len(out), len(wantRoles), anthropicRolesOf(out))
|
||||
}
|
||||
for i, want := range wantRoles {
|
||||
if out[i].Role != want {
|
||||
t.Fatalf("messages.%d role = %q, want %q (roles=%v)", i, out[i].Role, want, anthropicRolesOf(out))
|
||||
}
|
||||
}
|
||||
blocks, ok := out[2].Content.([]any)
|
||||
if !ok {
|
||||
t.Fatalf("messages.2 content = %#v, want a block array", out[2].Content)
|
||||
}
|
||||
if len(blocks) != len(ids) {
|
||||
t.Fatalf("messages.2 blocks = %d, want %d（一个 tool_use 一个应答块）", len(blocks), len(ids))
|
||||
}
|
||||
got := make([]string, 0, len(blocks))
|
||||
for _, b := range blocks {
|
||||
blk, ok := b.(map[string]any)
|
||||
if !ok {
|
||||
t.Fatalf("block = %#v, want an object", b)
|
||||
}
|
||||
if blk["type"] != "tool_result" {
|
||||
t.Fatalf("messages.2 里出现非 tool_result 块: %#v", blk)
|
||||
}
|
||||
s, _ := blk["tool_use_id"].(string)
|
||||
got = append(got, s)
|
||||
}
|
||||
if !reflect.DeepEqual(got, ids) {
|
||||
t.Fatalf("messages.2 的 tool_use_id 顺序 = %v, want %v（顺序必须与调用顺序一致）", got, ids)
|
||||
}
|
||||
}
|
||||
|
||||
// TestAnthropicBuild_ToolResults_SeparateTurnsNotMerged 守住合并的边界：
|
||||
// 只有**相邻**的工具结果才合并；两个不同 assistant 回合的结果之间隔着别的
|
||||
// 回合时绝不能并进同一条 user 消息（否则会把 A 批的结果塞到 B 批的
|
||||
// assistant 回合之后，反而制造出更坏的配对错乱）。
|
||||
func TestAnthropicBuild_ToolResults_SeparateTurnsNotMerged(t *testing.T) {
|
||||
t.Parallel()
|
||||
msgs := []Message{
|
||||
TextMessage(RoleUser, "first"),
|
||||
AssistantToolCallMessage([]ToolCall{{ID: "call_a", Name: "knowledge", Args: map[string]any{}}}),
|
||||
ToolResultMessage("call_a", `{"count":1}`, false),
|
||||
TextMessage(RoleUser, "second"),
|
||||
AssistantToolCallMessage([]ToolCall{{ID: "call_b", Name: "knowledge", Args: map[string]any{}}}),
|
||||
ToolResultMessage("call_b", `{"count":2}`, false),
|
||||
}
|
||||
_, out, err := buildAnthropicMessages(msgs)
|
||||
if err != nil {
|
||||
t.Fatalf("buildAnthropicMessages error: %v", err)
|
||||
}
|
||||
if len(out) != 6 {
|
||||
t.Fatalf("messages = %d, want 6 (两个回合各自一条结果回合): roles=%v", len(out), anthropicRolesOf(out))
|
||||
}
|
||||
wantRoles := []string{"user", "assistant", "user", "user", "assistant", "user"}
|
||||
for i, want := range wantRoles {
|
||||
if out[i].Role != want {
|
||||
t.Fatalf("messages.%d role = %q, want %q (roles=%v)", i, out[i].Role, want, anthropicRolesOf(out))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// anthropicRolesOf 只在断言失败时渲染角色序列，便于一眼看出错误形态。
|
||||
func anthropicRolesOf(msgs []anthropicMessage) []string {
|
||||
roles := make([]string, 0, len(msgs))
|
||||
for _, m := range msgs {
|
||||
roles = append(roles, m.Role)
|
||||
}
|
||||
return roles
|
||||
}
|
||||
|
||||
// TestAnthropicChat_ParallelToolResults_WirePairing 走真机同一条路（非流式
|
||||
// Chat → 抓请求体），断言并行回合在**线格式**上被一一配对。
|
||||
func TestAnthropicChat_ParallelToolResults_WirePairing(t *testing.T) {
|
||||
t.Parallel()
|
||||
ch := make(chan anthropicCaptured, 1)
|
||||
srv := newAnthropicServer(t, http.StatusOK, "application/json", anthropicTextResponse, ch)
|
||||
ad := NewAnthropicChat(config.LLMConfig{
|
||||
AnthropicBaseURL: srv.URL, AnthropicAPIKey: "ak-test", AnthropicModel: "claude-test",
|
||||
})
|
||||
|
||||
if _, err := ad.Chat(context.Background(), Request{
|
||||
Messages: parallelToolTurnMessages([]string{"call_01_a", "call_01_b"}),
|
||||
Model: "claude-test", MaxTokens: 64,
|
||||
}, Options{}); err != nil {
|
||||
t.Fatalf("Chat error: %v", err)
|
||||
}
|
||||
|
||||
got := <-ch
|
||||
assertToolResultPairing(t, got.body)
|
||||
msgs := wireMessages(t, got.body)
|
||||
if len(msgs) != 3 {
|
||||
t.Fatalf("messages = %d, want 3 (user / assistant / 合并后的 user)", len(msgs))
|
||||
}
|
||||
if msgs[1]["role"] != "assistant" || msgs[2]["role"] != "user" {
|
||||
t.Fatalf("roles = %v, want [user assistant user]", []any{msgs[0]["role"], msgs[1]["role"], msgs[2]["role"]})
|
||||
}
|
||||
uses := blockIDs(wireBlocks(t, msgs[1]), "tool_use")
|
||||
if !reflect.DeepEqual(uses, []string{"call_01_a", "call_01_b"}) {
|
||||
t.Fatalf("messages.1 tool_use ids = %v, want [call_01_a call_01_b]", uses)
|
||||
}
|
||||
}
|
||||
|
||||
// TestAnthropicChat_ParallelToolResults_StreamWirePairing 覆盖真机那次
|
||||
// 实际走的路：Web 聊天是流式，流式与非流式共用 buildRequest，因此也必须
|
||||
// 满足同一条契约。
|
||||
func TestAnthropicChat_ParallelToolResults_StreamWirePairing(t *testing.T) {
|
||||
t.Parallel()
|
||||
sse := strings.Join([]string{
|
||||
`data: {"type":"message_start","message":{"model":"claude-test","usage":{"input_tokens":5}}}`,
|
||||
``,
|
||||
`data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}`,
|
||||
``,
|
||||
`data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"ok"}}`,
|
||||
``,
|
||||
`data: {"type":"content_block_stop","index":0}`,
|
||||
``,
|
||||
`data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":2}}`,
|
||||
``,
|
||||
`data: {"type":"message_stop"}`,
|
||||
``,
|
||||
}, "\n")
|
||||
ch := make(chan anthropicCaptured, 1)
|
||||
srv := newAnthropicServer(t, http.StatusOK, "text/event-stream", sse, ch)
|
||||
ad := NewAnthropicChat(config.LLMConfig{AnthropicBaseURL: srv.URL, AnthropicModel: "claude-test"})
|
||||
|
||||
if _, err := ad.ChatStream(context.Background(), Request{
|
||||
Messages: parallelToolTurnMessages([]string{"call_01_a", "call_01_b"}),
|
||||
Model: "claude-test", Purpose: "chat",
|
||||
}, func(StreamEvent) {}); err != nil {
|
||||
t.Fatalf("ChatStream error: %v", err)
|
||||
}
|
||||
|
||||
captured := <-ch
|
||||
assertToolResultPairing(t, captured.body)
|
||||
}
|
||||
|
||||
// TestAnthropicChat_SingleToolResult_StillPaired 守住单调用形态：它本来就
|
||||
// 正常，合并逻辑不得把它改坏（既不能多出回合，也不能丢块）。
|
||||
func TestAnthropicChat_SingleToolResult_StillPaired(t *testing.T) {
|
||||
t.Parallel()
|
||||
ch := make(chan anthropicCaptured, 1)
|
||||
srv := newAnthropicServer(t, http.StatusOK, "application/json", anthropicTextResponse, ch)
|
||||
ad := NewAnthropicChat(config.LLMConfig{AnthropicBaseURL: srv.URL, AnthropicModel: "claude-test"})
|
||||
|
||||
if _, err := ad.Chat(context.Background(), Request{
|
||||
Messages: parallelToolTurnMessages([]string{"call_01_only"}),
|
||||
Model: "claude-test", MaxTokens: 64,
|
||||
}, Options{}); err != nil {
|
||||
t.Fatalf("Chat error: %v", err)
|
||||
}
|
||||
|
||||
got := <-ch
|
||||
assertToolResultPairing(t, got.body)
|
||||
msgs := wireMessages(t, got.body)
|
||||
if len(msgs) != 3 {
|
||||
t.Fatalf("messages = %d, want 3 (user / assistant / user)", len(msgs))
|
||||
}
|
||||
results := blockIDs(wireBlocks(t, msgs[2]), "tool_result")
|
||||
if !reflect.DeepEqual(results, []string{"call_01_only"}) {
|
||||
t.Fatalf("messages.2 tool_result ids = %v, want [call_01_only]", results)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user