Spaces:
Sleeping
Sleeping
| # src/chimera/api_clients/gemini_client.py | |
| import google.generativeai as genai | |
| from ..config import GEMINI_API_KEY, GEMINI_MODEL_NAME | |
| from ..utils.logging_config import logger | |
| import asyncio | |
| # Configure the Gemini client globally (or instantiate as needed) | |
| if GEMINI_API_KEY: | |
| genai.configure(api_key=GEMINI_API_KEY) | |
| else: | |
| logger.error("Gemini API Key not configured!") | |
| # Depending on your flow, you might raise an error or handle this state | |
| async def generate_analysis(prompt: str, retries=3, delay=5) -> str: | |
| """ | |
| Generates content using the Gemini API with async handling and retries. | |
| """ | |
| if not GEMINI_API_KEY: | |
| return "Error: Gemini API Key not configured." | |
| try: | |
| model = genai.GenerativeModel(GEMINI_MODEL_NAME) | |
| # Note: The current google-generativeai SDK might not be fully async yet. | |
| # If performance becomes an issue, consider running sync calls | |
| # in an executor pool using asyncio.to_thread. | |
| # For now, we'll call it directly but within an async function context. | |
| logger.info(f"Sending prompt to Gemini (first ~100 chars): {prompt[:100]}...") | |
| # Placeholder for potential future truly async call | |
| # response = await model.generate_content_async(prompt) | |
| # Using run_in_executor for the synchronous call | |
| loop = asyncio.get_running_loop() | |
| for attempt in range(retries): | |
| try: | |
| # Wrap the synchronous call in run_in_executor | |
| response = await loop.run_in_executor( | |
| None, # Use default executor | |
| lambda: model.generate_content(prompt) | |
| ) | |
| logger.info("Received response from Gemini.") | |
| # Basic check for blocked content or empty response | |
| if not response.parts: | |
| # Handle safety flags/blocks if necessary | |
| if response.prompt_feedback.block_reason: | |
| logger.warning(f"Gemini prompt blocked: {response.prompt_feedback.block_reason}") | |
| return f"Error: Content generation blocked by safety filters ({response.prompt_feedback.block_reason})." | |
| else: | |
| logger.warning("Gemini returned empty response.") | |
| return "Error: Gemini returned an empty response." | |
| return response.text | |
| except Exception as e: # Catch broad exceptions from the SDK | |
| logger.error(f"Gemini API call attempt {attempt + 1} failed: {e}") | |
| if attempt < retries - 1: | |
| await asyncio.sleep(delay * (attempt + 1)) # Exponential backoff | |
| else: | |
| logger.error("Gemini API call failed after multiple retries.") | |
| return f"Error: Failed to get response from Gemini after {retries} attempts. ({e})" | |
| except Exception as e: | |
| logger.exception("Critical error during Gemini generation setup or execution.") | |
| return f"Error: An unexpected error occurred while contacting Gemini: {e}" | |
| return "Error: Unexpected exit from generate_analysis function." # Should not happen |