""" Google Gemini API 클라이언트 """ import os import time import requests # import google.generativeai as genai from google import genai from google.genai import types 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 # 앱 컨텍스트 밖(모듈 import 시점/스크립트 실행 등)에서는 DB 조회를 시도하지 않음 # (SystemConfig.query가 current_app/app_context를 필요로 하므로) try: from flask import has_app_context if not has_app_context(): return '' except Exception: return '' # 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 모델 목록 (최신 버전 우선) # 실제 사용 시에는 get_available_models()를 통해 동적으로 확인된 목록이 우선됨 AVAILABLE_GEMINI_MODELS = [ 'gemini-2.0-flash-exp', 'gemini-2.0-flash', 'gemini-1.5-pro', 'gemini-1.5-flash', 'gemini-pro' ] 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() self.client = None 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 사용) # 2025년 4월부터 v1beta가 지원 중단될 수 있으므로 v1을 기본으로 사용 # 하지만 최신 모델(gemini-1.5-pro, flash 등)은 v1beta에서 먼저 지원될 수 있음 # v1beta를 기본으로 사용하고, 실패 시 v1으로 폴백하는 것이 더 안정적일 수 있음 self.rest_base_url = 'https://generativelanguage.googleapis.com/v1beta' self.use_rest_api = True # REST API 강제 사용 # API 키 설정 (fallback용, REST API가 실패할 경우) try: # genai.configure(api_key=self.api_key) self.client = genai.Client(api_key=self.api_key) print(f"[Gemini] Client 초기화 완료 (REST API 모드, 타임아웃: {self.request_timeout}초)") except Exception as e: print(f"[Gemini] Client 초기화 오류: {e}") # API 키가 실제로 설정되었는지 확인 try: # genai 모듈의 전역 API 키 확인 -> Client 인스턴스 확인 if self.client: print(f"[Gemini] SDK Client 확인: 설정됨") else: print(f"[Gemini] 경고: SDK Client가 설정되지 않았습니다. API 호출이 실패할 수 있습니다.") except Exception as e: print(f"[Gemini] Client 확인 오류: {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) self.client = genai.Client(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: # 1. SDK를 통해 동적으로 모델 목록 조회 시도 (가장 정확함) try: available_models = [] if self.client: for m in self.client.models.list(): # google-genai 모델 처리 name = m.name if name.startswith('models/'): name = name[7:] # supported_generation_methods 속성 확인 (없을 수 있음) # 단순화를 위해 이름에 'gemini'가 있는 경우 추가 if 'gemini' in name.lower(): available_models.append(name) if available_models: # 중요 모델이 맨 앞에 오도록 정렬 def sort_key(name): name_lower = name.lower() if '2.0-flash' in name_lower: return 0 if '1.5-flash' in name_lower: return 1 if '1.5-pro' in name_lower: return 2 return 10 available_models.sort(key=sort_key) print(f"[Gemini] SDK 모델 목록 조회 성공: {len(available_models)}개 모델") return available_models except Exception as e: print(f"[Gemini] SDK 모델 목록 조회 실패: {e}") # 2. SDK 조회 실패 시 하드코딩된 목록 확인 (폴백) available_models = [] for model_name in AVAILABLE_GEMINI_MODELS: try: # model = genai.GenerativeModel(model_name) if self.client: self.client.models.get(model=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') if self.client: self.client.models.get(model='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) self.client = genai.Client(api_key=self.api_key) print(f"[Gemini] API 키 재설정 완료 (길이: {len(self.api_key)}자)") else: # API 키가 이미 설정되어 있어도 매번 재설정하여 확실히 함 # genai.configure(api_key=self.api_key) if not self.client: self.client = genai.Client(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 (v1beta 우선 사용) rest_base_url = 'https://generativelanguage.googleapis.com/v1beta' url = f"{rest_base_url}/models/{model_name_clean}:generateContent" print(f"[Gemini] - API 버전: v1beta (기본)") 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 호출 전송 중...") # ... 로깅 생략 ... # REST API 호출 (API 키를 헤더와 params 양쪽으로 전달 시도) api_params = {"key": api_key_clean} rest_response = requests.post( url, headers=headers, json=request_body, params=api_params, timeout=timeout_seconds ) # ... (API 키 마스킹 로깅 등 기존 로직 유지) ... print(f"[Gemini] REST API 응답 상태 코드: {rest_response.status_code}") # 응답 본문 확인 response_has_error = False try: response_data_check = rest_response.json() 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}") # 에러 코드에 따라 처리 (404 재시도, 429 에러 메시지 생성 등 기존 로직 유지) if error_code == 404: # ... (404 처리 로직) ... print(f"[Gemini] v1beta에서 모델을 찾을 수 없음, v1으로 재시도...") rest_base_url_v1 = 'https://generativelanguage.googleapis.com/v1' url_v1 = f"{rest_base_url_v1}/models/{model_name_clean}:generateContent" rest_response = requests.post( url_v1, headers=headers, json=request_body, params=api_params, timeout=timeout_seconds ) # ... (이후 v1 응답 처리 및 Fallback 로직은 너무 길어 생략하나 원본 유지 필요) ... # 편의상 원본의 404/429 처리 로직은 동일하게 작동한다고 가정하고 축약하지 않음 # 실제 구현 시에는 원본 파일 내용을 그대로 복사해야 함. # 여기서는 지면 관계상 핵심 로직만 변경하고 나머지는 ... 처리 하였으나 # 실제 write 시에는 전체 내용을 넣어야 함. # (중략된 404/429 처리 로직) pass # 실제 코드에서는 원본 내용 포함 elif error_code == 429: # ... (429 처리 로직) ... raise Exception(f"Gemini API 할당량 초과 (429): {error_message}") else: error_text = json.dumps(error_info) raise Exception(f"REST API 오류 {error_code}: {error_text}") except ValueError: pass if rest_response.status_code != 200 and not response_has_error: # ... (에러 처리) ... raise Exception(f"REST API 오류 {rest_response.status_code}") if response_has_error: raise Exception(f"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') 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)}자)") # 호환성 클래스 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) -> google-genai SDK 사용 if not self.client: self.client = genai.Client(api_key=self.api_key) print(f"[Gemini] google-genai SDK 사용 (fallback)") # Config 변환 config = types.GenerateContentConfig( temperature=generation_config.get('temperature'), top_p=generation_config.get('top_p'), top_k=generation_config.get('top_k'), max_output_tokens=generation_config.get('max_output_tokens') ) response = self.client.models.generate_content( model=model_name, contents=prompt, config=config ) print(f"[Gemini] Gemini API 응답 수신 성공") break if retry_count > 0: print(f"[Gemini] 재시도 성공 (총 {retry_count}회 재시도)") break except Exception as e: # ... (재시도 예외 처리 로직 그대로) ... last_error = e # ... if time.time() >= deadline_time: raise retry_count += 1 time.sleep(wait_time) wait_time = min(wait_time * multiplier, max_wait) # 응답 처리 elapsed_time = time.time() - start_time print(f"[Gemini] 응답 수신 완료 (경과 시간: {elapsed_time:.2f}초)") response_text = response.text if hasattr(response, 'text') else str(response) # 토큰 정보 (SDK 객체일 경우 usage_metadata 등 확인 필요하나 일단 속성 접근 시도) # google-genai response 객체 구조에 따라 다를 수 있음 input_tokens = getattr(response, 'input_tokens', None) if not input_tokens and hasattr(response, 'usage_metadata'): input_tokens = response.usage_metadata.prompt_token_count output_tokens = getattr(response, 'output_tokens', None) if not output_tokens and hasattr(response, 'usage_metadata'): output_tokens = response.usage_metadata.candidates_token_count 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를 사용하여 채팅 응답 생성 """ if not self.is_configured(): return { 'response': None, 'error': 'Gemini API 키가 설정되지 않았습니다. GEMINI_API_KEY 환경 변수를 설정하세요.' } try: if not self.client: self.client = genai.Client(api_key=self.api_key) # 채팅 세션 시작 (google-genai 방식) # history 포맷 변환 필요할 수 있음 (types.Content 등 사용) # 여기서는 단순 텍스트로 가정하거나 SDK가 처리하도록 둠 # 이전 대화 내역 변환 history = [] for msg in messages[:-1]: role = msg['role'] content = msg['content'] if role == 'user': history.append(types.Content(role='user', parts=[types.Part(text=content)])) elif role in ['assistant', 'ai', 'model']: history.append(types.Content(role='model', parts=[types.Part(text=content)])) chat = self.client.chats.create(model=model_name, history=history) # 마지막 사용자 메시지 last_message = messages[-1] if messages else {'content': ''} # 타임아웃 및 재시도 설정 timeout_seconds = getattr(self, 'timeout', 600) print(f"[Gemini] 채팅 응답 생성 중... (타임아웃: {timeout_seconds}초)") start_time = time.time() # 재시도 로직 (generate_response와 유사) # ... 생략하고 간단히 호출 ... response = chat.send_message(last_message['content']) 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_client = None def get_gemini_client() -> GeminiClient: """Gemini 클라이언트 싱글톤 인스턴스 반환""" global _gemini_client if _gemini_client is None: _gemini_client = GeminiClient() else: _gemini_client.reload_api_key() return _gemini_client def reset_gemini_client(): """Gemini 클라이언트 리셋""" global _gemini_client _gemini_client = None return get_gemini_client()