AI API 타임아웃: 원인, 재시도 패턴, 폴백 설계

# 문제: 큰 입력 + 높은 max_tokens = 느린 응답
response = client.chat.completions.create(
    model="gpt-4o",
    messages=very_long_messages,  # 100K+ tokens
    max_tokens=4096  # 긴 출력 요청
)

# 해결: 입력 줄이기 또는 출력 제한
response = client.chat.completions.create(
    model="gpt-4o",
    messages=trimmed_messages,  # 50K 토큰으로 축소
    max_tokens=1024  # 짧은 출력
)

# 동기 — 전체 응답을 기다리는 동안 클라이언트가 타임아웃될 수 있음
response = client.chat.completions.create(
    model="gpt-4o",
    messages=messages,
    max_tokens=2048
)

# 스트리밍 — 첫 번째 토큰이 더 빨리 도착하여 연결을 유지
stream = client.chat.completions.create(
    model="gpt-4o",
    messages=messages,
    max_tokens=2048,
    stream=True
)

for chunk in stream:
    if chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="")

import asyncio
import httpx

async def generate_video_async(client, prompt: str, timeout: int = 600):
    """영상 생성 작업을 제출하고 완료될 때까지 폴링한다."""

    # 1단계: 작업 제출
    submit_response = await client.post(
        "/v1/video/generations",
        json={"model": "veo-3-fast", "prompt": prompt}
    )
    job_id = submit_response.json()["id"]

    # 2단계: 완료까지 폴링
    for _ in range(timeout // 5):  # 5초마다 확인
        status_response = await client.get(f"/v1/video/generations/{job_id}")
        status = status_response.json()

        if status["status"] == "completed":
            return status["result"]
        elif status["status"] == "failed":
            raise RuntimeError(f"Generation failed: {status.get('error')}")

        await asyncio.sleep(5)

    raise TimeoutError(f"Video generation did not complete within {timeout}s")

Status: queued → position 42
Status: queued → position 15
Status: processing → estimated 90s remaining
Status: completed → download URL available

import asyncio
import random

async def retry_with_timeout_awareness(
    make_request,
    max_retries: int = 3,
    base_timeout: float = 30.0
):
    """타임아웃 유형을 인식하는 재시도."""
    for attempt in range(max_retries):
        try:
            return await asyncio.wait_for(
                make_request(),
                timeout=base_timeout * (1.5 ** attempt)  # 재시도마다 타임아웃 증가
            )
        except asyncio.TimeoutError:
            if attempt == max_retries - 1:
                raise

            # 썬더링 허드 방지를 위한 지터 추가
            delay = min(30, (2 ** attempt) + random.uniform(0, 1))
            await asyncio.sleep(delay)

        except Exception as e:
            # 타임아웃이 아닌 에러: 자동 재시도하지 않음
            if "429" in str(e):
                delay = min(60, (2 ** attempt) + random.uniform(0, 2))
                await asyncio.sleep(delay)
            else:
                raise

async def call_with_fallback(messages, client, primary_model, fallback_model):
    """프라이머리 모델 시도; 타임아웃 시 더 빠른 모델로 폴백."""
    try:
        return await asyncio.wait_for(
            client.chat.completions.create(
                model=primary_model,
                messages=messages
            ),
            timeout=30.0
        )
    except asyncio.TimeoutError:
        # 더 빠른(아마도 더 작은) 모델로 폴백
        return await client.chat.completions.create(
            model=fallback_model,
            messages=messages
        )

from openai import OpenAI

client = OpenAI(
    api_key="your-evolink-key",
    base_url="https://api.evolink.ai/v1"
)

# Smart Router가 프로바이더 선택과 폴백을 처리
response = client.chat.completions.create(
    model="evolink/auto",
    messages=messages
)