Gemini API 동시성·비동기 병렬 호출 완전 가이드 (2026)

서문: 왜 병렬화가 필요한가

Gemini API를 순차적으로 호출하면 100개 요청은 100번의 왕복 시간을 그대로 누적한다. 각 요청이 1초 걸린다면 100초. 사용자가 기다릴 수 있는 시간을 한참 넘어선다.

비유하자면 마트 계산대 한 곳에서 100명이 한 명씩 계산하는 상황이다. 계산대(동시 요청)를 늘리면 전체 시간이 줄어든다. 이 글은 Gemini API에서 "계산대 늘리기"를 정확히 구현하는 네 가지 패턴을 다룬다.

이 글의 내용은 공식 레퍼런스 노트북 google-gemini/cookbook/quickstarts/Asynchronous_requests.ipynb를 기준으로 작성되었다. 사용자가 원래 요청한 google-gemma/cookbook/apps/concurrent 경로는 공개 레포에서 확인되지 않아 공식 등가 자료로 대체했다.

먼저 알아야 할 기본 사실

설치:

pip install -U google-genai
pip install aiohttp genai-processors  # 추가 패턴용

패턴 1: 기본 비동기 병렬 호출

asyncio.as_completed()를 사용하면 완료된 요청부터 순서대로 처리할 수 있다.

import asyncio
from google import genai

client = genai.Client(api_key="YOUR_API_KEY")

async def generate(prompt: str) -> str:
    response = await client.aio.models.generate_content(
        model="gemini-2.0-flash",
        contents=prompt,
    )
    return response.text

async def main():
    prompts = [
        "Python asyncio를 한 문장으로 설명해주세요",
        "Gemini API 장점 3가지를 알려주세요",
        "Batch API는 언제 쓰나요?",
    ]
    tasks = [generate(p) for p in prompts]
    for coro in asyncio.as_completed(tasks):
        result = await coro
        print(result[:100])

asyncio.run(main())

await client.aio.models.generate_content()만 기억하면 된다. 동기 API와 거의 동일한 시그니처인데 .aio가 붙었다는 점만 다르다.

asyncio.gather()와 as_completed() 차이:

패턴 2: asyncio.Semaphore로 Rate Limit 안전 처리

무료 티어에서 Gemini 2.0 Flash는 분당 15개 요청(RPM)까지만 허용한다. 한도 없이 100개를 동시에 쏘면 429 Resource Exhausted 에러가 돌아온다.

asyncio.Semaphore는 동시에 실행되는 코루틴 수를 제한한다.

import asyncio
from google import genai

client = genai.Client(api_key="YOUR_API_KEY")
semaphore = asyncio.Semaphore(10)  # 동시에 최대 10개만 실행

async def safe_generate(prompt: str) -> str:
    async with semaphore:
        response = await client.aio.models.generate_content(
            model="gemini-2.0-flash",
            contents=prompt,
        )
        return response.text

async def batch_process(prompts: list[str]) -> list[str]:
    tasks = [safe_generate(p) for p in prompts]
    return await asyncio.gather(*tasks)

if __name__ == "__main__":
    prompts = [f"주제 {i}에 대해 한 문장 요약" for i in range(50)]
    results = asyncio.run(batch_process(prompts))
    print(f"처리 완료: {len(results)}건")

Semaphore 값은 API 한도(RPM/TPM)에 맞춰 조정한다.

RPM을 기준으로 Semaphore를 설정하되, 실제 응답 시간도 고려한다. 한 요청이 평균 2초 걸리면 Semaphore(10)로도 분당 300개까지 처리 가능하므로 여유가 있다.

패턴 3: aiohttp로 이미지·파일 병렬 처리

Gemini Vision으로 이미지 수십 장을 분석할 때, 다운로드 자체도 직렬이면 느리다. aiohttp로 HTTP 요청을 병렬화하면서 Gemini API도 병렬 호출하면 I/O와 LLM 처리가 겹친다.

import asyncio
import aiohttp
from google import genai

client = genai.Client(api_key="YOUR_API_KEY")

async def download_and_analyze(session, url: str) -> str:
    # 1단계: 이미지 다운로드 (I/O 병렬)
    async with session.get(url) as resp:
        image_bytes = await resp.read()

    # 2단계: Gemini Vision 분석 (API 병렬)
    response = await client.aio.models.generate_content(
        model="gemini-2.0-flash",
        contents=[
            "이 이미지를 한 문장으로 설명하세요.",
            {"mime_type": "image/png", "data": image_bytes},
        ],
    )
    return response.text

async def main(urls: list[str]) -> list[str]:
    async with aiohttp.ClientSession() as session:
        tasks = [download_and_analyze(session, u) for u in urls]
        return await asyncio.gather(*tasks)

if __name__ == "__main__":
    urls = [f"https://example.com/img_{i}.png" for i in range(50)]
    captions = asyncio.run(main(urls))

이미지 50장을 순차 처리하면 다운로드 + 분석이 직렬로 누적되지만, 병렬 처리는 네트워크 대기와 API 대기가 서로 겹쳐 전체 시간이 크게 줄어든다.

패턴 4: GenAI Processors로 선언적 파이프라인

GenAI Processors는 구글이 공개한 공식 고수준 라이브러리다. 프로세서들을 연산자로 조합해 파이프라인을 만든다.

# 순차 체이닝: +
pipeline = input_processor + gemini_model + output_processor

# 병렬 분기: //
# 같은 입력을 두 프로세서에 동시에 흘려보냄
parallel = processor_a // processor_b

// 연산자 한 줄로 동일한 입력을 서로 다른 프로세서에 동시 전달한다. Real-Time Live 에이전트처럼 오디오·텍스트·함수 호출을 같은 틱에 처리해야 하는 경우에 특히 유용하다.

공식 소개: Google Developers Blog — GenAI Processors

성능·비용 비교

N개 요청을 처리할 때의 지표 비교다.

핵심 트레이드오프:

• 비동기 병렬: 속도를 얻지만 비용은 총량 동일. Rate Limit 내에서만 유효.
• Batch API: 비용을 반으로 낮추지만 지연을 감수해야 한다.

선택 기준: 내 상황엔 뭐가 맞나

실전 시나리오 5가지

1. 대용량 문서 병렬 분류

고객 지원 티켓 500건을 카테고리별로 분류. asyncio.gather + Semaphore(10)으로 분당 처리량을 RPM 한도 안에 맞춘다.

2. 멀티미디어 콘텐츠 파이프라인

SNS 콘텐츠 제작용 이미지 50장을 병렬 다운로드 + Vision 분석 + 캡션 생성. aiohttp + Gemini Vision 조합.

3. A/B 프롬프트 평가

같은 입력에 프롬프트 변형 5개를 동시 전송. as_completed로 가장 빠른 응답부터 UI에 표시.

4. 실시간 Live 에이전트

Gemini Live + 함수 호출 + 외부 센서 데이터를 동시에 처리. GenAI Processors의 // 연산자로 선언적 구성.

5. 야간 배치 번역

상품 설명 10,000건을 한국어로 번역. Batch API로 50% 비용 절감, 다음 날 아침 결과 수령.

주의사항

Rate Limit 이해

• RPM: Requests Per Minute. 분당 요청 수.
• TPM: Tokens Per Minute. 분당 처리 토큰 수. 긴 프롬프트 병렬 시 이쪽이 먼저 막힌다.
• 429 에러는 RPM 또는 TPM 중 하나라도 초과하면 발생한다.

공식 문서: Rate Limits

이벤트 루프 관리

• Jupyter/Colab: 이벤트 루프가 이미 돌고 있어 await를 직접 쓸 수 있다.
• 일반 스크립트: asyncio.run(main()) 필수.
• 중첩 이벤트 루프 필요 시: nest_asyncio.apply() 사용.

Batch API 제약

• 입력 포맷: JSONL 필수.
• 실시간 응답 불가, 최소 1분 지연.
• 공식 문서: Batch API

SDK 선택

구 google-generativeai는 비동기 네임스페이스가 없다. 반드시 google-genai 최신 버전을 설치한다.

pip install -U google-genai
python -c "from google import genai; print(genai.__version__)"

정리

Gemini API의 병렬화는 선택이 아니라 필수다. 50건 이상의 요청을 처리하는 서비스라면 비동기 패턴이 기본값이어야 한다.

네 가지 패턴을 요약하면:

1. asyncio.as_completed — 즉시 응답 스트리밍
2. asyncio.Semaphore — Rate Limit 안전 처리
3. aiohttp — 이미지·파일 병렬 I/O
4. GenAI Processors — 선언적 파이프라인

그리고 지연을 허용할 수 있는 대량 작업이라면 Batch API로 비용을 반으로 깎는다.

지금 실행 중인 Gemini 스크립트에 for prompt in prompts: 같은 순차 루프가 있는지 확인해보자. 패턴 1만 적용해도 체감 속도가 완전히 달라진다.

참고 자료

• google-gemini/cookbook — Asynchronous_requests.ipynb — 이 글의 기반이 된 공식 노트북
• google-gemini/cookbook — Gemini 공식 예제 저장소
• google-gemini/genai-processors — 공식 파이프라인 라이브러리
• Google Developers Blog — GenAI Processors — 공식 소개 글
• Gemini API — Batch API — 배치 처리 공식 문서
• Gemini API — Rate Limits — RPM/TPM 공식 한도표

위 자료와 본 가이드의 네 가지 패턴을 함께 활용하면 Gemini API의 성능과 비용을 동시에 최적화할 수 있습니다.