git.iko.soy/crush
on main help sign in
fix: handle anthropic 429s
57a2210adamdottv
internal/llm/provider/anthropic.go+124−62
compare with main
unifiedsplit
@@ -4,7 +4,10 @@ import (
4 "context"
4 "context"
5 "encoding/json"
5 "encoding/json"
6 "errors"
6 "errors"
7+ "fmt"
8+ "log"
7 "strings"
9 "strings"
10+ "time"
8
11
9 "github.com/anthropics/anthropic-sdk-go"
12 "github.com/anthropics/anthropic-sdk-go"
10 "github.com/anthropics/anthropic-sdk-go/option"
13 "github.com/anthropics/anthropic-sdk-go/option"
@@ -125,87 +128,145 @@ func (a *anthropicProvider) StreamResponse(ctx context.Context, messages []messa
125 temperature = anthropic.Float(1)
128 temperature = anthropic.Float(1)
126 }
129 }
127
130
128− stream := a.client.Messages.NewStreaming(
129− ctx,
130− anthropic.MessageNewParams{
131− Model: anthropic.Model(a.model.APIModel),
132− MaxTokens: a.maxTokens,
133− Temperature: temperature,
134− Messages: anthropicMessages,
135− Tools: anthropicTools,
136− Thinking: thinkingParam,
137− System: []anthropic.TextBlockParam{
138− {
139− Text: a.systemMessage,
140− CacheControl: anthropic.CacheControlEphemeralParam{
141− Type: "ephemeral",
142− },
143− },
144− },
145− },
146− option.WithMaxRetries(8),
147− )
148−
149 eventChan := make(chan ProviderEvent)
131 eventChan := make(chan ProviderEvent)
150
132
151 go func() {
133 go func() {
152 defer close(eventChan)
134 defer close(eventChan)
153
135
154− accumulatedMessage := anthropic.Message{}
136+ const maxRetries = 8
137+ attempts := 0
155
138
156− for stream.Next() {
139+ for {
157− event := stream.Current()
140+ // If this isn't the first attempt, we're retrying
158− err := accumulatedMessage.Accumulate(event)
141+ if attempts > 0 {
159− if err != nil {
142+ if attempts > maxRetries {
143+ eventChan <- ProviderEvent{
144+ Type: EventError,
145+ Error: errors.New("maximum retry attempts reached for rate limit (429)"),
146+ }
147+ return
148+ }
149+
150+ // Inform user we're retrying with attempt number
151+ eventChan <- ProviderEvent{
152+ Type: EventContentDelta,
153+ Content: fmt.Sprintf("\n\n[Retrying due to rate limit... attempt %d of %d]\n\n", attempts, maxRetries),
154+ }
155+
156+ // Calculate backoff with exponential backoff and jitter
157+ backoffMs := 2000 * (1 << (attempts - 1)) // 2s, 4s, 8s, 16s, 32s
158+ jitterMs := int(float64(backoffMs) * 0.2)
159+ totalBackoffMs := backoffMs + jitterMs
160+
161+ // Sleep with backoff, respecting context cancellation
162+ select {
163+ case <-ctx.Done():
160− eventChan <- ProviderEvent{Type: EventError, Error: err}
164+ eventChan <- ProviderEvent{Type: EventError, Error: ctx.Err()}
161− return
165+ return
166+ case <-time.After(time.Duration(totalBackoffMs) * time.Millisecond):
167+ // Continue with retry
168+ }
162 }
169 }
163
170
164− switch event := event.AsAny().(type) {
171+ attempts++
165− case anthropic.ContentBlockStartEvent:
172+
166− eventChan <- ProviderEvent{Type: EventContentStart}
173+ // Create new streaming request
174+ stream := a.client.Messages.NewStreaming(
175+ ctx,
176+ anthropic.MessageNewParams{
177+ Model: anthropic.Model(a.model.APIModel),
178+ MaxTokens: a.maxTokens,
179+ Temperature: temperature,
180+ Messages: anthropicMessages,
181+ Tools: anthropicTools,
182+ Thinking: thinkingParam,
183+ System: []anthropic.TextBlockParam{
184+ {
185+ Text: a.systemMessage,
186+ CacheControl: anthropic.CacheControlEphemeralParam{
187+ Type: "ephemeral",
188+ },
189+ },
190+ },
191+ },
192+ )
167
193
194+ // Process stream events
195+ accumulatedMessage := anthropic.Message{}
196+ streamSuccess := false
197+
198+ // Process the stream until completion or error
199+ for stream.Next() {
200+ event := stream.Current()
201+ err := accumulatedMessage.Accumulate(event)
202+ if err != nil {
203+ eventChan <- ProviderEvent{Type: EventError, Error: err}
204+ return // Don't retry on accumulation errors
205+ }
206+
207+ switch event := event.AsAny().(type) {
208+ case anthropic.ContentBlockStartEvent:
209+ eventChan <- ProviderEvent{Type: EventContentStart}
210+
168− case anthropic.ContentBlockDeltaEvent:
211+ case anthropic.ContentBlockDeltaEvent:
169− if event.Delta.Type == "thinking_delta" && event.Delta.Thinking != "" {
212+ if event.Delta.Type == "thinking_delta" && event.Delta.Thinking != "" {
170− eventChan <- ProviderEvent{
213+ eventChan <- ProviderEvent{
171− Type: EventThinkingDelta,
214+ Type: EventThinkingDelta,
172− Thinking: event.Delta.Thinking,
215+ Thinking: event.Delta.Thinking,
216+ }
217+ } else if event.Delta.Type == "text_delta" && event.Delta.Text != "" {
218+ eventChan <- ProviderEvent{
219+ Type: EventContentDelta,
220+ Content: event.Delta.Text,
221+ }
173 }
222 }
174− } else if event.Delta.Type == "text_delta" && event.Delta.Text != "" {
223+
224+ case anthropic.ContentBlockStopEvent:
175− eventChan <- ProviderEvent{
225+ eventChan <- ProviderEvent{Type: EventContentStop}
176− Type: EventContentDelta,
226+
177− Content: event.Delta.Text,
227+ case anthropic.MessageStopEvent:
228+ streamSuccess = true
229+ content := ""
230+ for _, block := range accumulatedMessage.Content {
231+ if text, ok := block.AsAny().(anthropic.TextBlock); ok {
232+ content += text.Text
233+ }
178 }
234 }
179− }
180
235
181− case anthropic.ContentBlockStopEvent:
236+ toolCalls := a.extractToolCalls(accumulatedMessage.Content)
182− eventChan <- ProviderEvent{Type: EventContentStop}
237+ tokenUsage := a.extractTokenUsage(accumulatedMessage.Usage)
183
238
184− case anthropic.MessageStopEvent:
239+ eventChan <- ProviderEvent{
185− content := ""
240+ Type: EventComplete,
186− for _, block := range accumulatedMessage.Content {
241+ Response: &ProviderResponse{
187− if text, ok := block.AsAny().(anthropic.TextBlock); ok {
242+ Content: content,
188− content += text.Text
243+ ToolCalls: toolCalls,
244+ Usage: tokenUsage,
245+ FinishReason: string(accumulatedMessage.StopReason),
246+ },
189 }
247 }
190 }
248 }
249+ }
191
250
192− toolCalls := a.extractToolCalls(accumulatedMessage.Content)
251+ // If the stream completed successfully, we're done
193− tokenUsage := a.extractTokenUsage(accumulatedMessage.Usage)
252+ if streamSuccess {
253+ return
254+ }
194
255
195− eventChan <- ProviderEvent{
256+ // Check for stream errors
196− Type: EventComplete,
257+ err := stream.Err()
197− Response: &ProviderResponse{
258+ if err != nil {
198− Content: content,
259+ log.Println("error", err)
199− ToolCalls: toolCalls,
260+
200− Usage: tokenUsage,
261+ var apierr *anthropic.Error
201− FinishReason: string(accumulatedMessage.StopReason),
262+ if errors.As(err, &apierr) && apierr.StatusCode == 429 {
202− },
263+ continue
203 }
264 }
204− }
205− }
206
265
207− if stream.Err() != nil {
266+ // For non-rate limit errors, report and exit
208− eventChan <- ProviderEvent{Type: EventError, Error: stream.Err()}
267+ eventChan <- ProviderEvent{Type: EventError, Error: err}
268+ return
269+ }
209 }
270 }
210 }()
271 }()
211
272
@@ -319,3 +380,4 @@ func (a *anthropicProvider) convertToAnthropicMessages(messages []message.Messag
319
380
320 return anthropicMessages
381 return anthropicMessages
321}
382}
383+