diff --git a/pkg/providers/openai_compat/provider.go b/pkg/providers/openai_compat/provider.go index 7003709a7..e15a40e41 100644 --- a/pkg/providers/openai_compat/provider.go +++ b/pkg/providers/openai_compat/provider.go @@ -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, diff --git a/pkg/providers/openai_compat/provider_test.go b/pkg/providers/openai_compat/provider_test.go index fbda447b4..7d7c31586 100644 --- a/pkg/providers/openai_compat/provider_test.go +++ b/pkg/providers/openai_compat/provider_test.go @@ -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) {