gemini_adapter.py

"""
Adapter to convert between Anthropic-style API calls and Google GenAI SDK
"""
from google import genai
from google.genai import types
import json
import os

# Lazy client initialization
_client = None

def get_client():
    """Get or create the Gemini client"""
    global _client
    if _client is None:
        api_key = os.getenv("GOOGLE_API_KEY")
        if not api_key:
            raise ValueError("GOOGLE_API_KEY environment variable not set")
        _client = genai.Client(api_key=api_key)
    return _client

def clean_schema_for_gemini(schema):
    """
    Remove fields that Gemini doesn't support from JSON schema
    """
    if not isinstance(schema, dict):
        return schema

    cleaned = {}
    # Gemini doesn't support: title, default, examples, additionalProperties
    skip_fields = {"title", "default", "examples", "additionalProperties"}

    for key, value in schema.items():
        if key in skip_fields:
            continue
        if isinstance(value, dict):
            cleaned[key] = clean_schema_for_gemini(value)
        elif isinstance(value, list):
            cleaned[key] = [clean_schema_for_gemini(item) if isinstance(item, dict) else item for item in value]
        else:
            cleaned[key] = value

    return cleaned

def convert_tools_to_gemini_format(anthropic_tools):
    """
    Convert Anthropic tool schema to Gemini function declarations
    """
    function_declarations = []

    for tool in anthropic_tools:
        # Clean the input schema
        cleaned_schema = clean_schema_for_gemini(tool["input_schema"])

        # Convert to Gemini FunctionDeclaration
        func_decl = types.FunctionDeclaration(
            name=tool["name"],
            description=tool["description"],
            parameters=cleaned_schema
        )
        function_declarations.append(func_decl)

    # Wrap in Tool object
    if function_declarations:
        return [types.Tool(function_declarations=function_declarations)]
    return None

def convert_messages_to_gemini_format(anthropic_messages):
    """
    Convert Anthropic messages format to Gemini Content format
    """
    contents = []

    for msg in anthropic_messages:
        role = msg["role"]
        content = msg["content"]

        # Convert role: assistant -> model
        gemini_role = "model" if role == "assistant" else "user"

        parts = []

        # Handle different content types
        if isinstance(content, str):
            # Simple text message
            parts.append(types.Part(text=content))
        elif isinstance(content, dict):
            # Could be Gradio file format {"path": ..., "mime_type": ...}
            # Skip audio/video files - they can't be sent to Gemini text API
            if content.get("path") and content.get("mime_type"):
                # print(f"DEBUG convert_messages: Skipping file content: {content.get('mime_type')}")
                continue
            # Could be text content {"type": "text", "text": "..."}
            elif content.get("type") == "text":
                parts.append(types.Part(text=content.get("text", "")))
        elif hasattr(content, '__class__') and 'Audio' in content.__class__.__name__:
            # Skip Gradio Audio component objects
            # print(f"DEBUG convert_messages: Skipping Gradio component: {content.__class__.__name__}")
            continue
        elif isinstance(content, list):
            # Complex content with tool calls/results
            for item in content:
                if isinstance(item, dict):
                    if item.get("type") == "text":
                        parts.append(types.Part(text=item.get("text", "")))
                    elif item.get("type") == "tool_use":
                        # Gemini function call
                        parts.append(types.Part(
                            function_call=types.FunctionCall(
                                name=item.get("name"),
                                args=item.get("input", {})
                            )
                        ))
                    elif item.get("type") == "tool_result":
                        # Gemini function response
                        try:
                            response_data = json.loads(item.get("content", "{}"))
                        except:
                            response_data = {"result": item.get("content", "")}

                        parts.append(types.Part(
                            function_response=types.FunctionResponse(
                                name=item.get("tool_use_id", "unknown"),
                                response=response_data
                            )
                        ))
                elif hasattr(item, 'type'):
                    # Anthropic SDK objects
                    if item.type == "text":
                        parts.append(types.Part(text=item.text))
                    elif item.type == "tool_use":
                        parts.append(types.Part(
                            function_call=types.FunctionCall(
                                name=item.name,
                                args=item.input
                            )
                        ))

        if parts:
            contents.append(types.Content(role=gemini_role, parts=parts))

    return contents

def call_gemini_with_tools(model_name, system_prompt, messages, tools, max_tokens=2048):
    """
    Call Gemini API with tools in Anthropic-compatible way
    """
    try:
        # Convert tools to Gemini format
        gemini_tools = convert_tools_to_gemini_format(tools) if tools else None

        # Convert messages to Gemini format
        contents = convert_messages_to_gemini_format(messages)

        # print(f"DEBUG gemini_adapter: Converted {len(messages)} messages to {len(contents)} Contents")
        # print(f"DEBUG gemini_adapter: Tools: {len(gemini_tools[0].function_declarations) if gemini_tools else 0}")

        # Call API
        # print(f"DEBUG gemini_adapter: Calling Gemini API...")
        client = get_client()
        
        # Generate content (no timeout parameter - not supported)
        config = types.GenerateContentConfig(
            system_instruction=system_prompt,
            max_output_tokens=max_tokens,
            temperature=1.0,
            tools=gemini_tools
        )
        
        response = client.models.generate_content(
            model=model_name,
            contents=contents,
            config=config
        )

        print(f"DEBUG gemini_adapter: Got response")

        # print(f"DEBUG gemini_adapter: Response type: {type(response)}")
        # print(f"DEBUG gemini_adapter: Response has candidates: {hasattr(response, 'candidates')}")
        if hasattr(response, 'candidates') and response.candidates:
            candidate = response.candidates[0]
            # print(f"DEBUG gemini_adapter: First candidate type: {type(candidate)}")
            # print(f"DEBUG gemini_adapter: finish_reason: {getattr(candidate, 'finish_reason', 'UNKNOWN')}")

            # Log function calls if present
            if hasattr(candidate, 'content') and candidate.content and hasattr(candidate.content, 'parts') and candidate.content.parts:
                for i, part in enumerate(candidate.content.parts):
                    if hasattr(part, 'function_call') and part.function_call:
                        # print(f"DEBUG gemini_adapter: Part {i} has function_call: {part.function_call.name}")
                        print(f"DEBUG gemini_adapter: Function args: {dict(part.function_call.args) if part.function_call.args else {}}")
        return response

    except Exception as e:
        print(f"DEBUG gemini_adapter ERROR: {type(e).__name__}: {str(e)}")
        import traceback
        traceback.print_exc()
        raise

def extract_tool_calls_from_gemini_response(response):
    """
    Extract function calls from Gemini response
    Returns list of {name, input, id} dicts compatible with Anthropic format
    """
    tool_calls = []

    # print(f"DEBUG extract_tool_calls: Response type: {type(response)}")
    # print(f"DEBUG extract_tool_calls: Has candidates: {hasattr(response, 'candidates')}")

    if not hasattr(response, 'candidates') or not response.candidates:
        # print(f"DEBUG extract_tool_calls: No candidates found")
        return tool_calls

    candidate = response.candidates[0]
    # print(f"DEBUG extract_tool_calls: Candidate has content: {hasattr(candidate, 'content')}")

    if not hasattr(candidate, 'content') or not candidate.content:
        # print(f"DEBUG extract_tool_calls: No content in candidate")
        return tool_calls

    # print(f"DEBUG extract_tool_calls: Content has parts: {hasattr(candidate.content, 'parts')}")

    if not hasattr(candidate.content, 'parts') or not candidate.content.parts:
        # print(f"DEBUG extract_tool_calls: No parts in content")
        return tool_calls

    for i, part in enumerate(candidate.content.parts):
        # print(f"DEBUG extract_tool_calls: Part {i} has function_call: {hasattr(part, 'function_call')}")
        if hasattr(part, 'function_call') and part.function_call:
            fc = part.function_call
            tool_calls.append({
                "name": fc.name,
                "input": dict(fc.args) if fc.args else {},
                "id": f"call_{i}"
            })

    # print(f"DEBUG extract_tool_calls: Found {len(tool_calls)} tool calls")
    return tool_calls

def get_text_from_gemini_response(response):
    """Extract text content from Gemini response

    Returns:
        tuple: (text, error_message) where error_message is None if successful
    """
    # print(f"DEBUG get_text: Response type: {type(response)}")
    # print(f"DEBUG get_text: Has candidates: {hasattr(response, 'candidates')}")

    if not hasattr(response, 'candidates') or not response.candidates:
        # print(f"ERROR get_text: No candidates in response")
        return "", "No response candidates received from AI. Please try again."

    candidate = response.candidates[0]
    finish_reason = getattr(candidate, 'finish_reason', 'UNKNOWN')
    # print(f"DEBUG get_text: Candidate has content: {hasattr(candidate, 'content')}")
    # print(f"DEBUG get_text: finish_reason: {finish_reason}")

    if not hasattr(candidate, 'content') or not candidate.content:
        # print(f"WARNING get_text: No content in candidate")
        # print(f"WARNING get_text: safety_ratings: {getattr(candidate, 'safety_ratings', 'N/A')}")

        # Provide specific error messages based on finish_reason
        if finish_reason == 'SAFETY':
            return "", "Response blocked by safety filters. Please try different phrasing."
        elif finish_reason == 'RECITATION':
            return "", "Response contained copied content. Please try again."
        elif finish_reason == 'MAX_TOKENS':
            return "", "Response too long. Please try again."
        elif finish_reason != 'STOP':
            return "", f"AI response incomplete (reason: {finish_reason}). Please try again."
        else:
            return "", "AI returned empty response. Please try again."

    # print(f"DEBUG get_text: Content has parts: {hasattr(candidate.content, 'parts') if candidate.content else False}")

    if not candidate.content or not hasattr(candidate.content, 'parts') or not candidate.content.parts:
        # print(f"WARNING get_text: No parts in content")
        return "", "AI response had no content. Please try again."

    text_parts = []
    has_function_call = False
    for i, part in enumerate(candidate.content.parts):
        # print(f"DEBUG get_text: Part {i} has text: {hasattr(part, 'text')}")
        if hasattr(part, 'text') and part.text:
            text_parts.append(part.text)
        if hasattr(part, 'function_call') and part.function_call:
            has_function_call = True

    result = " ".join(text_parts)
    # print(f"DEBUG get_text: Extracted text length: {len(result)}, has_function_call: {has_function_call}")

    # Empty text is OK if there's a function call (tool-only response)
    if (not result or result.strip() == "") and not has_function_call:
        # print(f"WARNING get_text: Empty text extracted and no function call")
        return "", "AI returned empty text. Please try again."

    return result, None


def chat_with_gemini_and_tools(system_prompt, messages, tools=None, max_tokens=1024, temperature=1.0, model_name="gemini-2.5-flash"):
    """
    Chat function with optional tool calling support

    Args:
        system_prompt: System instruction for the AI
        messages: List of message dicts with 'role' and 'content' keys
        tools: Optional list of tool definitions in Anthropic format
        max_tokens: Maximum tokens in response (default: 1024)
        temperature: Randomness in response (0.0-2.0, default: 1.0)
        model_name: Gemini model to use

    Returns:
        tuple: (response_text, tool_calls, error_message)
               - response_text: Text response from AI (or empty if tool call only)
               - tool_calls: List of tool call dicts [{name, input, id}] or None
               - error_message: Error string or None if successful
    """
    try:
        # Convert messages to Gemini format
        contents = convert_messages_to_gemini_format(messages)

        # Convert tools to Gemini format if provided
        gemini_tools = convert_tools_to_gemini_format(tools) if tools else None

        # print(f"DEBUG chat_with_gemini_and_tools: {len(messages)} messages, {len(tools) if tools else 0} tools")

        # Get client
        client = get_client()

        # Configure generation
        # Use tool_config to encourage tool usage when tools are available
        tool_config = None
        if gemini_tools:
            # AUTO mode: Gemini decides when to call functions
            # This helps ensure it actually calls the tool when appropriate
            tool_config = types.ToolConfig(
                function_calling_config=types.FunctionCallingConfig(
                    mode="AUTO"
                )
            )

        config = types.GenerateContentConfig(
            system_instruction=system_prompt,
            max_output_tokens=max_tokens,
            temperature=temperature,
            tools=gemini_tools,
            tool_config=tool_config
        )

        # Call API
        response = client.models.generate_content(
            model=model_name,
            contents=contents,
            config=config
        )

        # print(f"DEBUG chat_with_gemini_and_tools: Got response")

        # Check for tool calls first
        tool_calls = extract_tool_calls_from_gemini_response(response)

        # Extract text (may be empty if tool call only)
        text, error = get_text_from_gemini_response(response)

        # If there's an error but we have tool calls, that's OK (tool-only response)
        if error and not tool_calls:
            # print(f"ERROR chat_with_gemini_and_tools: {error}")
            return "", None, error
        elif error and tool_calls:
            # print(f"DEBUG chat_with_gemini_and_tools: Error '{error}' but have {len(tool_calls)} tool calls, proceeding")
            text = ""  # Clear any error text

        # print(f"DEBUG chat_with_gemini_and_tools: text={len(text)} chars, tool_calls={len(tool_calls) if tool_calls else 0}")
        return text, tool_calls if tool_calls else None, None

    except Exception as e:
        error_msg = f"Error calling Gemini: {type(e).__name__}: {str(e)}"
        print(f"ERROR chat_with_gemini_and_tools: {error_msg}")
        import traceback
        traceback.print_exc()
        return "", None, error_msg


def chat_with_gemini(system_prompt, messages, max_tokens=1024, temperature=1.0, model_name="gemini-2.5-flash"):
    """
    Simple chat function for conversational AI without tool calling

    Args:
        system_prompt: System instruction for the AI
        messages: List of message dicts with 'role' and 'content' keys
        max_tokens: Maximum tokens in response (default: 1024)
        temperature: Randomness in response (0.0-2.0, default: 1.0)
        model_name: Gemini model to use

    Returns:
        tuple: (response_text, error_message) where error_message is None if successful
    """
    try:
        # Convert messages to Gemini format
        contents = convert_messages_to_gemini_format(messages)

        # print(f"DEBUG chat_with_gemini: Converted {len(messages)} messages to {len(contents)} Contents")
        # print(f"DEBUG chat_with_gemini: Calling Gemini API with model {model_name}")

        # Get client
        client = get_client()

        # Configure generation
        config = types.GenerateContentConfig(
            system_instruction=system_prompt,
            max_output_tokens=max_tokens,
            temperature=temperature
        )

        # Call API
        response = client.models.generate_content(
            model=model_name,
            contents=contents,
            config=config
        )

        # print(f"DEBUG chat_with_gemini: Got response")

        # Extract text
        text, error = get_text_from_gemini_response(response)

        if error:
            # print(f"ERROR chat_with_gemini: {error}")
            return "", error

        # print(f"DEBUG chat_with_gemini: Successfully extracted text ({len(text)} chars)")
        return text, None

    except Exception as e:
        error_msg = f"Error calling Gemini: {type(e).__name__}: {str(e)}"
        print(f"ERROR chat_with_gemini: {error_msg}")
        import traceback
        traceback.print_exc()
        return "", error_msg