Merge pull request #2741 from lc6464/fix/deepseek-stream-reasoning-content

fix(openai_compat): parse reasoning_content in streaming responses

pkg/providers/openai_compat/provider.go

@@ -419,6 +419,9 @@ func parseStreamResponse(
 	onChunk func(accumulated string),
 ) (*LLMResponse, error) {
 	var textContent strings.Builder
+	var reasoningContent strings.Builder
+	var reasoning strings.Builder
+	var reasoningDetails []ReasoningDetail
 	var finishReason string
 	var usage *UsageInfo
 
@@ -430,28 +433,21 @@ func parseStreamResponse(
 	}
 	activeTools := map[int]*toolAccum{}
 
-	scanner := bufio.NewScanner(reader)
+	processEvent := func(data string) error {
-	scanner.Buffer(make([]byte, 0, 1024*1024), 10*1024*1024) // 1MB initial, 10MB max
+		if strings.TrimSpace(data) == "" {
-	for scanner.Scan() {
+			return nil
-		// Check for context cancellation between chunks
-		if err := ctx.Err(); err != nil {
-			return nil, err
 		}
-
+		if strings.TrimSpace(data) == "[DONE]" {
-		line := scanner.Text()
+			return io.EOF
-
-		if !strings.HasPrefix(line, "data: ") {
-			continue
-		}
-		data := strings.TrimPrefix(line, "data: ")
-		if data == "[DONE]" {
-			break
 		}
 
 		var chunk struct {
 			Choices []struct {
 				Delta struct {
 					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"`
@@ -467,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 {
@@ -475,7 +471,7 @@ func parseStreamResponse(
 		}
 
 		if len(chunk.Choices) == 0 {
-			continue
+			return nil
 		}
 
 		choice := chunk.Choices[0]
@@ -487,6 +483,15 @@ func parseStreamResponse(
 				onChunk(textContent.String())
 			}
 		}
+		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 {
@@ -511,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
@@ -545,6 +594,9 @@ func parseStreamResponse(
 
 	return &LLMResponse{
 		Content:          textContent.String(),
+		ReasoningContent: reasoningContent.String(),
+		Reasoning:        reasoning.String(),
+		ReasoningDetails: reasoningDetails,
 		ToolCalls:        toolCalls,
 		FinishReason:     finishReason,
 		Usage:            usage,

pkg/providers/openai_compat/provider_test.go

@@ -1195,6 +1195,168 @@ func TestProviderChatStream_CustomHeadersInjected(t *testing.T) {
 	}
 }
 
+func TestProviderChatStream_ParsesReasoningContent(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_content\":\"Let me \",\"content\":\"Checking \",\"tool_calls\":[{\"index\":0,\"id\":\"call_1\",\"function\":{\"name\":\"get_weather\",\"arguments\":\"{\\\"city\\\":\"}}]}}]}\n\n",
+		))
+		_, _ = w.Write([]byte(
+			"data: {\"choices\":[{\"delta\":{\"reasoning_content\":\"think step by step.\",\"content\":\"the weather\",\"tool_calls\":[{\"index\":0,\"function\":{\"arguments\":\"\\\"Hangzhou\\\"}\"}}]},\"finish_reason\":\"tool_calls\"}],\"usage\":{\"prompt_tokens\":10,\"completion_tokens\":6,\"total_tokens\":16}}\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: "weather?"}},
+		nil,
+		"deepseek-v4-flash",
+		nil,
+		nil,
+	)
+	if err != nil {
+		t.Fatalf("ChatStream() error = %v", err)
+	}
+	if out.Content != "Checking the weather" {
+		t.Fatalf("Content = %q, want %q", out.Content, "Checking the weather")
+	}
+	if out.ReasoningContent != "Let me think step by step." {
+		t.Fatalf("ReasoningContent = %q, want %q", out.ReasoningContent, "Let me think step by step.")
+	}
+	if len(out.ToolCalls) != 1 {
+		t.Fatalf("len(ToolCalls) = %d, want 1", len(out.ToolCalls))
+	}
+	if out.ToolCalls[0].ID != "call_1" {
+		t.Fatalf("ToolCalls[0].ID = %q, want %q", out.ToolCalls[0].ID, "call_1")
+	}
+	if out.ToolCalls[0].Name != "get_weather" {
+		t.Fatalf("ToolCalls[0].Name = %q, want %q", out.ToolCalls[0].Name, "get_weather")
+	}
+	if out.ToolCalls[0].Arguments["city"] != "Hangzhou" {
+		t.Fatalf("ToolCalls[0].Arguments[city] = %v, want %q", out.ToolCalls[0].Arguments["city"], "Hangzhou")
+	}
+	if out.FinishReason != "tool_calls" {
+		t.Fatalf("FinishReason = %q, want %q", out.FinishReason, "tool_calls")
+	}
+	if out.Usage == nil || out.Usage.TotalTokens != 16 {
+		t.Fatalf("Usage = %#v, want total tokens 16", out.Usage)
+	}
+}
+
+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) {