GeminiInterface

llm.GeminiInterface — Google Gemini interface with built-in rate limiting and retry logic.

This is the most defensively-written interface in the package: it includes hardcoded request throttling and multi-attempt retry/backoff logic that the other providers don't have, aimed specifically at Gemini's free-tier rate limits.

Constructor

GeminiInterface(
    model_name: str = "gemini-3-flash-preview",  # from llm_config.gemini_model
    api_key: Optional[str] = None,
    **kwargs,
)

Imports from google import genai and creates genai.Client(api_key=self.api_key). Raises ImportError with the message "Google Gemini Not Found!\nInstall: pip install google-genai" if the package is missing.

Also sets up, at construction time:

self.default_config = {
    "temperature": config.temperature,   # 0.3
    "top_p": config.top_p,               # 0.9
    "top_k": config.top_k,               # 50
    "max_output_tokens": config.max_token,  # 2048
}
self._min_request_interval = 5.0   # seconds
self._last_request_time = 0.0

Rate limiting

Before every generate or generate_async call, the interface checks how long it's been since _last_request_time and sleeps (time.sleep in generate, await asyncio.sleep in generate_async) to enforce a minimum 5-second gap between requests:

elapsed = time.time() - self._last_request_time
if elapsed < self._min_request_interval:
    time.sleep(self._min_request_interval - elapsed)
self._last_request_time = time.time()

This targets Gemini's free-tier limit of ~15 requests/minute (5s gaps ≈ 12 req/min, kept conservatively under the limit). This means every call to generate/generate_async on a GeminiInterface instance can block for up to 5 seconds even before the network request is made, if called in quick succession. There is no constructor parameter to change _min_request_interval — it's a hardcoded literal, not read from llm_config.

Message building: _build_contents(prompt, messages)

@staticmethod
def _build_contents(prompt, messages):
    if messages is None:
        if prompt is None:
            raise ValueError("Either 'prompt' or 'messages' must be provided.")
        return prompt, None

    system_parts = []
    contents = []
    for m in messages:
        role = m.get("role", "user")
        text = m.get("content", "")
        if role == "system":
            system_parts.append(text)
            continue
        gemini_role = "model" if role == "assistant" else "user"
        contents.append({"role": gemini_role, "parts": [{"text": text}]})

    system_instruction = "\n".join(system_parts) if system_parts else None
    return contents, system_instruction

Resolves prompt/messages into the (contents, system_instruction) pair the Gemini SDK expects. Notable behavior, distinct from OpenAI/Anthropic/Groq/AWS's _build_messages:

• Role translation: Gemini has no assistant role — any message with role="assistant" is sent as role="model". Any other non-"system" role (including a typo'd or unrecognized role) falls through to role="user".
• System messages are extracted, not left in the turn history: every role="system" message's content is collected and joined with \n into a single system_instruction string, passed separately to client.models.generate_content(..., config=generation_config) (merged into generation_config["system_instruction"]) rather than appearing as a turn in contents. If there are no system messages, system_instruction is None and the key is omitted from generation_config entirely.
• prompt-only calls bypass this shape entirely: when only prompt is given, _build_contents returns the raw string as contents (not wrapped in a {"role": ..., "parts": [...]} dict) — this matches what the Gemini SDK accepts directly for a single-turn call.
• When messages is None and prompt is also None, this raises ValueError immediately.

generate(messages=None, prompt=None, max_tokens=2048, temperature=None, tools=None, tool_choice=None, **kwargs) -> Union[str, dict]

Parameters

Raises: ValueError("Either 'prompt' or 'messages' must be provided.") if both are omitted. Unlike generate_async below (and unlike this same check on OpenAI/Anthropic/Groq/Mistral), the _build_contents call in this synchronous generate happens before the retry loop's try/except, so this ValueError propagates straight to the caller uncaught rather than being converted into an error string.

Retry behavior

Up to 5 attempts (max_retries = 5), with exponential backoff plus jitter (base_delay * (2 ** attempt) + random.uniform(0, 1.5)), specifically for errors whose message contains "503", "UNAVAILABLE", or "429". Other exceptions return immediately as f"Error: {error_msg}" without retrying. Additionally, if a successful plain-text response's text is under 3 characters and attempts remain, the method retries once more (short-but-valid answers like "4" or "Berlin" are not retried once fewer than 3 characters is no longer true relative to this specific check — the check is exactly len(result) < 3, so 1–2 character results, even if fully correct, will trigger a retry attempt if attempts remain). This length-based retry only applies to the plain-text path — a tool-call response is returned immediately on the first successful attempt regardless of how short (or absent) its accompanying text is, since parse_gemini_tool_calls(response) is checked, and returned on if non-empty, before the text-length check is ever reached.

Returns: If the model doesn't call a tool, extracted text (joined from response.candidates[*].content.parts[*].text, falling back to response.text if no candidates/parts are present), or "" if no text was ever produced after retries, or f"Error: {error_msg}" / "❌ Error: All attempts failed with Gemini" on failure paths — the no-tool-call return type is always str here (unlike OpenAI/Anthropic/Groq/AWS, which return a dict on error) — error conditions are represented as error-prefixed strings, not a distinguishable error object. If the model calls one or more tools, this is the one case where GeminiInterface.generate returns a dict: parse_gemini_tool_calls(response) scans every candidate's content parts for a function_call (coercing .args to a plain dict; id is always None, since Gemini function calls don't carry a call id), and the method returns {"content": <any accompanying text or None>, "tool_calls": [...], "stop_reason": <first candidate's finish_reason, or None>} — this dict is returned on the first successful attempt with a tool call, bypassing both the retry-on-short-text logic and the "always str" behavior described above. See Tool Calling.

Side effects: prints progress/warning messages to stdout during retries (e.g. "⚠️ Response too short...", "⚠️ Gemini error (...)") — this interface uses print() for these rather than the logging module used by OpenAI/Anthropic/Groq/AWS.

Example

from zerda.llm import GeminiInterface

llm = GeminiInterface(api_key="AIza...")
print(llm.generate(prompt="اشرح الفرق بين RAG و fine-tuning."))

# multi-turn, with a system message
print(llm.generate(messages=[
    {"role": "system", "content": "Answer in one sentence."},
    {"role": "user", "content": "What is RAG?"},
]))

# tool calling
tools = [{"name": "get_weather", "description": "...", "input_schema": {"type": "object", "properties": {"city": {"type": "string"}}}}]
result = llm.generate(prompt="Weather in Cairo?", tools=tools)  # -> dict if the model calls the tool

generate_async(messages=None, prompt=None, max_tokens=None, temperature=None, tools=None, tool_choice=None, **kwargs) -> Union[str, dict]

Applies the same 5-second rate-limit gate (via await asyncio.sleep), then runs the synchronous client.models.generate_content call inside loop.run_in_executor(None, ...) — i.e., this is a thread-pool-offloaded sync call rather than a call to a native async Gemini client method. Unlike generate, this method does not implement the retry/backoff logic — a single attempt only; on exception it returns f"Error: {str(e)}" immediately. tools/tool_choice are handled the same way as in generate (merged into generation_config before the call), and a tool-call response likewise returns the same {"content", "tool_calls", "stop_reason"} dict, checked before the plain-text extraction path.

Unlike the synchronous generate, the _build_contents(prompt, messages) call here sits inside this method's own try/except Exception, so a missing prompt/messages does not raise — it's caught and returned as the string "Error: Either 'prompt' or 'messages' must be provided.", same as any other exception on this path. This is an intentional-looking but easy-to-miss asymmetry between Gemini's sync and async generate methods; write code that checks the return value rather than assuming both paths raise the same way.

astream(messages=None, prompt=None, max_tokens=None, temperature=None, **kwargs) -> AsyncIterator[str]

response = await self.generate_async(messages=messages, prompt=prompt, max_tokens=max_tokens, temperature=temperature, **kwargs)
if isinstance(response, dict):
    response = response.get("error", "")
for word in response.split(" "):
    yield word + " "
    await asyncio.sleep(0)

Same word-split fallback pattern as BaseLLMInterface's default — Gemini's SDK is not used for true incremental streaming here (per the source comment, "Gemini SDK may not support true streaming yet"). Since it delegates to generate_async, a missing prompt/messages here is likewise reported by yielding the word-split "Error: ..." string, not by raising.

astream does not accept tools/tool_choice as named parameters. If a tools= kwarg is passed through anyway, generate_async would return the unified tool-call dict instead of a string — but astream's fallback line only special-cases the "error" key (response.get("error", "")), so a {"content", "tool_calls", "stop_reason"} dict without an "error" key falls through to response.get("error", "") returning "", and the method would then yield nothing at all rather than surfacing the tool call. Do not pass tools to astream() on GeminiInterface — use generate/generate_async instead. See Tool Calling § Streaming limitation.

Thread safety

_last_request_time is a plain instance attribute mutated without locking — concurrent calls on the same instance from multiple threads or concurrently-scheduled async tasks could race on this check-then-sleep sequence, potentially allowing more than one request through within the intended 5-second window. Not documented as thread-safe.

Related APIs

• BaseLLMInterface — astream's fallback pattern originates here.
• Configuration — for how default_config values trace back to llm_config.
• Tool Calling — the unified tools/tool_choice contract; note Gemini's astream incompatibility flagged above.