Create src/chimera/api_clients/gemini_client.py

src/chimera/api_clients/gemini_client.py

@@ -0,0 +1,67 @@
+# 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