""" Google Gemini API 클라이언트 """ import os import time import requests import google.generativeai as genai from google.api_core import retry from typing import Optional, List, Dict import functools import json # Gemini API 키 (환경 변수 또는 데이터베이스에서 가져오기) def get_gemini_api_key(): """Gemini API 키 가져오기 (환경 변수 우선, 없으면 DB에서)""" # 환경 변수에서 먼저 확인 api_key = os.getenv('GEMINI_API_KEY', '').strip() if api_key: print(f"[Gemini] 환경 변수에서 API 키 가져옴 (길이: {len(api_key)}자)") return api_key # DB에서 가져오기 (순환 참조 방지를 위해 여기서 임포트) try: from app.database import SystemConfig api_key = SystemConfig.get_config('gemini_api_key', '').strip() if api_key: print(f"[Gemini] DB에서 API 키 가져옴 (길이: {len(api_key)}자)") else: print(f"[Gemini] DB에 API 키가 없거나 비어있음") return api_key except Exception as e: print(f"[Gemini] DB에서 API 키 조회 실패: {e}") return '' GEMINI_API_KEY = get_gemini_api_key() # 사용 가능한 Gemini 모델 목록 (최신 버전 우선) AVAILABLE_GEMINI_MODELS = [ 'gemini-2.0-flash-exp', 'gemini-1.5-pro', 'gemini-1.5-flash', 'gemini-1.5-pro-latest', 'gemini-1.5-flash-latest', 'gemini-pro', 'gemini-pro-vision' ] class GeminiClient: """Google Gemini API 클라이언트 클래스""" def __init__(self, api_key: Optional[str] = None): """Gemini 클라이언트 초기화""" if api_key: self.api_key = api_key else: # 최신 API 키 가져오기 (DB에서 동적으로) self.api_key = get_gemini_api_key() if not self.api_key: print("[Gemini] 경고: GEMINI_API_KEY가 설정되지 않았습니다. 환경 변수나 관리 페이지에서 설정하세요.") return try: # API 키 설정 및 타임아웃 설정 (기본 60초 -> 300초(5분)로 증가) # 전역 타임아웃 설정 self.request_timeout = 300 # 5분(300초) 타임아웃 # 재시도 정책 설정 self.retry_policy = retry.Retry( initial=10.0, # 초기 대기 시간 (10초) maximum=60.0, # 최대 대기 시간 (60초) multiplier=2.0, # 대기 시간 배수 deadline=600.0 # 전체 재시도 기간 (600초 = 10분) ) # REST API 사용을 위한 설정 # 환경 변수를 통해 HTTP 클라이언트 타임아웃 설정 os.environ.setdefault('HTTPX_TIMEOUT', str(self.request_timeout)) os.environ.setdefault('GOOGLE_API_TIMEOUT', str(self.request_timeout)) # REST API 엔드포인트 설정 (v1 사용) self.rest_base_url = 'https://generativelanguage.googleapis.com/v1' self.use_rest_api = True # REST API 강제 사용 # API 키 설정 (fallback용, REST API가 실패할 경우) try: genai.configure(api_key=self.api_key) print(f"[Gemini] API 키 설정 완료 (REST API 모드, 타임아웃: {self.request_timeout}초)") except Exception as e: print(f"[Gemini] API 키 설정 오류: {e}") # API 키가 실제로 설정되었는지 확인 try: # genai 모듈의 전역 API 키 확인 configured_key = getattr(genai, '_api_key', None) or getattr(genai, 'api_key', None) if configured_key: print(f"[Gemini] 전역 API 키 확인: 설정됨 (길이: {len(str(configured_key))}자)") else: print(f"[Gemini] 경고: 전역 API 키가 확인되지 않습니다. API 호출이 실패할 수 있습니다.") except Exception as e: print(f"[Gemini] API 키 확인 오류: {e}") print(f"[Gemini] 재시도 정책 적용됨") except Exception as e: print(f"[Gemini] API 키 설정 오류: {e}") def reload_api_key(self): """API 키를 다시 로드 (DB에서 최신 값 가져오기)""" self.api_key = get_gemini_api_key() if self.api_key: try: genai.configure(api_key=self.api_key) print(f"[Gemini] API 키 재로드 완료") return True except Exception as e: print(f"[Gemini] API 키 재로드 오류: {e}") return False return False def is_configured(self) -> bool: """Gemini API가 제대로 설정되었는지 확인""" return bool(self.api_key) def get_available_models(self) -> List[str]: """사용 가능한 Gemini 모델 목록 반환""" if not self.is_configured(): return [] try: # 실제 사용 가능한 모델 확인 available_models = [] for model_name in AVAILABLE_GEMINI_MODELS: try: model = genai.GenerativeModel(model_name) # 모델 접근 가능 여부 확인 available_models.append(model_name) except Exception as e: # 모델을 찾을 수 없으면 건너뛰기 continue # 모델을 찾지 못한 경우 기본 모델 시도 if not available_models: try: # 기본적으로 gemini-1.5-flash 시도 model = genai.GenerativeModel('gemini-1.5-flash') available_models.append('gemini-1.5-flash') except: pass print(f"[Gemini] 사용 가능한 모델: {available_models}") return available_models except Exception as e: print(f"[Gemini] 모델 목록 조회 오류: {e}") return [] def generate_response(self, prompt: str, model_name: str = 'gemini-1.5-flash', **kwargs) -> Dict: """ Gemini API를 사용하여 응답 생성 Args: prompt: 입력 프롬프트 model_name: 사용할 모델 이름 **kwargs: 추가 파라미터 (temperature, max_tokens 등) Returns: Dict: {'response': str, 'error': str or None} """ if not self.is_configured(): return { 'response': None, 'error': 'Gemini API 키가 설정되지 않았습니다. GEMINI_API_KEY 환경 변수를 설정하세요.' } try: # API 키를 항상 최신으로 다시 가져와서 설정 (DB에서 변경되었을 수 있음) current_api_key = get_gemini_api_key() if not current_api_key: return { 'response': None, 'error': 'Gemini API 키가 설정되지 않았습니다. 관리 페이지에서 API 키를 설정하세요.' } # API 키가 변경되었거나 없는 경우 재설정 if not self.api_key or self.api_key != current_api_key: self.api_key = current_api_key genai.configure(api_key=self.api_key) print(f"[Gemini] API 키 재설정 완료 (길이: {len(self.api_key)}자)") else: # API 키가 이미 설정되어 있어도 매번 재설정하여 확실히 함 genai.configure(api_key=self.api_key) # 모델 생성 model = genai.GenerativeModel(model_name) print(f"[Gemini] 모델 생성 완료: {model_name}") # 생성 설정 generation_config = { 'temperature': kwargs.get('temperature', 0.7), 'top_p': kwargs.get('top_p', 0.95), 'top_k': kwargs.get('top_k', 40), 'max_output_tokens': kwargs.get('max_output_tokens', 8192), } # 응답 생성 (타임아웃 설정) timeout_seconds = getattr(self, 'request_timeout', 300) # 5분 타임아웃 print(f"[Gemini] 모델 {model_name}로 응답 생성 중... (타임아웃: {timeout_seconds}초)") print(f"[Gemini] API 키 확인: 설정됨 (길이: {len(self.api_key)}자)") print(f"[Gemini] 프롬프트 길이: {len(prompt)}자") # 타임아웃을 위한 시작 시간 기록 start_time = time.time() # google-generativeai는 retry 파라미터를 지원하지 않으므로 # 재시도 정책을 수동으로 구현하여 적용 retry_policy = getattr(self, 'retry_policy', None) print(f"[Gemini] API 호출 시작 (타임아웃: {timeout_seconds}초, 재시도 정책 적용)") # 재시도 로직 구현 (재시도 정책 객체의 설정 사용) # retry.Retry 객체를 생성했지만, 실제로는 그 설정값들을 직접 사용 initial_wait = 10.0 # 초기 대기 시간 max_wait = 60.0 # 최대 대기 시간 multiplier = 2.0 # 대기 시간 배수 deadline = 600.0 # 전체 재시도 기간 (10분) wait_time = initial_wait deadline_time = time.time() + deadline retry_count = 0 last_error = None while True: try: # API 호출 전 API 키 재확인 및 재설정 current_key = get_gemini_api_key() if not current_key: raise Exception("API 키가 설정되지 않았습니다. 관리 페이지에서 API 키를 설정하세요.") # API 키가 변경되었거나 설정되지 않은 경우 재설정 if not self.api_key or self.api_key != current_key: self.api_key = current_key.strip() if current_key else None # 공백 제거 print(f"[Gemini] API 키 재설정 완료 (길이: {len(self.api_key) if self.api_key else 0}자)") # API 키 유효성 검사 if not self.api_key or not self.api_key.strip(): raise Exception("API 키가 비어있습니다. 관리 페이지에서 API 키를 설정하세요.") # API 키 앞뒤 공백 제거 api_key_clean = self.api_key.strip() if not api_key_clean: raise Exception("API 키가 유효하지 않습니다 (공백만 포함).") # API 키 형식 확인 (Google API 키는 보통 AIza로 시작) if not api_key_clean.startswith('AIza'): print(f"[Gemini] 경고: API 키가 일반적인 Google API 키 형식이 아닙니다 (AIza로 시작하지 않음)") # API 키 길이 확인 (일반적으로 39자 이상) if len(api_key_clean) < 20: raise Exception(f"API 키 길이가 너무 짧습니다 ({len(api_key_clean)}자). 올바른 API 키인지 확인하세요.") print(f"[Gemini] API 키 검증 완료 (길이: {len(api_key_clean)}자, 시작: {api_key_clean[:10]}..., 끝: ...{api_key_clean[-5:]})") # REST API를 직접 사용하여 호출 use_rest = getattr(self, 'use_rest_api', True) if use_rest: print(f"[Gemini] REST API 직접 호출 모드") # API 버전 및 모델 이름 정규화 # 모델 이름에서 'gemini:' 접두사 제거 및 정규화 model_name_clean = model_name.strip() if ':' in model_name_clean: # "gemini:gemini-1.5-flash" 형식인 경우 model_name_clean = model_name_clean.split(':', 1)[1].strip() elif model_name_clean.startswith('gemini-'): # "gemini-1.5-flash" 형식인 경우 그대로 사용 pass # REST API 베이스 URL (v1 사용) rest_base_url = 'https://generativelanguage.googleapis.com/v1' url = f"{rest_base_url}/models/{model_name_clean}:generateContent" print(f"[Gemini] - API 버전: v1") print(f"[Gemini] - 원본 모델 이름: {model_name}") print(f"[Gemini] - 정규화된 모델 이름: {model_name_clean}") print(f"[Gemini] - 전체 URL: {url}") # REST API 요청 본문 구성 request_body = { "contents": [{ "parts": [{ "text": prompt }] }], "generationConfig": generation_config } # REST API 헤더 (API 키를 헤더로 전달 시도) headers = { "Content-Type": "application/json", "x-goog-api-key": api_key_clean } print(f"[Gemini] REST API 호출 전송 중...") print(f"[Gemini] - URL: {url}") print(f"[Gemini] - 모델: {model_name}") print(f"[Gemini] - API 키: 설정됨 (길이: {len(api_key_clean)}자)") print(f"[Gemini] - API 키 시작: {api_key_clean[:15]}...") print(f"[Gemini] - API 키 끝: ...{api_key_clean[-10:]}") print(f"[Gemini] - 프롬프트 길이: {len(prompt)}자") # REST API 호출 (API 키를 헤더와 params 양쪽으로 전달 시도) # Google Gemini API는 헤더 또는 params 중 하나만 필요하지만, 양쪽 모두 시도 api_params = {"key": api_key_clean} print(f"[Gemini] - API 키 전달 방식: 헤더 (x-goog-api-key) 및 파라미터 (key)") rest_response = requests.post( url, headers=headers, json=request_body, params=api_params, timeout=timeout_seconds ) # 요청 URL에서 API 키 부분만 제거하여 로깅 (보안) request_url = rest_response.request.url if 'key=' in request_url: # API 키 부분을 마스킹 from urllib.parse import urlparse, parse_qs, urlencode, urlunparse parsed = urlparse(request_url) params = parse_qs(parsed.query) if 'key' in params: masked_params = params.copy() masked_key = masked_params['key'][0][:10] + '...' + masked_params['key'][0][-5:] masked_params['key'] = [masked_key] masked_query = urlencode(masked_params, doseq=True) masked_url = urlunparse((parsed.scheme, parsed.netloc, parsed.path, parsed.params, masked_query, parsed.fragment)) print(f"[Gemini] - 요청 URL (마스킹됨): {masked_url[:150]}...") else: print(f"[Gemini] - 요청 URL: {request_url[:150]}...") print(f"[Gemini] REST API 응답 상태 코드: {rest_response.status_code}") # 응답 본문 확인 (상태 코드가 200이어도 에러가 있을 수 있음) response_has_error = False try: response_data_check = rest_response.json() # 응답에 error 필드가 있는지 확인 if 'error' in response_data_check: response_has_error = True error_info = response_data_check['error'] error_code = error_info.get('code', rest_response.status_code) error_message = error_info.get('message', '알 수 없는 오류') print(f"[Gemini] 응답 본문에 에러 감지: code={error_code}, message={error_message}") # 에러 코드에 따라 처리 if error_code == 404: # 404 오류인 경우 v1beta로 재시도 print(f"[Gemini] v1에서 모델을 찾을 수 없음, v1beta로 재시도...") rest_base_url_v1beta = 'https://generativelanguage.googleapis.com/v1beta' url_v1beta = f"{rest_base_url_v1beta}/models/{model_name_clean}:generateContent" rest_response = requests.post( url_v1beta, headers=headers, json=request_body, params=api_params, timeout=timeout_seconds ) print(f"[Gemini] v1beta REST API 응답 상태 코드: {rest_response.status_code}") # v1beta 응답도 확인 if rest_response.status_code == 200: response_data_check_v1beta = rest_response.json() if 'error' in response_data_check_v1beta: # v1beta에서도 에러 발생 error_info_v1beta = response_data_check_v1beta['error'] error_code_v1beta = error_info_v1beta.get('code', 404) error_message_v1beta = error_info_v1beta.get('message', '알 수 없는 오류') # 모델 목록 조회 시도 available_models_str = "확인 불가" try: list_models_url_v1 = f"{rest_base_url}/models" list_response = requests.get( list_models_url_v1, headers={"x-goog-api-key": api_key_clean}, params={"key": api_key_clean}, timeout=10 ) if list_response.status_code == 200: models_data = list_response.json() available_models = [] for m in models_data.get('models', []): model_name_full = m.get('name', '') if '/' in model_name_full: model_name_short = model_name_full.split('/')[-1] else: model_name_short = model_name_full # generateContent를 지원하는 모델만 필터링 supported_methods = m.get('supportedGenerationMethods', []) if 'generateContent' in supported_methods: available_models.append(model_name_short) available_models_str = ', '.join(available_models[:10]) if available_models else '없음' print(f"[Gemini] 사용 가능한 모델 목록 (v1): {available_models[:10]}") except Exception as list_error: print(f"[Gemini] 모델 목록 조회 실패: {list_error}") error_text_v1beta = json.dumps(error_info_v1beta) raise Exception(f"REST API 오류 {error_code_v1beta}: {error_text_v1beta}\n사용 가능한 모델: {available_models_str}") else: # v1beta에서 성공 response_has_error = False print(f"[Gemini] v1beta에서 정상 응답 받음") elif rest_response.status_code != 200: error_text_v1beta = rest_response.text[:1000] if rest_response.text else '상세 정보 없음' raise Exception(f"REST API 오류 {rest_response.status_code}: {error_text_v1beta}") elif error_code == 429: # 429 오류: 할당량 초과 (재시도 불가능) print(f"[Gemini] ❌ 할당량 초과 오류 (429) 감지") print(f"[Gemini] 재시도 불가능한 오류: REST API 오류 {error_code}: {json.dumps(error_info)}") # 에러 메시지에서 권장 모델 추출 recommended_model = None if 'gemini-2.0-flash' in error_message.lower() or 'gemini 2.0' in error_message.lower(): recommended_model = "gemini-2.0-flash-exp" # 사용자 친화적인 에러 메시지 생성 quota_error_msg = f"""Gemini API 할당량 초과 (429) 현재 사용 중인 모델 '{model_name_clean}'의 일일 요청 한도를 초과했습니다. 해결 방법: 1. 내일까지 대기 (할당량이 자동으로 재설정됩니다) 2. 다른 Gemini 모델로 변경: - gemini-2.0-flash-exp (더 높은 할당량 제공) - gemini-1.5-pro - gemini-1.5-flash 3. Google AI Studio에서 할당량 확인: https://aistudio.google.com/app/apikey 상세 오류: {error_message[:200]}""" if recommended_model: quota_error_msg += f"\n\n권장: {recommended_model} 모델로 변경을 고려해보세요." raise Exception(quota_error_msg) else: # 404, 429가 아닌 다른 에러 error_text = json.dumps(error_info) raise Exception(f"REST API 오류 {error_code}: {error_text}") except ValueError: # JSON 파싱 실패 pass # 이미 에러가 처리되지 않은 경우에만 추가 에러 처리 # (response_has_error가 True면 이미 위에서 처리되었거나 Exception이 발생했을 것) if rest_response.status_code != 200 and not response_has_error: error_text = rest_response.text[:1000] if rest_response.text else '상세 정보 없음' # API 키 오류인 경우 더 상세한 안내 제공 if rest_response.status_code == 400 and ('API key' in error_text or 'API_KEY' in error_text): print(f"[Gemini] ❌ API 키 오류 감지") print(f"[Gemini] 현재 API 키 정보:") print(f"[Gemini] - 길이: {len(api_key_clean)}자") print(f"[Gemini] - 시작: {api_key_clean[:15]}...") print(f"[Gemini] - 끝: ...{api_key_clean[-10:]}") print(f"[Gemini] - 형식 확인: {'AIza로 시작' if api_key_clean.startswith('AIza') else 'AIza로 시작하지 않음 (비정상)'}") raise Exception(f"""REST API 오류 400: API 키가 유효하지 않습니다. 확인 사항: 1. 관리 페이지에서 API 키가 올바르게 설정되었는지 확인하세요. 2. Google AI Studio (https://aistudio.google.com/app/apikey)에서 API 키가 활성화되어 있는지 확인하세요. 3. API 키 형식이 올바른지 확인하세요 (일반적으로 'AIza'로 시작). 4. API 키에 불필요한 공백이나 줄바꿈이 포함되지 않았는지 확인하세요. 오류 상세: {error_text[:300]}""") raise Exception(f"REST API 오류 {rest_response.status_code}: {error_text}") # 에러가 있었지만 처리되지 않은 경우 (정상 응답이 아님) if response_has_error: # 이미 위에서 Exception이 발생했어야 하지만, 혹시 모르니 확인 error_text = rest_response.text[:1000] if rest_response.text else '상세 정보 없음' raise Exception(f"REST API 오류: 응답에 에러가 포함되어 있습니다. {error_text}") # REST API 응답 파싱 response_data = rest_response.json() # 토큰 사용량 정보 추출 input_tokens = None output_tokens = None if 'usageMetadata' in response_data: usage = response_data['usageMetadata'] input_tokens = usage.get('promptTokenCount') output_tokens = usage.get('candidatesTokenCount') total_tokens = usage.get('totalTokenCount') print(f"[Gemini] 토큰 사용량: 입력={input_tokens}, 출력={output_tokens}, 총={total_tokens}") # 응답에서 텍스트 추출 if 'candidates' in response_data and len(response_data['candidates']) > 0: candidate = response_data['candidates'][0] if 'content' in candidate and 'parts' in candidate['content']: parts = candidate['content']['parts'] if len(parts) > 0 and 'text' in parts[0]: response_text = parts[0]['text'] print(f"[Gemini] REST API 응답 수신 성공 (길이: {len(response_text)}자)") # genai 라이브러리 형식으로 변환 (호환성을 위해) class MockResponse: def __init__(self, text, input_tokens=None, output_tokens=None): self.text = text self.input_tokens = input_tokens self.output_tokens = output_tokens response = MockResponse(response_text, input_tokens, output_tokens) break else: raise Exception("REST API 응답에 텍스트가 없습니다.") else: raise Exception("REST API 응답 형식이 올바르지 않습니다.") else: raise Exception("REST API 응답에 candidates가 없습니다.") else: # 기존 genai 라이브러리 사용 (fallback) genai.configure(api_key=self.api_key) print(f"[Gemini] genai 라이브러리 사용 (fallback)") response = model.generate_content( prompt, generation_config=generation_config ) print(f"[Gemini] Gemini API 응답 수신 성공") break # 성공 시 루프 종료 if retry_count > 0: print(f"[Gemini] 재시도 성공 (총 {retry_count}회 재시도)") break except Exception as e: last_error = e error_str = str(e).lower() # 재시도 가능한 오류인지 확인 (타임아웃, 네트워크 오류 등) # 429(할당량 초과), 400(잘못된 요청), 401(인증 실패), 403(권한 없음)은 재시도 불가 non_retryable_errors = ['429', 'quota', 'exceeded', '400', '401', '403', 'api key', 'invalid', 'unauthorized', 'forbidden'] is_non_retryable = any(err in error_str for err in non_retryable_errors) retryable_errors = ['timeout', '503', '502', '504', 'connection', 'network', 'illegal metadata'] is_retryable = any(err in error_str for err in retryable_errors) and not is_non_retryable # deadline 확인 if time.time() >= deadline_time: print(f"[Gemini] 재시도 deadline 초과 ({deadline}초), 마지막 오류 반환") raise if is_retryable: retry_count += 1 print(f"[Gemini] 재시도 {retry_count} - {wait_time:.1f}초 후 재시도 (오류: {str(e)[:100]})") time.sleep(wait_time) wait_time = min(wait_time * multiplier, max_wait) # 배수로 증가, 최대값 제한 else: # 재시도 불가능한 오류 if is_non_retryable: print(f"[Gemini] 재시도 불가능한 오류 (할당량/인증 오류): {str(e)[:200]}") else: print(f"[Gemini] 재시도 불가능한 오류: {str(e)[:100]}") raise # 응답 시간 확인 elapsed_time = time.time() - start_time print(f"[Gemini] 응답 수신 완료 (경과 시간: {elapsed_time:.2f}초)") # 응답 텍스트 추출 response_text = response.text if hasattr(response, 'text') else str(response) # 토큰 정보 추출 input_tokens = getattr(response, 'input_tokens', None) output_tokens = getattr(response, 'output_tokens', None) print(f"[Gemini] 응답 생성 완료: {len(response_text)}자, 입력 토큰: {input_tokens}, 출력 토큰: {output_tokens}") return { 'response': response_text, 'error': None, 'input_tokens': input_tokens, 'output_tokens': output_tokens } except Exception as e: error_msg = f'Gemini API 오류: {str(e)}' print(f"[Gemini] {error_msg}") return { 'response': None, 'error': error_msg } def generate_chat_response(self, messages: List[Dict], model_name: str = 'gemini-1.5-flash', **kwargs) -> Dict: """ Gemini API를 사용하여 채팅 응답 생성 Args: messages: 메시지 리스트 [{'role': 'user', 'content': '...'}, ...] model_name: 사용할 모델 이름 **kwargs: 추가 파라미터 Returns: Dict: {'response': str, 'error': str or None} """ if not self.is_configured(): return { 'response': None, 'error': 'Gemini API 키가 설정되지 않았습니다. GEMINI_API_KEY 환경 변수를 설정하세요.' } try: # 모델 생성 model = genai.GenerativeModel(model_name) # 채팅 세션 시작 chat = model.start_chat(history=[]) # 이전 대화 내역 추가 (user와 assistant 메시지) for msg in messages[:-1]: # 마지막 메시지 제외 if msg['role'] == 'user': chat.send_message(msg['content']) elif msg['role'] == 'assistant' or msg['role'] == 'ai': # Gemini는 사용자 메시지만 직접 보내므로, assistant 메시지는 히스토리로 처리하지 않음 pass # 마지막 사용자 메시지로 응답 생성 (타임아웃 설정) last_message = messages[-1] if messages else {'content': ''} timeout_seconds = getattr(self, 'timeout', 600) # 5분 타임아웃 print(f"[Gemini] 채팅 응답 생성 중... (타임아웃: {timeout_seconds}초)") # 타임아웃을 위한 시작 시간 기록 start_time = time.time() # google-generativeai는 retry 파라미터를 지원하지 않으므로 # 재시도 정책을 수동으로 구현하여 적용 print(f"[Gemini] 채팅 API 호출 시작 (타임아웃: {timeout_seconds}초, 재시도 정책 적용)") # 재시도 로직 구현 (재시도 정책 객체의 설정 사용) initial_wait = 10.0 # 초기 대기 시간 max_wait = 60.0 # 최대 대기 시간 multiplier = 2.0 # 대기 시간 배수 deadline = 600.0 # 전체 재시도 기간 (10분) wait_time = initial_wait deadline_time = time.time() + deadline retry_count = 0 last_error = None while True: try: response = chat.send_message(last_message['content']) # 성공 시 루프 종료 if retry_count > 0: print(f"[Gemini] 채팅 재시도 성공 (총 {retry_count}회 재시도)") break except Exception as e: last_error = e error_str = str(e).lower() # 재시도 가능한 오류인지 확인 (타임아웃, 네트워크 오류 등) # 429(할당량 초과), 400(잘못된 요청), 401(인증 실패), 403(권한 없음)은 재시도 불가 non_retryable_errors = ['429', 'quota', 'exceeded', '400', '401', '403', 'api key', 'invalid', 'unauthorized', 'forbidden'] is_non_retryable = any(err in error_str for err in non_retryable_errors) retryable_errors = ['timeout', '503', '502', '504', 'connection', 'network', 'illegal metadata'] is_retryable = any(err in error_str for err in retryable_errors) and not is_non_retryable # deadline 확인 if time.time() >= deadline_time: print(f"[Gemini] 채팅 재시도 deadline 초과 ({deadline}초), 마지막 오류 반환") raise if is_retryable: retry_count += 1 print(f"[Gemini] 채팅 재시도 {retry_count} - {wait_time:.1f}초 후 재시도 (오류: {str(e)[:100]})") time.sleep(wait_time) wait_time = min(wait_time * multiplier, max_wait) # 배수로 증가, 최대값 제한 else: # 재시도 불가능한 오류 if is_non_retryable: print(f"[Gemini] 채팅 재시도 불가능한 오류 (할당량/인증 오류): {str(e)[:200]}") else: print(f"[Gemini] 채팅 재시도 불가능한 오류: {str(e)[:100]}") raise elapsed_time = time.time() - start_time print(f"[Gemini] 채팅 응답 수신 완료 (경과 시간: {elapsed_time:.2f}초)") response_text = response.text if hasattr(response, 'text') else str(response) print(f"[Gemini] 채팅 응답 생성 완료: {len(response_text)}자") return { 'response': response_text, 'error': None } except Exception as e: error_msg = f'Gemini API 오류: {str(e)}' print(f"[Gemini] {error_msg}") return { 'response': None, 'error': error_msg } # 전역 Gemini 클라이언트 인스턴스 _gemini_client = None def get_gemini_client() -> GeminiClient: """Gemini 클라이언트 싱글톤 인스턴스 반환""" global _gemini_client if _gemini_client is None: _gemini_client = GeminiClient() else: # API 키가 변경되었을 수 있으므로 재로드 시도 _gemini_client.reload_api_key() return _gemini_client def reset_gemini_client(): """Gemini 클라이언트 리셋 (API 키 변경 후 호출)""" global _gemini_client _gemini_client = None return get_gemini_client()