Add support for Gemini

This adds support for the two types of Gemini endpoints:

- Standard API
- OpenAI-compatible

NOTE: The OpenAI-compatible endpoints are working in both standalone and
muxing scenarios. I'm still trying to get the standard-gemini endpoints
working.

For now, one can test the Gemini integration in Continue as follows:

```
   {
      "title": "Gemini 2.0 Flash",
      "provider": "openai",
      "model": "gemini-2.0-flash",
      "apiBase": "http://127.0.0.1:8989/gemini/v1beta/openai",
      "apiKey": "MY_API_KEY"
    }
```

Signed-off-by: Juan Antonio Osorio <ozz@stacklok.com>

Author: JAORMX

src/codegate/db/models.py

@@ -134,6 +134,7 @@ class ProviderType(str, Enum):
     lm_studio = "lm_studio"
     llamacpp = "llamacpp"
     openrouter = "openrouter"
+    gemini = "gemini"
 
 
 class IntermediatePromptWithOutputUsageAlerts(BaseModel):

src/codegate/muxing/adapter.py

@@ -39,6 +39,9 @@ def _get_provider_formatted_url(self, model_route: rulematcher.ModelRoute) -> st
             return urljoin(model_route.endpoint.endpoint, "/v1")
         if model_route.endpoint.provider_type == db_models.ProviderType.openrouter:
             return urljoin(model_route.endpoint.endpoint, "/api/v1")
+        if model_route.endpoint.provider_type == db_models.ProviderType.gemini:
+            # Gemini API uses /v1beta/openai as the base URL
+            return urljoin(model_route.endpoint.endpoint, "/v1beta/openai")
         return model_route.endpoint.endpoint
 
     def set_destination_info(self, model_route: rulematcher.ModelRoute, data: dict) -> dict:
@@ -209,6 +212,8 @@ def provider_format_funcs(self) -> Dict[str, Callable]:
             db_models.ProviderType.openrouter: self._format_openai,
             # VLLM is a dialect of OpenAI
             db_models.ProviderType.vllm: self._format_openai,
+            # Gemini provider emits OpenAI-compatible chunks
+            db_models.ProviderType.gemini: self._format_openai,
         }
 
     def _format_ollama(self, chunk: str) -> str:
@@ -245,6 +250,8 @@ def provider_format_funcs(self) -> Dict[str, Callable]:
             # VLLM is a dialect of OpenAI
             db_models.ProviderType.vllm: self._format_openai,
             db_models.ProviderType.anthropic: self._format_antropic,
+            # Gemini provider emits OpenAI-compatible chunks
+            db_models.ProviderType.gemini: self._format_openai,
         }
 
     def _format_ollama(self, chunk: str) -> str:

src/codegate/providers/crud/crud.py

@@ -453,6 +453,7 @@ def provider_default_endpoints(provider_type: str) -> str:
     defaults = {
         "openai": "https://api.openai.com",
         "anthropic": "https://api.anthropic.com",
+        "gemini": "https://generativelanguage.googleapis.com",
     }
 
     # If we have a default, we return it

src/codegate/providers/gemini/__init__.py

@@ -0,0 +1,3 @@
+"""
+Gemini provider for CodeGate.
+"""

src/codegate/providers/gemini/adapter.py

@@ -0,0 +1,141 @@
+from typing import Any, Dict, Optional
+
+import structlog
+from litellm import ChatCompletionRequest
+
+from codegate.providers.litellmshim import sse_stream_generator
+from codegate.providers.litellmshim.adapter import (
+    BaseAdapter,
+    LiteLLMAdapterInputNormalizer,
+    LiteLLMAdapterOutputNormalizer,
+)
+
+logger = structlog.get_logger("codegate")
+
+
+class GeminiAdapter(BaseAdapter):
+    """
+    Adapter for Gemini API to translate between Gemini's format and OpenAI's format.
+    """
+
+    def __init__(self) -> None:
+        super().__init__(sse_stream_generator)
+
+    def translate_completion_input_params(self, kwargs) -> Optional[ChatCompletionRequest]:
+        """
+        Translate Gemini API parameters to OpenAI format.
+
+        Gemini API uses a similar format to OpenAI, but with some differences:
+        - 'contents' instead of 'messages'
+        - Different role names
+        - Different parameter names for temperature, etc.
+        """
+        # Make a copy to avoid modifying the original
+        translated_params = dict(kwargs)
+
+        # Handle Gemini-specific parameters
+        if "contents" in translated_params:
+            # Convert Gemini 'contents' to OpenAI 'messages'
+            contents = translated_params.pop("contents")
+            messages = []
+
+            for content in contents:
+                role = content.get("role", "user")
+                # Map Gemini roles to OpenAI roles
+                if role == "model":
+                    role = "assistant"
+
+                message = {
+                    "role": role,
+                    "content": content.get("parts", [{"text": ""}])[0].get("text", ""),
+                }
+                messages.append(message)
+
+            translated_params["messages"] = messages
+
+        # Map other parameters
+        if "temperature" in translated_params:
+            # Temperature is the same in both APIs
+            pass
+
+        if "topP" in translated_params:
+            translated_params["top_p"] = translated_params.pop("topP")
+
+        if "topK" in translated_params:
+            translated_params["top_k"] = translated_params.pop("topK")
+
+        if "maxOutputTokens" in translated_params:
+            translated_params["max_tokens"] = translated_params.pop("maxOutputTokens")
+
+        # Check if we're using the OpenAI-compatible endpoint
+        is_openai_compatible = False
+        if (
+            "_is_openai_compatible" in translated_params
+            and translated_params["_is_openai_compatible"]
+        ):
+            is_openai_compatible = True
+            # Remove the custom field to avoid sending it to the API
+            translated_params.pop("_is_openai_compatible")
+        elif (
+            "base_url" in translated_params
+            and translated_params["base_url"]
+            and "v1beta/openai" in translated_params["base_url"]
+        ):
+            is_openai_compatible = True
+
+        # Apply the appropriate prefix based on the endpoint
+        if "model" in translated_params:
+            model_in_request = translated_params["model"]
+            if is_openai_compatible:
+                # For OpenAI-compatible endpoint, use 'openai/' prefix
+                if not model_in_request.startswith("openai/"):
+                    translated_params["model"] = f"openai/{model_in_request}"
+                    logger.debug(
+                        "Using OpenAI-compatible endpoint, prefixed model name with 'openai/': %s",
+                        translated_params["model"],
+                    )
+            else:
+                # For native Gemini API, use 'gemini/' prefix
+                if not model_in_request.startswith("gemini/"):
+                    translated_params["model"] = f"gemini/{model_in_request}"
+                    logger.debug(
+                        "Using native Gemini API, prefixed model name with 'gemini/': %s",
+                        translated_params["model"],
+                    )
+
+        return ChatCompletionRequest(**translated_params)
+
+    def translate_completion_output_params(self, response: Any) -> Dict:
+        """
+        Translate OpenAI format response to Gemini format.
+        """
+        # For non-streaming responses, we can just return the response as is
+        # LiteLLM should handle the conversion
+        return response
+
+    def translate_completion_output_params_streaming(self, completion_stream: Any) -> Any:
+        """
+        Translate streaming response from OpenAI format to Gemini format.
+        """
+        # For streaming, we can just return the stream as is
+        # The stream generator will handle the conversion
+        return completion_stream
+
+
+class GeminiInputNormalizer(LiteLLMAdapterInputNormalizer):
+    """
+    Normalizer for Gemini API input.
+    """
+
+    def __init__(self):
+        self.adapter = GeminiAdapter()
+        super().__init__(self.adapter)
+
+
+class GeminiOutputNormalizer(LiteLLMAdapterOutputNormalizer):
+    """
+    Normalizer for Gemini API output.
+    """
+
+    def __init__(self):
+        super().__init__(GeminiAdapter())

src/codegate/providers/gemini/completion_handler.py

@@ -0,0 +1,71 @@
+from typing import AsyncIterator, Optional, Union
+
+import structlog
+from litellm import ChatCompletionRequest, ModelResponse
+
+from codegate.providers.litellmshim import LiteLLmShim
+
+logger = structlog.get_logger("codegate")
+
+
+class GeminiCompletion(LiteLLmShim):
+    """
+    GeminiCompletion used by the Gemini provider to execute completions.
+
+    This class extends LiteLLmShim to handle Gemini-specific completion logic.
+    """
+
+    async def execute_completion(
+        self,
+        request: ChatCompletionRequest,
+        base_url: Optional[str],
+        api_key: Optional[str],
+        stream: bool = False,
+        is_fim_request: bool = False,
+    ) -> Union[ModelResponse, AsyncIterator[ModelResponse]]:
+        """
+        Execute the completion request with LiteLLM's API.
+
+        Ensures the model name is prefixed with the appropriate prefix to route to Google's API:
+        - 'openai/' for the OpenAI-compatible endpoint (v1beta/openai)
+        - 'gemini/' for the native Gemini API
+        """
+        model_in_request = request["model"]
+
+        # Check if we're using the OpenAI-compatible endpoint
+        is_openai_compatible = False
+        if "_is_openai_compatible" in request and request["_is_openai_compatible"]:
+            is_openai_compatible = True
+        elif base_url and "v1beta/openai" in base_url:
+            is_openai_compatible = True
+
+        # Apply the appropriate prefix based on the endpoint
+        if is_openai_compatible:
+            # For OpenAI-compatible endpoint, use 'openai/' prefix
+            if not model_in_request.startswith("openai/"):
+                request["model"] = f"openai/{model_in_request}"
+                logger.debug(
+                    "Using OpenAI-compatible endpoint, prefixed model name with 'openai/': %s",
+                    request["model"],
+                )
+        else:
+            # For native Gemini API, use 'gemini/' prefix
+            if not model_in_request.startswith("gemini/"):
+                request["model"] = f"gemini/{model_in_request}"
+                logger.debug(
+                    "Using native Gemini API, prefixed model name with 'gemini/': %s",
+                    request["model"],
+                )
+
+        # Set the API key and base URL
+        request["api_key"] = api_key
+        request["base_url"] = base_url
+
+        # Execute the completion
+        return await super().execute_completion(
+            request=request,
+            api_key=api_key,
+            stream=stream,
+            is_fim_request=is_fim_request,
+            base_url=base_url,
+        )