[deploy] gemini API 키값 RR 분산 적용 배포

[selnem]

[deploy] gemini API 키값 RR 분산 적용 배포

[gemini-code-assist]

Code Review

This pull request introduces a multi-key round-robin rotation and failover mechanism (GeminiKeyPool) for Gemini API calls to improve reliability and handle rate limits. This includes updating various AI services, embedding models, and retry logic to support key rotation, along with adding comprehensive unit tests. The review feedback highlights a layering violation in app/core/config.py due to importing from the service layer, and suggests refactoring duplicate failover logic in PolicyCardnewsImageService.py into a shared helper method.

Important

The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.

[gemini-code-assist]

app/core/config.py는 애플리케이션의 핵심 설정 모듈이므로 서비스 레이어(app/services/...)를 참조하는 것은 레이어링 원칙(Layering Violation)에 위배되며, 순환 참조(Circular Import)의 원인이 될 수 있습니다.

parse_gemini_model_list는 단순히 콤마로 문자열을 분리하고 공백을 제거하는 헬퍼 함수이므로, 외부 모듈 의존성 없이 내부에서 직접 처리하도록 개선하는 것이 안전합니다.

    def resolved_gemini_api_keys(self) -> tuple[str, ...]:
        if self.gemini_api_keys.strip():
            raw = [k.strip() for k in self.gemini_api_keys.split(",") if k.strip()]
        elif self.gemini_api_key is not None:
            raw = [self.gemini_api_key.get_secret_value()]
        else:
            raw = []
        return tuple(dict.fromkeys(raw))

[gemini-code-assist]

_generate_with_gemini_image와 _generate_with_imagen 메서드 내에 동일한 failover 및 재시도 로직이 완전히 중복되어 구현되어 있습니다.

공통 failover 로직을 _execute_with_failover와 같은 별도의 헬퍼 메서드로 추출하면 코드 중복을 제거하고 가독성과 유지보수성을 크게 향상시킬 수 있습니다.

    async def _execute_with_failover(self, invoke) -> bytes:
        pool = self.key_pool
        if pool is not None and pool.enabled:
            key_index, _, client = await pool.acquire()
            try:
                return await invoke(client)
            except (genai_errors.ServerError, genai_errors.APIError) as exc:
                if not is_retryable_gemini_error(exc):
                    raise
                last_error: Exception = exc
                for failover_idx in pool.failover_indices(key_index):
                    try:
                        return await invoke(pool.client_at(failover_idx))
                    except (genai_errors.ServerError, genai_errors.APIError) as failover_exc:
                        last_error = failover_exc
                        if not is_retryable_gemini_error(failover_exc):
                            raise
                raise last_error
        return await invoke(self.client)

    async def _generate_with_gemini_image(self, *, model: str, prompt: str) -> bytes:
        config = types.GenerateContentConfig(
            response_modalities=[types.Modality.IMAGE],
        )
        async def invoke(client: genai.Client) -> bytes:
            response = await client.aio.models.generate_content(
                model=model,
                contents=prompt,
                config=config,
            )
            return self._extract_image_bytes(response)
        return await self._execute_with_failover(invoke)

    async def _generate_with_imagen(self, *, model: str, prompt: str) -> bytes:
        image_config = types.GenerateImagesConfig(
            number_of_images=1,
            aspect_ratio="3:4",
            output_mime_type="image/png",
        )
        async def invoke(client: genai.Client) -> bytes:
            response = await client.aio.models.generate_images(
                model=model,
                prompt=prompt,
                config=image_config,
            )
            generated = response.generated_images or []
            if not generated:
                raise RuntimeError("Imagen 응답에 generated_images가 없음")
            data = generated[0].image.image_bytes
            if not data:
                raise RuntimeError("Imagen image_bytes가 비어 있음")
            if isinstance(data, str):
                import base64
                return base64.b64decode(data)
            return bytes(data)
        return await self._execute_with_failover(invoke)