0
# Error Handling
1
2
Comprehensive exception classes and error handling patterns for robust Replicate API integration.
3
4
## Capabilities
5
6
### Exception Hierarchy
7
8
The Replicate client provides a structured exception hierarchy for different types of errors.
9
10
```python { .api }
11
class ReplicateException(Exception):
12
"""A base class for all Replicate exceptions."""
13
14
class ModelError(ReplicateException):
15
"""An error from user's code in a model."""
16
17
prediction: Prediction
18
19
def __init__(self, prediction: Prediction) -> None:
20
"""
21
Initialize with the failed prediction.
22
23
Parameters:
24
- prediction: The prediction that encountered an error
25
"""
26
27
class ReplicateError(ReplicateException):
28
"""
29
An error from Replicate's API.
30
31
This class represents a problem details response as defined in RFC 7807.
32
"""
33
34
type: Optional[str]
35
"""A URI that identifies the error type."""
36
37
title: Optional[str]
38
"""A short, human-readable summary of the error."""
39
40
status: Optional[int]
41
"""The HTTP status code."""
42
43
detail: Optional[str]
44
"""A human-readable explanation specific to this occurrence of the error."""
45
46
instance: Optional[str]
47
"""A URI that identifies the specific occurrence of the error."""
48
49
def __init__(
50
self,
51
type: Optional[str] = None,
52
title: Optional[str] = None,
53
status: Optional[int] = None,
54
detail: Optional[str] = None,
55
instance: Optional[str] = None
56
) -> None: ...
57
58
@classmethod
59
def from_response(cls, response: httpx.Response) -> "ReplicateError":
60
"""Create a ReplicateError from an HTTP response."""
61
62
def to_dict(self) -> dict:
63
"""Get a dictionary representation of the error."""
64
```
65
66
### Webhook Validation Exceptions
67
68
Specific exceptions for webhook validation and processing.
69
70
```python { .api }
71
class WebhookValidationError(ValueError):
72
"""Base webhook validation error."""
73
74
class MissingWebhookHeaderError(WebhookValidationError):
75
"""Missing required webhook header."""
76
77
class InvalidSecretKeyError(WebhookValidationError):
78
"""Invalid webhook secret key."""
79
80
class MissingWebhookBodyError(WebhookValidationError):
81
"""Missing webhook request body."""
82
83
class InvalidTimestampError(WebhookValidationError):
84
"""Invalid or expired timestamp."""
85
86
class InvalidSignatureError(WebhookValidationError):
87
"""Invalid webhook signature."""
88
```
89
90
### Usage Examples
91
92
#### Basic Error Handling
93
94
```python
95
import replicate
96
from replicate.exceptions import ModelError, ReplicateError, ReplicateException
97
from replicate.webhook import (
98
WebhookValidationError,
99
MissingWebhookHeaderError,
100
InvalidSecretKeyError,
101
MissingWebhookBodyError,
102
InvalidTimestampError,
103
InvalidSignatureError
104
)
105
106
try:
107
output = replicate.run(
108
"stability-ai/stable-diffusion-3",
109
input={"prompt": "An astronaut riding a rainbow unicorn"}
110
)
111
112
with open("output.png", "wb") as f:
113
f.write(output.read())
114
115
except ModelError as e:
116
# Model execution failed
117
print(f"Model execution failed for prediction {e.prediction.id}")
118
print(f"Error: {e.prediction.error}")
119
print(f"Status: {e.prediction.status}")
120
121
# Check logs for debugging
122
if e.prediction.logs:
123
print(f"Logs:\n{e.prediction.logs}")
124
125
except ReplicateError as e:
126
# API error
127
print(f"API Error: {e.title}")
128
print(f"Status: {e.status}")
129
print(f"Detail: {e.detail}")
130
131
# Check error type for specific handling
132
if e.status == 429:
133
print("Rate limited - consider adding delays between requests")
134
elif e.status == 402:
135
print("Payment required - check your account balance")
136
elif e.status == 404:
137
print("Model not found - verify the model name and version")
138
139
except ReplicateException as e:
140
# Generic Replicate error
141
print(f"Replicate error: {e}")
142
143
except Exception as e:
144
# Unexpected error
145
print(f"Unexpected error: {e}")
146
```
147
148
#### Model Error Inspection
149
150
```python
151
import replicate
152
from replicate.exceptions import ModelError
153
154
try:
155
# This might fail due to invalid input
156
output = replicate.run(
157
"some-model/that-might-fail",
158
input={"invalid_parameter": "bad_value"}
159
)
160
161
except ModelError as e:
162
prediction = e.prediction
163
164
print(f"Model Error Details:")
165
print(f"Prediction ID: {prediction.id}")
166
print(f"Status: {prediction.status}")
167
print(f"Error: {prediction.error}")
168
169
# Analyze logs for common issues
170
if prediction.logs:
171
logs = prediction.logs.lower()
172
173
if "out of memory" in logs:
174
print("Suggestion: Try reducing batch size or image resolution")
175
elif "invalid input" in logs:
176
print("Suggestion: Check input parameters against model schema")
177
elif "timeout" in logs:
178
print("Suggestion: Model might need more time, try again")
179
180
# Print recent logs
181
log_lines = prediction.logs.split('\n')
182
print(f"Recent logs:\n" + '\n'.join(log_lines[-10:]))
183
184
# Check if prediction can be retried
185
if prediction.status == "failed":
186
print("Prediction failed permanently")
187
else:
188
print("Prediction might still be retryable")
189
```
190
191
#### API Error Handling
192
193
```python
194
import replicate
195
from replicate.exceptions import ReplicateError
196
import time
197
198
def robust_prediction(model_name, input_params, max_retries=3):
199
"""Create prediction with retry logic for transient errors."""
200
201
for attempt in range(max_retries):
202
try:
203
return replicate.predictions.create(
204
model=model_name,
205
input=input_params
206
)
207
208
except ReplicateError as e:
209
if e.status == 429: # Rate limited
210
wait_time = 2 ** attempt # Exponential backoff
211
print(f"Rate limited, waiting {wait_time} seconds...")
212
time.sleep(wait_time)
213
continue
214
215
elif e.status == 503: # Service unavailable
216
wait_time = 5 * (attempt + 1)
217
print(f"Service unavailable, waiting {wait_time} seconds...")
218
time.sleep(wait_time)
219
continue
220
221
elif e.status == 402: # Payment required
222
print("Payment required - check account balance")
223
raise
224
225
elif e.status == 404: # Not found
226
print(f"Model not found: {model_name}")
227
raise
228
229
elif e.status == 422: # Validation error
230
print(f"Invalid input parameters: {e.detail}")
231
print("Available parameters might have changed")
232
raise
233
234
else:
235
# Other API errors
236
print(f"API Error: {e.title} (Status: {e.status})")
237
print(f"Detail: {e.detail}")
238
239
if attempt == max_retries - 1:
240
raise
241
else:
242
time.sleep(2)
243
continue
244
245
raise Exception(f"Failed after {max_retries} attempts")
246
247
# Usage
248
try:
249
prediction = robust_prediction(
250
"stability-ai/stable-diffusion-3",
251
{"prompt": "a beautiful landscape"}
252
)
253
print(f"Prediction created: {prediction.id}")
254
255
except Exception as e:
256
print(f"Failed to create prediction: {e}")
257
```
258
259
#### Webhook Error Handling
260
261
```python
262
import replicate
263
from replicate.exceptions import (
264
WebhookValidationError,
265
MissingWebhookHeaderError,
266
InvalidSecretKeyError,
267
InvalidSignatureError,
268
InvalidTimestampError
269
)
270
271
def handle_webhook_request(request):
272
"""Handle webhook with comprehensive error handling."""
273
274
try:
275
# Extract required headers
276
signature = request.headers.get('Replicate-Signature')
277
timestamp = request.headers.get('Replicate-Timestamp')
278
279
if not signature:
280
raise MissingWebhookHeaderError("Missing Replicate-Signature header")
281
282
if not timestamp:
283
raise MissingWebhookHeaderError("Missing Replicate-Timestamp header")
284
285
# Get request body
286
body = request.body
287
if not body:
288
raise MissingWebhookBodyError("Empty request body")
289
290
# Get webhook secret and validate
291
secret_obj = replicate.webhooks.default.secret()
292
replicate.webhooks.validate(
293
headers={"webhook-signature": signature, "webhook-timestamp": timestamp, "webhook-id": "webhook-id"},
294
body=body.decode('utf-8'),
295
secret=secret_obj,
296
tolerance=300
297
)
298
299
# Process webhook payload
300
import json
301
payload = json.loads(body.decode('utf-8'))
302
303
# Handle different event types
304
if payload.get('status') == 'succeeded':
305
handle_success(payload)
306
elif payload.get('status') == 'failed':
307
handle_failure(payload)
308
309
return {"success": True}, 200
310
311
except MissingWebhookHeaderError as e:
312
print(f"Missing webhook header: {e}")
313
return {"error": "Missing required headers"}, 400
314
315
except InvalidSignatureError as e:
316
print(f"Invalid webhook signature: {e}")
317
return {"error": "Invalid signature"}, 401
318
319
except InvalidTimestampError as e:
320
print(f"Invalid timestamp: {e}")
321
return {"error": "Request too old"}, 400
322
323
except WebhookValidationError as e:
324
print(f"Webhook validation failed: {e}")
325
return {"error": "Validation failed"}, 400
326
327
except json.JSONDecodeError as e:
328
print(f"Invalid JSON payload: {e}")
329
return {"error": "Invalid JSON"}, 400
330
331
except Exception as e:
332
print(f"Unexpected webhook error: {e}")
333
return {"error": "Internal server error"}, 500
334
335
def handle_success(payload):
336
"""Handle successful prediction webhook."""
337
prediction_id = payload.get('id')
338
output = payload.get('output')
339
340
print(f"Prediction {prediction_id} succeeded")
341
342
if output:
343
# Process output files
344
for i, url in enumerate(output):
345
print(f"Output {i}: {url}")
346
347
def handle_failure(payload):
348
"""Handle failed prediction webhook."""
349
prediction_id = payload.get('id')
350
error = payload.get('error')
351
logs = payload.get('logs')
352
353
print(f"Prediction {prediction_id} failed: {error}")
354
355
if logs:
356
# Log failure details for debugging
357
print(f"Error logs:\n{logs}")
358
```
359
360
#### Error Recovery Patterns
361
362
```python
363
import replicate
364
from replicate.exceptions import ModelError, ReplicateError
365
import time
366
import logging
367
368
logging.basicConfig(level=logging.INFO)
369
logger = logging.getLogger(__name__)
370
371
class ReplicateClient:
372
"""Wrapper class with error recovery and retry logic."""
373
374
def __init__(self, max_retries=3, base_delay=1):
375
self.max_retries = max_retries
376
self.base_delay = base_delay
377
378
def run_with_retry(self, model, input_params, **kwargs):
379
"""Run model with automatic retry and error recovery."""
380
381
for attempt in range(self.max_retries):
382
try:
383
# Attempt to run model
384
prediction = replicate.predictions.create(
385
model=model,
386
input=input_params,
387
**kwargs
388
)
389
390
# Wait for completion
391
prediction.wait()
392
393
if prediction.status == "succeeded":
394
return prediction.output
395
elif prediction.status == "failed":
396
raise ModelError(prediction)
397
else:
398
raise Exception(f"Unexpected status: {prediction.status}")
399
400
except ModelError as e:
401
logger.error(f"Model error (attempt {attempt + 1}): {e.prediction.error}")
402
403
# Check if error is retryable
404
error_msg = e.prediction.error.lower() if e.prediction.error else ""
405
406
if "out of memory" in error_msg:
407
# Try to reduce input complexity
408
input_params = self._reduce_complexity(input_params)
409
logger.info("Reduced input complexity for retry")
410
elif "timeout" in error_msg:
411
# Just retry - might be transient
412
logger.info("Timeout error, retrying...")
413
else:
414
# Non-retryable error
415
logger.error("Non-retryable model error")
416
raise
417
418
if attempt < self.max_retries - 1:
419
delay = self.base_delay * (2 ** attempt)
420
logger.info(f"Waiting {delay} seconds before retry...")
421
time.sleep(delay)
422
else:
423
raise
424
425
except ReplicateError as e:
426
logger.error(f"API error (attempt {attempt + 1}): {e.title}")
427
428
if e.status == 429: # Rate limited
429
delay = self.base_delay * (2 ** attempt)
430
logger.info(f"Rate limited, waiting {delay} seconds...")
431
time.sleep(delay)
432
elif e.status in [500, 502, 503, 504]: # Server errors
433
delay = self.base_delay * (2 ** attempt)
434
logger.info(f"Server error, waiting {delay} seconds...")
435
time.sleep(delay)
436
else:
437
# Non-retryable API error
438
raise
439
440
except Exception as e:
441
logger.error(f"Unexpected error (attempt {attempt + 1}): {e}")
442
if attempt < self.max_retries - 1:
443
time.sleep(self.base_delay)
444
else:
445
raise
446
447
raise Exception(f"Failed after {self.max_retries} attempts")
448
449
def _reduce_complexity(self, input_params):
450
"""Reduce input complexity to avoid memory issues."""
451
params = input_params.copy()
452
453
# Reduce common parameters that might cause OOM
454
if 'num_inference_steps' in params:
455
params['num_inference_steps'] = min(params['num_inference_steps'], 20)
456
457
if 'guidance_scale' in params:
458
params['guidance_scale'] = min(params['guidance_scale'], 7.5)
459
460
if 'width' in params and 'height' in params:
461
# Reduce resolution
462
params['width'] = min(params['width'], 512)
463
params['height'] = min(params['height'], 512)
464
465
return params
466
467
# Usage
468
client = ReplicateClient(max_retries=3)
469
470
try:
471
output = client.run_with_retry(
472
"stability-ai/stable-diffusion-3",
473
{
474
"prompt": "a beautiful landscape",
475
"width": 1024,
476
"height": 1024,
477
"num_inference_steps": 50
478
}
479
)
480
481
# Save output
482
with open("output.png", "wb") as f:
483
f.write(output.read())
484
485
print("Generation completed successfully")
486
487
except Exception as e:
488
logger.error(f"Final failure: {e}")
489
```