From 21a6bb12a1f8c7aff913390ebf8b7eaf096f4bf0 Mon Sep 17 00:00:00 2001 From: Thomas Beaudouin Date: Sat, 21 Feb 2026 14:54:08 +0800 Subject: [PATCH] feat: add bounded retry logic with escalating timeouts for LLM calls Implement retry mechanism for LLM API calls with progressive timeout escalation (45s, 90s, 120s) and exponential backoff (2s, 5s) to handle transient network issues and server errors. Add user notifications via message bus when retries occur. Update HTTP client timeout to 130s to accommodate longest retry attempt. - Add chatWithRetry method in agent loop with retry configuration - Integrate DoWithRetry utility in both agent loop --- pkg/agent/loop.go | 37 +++++++-- pkg/providers/http_provider.go | 2 +- pkg/tools/toolloop.go | 35 +++++++- pkg/utils/llm_retry.go | 141 ++++++++++++++++++++++++++++++++ pkg/utils/llm_retry_test.go | 145 +++++++++++++++++++++++++++++++++ 5 files changed, 351 insertions(+), 9 deletions(-) create mode 100644 pkg/utils/llm_retry.go create mode 100644 pkg/utils/llm_retry_test.go diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index f3dd94090..ea58d5bd1 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -445,11 +445,8 @@ func (al *AgentLoop) runLLMIteration(ctx context.Context, messages []providers.M "tools_json": formatToolsForLog(providerToolDefs), }) - // Call LLM - response, err := al.provider.Chat(ctx, messages, providerToolDefs, al.model, map[string]interface{}{ - "max_tokens": 8192, - "temperature": 0.7, - }) + // Call LLM (with bounded retries on timeouts and server errors) + response, err := al.chatWithRetry(ctx, messages, providerToolDefs, opts) if err != nil { logger.ErrorCF("agent", "LLM call failed", @@ -568,6 +565,34 @@ func (al *AgentLoop) runLLMIteration(ctx context.Context, messages []providers.M return finalContent, iteration, nil } +func (al *AgentLoop) chatWithRetry(ctx context.Context, messages []providers.Message, toolDefs []providers.ToolDefinition, opts processOptions) (*providers.LLMResponse, error) { + llmOpts := map[string]interface{}{ + "max_tokens": 8192, + "temperature": 0.7, + } + + // 3 attempts total: 45s, 90s, 120s + retryCfg := utils.RetryConfig{ + Timeouts: []time.Duration{45 * time.Second, 90 * time.Second, 120 * time.Second}, + Backoffs: []time.Duration{2 * time.Second, 5 * time.Second}, + Notify: func(attempt, total int, decision utils.RetryDecision) { + if opts.Channel == "" || opts.ChatID == "" { + return + } + notice := utils.FormatLLMRetryNotice(attempt, total, decision) + al.bus.PublishOutbound(bus.OutboundMessage{ + Channel: opts.Channel, + ChatID: opts.ChatID, + Content: notice, + }) + }, + } + + return utils.DoWithRetry(ctx, retryCfg, func(attemptCtx context.Context) (*providers.LLMResponse, error) { + return al.provider.Chat(attemptCtx, messages, toolDefs, al.model, llmOpts) + }) +} + // updateToolContexts updates the context for tools that need channel/chatID info. func (al *AgentLoop) updateToolContexts(channel, chatID string) { // Use ContextualTool interface instead of type assertions @@ -631,7 +656,7 @@ func formatMessagesForLog(messages []providers.Message) string { result += "[\n" for i, msg := range messages { result += fmt.Sprintf(" [%d] Role: %s\n", i, msg.Role) - if msg.ToolCalls != nil && len(msg.ToolCalls) > 0 { + if len(msg.ToolCalls) > 0 { result += " ToolCalls:\n" for _, tc := range msg.ToolCalls { result += fmt.Sprintf(" - ID: %s, Type: %s, Name: %s\n", tc.ID, tc.Type, tc.Name) diff --git a/pkg/providers/http_provider.go b/pkg/providers/http_provider.go index 4cf2c6db2..4d2676cd8 100644 --- a/pkg/providers/http_provider.go +++ b/pkg/providers/http_provider.go @@ -29,7 +29,7 @@ type HTTPProvider struct { func NewHTTPProvider(apiKey, apiBase, proxy string) *HTTPProvider { client := &http.Client{ - Timeout: 120 * time.Second, + Timeout: 130 * time.Second, } if proxy != "" { diff --git a/pkg/tools/toolloop.go b/pkg/tools/toolloop.go index 1302079b4..7f048fd1e 100644 --- a/pkg/tools/toolloop.go +++ b/pkg/tools/toolloop.go @@ -10,6 +10,7 @@ import ( "context" "encoding/json" "fmt" + "time" "github.com/sipeed/picoclaw/pkg/logger" "github.com/sipeed/picoclaw/pkg/providers" @@ -37,6 +38,10 @@ func RunToolLoop(ctx context.Context, config ToolLoopConfig, messages []provider iteration := 0 var finalContent string + // 3 attempts total: 45s, 90s, 120s + perAttemptTimeouts := []time.Duration{45 * time.Second, 90 * time.Second, 120 * time.Second} + backoffs := []time.Duration{2 * time.Second, 5 * time.Second} + for iteration < config.MaxIterations { iteration++ @@ -61,8 +66,17 @@ func RunToolLoop(ctx context.Context, config ToolLoopConfig, messages []provider } } - // 3. Call LLM - response, err := config.Provider.Chat(ctx, messages, providerToolDefs, config.Model, llmOpts) + // 3. Call LLM (with bounded retries on timeouts and server errors) + retryCfg := utils.RetryConfig{ + Timeouts: perAttemptTimeouts, + Backoffs: backoffs, + Notify: func(attempt, total int, decision utils.RetryDecision) { + sendRetryNotice(ctx, config.Tools, channel, chatID, attempt, total, decision) + }, + } + response, err := utils.DoWithRetry(ctx, retryCfg, func(attemptCtx context.Context) (*providers.LLMResponse, error) { + return config.Provider.Chat(attemptCtx, messages, providerToolDefs, config.Model, llmOpts) + }) if err != nil { logger.ErrorCF("toolloop", "LLM call failed", map[string]any{ @@ -152,3 +166,20 @@ func RunToolLoop(ctx context.Context, config ToolLoopConfig, messages []provider Iterations: iteration, }, nil } + +func sendRetryNotice(ctx context.Context, tools *ToolRegistry, channel, chatID string, attempt, total int, decision utils.RetryDecision) { + if tools == nil || channel == "" || chatID == "" { + return + } + + notice := utils.FormatLLMRetryNotice(attempt, total, utils.RetryDecision{ + Retryable: decision.Retryable, + Status: decision.Status, + Reason: utils.RetryReason(decision.Reason), + }) + + args := map[string]interface{}{ + "content": notice, + } + _ = tools.ExecuteWithContext(ctx, "message", args, channel, chatID, nil) +} diff --git a/pkg/utils/llm_retry.go b/pkg/utils/llm_retry.go new file mode 100644 index 000000000..48b4ca5b4 --- /dev/null +++ b/pkg/utils/llm_retry.go @@ -0,0 +1,141 @@ +package utils + +import ( + "context" + "errors" + "fmt" + "strconv" + "strings" + "time" +) + +type RetryReason string + +const ( + RetryReasonTimeout RetryReason = "timeout" + RetryReasonServerError RetryReason = "server_error" +) + +type RetryDecision struct { + Retryable bool + Status int + Reason RetryReason +} + +func IsRetryableError(err error) RetryDecision { + if err == nil { + return RetryDecision{} + } + + if errors.Is(err, context.DeadlineExceeded) { + return RetryDecision{Retryable: true, Reason: RetryReasonTimeout} + } + + msg := err.Error() + if strings.Contains(msg, "context deadline exceeded") || strings.Contains(msg, "Client.Timeout") { + return RetryDecision{Retryable: true, Reason: RetryReasonTimeout} + } + + if s, ok := ParseHTTPStatusFromError(msg); ok { + if s >= 500 && s <= 599 { + return RetryDecision{Retryable: true, Status: s, Reason: RetryReasonServerError} + } + return RetryDecision{Retryable: false, Status: s} + } + + return RetryDecision{} +} + +func ParseHTTPStatusFromError(msg string) (int, bool) { + idx := strings.Index(msg, "Status:") + if idx < 0 { + return 0, false + } + + s := strings.TrimSpace(msg[idx+len("Status:"):]) + end := 0 + for end < len(s) { + c := s[end] + if c < '0' || c > '9' { + break + } + end++ + } + if end == 0 { + return 0, false + } + + code, err := strconv.Atoi(s[:end]) + if err != nil { + return 0, false + } + return code, true +} + +type RetryNotifyFunc func(attempt, total int, decision RetryDecision) + +type RetryConfig struct { + Timeouts []time.Duration + Backoffs []time.Duration + Notify RetryNotifyFunc +} + +func DoWithRetry[T any]( + ctx context.Context, + retry RetryConfig, + fn func(context.Context) (T, error), +) (T, error) { + var zero T + if len(retry.Timeouts) == 0 { + return fn(ctx) + } + + var lastErr error + for attempt := 1; attempt <= len(retry.Timeouts); attempt++ { + attemptCtx, cancel := context.WithTimeout(ctx, retry.Timeouts[attempt-1]) + val, err := fn(attemptCtx) + cancel() + + if err == nil { + return val, nil + } + + lastErr = err + if attempt == len(retry.Timeouts) { + break + } + + decision := IsRetryableError(err) + if !decision.Retryable { + break + } + + if retry.Notify != nil { + retry.Notify(attempt, len(retry.Timeouts), decision) + } + + if attempt-1 < len(retry.Backoffs) { + select { + case <-ctx.Done(): + return zero, ctx.Err() + case <-time.After(retry.Backoffs[attempt-1]): + } + } + } + + return zero, lastErr +} + +func FormatLLMRetryNotice(attempt, total int, decision RetryDecision) string { + switch decision.Reason { + case RetryReasonTimeout: + return fmt.Sprintf("LLM timed out, retrying (attempt %d/%d)", attempt+1, total) + case RetryReasonServerError: + if decision.Status > 0 { + return fmt.Sprintf("LLM server error (%d), retrying (attempt %d/%d)", decision.Status, attempt+1, total) + } + return fmt.Sprintf("LLM server error, retrying (attempt %d/%d)", attempt+1, total) + default: + return fmt.Sprintf("LLM call failed, retrying (attempt %d/%d)", attempt+1, total) + } +} diff --git a/pkg/utils/llm_retry_test.go b/pkg/utils/llm_retry_test.go new file mode 100644 index 000000000..99d9c2292 --- /dev/null +++ b/pkg/utils/llm_retry_test.go @@ -0,0 +1,145 @@ +package utils + +import ( + "context" + "errors" + "testing" + "time" +) + +type stubValueRunner struct { + errors []error + vals []string + calls int +} + +func (s *stubValueRunner) Run(ctx context.Context) (string, error) { + s.calls++ + idx := s.calls - 1 + if idx < len(s.errors) && s.errors[idx] != nil { + return "", s.errors[idx] + } + if idx < len(s.vals) { + return s.vals[idx], nil + } + return "", errors.New("no value") +} + +func TestLLMRetry_IsRetryableError(t *testing.T) { + cases := []struct { + name string + err error + want RetryDecision + }{ + { + name: "deadline exceeded", + err: context.DeadlineExceeded, + want: RetryDecision{Retryable: true, Reason: RetryReasonTimeout}, + }, + { + name: "client timeout string", + err: errors.New("failed to read response: context deadline exceeded (Client.Timeout)"), + want: RetryDecision{Retryable: true, Reason: RetryReasonTimeout}, + }, + { + name: "server 502", + err: errors.New("API request failed:\n Status: 502\n Body: bad"), + want: RetryDecision{Retryable: true, Status: 502, Reason: RetryReasonServerError}, + }, + { + name: "client 400", + err: errors.New("API request failed:\n Status: 400\n Body: bad"), + want: RetryDecision{Retryable: false, Status: 400}, + }, + { + name: "other error", + err: errors.New("something else"), + want: RetryDecision{Retryable: false}, + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + got := IsRetryableError(tc.err) + if got.Retryable != tc.want.Retryable || got.Status != tc.want.Status || got.Reason != tc.want.Reason { + t.Fatalf("IsRetryableError(%v) = %+v, want %+v", tc.err, got, tc.want) + } + }) + } +} + +func TestLLMRetry_DoWithRetry_TimeoutThenSuccess(t *testing.T) { + runner := &stubValueRunner{ + errors: []error{context.DeadlineExceeded, nil}, + vals: []string{"", "ok"}, + } + + notices := 0 + retryCfg := RetryConfig{ + Timeouts: []time.Duration{5 * time.Millisecond, 5 * time.Millisecond}, + Backoffs: []time.Duration{}, + Notify: func(attempt, total int, decision RetryDecision) { + notices++ + if decision.Reason != RetryReasonTimeout { + t.Fatalf("expected timeout reason, got %v", decision.Reason) + } + }, + } + + val, err := DoWithRetry(context.Background(), retryCfg, runner.Run) + if err != nil { + t.Fatalf("DoWithRetry error: %v", err) + } + if val != "ok" { + t.Fatalf("val = %q, want ok", val) + } + if runner.calls != 2 { + t.Fatalf("runner.calls = %d, want 2", runner.calls) + } + if notices != 1 { + t.Fatalf("notices = %d, want 1", notices) + } +} + +func TestLLMRetry_DoWithRetry_ServerErrorThenSuccess(t *testing.T) { + runner := &stubValueRunner{ + errors: []error{errors.New("API request failed:\n Status: 502\n Body: bad"), nil}, + vals: []string{"", "ok"}, + } + + retryCfg := RetryConfig{ + Timeouts: []time.Duration{5 * time.Millisecond, 5 * time.Millisecond}, + Backoffs: []time.Duration{}, + } + + val, err := DoWithRetry(context.Background(), retryCfg, runner.Run) + if err != nil { + t.Fatalf("DoWithRetry error: %v", err) + } + if val != "ok" { + t.Fatalf("val = %q, want ok", val) + } + if runner.calls != 2 { + t.Fatalf("runner.calls = %d, want 2", runner.calls) + } +} + +func TestLLMRetry_DoWithRetry_NoRetryOnClientError(t *testing.T) { + runner := &stubValueRunner{ + errors: []error{errors.New("API request failed:\n Status: 400\n Body: bad")}, + vals: []string{""}, + } + + retryCfg := RetryConfig{ + Timeouts: []time.Duration{5 * time.Millisecond, 5 * time.Millisecond}, + Backoffs: []time.Duration{}, + } + + _, err := DoWithRetry(context.Background(), retryCfg, runner.Run) + if err == nil { + t.Fatalf("expected error, got nil") + } + if runner.calls != 1 { + t.Fatalf("runner.calls = %d, want 1", runner.calls) + } +}