windlass.providers.llm.gemini

gemini

Google Gemini adapter.

Gemini's API names things differently from everyone else: turns are contents with parts, the assistant role is model, tools are function_declarations grouped under a Tool, and generation options live in a nested generation_config. This adapter absorbs all of that.

Install with::

pip install "windlass[gemini]"

Example

from windlass import Windlass # doctest: +SKIP Windlass.llm("gemini", model="gemini-2.0-flash") # doctest: +SKIP

GeminiLLM

GeminiLLM(
    model: str = "",
    *,
    api_key: str | None = None,
    vertexai: bool = False,
    project: str | None = None,
    location: str | None = None,
    **config: Any
)

Bases: LLM

Chat completions via the google-genai SDK.

Parameters:

Name	Type	Description	Default
model	str	Model id, e.g. gemini-2.0-flash.	''
api_key	str | None	Credential. Falls back to GOOGLE_API_KEY / GEMINI_API_KEY.	None
vertexai	bool	Route through Vertex AI instead of the Gemini Developer API.	False
project	str | None	GCP project id (Vertex only).	None
location	str | None	GCP region (Vertex only).	None
**config	Any	Forwarded to :class:~windlass.interfaces.llm.LLM.	{}

Raises:

Type	Description
MissingDependencyError	When google-genai is not installed.
AuthenticationError	When no API key can be found and Vertex is off.

Source code in src\windlass\providers\llm\gemini.py

def __init__(
    self,
    model: str = "",
    *,
    api_key: str | None = None,
    vertexai: bool = False,
    project: str | None = None,
    location: str | None = None,
    **config: Any,
) -> None:
    super().__init__(model=model, **config)
    genai = require("google.genai", extra="gemini", feature="The Gemini provider")
    self._genai = genai
    self._types = require("google.genai.types", extra="gemini", feature="The Gemini provider")
    if vertexai:
        self._client = genai.Client(vertexai=True, project=project, location=location)
    else:
        key = api_key or settings().require_secret(
            "google_api_key", provider="gemini", env_var="GOOGLE_API_KEY"
        )
        self._client = genai.Client(api_key=key)

default_model classmethod

default_model() -> str

Return "gemini-2.0-flash".

Source code in src\windlass\providers\llm\gemini.py

@classmethod
def default_model(cls) -> str:
    """Return ``"gemini-2.0-flash"``."""
    return "gemini-2.0-flash"

native

native() -> Any

Return the underlying google.genai.Client.

Source code in src\windlass\providers\llm\gemini.py

def native(self) -> Any:
    """Return the underlying ``google.genai.Client``."""
    return self._client

agenerate async

agenerate(
    messages: list[Message], *, tools: list[dict[str, Any]] | None = None, **kwargs: Any
) -> Completion

Request one generation.

Parameters:

Name	Type	Description	Default
messages	list[Message]	The conversation.	required
tools	list[dict[str, Any]] | None	Gemini-format function declarations.	None
**kwargs	Any	Request overrides.	{}

Returns:

Type	Description
Completion	The completion.

Raises:

Type	Description
ProviderError	For any API failure.

Source code in src\windlass\providers\llm\gemini.py

async def agenerate(
    self,
    messages: list[Message],
    *,
    tools: list[dict[str, Any]] | None = None,
    **kwargs: Any,
) -> Completion:
    """Request one generation.

    Args:
        messages: The conversation.
        tools: Gemini-format function declarations.
        **kwargs: Request overrides.

    Returns:
        The completion.

    Raises:
        ProviderError: For any API failure.
    """
    system, contents = to_gemini_contents(messages)
    config = self._config(system, tools, kwargs)
    try:
        response = await self._client.aio.models.generate_content(
            model=self.model, contents=contents, config=config
        )
    except Exception as exc:
        raise self._translate(exc) from exc

    text_parts: list[str] = []
    calls: list[ToolCall] = []
    for candidate in getattr(response, "candidates", None) or []:
        for part in getattr(candidate.content, "parts", None) or []:
            if getattr(part, "text", None):
                text_parts.append(part.text)
            fn = getattr(part, "function_call", None)
            if fn is not None:
                calls.append(ToolCall(name=fn.name, arguments=dict(fn.args or {})))

    finish = None
    if getattr(response, "candidates", None):
        finish = getattr(response.candidates[0], "finish_reason", None)

    return Completion(
        content="".join(text_parts),
        tool_calls=calls,
        finish_reason=self._finish_reason(str(finish) if finish else None),
        model=self.model,
        usage=self._usage_from(response),
        raw=response,
    )

astream_generate async

astream_generate(
    messages: list[Message], *, tools: list[dict[str, Any]] | None = None, **kwargs: Any
) -> AsyncIterator[StreamEvent]

Stream a generation.

Parameters:

Name	Type	Description	Default
messages	list[Message]	The conversation.	required
tools	list[dict[str, Any]] | None	Gemini-format function declarations.	None
**kwargs	Any	Request overrides.	{}

Yields:

Type	Description
AsyncIterator[StreamEvent]	Text deltas, then tool calls, then done.

Source code in src\windlass\providers\llm\gemini.py

async def astream_generate(
    self,
    messages: list[Message],
    *,
    tools: list[dict[str, Any]] | None = None,
    **kwargs: Any,
) -> AsyncIterator[StreamEvent]:
    """Stream a generation.

    Args:
        messages: The conversation.
        tools: Gemini-format function declarations.
        **kwargs: Request overrides.

    Yields:
        Text deltas, then tool calls, then ``done``.
    """
    system, contents = to_gemini_contents(messages)
    config = self._config(system, tools, kwargs)
    calls: list[ToolCall] = []
    usage = Usage()

    try:
        stream = await self._client.aio.models.generate_content_stream(
            model=self.model, contents=contents, config=config
        )
        async for chunk in stream:
            for candidate in getattr(chunk, "candidates", None) or []:
                for part in getattr(candidate.content, "parts", None) or []:
                    if getattr(part, "text", None):
                        yield StreamEvent(type="text", delta=part.text, raw=chunk)
                    fn = getattr(part, "function_call", None)
                    if fn is not None:
                        calls.append(ToolCall(name=fn.name, arguments=dict(fn.args or {})))
            if getattr(chunk, "usage_metadata", None):
                usage = self._usage_from(chunk)
    except Exception as exc:
        raise self._translate(exc) from exc

    for call in calls:
        yield StreamEvent(type="tool_call", tool_call=call)
    yield StreamEvent(type="done", usage=usage)

to_gemini_contents

to_gemini_contents(messages: list[Message]) -> tuple[str, list[dict[str, Any]]]

Translate Windlass messages into Gemini's contents format.

Parameters:

Name	Type	Description	Default
messages	list[Message]	The conversation.	required

Returns:

Type	Description
str	A (system_instruction, contents) pair. The assistant role is
list[dict[str, Any]]	renamed to model and tool results become function_response parts.

Example

from windlass.core.types import Message system, contents = to_gemini_contents([Message.user("hi")]) contents[0]["role"] 'user'

Source code in src\windlass\providers\llm\gemini.py

def to_gemini_contents(messages: list[Message]) -> tuple[str, list[dict[str, Any]]]:
    """Translate Windlass messages into Gemini's ``contents`` format.

    Args:
        messages: The conversation.

    Returns:
        A ``(system_instruction, contents)`` pair. The assistant role is
        renamed to ``model`` and tool results become ``function_response`` parts.

    Example:
        >>> from windlass.core.types import Message
        >>> system, contents = to_gemini_contents([Message.user("hi")])
        >>> contents[0]["role"]
        'user'
    """
    system_parts: list[str] = []
    contents: list[dict[str, Any]] = []

    for message in messages:
        role = message.role.value
        if role == "system":
            system_parts.append(message.content)
            continue
        if role == "tool":
            contents.append(
                {
                    "role": "user",
                    "parts": [
                        {
                            "function_response": {
                                "name": message.name or "tool",
                                "response": {"result": message.content},
                            }
                        }
                    ],
                }
            )
            continue

        parts: list[dict[str, Any]] = []
        if message.content:
            parts.append({"text": message.content})
        for call in message.tool_calls:
            parts.append({"function_call": {"name": call.name, "args": call.arguments}})
        if parts:
            contents.append({"role": "model" if role == "assistant" else "user", "parts": parts})

    return "\n\n".join(p for p in system_parts if p), contents