diff --git a/doc.go b/doc.go index 085e7b26c..b8c91a308 100644 --- a/doc.go +++ b/doc.go @@ -221,6 +221,109 @@ // } // } // +// # Automatic Retry +// +// LangchainGo provides a built-in retry mechanism for transient HTTP failures. +// Retry is opt-in: it is only enabled when a RetryConfig is explicitly provided +// via the WithRetryConfig option. Without it, behavior is unchanged. +// +// ## What gets retried +// +// The following transient failures are automatically retried: +// +// - HTTP 429 (Rate Limit) — respects the Retry-After response header +// - HTTP 500, 502, 503, 504 (Server Errors) +// - Network errors (connection refused, DNS failure, TLS timeout) +// - Provider-specific body-level errors (e.g., ERNIE's HTTP 200 + error_code:18) +// +// The following are NOT retried: +// +// - HTTP 400, 401, 403, 404 and other 4xx client errors +// - Context cancellation (context.Canceled) +// - Context deadline exceeded (context.DeadlineExceeded) +// - Successful responses +// +// ## Basic usage +// +// All LLM providers support retry via the WithRetryConfig option: +// +// import "github.com/tmc/langchaingo/httputil" +// +// // Use sensible defaults: 3 retries, 1s initial backoff, 30s max backoff +// llm, err := openai.New( +// openai.WithToken("sk-xxx"), +// openai.WithRetryConfig(httputil.DefaultRetryConfig()), +// ) +// +// The same pattern works for all providers: +// +// llm, _ := anthropic.New(anthropic.WithRetryConfig(cfg), ...) +// llm, _ := ernie.New(ernie.WithRetryConfig(cfg), ...) +// llm, _ := ollama.New(ollama.WithRetryConfig(cfg), ...) +// llm, _ := cohere.New(cohere.WithRetryConfig(cfg), ...) +// llm, _ := cloudflare.New(cloudflare.WithRetryConfig(cfg), ...) +// llm, _ := huggingface.New(huggingface.WithRetryConfig(cfg), ...) +// llm, _ := llamafile.New(llamafile.WithRetryConfig(cfg), ...) +// llm, _ := maritaca.New(maritaca.WithRetryConfig(cfg), ...) +// +// ## Custom configuration +// +// llm, err := openai.New( +// openai.WithToken("sk-xxx"), +// openai.WithRetryConfig(&httputil.RetryConfig{ +// MaxRetries: 5, +// InitialBackoff: 2 * time.Second, +// MaxBackoff: 60 * time.Second, +// BackoffFactor: 2.0, +// }), +// ) +// +// The backoff sequence with Factor 2.0 and InitialBackoff 2s is: +// +// attempt 0: 2s (initial) +// attempt 1: 4s (2s × 2.0) +// attempt 2: 8s (4s × 2.0) +// attempt 3: 16s (8s × 2.0) +// attempt 4: 32s (capped at MaxBackoff 60s) +// +// Random jitter is applied to prevent thundering herd. +// +// ## Retry-After header support +// +// When a provider returns HTTP 429 with a Retry-After header, the retry +// mechanism waits the server-specified duration instead of the computed backoff: +// +// // Server returns: HTTP 429 + Retry-After: 60 +// // Wait duration: max(computed_backoff, 60s) = 60s +// +// ## Logging retries +// +// Use the OnRetry callback for observability: +// +// llm, err := openai.New( +// openai.WithToken("sk-xxx"), +// openai.WithRetryConfig(&httputil.RetryConfig{ +// MaxRetries: 3, +// InitialBackoff: 1 * time.Second, +// MaxBackoff: 30 * time.Second, +// BackoffFactor: 2.0, +// OnRetry: func(attempt int, err error) { +// slog.Warn("retrying request", "attempt", attempt+1, "error", err) +// }, +// }), +// ) +// +// ## Custom retry conditions +// +// Override the default retryable status codes or error checks: +// +// cfg := httputil.DefaultRetryConfig() +// +// // Also retry HTTP 408 (Request Timeout) +// cfg.RetryableStatus = func(code int) bool { +// return code == 408 || code == 429 || (code >= 500 && code <= 504) +// } +// // # Testing // // LangchainGo includes comprehensive testing utilities including HTTP record/replay for internal tests. diff --git a/httputil/doc.go b/httputil/doc.go index 8330bb458..740aac52d 100644 --- a/httputil/doc.go +++ b/httputil/doc.go @@ -50,4 +50,139 @@ // The transports in this package are designed to work with the httprr // HTTP record/replay system used in tests. When using httprr, pass // httputil.DefaultTransport to ensure proper request interception. +// +// # Automatic Retry +// +// The package provides a unified retry mechanism for transient HTTP failures. +// It is opt-in: retry is only enabled when a RetryConfig is explicitly provided. +// +// ## What gets retried +// +// The following transient failures are automatically retried: +// +// - HTTP 429 (Rate Limit) — respects the Retry-After response header +// - HTTP 500, 502, 503, 504 (Server Errors) +// - Network errors (connection refused, DNS failure, TLS timeout) +// - Provider-specific body-level errors (e.g., ERNIE's HTTP 200 + error_code:18) +// +// The following are NOT retried: +// +// - HTTP 400, 401, 403, 404 and other 4xx client errors +// - Context cancellation (context.Canceled) +// - Context deadline exceeded (context.DeadlineExceeded) +// - Successful responses +// +// ## Basic usage +// +// All LLM providers support retry via the WithRetryConfig option: +// +// import "github.com/tmc/langchaingo/httputil" +// +// // Use sensible defaults: 3 retries, 1s initial backoff, 30s max backoff +// llm, err := openai.New( +// openai.WithToken("sk-xxx"), +// openai.WithRetryConfig(httputil.DefaultRetryConfig()), +// ) +// +// The same pattern works for all providers: +// +// llm, _ := anthropic.New(anthropic.WithRetryConfig(cfg), ...) +// llm, _ := ernie.New(ernie.WithRetryConfig(cfg), ...) +// llm, _ := ollama.New(ollama.WithRetryConfig(cfg), ...) +// llm, _ := cohere.New(cohere.WithRetryConfig(cfg), ...) +// llm, _ := cloudflare.New(cloudflare.WithRetryConfig(cfg), ...) +// llm, _ := huggingface.New(huggingface.WithRetryConfig(cfg), ...) +// llm, _ := llamafile.New(llamafile.WithRetryConfig(cfg), ...) +// llm, _ := maritaca.New(maritaca.WithRetryConfig(cfg), ...) +// +// When WithRetryConfig is not provided, no retry is performed (the default +// behavior is unchanged). +// +// ## Custom configuration +// +// llm, err := openai.New( +// openai.WithToken("sk-xxx"), +// openai.WithRetryConfig(&httputil.RetryConfig{ +// MaxRetries: 5, +// InitialBackoff: 2 * time.Second, +// MaxBackoff: 60 * time.Second, +// BackoffFactor: 2.0, +// }), +// ) +// +// The backoff sequence with Factor 2.0 and InitialBackoff 2s is: +// +// attempt 0: 2s (initial) +// attempt 1: 4s (2s × 2.0) +// attempt 2: 8s (4s × 2.0) +// attempt 3: 16s (8s × 2.0) +// attempt 4: 32s (capped at MaxBackoff 60s) +// +// Random jitter (±50%) is applied to prevent thundering herd. +// +// ## Retry-After header support +// +// When a provider returns HTTP 429 with a Retry-After header, the retry +// mechanism waits the duration specified by the server instead of the +// computed backoff: +// +// // Server returns: HTTP 429 + Retry-After: 60 +// // Wait duration: max(computed_backoff, 60s) = 60s +// +// This applies to OpenAI, Anthropic, and ERNIE providers. +// +// ## Logging retries +// +// Use the OnRetry callback for observability: +// +// llm, err := openai.New( +// openai.WithToken("sk-xxx"), +// openai.WithRetryConfig(&httputil.RetryConfig{ +// MaxRetries: 3, +// InitialBackoff: 1 * time.Second, +// MaxBackoff: 30 * time.Second, +// BackoffFactor: 2.0, +// OnRetry: func(attempt int, err error) { +// slog.Warn("retrying request", +// "attempt", attempt+1, +// "error", err, +// ) +// }, +// }), +// ) +// +// ## Custom retry conditions +// +// Override the default retryable status codes or error checks: +// +// cfg := httputil.DefaultRetryConfig() +// +// // Also retry HTTP 408 (Request Timeout) +// cfg.RetryableStatus = func(code int) bool { +// return code == 408 || code == 429 || (code >= 500 && code <= 504) +// } +// +// // Custom network error check +// cfg.RetryableError = func(err error) bool { +// return myCustomCheck(err) +// } +// +// ## Two internal mechanisms +// +// The package provides two retry implementations. Users do not need to +// choose — the appropriate one is selected automatically by each provider: +// +// - RetryOnError (application-layer): Used by OpenAI, Anthropic, and ERNIE. +// Retries based on network errors, HTTP status codes, AND provider-specific +// body-level errors (e.g., ERNIE returns HTTP 200 with error_code in the +// response body). Also respects Retry-After headers. +// +// - RetryTransport (transport-layer): Used by Ollama, Cohere, Cloudflare, +// HuggingFace, Llamafile, and Maritaca. Retries based on HTTP status codes +// and network errors at the http.RoundTripper level. Also respects +// Retry-After headers. +// +// Both mechanisms share the same RetryConfig, backoff algorithm, and jitter +// strategy. The only difference is that RetryOnError can additionally detect +// body-level errors via provider-specific MapError functions. package httputil diff --git a/httputil/errors.go b/httputil/errors.go new file mode 100644 index 000000000..e2e5c0c06 --- /dev/null +++ b/httputil/errors.go @@ -0,0 +1,71 @@ +package httputil + +import ( + "fmt" + "net/http" + "strconv" + "time" +) + +// ResponseError represents an HTTP error response that carries metadata +// useful for retry decisions, including the Retry-After header value. +// +// Provider internal clients should return *ResponseError when they receive +// non-200 HTTP responses so that [RetryOnError] can respect Retry-After. +type ResponseError struct { + // StatusCode is the HTTP status code. + StatusCode int + + // Message is the error message (typically includes status code and API body). + Message string + + // RetryAfter is the duration indicated by the Retry-After response header. + // Zero means the header was absent or could not be parsed. + RetryAfter time.Duration +} + +// Error implements the error interface. +func (e *ResponseError) Error() string { return e.Message } + +// ParseRetryAfterHeader extracts the Retry-After duration from an HTTP response. +// Returns 0 if the header is absent or cannot be parsed. +func ParseRetryAfterHeader(resp *http.Response) time.Duration { + val := resp.Header.Get("Retry-After") + if val == "" { + return 0 + } + + // Try parsing as integer seconds. + if seconds, err := strconv.Atoi(val); err == nil && seconds > 0 { + return time.Duration(seconds) * time.Second + } + + // Try parsing as HTTP date. + if t, err := http.ParseTime(val); err == nil { + d := time.Until(t) + if d > 0 { + return d + } + } + + return 0 +} + +// NewResponseError creates a ResponseError from an HTTP response. +// The body should be pre-read; this function only extracts the status code +// and Retry-After header. +func NewResponseError(resp *http.Response, message string) *ResponseError { + return &ResponseError{ + StatusCode: resp.StatusCode, + Message: message, + RetryAfter: ParseRetryAfterHeader(resp), + } +} + +// FormatHTTPError formats a standard HTTP error message with status code and optional body. +func FormatHTTPError(statusCode int, body string) string { + if body != "" { + return fmt.Sprintf("API returned unexpected status code: %d: %s", statusCode, body) + } + return fmt.Sprintf("API returned unexpected status code: %d", statusCode) +} diff --git a/httputil/retry.go b/httputil/retry.go new file mode 100644 index 000000000..16bd464df --- /dev/null +++ b/httputil/retry.go @@ -0,0 +1,320 @@ +package httputil + +import ( + "bytes" + "context" + "errors" + "fmt" + "io" + "math" + "math/rand" + "net" + "net/http" + "strconv" + "strings" + "time" +) + +// RetryConfig controls the retry behavior for HTTP requests. +// +// Retry is performed at the transport layer via [RetryTransport], which wraps +// an underlying [http.RoundTripper]. Only transient failures are retried: +// +// - HTTP 429 (Rate Limit) — respects the Retry-After header +// - HTTP 500, 502, 503, 504 (Server Errors) +// - Network errors (connection refused, DNS failure, TLS timeout) +// +// Client errors (4xx except 429), context cancellation, and successful +// responses are never retried. +type RetryConfig struct { + // MaxRetries is the maximum number of retry attempts. + // The total number of requests will be MaxRetries + 1. + // Default: 3 + MaxRetries int + + // InitialBackoff is the duration to wait before the first retry. + // Default: 1 * time.Second + InitialBackoff time.Duration + + // MaxBackoff is the upper bound on backoff duration. + // Default: 30 * time.Second + MaxBackoff time.Duration + + // BackoffFactor is the multiplier applied to the backoff after each attempt. + // Default: 2.0 + BackoffFactor float64 + + // RetryableError, if provided, overrides the default error retryability check. + // Return true to indicate the error is transient and should be retried. + // The default check retries network errors (connection refused, DNS, TLS). + RetryableError func(error) bool + + // RetryableStatus, if provided, overrides the default HTTP status code + // retryability check. Return true to indicate the status code is transient. + // The default check retries 429, 500, 502, 503, 504. + RetryableStatus func(statusCode int) bool + + // OnRetry is called before each retry attempt with the attempt number + // (0-based) and the error that triggered the retry. Use this for logging + // or metrics. + OnRetry func(attempt int, err error) +} + +// DefaultRetryConfig returns a RetryConfig with sensible defaults: +// - MaxRetries: 3 +// - InitialBackoff: 1 second +// - MaxBackoff: 30 seconds +// - BackoffFactor: 2.0 +func DefaultRetryConfig() *RetryConfig { + return &RetryConfig{ + MaxRetries: 3, + InitialBackoff: 1 * time.Second, + MaxBackoff: 30 * time.Second, + BackoffFactor: 2.0, + } +} + +// RetryTransport is an [http.RoundTripper] that automatically retries +// failed requests using exponential backoff with jitter. +// +// RetryTransport composes with the existing [Transport] (User-Agent injection) +// and any other [http.RoundTripper]. Typical transport chain: +// +// RetryTransport → Transport → http.DefaultTransport +// +// Retry is only performed before a successful response is received. +// Once a 200 OK is returned (including for streaming/SSE responses), +// no mid-stream retry is attempted. +type RetryTransport struct { + // Transport is the underlying [http.RoundTripper]. + // If nil, [http.DefaultTransport] is used. + Transport http.RoundTripper + + // Config controls retry behavior. Must not be nil. + Config *RetryConfig +} + +// RoundTrip implements the [http.RoundTripper] interface. +// It retries the request according to the configured [RetryConfig]. +func (t *RetryTransport) RoundTrip(req *http.Request) (*http.Response, error) { + cfg := t.Config + if cfg == nil { + cfg = DefaultRetryConfig() + } + + transport := t.Transport + if transport == nil { + transport = http.DefaultTransport + } + + ctx := req.Context() + + // Cache the request body for potential replays. + bodyBytes, err := readBody(req) + if err != nil { + return nil, fmt.Errorf("retry: failed to read request body: %w", err) + } + + var lastErr error + for attempt := 0; attempt <= cfg.MaxRetries; attempt++ { + // Check if the context has been cancelled. + if ctx.Err() != nil { + return nil, ctx.Err() + } + + // Clone the request, restoring the body. + reqClone := cloneRequest(req, bodyBytes) + + resp, err := transport.RoundTrip(reqClone) + if err != nil { + lastErr = err + if isRetryableError(err, cfg) && attempt < cfg.MaxRetries { + fireOnRetry(cfg, attempt, err) + if !wait(ctx, cfg.backoff(attempt)) { + return nil, ctx.Err() + } + continue + } + return nil, err + } + + // Check if the status code is retryable. + if isRetryableStatus(resp.StatusCode, cfg) && attempt < cfg.MaxRetries { + lastErr = fmt.Errorf("server returned %d: %s", resp.StatusCode, resp.Status) + // Parse Retry-After header for 429 responses. + waitDuration := cfg.backoff(attempt) + if ra, ok := parseRetryAfter(resp); ok && ra > waitDuration { + waitDuration = ra + } + resp.Body.Close() + fireOnRetry(cfg, attempt, lastErr) + if !wait(ctx, waitDuration) { + return nil, ctx.Err() + } + continue + } + + return resp, nil + } + + return nil, fmt.Errorf("retry: max retries (%d) exceeded, last error: %w", cfg.MaxRetries, lastErr) +} + +// backoff calculates the wait duration for the given attempt using exponential +// backoff with random jitter. +func (c *RetryConfig) backoff(attempt int) time.Duration { + if attempt <= 0 { + return c.InitialBackoff + } + + backoff := float64(c.InitialBackoff) * math.Pow(c.BackoffFactor, float64(attempt)) + + // Cap at MaxBackoff, then apply jitter. + if backoff > float64(c.MaxBackoff) { + backoff = float64(c.MaxBackoff) + } + + // Add random jitter in [0.5, 1.0) to ensure we never exceed MaxBackoff. + jitter := 0.5 + rand.Float64()*0.5 //nolint:gosec // G404: jitter doesn't need crypto-rand + return time.Duration(backoff * jitter) +} + +// isRetryableError checks if the given error is transient and worth retrying. +func isRetryableError(err error, cfg *RetryConfig) bool { + if cfg.RetryableError != nil { + return cfg.RetryableError(err) + } + return defaultIsRetryableError(err) +} + +// defaultIsRetryableError checks for network-level transient errors. +func defaultIsRetryableError(err error) bool { + if err == nil { + return false + } + + // Context errors are not retryable — the caller cancelled the request. + if err == context.Canceled || err == context.DeadlineExceeded { + return false + } + + // Network errors (connection refused, DNS, TLS) are generally transient. + var netErr net.Error + if errors.As(err, &netErr) { + return true + } + + // Check for common network error patterns via string matching. + // Many HTTP client errors wrap underlying network errors without + // implementing the net.Error interface. + errStr := err.Error() + retryablePatterns := []string{ + "connection refused", + "connection reset", + "broken pipe", + "TLS handshake", + "no such host", + "temporary", + "i/o timeout", + "EOF", + } + for _, pattern := range retryablePatterns { + if strings.Contains(strings.ToLower(errStr), strings.ToLower(pattern)) { + return true + } + } + + return false +} + +// isRetryableStatus checks if the HTTP status code indicates a transient failure. +func isRetryableStatus(statusCode int, cfg *RetryConfig) bool { + if cfg.RetryableStatus != nil { + return cfg.RetryableStatus(statusCode) + } + return defaultIsRetryableStatus(statusCode) +} + +// defaultIsRetryableStatus retries on 429 (rate limit) and 5xx (server errors). +func defaultIsRetryableStatus(statusCode int) bool { + switch statusCode { + case http.StatusTooManyRequests, // 429 + http.StatusInternalServerError, // 500 + http.StatusBadGateway, // 502 + http.StatusServiceUnavailable, // 503 + http.StatusGatewayTimeout: // 504 + return true + default: + return false + } +} + +// parseRetryAfter parses the Retry-After header from the response. +// It supports both seconds (integer) and HTTP date formats. +func parseRetryAfter(resp *http.Response) (time.Duration, bool) { + val := resp.Header.Get("Retry-After") + if val == "" { + return 0, false + } + + // Try parsing as integer seconds. + if seconds, err := strconv.Atoi(val); err == nil { + return time.Duration(seconds) * time.Second, true + } + + // Try parsing as HTTP date. + if t, err := http.ParseTime(val); err == nil { + d := time.Until(t) + if d > 0 { + return d, true + } + } + + return 0, false +} + +// readBody reads and caches the request body so it can be replayed on retry. +func readBody(req *http.Request) ([]byte, error) { + if req.Body == nil || req.Body == http.NoBody { + return nil, nil + } + data, err := io.ReadAll(req.Body) + req.Body.Close() + return data, err +} + +// cloneRequest creates a shallow clone of the request with a fresh body. +func cloneRequest(req *http.Request, body []byte) *http.Request { + r := req.Clone(req.Context()) + if body != nil { + r.Body = io.NopCloser(bytes.NewReader(body)) + r.GetBody = func() (io.ReadCloser, error) { + return io.NopCloser(bytes.NewReader(body)), nil + } + r.ContentLength = int64(len(body)) + } + return r +} + +// wait blocks for the given duration or until the context is cancelled. +// Returns false if the context was cancelled. +func wait(ctx context.Context, d time.Duration) bool { + if d <= 0 { + return ctx.Err() == nil + } + timer := time.NewTimer(d) + defer timer.Stop() + select { + case <-ctx.Done(): + return false + case <-timer.C: + return true + } +} + +// fireOnRetry calls the OnRetry callback if configured. +func fireOnRetry(cfg *RetryConfig, attempt int, err error) { + if cfg.OnRetry != nil { + cfg.OnRetry(attempt, err) + } +} diff --git a/httputil/retry_test.go b/httputil/retry_test.go new file mode 100644 index 000000000..bd6cf2976 --- /dev/null +++ b/httputil/retry_test.go @@ -0,0 +1,585 @@ +package httputil + +import ( + "context" + "errors" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" +) + +func TestDefaultRetryConfig(t *testing.T) { + cfg := DefaultRetryConfig() + if cfg.MaxRetries != 3 { + t.Errorf("expected MaxRetries=3, got %d", cfg.MaxRetries) + } + if cfg.InitialBackoff != 1*time.Second { + t.Errorf("expected InitialBackoff=1s, got %v", cfg.InitialBackoff) + } + if cfg.MaxBackoff != 30*time.Second { + t.Errorf("expected MaxBackoff=30s, got %v", cfg.MaxBackoff) + } + if cfg.BackoffFactor != 2.0 { + t.Errorf("expected BackoffFactor=2.0, got %f", cfg.BackoffFactor) + } +} + +func TestBackoff(t *testing.T) { + cfg := &RetryConfig{ + InitialBackoff: 1 * time.Second, + MaxBackoff: 30 * time.Second, + BackoffFactor: 2.0, + } + + tests := []struct { + attempt int + min time.Duration + max time.Duration + }{ + {0, 1 * time.Second, 1 * time.Second}, // exact InitialBackoff, no jitter for attempt 0 + {1, 1000 * time.Millisecond, 2000 * time.Millisecond}, // 2s * jitter [0.5, 1.0) + {2, 2000 * time.Millisecond, 4000 * time.Millisecond}, // 4s * jitter [0.5, 1.0) + {3, 4000 * time.Millisecond, 8000 * time.Millisecond}, // 8s * jitter [0.5, 1.0) + {10, 15000 * time.Millisecond, 30000 * time.Millisecond}, // capped at MaxBackoff, jitter [0.5, 1.0) + } + + for _, tt := range tests { + got := cfg.backoff(tt.attempt) + if got < tt.min || got > tt.max { + t.Errorf("backoff(%d) = %v, want in [%v, %v]", tt.attempt, got, tt.min, tt.max) + } + } +} + +func TestDefaultIsRetryableStatus(t *testing.T) { + retryable := []int{429, 500, 502, 503, 504} + for _, code := range retryable { + if !defaultIsRetryableStatus(code) { + t.Errorf("expected status %d to be retryable", code) + } + } + + notRetryable := []int{200, 400, 401, 403, 404, 405, 408, 409, 422} + for _, code := range notRetryable { + if defaultIsRetryableStatus(code) { + t.Errorf("expected status %d to NOT be retryable", code) + } + } +} + +func TestDefaultIsRetryableError(t *testing.T) { + tests := []struct { + err error + retryable bool + }{ + {nil, false}, + {context.Canceled, false}, + {context.DeadlineExceeded, false}, + {fmt.Errorf("connection refused"), true}, + {fmt.Errorf("connection reset by peer"), true}, + {fmt.Errorf("broken pipe"), true}, + {fmt.Errorf("TLS handshake timeout"), true}, + {fmt.Errorf("no such host"), true}, + {fmt.Errorf("i/o timeout"), true}, + {fmt.Errorf("unexpected EOF"), true}, + {fmt.Errorf("temporary failure"), true}, + {fmt.Errorf("some other error"), false}, + } + + for _, tt := range tests { + got := defaultIsRetryableError(tt.err) + if got != tt.retryable { + t.Errorf("defaultIsRetryableError(%v) = %v, want %v", tt.err, got, tt.retryable) + } + } +} + +func TestDefaultIsRetryableError_NetError(t *testing.T) { + // net.Error should be retryable. + netErr := &netOpError{Err: errors.New("network error")} + if !defaultIsRetryableError(netErr) { + t.Error("expected net.Error to be retryable") + } +} + +// netOpError implements net.Error for testing. +type netOpError struct { + Err error +} + +func (e *netOpError) Error() string { return e.Err.Error() } +func (e *netOpError) Timeout() bool { return false } +func (e *netOpError) Temporary() bool { return true } +func (e *netOpError) Unwrap() error { return e.Err } + +func TestParseRetryAfter(t *testing.T) { + tests := []struct { + header string + hasResult bool + min time.Duration + }{ + {"", false, 0}, + {"5", true, 4 * time.Second}, // 5 seconds + {"0", true, -1 * time.Second}, // 0 seconds is valid but <=0 + } + + for _, tt := range tests { + resp := &http.Response{Header: http.Header{"Retry-After": {tt.header}}} + d, ok := parseRetryAfter(resp) + if ok != tt.hasResult { + t.Errorf("parseRetryAfter(%q) returned ok=%v, want %v", tt.header, ok, tt.hasResult) + } + if ok && d < tt.min { + t.Errorf("parseRetryAfter(%q) = %v, want >= %v", tt.header, d, tt.min) + } + } +} + +func TestRetryTransport_SuccessOnFirstAttempt(t *testing.T) { + callCount := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + callCount++ + w.WriteHeader(http.StatusOK) + fmt.Fprint(w, `{"ok":true}`) + })) + defer server.Close() + + cfg := &RetryConfig{ + MaxRetries: 3, + InitialBackoff: 1 * time.Millisecond, + MaxBackoff: 10 * time.Millisecond, + BackoffFactor: 2.0, + } + client := &http.Client{ + Transport: &RetryTransport{ + Transport: http.DefaultTransport, + Config: cfg, + }, + } + + req, _ := http.NewRequestWithContext(context.Background(), http.MethodGet, server.URL, nil) + resp, err := client.Do(req) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + t.Errorf("expected 200, got %d", resp.StatusCode) + } + if callCount != 1 { + t.Errorf("expected 1 call, got %d", callCount) + } +} + +func TestRetryTransport_RetriesOnServerError(t *testing.T) { + callCount := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + callCount++ + if callCount < 3 { + w.WriteHeader(http.StatusServiceUnavailable) + return + } + w.WriteHeader(http.StatusOK) + fmt.Fprint(w, `{"ok":true}`) + })) + defer server.Close() + + cfg := &RetryConfig{ + MaxRetries: 3, + InitialBackoff: 1 * time.Millisecond, + MaxBackoff: 10 * time.Millisecond, + BackoffFactor: 2.0, + } + client := &http.Client{ + Transport: &RetryTransport{ + Transport: http.DefaultTransport, + Config: cfg, + }, + } + + req, _ := http.NewRequestWithContext(context.Background(), http.MethodGet, server.URL, nil) + resp, err := client.Do(req) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + t.Errorf("expected 200, got %d", resp.StatusCode) + } + if callCount != 3 { + t.Errorf("expected 3 calls, got %d", callCount) + } +} + +func TestRetryTransport_RetriesOnRateLimit(t *testing.T) { + callCount := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + callCount++ + if callCount < 2 { + w.Header().Set("Retry-After", "0") // immediate retry + w.WriteHeader(http.StatusTooManyRequests) + return + } + w.WriteHeader(http.StatusOK) + fmt.Fprint(w, `{"ok":true}`) + })) + defer server.Close() + + cfg := &RetryConfig{ + MaxRetries: 3, + InitialBackoff: 1 * time.Millisecond, + MaxBackoff: 10 * time.Millisecond, + BackoffFactor: 2.0, + } + client := &http.Client{ + Transport: &RetryTransport{ + Transport: http.DefaultTransport, + Config: cfg, + }, + } + + req, _ := http.NewRequestWithContext(context.Background(), http.MethodGet, server.URL, nil) + resp, err := client.Do(req) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + defer resp.Body.Close() + + if callCount != 2 { + t.Errorf("expected 2 calls, got %d", callCount) + } +} + +func TestRetryTransport_ExhaustsRetries(t *testing.T) { + callCount := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + callCount++ + w.WriteHeader(http.StatusInternalServerError) + })) + defer server.Close() + + cfg := &RetryConfig{ + MaxRetries: 2, + InitialBackoff: 1 * time.Millisecond, + MaxBackoff: 5 * time.Millisecond, + BackoffFactor: 2.0, + } + client := &http.Client{ + Transport: &RetryTransport{ + Transport: http.DefaultTransport, + Config: cfg, + }, + } + + req, _ := http.NewRequestWithContext(context.Background(), http.MethodGet, server.URL, nil) + resp, err := client.Do(req) + // After exhausting retries, the last 500 response is returned (not an error). + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + defer resp.Body.Close() + + // The last attempt returns the 500 response. + if resp.StatusCode != http.StatusInternalServerError { + t.Errorf("expected 500, got %d", resp.StatusCode) + } + // Total attempts = MaxRetries + 1 = 3 + if callCount != 3 { + t.Errorf("expected 3 calls, got %d", callCount) + } +} + +func TestRetryTransport_NoRetryOnClientError(t *testing.T) { + callCount := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + callCount++ + w.WriteHeader(http.StatusBadRequest) + })) + defer server.Close() + + cfg := &RetryConfig{ + MaxRetries: 3, + InitialBackoff: 1 * time.Millisecond, + MaxBackoff: 5 * time.Millisecond, + BackoffFactor: 2.0, + } + client := &http.Client{ + Transport: &RetryTransport{ + Transport: http.DefaultTransport, + Config: cfg, + }, + } + + req, _ := http.NewRequestWithContext(context.Background(), http.MethodGet, server.URL, nil) + resp, err := client.Do(req) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusBadRequest { + t.Errorf("expected 400, got %d", resp.StatusCode) + } + if callCount != 1 { + t.Errorf("expected 1 call (no retry), got %d", callCount) + } +} + +func TestRetryTransport_ContextCancellation(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusInternalServerError) + })) + defer server.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + + cfg := &RetryConfig{ + MaxRetries: 5, + InitialBackoff: 100 * time.Millisecond, // longer than context timeout + MaxBackoff: 500 * time.Millisecond, + BackoffFactor: 2.0, + } + client := &http.Client{ + Transport: &RetryTransport{ + Transport: http.DefaultTransport, + Config: cfg, + }, + } + + req, _ := http.NewRequestWithContext(ctx, http.MethodGet, server.URL, nil) + _, err := client.Do(req) + if err == nil { + t.Fatal("expected error due to context cancellation") + } + if !errors.Is(err, context.DeadlineExceeded) { + t.Errorf("expected context.DeadlineExceeded, got: %v", err) + } +} + +func TestRetryTransport_RequestBodyReplay(t *testing.T) { + var bodies []string + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(r.Body) + bodies = append(bodies, string(body)) + + if len(bodies) < 2 { + w.WriteHeader(http.StatusServiceUnavailable) + return + } + w.WriteHeader(http.StatusOK) + })) + defer server.Close() + + cfg := &RetryConfig{ + MaxRetries: 3, + InitialBackoff: 1 * time.Millisecond, + MaxBackoff: 5 * time.Millisecond, + BackoffFactor: 2.0, + } + client := &http.Client{ + Transport: &RetryTransport{ + Transport: http.DefaultTransport, + Config: cfg, + }, + } + + req, _ := http.NewRequestWithContext( + context.Background(), + http.MethodPost, + server.URL, + strings.NewReader(`{"model":"gpt-4","message":"hello"}`), + ) + resp, err := client.Do(req) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + defer resp.Body.Close() + + if len(bodies) != 2 { + t.Fatalf("expected 2 bodies, got %d", len(bodies)) + } + for i, body := range bodies { + expected := `{"model":"gpt-4","message":"hello"}` + if body != expected { + t.Errorf("body[%d] = %q, want %q", i, body, expected) + } + } +} + +func TestRetryTransport_OnRetryCallback(t *testing.T) { + callCount := 0 + var retryAttempts []int + var mu sync.Mutex + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + callCount++ + if callCount < 3 { + w.WriteHeader(http.StatusInternalServerError) + return + } + w.WriteHeader(http.StatusOK) + })) + defer server.Close() + + cfg := &RetryConfig{ + MaxRetries: 3, + InitialBackoff: 1 * time.Millisecond, + MaxBackoff: 5 * time.Millisecond, + BackoffFactor: 2.0, + OnRetry: func(attempt int, err error) { + mu.Lock() + retryAttempts = append(retryAttempts, attempt) + mu.Unlock() + }, + } + client := &http.Client{ + Transport: &RetryTransport{ + Transport: http.DefaultTransport, + Config: cfg, + }, + } + + req, _ := http.NewRequestWithContext(context.Background(), http.MethodGet, server.URL, nil) + resp, err := client.Do(req) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + defer resp.Body.Close() + + mu.Lock() + defer mu.Unlock() + if len(retryAttempts) != 2 { + t.Errorf("expected 2 OnRetry calls, got %d", len(retryAttempts)) + } +} + +func TestRetryTransport_CustomRetryableStatus(t *testing.T) { + callCount := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + callCount++ + // Return 408 (Request Timeout) — not in default retryable list + w.WriteHeader(http.StatusRequestTimeout) + })) + defer server.Close() + + // Default config should NOT retry 408 + cfg := DefaultRetryConfig() + cfg.InitialBackoff = 1 * time.Millisecond + cfg.MaxBackoff = 5 * time.Millisecond + client := &http.Client{ + Transport: &RetryTransport{ + Transport: http.DefaultTransport, + Config: cfg, + }, + } + + req, _ := http.NewRequestWithContext(context.Background(), http.MethodGet, server.URL, nil) + resp, _ := client.Do(req) + if callCount != 1 { + t.Errorf("default config should NOT retry 408, got %d calls", callCount) + } + resp.Body.Close() + + // Custom config that retries 408 + callCount = 0 + cfg2 := DefaultRetryConfig() + cfg2.InitialBackoff = 1 * time.Millisecond + cfg2.MaxBackoff = 5 * time.Millisecond + cfg2.RetryableStatus = func(code int) bool { + return code == http.StatusRequestTimeout || defaultIsRetryableStatus(code) + } + + client2 := &http.Client{ + Transport: &RetryTransport{ + Transport: http.DefaultTransport, + Config: cfg2, + }, + } + + req2, _ := http.NewRequestWithContext(context.Background(), http.MethodGet, server.URL, nil) + resp2, err := client2.Do(req2) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + resp2.Body.Close() + if callCount != cfg2.MaxRetries+1 { + t.Errorf("custom config should retry 408, got %d calls, want %d", callCount, cfg2.MaxRetries+1) + } +} + +func TestRetryTransport_NilConfigUsesDefault(t *testing.T) { + callCount := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + callCount++ + w.WriteHeader(http.StatusOK) + })) + defer server.Close() + + // RetryTransport with nil Config should use DefaultRetryConfig. + client := &http.Client{ + Transport: &RetryTransport{ + Transport: http.DefaultTransport, + Config: nil, + }, + } + + req, _ := http.NewRequestWithContext(context.Background(), http.MethodGet, server.URL, nil) + resp, err := client.Do(req) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + defer resp.Body.Close() + + if callCount != 1 { + t.Errorf("expected 1 call, got %d", callCount) + } +} + +func TestNewRetryClient(t *testing.T) { + cfg := DefaultRetryConfig() + client := NewRetryClient(cfg) + if client == nil { + t.Fatal("expected non-nil client") + } + + // Verify the transport chain. + rt, ok := client.Transport.(*RetryTransport) + if !ok { + t.Fatal("expected RetryTransport") + } + if rt.Config != cfg { + t.Error("config mismatch") + } +} + +func TestNewClientWithRetry(t *testing.T) { + cfg := DefaultRetryConfig() + cfg.InitialBackoff = 1 * time.Millisecond + cfg.MaxBackoff = 5 * time.Millisecond + + client := NewClientWithRetry(cfg) + if client == nil { + t.Fatal("expected non-nil client") + } + + // Verify the transport chain: RetryTransport -> Transport -> http.DefaultTransport + rt, ok := client.Transport.(*RetryTransport) + if !ok { + t.Fatal("expected RetryTransport as outer transport") + } + inner, ok := rt.Transport.(*Transport) + if !ok { + t.Fatal("expected Transport as inner transport") + } + if inner.Transport != http.DefaultTransport { + t.Error("inner Transport should use http.DefaultTransport") + } +} diff --git a/httputil/retry_wrapper.go b/httputil/retry_wrapper.go new file mode 100644 index 000000000..d97ffb1ec --- /dev/null +++ b/httputil/retry_wrapper.go @@ -0,0 +1,152 @@ +package httputil + +import ( + "context" + "errors" + "fmt" + "math" + "math/rand" + "time" + + "github.com/tmc/langchaingo/llms" +) + +// ErrorClassifier attempts to classify an error into a standardized [*llms.Error]. +// Provider implementations should use their [MapError] functions for this. +// Returns nil if the error cannot be classified. +type ErrorClassifier func(error) error + +// RetryOnError executes fn with retry on transient failures. It is the single +// unified retry mechanism that checks ALL retryable conditions in one pass: +// +// 1. Network errors (connection refused, DNS failure, TLS timeout) +// 2. HTTP status code patterns in error messages ("429", "500", "503", etc.) +// 3. Provider-specific error classification via the classifier (MapError) +// — catches body-level errors like ERNIE's HTTP 200 + error_code:18 +// +// If ANY condition indicates the error is retryable, the request is retried +// with exponential backoff. This replaces both [RetryTransport] and per-provider +// retry logic, ensuring a single retry budget is used. +// +// When the classified error carries a Retry-After hint (from [llms.Error.WithRetryAfter] +// or [*ResponseError]), the wait duration is max(computed_backoff, retryAfter). +// +// Usage: +// +// err = httputil.RetryOnError(ctx, retryCfg, openai.MapError, func() error { +// result, err = client.CreateChat(ctx, req) +// return err +// }) +func RetryOnError(ctx context.Context, cfg *RetryConfig, classifier ErrorClassifier, fn func() error) error { + if cfg == nil { + return fn() + } + + var lastErr error + for attempt := 0; attempt <= cfg.MaxRetries; attempt++ { + if ctx.Err() != nil { + return ctx.Err() + } + + err := fn() + if err == nil { + return nil + } + + lastErr = err + + // Classify the error once per iteration. + var classified error + if classifier != nil { + classified = classifier(err) + } + + // Exhausted retries — return classified error if available. + if attempt >= cfg.MaxRetries { + if classified != nil { + return classified + } + return lastErr + } + + // Check retryability: network error or classified error code. + if !isProviderRetryable(err, cfg, classified) { + if classified != nil { + return classified + } + return err + } + + // Calculate wait duration, respecting Retry-After if present. + waitDuration := providerBackoff(cfg, attempt) + if ra := extractRetryAfter(err, classified); ra > waitDuration { + waitDuration = ra + } + + fireOnRetry(cfg, attempt, err) + if !wait(ctx, waitDuration) { + return ctx.Err() + } + } + + return fmt.Errorf("retry: max retries (%d) exceeded, last error: %w", cfg.MaxRetries, lastErr) +} + +// isProviderRetryable checks all retry conditions in one pass. +// Returns true if the error should be retried. +func isProviderRetryable(err error, cfg *RetryConfig, classified error) bool { + // 1. Network errors (connection refused, DNS, TLS, etc.) + if isRetryableError(err, cfg) { + return true + } + + // 2. Classified error is a retryable code + // (rate limit, provider unavailable, timeout). + if classified != nil { + if llms.IsRetryableError(classified) { + return true + } + } + + return false +} + +// retryAfterer is an interface for errors that carry Retry-After information. +type retryAfterer interface { + RetryAfter() time.Duration +} + +// extractRetryAfter checks both the classified error and the original error +// for Retry-After information. Returns 0 if neither carries it. +func extractRetryAfter(original, classified error) time.Duration { + // Prefer classified error (MapError may have enriched it from ResponseError). + if classified != nil { + var ra retryAfterer + if errors.As(classified, &ra) { + if d := ra.RetryAfter(); d > 0 { + return d + } + } + } + // Fall back to original error (direct *ResponseError from HTTP layer). + var ra retryAfterer + if errors.As(original, &ra) { + return ra.RetryAfter() + } + return 0 +} + +// providerBackoff calculates backoff for provider-layer retries. +func providerBackoff(cfg *RetryConfig, attempt int) time.Duration { + if attempt <= 0 { + return cfg.InitialBackoff + } + + backoff := float64(cfg.InitialBackoff) * math.Pow(cfg.BackoffFactor, float64(attempt)) + if backoff > float64(cfg.MaxBackoff) { + backoff = float64(cfg.MaxBackoff) + } + + jitter := 0.5 + rand.Float64()*0.5 //nolint:gosec // G404: jitter doesn't need crypto-rand + return time.Duration(backoff * jitter) +} diff --git a/httputil/retry_wrapper_test.go b/httputil/retry_wrapper_test.go new file mode 100644 index 000000000..eeda8187c --- /dev/null +++ b/httputil/retry_wrapper_test.go @@ -0,0 +1,316 @@ +package httputil + +import ( + "context" + "errors" + "fmt" + "net/http" + "net/http/httptest" + "sync" + "testing" + "time" + + "github.com/tmc/langchaingo/llms" +) + +// --- ResponseError tests --- + +func TestResponseError_Error(t *testing.T) { + err := &ResponseError{ + StatusCode: 429, + Message: "API returned unexpected status code: 429: Rate limit exceeded", + RetryAfter: 30 * time.Second, + } + if err.Error() != "API returned unexpected status code: 429: Rate limit exceeded" { + t.Errorf("unexpected error message: %s", err.Error()) + } +} + +func TestParseRetryAfterHeader_Integer(t *testing.T) { + h := http.Header{} + h.Set("Retry-After", "60") + resp := &http.Response{Header: h} + d := ParseRetryAfterHeader(resp) + if d != 60*time.Second { + t.Errorf("expected 60s, got %v", d) + } +} + +func TestParseRetryAfterHeader_Absent(t *testing.T) { + resp := &http.Response{Header: http.Header{}} + d := ParseRetryAfterHeader(resp) + if d != 0 { + t.Errorf("expected 0, got %v", d) + } +} + +func TestNewResponseError(t *testing.T) { + h := http.Header{} + h.Set("Retry-After", "10") + resp := &http.Response{ + StatusCode: 429, + Header: h, + } + err := NewResponseError(resp, "rate limited") + if err.StatusCode != 429 { + t.Errorf("expected status 429, got %d", err.StatusCode) + } + if err.Message != "rate limited" { + t.Errorf("unexpected message: %s", err.Message) + } + if err.RetryAfter != 10*time.Second { + t.Errorf("expected 10s RetryAfter, got %v", err.RetryAfter) + } +} + +// mockRetryAfterClassifier simulates a provider MapError that extracts +// RetryAfter from a ResponseError and sets it on a real llms.Error. +func mockRetryAfterClassifier(err error) error { + var respErr *ResponseError + if errors.As(err, &respErr) && respErr.StatusCode == 429 { + return llms.NewError(llms.ErrCodeRateLimit, "test", "rate limited"). + WithCause(err). + WithRetryAfter(respErr.RetryAfter) + } + return nil +} + +// --- RetryOnError tests --- + +func TestRetryOnError_NilConfig_ExecutesOnce(t *testing.T) { + callCount := 0 + err := RetryOnError(context.Background(), nil, nil, func() error { + callCount++ + return fmt.Errorf("some error") + }) + if err == nil { + t.Fatal("expected error") + } + if callCount != 1 { + t.Errorf("expected 1 call with nil config, got %d", callCount) + } +} + +func TestRetryOnError_RetriesOnNetworkError(t *testing.T) { + cfg := &RetryConfig{ + MaxRetries: 2, + InitialBackoff: 1 * time.Millisecond, + MaxBackoff: 5 * time.Millisecond, + BackoffFactor: 2.0, + } + + callCount := 0 + err := RetryOnError(context.Background(), cfg, nil, func() error { + callCount++ + return fmt.Errorf("connection refused") + }) + if err == nil { + t.Fatal("expected error after exhausting retries") + } + if callCount != 3 { // 1 initial + 2 retries + t.Errorf("expected 3 calls, got %d", callCount) + } +} + +func TestRetryOnError_SucceedsAfterRetry(t *testing.T) { + cfg := &RetryConfig{ + MaxRetries: 3, + InitialBackoff: 1 * time.Millisecond, + MaxBackoff: 5 * time.Millisecond, + BackoffFactor: 2.0, + } + + callCount := 0 + err := RetryOnError(context.Background(), cfg, nil, func() error { + callCount++ + if callCount < 3 { + return fmt.Errorf("connection refused") + } + return nil + }) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if callCount != 3 { + t.Errorf("expected 3 calls, got %d", callCount) + } +} + +func TestRetryOnError_UsesClassifier(t *testing.T) { + cfg := &RetryConfig{ + MaxRetries: 2, + InitialBackoff: 1 * time.Millisecond, + MaxBackoff: 5 * time.Millisecond, + BackoffFactor: 2.0, + } + + classifier := func(err error) error { + if err != nil && err.Error() == "server busy" { + return llms.NewError(llms.ErrCodeRateLimit, "test", "classified: rate_limit") + } + return nil + } + + callCount := 0 + err := RetryOnError(context.Background(), cfg, classifier, func() error { + callCount++ + return fmt.Errorf("server busy") + }) + if err == nil { + t.Fatal("expected error") + } + // Should be the llms.Error from classifier + var llmsErr *llms.Error + if !errors.As(err, &llmsErr) { + t.Fatalf("expected llms.Error, got: %T: %v", err, err) + } + if llmsErr.Message != "classified: rate_limit" { + t.Errorf("expected classified message, got: %s", llmsErr.Message) + } +} + +func TestRetryOnError_NoRetryOnNonRetryableError(t *testing.T) { + cfg := &RetryConfig{ + MaxRetries: 3, + InitialBackoff: 1 * time.Millisecond, + MaxBackoff: 5 * time.Millisecond, + BackoffFactor: 2.0, + } + + callCount := 0 + err := RetryOnError(context.Background(), cfg, nil, func() error { + callCount++ + return fmt.Errorf("some non-retryable error") + }) + if err == nil { + t.Fatal("expected error") + } + if callCount != 1 { + t.Errorf("expected 1 call (no retry), got %d", callCount) + } +} + +func TestRetryOnError_ContextCancellation(t *testing.T) { + cfg := &RetryConfig{ + MaxRetries: 3, + InitialBackoff: 100 * time.Millisecond, + MaxBackoff: 500 * time.Millisecond, + BackoffFactor: 2.0, + } + + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + + err := RetryOnError(ctx, cfg, nil, func() error { + return fmt.Errorf("connection refused") + }) + if !errors.Is(err, context.DeadlineExceeded) { + t.Errorf("expected context.DeadlineExceeded, got: %v", err) + } +} + +func TestRetryOnError_RespectsRetryAfter(t *testing.T) { + cfg := &RetryConfig{ + MaxRetries: 2, + InitialBackoff: 1 * time.Millisecond, + MaxBackoff: 5 * time.Millisecond, + BackoffFactor: 2.0, + } + + callCount := 0 + err := RetryOnError(context.Background(), cfg, mockRetryAfterClassifier, func() error { + callCount++ + return &ResponseError{ + StatusCode: 429, + Message: "rate limited", + RetryAfter: 50 * time.Millisecond, + } + }) + if err == nil { + t.Fatal("expected error after exhausting retries") + } + if callCount != 3 { // 1 + 2 retries + t.Errorf("expected 3 calls, got %d", callCount) + } + + // Verify classified error carries RetryAfter + var llmsErr *llms.Error + if !errors.As(err, &llmsErr) { + t.Fatal("expected llms.Error") + } + if llmsErr.RetryAfter() != 50*time.Millisecond { + t.Errorf("expected RetryAfter 50ms, got %v", llmsErr.RetryAfter()) + } +} + +func TestRetryOnError_OnRetryCallback(t *testing.T) { + var mu sync.Mutex + var attempts []int + + cfg := &RetryConfig{ + MaxRetries: 2, + InitialBackoff: 1 * time.Millisecond, + MaxBackoff: 5 * time.Millisecond, + BackoffFactor: 2.0, + OnRetry: func(attempt int, err error) { + mu.Lock() + attempts = append(attempts, attempt) + mu.Unlock() + }, + } + + _ = RetryOnError(context.Background(), cfg, nil, func() error { + return fmt.Errorf("connection refused") + }) + + mu.Lock() + defer mu.Unlock() + if len(attempts) != 2 { + t.Errorf("expected 2 OnRetry calls, got %d", len(attempts)) + } +} + +// --- Integration: RetryOnError with real HTTP server --- + +func TestRetryOnError_Integration(t *testing.T) { + callCount := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + callCount++ + if callCount < 3 { + w.Header().Set("Retry-After", "0") + w.WriteHeader(http.StatusTooManyRequests) + return + } + w.WriteHeader(http.StatusOK) + fmt.Fprint(w, `{"result":"ok"}`) + })) + defer server.Close() + + cfg := &RetryConfig{ + MaxRetries: 3, + InitialBackoff: 1 * time.Millisecond, + MaxBackoff: 5 * time.Millisecond, + BackoffFactor: 2.0, + } + + err := RetryOnError(context.Background(), cfg, mockRetryAfterClassifier, func() error { + req, _ := http.NewRequestWithContext(context.Background(), http.MethodGet, server.URL, nil) + resp, err := http.DefaultClient.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return NewResponseError(resp, fmt.Sprintf("status %d", resp.StatusCode)) + } + return nil + }) + + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if callCount != 3 { + t.Errorf("expected 3 calls, got %d", callCount) + } +} diff --git a/httputil/transport.go b/httputil/transport.go index bfe660fe7..8ed08028d 100644 --- a/httputil/transport.go +++ b/httputil/transport.go @@ -22,6 +22,35 @@ var ( } ) +// NewRetryClient returns an [*http.Client] with retry support. +// The client wraps [http.DefaultTransport] with a [RetryTransport]. +// User-Agent headers are NOT automatically added; use [NewClientWithRetry] +// for the full transport chain. +func NewRetryClient(cfg *RetryConfig) *http.Client { + return &http.Client{ + Transport: &RetryTransport{ + Transport: http.DefaultTransport, + Config: cfg, + }, + } +} + +// NewClientWithRetry returns an [*http.Client] with both retry support +// and User-Agent injection. The transport chain is: +// +// RetryTransport → Transport → http.DefaultTransport +// +// This is the recommended constructor for HTTP clients that need automatic +// retries. +func NewClientWithRetry(cfg *RetryConfig) *http.Client { + return &http.Client{ + Transport: &RetryTransport{ + Transport: DefaultTransport, + Config: cfg, + }, + } +} + // Transport is an [http.RoundTripper] that adds LangChainGo User-Agent headers // to outgoing HTTP requests. It wraps another RoundTripper (typically // [http.DefaultTransport]) and can be used to add User-Agent headers to any diff --git a/llms/anthropic/anthropicllm.go b/llms/anthropic/anthropicllm.go index 417de1b25..ad7de4163 100644 --- a/llms/anthropic/anthropicllm.go +++ b/llms/anthropic/anthropicllm.go @@ -33,7 +33,8 @@ const ( type LLM struct { CallbacksHandler callbacks.Handler client *anthropicclient.Client - model string // Track current model for reasoning detection + model string + retryConfig *httputil.RetryConfig } var ( @@ -43,17 +44,18 @@ var ( // New returns a new Anthropic LLM. func New(opts ...Option) (*LLM, error) { - c, err := newClient(opts...) + c, retryCfg, err := newClient(opts...) if err != nil { return nil, fmt.Errorf("anthropic: failed to create client: %w", err) } return &LLM{ - client: c, - model: c.Model, // Store the model for reasoning detection + client: c, + model: c.Model, + retryConfig: retryCfg, }, nil } -func newClient(opts ...Option) (*anthropicclient.Client, error) { +func newClient(opts ...Option) (*anthropicclient.Client, *httputil.RetryConfig, error) { options := &options{ token: os.Getenv(tokenEnvVarName), baseURL: anthropicclient.DefaultBaseURL, @@ -65,14 +67,15 @@ func newClient(opts ...Option) (*anthropicclient.Client, error) { } if len(options.token) == 0 { - return nil, ErrMissingToken + return nil, nil, ErrMissingToken } - return anthropicclient.New(options.token, options.model, options.baseURL, + c, err := anthropicclient.New(options.token, options.model, options.baseURL, anthropicclient.WithHTTPClient(options.httpClient), anthropicclient.WithLegacyTextCompletionsAPI(options.useLegacyTextCompletionsAPI), anthropicclient.WithAnthropicBetaHeader(options.anthropicBetaHeader), ) + return c, options.retryConfig, err } // Call requests a completion for the given prompt. @@ -109,14 +112,19 @@ func generateCompletionsContent(ctx context.Context, o *LLM, messages []llms.Mes return nil, fmt.Errorf("anthropic: unexpected message type: %T", part) } prompt := fmt.Sprintf("\n\nHuman: %s\n\nAssistant:", partText.Text) - result, err := o.client.CreateCompletion(ctx, &anthropicclient.CompletionRequest{ - Model: opts.Model, - Prompt: prompt, - MaxTokens: opts.MaxTokens, - StopWords: opts.StopWords, - Temperature: opts.Temperature, - TopP: opts.TopP, - StreamingFunc: opts.StreamingFunc, + var result *anthropicclient.Completion + err := httputil.RetryOnError(ctx, o.retryConfig, MapError, func() error { + var retryErr error + result, retryErr = o.client.CreateCompletion(ctx, &anthropicclient.CompletionRequest{ + Model: opts.Model, + Prompt: prompt, + MaxTokens: opts.MaxTokens, + StopWords: opts.StopWords, + Temperature: opts.Temperature, + TopP: opts.TopP, + StreamingFunc: opts.StreamingFunc, + }) + return retryErr }) if err != nil { if o.CallbacksHandler != nil { @@ -145,19 +153,24 @@ 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, - StreamingReasoningFunc: opts.StreamingReasoningFunc, + var msgResult *anthropicclient.MessageResponsePayload + err = httputil.RetryOnError(ctx, o.retryConfig, MapError, func() error { + var retryErr error + msgResult, retryErr = 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, + StreamingReasoningFunc: opts.StreamingReasoningFunc, + }) + return retryErr }) if err != nil { if o.CallbacksHandler != nil { @@ -165,7 +178,7 @@ func generateMessagesContent(ctx context.Context, o *LLM, messages []llms.Messag } return nil, fmt.Errorf("anthropic: failed to create message: %w", err) } - return processAnthropicResponse(result) + return processAnthropicResponse(msgResult) } // processAnthropicResponse converts Anthropic API response to standard ContentResponse diff --git a/llms/anthropic/anthropicllm_option.go b/llms/anthropic/anthropicllm_option.go index 75ae25ded..2ae933de5 100644 --- a/llms/anthropic/anthropicllm_option.go +++ b/llms/anthropic/anthropicllm_option.go @@ -1,6 +1,7 @@ package anthropic import ( + "github.com/tmc/langchaingo/httputil" "github.com/tmc/langchaingo/llms/anthropic/internal/anthropicclient" ) @@ -13,10 +14,11 @@ const ( const MaxTokensAnthropicSonnet35 = "max-tokens-3-5-sonnet-2024-07-15" //nolint:gosec // This is not a sensitive value. type options struct { - token string - model string - baseURL string - httpClient anthropicclient.Doer + token string + model string + baseURL string + httpClient anthropicclient.Doer + retryConfig *httputil.RetryConfig useLegacyTextCompletionsAPI bool @@ -70,3 +72,12 @@ func WithAnthropicBetaHeader(value string) Option { opts.anthropicBetaHeader = value } } + +// WithRetryConfig sets the retry configuration for the HTTP client. +// When set, the internal HTTP client will be wrapped with automatic retry +// behavior using exponential backoff with jitter. +func WithRetryConfig(retryConfig *httputil.RetryConfig) Option { + return func(opts *options) { + opts.retryConfig = retryConfig + } +} diff --git a/llms/anthropic/errors.go b/llms/anthropic/errors.go index 3706e8133..09e134bdc 100644 --- a/llms/anthropic/errors.go +++ b/llms/anthropic/errors.go @@ -1,8 +1,10 @@ package anthropic import ( + "errors" "strings" + "github.com/tmc/langchaingo/httputil" "github.com/tmc/langchaingo/llms" ) @@ -69,7 +71,9 @@ func MapError(err error) error { for _, mapping := range anthropicErrorMappings { for _, pattern := range mapping.patterns { if strings.Contains(errStr, pattern) { - return llms.NewError(mapping.code, "anthropic", mapping.message).WithCause(err) + classified := llms.NewError(mapping.code, "anthropic", mapping.message).WithCause(err) + transferRetryAfter(err, classified) + return classified } } } @@ -78,3 +82,12 @@ func MapError(err error) error { mapper := llms.NewErrorMapper("anthropic") return mapper.Map(err) } + +// transferRetryAfter extracts the Retry-After value from an *httputil.ResponseError +// and sets it on the classified *llms.Error. +func transferRetryAfter(src error, dst *llms.Error) { + var respErr *httputil.ResponseError + if errors.As(src, &respErr) && respErr.RetryAfter > 0 { + _ = dst.WithRetryAfter(respErr.RetryAfter) //nolint:errcheck + } +} diff --git a/llms/anthropic/internal/anthropicclient/anthropicclient.go b/llms/anthropic/internal/anthropicclient/anthropicclient.go index fd3404d36..00b911e5b 100644 --- a/llms/anthropic/internal/anthropicclient/anthropicclient.go +++ b/llms/anthropic/internal/anthropicclient/anthropicclient.go @@ -223,7 +223,7 @@ func (c *Client) decodeError(resp *http.Response) error { var errResp errorMessage if err := json.NewDecoder(resp.Body).Decode(&errResp); err != nil { - return errors.New(msg) + return httputil.NewResponseError(resp, msg) } - return fmt.Errorf("%s: %s", msg, errResp.Error.Message) + return httputil.NewResponseError(resp, fmt.Sprintf("%s: %s", msg, errResp.Error.Message)) } diff --git a/llms/cloudflare/cloudflarellm.go b/llms/cloudflare/cloudflarellm.go index 78135a1d7..330dabc02 100644 --- a/llms/cloudflare/cloudflarellm.go +++ b/llms/cloudflare/cloudflarellm.go @@ -34,6 +34,10 @@ func New(opts ...Option) (*LLM, error) { opt(&o) } + if o.httpClient == httputil.DefaultClient && o.retryConfig != nil { + o.httpClient = httputil.NewClientWithRetry(o.retryConfig) + } + // Default URL if not provided serverURL := "" if o.cloudflareServerURL != nil { diff --git a/llms/cloudflare/options.go b/llms/cloudflare/options.go index ae16a4e86..798abd965 100644 --- a/llms/cloudflare/options.go +++ b/llms/cloudflare/options.go @@ -4,6 +4,8 @@ import ( "log" "net/http" "net/url" + + "github.com/tmc/langchaingo/httputil" ) type options struct { @@ -11,6 +13,7 @@ type options struct { cloudflareServerURL *url.URL cloudflareToken string httpClient *http.Client + retryConfig *httputil.RetryConfig model string embeddingModel string system string @@ -79,3 +82,12 @@ func WithHTTPClient(client *http.Client) Option { opts.httpClient = client } } + +// WithRetryConfig sets the retry configuration for the HTTP client. +// When set, the internal HTTP client will be wrapped with automatic retry +// behavior using exponential backoff with jitter. +func WithRetryConfig(retryConfig *httputil.RetryConfig) Option { + return func(opts *options) { + opts.retryConfig = retryConfig + } +} diff --git a/llms/cohere/coherellm.go b/llms/cohere/coherellm.go index 8e9054165..82bd96ea6 100644 --- a/llms/cohere/coherellm.go +++ b/llms/cohere/coherellm.go @@ -6,6 +6,7 @@ import ( "os" "github.com/tmc/langchaingo/callbacks" + "github.com/tmc/langchaingo/httputil" "github.com/tmc/langchaingo/llms" "github.com/tmc/langchaingo/llms/cohere/internal/cohereclient" ) @@ -90,5 +91,10 @@ func newClient(opts ...Option) (*cohereclient.Client, error) { return nil, ErrMissingToken } - return cohereclient.New(options.token, options.baseURL, options.model) + var clientOpts []cohereclient.Option + if options.retryConfig != nil { + clientOpts = append(clientOpts, cohereclient.WithHTTPClient(httputil.NewClientWithRetry(options.retryConfig))) + } + + return cohereclient.New(options.token, options.baseURL, options.model, clientOpts...) } diff --git a/llms/cohere/coherellm_option.go b/llms/cohere/coherellm_option.go index 584e8dffd..3adc55525 100644 --- a/llms/cohere/coherellm_option.go +++ b/llms/cohere/coherellm_option.go @@ -1,5 +1,9 @@ package cohere +import ( + "github.com/tmc/langchaingo/httputil" +) + const ( tokenEnvVarName = "COHERE_API_KEY" //nolint:gosec modelEnvVarName = "COHERE_MODEL" //nolint:gosec @@ -7,9 +11,10 @@ const ( ) type options struct { - token string - model string - baseURL string + token string + model string + baseURL string + retryConfig *httputil.RetryConfig } type Option func(*options) @@ -38,3 +43,12 @@ func WithBaseURL(baseURL string) Option { opts.baseURL = baseURL } } + +// WithRetryConfig sets the retry configuration for the HTTP client. +// When set, the internal HTTP client will be wrapped with automatic retry +// behavior using exponential backoff with jitter. +func WithRetryConfig(retryConfig *httputil.RetryConfig) Option { + return func(opts *options) { + opts.retryConfig = retryConfig + } +} diff --git a/llms/ernie/erniellm.go b/llms/ernie/erniellm.go index c7cbd95aa..928afc672 100644 --- a/llms/ernie/erniellm.go +++ b/llms/ernie/erniellm.go @@ -7,6 +7,7 @@ import ( "os" "github.com/tmc/langchaingo/callbacks" + "github.com/tmc/langchaingo/httputil" "github.com/tmc/langchaingo/llms" "github.com/tmc/langchaingo/llms/ernie/internal/ernieclient" ) @@ -20,6 +21,7 @@ type LLM struct { client *ernieclient.Client model ModelName CallbacksHandler callbacks.Handler + retryConfig *httputil.RetryConfig } var _ llms.Model = (*LLM)(nil) @@ -41,6 +43,7 @@ func New(opts ...Option) (*LLM, error) { client: c, model: options.modelName, CallbacksHandler: options.callbacksHandler, + retryConfig: options.retryConfig, }, err } @@ -85,13 +88,27 @@ func (o *LLM) GenerateContent(ctx context.Context, messages []llms.MessageConten // Assume we get a single text message msg0 := messages[0] part := msg0.Parts[0] - result, err := o.client.CreateCompletion(ctx, o.getModelPath(*opts), &ernieclient.CompletionRequest{ - Messages: []ernieclient.Message{{Role: "user", Content: part.(llms.TextContent).Text}}, - Temperature: opts.Temperature, - TopP: opts.TopP, - PenaltyScore: opts.RepetitionPenalty, - StreamingFunc: opts.StreamingFunc, - Stream: opts.StreamingFunc != nil, + + var result *ernieclient.Completion + err := httputil.RetryOnError(ctx, o.retryConfig, MapError, func() error { + var retryErr error + result, retryErr = o.client.CreateCompletion(ctx, o.getModelPath(*opts), &ernieclient.CompletionRequest{ + Messages: []ernieclient.Message{{Role: "user", Content: part.(llms.TextContent).Text}}, + Temperature: opts.Temperature, + TopP: opts.TopP, + PenaltyScore: opts.RepetitionPenalty, + StreamingFunc: opts.StreamingFunc, + Stream: opts.StreamingFunc != nil, + }) + if retryErr != nil { + return retryErr + } + // ERNIE returns HTTP 200 with error_code in body for some errors. + if result.ErrorCode > 0 { + return fmt.Errorf("error_code:%v, error_msg:%v, id:%v", + result.ErrorCode, result.ErrorMsg, result.ID) + } + return nil }) if err != nil { if o.CallbacksHandler != nil { @@ -99,14 +116,6 @@ func (o *LLM) GenerateContent(ctx context.Context, messages []llms.MessageConten } return nil, err } - if result.ErrorCode > 0 { - err = fmt.Errorf("%w, error_code:%v, erro_msg:%v, id:%v", - ErrCodeResponse, result.ErrorCode, result.ErrorMsg, result.ID) - if o.CallbacksHandler != nil { - o.CallbacksHandler.HandleLLMError(ctx, err) - } - return nil, err - } resp := &llms.ContentResponse{ Choices: []*llms.ContentChoice{ diff --git a/llms/ernie/erniellm_option.go b/llms/ernie/erniellm_option.go index fcde590b5..0e3fbcc74 100644 --- a/llms/ernie/erniellm_option.go +++ b/llms/ernie/erniellm_option.go @@ -4,6 +4,7 @@ import ( "net/http" "github.com/tmc/langchaingo/callbacks" + "github.com/tmc/langchaingo/httputil" ) const ( @@ -33,6 +34,7 @@ type options struct { modelPath string cacheType string httpClient *http.Client + retryConfig *httputil.RetryConfig } type Option func(*options) @@ -122,3 +124,12 @@ func WithHTTPClient(client *http.Client) Option { opts.httpClient = client } } + +// WithRetryConfig sets the retry configuration for the HTTP client. +// When set, the internal HTTP client will be wrapped with automatic retry +// behavior using exponential backoff with jitter. +func WithRetryConfig(retryConfig *httputil.RetryConfig) Option { + return func(opts *options) { + opts.retryConfig = retryConfig + } +} diff --git a/llms/ernie/errors.go b/llms/ernie/errors.go new file mode 100644 index 000000000..0dfcfc379 --- /dev/null +++ b/llms/ernie/errors.go @@ -0,0 +1,88 @@ +package ernie + +import ( + "errors" + "strings" + + "github.com/tmc/langchaingo/httputil" + "github.com/tmc/langchaingo/llms" +) + +// errorMapping represents a mapping from error patterns to error codes. +type errorMapping struct { + patterns []string + code llms.ErrorCode + message string +} + +// ernieErrorMappings defines the error mappings for ERNIE. +// ERNIE returns errors both as HTTP status codes and as body-level error_code fields. +// Mappings are checked in order; first match wins. +// +// Reference: https://ai.baidu.com/ai-doc/NLP/Bk6z52e59 +var ernieErrorMappings = []errorMapping{ + { + // error_code:18 = QPS rate limit (per-second) — retryable + patterns: []string{"error_code:18", "qps limit"}, + code: llms.ErrCodeRateLimit, + message: "ERNIE QPS limit exceeded", + }, + { + patterns: []string{"error_code:110", "error_code:111", "access token"}, + code: llms.ErrCodeAuthentication, + message: "ERNIE authentication failed", + }, + { + // ERNIE invalid parameter errors + patterns: []string{"error_code:1", "error_code:2", "error_code:3", "invalid parameter"}, + code: llms.ErrCodeInvalidRequest, + message: "Invalid request parameter", + }, + { + // error_code:17 = daily request limit (per-day) — NOT retryable + // error_code:19 = total request limit — NOT retryable + patterns: []string{"error_code:17", "error_code:19", "quota", "limit exceeded"}, + code: llms.ErrCodeQuotaExceeded, + message: "API quota exceeded", + }, + { + // Server-side errors + patterns: []string{"error_code:500", "error_code:503", "internal error", "service unavailable"}, + code: llms.ErrCodeProviderUnavailable, + message: "ERNIE service temporarily unavailable", + }, +} + +// MapError maps ERNIE-specific errors to standardized error codes. +// It handles both HTTP status code errors and body-level error_code fields +// that ERNIE returns with HTTP 200 responses. +func MapError(err error) error { + if err == nil { + return nil + } + + errStr := strings.ToLower(err.Error()) + + for _, mapping := range ernieErrorMappings { + for _, pattern := range mapping.patterns { + if strings.Contains(errStr, strings.ToLower(pattern)) { + classified := llms.NewError(mapping.code, "ernie", mapping.message).WithCause(err) + transferRetryAfter(err, classified) + return classified + } + } + } + + // Fall back to generic error mapper. + mapper := llms.NewErrorMapper("ernie") + return mapper.Map(err) +} + +// transferRetryAfter extracts the Retry-After value from an *httputil.ResponseError +// and sets it on the classified *llms.Error. +func transferRetryAfter(src error, dst *llms.Error) { + var respErr *httputil.ResponseError + if errors.As(src, &respErr) && respErr.RetryAfter > 0 { + _ = dst.WithRetryAfter(respErr.RetryAfter) //nolint:errcheck + } +} diff --git a/llms/ernie/internal/ernieclient/chat.go b/llms/ernie/internal/ernieclient/chat.go index ccbd3343a..5f7a3511a 100644 --- a/llms/ernie/internal/ernieclient/chat.go +++ b/llms/ernie/internal/ernieclient/chat.go @@ -5,12 +5,12 @@ import ( "bytes" "context" "encoding/json" - "errors" "fmt" "log" "net/http" "strings" + "github.com/tmc/langchaingo/httputil" "github.com/tmc/langchaingo/llms" ) @@ -177,10 +177,10 @@ func (c *Client) createChat(ctx context.Context, payload *ChatRequest) (*ChatRes // status code. var errResp errorMessage if err := json.NewDecoder(r.Body).Decode(&errResp); err != nil { - return nil, errors.New(msg) + return nil, httputil.NewResponseError(r, msg) } - return nil, fmt.Errorf("%s: %s", msg, errResp.Error.Message) + return nil, httputil.NewResponseError(r, fmt.Sprintf("%s: %s", msg, errResp.Error.Message)) } if payload.StreamingFunc != nil { return parseStreamingChatResponse(ctx, r, payload) diff --git a/llms/errors.go b/llms/errors.go index 94225e659..56895980e 100644 --- a/llms/errors.go +++ b/llms/errors.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "time" ) // ErrorCode represents a standardized error code for LLM operations. @@ -63,6 +64,10 @@ type Error struct { // Cause is the underlying error, if any. Cause error + + // retryAfter is the duration to wait before retrying, extracted from + // the Retry-After HTTP header. Zero means no hint was provided. + retryAfter time.Duration } // Error implements the error interface. @@ -116,6 +121,17 @@ func (e *Error) WithCause(cause error) *Error { return e } +// WithRetryAfter sets the Retry-After duration hint from the HTTP header. +func (e *Error) WithRetryAfter(d time.Duration) *Error { + e.retryAfter = d + return e +} + +// RetryAfter returns the Retry-After duration hint, or 0 if not set. +func (e *Error) RetryAfter() time.Duration { + return e.retryAfter +} + // WithDetail adds a detail to the error. func (e *Error) WithDetail(key string, value interface{}) *Error { if e.Details == nil { @@ -185,6 +201,26 @@ func IsNotImplementedError(err error) bool { return errors.As(err, &e) && e.Code == ErrCodeNotImplemented } +// IsRetryableErrorCode returns true if the error code represents a transient +// failure that may succeed on retry. This includes rate limits, provider +// unavailability, and timeouts. +func IsRetryableErrorCode(code ErrorCode) bool { + switch code { + case ErrCodeRateLimit, ErrCodeProviderUnavailable, ErrCodeTimeout: + return true + default: + return false + } +} + +// IsRetryableError returns true if the error is classified as retryable. +// It checks if the error is an [*Error] with a retryable error code. +// This works with errors produced by provider [MapError] functions. +func IsRetryableError(err error) bool { + var e *Error + return errors.As(err, &e) && IsRetryableErrorCode(e.Code) +} + // Common error variables for easy comparison. var ( // ErrAuthentication is returned when authentication fails. diff --git a/llms/huggingface/huggingfacellm.go b/llms/huggingface/huggingfacellm.go index 4ca171d18..90bf6c507 100644 --- a/llms/huggingface/huggingfacellm.go +++ b/llms/huggingface/huggingfacellm.go @@ -8,6 +8,7 @@ import ( "path/filepath" "github.com/tmc/langchaingo/callbacks" + "github.com/tmc/langchaingo/httputil" "github.com/tmc/langchaingo/llms" "github.com/tmc/langchaingo/llms/huggingface/internal/huggingfaceclient" ) @@ -97,6 +98,8 @@ func New(opts ...Option) (*LLM, error) { var clientOpts []huggingfaceclient.Option if options.httpClient != nil { clientOpts = append(clientOpts, huggingfaceclient.WithHTTPClient(options.httpClient)) + } else if options.retryConfig != nil { + clientOpts = append(clientOpts, huggingfaceclient.WithHTTPClient(httputil.NewClientWithRetry(options.retryConfig))) } if options.provider != "" { clientOpts = append(clientOpts, huggingfaceclient.WithProvider(options.provider)) diff --git a/llms/huggingface/huggingfacellm_option.go b/llms/huggingface/huggingfacellm_option.go index e083780d1..956679017 100644 --- a/llms/huggingface/huggingfacellm_option.go +++ b/llms/huggingface/huggingfacellm_option.go @@ -2,6 +2,8 @@ package huggingface import ( "net/http" + + "github.com/tmc/langchaingo/httputil" ) const ( @@ -18,11 +20,12 @@ const ( ) type options struct { - token string - model string - url string - httpClient *http.Client - provider string // Inference provider (e.g., "hyperbolic", "nebius") + token string + model string + url string + httpClient *http.Client + retryConfig *httputil.RetryConfig + provider string // Inference provider (e.g., "hyperbolic", "nebius") } type Option func(*options) @@ -58,6 +61,15 @@ func WithHTTPClient(httpClient *http.Client) Option { } } +// WithRetryConfig sets the retry configuration for the HTTP client. +// When set, the internal HTTP client will be wrapped with automatic retry +// behavior using exponential backoff with jitter. +func WithRetryConfig(retryConfig *httputil.RetryConfig) Option { + return func(opts *options) { + opts.retryConfig = retryConfig + } +} + // WithInferenceProvider passes the inference provider to use with HuggingFace's router. // When set, the client will use the router URL (https://router.huggingface.co/{provider}/v1/...) // instead of the default inference API. Common providers include "hyperbolic", "nebius", etc. diff --git a/llms/llamafile/options.go b/llms/llamafile/options.go index 8790491bd..c707093b6 100644 --- a/llms/llamafile/options.go +++ b/llms/llamafile/options.go @@ -1,6 +1,9 @@ package llamafile -import "github.com/tmc/langchaingo/llms/llamafile/internal/llamafileclient" +import ( + "github.com/tmc/langchaingo/httputil" + "github.com/tmc/langchaingo/llms/llamafile/internal/llamafileclient" +) type Option func(*llamafileclient.GenerationSettings) @@ -199,3 +202,12 @@ func WithEmbeddingSize(val int) Option { g.EmbeddingSize = val } } + +// WithRetryConfig sets the retry configuration for the HTTP client. +// When set, the internal HTTP client will be wrapped with automatic retry +// behavior using exponential backoff with jitter. +func WithRetryConfig(retryConfig *httputil.RetryConfig) Option { + return func(g *llamafileclient.GenerationSettings) { + g.HTTPClient = httputil.NewClientWithRetry(retryConfig) + } +} diff --git a/llms/maritaca/maritacallm.go b/llms/maritaca/maritacallm.go index 495bad16c..e0c81f41d 100644 --- a/llms/maritaca/maritacallm.go +++ b/llms/maritaca/maritacallm.go @@ -32,7 +32,11 @@ func New(opts ...Option) (*LLM, error) { } if o.httpClient == nil { - o.httpClient = httputil.DefaultClient + if o.retryConfig != nil { + o.httpClient = httputil.NewClientWithRetry(o.retryConfig) + } else { + o.httpClient = httputil.DefaultClient + } } client, err := maritacaclient.NewClient(o.httpClient) diff --git a/llms/maritaca/options.go b/llms/maritaca/options.go index 5990dfce2..c00acb3b4 100644 --- a/llms/maritaca/options.go +++ b/llms/maritaca/options.go @@ -5,12 +5,14 @@ import ( "net/http" "net/url" + "github.com/tmc/langchaingo/httputil" "github.com/tmc/langchaingo/llms/maritaca/internal/maritacaclient" ) type options struct { maritacaServerURL *url.URL httpClient *http.Client + retryConfig *httputil.RetryConfig model string maritacaOptions maritacaclient.Options customModelTemplate string @@ -69,6 +71,15 @@ func WithHTTPClient(client *http.Client) Option { } } +// WithRetryConfig sets the retry configuration for the HTTP client. +// When set, the internal HTTP client will be wrapped with automatic retry +// behavior using exponential backoff with jitter. +func WithRetryConfig(retryConfig *httputil.RetryConfig) Option { + return func(opts *options) { + opts.retryConfig = retryConfig + } +} + // WithChatMode Set the chat mode. // default: true // If True, the model will run in chat mode, where messages is a string containing the diff --git a/llms/ollama/ollamallm.go b/llms/ollama/ollamallm.go index 493be28e0..c4fd33ebc 100644 --- a/llms/ollama/ollamallm.go +++ b/llms/ollama/ollamallm.go @@ -7,6 +7,7 @@ import ( "strings" "github.com/tmc/langchaingo/callbacks" + "github.com/tmc/langchaingo/httputil" "github.com/tmc/langchaingo/llms" "github.com/tmc/langchaingo/llms/ollama/internal/ollamaclient" ) @@ -37,6 +38,10 @@ func New(opts ...Option) (*LLM, error) { opt(&o) } + if o.httpClient == nil && o.retryConfig != nil { + o.httpClient = httputil.NewClientWithRetry(o.retryConfig) + } + client, err := ollamaclient.NewClient(o.ollamaServerURL, o.httpClient) if err != nil { return nil, err diff --git a/llms/ollama/options.go b/llms/ollama/options.go index 5d2940f32..d93c31d75 100644 --- a/llms/ollama/options.go +++ b/llms/ollama/options.go @@ -6,12 +6,14 @@ import ( "net/url" "time" + "github.com/tmc/langchaingo/httputil" "github.com/tmc/langchaingo/llms/ollama/internal/ollamaclient" ) type options struct { ollamaServerURL *url.URL httpClient *http.Client + retryConfig *httputil.RetryConfig model string ollamaOptions ollamaclient.Options customModelTemplate string @@ -86,6 +88,15 @@ func WithHTTPClient(client *http.Client) Option { } } +// WithRetryConfig sets the retry configuration for the HTTP client. +// When set, the internal HTTP client will be wrapped with automatic retry +// behavior using exponential backoff with jitter. +func WithRetryConfig(retryConfig *httputil.RetryConfig) Option { + return func(opts *options) { + opts.retryConfig = retryConfig + } +} + // WithBackendUseNUMA Use NUMA optimization on certain systems. func WithRunnerUseNUMA(numa bool) Option { return func(opts *options) { diff --git a/llms/openai/errors.go b/llms/openai/errors.go index aaabc4929..bbe705e76 100644 --- a/llms/openai/errors.go +++ b/llms/openai/errors.go @@ -1,8 +1,10 @@ package openai import ( + "errors" "strings" + "github.com/tmc/langchaingo/httputil" "github.com/tmc/langchaingo/llms" ) @@ -69,7 +71,10 @@ func MapError(err error) error { for _, mapping := range openaiErrorMappings { for _, pattern := range mapping.patterns { if strings.Contains(errStr, pattern) { - return llms.NewError(mapping.code, "openai", mapping.message).WithCause(err) + classified := llms.NewError(mapping.code, "openai", mapping.message).WithCause(err) + // Transfer Retry-After from HTTP response error. + transferRetryAfter(err, classified) + return classified } } } @@ -78,3 +83,12 @@ func MapError(err error) error { mapper := llms.NewErrorMapper("openai") return mapper.Map(err) } + +// transferRetryAfter extracts the Retry-After value from an *httputil.ResponseError +// and sets it on the classified *llms.Error. +func transferRetryAfter(src error, dst *llms.Error) { + var respErr *httputil.ResponseError + if errors.As(src, &respErr) && respErr.RetryAfter > 0 { + _ = dst.WithRetryAfter(respErr.RetryAfter) //nolint:errcheck + } +} diff --git a/llms/openai/internal/openaiclient/chat.go b/llms/openai/internal/openaiclient/chat.go index 991c236ab..7bce4a0b3 100644 --- a/llms/openai/internal/openaiclient/chat.go +++ b/llms/openai/internal/openaiclient/chat.go @@ -10,6 +10,7 @@ import ( "net/http" "strings" + "github.com/tmc/langchaingo/httputil" "github.com/tmc/langchaingo/llms" ) @@ -87,6 +88,18 @@ type ChatRequest struct { // WebSearchOptions configures web search behavior for search-enabled models // like gpt-4o-search-preview and gpt-4o-mini-search-preview. WebSearchOptions *WebSearchOptions `json:"web_search_options,omitempty"` + + // EnableThinking enables Qwen's deep thinking mode via OpenAI-compatible API. + // Must be used with streaming (Qwen requires enable_thinking=false for non-streaming calls). + EnableThinking *bool `json:"enable_thinking,omitempty"` + + // EnableDeepSeekThinking enables Qwen's deep thinking mode via OpenAI-compatible API. + // Must be used with streaming (Qwen requires enable_thinking=false for non-streaming calls). + DeepSeekThinking map[string]any `json:"thinking,omitempty"` + + // ThinkingBudget limits the thinking tokens for Qwen models. + // Supported by Qwen3+ models. + ThinkingBudget int `json:"thinking_budget,omitempty"` } // MarshalJSON ensures that only one of MaxTokens or MaxCompletionTokens is sent. @@ -555,10 +568,10 @@ func (c *Client) createChat(ctx context.Context, payload *ChatRequest) (*ChatCom // status code. var errResp errorMessage if err := json.NewDecoder(r.Body).Decode(&errResp); err != nil { - return nil, errors.New(msg) + return nil, httputil.NewResponseError(r, msg) } - return nil, fmt.Errorf("%s: %s", msg, errResp.Error.Message) + return nil, httputil.NewResponseError(r, fmt.Sprintf("%s: %s", msg, errResp.Error.Message)) } if payload.StreamingFunc != nil || payload.StreamingReasoningFunc != nil { return parseStreamingChatResponse(ctx, r, payload) diff --git a/llms/openai/llm.go b/llms/openai/llm.go index 1f2897c77..d40424fae 100644 --- a/llms/openai/llm.go +++ b/llms/openai/llm.go @@ -31,6 +31,7 @@ func newClient(opts ...Option) (*options, *openaiclient.Client, error) { for _, opt := range opts { opt(options) } + // set of options needed for Azure client if openaiclient.IsAzure(openaiclient.APIType(options.apiType)) && options.apiVersion == "" { options.apiVersion = DefaultAPIVersion diff --git a/llms/openai/openaillm.go b/llms/openai/openaillm.go index 84690072a..52455b597 100644 --- a/llms/openai/openaillm.go +++ b/llms/openai/openaillm.go @@ -7,6 +7,7 @@ import ( "strings" "github.com/tmc/langchaingo/callbacks" + "github.com/tmc/langchaingo/httputil" "github.com/tmc/langchaingo/llms" "github.com/tmc/langchaingo/llms/openai/internal/openaiclient" ) @@ -17,6 +18,7 @@ type LLM struct { CallbacksHandler callbacks.Handler client *openaiclient.Client model string // Track current model for reasoning detection + retryConfig *httputil.RetryConfig } const ( @@ -91,7 +93,8 @@ func New(opts ...Option) (*LLM, error) { return &LLM{ client: c, CallbacksHandler: opt.callbackHandler, - model: c.Model, // Store the model for reasoning detection + model: c.Model, + retryConfig: opt.retryConfig, }, err } @@ -250,7 +253,7 @@ func (o *LLM) GenerateContent(ctx context.Context, messages []llms.MessageConten if opts.Metadata != nil { for k, v := range opts.Metadata { // Skip internal metadata keys - if k == "thinking_config" || strings.HasPrefix(k, "openai:") { + if k == "thinking_config" || strings.HasPrefix(k, "openai:") || strings.HasPrefix(k, "qwen:") { continue } apiMetadata[k] = v @@ -261,6 +264,22 @@ func (o *LLM) GenerateContent(ctx context.Context, messages []llms.MessageConten apiMetadata = nil } + // Extract Qwen enable_thinking parameter from metadata + var enableThinking *bool + var thinkingBudget int + var enableDeepSeekThinking map[string]any + if opts.Metadata != nil { + if v, ok := opts.Metadata["qwen:enable_thinking"].(bool); ok { + enableThinking = &v + } + if v, ok := opts.Metadata["deepseek:enable_thinking"].(map[string]any); ok { + enableDeepSeekThinking = v + } + if v, ok := opts.Metadata["qwen:thinking_budget"].(int); ok { + thinkingBudget = v + } + } + req := &openaiclient.ChatRequest{ Model: opts.Model, StopWords: opts.StopWords, @@ -293,6 +312,9 @@ func (o *LLM) GenerateContent(ctx context.Context, messages []llms.MessageConten FunctionCallBehavior: openaiclient.FunctionCallBehavior(opts.FunctionCallBehavior), Seed: opts.Seed, Metadata: apiMetadata, + EnableThinking: enableThinking, + DeepSeekThinking: enableDeepSeekThinking, + ThinkingBudget: thinkingBudget, WebSearchOptions: webSearchOptionsFromCallOptions(opts.WebSearchOptions), } if opts.JSONMode { @@ -325,7 +347,12 @@ func (o *LLM) GenerateContent(ctx context.Context, messages []llms.MessageConten req.ResponseFormat = o.client.ResponseFormat } - result, err := o.client.CreateChat(ctx, req) + var result *openaiclient.ChatCompletionResponse + err := httputil.RetryOnError(ctx, o.retryConfig, MapError, func() error { + var retryErr error + result, retryErr = o.client.CreateChat(ctx, req) + return retryErr + }) if err != nil { return nil, err } @@ -425,9 +452,14 @@ func (o *LLM) SupportsReasoning() bool { // CreateEmbedding creates embeddings for the given input texts. func (o *LLM) CreateEmbedding(ctx context.Context, inputTexts []string) ([][]float32, error) { - embeddings, err := o.client.CreateEmbedding(ctx, &openaiclient.EmbeddingRequest{ - Input: inputTexts, - Model: o.client.EmbeddingModel, + var embeddings [][]float32 + err := httputil.RetryOnError(ctx, o.retryConfig, MapError, func() error { + var retryErr error + embeddings, retryErr = o.client.CreateEmbedding(ctx, &openaiclient.EmbeddingRequest{ + Input: inputTexts, + Model: o.client.EmbeddingModel, + }) + return retryErr }) if err != nil { return nil, fmt.Errorf("failed to create openai embeddings: %w", err) diff --git a/llms/openai/openaillm_option.go b/llms/openai/openaillm_option.go index f2ffe93fe..f8a8c413a 100644 --- a/llms/openai/openaillm_option.go +++ b/llms/openai/openaillm_option.go @@ -2,6 +2,7 @@ package openai import ( "github.com/tmc/langchaingo/callbacks" + "github.com/tmc/langchaingo/httputil" "github.com/tmc/langchaingo/llms/openai/internal/openaiclient" ) @@ -32,6 +33,7 @@ type options struct { organization string apiType APIType httpClient openaiclient.Doer + retryConfig *httputil.RetryConfig responseFormat *ResponseFormat @@ -146,3 +148,12 @@ func WithResponseFormat(responseFormat *ResponseFormat) Option { opts.responseFormat = responseFormat } } + +// WithRetryConfig sets the retry configuration for the HTTP client. +// When set, the internal HTTP client will be wrapped with automatic retry +// behavior using exponential backoff with jitter. +func WithRetryConfig(retryConfig *httputil.RetryConfig) Option { + return func(opts *options) { + opts.retryConfig = retryConfig + } +} diff --git a/llms/openai/options.go b/llms/openai/options.go index 0968b3724..fd2bc83de 100644 --- a/llms/openai/options.go +++ b/llms/openai/options.go @@ -37,3 +37,55 @@ func WithLegacyMaxTokensField() llms.CallOption { opts.Metadata["openai:use_legacy_max_tokens"] = true } } + +// WithEnableThinking enables Qwen's deep thinking mode via OpenAI-compatible API. +// Must be used with streaming — Qwen requires enable_thinking=false for non-streaming calls. +// +// Usage: +// +// llm.GenerateContent(ctx, messages, +// openai.WithEnableThinking(true), +// llms.WithStreamingFunc(func(ctx context.Context, chunk []byte) error { +// fmt.Print(string(chunk)) +// return nil +// }), +// ) +func WithEnableThinking(enabled bool) llms.CallOption { + return func(opts *llms.CallOptions) { + if opts.Metadata == nil { + opts.Metadata = make(map[string]interface{}) + } + opts.Metadata["qwen:enable_thinking"] = enabled + if enabled { + opts.Metadata["deepseek:enable_thinking"] = map[string]any{"type": "enabled"} + } else { + opts.Metadata["deepseek:enable_thinking"] = map[string]any{"type": "disabled"} + } + } +} + +func WithEnableDeepSeekThinking(data map[string]any) llms.CallOption { + return func(opts *llms.CallOptions) { + if opts.Metadata == nil { + opts.Metadata = make(map[string]interface{}) + } + opts.Metadata["deepseek:enable_thinking"] = data + } +} + +// WithThinkingBudget limits thinking tokens for Qwen models (Qwen3+). +// +// Usage: +// +// llm.GenerateContent(ctx, messages, +// openai.WithEnableThinking(true), +// openai.WithThinkingBudget(1000), +// ) +func WithThinkingBudget(budget int) llms.CallOption { + return func(opts *llms.CallOptions) { + if opts.Metadata == nil { + opts.Metadata = make(map[string]interface{}) + } + opts.Metadata["qwen:thinking_budget"] = budget + } +}