"""
Google (Gemini) provider implementation.
This module implements the BaseLLMProvider interface for Google's Gemini API.
Key Differences:
- Uses google-generativeai SDK
- Different content format (parts-based)
- Function calling called "function declarations"
- Different safety settings
"""
from __future__ import annotations
import json
from typing import Any, Dict, List, Optional
import google.generativeai as genai
from google.api_core import exceptions as google_exceptions
from tenacity import retry, stop_after_attempt, wait_exponential
from .base import (
BaseLLMProvider,
LLMResponse,
ToolCall,
ToolCallStatus,
LLMProviderError,
RateLimitError,
AuthenticationError,
)
class GoogleProvider(BaseLLMProvider):
"""
Google (Gemini) implementation of the LLM provider interface.
Supports:
- Gemini Pro, Gemini Flash
- Function calling
- Multimodal inputs (text, images)
- Long context windows
"""
def __init__(self, config: Dict[str, Any]):
"""
Initialize Google provider.
Args:
config: Must contain:
- api_key: Google API key
- model: Model name (e.g., 'gemini-2.0-flash-exp')
- max_tokens: Optional max tokens
"""
super().__init__(config)
genai.configure(api_key=config["api_key"])
self.model_name_str = config["model"]
self.model = genai.GenerativeModel(self.model_name_str)
self.default_max_tokens = config.get("max_tokens", 4096)
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=2, max=10),
)
async def generate(
self,
messages: List[Dict[str, Any]],
tools: Optional[List[Dict[str, Any]]] = None,
temperature: Optional[float] = None,
max_tokens: Optional[int] = None,
**kwargs
) -> LLMResponse:
"""
Generate a response using Google's Gemini API.
Note: Gemini uses a different message format with 'parts' instead of 'content'.
"""
try:
# Convert messages to Gemini format
gemini_messages = self._convert_messages_to_gemini(messages)
# Configure generation
generation_config = {
"max_output_tokens": max_tokens if max_tokens is not None else self.default_max_tokens,
}
if temperature is not None:
generation_config["temperature"] = temperature
# Add tools if provided
if tools:
gemini_tools = self._convert_mcp_tools_to_gemini(tools)
self.model = genai.GenerativeModel(
self.model_name_str,
tools=gemini_tools
)
# Generate response
# Note: Gemini's async API is different, using generate_content_async
response = await self.model.generate_content_async(
gemini_messages,
generation_config=generation_config,
)
# Extract content
text_content = ""
tool_calls = []
if response.parts:
for part in response.parts:
if hasattr(part, 'text') and part.text:
text_content += part.text
elif hasattr(part, 'function_call'):
fc = part.function_call
tool_calls.append(ToolCall(
id=fc.name, # Gemini doesn't provide IDs, use name
name=fc.name,
arguments=dict(fc.args),
status=ToolCallStatus.PENDING,
))
# Extract usage (Gemini provides this differently)
usage = {
"prompt_tokens": getattr(response.usage_metadata, 'prompt_token_count', 0) if hasattr(response, 'usage_metadata') else 0,
"completion_tokens": getattr(response.usage_metadata, 'candidates_token_count', 0) if hasattr(response, 'usage_metadata') else 0,
"total_tokens": getattr(response.usage_metadata, 'total_token_count', 0) if hasattr(response, 'usage_metadata') else 0,
}
finish_reason = response.candidates[0].finish_reason.name if response.candidates else "STOP"
return LLMResponse(
content=text_content.strip(),
tool_calls=tool_calls,
finish_reason=finish_reason,
usage=usage,
raw_response=response,
)
except google_exceptions.ResourceExhausted as e:
raise RateLimitError(
"Google API rate limit exceeded",
provider="google",
original_error=e,
)
except google_exceptions.PermissionDenied as e:
raise AuthenticationError(
"Invalid Google API key or insufficient permissions",
provider="google",
original_error=e,
)
except Exception as e:
raise LLMProviderError(
f"Google API error: {str(e)}",
provider="google",
original_error=e,
)
async def generate_with_tools(
self,
messages: List[Dict[str, Any]],
tools: List[Dict[str, Any]],
max_turns: int = 10,
**kwargs
) -> tuple[str, List[Dict[str, Any]]]:
"""
Generate response with automatic tool calling loop.
Gemini handles tool calling similarly to other providers but with
different message formats.
"""
conversation = list(messages)
for turn in range(max_turns):
response = await self.generate(
messages=conversation,
tools=tools,
**kwargs
)
if not response.has_tool_calls:
return response.content, conversation
# Add assistant's response with tool calls
assistant_message = {
"role": "model", # Gemini uses "model" instead of "assistant"
"parts": []
}
if response.content:
assistant_message["parts"].append({"text": response.content})
for tc in response.tool_calls:
assistant_message["parts"].append({
"function_call": {
"name": tc.name,
"args": tc.arguments,
}
})
conversation.append(assistant_message)
# Return early - orchestrator will execute tools
return "", conversation
from .base import MaxTurnsExceededError
raise MaxTurnsExceededError(
f"Tool calling loop exceeded {max_turns} turns",
provider="google",
)
def format_tool_result(self, tool_call: ToolCall) -> Dict[str, Any]:
"""
Format tool result for Gemini's expected format.
Gemini expects function responses in a specific format.
"""
return {
"role": "function",
"parts": [
{
"function_response": {
"name": tool_call.name,
"response": tool_call.result if tool_call.result else {},
}
}
]
}
def _convert_messages_to_gemini(self, messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""
Convert standard message format to Gemini's format.
Standard: {"role": "user", "content": "text"}
Gemini: {"role": "user", "parts": [{"text": "text"}]}
"""
gemini_messages = []
for msg in messages:
role = msg["role"]
# Map roles
if role == "assistant":
role = "model"
elif role == "system":
# Gemini doesn't have system role, convert to user
role = "user"
content = msg.get("content", "")
if isinstance(content, str):
gemini_messages.append({
"role": role,
"parts": [{"text": content}]
})
else:
# Already in parts format
gemini_messages.append(msg)
return gemini_messages
def _convert_mcp_tools_to_gemini(self, mcp_tools: List[Dict[str, Any]]) -> List[Any]:
"""
Convert MCP tool format to Gemini's function declarations.
Gemini uses a different schema format for function declarations.
"""
from google.generativeai.types import FunctionDeclaration, Tool
function_declarations = []
for tool in mcp_tools:
parameters = tool.get("parameters", {})
# Convert JSON Schema to Gemini's format
function_declarations.append(
FunctionDeclaration(
name=tool["name"],
description=tool.get("description", ""),
parameters=parameters,
)
)
return [Tool(function_declarations=function_declarations)]
@property
def provider_name(self) -> str:
return "google"
@property
def model_name(self) -> str:
return self.model_name_str
