anthropic: add improved streaming thinking/reasoning token support (#1418)

llms/anthropic: add streaming thinking/reasoning token support

Implement StreamingReasoningFunc support in the Anthropic client to enable
real-time streaming of thinking tokens during extended thinking responses.

Changes:
- Add StreamingReasoningFunc field to messagePayload, MessageRequest structs
- Modify handleThinkingDelta() to call StreamingReasoningFunc when thinking
  chunks arrive during streaming
- Wire up StreamingReasoningFunc from llms.CallOptions through to the
  Anthropic client payload
- Update setMessageDefaults to enable streaming when StreamingReasoningFunc
  is provided

This follows the same pattern as the OpenAI client (chat.go:638-663) and
enables thinking tokens to stream in real-time at the BEGINNING of the
response, rather than appearing after the response completes.

Fixes issue where thinking_delta events were not calling the streaming
reasoning callback, causing thinking content to only be available after
response completion.

Author: tmc

llms/anthropic/anthropicllm.go

@@ -146,17 +146,18 @@ func generateMessagesContent(ctx context.Context, o *LLM, messages []llms.Messag
 	betaHeaders, thinking := extractThinkingOptions(o, opts)
 
 	result, err := o.client.CreateMessage(ctx, &anthropicclient.MessageRequest{
-		Model:         opts.Model,
-		Messages:      chatMessages,
-		System:        systemPrompt,
-		MaxTokens:     opts.MaxTokens,
-		StopWords:     opts.StopWords,
-		Temperature:   opts.Temperature,
-		TopP:          opts.TopP,
-		Tools:         tools,
-		Thinking:      thinking,
-		BetaHeaders:   betaHeaders,
-		StreamingFunc: opts.StreamingFunc,
+		Model:                  opts.Model,
+		Messages:               chatMessages,
+		System:                 systemPrompt,
+		MaxTokens:              opts.MaxTokens,
+		StopWords:              opts.StopWords,
+		Temperature:            opts.Temperature,
+		TopP:                   opts.TopP,
+		Tools:                  tools,
+		Thinking:               thinking,
+		BetaHeaders:            betaHeaders,
+		StreamingFunc:          opts.StreamingFunc,
+		StreamingReasoningFunc: opts.StreamingReasoningFunc,
 	})
 	if err != nil {
 		if o.CallbacksHandler != nil {

llms/anthropic/internal/anthropicclient/anthropicclient.go

@@ -141,24 +141,26 @@ type MessageRequest struct {
 	Thinking *ThinkingConfig `json:"thinking,omitempty"`
 
 	// BetaHeaders are additional beta feature headers to include
-	BetaHeaders   []string                                      `json:"-"`
-	StreamingFunc func(ctx context.Context, chunk []byte) error `json:"-"`
+	BetaHeaders            []string                                                      `json:"-"`
+	StreamingFunc          func(ctx context.Context, chunk []byte) error                `json:"-"`
+	StreamingReasoningFunc func(ctx context.Context, reasoningChunk, chunk []byte) error `json:"-"`
 }
 
 // CreateMessage creates message for the messages api.
 func (c *Client) CreateMessage(ctx context.Context, r *MessageRequest) (*MessageResponsePayload, error) {
 	resp, err := c.createMessage(ctx, &messagePayload{
-		Model:         r.Model,
-		Messages:      r.Messages,
-		System:        r.System,
-		Temperature:   r.Temperature,
-		MaxTokens:     r.MaxTokens,
-		StopWords:     r.StopWords,
-		TopP:          r.TopP,
-		Tools:         r.Tools,
-		Stream:        r.Stream,
-		Thinking:      r.Thinking,
-		StreamingFunc: r.StreamingFunc,
+		Model:                  r.Model,
+		Messages:               r.Messages,
+		System:                 r.System,
+		Temperature:            r.Temperature,
+		MaxTokens:              r.MaxTokens,
+		StopWords:              r.StopWords,
+		TopP:                   r.TopP,
+		Tools:                  r.Tools,
+		Stream:                 r.Stream,
+		Thinking:               r.Thinking,
+		StreamingFunc:          r.StreamingFunc,
+		StreamingReasoningFunc: r.StreamingReasoningFunc,
 	}, r.BetaHeaders)
 	if err != nil {
 		return nil, err

llms/anthropic/internal/anthropicclient/messages.go

@@ -47,7 +47,8 @@ type messagePayload struct {
 	// Extended thinking parameters (Claude 3.7+)
 	Thinking *ThinkingConfig `json:"thinking,omitempty"`
 
-	StreamingFunc func(ctx context.Context, chunk []byte) error `json:"-"`
+	StreamingFunc          func(ctx context.Context, chunk []byte) error                      `json:"-"`
+	StreamingReasoningFunc func(ctx context.Context, reasoningChunk, chunk []byte) error `json:"-"`
 }
 
 // ThinkingConfig represents the thinking configuration for Claude 3.7+
@@ -219,7 +220,7 @@ func (c *Client) setMessageDefaults(payload *messagePayload) {
 	default:
 		payload.Model = defaultModel
 	}
-	if payload.StreamingFunc != nil {
+	if payload.StreamingFunc != nil || payload.StreamingReasoningFunc != nil {
 		payload.Stream = true
 	}
 }
@@ -440,7 +441,7 @@ func handleContentBlockDeltaEvent(ctx context.Context, event map[string]interfac
 	case "input_json_delta":
 		return handleJSONDelta(delta, response, index)
 	case "thinking_delta":
-		return handleThinkingDelta(delta, response, index)
+		return handleThinkingDelta(ctx, delta, response, payload, index)
 	}
 
 	return response, nil
@@ -485,7 +486,7 @@ func handleJSONDelta(delta map[string]interface{}, response MessageResponsePaylo
 }
 
 // handleThinkingDelta processes thinking delta events for content blocks.
-func handleThinkingDelta(delta map[string]interface{}, response MessageResponsePayload, index int) (MessageResponsePayload, error) {
+func handleThinkingDelta(ctx context.Context, delta map[string]interface{}, response MessageResponsePayload, payload *messagePayload, index int) (MessageResponsePayload, error) {
 	thinking, ok := delta["thinking"].(string)
 	if !ok {
 		return response, ErrInvalidDeltaTextField
@@ -495,6 +496,17 @@ func handleThinkingDelta(delta map[string]interface{}, response MessageResponseP
 		return response, fmt.Errorf("failed to cast to ThinkingContent at index %d", index)
 	}
 	thinkingContent.Thinking += thinking
+
+	// Call StreamingReasoningFunc if provided (similar to OpenAI pattern)
+	if payload.StreamingReasoningFunc != nil {
+		reasoningChunk := []byte(thinking)
+		// For thinking deltas, the content chunk is empty since this is pure reasoning
+		err := payload.StreamingReasoningFunc(ctx, reasoningChunk, []byte{})
+		if err != nil {
+			return response, fmt.Errorf("streaming reasoning func returned an error: %w", err)
+		}
+	}
+
 	return response, nil
 }