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) + } +}