mirror of
https://github.com/Wei-Shaw/sub2api.git
synced 2026-10-14 03:52:26 +08:00
优化 Responses SSE 事件类型解析
This commit is contained in:
@@ -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:
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
})
|
||||
}
|
||||
Reference in New Issue
Block a user