A production-oriented Python example for adding retries and model fallback to an OpenAI-compatible API client.
This guide uses CometAPI as the API endpoint, but the reliability pattern is intentionally portable. The application owns the retry policy, model sequence, response validation, and logging rather than assuming every error should be retried.
Model availability, model IDs, endpoint behavior, pricing, streaming, tool calling, and structured-output support can change. Verify the current CometAPI documentation and test the exact features your application depends on before deploying this pattern to production.
An application that sends every request to one model has a simple dependency chain:
Application → One model → One upstream provider
When that model is rate-limited or its provider returns a temporary error, the entire application can become unavailable.
A fallback chain gives the client another path:
flowchart LR
A[Application] --> B[OpenAI-compatible client]
B --> C[Primary model]
C -->|429, 5xx, timeout| D[Retry with backoff]
D -->|Still unavailable| E[Fallback model]
C -->|400, 401, 403, 404, 422| F[Fail fast]
E --> G[Validated response]
The important distinction is that not every error is recoverable.
A temporary 503 Service Unavailable response may justify a retry or fallback. An invalid API key does not. Retrying the same invalid credential against several models only increases latency and hides the actual configuration problem.
This example divides failures into three groups.
| Failure type | Examples | Client behavior |
|---|---|---|
| Configuration or request error | 400, 401, 403, 404, 422 |
Stop immediately and surface the error |
| Temporary transport or provider error | Timeout, connection error, 408, 429, 500, 502, 503, 504 |
Retry with backoff, then move to the next model |
| Invalid successful response | Empty choices or missing text | Retry, then move to the next model |
This is a client policy, not a claim that every provider always uses the same status code for the same condition. Validate the actual error behavior of each endpoint and model you use.
cometapi-multi-model-fallback/
├── README.md
├── fallback.py
├── requirements.txt
├── .env.example
├── .gitignore
└── LICENSE
- Python 3.10 or later
- A CometAPI credential
- At least two valid model IDs available to your account
- The official OpenAI Python package
Install the dependency:
pip install openairequirements.txt:
openai>=1.0.0
For reproducible production deployments, pin and test an exact package version rather than relying indefinitely on a broad version range.
Create a .env.example file:
COMETAPI_API_KEY=replace-with-your-api-key
COMETAPI_BASE_URL=https://api.cometapi.com/v1
COMETAPI_MODEL_CHAIN=primary-model-id,fallback-model-idDo not commit a real API key.
Add this to .gitignore:
.env
.venv/
__pycache__/
*.pycSet the variables in your shell:
export COMETAPI_API_KEY="your-api-key"
export COMETAPI_BASE_URL="https://api.cometapi.com/v1"
export COMETAPI_MODEL_CHAIN="primary-model-id,fallback-model-id"Replace the example model IDs with IDs confirmed in the current CometAPI documentation or your account dashboard.
Create fallback.py:
from __future__ import annotations
import logging
import os
import random
import time
from dataclasses import dataclass
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime
from typing import Any
from openai import (
APIConnectionError,
APIStatusError,
APITimeoutError,
OpenAI,
)
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)s %(message)s",
)
logger = logging.getLogger(__name__)
# Errors in this group normally indicate that the request or configuration
# must be corrected. Sending the same request again is unlikely to help.
FAIL_FAST_STATUS_CODES = {
400, # Invalid request or unsupported parameter
401, # Invalid or missing credential
403, # Permission denied
404, # Endpoint or model not found
422, # Request could not be processed
}
# Errors in this group may be temporary.
TRANSIENT_STATUS_CODES = {
408, # Request timeout
429, # Rate limit
500, # Internal server error
502, # Bad gateway
503, # Service unavailable
504, # Gateway timeout
}
class InvalidModelResponse(RuntimeError):
"""Raised when an HTTP-successful response is unusable."""
@dataclass(frozen=True)
class Settings:
api_key: str
base_url: str
models: tuple[str, ...]
timeout_seconds: float = 30.0
attempts_per_model: int = 3
backoff_base_seconds: float = 1.0
backoff_cap_seconds: float = 20.0
def load_settings() -> Settings:
"""Load and validate application configuration."""
api_key = os.getenv("COMETAPI_API_KEY", "").strip()
base_url = os.getenv(
"COMETAPI_BASE_URL",
"https://api.cometapi.com/v1",
).strip()
model_chain = os.getenv("COMETAPI_MODEL_CHAIN", "")
models = tuple(
model.strip()
for model in model_chain.split(",")
if model.strip()
)
if not api_key:
raise RuntimeError(
"COMETAPI_API_KEY is missing. "
"Set it as an environment variable."
)
if not models:
raise RuntimeError(
"COMETAPI_MODEL_CHAIN is missing. "
"Provide one or more comma-separated model IDs."
)
return Settings(
api_key=api_key,
base_url=base_url,
models=models,
)
def create_client(settings: Settings) -> OpenAI:
"""
Create an OpenAI-compatible client.
max_retries is set to zero because this application implements its own
retry and fallback policy. Leaving SDK-level retries enabled would create
nested retries and make latency, logs, and attempt counts harder to reason
about.
"""
return OpenAI(
api_key=settings.api_key,
base_url=settings.base_url,
timeout=settings.timeout_seconds,
max_retries=0,
)
def parse_retry_after(value: str | None) -> float | None:
"""
Parse Retry-After as either a number of seconds or an HTTP date.
Returns None when the header is missing or invalid.
"""
if not value:
return None
value = value.strip()
try:
return max(0.0, float(value))
except ValueError:
pass
try:
retry_time = parsedate_to_datetime(value)
if retry_time.tzinfo is None:
retry_time = retry_time.replace(tzinfo=timezone.utc)
now = datetime.now(timezone.utc)
return max(0.0, (retry_time - now).total_seconds())
except (TypeError, ValueError, OverflowError):
return None
def calculate_backoff(
attempt_index: int,
settings: Settings,
retry_after: float | None = None,
) -> float:
"""
Calculate a retry delay.
Retry-After takes precedence when the server supplies it. Otherwise,
exponential backoff with full jitter is used.
"""
if retry_after is not None:
return min(retry_after, settings.backoff_cap_seconds)
exponential_cap = min(
settings.backoff_cap_seconds,
settings.backoff_base_seconds * (2**attempt_index),
)
return random.uniform(0.0, exponential_cap)
def extract_text(response: Any) -> str:
"""
Validate and extract text from a Chat Completions response.
An HTTP 200 response is not automatically a usable application result.
The response still needs structural validation.
"""
choices = getattr(response, "choices", None)
if not choices:
raise InvalidModelResponse(
"The response did not contain any choices."
)
message = getattr(choices[0], "message", None)
if message is None:
raise InvalidModelResponse(
"The first choice did not contain a message."
)
content = getattr(message, "content", None)
if not isinstance(content, str) or not content.strip():
raise InvalidModelResponse(
"The response did not contain non-empty text."
)
return content.strip()
def generate_with_fallback(
client: OpenAI,
settings: Settings,
messages: list[dict[str, str]],
) -> str:
"""
Call each configured model in sequence until one returns a valid result.
Fail-fast errors stop the entire operation. Temporary failures are retried
on the current model before the client moves to the next model.
"""
failure_summary: list[str] = []
for model in settings.models:
for attempt_index in range(settings.attempts_per_model):
attempt_number = attempt_index + 1
started_at = time.perf_counter()
try:
response = client.chat.completions.create(
model=model,
messages=messages,
)
text = extract_text(response)
latency_ms = round(
(time.perf_counter() - started_at) * 1000
)
logger.info(
"request_succeeded model=%s attempt=%d latency_ms=%d",
model,
attempt_number,
latency_ms,
)
return text
except APIStatusError as exc:
status_code = exc.status_code
response_headers = getattr(exc.response, "headers", {})
request_id = response_headers.get("x-request-id", "-")
logger.warning(
"request_failed model=%s attempt=%d "
"status=%s request_id=%s",
model,
attempt_number,
status_code,
request_id,
)
if status_code in FAIL_FAST_STATUS_CODES:
raise RuntimeError(
"The request failed with a non-retryable HTTP "
f"status. model={model} status={status_code} "
f"request_id={request_id}"
) from exc
if status_code not in TRANSIENT_STATUS_CODES:
raise RuntimeError(
"The request failed with an unclassified HTTP "
f"status. model={model} status={status_code} "
f"request_id={request_id}"
) from exc
failure_summary.append(
f"{model}:HTTP_{status_code}"
)
if attempt_number < settings.attempts_per_model:
retry_after = parse_retry_after(
response_headers.get("retry-after")
)
delay = calculate_backoff(
attempt_index=attempt_index,
settings=settings,
retry_after=retry_after,
)
logger.info(
"retry_scheduled model=%s delay_seconds=%.2f",
model,
delay,
)
time.sleep(delay)
continue
logger.warning(
"model_attempts_exhausted model=%s; "
"moving_to_next_model=true",
model,
)
break
except (APITimeoutError, APIConnectionError) as exc:
error_type = type(exc).__name__
logger.warning(
"transport_error model=%s attempt=%d error_type=%s",
model,
attempt_number,
error_type,
)
failure_summary.append(
f"{model}:{error_type}"
)
if attempt_number < settings.attempts_per_model:
delay = calculate_backoff(
attempt_index=attempt_index,
settings=settings,
)
logger.info(
"retry_scheduled model=%s delay_seconds=%.2f",
model,
delay,
)
time.sleep(delay)
continue
logger.warning(
"model_attempts_exhausted model=%s; "
"moving_to_next_model=true",
model,
)
break
except InvalidModelResponse as exc:
logger.warning(
"invalid_response model=%s attempt=%d reason=%s",
model,
attempt_number,
str(exc),
)
failure_summary.append(
f"{model}:INVALID_RESPONSE"
)
if attempt_number < settings.attempts_per_model:
delay = calculate_backoff(
attempt_index=attempt_index,
settings=settings,
)
time.sleep(delay)
continue
break
summary = ", ".join(failure_summary) or "no failure details"
raise RuntimeError(
"Every configured model failed. "
f"Failure summary: {summary}"
)
def main() -> None:
settings = load_settings()
client = create_client(settings)
messages = [
{
"role": "system",
"content": (
"Answer clearly and do not invent information that is "
"not supported by the prompt."
),
},
{
"role": "user",
"content": (
"Explain the difference between retrying a request and "
"falling back to another model."
),
},
]
result = generate_with_fallback(
client=client,
settings=settings,
messages=messages,
)
print(result)
if __name__ == "__main__":
main()Run it:
python fallback.pyAssume the model chain contains:
primary-model-id,fallback-model-id
The client follows this sequence:
- Send the request to
primary-model-id. - Return immediately when the response is valid.
- Stop immediately when the request has a fail-fast error.
- Retry temporary errors using backoff and jitter.
- Respect
Retry-Afterwhen the response supplies it. - Move to
fallback-model-idafter the primary model exhausts its attempts. - Raise a final error when every configured model fails.
The distinction between a retry and a fallback is important:
- A retry sends the request to the same model again.
- A fallback sends the request to a different configured model.
Retries address temporary failures. Fallbacks reduce dependence on one model or upstream route.
The OpenAI Python client can perform some retries internally.
This example sets:
max_retries=0because the application already implements:
- explicit status classification;
- retry limits;
Retry-Afterhandling;- exponential backoff;
- jitter;
- model fallback;
- structured logging.
Allowing both layers to retry would create nested behavior.
For example, three application attempts combined with two hidden SDK retries could produce more network calls than the logs suggest. It would also make total latency and rate-limit behavior harder to predict.
A production system should generally have one clearly owned retry policy.
A 429 Too Many Requests response means the current request rate has exceeded an applicable limit.
Immediately resending every failed request can make the condition worse:
Rate limit
→ Immediate retry
→ More requests
→ More rate limits
→ More retries
Exponential backoff increases the delay after repeated failures. Jitter prevents many workers from retrying at exactly the same moment.
The example uses full jitter:
random.uniform(0.0, exponential_cap)This distributes retries across a time window rather than synchronizing them.
Consider an invalid credential:
Request
→ 401 Unauthorized
→ Retry primary model
→ 401 Unauthorized
→ Fallback model
→ 401 Unauthorized
Changing the model does not repair the credential.
The correct response is to stop, surface the configuration error, and alert the operator.
The same principle generally applies to:
- malformed payloads;
- unsupported parameters;
- invalid model IDs;
- missing permissions;
- requests that exceed a known format constraint.
The precise status code can vary, so production code should be tested against the actual endpoint behavior.
A successful HTTP status does not prove that the application received a usable result.
A response can still be incomplete or structurally unexpected. Examples include:
- an empty
choicesarray; - a missing message;
- empty text;
- malformed structured output;
- incomplete tool-call arguments;
- content that fails a task-specific acceptance rule.
This example validates only the minimum text structure.
A real application should validate the result required by its workload. For example:
def validate_json_result(text: str) -> dict:
import json
result = json.loads(text)
required_fields = {"title", "summary", "confidence"}
if not required_fields.issubset(result):
raise InvalidModelResponse(
"The JSON result is missing required fields."
)
return resultFor important workloads, fallback should be semantic as well as network-based.
Do not wait for a production outage to test fallback behavior.
At minimum, test the following cases in a controlled environment.
Expected behavior:
401 or equivalent authentication error
→ No retry
→ No model fallback
→ Immediate operator-visible failure
Expected behavior:
Model-not-found or invalid-request response
→ No blind retry
→ Configuration error surfaced
Some APIs may return 400 rather than 404 for an invalid model ID. Test the actual response.
Expected behavior:
429
→ Respect Retry-After when present
→ Retry with bounded delay
→ Move to the next model after attempts are exhausted
Expected behavior:
500, 502, 503, or 504
→ Retry with backoff
→ Fall back after repeated failure
Expected behavior:
Timeout
→ Retry only when repeating the operation is safe
→ Fall back after repeated failure
Expected behavior:
HTTP success with missing or unusable content
→ Response validation fails
→ Retry or fallback according to policy
Retrying a text-generation request is usually different from retrying a request that can trigger an external action.
Consider a model that calls a payment, email, database, or deployment tool:
Model request
→ Tool executes successfully
→ Client loses the response
→ Client retries
→ Tool may execute a second time
For agentic or tool-calling workflows:
- assign an idempotency key to side-effecting operations;
- store tool execution state outside the model response;
- validate whether a tool call has already completed;
- separate model retries from tool retries;
- avoid replaying side effects automatically.
An API-level fallback policy does not by itself make agent actions safe.
This example is intentionally small. A production implementation may also need:
Temporarily stop sending traffic to a model after repeated failures.
A basic policy could open the circuit after five consecutive temporary failures and probe the model again after a cooldown period.
Run controlled health checks using inexpensive requests rather than waiting for user traffic to discover every failure.
Health checks should verify more than connectivity. Where relevant, validate:
- authentication;
- model availability;
- response structure;
- streaming;
- tool calling;
- structured output.
A retry policy should limit total work across the entire request.
For example, a request with three models and three attempts per model could create up to nine API calls. Define a maximum total attempt count or total deadline.
A thirty-second timeout per attempt does not mean the user waits only thirty seconds.
Multiple retries and fallback models can extend total latency substantially. Use an end-to-end deadline and stop when the remaining time is insufficient for another attempt.
At minimum, log:
- model ID;
- HTTP status;
- request ID when available;
- attempt number;
- latency;
- token usage;
- retry delay;
- fallback transition;
- final result status.
Do not log API keys, confidential prompts, personal data, or unrestricted model responses.
An alternative model is not automatically an equivalent replacement.
Before adding a fallback model, verify the capabilities your workload requires:
- context length;
- input modalities;
- output limits;
- streaming;
- tool calling;
- structured output;
- reasoning controls;
- safety behavior;
- region availability.
A fallback that returns HTTP 200 but cannot satisfy the task is not a reliable fallback.
This repository demonstrates an implementation pattern. It does not prove that:
- every model supports the same parameters;
- every provider returns identical response structures;
- streaming works without model-specific testing;
- tool calling works identically across models;
- one model is an acceptable semantic substitute for another;
- CometAPI is always faster than a direct provider API;
- CometAPI is always less expensive than direct access;
- every error code has the same meaning across all upstream providers.
Those questions require current documentation and workload-specific tests.
An OpenAI-compatible multi-model endpoint can be useful when an application needs:
- one client abstraction for several supported models;
- faster model evaluation;
- configurable model selection;
- centralized retry and fallback behavior;
- less provider-specific integration code.
A direct provider integration may still be preferable when the application depends on:
- newly released provider-native features;
- provider-specific beta parameters;
- specialized fine-tuning endpoints;
- the shortest possible request path;
- strict data-governance requirements;
- behavior not yet supported by a normalized interface.
A hybrid architecture is often reasonable: use a unified endpoint for normalized workloads and keep direct integrations for provider-specific capabilities.
- Never commit a real API key.
- Load credentials from a secret manager in production.
- Rotate exposed credentials immediately.
- Restrict repository Actions from printing environment variables.
- Review whether prompts, responses, or metadata contain confidential data.
- Confirm the current provider’s retention and data-handling policies.
- Add a
SECURITY.mdfile explaining how vulnerabilities should be reported.
Production-oriented Python example for multi-model retries, backoff, response validation, and fail-fast error handling with CometAPI.
cometapi
llm
llm-api
openai-compatible
model-fallback
retry
exponential-backoff
api-reliability
python
generative-ai
Only use topics that accurately describe the repository.