Telegram 봇에 Gemini API 통합: 개발자 가이드

aiogram 3.x를 사용한 Telegram 봇에 Gemini API 비동기 통합: 실전 가이드

Telegram 봇이 Gemini API의 429 오류로 갑자기 응답하지 않게 되면, 단순한 버그가 아니라 기술 부채입니다. 이 글에서는 aiogram 3.x의 비동기 봇에 Gemini를 제대로 통합하는 방법을 단계별로 설명합니다: 이벤트 루프를 차단하지 않고, 분당 요청/토큰 제한을 우회하며, 성능을 희생하지 않고 함수 호출을 구현하는 법입니다.

아키텍처: 간단한 핸들러에서 프로덕션 솔루션으로

기본적인 사용자-봇 상호작용 흐름은 Gemini를 통해 request → 처리 → response로 보이지만, 실제 시나리오에서는 추가 레이어가 필요합니다. 적절한 래핑 없이 동기 API를 비동기로 호출하면 aiogram 이벤트 루프가 멈추고, 특히 부하가 클 때 문제가 됩니다. 그래서 핵심 구성 요소는 다음과 같습니다:

• Asyncio Queue — 들어오는 요청을 버퍼링하기 위해
• Rate Limiter — Gemini 제한(분당 5~60 요청)에 맞추기 위해
• Response Cache — 반복 요청에서 토큰을 절약하기 위해
• Streaming Handler — Telegram 타임아웃(10초)을 우회하기 위해

Gemini API는 유료 플랜에서도 엄격한 제한을 적용합니다: 최대 응답 지연은 15초까지, 무료 티어는 분당 5회로 제한됩니다. 큐와 캐싱 없이는 사용자 경험이 엉망이 됩니다.

클라이언트 설정: 비동기 환경에서의 동기 SDK

Google GenAI SDK는 커뮤니티의 기대에도 불구하고 여전히 동기식입니다. aiogram 핸들러에서 generate_content()를 직접 호출하면 전체 이벤트 루프가 차단됩니다. 해결책: asyncio.to_thread나 run_in_executor로 감싸기.

import asyncio
from google import genai

class AsyncGeminiClient:
    def __init__(self, model: str = "gemini-3-flash-preview"):
        self.client = genai.Client()
        self.model = model
    
    async def generate(self, prompt: str) -> str:
        loop = asyncio.get_event_loop()
        response = await loop.run_in_executor(
            None,
            self._sync_generate,
            prompt
        )
        return response
    
    def _sync_generate(self, prompt: str) -> str:
        response = self.client.models.generate_content(
            model=self.model,
            contents=prompt
        )
        return response.text

이것은 최소 작동 래퍼입니다. 프로덕션에서는 예외 처리와 실행 메트릭을 추가하세요.

aiogram과의 통합: /start부터 스트리밍 응답까지

기본 핸들러는 메시지를 받는 것 이상을 해야 합니다: "typing" 표시로 상태 관리, 긴 응답 분할, 결과 캐싱. Telegram은 메시지 길이를 4096자로 제한하니 무시하면 잘립니다.

@router.message()
async def handle_message(message: types.Message):
    await message.bot.send_chat_action(chat_id=message.chat.id, action="typing")
    
    try:
        response = await gemini.generate(message.text)
        if len(response) > 4000:
            for i in range(0, len(response), 4000):
                await message.answer(response[i:i+4000])
        else:
            await message.answer(response)
    except Exception as e:
        logging.error(f"Gemini error: {e}")
        await message.answer("Chto-then poshlo not so. Poprobuyte pozzhe.")

장시간 요청의 경우 스트리밍을 사용하세요—텍스트가 생성됨에 따라 단일 메시지를 업데이트합니다. 이렇게 하면 타임아웃을 피하고 UX를 향상시킵니다.

함수 호출: Gemini가 에이전트가 될 때

Function Calling은 요청 맥락에 따라 모델이 외부 API를 호출할 수 있게 합니다. 예를 들어 사용자가 "내일 오후 3시에 회의 예약해"라고 하면 Gemini가 파라미터를 분석해 schedule_meeting 함수를 호출합니다.

주요 단계:

• JSON Schema 형식으로 함수 설명—모델이 시그니처를 이해하도록.
• GenerateContentConfig의 tools로 설명 전달.
• 응답의 function_call 처리하고 실제 호출 실행.
• 결과를 대화 일부로 채팅에 피드백.

예제 함수 선언:

schedule_meeting_function = {
    "name": "schedule_meeting",
    "description": "Withzdayot meetingsu with ukazannymi uchastnikami",
    "parameters": {
        "type": "object",
        "properties": {
            "attendees": {"type": "array", "items": {"type": "string"}},
            "date": {"type": "string"},
            "time": {"type": "string"},
            "topic": {"type": "string"}
        },
        "required": ["attendees", "date", "time", "topic"]
    }
}

이렇게 하면 정적 봇이 외부 시스템과 상호작용하는 동적 어시스턴트로 변합니다.

캐싱과 속도 제한: 파산과 밴으로부터 보호

Gemini는 토큰 단위 과금입니다. 캐싱 없이는 반복 질문마다 돈이 나갑니다. 프롬프트의 MD5 해시와 TTL을 사용한 간단한 인메모리 캐시로 해결:

class SimpleCache:
    def __init__(self, ttl_seconds: int = 3600):
        self._cache = {}
        self._ttl = ttl_seconds
    
    def get(self, key: str) -> Optional[str]:
        # check TTL and vozvrat values
    
    def set(self, key: str, value: str):
        self._cache[key] = (value, datetime.now())
    
    @staticmethod
    def hash_prompt(prompt: str) -> str:
        return hashlib.md5(prompt.lower().strip().encode()).hexdigest()

유료 티어에서도 속도 제한이 필수입니다. 비동기 컨텍스트 매니저나 미들웨어로 구현:

class AsyncRateLimiter:
    def __init__(self, max_requests: int, time_window: int = 60):
        self.max_requests = max_requests
        self.time_window = time_window
        self.requests = []
        self._lock = asyncio.Lock()
    
    async def acquire(self):
        async with self._lock:
            now = time.time()
            self.requests = [t for t in self.requests if now - t < self.time_window]
            if len(self.requests) >= self.max_requests:
                sleep_time = self.time_window - (now - self.requests[0])
                await asyncio.sleep(sleep_time + 0.1)
                return await self.acquire()
            self.requests.append(now)

흔한 실수와 피하는 법

모두가 겪는 세 가지 고전적 함정:

• 429 오류 (RESOURCE_EXHAUSTED) — 분당 요청(RPM), 분당 토큰(TPM), 일일 요청(RPD) 초과로 발생. 해결: 로컬 속도 제한자 + 지수 백오프.
• Telegram 타임아웃 — 서버가 응답을 10초 기다림. Gemini가 느리면 스트리밍이나 중간 메시지 사용.
• 오래된 모델 — Google이 정기적으로 이전 버전 종료. 2026년 3월에 Pro 모델이 구독 전용으로 변경. 배포 전 모델 가용성 확인.

주요 요점:

• 동기 Gemini SDK는 run_in_executor 래핑 필요—아니면 이벤트 루프 차단.
• API에 의존 말고 봇 측에서 속도 제한 구현.
• 긴 응답에 대한 Telegram 10초 타임아웃은 스트리밍으로만 우회.
• 캐싱으로 반복 시 비용과 지연 절감.
• Function Calling으로 텍스트 생성기에서 완전한 에이전트로 진화.

이러한 관행을 채택하면 프로덕션 사고가 줄고 부하 증가에도 안정적입니다. 아키텍처에서 타협 말고 토큰에서 절약하세요.

— Editorial Team