Google Gemini LLM

​

Overview

​

Installation

uv add "pipecat-ai[google]"

​

Prerequisites

​

Google Gemini Setup

1. Google Account: Sign up at Google AI Studio
2. API Key: Generate a Gemini API key from AI Studio
3. Model Selection: Choose from available Gemini models

​

Required Environment Variables

• GOOGLE_API_KEY: Your Google Gemini API key for authentication

​

Configuration

​

Settings

​

GoogleThinkingConfig

​

Usage

​

Basic Setup

import os

from pipecat.services.google.llm import GoogleLLMService

llm = GoogleLLMService(
    api_key=os.getenv("GOOGLE_API_KEY"),
    settings=GoogleLLMService.Settings(model="gemini-3.5-flash"),
)

​

With Custom Settings

from pipecat.services.google.llm import GoogleLLMService

llm = GoogleLLMService(
    api_key=os.getenv("GOOGLE_API_KEY"),
    settings=GoogleLLMService.Settings(
        model="gemini-2.5-pro",
        system_instruction="You are a helpful assistant.",
        temperature=0.7,
        max_tokens=2048,
        top_p=0.9,
    ),
)

​

With Thinking Configuration

from pipecat.services.google.llm import GoogleLLMService, GoogleThinkingConfig

# Gemini 2.5 series (using thinking_budget)
llm = GoogleLLMService(
    api_key=os.getenv("GOOGLE_API_KEY"),
    settings=GoogleLLMService.Settings(
        model="gemini-2.5-pro",
        max_tokens=8192,
        thinking=GoogleThinkingConfig(
            thinking_budget=4096,
            include_thoughts=True,
        ),
    ),
)

# Gemini 3 series (using thinking_level)
llm = GoogleLLMService(
    api_key=os.getenv("GOOGLE_API_KEY"),
    settings=GoogleLLMService.Settings(
        model="gemini-3.5-flash",
        max_tokens=8192,
        thinking=GoogleThinkingConfig(
            thinking_level="high",
            include_thoughts=True,
        ),
    ),
)

​

With Safety Settings

from google.genai.types import HarmBlockThreshold, HarmCategory, SafetySetting

llm = GoogleLLMService(
    api_key=os.getenv("GOOGLE_API_KEY"),
    settings=GoogleLLMService.Settings(
        model="gemini-3.5-flash",
        safety_settings=[
            SafetySetting(
                category=HarmCategory.HARM_CATEGORY_HATE_SPEECH,
                threshold=HarmBlockThreshold.BLOCK_LOW_AND_ABOVE,
            ),
            SafetySetting(
                category=HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT,
                threshold=HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
            ),
        ],
    ),
)

​

Updating Settings at Runtime

from pipecat.frames.frames import LLMUpdateSettingsFrame
from pipecat.services.google.llm import GoogleLLMSettings

await worker.queue_frame(
    LLMUpdateSettingsFrame(
        delta=GoogleLLMSettings(
            temperature=0.3,
            max_tokens=1024,
        )
    )
)

# Update safety settings (accepts SafetySetting objects or plain dicts)
await worker.queue_frame(
    LLMUpdateSettingsFrame(
        delta=GoogleLLMSettings(
            safety_settings=[
                {
                    "category": "HARM_CATEGORY_HATE_SPEECH",
                    "threshold": "BLOCK_LOW_AND_ABOVE",
                }
            ]
        )
    )
)

​

Notes

• System instruction priority: The system_instruction set via the constructor or GoogleLLMSettings takes priority over any system message in the context. If both are set, a warning is logged and the constructor/settings value is used.
• Thinking defaults: By default, Pipecat sends each listed model its lowest thinking setting to reduce latency (see the note under Settings for the values). Aliases such as gemini-flash-latest aren’t listed, so they run at Google’s default. The same defaults apply on the run_inference() path as in the pipeline. To override this behavior, explicitly pass a GoogleThinkingConfig via settings.
• Response schema: run_inference() accepts an optional response_schema parameter (a JSON schema) that asks Gemini to enforce the reply’s structure. Gemini 2.5 and later models support this. When the schema is provided, the reply is JSON text matching it. Older models (Gemini 2.0 and earlier) ignore the schema with a warning.
• Thinking budget warning: When thinking_budget is set on a Gemini 3 model, the service emits a warning because Gemini 3 models take thinking_level instead. Whether a thinking_budget set alongside one is honored, ignored, or rejected varies by model and backend. Use thinking_level with Gemini 3 models.
• Incomplete responses: When a response ends for a notable reason — withheld for safety, recitation, or prohibited content, a rejected tool call, or the output hitting the token limit — the service logs a warning naming Gemini’s finish_reason, for both streamed responses and run_inference(). Whatever text did arrive is still pushed downstream or returned, so a turn that comes back short has an explanation in the logs rather than only in the transcript.
• Safety settings: Categories left unspecified in safety_settings keep the Gemini API defaults. The setting accepts SafetySetting objects or plain dicts, and is runtime-updatable via LLMUpdateSettingsFrame.
• Seed parameter: Gemini treats the seed parameter as best effort. Identical seeds usually but not always produce identical responses.
• Multimodal support: Gemini models natively support image and audio inputs through Google’s Content/Part format. Images and audio are automatically converted from OpenAI-style contexts. An image_url content part holding a data: URI is inlined; any other URL is passed to Gemini as an external reference, so an image already hosted somewhere doesn’t have to be fetched and re-encoded into the context.
• Grounding with Google Search: When grounding metadata is present in the response (e.g., from Google Search tool), the service emits LLMSearchResponseFrame with search results and source attributions.
• Context format: The service automatically converts OpenAI-style messages to Google’s native Content/Part format. To add a message already in Google’s format, wrap it with llm.create_llm_specific_message(...); a bare Google-format dict in the context is dropped.

​

Event Handlers

@llm.event_handler("on_completion_timeout")
async def on_completion_timeout(service):
    print("LLM completion timed out")