Python에서 AsyncOpenAI로 수천 개의 텍스트 분류하기

수천 개의 텍스트를 분류하려면 AsyncOpenAI를 사용하고, 모든 호출 주위에 asyncio.Semaphore를 두어 N개의 요청만 진행 중이도록 하고, 한 줄짜리 JSON 레이블을 요청하고, 실패한 항목만 다시 실행하세요. 아래 스크립트는 https://api.cheaptokens.dev/v1에 대해 정확히 그렇게 하며 마지막에 usage 필드에서 총 토큰을 출력합니다.

2026년 10월 31일 Andrej Karpathy(OpenAI 공동 창립자, 전 Tesla AI 디렉터)가 HackerTube를 소개했습니다: 그는 약 두 시간 만에 그것을 만들었고 Sonnet을 사용해 Hacker News 댓글을 YouTube 동영상을 추천하는 약 40K 개로 좁혔습니다. 그의 글에는 프롬프트, 배치 크기 또는 동시성이 나와 있지 않으므로, 다음은 이런 작업의 표준 패턴이며 그의 레시피가 아닙니다.

스크립트#

5,000개의 지원 티켓이 있고 각각을 bug, question 또는 praise로 태그하고 싶다고 가정해 봅시다. 전체 내용은 다음과 같습니다. pip install openai로 SDK를 설치하고(Python 3.10+; python -c "import openai; print(openai.__version__)"는 현재 버전을 보여줍니다) 먼저 키를 환경 변수에 넣으세요.

import asyncio
import json
import os

from openai import AsyncOpenAI

client = AsyncOpenAI(
    base_url='https://api.cheaptokens.dev/v1',
    api_key=os.environ['CHEAPTOKENS_API_KEY'],
    max_retries=2,
    timeout=60,
)
MODEL = 'deepseek-ai/DeepSeek-V4-Flash-0731'
LABELS = {'bug', 'question', 'praise'}
SYSTEM = (
    'Classify the text as bug, question or praise. '
    'Reply with JSON only: {"label": "<bug|question|praise>"}'
)


async def classify(text, sem, usage):
    async with sem:
        r = await client.chat.completions.create(
            model=MODEL,
            temperature=0,
            max_tokens=100,
            messages=[
                {'role': 'system', 'content': SYSTEM},
                {'role': 'user', 'content': text},
            ],
        )
    usage['in'] += r.usage.prompt_tokens
    usage['out'] += r.usage.completion_tokens
    raw = r.choices[0].message.content
    label = json.loads(raw[raw.find('{'):raw.rfind('}') + 1])['label']
    if label not in LABELS:
        raise ValueError(f'unexpected label: {label!r}')
    return label


async def run(texts, concurrency=5, rounds=3):
    sem = asyncio.Semaphore(concurrency)
    usage = {'in': 0, 'out': 0}
    labels, pending = {}, list(range(len(texts)))
    for _ in range(rounds):
        if not pending:
            break
        results = await asyncio.gather(
            *(classify(texts[i], sem, usage) for i in pending),
            return_exceptions=True,
        )
        failed = []
        for i, res in zip(pending, results):
            if isinstance(res, Exception):
                failed.append(i)
            else:
                labels[i] = res
        pending = failed
    return labels, pending, usage


texts = [
    'The export button crashes the app on Android 15.',
    'Does the Pro plan include API access?',
    'Love the new dark theme, thank you!',
]
labels, failed, usage = asyncio.run(run(texts))
print(labels)
print('failed:', failed)
print('tokens in:', usage['in'], 'out:', usage['out'])

결과로 labels(텍스트 인덱스 → 레이블), failed(유효한 레이블을 생성하지 못한 인덱스, 모두 통과했을 때는 빈 목록) 및 토큰 합계를 받습니다. 모델 id는 전체 deepseek-ai/DeepSeek-V4-Flash-0731입니다. 다른 모델의 id는 CheapTokens 문서에서 가져오세요. 짧은 이름은 제공자마다 다르기 때문입니다.

파이프라인을 순서대로 배치하여 머릿속에 구조가 잡혔는지 확인하세요:

왜 세마포어이고 한 번에 5,000개의 요청이 아닌가#

asyncio.gather는 주어진 모든 것을 기꺼이 시작합니다. 제한 없이 수천 개의 코루틴을 넘기면 모두 동시에 시작되며, 이는 429 오류로 이어지기 쉬운 요청 폭주입니다. 세마포어는 개찰구입니다: concurrency=5는 어느 순간에도 최대 다섯 개의 호출이 async with sem 안에 있음을 의미하고 나머지는 차례를 기다립니다. Python은 여전히 모든 작업을 생성하지만 다섯 개만 API와 통신합니다.

concurrency는 어떤 숫자여야 할까요? 위의 5처럼 낮게 시작하고 오류가 없으면 올리세요. 복사할 상수가 아니라 작업마다 조정하는 설정으로 취급하세요.

재시도: 두 계층, 두 가지 역할#

첫 번째 계층은 SDK에 내장되어 있습니다. openai-python README(2026년 11월 1일 확인)에 따르면 연결 오류, 408, 409, 429 및 5xx 응답은 기본적으로 짧은 지수 백오프와 함께 두 번 재시도됩니다. 스크립트의 max_retries=2는 이를 가시화할 뿐입니다. client.with_options(max_retries=5)로 더 높이거나 0로 끌 수 있습니다. 기본 타임아웃은 10분이며 타임아웃은 APITimeoutError를 발생시키므로 스크립트는 짧은 레이블을 위해 timeout=60를 설정합니다. 자세한 내용은 OpenAI Python SDK 타임아웃 및 재시도에, 그 뒤의 백오프 개념은 AI API의 429 Too Many Requests에 있습니다.

두 번째 계층은 여러분의 것이며 SDK가 볼 수 없는 것을 다룹니다: 호출은 성공했지만 모델이 JSON 없이 산문으로 답했거나(I can't classify this) complaint라는 레이블을 만들어낸 경우입니다. {...} 주변의 텍스트는 중요하지 않습니다. 코드가 답변에서 JSON을 잘라내기 때문입니다. 그러면 json.loads 또는 LABELS 검사가 예외를 발생시키고, gather(..., return_exceptions=True)는 오류를 배치를 죽이는 대신 값으로 유지하며, 항목은 다음 라운드로 갑니다. rounds번 시도 후에도 failed에 남아 있는 것은 직접 검사할 몫입니다. 5,000개 중 하나의 잘못된 답변은 전체 작업이 아니라 추가 요청 하나의 비용만 발생시킵니다.

스크립트는 프롬프트에서 JSON을 요청하고 코드에서 검증합니다. response_format로 형식을 더 강하게 밀어붙이고 싶다면 LLM API에서 유효한 JSON 얻기를 참조하세요.

작업 비용#

토큰이 유일한 미터이므로 스크립트는 usage를 합산하고 비용이 따라옵니다. 일반적인 경우를 가정해 봅시다: 짧은 텍스트, 짧은 레이블. 10,000개의 텍스트가 각각 약 300 입력 토큰(텍스트 + 지시문)이고 JSON 레이블에 약 20 출력 토큰이라고 가정합니다:

10,000 × 300 = 3,000,000 입력 토큰; 10,000 × 20 = 200,000 출력 토큰.

DeepSeek-V4-Flash에서는 이 작업이 $0.065, GLM-5.3-Flash에서는 $0.068, MiniMax-M2.7에서는 $0.18입니다. 표는 OpenRouter 옆에 모든 모델에 대한 동일한 작업을 보여줍니다. 입력과 출력을 합산하면 총액이 OpenRouter보다 저렴합니다.

계산기의 슬라이더를 자신의 숫자로 옮기세요. in 및 out 필드는 한 요청에 대한 것이고 runs는 보내는 요청 수입니다.

여기서는 입력 측이 지배적입니다: 텍스트가 들어가고 한 단어가 나옵니다. 그래서 저렴한 입력과 380K 컨텍스트 창을 가진 모델, 예를 들어 어느 시간에도 동일한 가격의 DeepSeek-V4-Flash가 대량 필터링에 적합합니다. 짧은 레이블의 경우 GLM-5.3-Flash와 MiniMax-M2.7이 같은 스크립트의 대안이며 MODEL만 변경하면 됩니다. 이런 청구서가 어떻게 커지는지 궁금하다면 AI API가 청구하는 항목에 그림이 있습니다.

대규모 실행 전에#

먼저 20개의 텍스트를 실행하고 failed와 토큰 합계를 확인하세요. 20개의 텍스트가 약 6,000 입력 토큰을 사용한다면 5,000개는 약 1,500,000을 사용할 것이며, 커밋하기 전에 곱해볼 수 있습니다.

작동하지 않을 때#

• KeyError: 'CHEAPTOKENS_API_KEY': 이 터미널에서 변수가 설정되지 않았습니다. 내보내고 Python을 다시 시작하세요.
• 401 또는 404: 잘못된 키, 기본 URL 또는 모델 id입니다. API 401 또는 404 오류에서 각각을 안내합니다.
• 많은 항목에서 json.JSONDecodeError: 답변에 {...}이 없으며, 종종 max_tokens=100이 잘라냈기 때문입니다. r.choices[0].finish_reason를 출력하세요. length은 한도에 도달했음을 의미하므로 max_tokens를 올리세요.
• 많은 429와 느린 실행: concurrency를 낮추세요.

일반 모델 및 엔드포인트 세부 정보는 CheapTokens 문서와 가격 페이지에 있습니다.