@@ -4,7 +4,9 @@ import (
44 "context"
55 "encoding/json"
66 "errors"
7+ "fmt"
78 "strings"
9+ "time"
810
911 "github.com/anthropics/anthropic-sdk-go"
1012 "github.com/anthropics/anthropic-sdk-go/option"
@@ -68,21 +70,24 @@ func (a *anthropicProvider) SendMessages(ctx context.Context, messages []message
6870 anthropicMessages := a.convertToAnthropicMessages(messages)
6971 anthropicTools := a.convertToAnthropicTools(tools)
7072
71− response, err := a.client.Messages.New(ctx, anthropic.MessageNewParams{
72− Model: anthropic.Model(a.model.APIModel),
73− MaxTokens: a.maxTokens,
74− Temperature: anthropic.Float(0),
75− Messages: anthropicMessages,
76− Tools: anthropicTools,
77− System: []anthropic.TextBlockParam{
78− {
79− Text: a.systemMessage,
80− CacheControl: anthropic.CacheControlEphemeralParam{
81− Type: "ephemeral",
73+ response, err := a.client.Messages.New(
74+ ctx,
75+ anthropic.MessageNewParams{
76+ Model: anthropic.Model(a.model.APIModel),
77+ MaxTokens: a.maxTokens,
78+ Temperature: anthropic.Float(0),
79+ Messages: anthropicMessages,
80+ Tools: anthropicTools,
81+ System: []anthropic.TextBlockParam{
82+ {
83+ Text: a.systemMessage,
84+ CacheControl: anthropic.CacheControlEphemeralParam{
85+ Type: "ephemeral",
86+ },
8287 },
8388 },
8489 },
85− })
90+ )
8691 if err != nil {
8792 return nil, err
8893 }
@@ -121,83 +126,171 @@ func (a *anthropicProvider) StreamResponse(ctx context.Context, messages []messa
121126 temperature = anthropic.Float(1)
122127 }
123128
124− stream := a.client.Messages.NewStreaming(ctx, anthropic.MessageNewParams{
125− Model: anthropic.Model(a.model.APIModel),
126− MaxTokens: a.maxTokens,
127− Temperature: temperature,
128− Messages: anthropicMessages,
129− Tools: anthropicTools,
130− Thinking: thinkingParam,
131− System: []anthropic.TextBlockParam{
132− {
133− Text: a.systemMessage,
134− CacheControl: anthropic.CacheControlEphemeralParam{
135− Type: "ephemeral",
136− },
137− },
138− },
139− })
140−
141129 eventChan := make(chan ProviderEvent)
142130
143131 go func() {
144132 defer close(eventChan)
145133
146− accumulatedMessage := anthropic.Message{}
134+ const maxRetries = 8
135+ attempts := 0
147136
148− for stream.Next() {
149− event := stream.Current()
150− err := accumulatedMessage.Accumulate(event)
151− if err != nil {
152− eventChan <- ProviderEvent{Type: EventError, Error: err}
153− return
137+ for {
138+ // If this isn't the first attempt, we're retrying
139+ if attempts > 0 {
140+ if attempts > maxRetries {
141+ eventChan <- ProviderEvent{
142+ Type: EventError,
143+ Error: errors.New("maximum retry attempts reached for rate limit (429)"),
144+ }
145+ return
146+ }
147+
148+ // Inform user we're retrying with attempt number
149+ eventChan <- ProviderEvent{
150+ Type: EventWarning,
151+ Info: fmt.Sprintf("[Retrying due to rate limit... attempt %d of %d]", attempts, maxRetries),
152+ }
153+
154+ // Calculate backoff with exponential backoff and jitter
155+ backoffMs := 2000 * (1 << (attempts - 1)) // 2s, 4s, 8s, 16s, 32s
156+ jitterMs := int(float64(backoffMs) * 0.2)
157+ totalBackoffMs := backoffMs + jitterMs
158+
159+ // Sleep with backoff, respecting context cancellation
160+ select {
161+ case <-ctx.Done():
162+ eventChan <- ProviderEvent{Type: EventError, Error: ctx.Err()}
163+ return
164+ case <-time.After(time.Duration(totalBackoffMs) * time.Millisecond):
165+ // Continue with retry
166+ }
154167 }
155168
156− switch event := event.AsAny().(type) {
157− case anthropic.ContentBlockStartEvent:
158− eventChan <- ProviderEvent{Type: EventContentStart}
169+ attempts++
170+
171+ // Create new streaming request
172+ stream := a.client.Messages.NewStreaming(
173+ ctx,
174+ anthropic.MessageNewParams{
175+ Model: anthropic.Model(a.model.APIModel),
176+ MaxTokens: a.maxTokens,
177+ Temperature: temperature,
178+ Messages: anthropicMessages,
179+ Tools: anthropicTools,
180+ Thinking: thinkingParam,
181+ System: []anthropic.TextBlockParam{
182+ {
183+ Text: a.systemMessage,
184+ CacheControl: anthropic.CacheControlEphemeralParam{
185+ Type: "ephemeral",
186+ },
187+ },
188+ },
189+ },
190+ )
159191
160− case anthropic.ContentBlockDeltaEvent:
161− if event.Delta.Type == "thinking_delta" && event.Delta.Thinking != "" {
162− eventChan <- ProviderEvent{
163− Type: EventThinkingDelta,
164− Thinking: event.Delta.Thinking,
192+ // Process stream events
193+ accumulatedMessage := anthropic.Message{}
194+ streamSuccess := false
195+
196+ // Process the stream until completion or error
197+ for stream.Next() {
198+ event := stream.Current()
199+ err := accumulatedMessage.Accumulate(event)
200+ if err != nil {
201+ eventChan <- ProviderEvent{Type: EventError, Error: err}
202+ return // Don't retry on accumulation errors
203+ }
204+
205+ switch event := event.AsAny().(type) {
206+ case anthropic.ContentBlockStartEvent:
207+ eventChan <- ProviderEvent{Type: EventContentStart}
208+
209+ case anthropic.ContentBlockDeltaEvent:
210+ if event.Delta.Type == "thinking_delta" && event.Delta.Thinking != "" {
211+ eventChan <- ProviderEvent{
212+ Type: EventThinkingDelta,
213+ Thinking: event.Delta.Thinking,
214+ }
215+ } else if event.Delta.Type == "text_delta" && event.Delta.Text != "" {
216+ eventChan <- ProviderEvent{
217+ Type: EventContentDelta,
218+ Content: event.Delta.Text,
219+ }
165220 }
166− } else if event.Delta.Type == "text_delta" && event.Delta.Text != "" {
167− eventChan <- ProviderEvent{
168− Type: EventContentDelta,
169− Content: event.Delta.Text,
221+
222+ case anthropic.ContentBlockStopEvent:
223+ eventChan <- ProviderEvent{Type: EventContentStop}
224+
225+ case anthropic.MessageStopEvent:
226+ streamSuccess = true
227+ content := ""
228+ for _, block := range accumulatedMessage.Content {
229+ if text, ok := block.AsAny().(anthropic.TextBlock); ok {
230+ content += text.Text
231+ }
170232 }
171− }
172233
173− case anthropic.ContentBlockStopEvent:
174− eventChan <- ProviderEvent{Type: EventContentStop}
234+ toolCalls := a.extractToolCalls(accumulatedMessage.Content)
235+ tokenUsage := a.extractTokenUsage(accumulatedMessage.Usage)
175236
176− case anthropic.MessageStopEvent:
177− content := ""
178− for _, block := range accumulatedMessage.Content {
179− if text, ok := block.AsAny().(anthropic.TextBlock); ok {
180− content += text.Text
237+ eventChan <- ProviderEvent{
238+ Type: EventComplete,
239+ Response: &ProviderResponse{
240+ Content: content,
241+ ToolCalls: toolCalls,
242+ Usage: tokenUsage,
243+ FinishReason: string(accumulatedMessage.StopReason),
244+ },
181245 }
182246 }
247+ }
183248
184− toolCalls := a.extractToolCalls(accumulatedMessage.Content)
185− tokenUsage := a.extractTokenUsage(accumulatedMessage.Usage)
249+ // If the stream completed successfully, we're done
250+ if streamSuccess {
251+ return
252+ }
186253
187− eventChan <- ProviderEvent{
188− Type: EventComplete,
189− Response: &ProviderResponse{
190− Content: content,
191− ToolCalls: toolCalls,
192− Usage: tokenUsage,
193− FinishReason: string(accumulatedMessage.StopReason),
194− },
254+ // Check for stream errors
255+ err := stream.Err()
256+ if err != nil {
257+ var apierr *anthropic.Error
258+ if errors.As(err, &apierr) {
259+ if apierr.StatusCode == 429 || apierr.StatusCode == 529 {
260+ // Check for Retry-After header
261+ if retryAfterValues := apierr.Response.Header.Values("Retry-After"); len(retryAfterValues) > 0 {
262+ // Parse the retry after value (seconds)
263+ var retryAfterSec int
264+ if _, err := fmt.Sscanf(retryAfterValues[0], "%d", &retryAfterSec); err == nil {
265+ retryMs := retryAfterSec * 1000
266+
267+ // Inform user of retry with specific wait time
268+ eventChan <- ProviderEvent{
269+ Type: EventWarning,
270+ Info: fmt.Sprintf("[Rate limited: waiting %d seconds as specified by API]", retryAfterSec),
271+ }
272+
273+ // Sleep respecting context cancellation
274+ select {
275+ case <-ctx.Done():
276+ eventChan <- ProviderEvent{Type: EventError, Error: ctx.Err()}
277+ return
278+ case <-time.After(time.Duration(retryMs) * time.Millisecond):
279+ // Continue with retry after specified delay
280+ continue
281+ }
282+ }
283+ }
284+
285+ // Fall back to exponential backoff if Retry-After parsing failed
286+ continue
287+ }
195288 }
196− }
197− }
198289
199− if stream.Err() != nil {
200− eventChan <- ProviderEvent{Type: EventError, Error: stream.Err()}
290+ // For non-rate limit errors, report and exit
291+ eventChan <- ProviderEvent{Type: EventError, Error: err}
292+ return
293+ }
201294 }
202295 }()
203296
@@ -311,3 +404,4 @@ func (a *anthropicProvider) convertToAnthropicMessages(messages []message.Messag
311404
312405 return anthropicMessages
313406}
407+