호출 한도와 폴백

429는 비정상이 아니라 예상해야 할 응답이야

언젠가는 429를 만나. 잘못해서가 아니라 호출 한도가 있기 때문이야. 올바른 대응은 지터를 더한 지수 백오프고, 당황할 필요 없어. 쉬지 않고 다시 시도해 상황을 악화시키거나 사용자에게 오류를 그대로 보여 주는 게 잘못된 대응이야.

모델 사이의 폴백 체인

Pro가 호출 한도에 걸렸다고 Flash도 함께 막히는 경우는 드물어. Flash가 막혀도 Flash-Lite는 살아 있을 수 있어. 호출 한도 실패가 나면 비용 사다리를 한 단계씩 내려가는 체인을 만들어.

제공자 사이의 폴백 체인

다음 계층은 Gemini에서 OpenAI, 다시 Claude로 넘어가는 제공자 간 폴백이야. 흐름은 같아. 시도하고, 실패하면 전환하고, 기록하고, 사용자에게 보여 줘.

안전 차단은 재시도할 일이 아니야

finishReason: SAFETY는 모델이 출력을 걸렀다는 뜻이야. 같은 프롬프트를 다시 보내면 같은 결과가 나와. 표현을 바꾸거나, 쓰임새가 정당할 때만 안전 기준을 완화하거나, 사용자에게 자연스러운 안내를 보여 줘.

Code

백오프를 적용한 복원력 있는 단일 호출·python

import asyncio, random
from google.genai import errors

async def resilient_generate(client, model, contents, max_retries=4):
    for attempt in range(max_retries):
        try:
            return await client.aio.models.generate_content(
                model=model, contents=contents,
            )
        except errors.ClientError as e:
            if e.code == 429:
                # Rate limit — exponential backoff with jitter
                base = 2 ** attempt
                sleep = base + random.uniform(0, base * 0.5)
                print(f'[429 on {model}] sleeping {sleep:.1f}s')
                await asyncio.sleep(sleep)
                continue
            # 4xx other than 429 — don't retry
            raise
        except errors.ServerError:
            await asyncio.sleep(min(2 ** attempt, 30))
            continue
    raise RuntimeError(f'Gave up after {max_retries} retries')

여러 모델의 폴백 체인·python

MODEL_CHAIN = [
    'gemini-2.5-pro',
    'gemini-2.5-flash',
    'gemini-2.5-flash-lite',
]

async def generate_with_fallback(client, contents):
    last_error = None
    for model in MODEL_CHAIN:
        try:
            return await client.aio.models.generate_content(
                model=model, contents=contents,
            ), model
        except errors.ClientError as e:
            if e.code == 429:
                last_error = e
                continue  # Try next model
            raise  # Other 4xx — don't shop around
        except errors.ServerError as e:
            last_error = e
            continue  # 5xx — try next

    raise RuntimeError(f'All models exhausted. Last error: {last_error}')

SAFETY를 명시적으로 처리하기·python

candidate = response.candidates[0]
if candidate.finish_reason.name == 'SAFETY':
    # Surface what got blocked, for logging
    for rating in candidate.safety_ratings:
        if rating.probability.name in ('HIGH', 'MEDIUM'):
            log.warning(f'Safety block: {rating.category.name} -> {rating.probability.name}')

    # Don't retry. Surface gracefully.
    return "Sorry, I can't help with that request."

제공자 간 폴백(개요)·python

ADAPTER_CHAIN: list[ModelAdapter] = [
    GeminiAdapter(api_key=GEMINI_KEY),
    OpenAIAdapter(api_key=OPENAI_KEY),     # if you have one
    OllamaAdapter(model='llama3.1:70b'),   # local fallback
]

async def stream_with_provider_fallback(messages):
    for adapter in ADAPTER_CHAIN:
        try:
            async for chunk in adapter.generate_stream(messages):
                yield chunk
            return
        except Exception as e:
            # Toast it visibly — never silent
            print(f'[{adapter.name()} failed: {e}. Trying next provider.]')
            continue
    raise RuntimeError('All providers exhausted')

Exercise

두 번째 코드 블록의 여러 모델 폴백 체인을 만들어. 할당량이 쌓이지 않은 새 프로젝트에서 Pro를 쉬지 않고 100번 호출해 호출 한도를 유도하고, 알맞은 기록과 함께 Flash, 그다음 Flash-Lite로 넘어가는지 확인해. 보너스로 모델별 대기 타이머를 넣어 한 번 호출 한도에 걸린 모델은 60초 동안 건너뛰게 해.