Spaces:
Running
Running
import os | |
import sys | |
cur_dir = os.getcwd() | |
parent_dir = os.path.realpath(os.path.join(os.path.dirname(cur_dir))) | |
if parent_dir not in sys.path: | |
sys.path.append(parent_dir) | |
sys.path.append(cur_dir) | |
sys.path.insert(1, ".") | |
import json | |
import time | |
import re | |
import google.generativeai as genai | |
from PIL import Image | |
import logging | |
# Set up logging | |
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') | |
logger = logging.getLogger(__name__) | |
## create connection with the model | |
def create_connection(api_key, model_name='gemini-1.5-flash'): | |
try: | |
genai.configure(api_key=api_key) | |
model = genai.GenerativeModel(model_name) | |
logger.info(f"Connected to model: {model_name}") | |
return model | |
except Exception as e: | |
logger.error(f"Failed to create connection: {str(e)}") | |
raise | |
## attempt to clean up malformed JSON | |
def clean_json_response(text): | |
""" | |
Attempts to extract valid JSON from a potentially malformed response. | |
Args: | |
text (str): Raw response from the model. | |
Returns: | |
str: Cleaned JSON string or None if unrecoverable. | |
""" | |
try: | |
# Strip any non-JSON text (e.g., "Here is the result: {...}") | |
json_match = re.search(r'\{.*\}', text, re.DOTALL) | |
if json_match: | |
cleaned_text = json_match.group(0) | |
json.loads(cleaned_text) # Validate | |
return cleaned_text | |
return None | |
except json.JSONDecodeError: | |
return None |