Merge pull request #4601 from jianjianai/new/perf-responses-sse-type

perf(openai): 复用 Responses SSE 事件类型解析

backend/internal/service/gateway_service.go

@@ -173,7 +173,23 @@ func openAIStreamEventIsTerminal(data string) bool {
 	if trimmed == "[DONE]" {
 		return true
 	}
-	switch gjson.Get(trimmed, "type").String() {
+	return openAIStreamEventTypeIsTerminal(gjson.Get(trimmed, "type").String())
+}
+
+// openAIStreamEventIsTerminalWithType 复用已提取的 type，避免 SSE 热路径重复扫描 JSON。
+func openAIStreamEventIsTerminalWithType(data, eventType string) bool {
+	trimmed := strings.TrimSpace(data)
+	if trimmed == "" {
+		return false
+	}
+	if trimmed == "[DONE]" {
+		return true
+	}
+	return openAIStreamEventTypeIsTerminal(eventType)
+}
+
+func openAIStreamEventTypeIsTerminal(eventType string) bool {
+	switch eventType {
 	case "response.completed", "response.done", "response.failed", "response.incomplete", "response.cancelled", "response.canceled":
 		return true
 	default:

backend/internal/service/openai_gateway_response_flush_test.go

@@ -402,6 +402,39 @@ func TestOpenAIResponseFlush_FailedAndErrorEventsFlushAtBoundaries(t *testing.T)
 	})
 }
 
+func TestOpenAIResponseFlush_ReusedTypeKeepsSSEBytesAndTerminalSemantics(t *testing.T) {
+	tests := []struct {
+		name       string
+		body       string
+		flushCount int
+	}{
+		{
+			name:       "whitespace around done",
+			body:       "data: \t[DONE]  \n\n",
+			flushCount: 1,
+		},
+		{
+			name:       "invalid JSON before done",
+			body:       "data: {\"type\":\n\ndata: [DONE]\n\n",
+			flushCount: 2,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			recorder := newOpenAIResponseFlushRecorder()
+
+			result, err := runOpenAIResponseFlushTest(recorder, io.NopCloser(strings.NewReader(tt.body)), config.GatewayConfig{})
+
+			require.NoError(t, err)
+			require.NotNil(t, result)
+			gotBody, flushes := recorder.snapshot()
+			require.Equal(t, tt.body, gotBody)
+			require.Len(t, flushes, tt.flushCount)
+		})
+	}
+}
+
 func TestOpenAIResponseFlush_ClientDisconnectStillDrainsUsage(t *testing.T) {
 	first := "data: {\"type\":\"response.output_text.delta\",\"delta\":\"a\"}\n\n"
 	terminal := "data: {\"type\":\"response.completed\",\"response\":{\"status\":\"completed\",\"output\":[],\"usage\":{\"input_tokens\":7,\"output_tokens\":5,\"input_tokens_details\":{\"cached_tokens\":2}}}}\n\n"

backend/internal/service/openai_gateway_response_handling.go

@@ -396,10 +396,12 @@ func (s *OpenAIGatewayService) handleStreamingResponseWithReasoning(ctx context.
 		// Extract data from SSE line (supports both "data: " and "data:" formats)
 		if data, ok := extractOpenAISSEDataLine(line); ok {
 			dataBytes := []byte(data)
-			if openAIStreamEventIsTerminal(data) {
+			eventTypeRaw := gjson.GetBytes(dataBytes, "type").String()
+			eventType := strings.TrimSpace(eventTypeRaw)
+			// 初始上游 data 的 type 只解析一次：原始值保持终止事件的精确匹配，规范化值供后续分支复用。
+			if openAIStreamEventIsTerminalWithType(data, eventTypeRaw) {
 				sawTerminalEvent = true
 			}
-			eventType := strings.TrimSpace(gjson.GetBytes(dataBytes, "type").String())
 			if responseID == "" {
 				responseID = extractOpenAIResponseIDFromJSONBytes(dataBytes)
 			}

backend/internal/service/openai_gateway_response_handling_type_test.go

@@ -0,0 +1,74 @@
+package service
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"github.com/tidwall/gjson"
+)
+
+func TestOpenAIStreamEventIsTerminalWithTypeMatchesExistingSemantics(t *testing.T) {
+	tests := []struct {
+		name string
+		data string
+		want bool
+	}{
+		{name: "empty", data: "", want: false},
+		{name: "whitespace", data: " \t ", want: false},
+		{name: "done", data: " [DONE] ", want: true},
+		{name: "JSON outer whitespace", data: " \n\t {\"type\":\"response.completed\"} \r\n", want: true},
+		{name: "completed", data: `{"type":"response.completed"}`, want: true},
+		{name: "response done", data: `{"type":"response.done"}`, want: true},
+		{name: "failed", data: `{"type":"response.failed"}`, want: true},
+		{name: "incomplete", data: `{"type":"response.incomplete"}`, want: true},
+		{name: "cancelled", data: `{"type":"response.cancelled"}`, want: true},
+		{name: "canceled", data: `{"type":"response.canceled"}`, want: true},
+		{name: "delta", data: `{"type":"response.output_text.delta"}`, want: false},
+		{name: "invalid JSON", data: `{"type":`, want: false},
+		{name: "terminal with trailing garbage", data: `{"type":"response.completed"} trailing`, want: true},
+		{name: "nonterminal with trailing garbage", data: `{"type":"response.output_text.delta"} trailing`, want: false},
+		{name: "type whitespace remains nonterminal", data: `{"type":" response.completed "}`, want: false},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			eventType := gjson.GetBytes([]byte(tt.data), "type").String()
+			got := openAIStreamEventIsTerminalWithType(tt.data, eventType)
+
+			require.Equal(t, tt.want, got)
+			require.Equal(t, openAIStreamEventIsTerminal(tt.data), got)
+		})
+	}
+}
+
+var (
+	benchmarkOpenAIResponseSSEEventTypeSink string
+	benchmarkOpenAIResponseSSETerminalSink  bool
+)
+
+func BenchmarkOpenAIResponseSSETypeExtraction(b *testing.B) {
+	data := `{"type":"response.output_text.delta","sequence_number":42,"delta":"streaming response benchmark payload"}`
+	dataBytes := []byte(data)
+
+	b.Run("legacy double parse", func(b *testing.B) {
+		b.ReportAllocs()
+		b.SetBytes(int64(len(dataBytes)))
+		b.ResetTimer()
+		for i := 0; i < b.N; i++ {
+			benchmarkOpenAIResponseSSETerminalSink = openAIStreamEventIsTerminal(data)
+			benchmarkOpenAIResponseSSEEventTypeSink = strings.TrimSpace(gjson.GetBytes(dataBytes, "type").String())
+		}
+	})
+
+	b.Run("reused single parse", func(b *testing.B) {
+		b.ReportAllocs()
+		b.SetBytes(int64(len(dataBytes)))
+		b.ResetTimer()
+		for i := 0; i < b.N; i++ {
+			eventTypeRaw := gjson.GetBytes(dataBytes, "type").String()
+			benchmarkOpenAIResponseSSEEventTypeSink = strings.TrimSpace(eventTypeRaw)
+			benchmarkOpenAIResponseSSETerminalSink = openAIStreamEventIsTerminalWithType(data, eventTypeRaw)
+		}
+	})
+}