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
@@ -0,0 +1,213 @@
|
||||
package loop
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"reflect"
|
||||
"sync"
|
||||
"testing"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/config"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||||
)
|
||||
|
||||
// 本文件是并行工具调用的**端到端**回归：用真适配器（AnthropicChatAdapter）
|
||||
// 打一个按 Anthropic 契约校验请求的假上游，跑完整回合。真机 P0 的原文
|
||||
// 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.
|
||||
//
|
||||
// 在这里被逐字复刻：上游只放行「assistant 回合里的每个 tool_use 都在紧随其后的
|
||||
// 那一条消息里被同 id 的 tool_result 应答」的请求。因此修复一旦回退，本用例会
|
||||
// 以同一个 400 失败 —— 而不是靠对内部结构做断言。
|
||||
|
||||
// strictAnthropicUpstream 是一个按 Anthropic 配对契约校验的假上游。
|
||||
// 它按顺序逐次回放 responses（超出则重复最后一条）。
|
||||
type strictAnthropicUpstream struct {
|
||||
mu sync.Mutex
|
||||
bodies []map[string]any
|
||||
responses []string
|
||||
srv *httptest.Server
|
||||
}
|
||||
|
||||
func newStrictAnthropicUpstream(t *testing.T, responses ...string) *strictAnthropicUpstream {
|
||||
t.Helper()
|
||||
up := &strictAnthropicUpstream{responses: responses}
|
||||
up.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
var body map[string]any
|
||||
_ = json.NewDecoder(r.Body).Decode(&body)
|
||||
|
||||
up.mu.Lock()
|
||||
up.bodies = append(up.bodies, body)
|
||||
idx := len(up.bodies) - 1
|
||||
if idx >= len(up.responses) {
|
||||
idx = len(up.responses) - 1
|
||||
}
|
||||
payload := up.responses[idx]
|
||||
up.mu.Unlock()
|
||||
|
||||
if bad := unpairedToolUse(body); bad != "" {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.WriteHeader(http.StatusBadRequest)
|
||||
_, _ = fmt.Fprintf(w, `{"error":{"message":%q,"type":"invalid_request_error"}}`, bad)
|
||||
return
|
||||
}
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.WriteHeader(http.StatusOK)
|
||||
_, _ = io.WriteString(w, payload)
|
||||
}))
|
||||
t.Cleanup(up.srv.Close)
|
||||
return up
|
||||
}
|
||||
|
||||
// requests 返回假上游收到的请求体快照数量。
|
||||
func (up *strictAnthropicUpstream) requests() int {
|
||||
up.mu.Lock()
|
||||
defer up.mu.Unlock()
|
||||
return len(up.bodies)
|
||||
}
|
||||
|
||||
// bodyAt 返回第 i 个请求体。
|
||||
func (up *strictAnthropicUpstream) bodyAt(i int) map[string]any {
|
||||
up.mu.Lock()
|
||||
defer up.mu.Unlock()
|
||||
return up.bodies[i]
|
||||
}
|
||||
|
||||
// unpairedToolUse 复刻上游的校验：对每条 assistant 消息，取出它声明的全部
|
||||
// tool_use id，要求紧随其后的那一条消息里出现**同 id、同数量、同顺序**的
|
||||
// tool_result 块；违反时返回与上游同形的错误文案（空串表示通过）。
|
||||
func unpairedToolUse(body map[string]any) string {
|
||||
raw, _ := body["messages"].([]any)
|
||||
for i, m := range raw {
|
||||
msg, _ := m.(map[string]any)
|
||||
uses := wireBlockIDs(msg, "tool_use")
|
||||
if len(uses) == 0 {
|
||||
continue
|
||||
}
|
||||
if i+1 >= len(raw) {
|
||||
return fmt.Sprintf("messages.%d:`tool_use` ids were found without `tool_result` blocks immediately after: %s",
|
||||
i, joinIDs(uses))
|
||||
}
|
||||
next, _ := raw[i+1].(map[string]any)
|
||||
results := wireBlockIDs(next, "tool_result")
|
||||
if !reflect.DeepEqual(uses, results) {
|
||||
return fmt.Sprintf("messages.%d:`tool_use` ids were found without `tool_result` blocks immediately after: %s",
|
||||
i, joinIDs(missingIDs(uses, results)))
|
||||
}
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
// wireBlockIDs 按出现顺序取出 assistant 的 tool_use id / 结果的 tool_result id。
|
||||
func wireBlockIDs(msg map[string]any, blockType string) []string {
|
||||
blocks, _ := msg["content"].([]any)
|
||||
ids := make([]string, 0, len(blocks))
|
||||
for _, b := range blocks {
|
||||
blk, _ := b.(map[string]any)
|
||||
if blk["type"] != blockType {
|
||||
continue
|
||||
}
|
||||
key := "id"
|
||||
if blockType == "tool_result" {
|
||||
key = "tool_use_id"
|
||||
}
|
||||
if s, ok := blk[key].(string); ok {
|
||||
ids = append(ids, s)
|
||||
}
|
||||
}
|
||||
return ids
|
||||
}
|
||||
|
||||
// joinIDs 渲染 id 列表，供错误文案使用。
|
||||
func joinIDs(ids []string) string {
|
||||
out := ""
|
||||
for i, id := range ids {
|
||||
if i > 0 {
|
||||
out += ", "
|
||||
}
|
||||
out += id
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// missingIDs 返回 uses 里没有在 results 中被一一应答的 id。
|
||||
func missingIDs(uses, results []string) []string {
|
||||
answered := make(map[string]int, len(results))
|
||||
for _, r := range results {
|
||||
answered[r]++
|
||||
}
|
||||
missing := make([]string, 0, len(uses))
|
||||
for _, u := range uses {
|
||||
if answered[u] > 0 {
|
||||
answered[u]--
|
||||
continue
|
||||
}
|
||||
missing = append(missing, u)
|
||||
}
|
||||
return missing
|
||||
}
|
||||
|
||||
// parallelToolUseResponse 是首步模型响应：**同一步并行发出两个** tool_use
|
||||
// （真机那次是两个 knowledge 调用）。id 采用 OpenAI 风格的
|
||||
// call_01_*，与真机日志一致。
|
||||
func parallelToolUseResponse() string {
|
||||
return `{
|
||||
"id":"msg_1","model":"claude-test","role":"assistant",
|
||||
"content":[
|
||||
{"type":"tool_use","id":"call_01_a","name":"knowledge","input":{"query":"Cyrene"}},
|
||||
{"type":"tool_use","id":"call_01_b","name":"datetime","input":{"timezone":"Asia/Shanghai"}}
|
||||
],
|
||||
"stop_reason":"tool_use",
|
||||
"usage":{"input_tokens":10,"output_tokens":8}
|
||||
}`
|
||||
}
|
||||
|
||||
const parallelFollowUpResponse = `{
|
||||
"id":"msg_2","model":"claude-test","role":"assistant",
|
||||
"content":[{"type":"text","text":"两条记录都在。"}],
|
||||
"stop_reason":"end_turn",
|
||||
"usage":{"input_tokens":20,"output_tokens":5}
|
||||
}`
|
||||
|
||||
// TestRun_ParallelToolCalls_SecondStepAccepted 端到端复现真机 P0 的场景：
|
||||
// 同一步并行两个 tool_use，回灌结果后必须能完成第二步模型调用并给出最终答复。
|
||||
// 修复回退时上游按原样回 400，本用例即以那条错误失败。
|
||||
func TestRun_ParallelToolCalls_SecondStepAccepted(t *testing.T) {
|
||||
t.Parallel()
|
||||
up := newStrictAnthropicUpstream(t, parallelToolUseResponse(), parallelFollowUpResponse)
|
||||
ad := model.NewAnthropicChat(config.LLMConfig{
|
||||
AnthropicBaseURL: up.srv.URL, AnthropicModel: "claude-test",
|
||||
})
|
||||
td := dispatcherWith(t,
|
||||
&mockTool{name: "knowledge", result: map[string]any{"count": 10}},
|
||||
&mockTool{name: "datetime", result: map[string]any{"now": "2026-10-04 22:17"}},
|
||||
)
|
||||
l := New(Options{Agent: newTestAgent(t, ad, td), MaxRounds: 8})
|
||||
|
||||
res, err := l.Run(context.Background(), "sess", []model.Message{
|
||||
model.TextMessage(model.RoleUser, "请调用 knowledge 工具检索关于「Cyrene」的记录"),
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("Run: %v（旧行为下这里是上游 400: tool_use ids were found without tool_result blocks）", err)
|
||||
}
|
||||
if res.Text != "两条记录都在。" {
|
||||
t.Fatalf("Text = %q, want 两条记录都在。", res.Text)
|
||||
}
|
||||
if res.ToolCallsExecuted != 2 {
|
||||
t.Fatalf("ToolCallsExecuted = %d, want 2（两个并行调用都要执行）", res.ToolCallsExecuted)
|
||||
}
|
||||
if up.requests() != 2 {
|
||||
t.Fatalf("上游请求数 = %d, want 2（第二步必须真的发出去）", up.requests())
|
||||
}
|
||||
// 第二步请求的配对形态也要成立（不只是「没报错」）。
|
||||
if bad := unpairedToolUse(up.bodyAt(1)); bad != "" {
|
||||
t.Fatalf("第二步请求仍违反配对契约: %s", bad)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,111 @@
|
||||
package loop
|
||||
|
||||
import (
|
||||
"reflect"
|
||||
"testing"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||||
)
|
||||
|
||||
// 本文件钉死工具回填侧的形态契约（appendToolCallHistory）：一条 assistant
|
||||
// 回合声明**本步的全部** tool_use，其后**每个调用一条** RoleTool 消息，
|
||||
// 顺序与调用顺序一致。真机 P0（并行 tool_use 的原生回填让上游判 400
|
||||
// 「tool_use ids were found without tool_result blocks immediately after」）
|
||||
// 的修复落在 Anthropic 编码器（model/anthropic_encode.go：相邻的 RoleTool
|
||||
// 回合合并成同一条 user 消息）；这里守住的是编码器的输入 —— 回填侧既不能
|
||||
// 漏掉某个调用的结果，也不能打乱顺序或丢失 id 配对，否则编码器无从补救。
|
||||
|
||||
// nativeCalls 构造一批带 provider id 的调用（原生路径）。
|
||||
func nativeCalls(ids ...string) []agent.ToolCall {
|
||||
calls := make([]agent.ToolCall, 0, len(ids))
|
||||
for _, id := range ids {
|
||||
calls = append(calls, agent.ToolCall{ID: id, Name: "knowledge", Args: map[string]any{"query": "Cyrene"}})
|
||||
}
|
||||
return calls
|
||||
}
|
||||
|
||||
// toolResultIDs 按消息顺序取出 RoleTool 回合应答的 provider id。
|
||||
func toolResultIDs(msgs []model.Message) []string {
|
||||
ids := make([]string, 0, len(msgs))
|
||||
for _, m := range msgs {
|
||||
if m.Role != model.RoleTool {
|
||||
continue
|
||||
}
|
||||
ids = append(ids, m.ToolCallID)
|
||||
}
|
||||
return ids
|
||||
}
|
||||
|
||||
// TestAppendToolCallHistory_ParallelBatch_PairsEveryCall 覆盖同一步里
|
||||
// **≥2 个并行调用**：assistant 回合必须声明全部调用，且每个 id 都要在紧随其后的
|
||||
// 结果里出现恰好一次、顺序一致。
|
||||
func TestAppendToolCallHistory_ParallelBatch_PairsEveryCall(t *testing.T) {
|
||||
t.Parallel()
|
||||
ids := []string{"call_01_a", "call_01_b", "call_01_c"}
|
||||
outcomes := make([]ToolOutcome, len(ids))
|
||||
for i := range outcomes {
|
||||
outcomes[i] = ToolOutcome{Result: map[string]any{"count": 10}}
|
||||
}
|
||||
|
||||
msgs := appendToolCallHistory(nil, nativeCalls(ids...), outcomes)
|
||||
|
||||
if len(msgs) != len(ids)+1 {
|
||||
t.Fatalf("messages = %d, want %d (1 条 assistant + 每调用 1 条结果)", len(msgs), len(ids)+1)
|
||||
}
|
||||
if msgs[0].Role != model.RoleAssistant {
|
||||
t.Fatalf("messages[0].role = %q, want assistant（结果必须先有调用可关联）", msgs[0].Role)
|
||||
}
|
||||
if got := toolCallIDs(msgs[0].Content); !reflect.DeepEqual(got, ids) {
|
||||
t.Fatalf("assistant tool_use id = %v, want %v", got, ids)
|
||||
}
|
||||
if got := toolResultIDs(msgs[1:]); !reflect.DeepEqual(got, ids) {
|
||||
t.Fatalf("tool_result id = %v, want %v（顺序与调用一致、不得去重或丢失）", got, ids)
|
||||
}
|
||||
}
|
||||
|
||||
// TestAppendToolCallHistory_SingleCall_Pairs 覆盖单调用形态，防止并行修复
|
||||
// 把单调用路径改坏。
|
||||
func TestAppendToolCallHistory_SingleCall_Pairs(t *testing.T) {
|
||||
t.Parallel()
|
||||
msgs := appendToolCallHistory(nil, nativeCalls("call_01_only"), []ToolOutcome{{Result: "ok"}})
|
||||
|
||||
if len(msgs) != 2 {
|
||||
t.Fatalf("messages = %d, want 2", len(msgs))
|
||||
}
|
||||
if got := toolCallIDs(msgs[0].Content); !reflect.DeepEqual(got, []string{"call_01_only"}) {
|
||||
t.Fatalf("assistant tool_use id = %v, want [call_01_only]", got)
|
||||
}
|
||||
if got := toolResultIDs(msgs[1:]); !reflect.DeepEqual(got, []string{"call_01_only"}) {
|
||||
t.Fatalf("tool_result id = %v, want [call_01_only]", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestAppendToolCallHistory_LegacyEnvelope_NoNativeTurn 守住旧式 JSON 信封
|
||||
// 路径（调用不带 provider id）：它没有可关联的原生 tool_use，因此保持纯 RoleTool
|
||||
// 文本回灌，绝不凭空造一条 assistant 回合。
|
||||
func TestAppendToolCallHistory_LegacyEnvelope_NoNativeTurn(t *testing.T) {
|
||||
t.Parallel()
|
||||
calls := []agent.ToolCall{{Name: "echo", Args: map[string]any{"q": "x"}}, {Name: "echo2", Args: map[string]any{"q": "y"}}}
|
||||
msgs := appendToolCallHistory(nil, calls, []ToolOutcome{{Result: "a"}, {Result: "b"}})
|
||||
|
||||
if len(msgs) != 2 {
|
||||
t.Fatalf("messages = %d, want 2（每个调用一条纯文本回灌）", len(msgs))
|
||||
}
|
||||
for i, m := range msgs {
|
||||
if m.Role != model.RoleTool {
|
||||
t.Fatalf("messages[%d].role = %q, want tool", i, m.Role)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// toolCallIDs 按内容块顺序取出 assistant 回合声明的 tool_use id。
|
||||
func toolCallIDs(parts []model.Content) []string {
|
||||
ids := make([]string, 0, len(parts))
|
||||
for _, p := range parts {
|
||||
if p.Type == model.ContentTypeToolCall {
|
||||
ids = append(ids, p.ToolCallID)
|
||||
}
|
||||
}
|
||||
return ids
|
||||
}
|
||||
@@ -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