From 030d17ed947adf8f441c853afb6ddc0637f72fe9 Mon Sep 17 00:00:00 2001 From: Luis Pater Date: Mon, 19 Oct 2026 08:11:17 +0800 Subject: [PATCH] perf(openai): cache stream preferences and tool maps per stream in Claude response translator - Cache and reuse resolved stream configuration and tool name maps across stream chunks to avoid redundant request body parsing. - Introduce dedicated state flags in conversion parameters to track resolution status safely. Closes: #6746 --- .../openai/claude/openai_claude_response.go | 31 ++++++-- .../claude/openai_claude_response_test.go | 76 +++++++++++++++++++ 2 files changed, 99 insertions(+), 8 deletions(-) diff --git a/internal/translator/openai/claude/openai_claude_response.go b/internal/translator/openai/claude/openai_claude_response.go index d416d4169..83b2c03e9 100644 --- a/internal/translator/openai/claude/openai_claude_response.go +++ b/internal/translator/openai/claude/openai_claude_response.go @@ -68,6 +68,14 @@ type ConvertOpenAIResponseToAnthropicParams struct { UsageOutputTokens int64 UsageCachedTokens int64 UsageCacheWriteTokens int64 + // ToolNameMapResolved tracks whether ToolNameMap has been resolved from + // the original request once per stream. + ToolNameMapResolved bool + // StreamResolved and StreamRequested cache whether the original request + // asked for streaming, resolved once per stream so large request bodies + // are not re-parsed on every chunk. + StreamResolved bool + StreamRequested bool } // InterleavedContentChunk stores a buffered chunk of text or thinking arriving while a tool call is open @@ -131,21 +139,28 @@ func ConvertOpenAIResponseToClaude(_ context.Context, _ string, originalRequestR } rawJSON = bytes.TrimSpace(rawJSON[5:]) - if (*param).(*ConvertOpenAIResponseToAnthropicParams).ToolNameMap == nil { - (*param).(*ConvertOpenAIResponseToAnthropicParams).ToolNameMap = util.ToolNameMapFromClaudeRequest(originalRequestRawJSON) + state := (*param).(*ConvertOpenAIResponseToAnthropicParams) + if !state.ToolNameMapResolved { + if state.ToolNameMap == nil { + state.ToolNameMap = util.ToolNameMapFromClaudeRequest(originalRequestRawJSON) + } + state.ToolNameMapResolved = true } // Check if this is the [DONE] marker if bytes.Equal(bytes.TrimSpace(rawJSON), []byte("[DONE]")) { - return convertOpenAIDoneToAnthropic((*param).(*ConvertOpenAIResponseToAnthropicParams)) + return convertOpenAIDoneToAnthropic(state) } - streamResult := gjson.GetBytes(originalRequestRawJSON, "stream") - if !streamResult.Exists() || (streamResult.Exists() && streamResult.Type == gjson.False) { - return convertOpenAINonStreamingToAnthropic(rawJSON) - } else { - return convertOpenAIStreamingChunkToAnthropic(rawJSON, (*param).(*ConvertOpenAIResponseToAnthropicParams)) + if !state.StreamResolved { + streamResult := gjson.GetBytes(originalRequestRawJSON, "stream") + state.StreamRequested = streamResult.Exists() && streamResult.Type != gjson.False + state.StreamResolved = true } + if !state.StreamRequested { + return convertOpenAINonStreamingToAnthropic(rawJSON) + } + return convertOpenAIStreamingChunkToAnthropic(rawJSON, state) } func hasValidToolCallArguments(param *ConvertOpenAIResponseToAnthropicParams) bool { diff --git a/internal/translator/openai/claude/openai_claude_response_test.go b/internal/translator/openai/claude/openai_claude_response_test.go index 6dc811238..bfd5582c3 100644 --- a/internal/translator/openai/claude/openai_claude_response_test.go +++ b/internal/translator/openai/claude/openai_claude_response_test.go @@ -1468,3 +1468,79 @@ func TestExtractOpenAIUsage(t *testing.T) { }) } } + +const streamFlagTestChunk = `data: {"id":"chatcmpl_1","model":"test-model","created":1,"choices":[{"index":0,"delta":{"content":"hi"},"finish_reason":null}]}` + +func TestConvertOpenAIResponseToClaude_StreamFlagResolvedOncePerStream(t *testing.T) { + var param any + first := ConvertOpenAIResponseToClaude(context.Background(), "test-model", []byte(`{"model":"m","stream":true}`), nil, []byte(streamFlagTestChunk), ¶m) + if !bytes.Contains(bytes.Join(first, nil), []byte("content_block_delta")) { + t.Fatalf("expected streaming events on first chunk, got %q", bytes.Join(first, nil)) + } + + // An empty original request on subsequent chunks would parse as non-streaming if re-read. + second := ConvertOpenAIResponseToClaude(context.Background(), "test-model", nil, nil, []byte(streamFlagTestChunk), ¶m) + if !bytes.Contains(bytes.Join(second, nil), []byte("content_block_delta")) { + t.Fatalf("expected cached stream flag to be reused on subsequent chunks, got %q", bytes.Join(second, nil)) + } +} + +func TestConvertOpenAIResponseToClaude_ToolNameMapResolvedOnceWhenNoTools(t *testing.T) { + var param any + first := ConvertOpenAIResponseToClaude(context.Background(), "test-model", []byte(`{"model":"m","stream":true}`), nil, []byte(streamFlagTestChunk), ¶m) + if !bytes.Contains(bytes.Join(first, nil), []byte("content_block_delta")) { + t.Fatalf("expected streaming events on first chunk, got %q", bytes.Join(first, nil)) + } + + // Even when the first chunk resolved ToolNameMap to nil (no tools in request), + // later chunks must not re-parse originalRequestRawJSON to rebuild ToolNameMap. + toolChunk := `data: {"id":"chatcmpl_1","model":"test-model","created":1,"choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"id":"call_1","function":{"name":"bash","arguments":"{}"}}]},"finish_reason":null}]}` + laterReqWithTools := []byte(`{"model":"m","stream":true,"tools":[{"name":"Bash"}]}`) + second := ConvertOpenAIResponseToClaude(context.Background(), "test-model", laterReqWithTools, nil, []byte(toolChunk), ¶m) + joined := bytes.Join(second, nil) + if !bytes.Contains(joined, []byte(`"name":"bash"`)) { + t.Fatalf("expected cached nil ToolNameMap to be reused without re-parsing originalRequestRawJSON, got %q", joined) + } +} + +func TestConvertOpenAIResponseToClaude_NonStreamingRequestStaysNonStreaming(t *testing.T) { + for _, original := range []string{`{"model":"m"}`, `{"model":"m","stream":false}`} { + var param any + out := ConvertOpenAIResponseToClaude(context.Background(), "test-model", []byte(original), nil, []byte(streamFlagTestChunk), ¶m) + joined := bytes.Join(out, nil) + if len(out) != 1 || gjson.GetBytes(joined, "type").String() != "message" { + t.Fatalf("request %s expected a single non-streaming message payload, got %q", original, joined) + } + if bytes.Contains(joined, []byte("content_block_delta")) { + t.Fatalf("request %s must not produce streaming events, got %q", original, joined) + } + + // Subsequent call with stream:true must still use the cached non-streaming decision. + second := ConvertOpenAIResponseToClaude(context.Background(), "test-model", []byte(`{"model":"m","stream":true}`), nil, []byte(streamFlagTestChunk), ¶m) + secondJoined := bytes.Join(second, nil) + if len(second) != 1 || gjson.GetBytes(secondJoined, "type").String() != "message" || bytes.Contains(secondJoined, []byte("content_block_delta")) { + t.Fatalf("request %s expected cached non-streaming flag on second call, got %q", original, secondJoined) + } + } +} + +func BenchmarkConvertOpenAIResponseToClaude_LargeOriginalRequest(b *testing.B) { + var sb strings.Builder + sb.WriteString(`{"model":"m","tools":[`) + for i := 0; i < 60; i++ { + if i > 0 { + sb.WriteByte(',') + } + fmt.Fprintf(&sb, `{"name":"tool_%d","description":%q,"input_schema":{"type":"object","properties":{"a":{"type":"string"}}}}`, i, strings.Repeat("x", 5000)) + } + sb.WriteString(`],"stream":true}`) + original := []byte(sb.String()) + chunk := []byte(streamFlagTestChunk) + + var param any + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + ConvertOpenAIResponseToClaude(context.Background(), "test-model", original, nil, chunk, ¶m) + } +}