From b7743e685805bfb085dbbaff798452db066b1ede Mon Sep 17 00:00:00 2001 From: chengqian Date: Fri, 18 Sep 2026 16:25:37 +0800 Subject: [PATCH] =?UTF-8?q?fix(relay):=E4=BF=AE=E5=A4=8Dresponses=E5=8D=8F?= =?UTF-8?q?=E8=AE=AE=E8=B0=83=E7=94=A8claude=E5=8D=8F=E8=AE=AE=E6=8A=A5?= =?UTF-8?q?=E9=94=99=E9=97=AE=E9=A2=98=EF=BC=88=E9=80=82=E9=85=8DZCode?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../claude_to_responses_stream.go | 137 +++++++++++++++--- .../claude_to_responses_test.go | 27 +++- .../thinking.stream.json | 80 +++++++++- .../register/goldensuite_stream_test.go | 6 +- 4 files changed, 224 insertions(+), 26 deletions(-) diff --git a/relaykit/relayconvert/internal/native_responses/claude_to_responses_stream.go b/relaykit/relayconvert/internal/native_responses/claude_to_responses_stream.go index 24baf30..1c1fce7 100644 --- a/relaykit/relayconvert/internal/native_responses/claude_to_responses_stream.go +++ b/relaykit/relayconvert/internal/native_responses/claude_to_responses_stream.go @@ -63,8 +63,10 @@ func (c *ClaudeToResponsesStreamConverter) ConvertStreamResponse( sentCreated := false sentTextDone := false sentCompleted := false - outputIndex := 0 contentIndex := 0 + // 输出项索引：message 项固定占 0，其余输出项（reasoning / tool call / server tool） + // 由 itemIdx 从 1 起分配——与 chat→responses 桥接的口径一致 + itemIdx := 1 toolCalls := make([]*claudeToolCallState, 0) // 有序聚合，completed 的 output 数组按此顺序 toolIndexByID := make(map[string]int) // callID → output_index var currentTool *claudeToolCallState // 正在接收参数增量的工具调用 @@ -80,7 +82,64 @@ func (c *ClaudeToResponsesStreamConverter) ConvertStreamResponse( return chunkWriter(&relayconvert.StreamEvent{Event: eventType, Data: data, Usage: u}) } - // closeTextPart 关闭文本 content part（进入工具调用或流结束时调用） + // 思考内容：必须落成独立的 reasoning 输出项（output_item.added → summary_part.added → + // summary 增量 → done 收口）。把 summary 增量挂在 message 项上是协议违规——OpenAI SDK + // 的流式状态机按 item 类型校验事件归属，错挂会使客户端解析失败并主动断流（client_gone）。 + // Claude 的 interleaved thinking 会出现多个 thinking 块，收口后重开新项。 + var reasoningBuf strings.Builder + reasoningID := "" + reasoningItemIdx := -1 + reasoningItems := make([]map[string]any, 0) // 已收口的 reasoning 项，completed 输出数组用 + closeReasoningItem := func() error { + if reasoningItemIdx < 0 { + return nil + } + text := reasoningBuf.String() + if err := emitEvent("response.reasoning_summary_text.done", map[string]any{ + "type": "response.reasoning_summary_text.done", + "item_id": reasoningID, + "output_index": reasoningItemIdx, + "summary_index": 0, + "text": text, + }, nil); err != nil { + return err + } + if err := emitEvent("response.reasoning_summary_part.done", map[string]any{ + "type": "response.reasoning_summary_part.done", + "item_id": reasoningID, + "output_index": reasoningItemIdx, + "summary_index": 0, + "part": map[string]any{ + "type": "summary_text", + "text": text, + }, + }, nil); err != nil { + return err + } + item := map[string]any{ + "type": "reasoning", + "id": reasoningID, + "status": "completed", + "summary": []map[string]any{{ + "type": "summary_text", + "text": text, + }}, + } + if err := emitEvent("response.output_item.done", map[string]any{ + "type": "response.output_item.done", + "output_index": reasoningItemIdx, + "item": item, + }, nil); err != nil { + return err + } + reasoningItems = append(reasoningItems, item) + reasoningItemIdx = -1 + reasoningID = "" + reasoningBuf.Reset() + return nil + } + + // closeTextPart 关闭文本 content part（进入工具调用或流结束时调用）；message 项固定占 output_index 0 closeTextPart := func() error { if sentTextDone { return nil @@ -89,7 +148,7 @@ func (c *ClaudeToResponsesStreamConverter) ConvertStreamResponse( if err := emitEvent("response.output_text.done", map[string]any{ "type": "response.output_text.done", "item_id": msgID, - "output_index": outputIndex, + "output_index": 0, "content_index": contentIndex, "text": finishedText, }, nil); err != nil { @@ -98,7 +157,7 @@ func (c *ClaudeToResponsesStreamConverter) ConvertStreamResponse( if err := emitEvent("response.content_part.done", map[string]any{ "type": "response.content_part.done", "item_id": msgID, - "output_index": outputIndex, + "output_index": 0, "content_index": contentIndex, "part": map[string]any{ "type": "output_text", @@ -110,7 +169,7 @@ func (c *ClaudeToResponsesStreamConverter) ConvertStreamResponse( } if err := emitEvent("response.output_item.done", map[string]any{ "type": "response.output_item.done", - "output_index": outputIndex, + "output_index": 0, "item": map[string]any{ "type": "message", "id": msgID, @@ -126,7 +185,6 @@ func (c *ClaudeToResponsesStreamConverter) ConvertStreamResponse( return err } sentTextDone = true - outputIndex++ return nil } @@ -135,12 +193,15 @@ func (c *ClaudeToResponsesStreamConverter) ConvertStreamResponse( if sentCompleted { return nil } + if err := closeReasoningItem(); err != nil { + return err + } if err := closeTextPart(); err != nil { return err } // 输出数组：文本消息（closeTextPart 已置 sentTextDone，此处等价于有文本才保留， - // 与 chat→responses 桥接的 finalOutput 构建口径一致） + // 与 chat→responses 桥接的 finalOutput 构建口径一致）+ reasoning 项 + 工具调用 finalOutput := make([]map[string]any, 0) if textBuf.Len() > 0 { finalOutput = append(finalOutput, map[string]any{ @@ -155,6 +216,8 @@ func (c *ClaudeToResponsesStreamConverter) ConvertStreamResponse( }}, }) } + // reasoning 项：completed 的 output 数组需携带完整项，客户端据此重建下一轮 input 历史 + finalOutput = append(finalOutput, reasoningItems...) for _, tc := range toolCalls { if err := emitEvent("response.function_call_arguments.done", map[string]any{ @@ -260,7 +323,7 @@ func (c *ClaudeToResponsesStreamConverter) ConvertStreamResponse( } if err := emitEvent("response.output_item.added", map[string]any{ "type": "response.output_item.added", - "output_index": outputIndex, + "output_index": 0, "item": map[string]any{ "type": "message", "id": msgID, @@ -274,7 +337,7 @@ func (c *ClaudeToResponsesStreamConverter) ConvertStreamResponse( if err := emitEvent("response.content_part.added", map[string]any{ "type": "response.content_part.added", "item_id": msgID, - "output_index": outputIndex, + "output_index": 0, "content_index": contentIndex, "part": map[string]any{ "type": "output_text", @@ -292,17 +355,20 @@ func (c *ClaudeToResponsesStreamConverter) ConvertStreamResponse( } switch event.ContentBlock.Type { case "tool_use": - // 先关闭文本 content part，再开 function_call 项 + // 先收口 reasoning 项与文本 content part，再开 function_call 项 + if err := closeReasoningItem(); err != nil { + return err + } if err := closeTextPart(); err != nil { return err } tc := &claudeToolCallState{id: event.ContentBlock.ID, name: event.ContentBlock.Name} toolCalls = append(toolCalls, tc) - toolIndexByID[tc.id] = outputIndex + toolIndexByID[tc.id] = itemIdx currentTool = tc if err := emitEvent("response.output_item.added", map[string]any{ "type": "response.output_item.added", - "output_index": outputIndex, + "output_index": itemIdx, "item": map[string]any{ "type": "function_call", "id": tc.id, @@ -313,23 +379,26 @@ func (c *ClaudeToResponsesStreamConverter) ConvertStreamResponse( }, nil); err != nil { return err } - outputIndex++ + itemIdx++ case "server_tool_use": // claude 服务端工具：仅 web_search 有 Responses 对应物（web_search_call）， // 其余服务端工具跳过。参数以 input_json_delta 聚合，收尾在 finish 统一冲刷 if event.ContentBlock.Name != "web_search" { continue } + if err := closeReasoningItem(); err != nil { + return err + } if err := closeTextPart(); err != nil { return err } st := &claudeToolCallState{id: event.ContentBlock.ID, name: event.ContentBlock.Name} serverToolCalls = append(serverToolCalls, st) - serverToolIndexByID[st.id] = outputIndex + serverToolIndexByID[st.id] = itemIdx currentServerTool = st if err := emitEvent("response.output_item.added", map[string]any{ "type": "response.output_item.added", - "output_index": outputIndex, + "output_index": itemIdx, "item": map[string]any{ "type": "web_search_call", "id": st.id, @@ -338,7 +407,7 @@ func (c *ClaudeToResponsesStreamConverter) ConvertStreamResponse( }, nil); err != nil { return err } - outputIndex++ + itemIdx++ case "text", "thinking", "redacted_thinking": // 文本/思考块：文本复用首个 content part，思考以 reasoning summary 事件透出 } @@ -363,10 +432,42 @@ func (c *ClaudeToResponsesStreamConverter) ConvertStreamResponse( } case "thinking_delta": if event.Delta.Thinking != nil && *event.Delta.Thinking != "" { + // 懒开 reasoning 输出项：首个增量到达时发 output_item.added + + // summary_part.added，后续增量挂在该项的 id / output_index 上 + if reasoningItemIdx < 0 { + reasoningItemIdx = itemIdx + itemIdx++ + reasoningID = fmt.Sprintf("rs_%s_%d", respID, reasoningItemIdx) + if err := emitEvent("response.output_item.added", map[string]any{ + "type": "response.output_item.added", + "output_index": reasoningItemIdx, + "item": map[string]any{ + "type": "reasoning", + "id": reasoningID, + "status": "in_progress", + "summary": []any{}, + }, + }, nil); err != nil { + return err + } + if err := emitEvent("response.reasoning_summary_part.added", map[string]any{ + "type": "response.reasoning_summary_part.added", + "item_id": reasoningID, + "output_index": reasoningItemIdx, + "summary_index": 0, + "part": map[string]any{ + "type": "summary_text", + "text": "", + }, + }, nil); err != nil { + return err + } + } + reasoningBuf.WriteString(*event.Delta.Thinking) if err := emitEvent("response.reasoning_summary_text.delta", map[string]any{ "type": "response.reasoning_summary_text.delta", - "item_id": msgID, - "output_index": 0, + "item_id": reasoningID, + "output_index": reasoningItemIdx, "summary_index": 0, "delta": *event.Delta.Thinking, }, nil); err != nil { diff --git a/relaykit/relayconvert/internal/native_responses/claude_to_responses_test.go b/relaykit/relayconvert/internal/native_responses/claude_to_responses_test.go index a81c6ec..523c5eb 100644 --- a/relaykit/relayconvert/internal/native_responses/claude_to_responses_test.go +++ b/relaykit/relayconvert/internal/native_responses/claude_to_responses_test.go @@ -324,18 +324,37 @@ func TestClaudeToResponsesStream_ThinkingAsReasoningSummary(t *testing.T) { got := responsesEventTypes(t, events) want := []string{ "response.created", - "response.output_item.added", + "response.output_item.added", // message 项（固定 output_index 0） "response.content_part.added", + "response.output_item.added", // reasoning 项（懒开，独立 output_index） + "response.reasoning_summary_part.added", "response.reasoning_summary_text.delta", // signature_delta 无对应物被忽略 + "response.reasoning_summary_text.done", + "response.reasoning_summary_part.done", + "response.output_item.done", // reasoning 项收口 "response.output_text.done", "response.content_part.done", - "response.output_item.done", + "response.output_item.done", // message 项收口 "response.completed", } if strings.Join(got, ",") != strings.Join(want, ",") { t.Fatalf("事件序列不符\ngot: %v\nwant: %v", got, want) } - if responsesDataOf(t, events[3])["delta"] != "推理中" { - t.Errorf("reasoning delta 不对: %+v", responsesDataOf(t, events[3])) + // reasoning 增量必须挂在独立 reasoning 项上（rs_ 前缀 id），不得错挂 message 项—— + // OpenAI SDK 流式状态机按 item 归属校验，错挂会使客户端解析失败断流 + reasoningAdded := responsesDataOf(t, events[3]) + item, _ := reasoningAdded["item"].(map[string]any) + if item == nil || item["type"] != "reasoning" { + t.Fatalf("第二个 output_item.added 必须是 reasoning 项: %+v", reasoningAdded) + } + delta := responsesDataOf(t, events[5]) + if delta["delta"] != "推理中" { + t.Errorf("reasoning delta 不对: %+v", delta) + } + if delta["item_id"] != item["id"] { + t.Errorf("reasoning delta 的 item_id 应指向 reasoning 项: delta=%+v item=%+v", delta, item) + } + if delta["output_index"] != reasoningAdded["output_index"] { + t.Errorf("reasoning delta 的 output_index 应与 reasoning 项一致: delta=%+v added=%+v", delta, reasoningAdded) } } diff --git a/relaykit/relayconvert/register/golden/expected/claude_messages_to_oai_responses_stream_resp/thinking.stream.json b/relaykit/relayconvert/register/golden/expected/claude_messages_to_oai_responses_stream_resp/thinking.stream.json index 22cfe25..4e70250 100644 --- a/relaykit/relayconvert/register/golden/expected/claude_messages_to_oai_responses_stream_resp/thinking.stream.json +++ b/relaykit/relayconvert/register/golden/expected/claude_messages_to_oai_responses_stream_resp/thinking.stream.json @@ -64,11 +64,37 @@ }, "event": "response.content_part.added" }, + { + "data": { + "item": { + "id": "\u003cscrubbed\u003e", + "status": "in_progress", + "summary": [], + "type": "reasoning" + }, + "output_index": 1, + "type": "response.output_item.added" + }, + "event": "response.output_item.added" + }, + { + "data": { + "item_id": "\u003cscrubbed\u003e", + "output_index": 1, + "part": { + "text": "", + "type": "summary_text" + }, + "summary_index": 0, + "type": "response.reasoning_summary_part.added" + }, + "event": "response.reasoning_summary_part.added" + }, { "data": { "delta": "Let me think...", "item_id": "\u003cscrubbed\u003e", - "output_index": 0, + "output_index": 1, "summary_index": 0, "type": "response.reasoning_summary_text.delta" }, @@ -84,6 +110,47 @@ }, "event": "response.output_text.delta" }, + { + "data": { + "item_id": "\u003cscrubbed\u003e", + "output_index": 1, + "summary_index": 0, + "text": "Let me think...", + "type": "response.reasoning_summary_text.done" + }, + "event": "response.reasoning_summary_text.done" + }, + { + "data": { + "item_id": "\u003cscrubbed\u003e", + "output_index": 1, + "part": { + "text": "Let me think...", + "type": "summary_text" + }, + "summary_index": 0, + "type": "response.reasoning_summary_part.done" + }, + "event": "response.reasoning_summary_part.done" + }, + { + "data": { + "item": { + "id": "\u003cscrubbed\u003e", + "status": "completed", + "summary": [ + { + "text": "Let me think...", + "type": "summary_text" + } + ], + "type": "reasoning" + }, + "output_index": 1, + "type": "response.output_item.done" + }, + "event": "response.output_item.done" + }, { "data": { "content_index": 0, @@ -154,6 +221,17 @@ "role": "assistant", "status": "completed", "type": "message" + }, + { + "id": "\u003cscrubbed\u003e", + "status": "completed", + "summary": [ + { + "text": "Let me think...", + "type": "summary_text" + } + ], + "type": "reasoning" } ], "parallel_tool_calls": true, diff --git a/relaykit/relayconvert/register/goldensuite_stream_test.go b/relaykit/relayconvert/register/goldensuite_stream_test.go index 890bc34..0d3e351 100644 --- a/relaykit/relayconvert/register/goldensuite_stream_test.go +++ b/relaykit/relayconvert/register/goldensuite_stream_test.go @@ -125,9 +125,9 @@ func TestGoldenSuite_Streams(t *testing.T) { } // synthesizedStreamID 流式转换器按时间戳合成的 ID 形态（resp_/msg_/fc_/chatcmpl- 前缀 -// 接毫秒/纳秒时间戳，或 call_[_]）。透传自语料的短 ID（msg_up / toolu_1 / -// call_1 等）不匹配、不清洗，保持可断言。 -var synthesizedStreamID = regexp.MustCompile(`^(?:resp_|msg_|fc_|chatcmpl-)\d{10,}$|^call_\d{10,}(?:_\d+)?$`) +// 接毫秒/纳秒时间戳，call_[_]，或 reasoning 项的 rs_resp_[_]）。 +// 透传自语料的短 ID（msg_up / toolu_1 / call_1 等）不匹配、不清洗，保持可断言。 +var synthesizedStreamID = regexp.MustCompile(`^(?:resp_|msg_|fc_|chatcmpl-)\d{10,}$|^(?:call_|rs_resp_)\d{10,}(?:_\d+)?$`) // scrubStreamNondeterministic 递归清洗流式帧树中的非确定字段： // - 命中 nondeterministicKeys 的 key（id/created/created_at/completed_at 等）按类型置占位；