From 7314c974f3ee3148db921b6527e322089be87be7 Mon Sep 17 00:00:00 2001 From: Seefs Date: Thu, 5 Feb 2026 19:32:26 +0800 Subject: [PATCH] fix: Claude stream block index/type transitions --- relay/common/relay_info.go | 3 ++ service/convert.go | 89 +++++++++++++++++++++++++++----------- 2 files changed, 67 insertions(+), 25 deletions(-) diff --git a/relay/common/relay_info.go b/relay/common/relay_info.go index f5c1d769e..ade3876ff 100644 --- a/relay/common/relay_info.go +++ b/relay/common/relay_info.go @@ -37,6 +37,9 @@ type ClaudeConvertInfo struct { Usage *dto.Usage FinishReason string Done bool + + ToolCallBaseIndex int + ToolCallMaxIndexOffset int } type RerankerInfo struct { diff --git a/service/convert.go b/service/convert.go index 97d1144dc..52824374e 100644 --- a/service/convert.go +++ b/service/convert.go @@ -207,6 +207,44 @@ func StreamResponseOpenAI2Claude(openAIResponse *dto.ChatCompletionsStreamRespon } var claudeResponses []*dto.ClaudeResponse + // stopOpenBlocks emits the required content_block_stop event(s) for the currently open block(s) + // according to Anthropic's SSE streaming state machine: + // content_block_start -> content_block_delta* -> content_block_stop (per index). + // + // For text/thinking, there is at most one open block at info.ClaudeConvertInfo.Index. + // For tools, OpenAI tool_calls can stream multiple parallel tool_use blocks (indexed from 0), + // so we may have multiple open blocks and must stop each one explicitly. + stopOpenBlocks := func() { + switch info.ClaudeConvertInfo.LastMessagesType { + case relaycommon.LastMessageTypeText, relaycommon.LastMessageTypeThinking: + claudeResponses = append(claudeResponses, generateStopBlock(info.ClaudeConvertInfo.Index)) + case relaycommon.LastMessageTypeTools: + base := info.ClaudeConvertInfo.ToolCallBaseIndex + for offset := 0; offset <= info.ClaudeConvertInfo.ToolCallMaxIndexOffset; offset++ { + claudeResponses = append(claudeResponses, generateStopBlock(base+offset)) + } + } + } + // stopOpenBlocksAndAdvance closes the currently open block(s) and advances the content block index + // to the next available slot for subsequent content_block_start events. + // + // This prevents invalid streams where a content_block_delta (e.g. thinking_delta) is emitted for an + // index whose active content_block type is different (the typical cause of "Mismatched content block type"). + stopOpenBlocksAndAdvance := func() { + if info.ClaudeConvertInfo.LastMessagesType == relaycommon.LastMessageTypeNone { + return + } + stopOpenBlocks() + switch info.ClaudeConvertInfo.LastMessagesType { + case relaycommon.LastMessageTypeTools: + info.ClaudeConvertInfo.Index = info.ClaudeConvertInfo.ToolCallBaseIndex + info.ClaudeConvertInfo.ToolCallMaxIndexOffset + 1 + info.ClaudeConvertInfo.ToolCallBaseIndex = 0 + info.ClaudeConvertInfo.ToolCallMaxIndexOffset = 0 + default: + info.ClaudeConvertInfo.Index++ + } + info.ClaudeConvertInfo.LastMessagesType = relaycommon.LastMessageTypeNone + } if info.SendResponseCount == 1 { msg := &dto.ClaudeMediaMessage{ Id: openAIResponse.Id, @@ -228,6 +266,8 @@ func StreamResponseOpenAI2Claude(openAIResponse *dto.ChatCompletionsStreamRespon //}) if openAIResponse.IsToolCall() { info.ClaudeConvertInfo.LastMessagesType = relaycommon.LastMessageTypeTools + info.ClaudeConvertInfo.ToolCallBaseIndex = 0 + info.ClaudeConvertInfo.ToolCallMaxIndexOffset = 0 var toolCall dto.ToolCallResponse if len(openAIResponse.Choices) > 0 && len(openAIResponse.Choices[0].Delta.ToolCalls) > 0 { toolCall = openAIResponse.Choices[0].Delta.ToolCalls[0] @@ -271,9 +311,8 @@ func StreamResponseOpenAI2Claude(openAIResponse *dto.ChatCompletionsStreamRespon content := openAIResponse.Choices[0].Delta.GetContentString() if reasoning != "" { - if info.ClaudeConvertInfo.LastMessagesType == relaycommon.LastMessageTypeTools || info.ClaudeConvertInfo.LastMessagesType == relaycommon.LastMessageTypeText { - claudeResponses = append(claudeResponses, generateStopBlock(info.ClaudeConvertInfo.Index)) - info.ClaudeConvertInfo.Index++ + if info.ClaudeConvertInfo.LastMessagesType != relaycommon.LastMessageTypeThinking { + stopOpenBlocksAndAdvance() } idx := info.ClaudeConvertInfo.Index claudeResponses = append(claudeResponses, &dto.ClaudeResponse{ @@ -295,9 +334,8 @@ func StreamResponseOpenAI2Claude(openAIResponse *dto.ChatCompletionsStreamRespon }) info.ClaudeConvertInfo.LastMessagesType = relaycommon.LastMessageTypeThinking } else if content != "" { - if info.ClaudeConvertInfo.LastMessagesType == relaycommon.LastMessageTypeTools || info.ClaudeConvertInfo.LastMessagesType == relaycommon.LastMessageTypeThinking { - claudeResponses = append(claudeResponses, generateStopBlock(info.ClaudeConvertInfo.Index)) - info.ClaudeConvertInfo.Index++ + if info.ClaudeConvertInfo.LastMessagesType != relaycommon.LastMessageTypeText { + stopOpenBlocksAndAdvance() } idx := info.ClaudeConvertInfo.Index claudeResponses = append(claudeResponses, &dto.ClaudeResponse{ @@ -324,7 +362,7 @@ func StreamResponseOpenAI2Claude(openAIResponse *dto.ChatCompletionsStreamRespon // 如果首块就带 finish_reason，需要立即发送停止块 if len(openAIResponse.Choices) > 0 && openAIResponse.Choices[0].FinishReason != nil && *openAIResponse.Choices[0].FinishReason != "" { info.FinishReason = *openAIResponse.Choices[0].FinishReason - claudeResponses = append(claudeResponses, generateStopBlock(info.ClaudeConvertInfo.Index)) + stopOpenBlocks() oaiUsage := openAIResponse.Usage if oaiUsage == nil { oaiUsage = info.ClaudeConvertInfo.Usage @@ -355,7 +393,7 @@ func StreamResponseOpenAI2Claude(openAIResponse *dto.ChatCompletionsStreamRespon // no choices // 可能为非标准的 OpenAI 响应，判断是否已经完成 if info.ClaudeConvertInfo.Done { - claudeResponses = append(claudeResponses, generateStopBlock(info.ClaudeConvertInfo.Index)) + stopOpenBlocks() oaiUsage := info.ClaudeConvertInfo.Usage if oaiUsage != nil { claudeResponses = append(claudeResponses, &dto.ClaudeResponse{ @@ -389,18 +427,25 @@ func StreamResponseOpenAI2Claude(openAIResponse *dto.ChatCompletionsStreamRespon if len(chosenChoice.Delta.ToolCalls) > 0 { toolCalls := chosenChoice.Delta.ToolCalls if info.ClaudeConvertInfo.LastMessagesType != relaycommon.LastMessageTypeTools { - claudeResponses = append(claudeResponses, generateStopBlock(info.ClaudeConvertInfo.Index)) - info.ClaudeConvertInfo.Index++ + stopOpenBlocksAndAdvance() + info.ClaudeConvertInfo.ToolCallBaseIndex = info.ClaudeConvertInfo.Index + info.ClaudeConvertInfo.ToolCallMaxIndexOffset = 0 } info.ClaudeConvertInfo.LastMessagesType = relaycommon.LastMessageTypeTools + base := info.ClaudeConvertInfo.ToolCallBaseIndex + maxOffset := info.ClaudeConvertInfo.ToolCallMaxIndexOffset for i, toolCall := range toolCalls { - blockIndex := info.ClaudeConvertInfo.Index + offset := 0 if toolCall.Index != nil { - blockIndex = *toolCall.Index - } else if len(toolCalls) > 1 { - blockIndex = info.ClaudeConvertInfo.Index + i + offset = *toolCall.Index + } else { + offset = i } + if offset > maxOffset { + maxOffset = offset + } + blockIndex := base + offset idx := blockIndex if toolCall.Function.Name != "" { @@ -426,19 +471,16 @@ func StreamResponseOpenAI2Claude(openAIResponse *dto.ChatCompletionsStreamRespon }, }) } - - info.ClaudeConvertInfo.Index = blockIndex } + info.ClaudeConvertInfo.ToolCallMaxIndexOffset = maxOffset + info.ClaudeConvertInfo.Index = base + maxOffset } else { reasoning := chosenChoice.Delta.GetReasoningContent() textContent := chosenChoice.Delta.GetContentString() if reasoning != "" || textContent != "" { if reasoning != "" { if info.ClaudeConvertInfo.LastMessagesType != relaycommon.LastMessageTypeThinking { - if info.ClaudeConvertInfo.LastMessagesType == relaycommon.LastMessageTypeText || info.ClaudeConvertInfo.LastMessagesType == relaycommon.LastMessageTypeTools { - claudeResponses = append(claudeResponses, generateStopBlock(info.ClaudeConvertInfo.Index)) - info.ClaudeConvertInfo.Index++ - } + stopOpenBlocksAndAdvance() idx := info.ClaudeConvertInfo.Index claudeResponses = append(claudeResponses, &dto.ClaudeResponse{ Index: &idx, @@ -456,10 +498,7 @@ func StreamResponseOpenAI2Claude(openAIResponse *dto.ChatCompletionsStreamRespon } } else { if info.ClaudeConvertInfo.LastMessagesType != relaycommon.LastMessageTypeText { - if info.ClaudeConvertInfo.LastMessagesType == relaycommon.LastMessageTypeThinking || info.ClaudeConvertInfo.LastMessagesType == relaycommon.LastMessageTypeTools { - claudeResponses = append(claudeResponses, generateStopBlock(info.ClaudeConvertInfo.Index)) - info.ClaudeConvertInfo.Index++ - } + stopOpenBlocksAndAdvance() idx := info.ClaudeConvertInfo.Index claudeResponses = append(claudeResponses, &dto.ClaudeResponse{ Index: &idx, @@ -487,7 +526,7 @@ func StreamResponseOpenAI2Claude(openAIResponse *dto.ChatCompletionsStreamRespon } if doneChunk || info.ClaudeConvertInfo.Done { - claudeResponses = append(claudeResponses, generateStopBlock(info.ClaudeConvertInfo.Index)) + stopOpenBlocks() oaiUsage := openAIResponse.Usage if oaiUsage == nil { oaiUsage = info.ClaudeConvertInfo.Usage