Резервные поставщики и автоматические размыкатели

Риск использования одного поставщика

Определение каскада поставщиков

from dataclasses import dataclass
from typing import Optional

@dataclass
class Provider:
    name: str
    base_url: Optional[str]
    api_key_env: str
    model: str
    priority: int  # lower = higher priority

CASCADE = [
    Provider('openai',    None,                              'OPENAI_API_KEY',    'gpt-4o',              1),
    Provider('anthropic', 'https://api.anthropic.com/v1',   'ANTHROPIC_API_KEY', 'claude-3-5-sonnet',   2),
    Provider('local',     'http://localhost:8000/v1',        'LOCAL_KEY',         'llama-3.1-8b-inst',   3),
]

Реализация цикла переключения

import openai
import os

TRANSIENT_ERRORS = (openai.APITimeoutError, openai.InternalServerError, openai.APIConnectionError)

async def call_with_fallback(messages: list, **kwargs) -> str:
    for provider in CASCADE:
        try:
            client = openai.AsyncOpenAI(
                api_key=os.environ[provider.api_key_env],
                base_url=provider.base_url
            )
            resp = await client.chat.completions.create(
                model=provider.model,
                messages=messages,
                timeout=10.0,
                **kwargs
            )
            return resp.choices[0].message.content
        except TRANSIENT_ERRORS as e:
            print(f'Provider {provider.name} failed: {e}, trying next...')
    raise RuntimeError('All providers failed')

Что такое автоматический размыкатель цепи

# Circuit breaker state machine:
#
# CLOSED --> (failure_count >= threshold) --> OPEN
#    ^                                          |
#    |     (test_request succeeds)              | (timeout expires)
#    +------------ HALF_OPEN <-----------------+
#
# In OPEN state: immediately return fallback/error
# In HALF_OPEN: allow one request through to test recovery
# In CLOSED: normal operation, count failures

Реализация автоматического размыкателя цепи

import time
from enum import Enum

class State(Enum):
    CLOSED = 'closed'
    OPEN = 'open'
    HALF_OPEN = 'half_open'

class CircuitBreaker:
    def __init__(self, failure_threshold=5, reset_timeout=60):
        self.state = State.CLOSED
        self.failure_count = 0
        self.failure_threshold = failure_threshold
        self.reset_timeout = reset_timeout
        self.opened_at = None

    def record_success(self):
        self.failure_count = 0
        self.state = State.CLOSED

    def record_failure(self):
        self.failure_count += 1
        if self.failure_count >= self.failure_threshold:
            self.state = State.OPEN
            self.opened_at = time.time()

    def can_attempt(self) -> bool:
        if self.state == State.CLOSED:
            return True
        if self.state == State.OPEN:
            if time.time() - self.opened_at > self.reset_timeout:
                self.state = State.HALF_OPEN
                return True  # allow one probe
            return False
        return True  # HALF_OPEN: allow probe

Интеграция автоматических размыкателей цепи с поставщиками

breakers = {p.name: CircuitBreaker(failure_threshold=5, reset_timeout=60) for p in CASCADE}

async def call_with_circuit_breaker(messages: list) -> str:
    for provider in CASCADE:
        breaker = breakers[provider.name]
        if not breaker.can_attempt():
            continue  # skip this provider, circuit is open
        try:
            result = await call_provider(provider, messages)
            breaker.record_success()
            return result
        except TRANSIENT_ERRORS as e:
            breaker.record_failure()
            print(f'{provider.name} failed ({breaker.failure_count}/{breaker.failure_threshold})')
    raise RuntimeError('All providers exhausted')

Обнаружение медленных вызовов как сбоев

async def call_provider(provider: Provider, messages: list) -> str:
    client = openai.AsyncOpenAI(
        api_key=os.environ[provider.api_key_env],
        base_url=provider.base_url
    )
    try:
        resp = await asyncio.wait_for(
            client.chat.completions.create(model=provider.model, messages=messages),
            timeout=10.0  # fail fast, let circuit breaker count it
        )
        return resp.choices[0].message.content
    except asyncio.TimeoutError:
        raise openai.APITimeoutError('Provider timed out')

Панель состояния поставщиков

from fastapi import FastAPI

app = FastAPI()

@app.get('/health/providers')
def provider_health():
    return {
        name: {
            'state': cb.state.value,
            'failure_count': cb.failure_count,
            'seconds_until_reset': (
                max(0, cb.reset_timeout - (time.time() - cb.opened_at))
                if cb.state == State.OPEN else None
            )
        }
        for name, cb in breakers.items()
    }

Согласование выходных данных поставщиков

def adapt_messages_for_provider(provider: Provider, messages: list) -> list:
    if provider.name == 'anthropic':
        # Claude prefers explicit task descriptions
        system = next((m['content'] for m in messages if m['role'] == 'system'), '')
        if 'JSON' not in system:
            messages = [{'role': 'system', 'content': system + ' Respond in JSON.'}] + [
                m for m in messages if m['role'] != 'system'
            ]
    return messages

Тестирование поведения при переключении

import pytest
from unittest.mock import AsyncMock, patch

@pytest.mark.asyncio
async def test_fallback_on_primary_timeout():
    # Primary provider times out
    with patch('your_module.call_provider', side_effect=[
        openai.APITimeoutError('Timeout'),  # primary fails
        'Claude response'                   # fallback succeeds
    ]):
        result = await call_with_circuit_breaker([{'role': 'user', 'content': 'Hello'}])
    assert result == 'Claude response'

Факторы стоимости каскада поставщиков

# Approximate costs per 1M tokens (2026):
PROVIDER_COSTS = {
    'openai/gpt-4o':          {'input': 2.50, 'output': 10.00},
    'anthropic/claude-3.5-sonnet': {'input': 3.00, 'output': 15.00},
    'openai/gpt-4o-mini':     {'input': 0.15, 'output': 0.60},
    'local/llama-3.1-8b':     {'input': 0.00, 'output': 0.00},  # infra cost only
}

# If fallback adds $0.50/day and a Tier 2 upgrade costs $100/month:
# Tier 2 pays off if you use fallback > 200 requests/day

Быстрая проверка

Итоги урока