opencode/picoclaw
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(openai_compat): parse SSE events and reasoning variants in streams

Browse source
This commit is contained in:
lc6464 2026-05-14 10:20:01 +08:00
parent b00ff5bc5d
commit 10f4466a7e
No known key found for this signature in database
GPG key ID: 53C61B42FEC71D6D
2 changed files with 176 additions and 19 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

84
pkg/providers/openai_compat/provider.go
View file

@ -420,6 +420,8 @@ func parseStreamResponse(
) (*LLMResponse, error) {
var textContent strings.Builder
var reasoningContent strings.Builder
var reasoning strings.Builder
var reasoningDetails []ReasoningDetail
var finishReason string
var usage *UsageInfo
@ -431,29 +433,21 @@ func parseStreamResponse(
}
activeTools := map[int]*toolAccum{}
scanner := bufio.NewScanner(reader)
scanner.Buffer(make([]byte, 0, 1024*1024), 10*1024*1024) // 1MB initial, 10MB max
for scanner.Scan() {
// Check for context cancellation between chunks
if err := ctx.Err(); err != nil {
return nil, err
processEvent := func(data string) error {
if strings.TrimSpace(data) == "" {
return nil
}
line := scanner.Text()
if !strings.HasPrefix(line, "data: ") {
continue
}
data := strings.TrimPrefix(line, "data: ")
if data == "[DONE]" {
break
if strings.TrimSpace(data) == "[DONE]" {
return io.EOF
}
var chunk struct {
Choices []struct {
Delta struct {
Content string `json:"content"`
ReasoningContent string `json:"reasoning_content"`
Content string `json:"content"`
ReasoningContent string `json:"reasoning_content"`
Reasoning string `json:"reasoning"`
ReasoningDetails []ReasoningDetail `json:"reasoning_details"`
ToolCalls []struct {
Index int `json:"index"`
ID string `json:"id"`
@ -469,7 +463,7 @@ func parseStreamResponse(
}
if err := json.Unmarshal([]byte(data), &chunk); err != nil {
continue // skip malformed chunks
return fmt.Errorf("failed to decode stream event: %w", err)
}
if chunk.Usage != nil {
@ -477,7 +471,7 @@ func parseStreamResponse(
}
if len(chunk.Choices) == 0 {
continue
return nil
}
choice := chunk.Choices[0]
@ -492,6 +486,12 @@ func parseStreamResponse(
if choice.Delta.ReasoningContent != "" {
reasoningContent.WriteString(choice.Delta.ReasoningContent)
}
if choice.Delta.Reasoning != "" {
reasoning.WriteString(choice.Delta.Reasoning)
}
if len(choice.Delta.ReasoningDetails) > 0 {
reasoningDetails = append(reasoningDetails, choice.Delta.ReasoningDetails...)
}
// Accumulate tool call deltas
for _, tc := range choice.Delta.ToolCalls {
@ -516,11 +516,55 @@ func parseStreamResponse(
if choice.FinishReason != nil {
finishReason = *choice.FinishReason
}
return nil
}
scanner := bufio.NewScanner(reader)
scanner.Buffer(make([]byte, 0, 1024*1024), 10*1024*1024) // 1MB initial, 10MB max
var eventData strings.Builder
for scanner.Scan() {
// Check for context cancellation between chunks
if err := ctx.Err(); err != nil {
return nil, err
}
line := scanner.Text()
if line == "" {
err := processEvent(eventData.String())
if err == io.EOF {
break
}
if err != nil {
return nil, err
}
eventData.Reset()
continue
}
if strings.HasPrefix(line, ":") {
continue
}
if !strings.HasPrefix(line, "data:") {
continue
}
data := strings.TrimPrefix(line, "data:")
data = strings.TrimPrefix(data, " ")
if eventData.Len() > 0 {
eventData.WriteByte('\n')
}
eventData.WriteString(data)
}
if err := scanner.Err(); err != nil {
return nil, fmt.Errorf("streaming read error: %w", err)
}
if eventData.Len() > 0 {
err := processEvent(eventData.String())
if err != nil && err != io.EOF {
return nil, err
}
}
// Assemble tool calls from accumulated deltas
var toolCalls []ToolCall
@ -551,6 +595,8 @@ func parseStreamResponse(
return &LLMResponse{
Content: textContent.String(),
ReasoningContent: reasoningContent.String(),
Reasoning: reasoning.String(),
ReasoningDetails: reasoningDetails,
ToolCalls: toolCalls,
FinishReason: finishReason,
Usage: usage,

111
pkg/providers/openai_compat/provider_test.go
View file

@ -1246,6 +1246,117 @@ func TestProviderChatStream_ParsesReasoningContent(t *testing.T) {
}
}
func TestProviderChatStream_ParsesMultilineSSEEvent(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "text/event-stream")
_, _ = w.Write([]byte(
"data: {\"choices\":[{\"delta\":{\n" +
"data: \"content\":\"Hello\",\"reasoning_content\":\"Thinking\",\n" +
"data: \"tool_calls\":[{\"index\":0,\"id\":\"call_1\",\"function\":{\"name\":\"echo\",\"arguments\":\"{\\\"message\\\":\\\"hello\\\"}\"}}]},\"finish_reason\":\"tool_calls\"}],\n" +
"data: \"usage\":{\"prompt_tokens\":3,\"completion_tokens\":4,\"total_tokens\":7}}\n\n",
))
_, _ = w.Write([]byte("data: [DONE]\n\n"))
}))
defer server.Close()
p := NewProvider("key", server.URL, "")
out, err := p.ChatStream(
t.Context(),
[]Message{{Role: "user", Content: "say hello"}},
nil,
"gpt-4o",
nil,
nil,
)
if err != nil {
t.Fatalf("ChatStream() error = %v", err)
}
if out.Content != "Hello" {
t.Fatalf("Content = %q, want %q", out.Content, "Hello")
}
if out.ReasoningContent != "Thinking" {
t.Fatalf("ReasoningContent = %q, want %q", out.ReasoningContent, "Thinking")
}
if len(out.ToolCalls) != 1 {
t.Fatalf("len(ToolCalls) = %d, want 1", len(out.ToolCalls))
}
if out.ToolCalls[0].Name != "echo" {
t.Fatalf("ToolCalls[0].Name = %q, want %q", out.ToolCalls[0].Name, "echo")
}
if out.ToolCalls[0].Arguments["message"] != "hello" {
t.Fatalf("ToolCalls[0].Arguments[message] = %v, want %q", out.ToolCalls[0].Arguments["message"], "hello")
}
if out.FinishReason != "tool_calls" {
t.Fatalf("FinishReason = %q, want %q", out.FinishReason, "tool_calls")
}
if out.Usage == nil || out.Usage.TotalTokens != 7 {
t.Fatalf("Usage = %#v, want total tokens 7", out.Usage)
}
}
func TestProviderChatStream_ParsesReasoningVariants(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "text/event-stream")
_, _ = w.Write([]byte(
"data: {\"choices\":[{\"delta\":{\"reasoning\":\"step 1\",\"reasoning_details\":[{\"format\":\"text\",\"index\":0,\"type\":\"summary\",\"text\":\"first\"}]}}]}\n\n",
))
_, _ = w.Write([]byte(
"data: {\"choices\":[{\"delta\":{\"reasoning\":\" + step 2\",\"reasoning_details\":[{\"format\":\"text\",\"index\":1,\"type\":\"summary\",\"text\":\"second\"}],\"content\":\"done\"},\"finish_reason\":\"stop\"}]}\n\n",
))
_, _ = w.Write([]byte("data: [DONE]\n\n"))
}))
defer server.Close()
p := NewProvider("key", server.URL, "")
out, err := p.ChatStream(
t.Context(),
[]Message{{Role: "user", Content: "think"}},
nil,
"gpt-4o",
nil,
nil,
)
if err != nil {
t.Fatalf("ChatStream() error = %v", err)
}
if out.Content != "done" {
t.Fatalf("Content = %q, want %q", out.Content, "done")
}
if out.Reasoning != "step 1 + step 2" {
t.Fatalf("Reasoning = %q, want %q", out.Reasoning, "step 1 + step 2")
}
if len(out.ReasoningDetails) != 2 {
t.Fatalf("len(ReasoningDetails) = %d, want 2", len(out.ReasoningDetails))
}
if out.ReasoningDetails[0].Text != "first" || out.ReasoningDetails[1].Text != "second" {
t.Fatalf("ReasoningDetails = %#v, want texts first/second", out.ReasoningDetails)
}
}
func TestProviderChatStream_InvalidEventReturnsError(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "text/event-stream")
_, _ = w.Write([]byte("data: {\"choices\":[\n\n"))
}))
defer server.Close()
p := NewProvider("key", server.URL, "")
_, err := p.ChatStream(
t.Context(),
[]Message{{Role: "user", Content: "hi"}},
nil,
"gpt-4o",
nil,
nil,
)
if err == nil {
t.Fatal("expected error for malformed stream event")
}
if !strings.Contains(err.Error(), "failed to decode stream event") {
t.Fatalf("error = %v, want decode stream event error", err)
}
}
type roundTripperFunc func(*http.Request) (*http.Response, error)
func (f roundTripperFunc) RoundTrip(r *http.Request) (*http.Response, error) {