Skip to content

GoogleAIModel

llmfy.llmfy_core.models.google.googleai_model

GoogleAIModel

Bases: BaseAIModel

GoogleAIModel class.

Uses the google-genai package to interact with Google AI (Gemini) models.

Example: 

# Configuration
config = GoogleAIConfig(
        temperature=0.7
)
llm = GoogleAIModel(model="gemini-2.0-flash", config=config)
...

Source code in llmfy/llmfy_core/models/google/googleai_model.py 
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
class GoogleAIModel(BaseAIModel):
    """
    GoogleAIModel class.

    Uses the `google-genai` package to interact with Google AI (Gemini) models.

    Example:
    ```python
    # Configuration
    config = GoogleAIConfig(
            temperature=0.7
    )
    llm = GoogleAIModel(model="gemini-2.0-flash", config=config)
    ...
    ```
    """

    def __init__(self, model: str, config: GoogleAIConfig = GoogleAIConfig()):
        """
        GoogleAIModel

        Args:
            model (str): Model ID (e.g. "gemini-2.0-flash")
            config (GoogleAIConfig, optional): Configuration. Defaults to GoogleAIConfig().
        """
        if genai is None:
            raise LLMfyException(
                'google-genai package is not installed. Install it using `pip install "llmfy[google-genai]"`'
            )

        import os

        if not os.getenv("GOOGLE_API_KEY"):
            raise LLMfyException("Please provide `GOOGLE_API_KEY` on your environment!")

        self.client = genai.Client(api_key=os.getenv("GOOGLE_API_KEY"))
        self.provider = ServiceProvider.GOOGLE
        self.model_name = model
        self.config = config

    def __build_config(self, tools=None, system_instruction=None):
        """Build GenerateContentConfig from self.config."""
        from google.genai import types

        config_kwargs: Dict[str, Any] = {
            "temperature": self.config.temperature,
        }
        if self.config.max_tokens is not None:
            config_kwargs["max_output_tokens"] = self.config.max_tokens
        if self.config.top_p is not None:
            config_kwargs["top_p"] = self.config.top_p
        if self.config.top_k is not None:
            config_kwargs["top_k"] = self.config.top_k
        if self.config.stop_sequences is not None:
            config_kwargs["stop_sequences"] = self.config.stop_sequences
        if self.config.candidate_count is not None:
            config_kwargs["candidate_count"] = self.config.candidate_count
        if self.config.seed is not None:
            config_kwargs["seed"] = self.config.seed
        if self.config.presence_penalty is not None:
            config_kwargs["presence_penalty"] = self.config.presence_penalty
        if self.config.frequency_penalty is not None:
            config_kwargs["frequency_penalty"] = self.config.frequency_penalty
        if self.config.response_mime_type is not None:
            config_kwargs["response_mime_type"] = self.config.response_mime_type
        if self.config.response_schema is not None:
            config_kwargs["response_schema"] = self.config.response_schema
        if self.config.safety_settings is not None:
            config_kwargs["safety_settings"] = self.config.safety_settings
        if self.config.thinking_config is not None:
            config_kwargs["thinking_config"] = self.config.thinking_config
        if system_instruction:
            config_kwargs["system_instruction"] = system_instruction
        if tools:
            config_kwargs["tools"] = [types.Tool(function_declarations=tools)]

        return types.GenerateContentConfig(**config_kwargs)

    def __call_googleai(self, params: dict[str, Any]):
        from google.genai import errors

        from llmfy.exception.exception_handler import handle_google_error
        from llmfy.llmfy_core.models.google.googleai_usage import track_googleai_usage

        @track_googleai_usage
        def _call_googleai_impl(params: dict[str, Any]):
            try:
                response = self.client.models.generate_content(
                    model=params["model"],
                    contents=params["contents"],
                    config=params["config"],
                )
                return response
            except errors.APIError as e:
                raise handle_google_error(e)
            # Any non-APIError exceptions will naturally propagate up the call stack.

        return _call_googleai_impl(params)

    def __call_stream_googleai(self, params: dict[str, Any]):
        from google.genai import errors

        from llmfy.exception.exception_handler import handle_google_error
        from llmfy.llmfy_core.models.google.googleai_usage import (
            track_googleai_stream_usage,
        )

        @track_googleai_stream_usage
        def _call_stream_googleai_impl(params: dict[str, Any]):
            try:
                return self.client.models.generate_content_stream(
                    model=params["model"],
                    contents=params["contents"],
                    config=params["config"],
                )
            except errors.APIError as e:
                raise handle_google_error(e)
            # Any non-APIError exceptions will naturally propagate up the call stack.

        return _call_stream_googleai_impl(params)

    def generate(
        self,
        messages: List[Dict[str, Any]],
        tools: Optional[List[Dict[str, Any]]] = None,
        **kwargs,
    ) -> AIResponse:
        """
        Generate messages.

        Args:
            messages (List[Dict[str, Any]]): Formatted messages from MessageTemp.get_messages().
            tools (Optional[List[Dict[str, Any]]], optional): Tool function definitions. Defaults to None.

        Returns:
            AIResponse: Response with content or tool_calls.
        """
        try:
            system_instruction = next(
                (
                    msg["parts"][0]["text"]
                    for msg in messages
                    if msg.get("role") == "system"
                ),
                None,
            )
            contents = [msg for msg in messages if msg.get("role") != "system"]

            params = {
                "model": self.model_name,
                "contents": contents,
                "config": self.__build_config(
                    tools=tools, system_instruction=system_instruction
                ),
            }

            response = self.__call_googleai(params)

            request_call_id = str(uuid.uuid4())
            tool_calls = None
            content = None

            if response.candidates:
                candidate = response.candidates[0]
                if candidate.content and candidate.content.parts:
                    function_call_parts = [
                        p
                        for p in candidate.content.parts
                        if p.function_call is not None
                    ]
                    if function_call_parts:
                        tool_calls = [
                            ToolCall(
                                request_call_id=request_call_id,
                                tool_call_id=fc.id if fc.id else str(uuid.uuid4()),
                                name=fc.name or "",
                                arguments=dict(fc.args) if fc.args else {},
                            )
                            for part in function_call_parts
                            for fc in [part.function_call]
                            if fc is not None
                        ]
                    else:
                        content = response.text

            return AIResponse(
                content=content,
                tool_calls=tool_calls,
            )

        except Exception as e:
            if isinstance(e, LLMfyException):
                raise  # Already handled, re-raise as-is
            raise LLMfyException(str(e), raw_error=e)

    def generate_stream(
        self,
        messages: List[Dict[str, Any]],
        tools: Optional[List[Dict[str, Any]]] = None,
        **kwargs,
    ) -> Any:
        """
        Generate messages with streaming.

        Note:
            Google AI delivers complete function_call objects in a single chunk
            (no incremental argument accumulation needed, unlike OpenAI).

        Args:
            messages (List[Dict[str, Any]]): Formatted messages from MessageTemp.get_messages().
            tools (Optional[List[Dict[str, Any]]], optional): Tool function definitions. Defaults to None.

        Returns:
            Generator[AIResponse]: Yields AIResponse chunks.
        """
        try:
            system_instruction = next(
                (
                    msg["parts"][0]["text"]
                    for msg in messages
                    if msg.get("role") == "system"
                ),
                None,
            )
            contents = [msg for msg in messages if msg.get("role") != "system"]

            params = {
                "model": self.model_name,
                "contents": contents,
                "config": self.__build_config(
                    tools=tools, system_instruction=system_instruction
                ),
            }

            stream = self.__call_stream_googleai(params)
            request_call_id = str(uuid.uuid4())

            for chunk in stream:
                content = None
                tool_calls = None

                if chunk.candidates:
                    for candidate in chunk.candidates:
                        if candidate.content and candidate.content.parts:
                            for part in candidate.content.parts:
                                if part.text is not None:
                                    content = part.text

                                if part.function_call is not None:
                                    fc = part.function_call
                                    fc_id = fc.id if fc.id else str(uuid.uuid4())
                                    # Google delivers complete function calls in one chunk
                                    tool_calls = [
                                        ToolCall(
                                            request_call_id=request_call_id,
                                            tool_call_id=fc_id,
                                            name=fc.name or "",
                                            arguments=dict(fc.args) if fc.args else {},
                                        )
                                    ]

                yield AIResponse(
                    content=content,
                    tool_calls=tool_calls if tool_calls else None,
                )

        except Exception as e:
            if isinstance(e, LLMfyException):
                raise  # Already handled, re-raise as-is
            raise LLMfyException(str(e), raw_error=e)

client = genai.Client(api_key=(os.getenv('GOOGLE_API_KEY'))) instance-attribute

provider = ServiceProvider.GOOGLE instance-attribute

model_name = model instance-attribute

config = config instance-attribute

__init__(model, config=GoogleAIConfig())

GoogleAIModel

Parameters:

Name Type Description Default
model str

Model ID (e.g. "gemini-2.0-flash")

 required
config GoogleAIConfig

Configuration. Defaults to GoogleAIConfig().

 GoogleAIConfig()
Source code in llmfy/llmfy_core/models/google/googleai_model.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def __init__(self, model: str, config: GoogleAIConfig = GoogleAIConfig()):
    """
    GoogleAIModel

    Args:
        model (str): Model ID (e.g. "gemini-2.0-flash")
        config (GoogleAIConfig, optional): Configuration. Defaults to GoogleAIConfig().
    """
    if genai is None:
        raise LLMfyException(
            'google-genai package is not installed. Install it using `pip install "llmfy[google-genai]"`'
        )

    import os

    if not os.getenv("GOOGLE_API_KEY"):
        raise LLMfyException("Please provide `GOOGLE_API_KEY` on your environment!")

    self.client = genai.Client(api_key=os.getenv("GOOGLE_API_KEY"))
    self.provider = ServiceProvider.GOOGLE
    self.model_name = model
    self.config = config

__build_config(tools=None, system_instruction=None)

Build GenerateContentConfig from self.config.

Source code in llmfy/llmfy_core/models/google/googleai_model.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
def __build_config(self, tools=None, system_instruction=None):
    """Build GenerateContentConfig from self.config."""
    from google.genai import types

    config_kwargs: Dict[str, Any] = {
        "temperature": self.config.temperature,
    }
    if self.config.max_tokens is not None:
        config_kwargs["max_output_tokens"] = self.config.max_tokens
    if self.config.top_p is not None:
        config_kwargs["top_p"] = self.config.top_p
    if self.config.top_k is not None:
        config_kwargs["top_k"] = self.config.top_k
    if self.config.stop_sequences is not None:
        config_kwargs["stop_sequences"] = self.config.stop_sequences
    if self.config.candidate_count is not None:
        config_kwargs["candidate_count"] = self.config.candidate_count
    if self.config.seed is not None:
        config_kwargs["seed"] = self.config.seed
    if self.config.presence_penalty is not None:
        config_kwargs["presence_penalty"] = self.config.presence_penalty
    if self.config.frequency_penalty is not None:
        config_kwargs["frequency_penalty"] = self.config.frequency_penalty
    if self.config.response_mime_type is not None:
        config_kwargs["response_mime_type"] = self.config.response_mime_type
    if self.config.response_schema is not None:
        config_kwargs["response_schema"] = self.config.response_schema
    if self.config.safety_settings is not None:
        config_kwargs["safety_settings"] = self.config.safety_settings
    if self.config.thinking_config is not None:
        config_kwargs["thinking_config"] = self.config.thinking_config
    if system_instruction:
        config_kwargs["system_instruction"] = system_instruction
    if tools:
        config_kwargs["tools"] = [types.Tool(function_declarations=tools)]

    return types.GenerateContentConfig(**config_kwargs)

__call_googleai(params)

Source code in llmfy/llmfy_core/models/google/googleai_model.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def __call_googleai(self, params: dict[str, Any]):
    from google.genai import errors

    from llmfy.exception.exception_handler import handle_google_error
    from llmfy.llmfy_core.models.google.googleai_usage import track_googleai_usage

    @track_googleai_usage
    def _call_googleai_impl(params: dict[str, Any]):
        try:
            response = self.client.models.generate_content(
                model=params["model"],
                contents=params["contents"],
                config=params["config"],
            )
            return response
        except errors.APIError as e:
            raise handle_google_error(e)
        # Any non-APIError exceptions will naturally propagate up the call stack.

    return _call_googleai_impl(params)

__call_stream_googleai(params)

Source code in llmfy/llmfy_core/models/google/googleai_model.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
def __call_stream_googleai(self, params: dict[str, Any]):
    from google.genai import errors

    from llmfy.exception.exception_handler import handle_google_error
    from llmfy.llmfy_core.models.google.googleai_usage import (
        track_googleai_stream_usage,
    )

    @track_googleai_stream_usage
    def _call_stream_googleai_impl(params: dict[str, Any]):
        try:
            return self.client.models.generate_content_stream(
                model=params["model"],
                contents=params["contents"],
                config=params["config"],
            )
        except errors.APIError as e:
            raise handle_google_error(e)
        # Any non-APIError exceptions will naturally propagate up the call stack.

    return _call_stream_googleai_impl(params)

generate(messages, tools=None, **kwargs)

Generate messages.

Parameters:

Name Type Description Default
messages List[Dict[str, Any]]

Formatted messages from MessageTemp.get_messages().

 required
tools Optional[List[Dict[str, Any]]]

Tool function definitions. Defaults to None.

 None

Returns:

Name Type Description
AIResponse AIResponse

Response with content or tool_calls.
Source code in llmfy/llmfy_core/models/google/googleai_model.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
def generate(
    self,
    messages: List[Dict[str, Any]],
    tools: Optional[List[Dict[str, Any]]] = None,
    **kwargs,
) -> AIResponse:
    """
    Generate messages.

    Args:
        messages (List[Dict[str, Any]]): Formatted messages from MessageTemp.get_messages().
        tools (Optional[List[Dict[str, Any]]], optional): Tool function definitions. Defaults to None.

    Returns:
        AIResponse: Response with content or tool_calls.
    """
    try:
        system_instruction = next(
            (
                msg["parts"][0]["text"]
                for msg in messages
                if msg.get("role") == "system"
            ),
            None,
        )
        contents = [msg for msg in messages if msg.get("role") != "system"]

        params = {
            "model": self.model_name,
            "contents": contents,
            "config": self.__build_config(
                tools=tools, system_instruction=system_instruction
            ),
        }

        response = self.__call_googleai(params)

        request_call_id = str(uuid.uuid4())
        tool_calls = None
        content = None

        if response.candidates:
            candidate = response.candidates[0]
            if candidate.content and candidate.content.parts:
                function_call_parts = [
                    p
                    for p in candidate.content.parts
                    if p.function_call is not None
                ]
                if function_call_parts:
                    tool_calls = [
                        ToolCall(
                            request_call_id=request_call_id,
                            tool_call_id=fc.id if fc.id else str(uuid.uuid4()),
                            name=fc.name or "",
                            arguments=dict(fc.args) if fc.args else {},
                        )
                        for part in function_call_parts
                        for fc in [part.function_call]
                        if fc is not None
                    ]
                else:
                    content = response.text

        return AIResponse(
            content=content,
            tool_calls=tool_calls,
        )

    except Exception as e:
        if isinstance(e, LLMfyException):
            raise  # Already handled, re-raise as-is
        raise LLMfyException(str(e), raw_error=e)

generate_stream(messages, tools=None, **kwargs)

Generate messages with streaming.

Note

Google AI delivers complete function_call objects in a single chunk (no incremental argument accumulation needed, unlike OpenAI).

Parameters:

Name Type Description Default
messages List[Dict[str, Any]]

Formatted messages from MessageTemp.get_messages().

 required
tools Optional[List[Dict[str, Any]]]

Tool function definitions. Defaults to None.

 None

Returns:

Type Description
Any

Generator[AIResponse]: Yields AIResponse chunks.
Source code in llmfy/llmfy_core/models/google/googleai_model.py 
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
def generate_stream(
    self,
    messages: List[Dict[str, Any]],
    tools: Optional[List[Dict[str, Any]]] = None,
    **kwargs,
) -> Any:
    """
    Generate messages with streaming.

    Note:
        Google AI delivers complete function_call objects in a single chunk
        (no incremental argument accumulation needed, unlike OpenAI).

    Args:
        messages (List[Dict[str, Any]]): Formatted messages from MessageTemp.get_messages().
        tools (Optional[List[Dict[str, Any]]], optional): Tool function definitions. Defaults to None.

    Returns:
        Generator[AIResponse]: Yields AIResponse chunks.
    """
    try:
        system_instruction = next(
            (
                msg["parts"][0]["text"]
                for msg in messages
                if msg.get("role") == "system"
            ),
            None,
        )
        contents = [msg for msg in messages if msg.get("role") != "system"]

        params = {
            "model": self.model_name,
            "contents": contents,
            "config": self.__build_config(
                tools=tools, system_instruction=system_instruction
            ),
        }

        stream = self.__call_stream_googleai(params)
        request_call_id = str(uuid.uuid4())

        for chunk in stream:
            content = None
            tool_calls = None

            if chunk.candidates:
                for candidate in chunk.candidates:
                    if candidate.content and candidate.content.parts:
                        for part in candidate.content.parts:
                            if part.text is not None:
                                content = part.text

                            if part.function_call is not None:
                                fc = part.function_call
                                fc_id = fc.id if fc.id else str(uuid.uuid4())
                                # Google delivers complete function calls in one chunk
                                tool_calls = [
                                    ToolCall(
                                        request_call_id=request_call_id,
                                        tool_call_id=fc_id,
                                        name=fc.name or "",
                                        arguments=dict(fc.args) if fc.args else {},
                                    )
                                ]

            yield AIResponse(
                content=content,
                tool_calls=tool_calls if tool_calls else None,
            )

    except Exception as e:
        if isinstance(e, LLMfyException):
            raise  # Already handled, re-raise as-is
        raise LLMfyException(str(e), raw_error=e)