중계 API 안정성 실무: 시간 초과, 재시도, 속도 제한 큐 및 스트리밍 단절

오류를 먼저 세 가지로 분류합니다

안정성 문제의 첫 단계는 재시도 횟수를 늘리는 것이 아니라, 무엇을 재시도해야 하는지 아는 것입니다. 처리 방식에 따라 일반적인 실패를 세 가지로 분류할 수 있습니다:

유형	전형적인 증상	처리 방식
일시적 복구 가능	429 속도 제한, 503 upstream_busy, 연결 재설정, 시간 초과	지수 백오프 후 재시도, 총 횟수 제한
요청 자체에 문제 있음	400 (입력 + max_tokens이 100,000을 초과하거나 요청 본문이 큼), 403 content_blocked	재시도하지 않고 요청을 수정하거나 사용자에게 직접 피드백
계정 상태 문제	401 키 무효, 402 no_credit	재시도하지 않고 담당자에게 알림, 충전 또는 키 교체

이 표는 코드 주석으로 작성하는 것이 좋습니다. 흔한 사고 사례는 다음과 같습니다: 잔액이 고갈된 후 인터페이스가 402를 안정적으로 반환하는데, 재시도 로직이 상태 코드를 구분하지 않아 각 요청이 5번씩 재시도합니다. 이로 인해 트래픽이 5배로 증폭되고 로그가 빨간색으로 가득 찹니다.

시간 초과를 계층적으로 설정해야 합니다

전체 시간 초과 하나만 설정하는 것은 충분하지 않습니다. 시간 초과를 세 층으로 나누어 고려하는 것을 권장합니다:

• 연결 시간 초과: TCP 및 TLS 연결을 수립하는 시간으로, 일반적으로 5~10초면 충분하며 연결 실패 시 빠르게 실패해야 합니다.
• 읽기 시간 초과: 응답 데이터를 기다리는 시간입니다. 비스트리밍 요청은 전체 텍스트 생성이 완료될 때까지 기다려야 하므로 가장 긴 출력 시간을 커버해야 합니다. 스트리밍 요청의 경우 '두 데이터 블록 사이의 최대 정적 시간'이 되며, 10~30초가 더 적절합니다.
• 비즈니스 전체 시간 초과: 비즈니스에서 허용할 수 있는 최대 대기 시간입니다. asyncio.wait_for 또는 게이트웨이 레이어를 통해 보장해야 하며, 모든 재시도를 포함해야 합니다.

출력이 길어질수록 비스트리밍 요청의 소요 시간이 길어집니다. max_tokens을 최대치인 32,000으로 설정하고 전체 시간 초과를 30초로만 설정하면 긴 텍스트 상황에서 시간 초과가 빈번하게 발생합니다. 더 안전한 방법은 긴 출력을 모두 스트리밍으로 처리하여 사용자에게 즉각적인 피드백을 제공하고 시간 초과 의미를 명확히 하는 것입니다. SDK의 timeout 파라미터는 숫자나 세부 구성을 전달할 수 있으며, 사용 중인 버전에 따라 문서를 참조하세요.

429 및 503 지수 백오프 재시도

백오프 핵심은 지연 시간 증가, 상한 설정, 랜덤 추가입니다. 랜덤이 없으면 재시도 시 서비스 부하가 다시 발생합니다. SDK 재시도를 끄고 429, 503, 연결 오류만 처리합니다:

复制import asyncio
import os
import random

from openai import AsyncOpenAI, APIConnectionError, APIStatusError, APITimeoutError

client = AsyncOpenAI(
    base_url="https://api.llmzhongzhuan.com/v1",
    api_key=os.environ["API_KEY"],
    timeout=60.0,
    max_retries=0,  # 重试由下面的函数统一控制
)

RETRY_STATUS = {429, 503}


async def chat_with_retry(messages, max_attempts=5, base_delay=1.0, cap=20.0):
    for attempt in range(1, max_attempts + 1):
        try:
            resp = await client.chat.completions.create(
                model="uncensored", messages=messages, max_tokens=800
            )
            return resp
        except APIStatusError as e:
            if e.status_code not in RETRY_STATUS or attempt == max_attempts:
                raise  # 400/401/402/403 重试没有意义
            reason = f"HTTP {e.status_code}"
        except (APIConnectionError, APITimeoutError) as e:
            if attempt == max_attempts:
                raise
            reason = type(e).__name__
        delay = min(cap, base_delay * 2 ** (attempt - 1))
        delay = random.uniform(0, delay)  # 全抖动，避免所有任务同时醒来
        print(f"第 {attempt} 次失败（{reason}），{delay:.1f}s 后重试")
        await asyncio.sleep(delay)

503은 서버가 일시적으로 바쁘다는 의미이므로 몇 초 후 재시도하는 것을 권장합니다. 위의 기준 지연 시간 1초, 상한 20초, 최대 5회로 이 상황을 충분히 커버할 수 있습니다. 429이 빈번하게 발생한다면 재시도 문제가 아니라 전송 측에서 속도 제어를 하지 않았다는 의미이므로 다음 섹션을 참조하세요.

분당 300회 속도 제한 하의 동시 요청 큐

각 키당 분당 300회 요청이므로 초당 평균 5회입니다. 배치 처리 작업은 여기서最容易 실수합니다. asyncio.gather로 수천 개의 요청을 한 번에 보내면, 처음 몇십 개가 한도액을 순식간에 소진하고 나머지는 모두 429을 받습니다.

신뢰할 수 있는 방법은 두 층 제어입니다. 세마포어로 '동시 진행' 수를 제한하여 로컬 연결과 메모리가 폭주하는 것을 방지하고, 타이머로 '출력 속도'를 제한해 요청을 고르게 분산합니다. 둘 다 필수적입니다: 세마포어만 있으면 단일 요청이 매우 빠를 때 분당 300회를 초과할 수 있으며, 타이머만 있으면 요청이 매우 느릴 때 진행 중인 요청이 계속 쌓입니다. 예시에서는 속도를 분당 270회로 설정해 다중 인스턴스, 재시도 및 시계 오차에 10% 여유를 두었습니다:

复制import asyncio
import os
import time

from openai import AsyncOpenAI

client = AsyncOpenAI(
    base_url="https://api.llmzhongzhuan.com/v1",
    api_key=os.environ["API_KEY"],
    timeout=60.0,
    max_retries=0,
)

MAX_IN_FLIGHT = 8          # 同时在途的请求数
PER_MINUTE = 270           # 留 10% 余量，低于每分钟 300 次的上限


class Pacer:
    """把请求的发出时刻均匀铺开：每 60/PER_MINUTE 秒放行一个。"""

    def __init__(self, per_minute):
        self.interval = 60.0 / per_minute
        self.next_at = 0.0
        self.lock = asyncio.Lock()

    async def wait(self):
        async with self.lock:
            now = time.monotonic()
            delay = self.next_at - now
            if delay > 0:
                await asyncio.sleep(delay)
            self.next_at = max(now, self.next_at) + self.interval


sem = asyncio.Semaphore(MAX_IN_FLIGHT)
pacer = Pacer(PER_MINUTE)


async def summarize(idx, text):
    async with sem:            # 控制并发
        await pacer.wait()     # 控制速率
        resp = await client.chat.completions.create(
            model="uncensored",
            messages=[{"role": "user", "content": f"用两句话概括：{text}"}],
            max_tokens=120,
        )
        return idx, resp.choices[0].message.content


async def main():
    texts = [f"第 {i} 条工单的正文……" for i in range(1000)]
    tasks = [summarize(i, t) for i, t in enumerate(texts)]
    done = 0
    for coro in asyncio.as_completed(tasks):
        idx, out = await coro
        done += 1
        if done % 100 == 0:
            print(f"已完成 {done} 条")


asyncio.run(main())

여러 프로세스가 키를 공유하면 메모리 기반 제한이 부족합니다. Redis 기반 토큰 버킷이나 전용 큐 서비스를 통해 요청을 통합해야 합니다.

스트리밍 출력 중단 시 대응 방법

스트리밍 연결은 일반 요청보다 더 취약합니다. 지속 시간이 길고 통과하는 네트워크 장치가 많으며, 어떤 구성 요소의 연결 시간 초과라도 연결을 끊을 수 있습니다. 처리 원칙은 '이미 수신한 내용은 자산'이며, 중단 때문에 모든 내용을 버리지 않아야 합니다.

아래 구현은 수신된 조각을 누적하고 연결 및 시간 초과 예외를 포착한 후, 기존 텍스트를 assistant 메시지로 요청해 모델이 이어서 생성하도록 합니다. 이는 최선의 노력을 다하는 복구 방식이며, 이어서 생성된 결과의 연결이 완벽하지 않을 수 있으므로 긴 텍스트와 대화에는 적합하지만 JSON 등 엄격한 구조의 출력에는 적합하지 않습니다. 구조화된 출력이 중단되면 전체를 다시 요청하는 것이 더 안전합니다. 스트리밍 종료 시 자동으로 usage 분할 데이터가 포함되므로 코드에서 이를 함께 추출했습니다:

复制import asyncio
import os

from openai import AsyncOpenAI, APIConnectionError, APITimeoutError

client = AsyncOpenAI(
    base_url="https://api.llmzhongzhuan.com/v1",
    api_key=os.environ["API_KEY"],
    timeout=30.0,   # 流式场景下，这是相邻数据块之间允许的最长静默
    max_retries=0,
)


async def stream_once(messages):
    parts, usage = [], None
    stream = await client.chat.completions.create(
        model="uncensored", messages=messages, max_tokens=600, stream=True
    )
    try:
        async for chunk in stream:
            if chunk.usage:                      # 最后一个分片携带 usage
                usage = chunk.usage
            if chunk.choices and chunk.choices[0].delta.content:
                parts.append(chunk.choices[0].delta.content)
    except (APIConnectionError, APITimeoutError):
        return "".join(parts), None, False       # 中断：返回已收到的部分
    return "".join(parts), usage, True


async def stream_with_resume(question, max_resume=2):
    messages = [{"role": "user", "content": question}]
    text = ""
    for _ in range(max_resume + 1):
        part, usage, finished = await stream_once(messages)
        text += part
        if finished:
            return text, usage
        # 把已生成部分作为 assistant 消息，请模型接着写
        messages = [
            {"role": "user", "content": question},
            {"role": "assistant", "content": text},
            {"role": "user", "content": "请从上一条回复的结尾处继续，不要重复已写的内容。"},
        ]
    return text, None


if __name__ == "__main__":
    out, usage = asyncio.run(stream_with_resume("用三段话说明为什么日志要带 request id。"))
    print(out)
    print("usage:", usage)

이어서 생성할 때 이미 전송된 텍스트도 입력 토큰으로 계산되므로, 매번 이어서 생성할 때마다 약간의 추가 비용이 발생합니다. max_resume 횟수를 제한하는 것이 필요합니다.

다운그레이드, 서킷 브레이커 및 출시 전 자가 점검

재시도는 일시적 오류를 해결합니다. 오류가 몇 분간 지속된다면 계속 재시도하면 요청만 쌓입니다. 재시도 위에 회로 차단기를 추가하는 것을 권장합니다. 예를 들어 1분 동안 10회 연속 실패하면 일정 시간 동안 외부 요청 전송을 일시 중단하고, 캐시된 결과 반환, 사용자 안내 또는 큐 후처리 등 대체 논드로 처리합니다. 대기 시간이 끝나면 소량의 탐지 요청만 통과시키고, 성공하면 전체 요청을 복원합니다.

대체 처리도 미리 명확히 정의해야 합니다. 사용자 대상 채팅 기능은 친절한 바쁜 상태 메시지를 반환할 수 있으며, 오프라인 배치 작업은 실패 항목을 실패 테이블에 기록한 후 전체 실행 완료 후 일괄 재실행해야 합니다. 메인 프로세스에서 무한 대기하는 방식이 아닙니다. 어떤 경우든 실패가 조용히 사라지지 않도록, request id가 포함된 로그를 최소한 한 줄 이상 남기는 것을 보장해야 합니다.

마지막으로 출시 전 자가 점검 체크리스트를 제공합니다: SDK의 암묵적 재시도를 비활성화하고 자체 재시도만 남겼는지, 400/401/402/403에 대해 즉시 포기했는지, 전체 시간 초과에 모든 재시도가 포함되었는지, 배치 처리에 속도 제어가 있고 일괄 동시 요청이 아닌지, 스트리밍 중단 처리가 되었는지, usage가 저장되었는지, 잔액 알림이 설정되었는지 확인하세요. 항목별 통과 후 압력 테스트를 진행하세요.

usage를 활용한 모니터링 및 대조

각 응답에는 usage가 포함되며, 스트리밍의 경우 마지막 분할 데이터에 나타납니다. 기능 이름과 함께 이를 저장하는 것은 비용이 가장 저렴한 모니터링 방법입니다. 본 사이트의 입력 가격은 백만 토큰당 0.25달러, 출력은 1.00달러이므로 로컬에서 각 요청의 비용을 직접 추정할 수 있습니다:

复制import csv
import time

LOG = "usage_log.csv"


def record_usage(feature, resp):
    u = resp.usage
    cost = u.prompt_tokens * 0.25 / 1_000_000 + u.completion_tokens * 1.00 / 1_000_000
    with open(LOG, "a", newline="", encoding="utf-8") as f:
        csv.writer(f).writerow(
            [int(time.time()), feature, u.prompt_tokens, u.completion_tokens, f"{cost:.6f}"]
        )

데이터를 저장한 후에는 다음 세 가지 지표를 매일 확인하는 것을 권장합니다: 각 기능의 입력 토큰 중앙값으로 프롬프트가 서서히 확장되는지 파악하고, 출력 토큰 비율을 확인합니다. 출력 단가는 입력의 4배이므로 일반적으로 청구서의 주요 부분을 차지합니다. 그리고 429 및 503 비율을 확인하여 비율이 상승하면 전송 속도나 재시도 전략을 조정해야 합니다. 잔액 알림을 추가하여 402 오류 발생 전에 담당자에게 충전 알림을 보내세요. 더 많은 연동 관련 세부 사항은 프레임워크 구성을 참조하고, 자주 묻는 질문은 자주 묻는 질문에 정리되어 있습니다.