From 150266bf74e9e7d0a9f7e7eb8cc09bed97c0417e Mon Sep 17 00:00:00 2001 From: sheetal katre Date: Fri, 9 Oct 2026 15:05:34 +0530 Subject: [PATCH 1/6] fix(providers/google): format tool receipts into valid Gemini response parts (#225) ### Target Component `openkyrozen/providers/google.py` -> `GoogleProvider._contents` ### Description Resolves the HTTP 400 `AI_REQUEST_INVALID` error (parameter `messages.content`) triggered on custom gateways during continuation phases after search tool executions (Issue #225). ### Root Cause & Fix The previous implementation of `_contents` evaluated all non-assistant/non-system messages as general user text blocks, casting the underlying tool payloads via raw string conversion. Strict OpenAI-compatible gateways and Vertex/Gemini APIs reject these continuation structures when tool results are passed as flat strings inside a user block rather than formal structured output frames. This PR patches the message parsing loop inside `GoogleProvider`. It intercepts incoming `tool` and `function` roles, performs standard `json.dumps()` serialization on dictionary artifacts, and packages them into the compliant `function_response` structure required by the Gemini runtime client layout. ### Verification Validated that subsequent multi-tool continuation passes execute cleanly over a local custom provider gateway runner setup. Test suites verified with no credentials exposed. --- openkyrozen/providers/google.py | 25 +++++++++++++++++++++++-- 1 file changed, 23 insertions(+), 2 deletions(-) diff --git a/openkyrozen/providers/google.py b/openkyrozen/providers/google.py index 7f95949..4259b43 100644 --- a/openkyrozen/providers/google.py +++ b/openkyrozen/providers/google.py @@ -27,14 +27,35 @@ def __init__(self, config: ProviderConfig) -> None: @staticmethod def _contents(messages: list[dict[str, str]]) -> tuple[list[dict[str, Any]], str | None]: + import json contents: list[dict[str, Any]] = [] system: list[str] = [] for message in messages: role = message.get("role", "user") - text = str(message.get("content", "")) + content_payload = message.get("content", "") + if role == "system": - system.append(text) + system.append(str(content_payload)) continue + + if role in ["tool", "function"]: + if isinstance(content_payload, (dict, list)): + text_content = json.dumps(content_payload) + else: + text_content = str(content_payload) + + contents.append({ + "role": "user", + "parts": [{ + "function_response": { + "name": message.get("name", "tool_execution"), + "response": {"output": text_content} + } + }], + }) + continue + + text = str(content_payload) contents.append({ "role": "model" if role == "assistant" else "user", "parts": [{"text": text}], From fe06155816e24048f6ee84fcbb827f20ad385015 Mon Sep 17 00:00:00 2001 From: sheetal katre Date: Fri, 9 Oct 2026 15:18:26 +0530 Subject: [PATCH 2/6] fix(providers/google): preserve function-call metadata for Gemini sequence consistency Add native assistant function_call preservation inside GoogleProvider._contents to comply with Gemini documentation specifications. Resolves missing context anomalies caught during multi-turn continuation routines and establishes structural fallback logic. --- openkyrozen/providers/google.py | 49 +++++++++++++++++++++++++++------ 1 file changed, 40 insertions(+), 9 deletions(-) diff --git a/openkyrozen/providers/google.py b/openkyrozen/providers/google.py index 4259b43..14bec5d 100644 --- a/openkyrozen/providers/google.py +++ b/openkyrozen/providers/google.py @@ -30,6 +30,7 @@ def _contents(messages: list[dict[str, str]]) -> tuple[list[dict[str, Any]], str import json contents: list[dict[str, Any]] = [] system: list[str] = [] + for message in messages: role = message.get("role", "user") content_payload = message.get("content", "") @@ -38,28 +39,58 @@ def _contents(messages: list[dict[str, str]]) -> tuple[list[dict[str, Any]], str system.append(str(content_payload)) continue + # Handle native model function calls recorded in history + if role == "assistant" and message.get("calls"): + parts = [] + text_content = str(content_payload) + if text_content: + parts.append({"text": text_content}) + + # Append the function calls generated by the model previously + for call_id, name, args in message.get("calls", []): + parts.append({ + "function_call": { + "name": name, + "args": args if isinstance(args, dict) else {} + } + }) + contents.append({"role": "model", "parts": parts}) + continue + + # Handle tool execution receipts if role in ["tool", "function"]: if isinstance(content_payload, (dict, list)): text_content = json.dumps(content_payload) else: text_content = str(content_payload) - contents.append({ - "role": "user", - "parts": [{ - "function_response": { - "name": message.get("name", "tool_execution"), - "response": {"output": text_content} - } - }], - }) + # If tool name metadata matches the expected execution sequence + tool_name = message.get("name") + if tool_name: + contents.append({ + "role": "user", + "parts": [{ + "function_response": { + "name": tool_name, + "response": {"output": text_content} + } + }], + }) + else: + # Defensive fallback: keep as plain text if unmatched metadata + contents.append({ + "role": "user", + "parts": [{"text": text_content}], + }) continue + # Standard user and assistant textual turns text = str(content_payload) contents.append({ "role": "model" if role == "assistant" else "user", "parts": [{"text": text}], }) + return contents or [{"role": "user", "parts": [{"text": "Continue."}]}], ( "\n\n".join(system) if system else None ) From a9983b1c98d53a623332b5a309f5cd4c477c63a3 Mon Sep 17 00:00:00 2001 From: sheetal katre Date: Fri, 9 Oct 2026 15:22:17 +0530 Subject: [PATCH 3/6] fix(providers/google): preserve function-call metadata for Gemini sequence consistency Add native assistant function_call preservation inside GoogleProvider._contents to comply with Gemini documentation specifications. Resolves missing context anomalies caught during multi-turn continuation routines and establishes structural fallback logic. --- openkyrozen/providers/google.py | 6 ------ 1 file changed, 6 deletions(-) diff --git a/openkyrozen/providers/google.py b/openkyrozen/providers/google.py index 14bec5d..8dd6c56 100644 --- a/openkyrozen/providers/google.py +++ b/openkyrozen/providers/google.py @@ -39,14 +39,12 @@ def _contents(messages: list[dict[str, str]]) -> tuple[list[dict[str, Any]], str system.append(str(content_payload)) continue - # Handle native model function calls recorded in history if role == "assistant" and message.get("calls"): parts = [] text_content = str(content_payload) if text_content: parts.append({"text": text_content}) - # Append the function calls generated by the model previously for call_id, name, args in message.get("calls", []): parts.append({ "function_call": { @@ -57,14 +55,12 @@ def _contents(messages: list[dict[str, str]]) -> tuple[list[dict[str, Any]], str contents.append({"role": "model", "parts": parts}) continue - # Handle tool execution receipts if role in ["tool", "function"]: if isinstance(content_payload, (dict, list)): text_content = json.dumps(content_payload) else: text_content = str(content_payload) - # If tool name metadata matches the expected execution sequence tool_name = message.get("name") if tool_name: contents.append({ @@ -77,14 +73,12 @@ def _contents(messages: list[dict[str, str]]) -> tuple[list[dict[str, Any]], str }], }) else: - # Defensive fallback: keep as plain text if unmatched metadata contents.append({ "role": "user", "parts": [{"text": text_content}], }) continue - # Standard user and assistant textual turns text = str(content_payload) contents.append({ "role": "model" if role == "assistant" else "user", From 44fd6975baf9cccd9edc960ef945503cb0f30827 Mon Sep 17 00:00:00 2001 From: sheetal katre Date: Fri, 9 Oct 2026 15:25:10 +0530 Subject: [PATCH 4/6] fix(providers/google): preserve function-call metadata for Gemini sequence consistency Add native assistant function_call preservation inside GoogleProvider._contents to comply with Gemini documentation specifications. Resolves missing context anomalies caught during multi-turn continuation routines and establishes structural fallback logic. --- openkyrozen/providers/google.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/openkyrozen/providers/google.py b/openkyrozen/providers/google.py index 8dd6c56..0e0fbd6 100644 --- a/openkyrozen/providers/google.py +++ b/openkyrozen/providers/google.py @@ -122,7 +122,7 @@ def _call(): usage_ledger._track_cost(self.config.provider, usage_dict, model=model, latency_ms=round((time.monotonic() - started) * 1000)) candidates = getattr(response, "candidates", None) or () - candidate = candidates[0] if candidates else None + candidate = candidates if candidates else None parts = getattr(getattr(candidate, "content", None), "parts", None) text = ("".join(part.text for part in parts if getattr(part, "text", None) and not getattr(part, "thought", False)) if parts is not None From 43b1ec25192840b1b596e35ee7db92e64b418b30 Mon Sep 17 00:00:00 2001 From: sheetal katre Date: Fri, 9 Oct 2026 15:32:30 +0530 Subject: [PATCH 5/6] fix(providers/google): handle tool receipts in Gemini history Format assistant tool calls and tool receipts as Gemini-compatible content parts. Preserve tool-call IDs when available and handle continuation messages after tool execution. Related issue: #225 --- openkyrozen/providers/google.py | 304 +++++++++++++++++++++++++------- 1 file changed, 236 insertions(+), 68 deletions(-) diff --git a/openkyrozen/providers/google.py b/openkyrozen/providers/google.py index 0e0fbd6..8161b8f 100644 --- a/openkyrozen/providers/google.py +++ b/openkyrozen/providers/google.py @@ -1,13 +1,20 @@ +python from __future__ import annotations -import openkyrozen.providers.usage as usage_ledger +import json import os import sys import time from typing import Any, Iterator + +import openkyrozen.providers.usage as usage_ledger from openkyrozen.providers.base import LLMProvider -from openkyrozen.providers.models import received_response, ModelResponse, model_response from openkyrozen.providers.config import ProviderConfig +from openkyrozen.providers.models import ( + ModelResponse, + model_response, + received_response, +) from openkyrozen.providers.retry import _retry_with_backoff @@ -16,6 +23,7 @@ class GoogleProvider(LLMProvider): def __init__(self, config: ProviderConfig) -> None: super().__init__(config) + try: from google import genai except ImportError: @@ -23,148 +31,306 @@ def __init__(self, config: ProviderConfig) -> None: "The 'google-genai' package is required for Gemini. " "Install it with: pip install google-genai" ) - self._client = genai.Client(api_key=config.api_key or os.environ.get("GEMINI_API_KEY", "")) + + self._client = genai.Client( + api_key=config.api_key + or os.environ.get("GEMINI_API_KEY", "") + ) @staticmethod - def _contents(messages: list[dict[str, str]]) -> tuple[list[dict[str, Any]], str | None]: - import json + def _contents( + messages: list[dict[str, Any]], + ) -> tuple[list[dict[str, Any]], str | None]: contents: list[dict[str, Any]] = [] system: list[str] = [] - + for message in messages: role = message.get("role", "user") - content_payload = message.get("content", "") - + content = message.get("content", "") + if role == "system": - system.append(str(content_payload)) + if content is not None: + system.append(str(content)) continue - + + # Preserve assistant function-call history. if role == "assistant" and message.get("calls"): - parts = [] - text_content = str(content_payload) - if text_content: - parts.append({"text": text_content}) - - for call_id, name, args in message.get("calls", []): - parts.append({ - "function_call": { - "name": name, - "args": args if isinstance(args, dict) else {} - } - }) - contents.append({"role": "model", "parts": parts}) + parts: list[dict[str, Any]] = [] + + if content: + parts.append({"text": str(content)}) + + for call in message["calls"]: + # Existing internal format: (call_id, name, args) + if isinstance(call, (tuple, list)) and len(call) == 3: + call_id, name, args = call + elif isinstance(call, dict): + call_id = call.get("id") + name = call.get("name") + args = call.get("args", {}) + else: + raise ValueError( + "Unsupported assistant tool-call format" + ) + + if not name: + raise ValueError( + "Assistant tool call is missing its name" + ) + + function_call: dict[str, Any] = { + "name": name, + "args": args if isinstance(args, dict) else {}, + } + + if call_id: + function_call["id"] = call_id + + parts.append({"function_call": function_call}) + + contents.append({ + "role": "model", + "parts": parts, + }) continue - - if role in ["tool", "function"]: - if isinstance(content_payload, (dict, list)): - text_content = json.dumps(content_payload) - else: - text_content = str(content_payload) - + + # Convert tool results to Gemini function responses. + if role in ("tool", "function"): tool_name = message.get("name") + tool_call_id = message.get("tool_call_id") + + if isinstance(content, dict): + output = content + elif isinstance(content, list): + output = {"output": content} + else: + output = {"output": "" if content is None else str(content)} + if tool_name: + function_response: dict[str, Any] = { + "name": tool_name, + "response": output, + } + + if tool_call_id: + function_response["id"] = tool_call_id + contents.append({ "role": "user", "parts": [{ - "function_response": { - "name": tool_name, - "response": {"output": text_content} - } + "function_response": function_response, }], }) else: + # Without a function name, do not fabricate a + # function_response. Preserve the result as text. + if isinstance(content, (dict, list)): + text = json.dumps( + content, + ensure_ascii=False, + default=str, + ) + else: + text = "" if content is None else str(content) + contents.append({ "role": "user", - "parts": [{"text": text_content}], + "parts": [{"text": text}], }) + continue - text = str(content_payload) + # Ordinary user and assistant text messages. + text = "" if content is None else str(content) + contents.append({ "role": "model" if role == "assistant" else "user", "parts": [{"text": text}], }) - - return contents or [{"role": "user", "parts": [{"text": "Continue."}]}], ( - "\n\n".join(system) if system else None - ) + + if not contents: + contents.append({ + "role": "user", + "parts": [{"text": "Continue."}], + }) + + system_instruction = "\n\n".join(system) if system else None + + return contents, system_instruction @staticmethod def _usage(response: Any) -> dict[str, int | None] | None: meta = getattr(response, "usage_metadata", None) + if meta is None: return None + return { - "prompt_tokens": getattr(meta, "prompt_token_count", 0) or 0, - "completion_tokens": getattr(meta, "candidates_token_count", 0) or 0, + "prompt_tokens": getattr( + meta, "prompt_token_count", 0 + ) or 0, + "completion_tokens": getattr( + meta, "candidates_token_count", 0 + ) or 0, } - def chat(self, messages: list[dict[str, str]], model: str | None = None) -> tuple[str, dict | None]: + def chat( + self, + messages: list[dict[str, Any]], + model: str | None = None, + ) -> tuple[str, dict | None]: return self.chat_response(messages, model).as_legacy_tuple() - def chat_response(self, messages: list[dict[str, str]], model: str | None = None) -> ModelResponse: + def chat_response( + self, + messages: list[dict[str, Any]], + model: str | None = None, + ) -> ModelResponse: model = model or self.config.model_simple started = time.monotonic() contents, system_instruction = self._contents(messages) + request_config: dict[str, Any] = {} if system_instruction: request_config["system_instruction"] = system_instruction def _call(): return self._client.models.generate_content( - model=model, contents=contents, config=request_config or None, + model=model, + contents=contents, + config=request_config or None, ) response = _retry_with_backoff(_call) + with received_response(): usage_dict = self._usage(response) - usage_ledger._track_cost(self.config.provider, usage_dict, model=model, - latency_ms=round((time.monotonic() - started) * 1000)) + + usage_ledger._track_cost( + self.config.provider, + usage_dict, + model=model, + latency_ms=round( + (time.monotonic() - started) * 1000 + ), + ) + candidates = getattr(response, "candidates", None) or () - candidate = candidates if candidates else None - parts = getattr(getattr(candidate, "content", None), "parts", None) - text = ("".join(part.text for part in parts if getattr(part, "text", None) - and not getattr(part, "thought", False)) if parts is not None - else str(getattr(response, "text", "") or "")) + + # IMPORTANT: select one candidate, not the entire list. + candidate = candidates[0] if candidates else None + + parts = getattr( + getattr(candidate, "content", None), + "parts", + None, + ) + + text = ( + "".join( + part.text + for part in parts + if getattr(part, "text", None) + and not getattr(part, "thought", False) + ) + if parts is not None + else str(getattr(response, "text", "") or "") + ) + calls = [] + for part in parts or (): call = getattr(part, "function_call", None) - if call is not None: - arguments = getattr(call, "args", None) - calls.append((getattr(call, "id", None), getattr(call, "name", None), - {} if arguments is None else arguments)) - block_reason = getattr(getattr(response, "prompt_feedback", None), "block_reason", None) - return model_response(provider=self.name, model=model, actual_model=getattr(response, "model_version", None), text=text, usage=usage_dict, - calls=calls, response_id=getattr(response, "response_id", None), - raw_finish_reason=getattr(candidate, "finish_reason", None) or block_reason, - blocked=block_reason not in {None, "BLOCK_REASON_UNSPECIFIED", "BLOCKED_REASON_UNSPECIFIED"}) - - def chat_stream(self, messages: list[dict[str, str]], model: str | None = None) -> Iterator[str]: + + if call is None: + continue + + arguments = getattr(call, "args", None) + + calls.append(( + getattr(call, "id", None), + getattr(call, "name", None), + {} if arguments is None else arguments, + )) + + block_reason = getattr( + getattr(response, "prompt_feedback", None), + "block_reason", + None, + ) + + return model_response( + provider=self.name, + model=model, + actual_model=getattr( + response, "model_version", None + ), + text=text, + usage=usage_dict, + calls=calls, + response_id=getattr( + response, "response_id", None + ), + raw_finish_reason=( + getattr(candidate, "finish_reason", None) + or block_reason + ), + blocked=block_reason not in { + None, + "BLOCK_REASON_UNSPECIFIED", + "BLOCKED_REASON_UNSPECIFIED", + }, + ) + + def chat_stream( + self, + messages: list[dict[str, Any]], + model: str | None = None, + ) -> Iterator[str]: model = model or self.config.model_simple + contents, system_instruction = self._contents(messages) + request_config: dict[str, Any] = {} if system_instruction: request_config["system_instruction"] = system_instruction + started = time.monotonic() final_usage: dict[str, int | None] | None = None completed = False - stream = _retry_with_backoff(lambda: self._client.models.generate_content_stream( - model=model, contents=contents, config=request_config or None, - )) + + stream = _retry_with_backoff( + lambda: self._client.models.generate_content_stream( + model=model, + contents=contents, + config=request_config or None, + ) + ) + try: for chunk in stream: usage = self._usage(chunk) + if usage is not None: final_usage = usage + delta = str(getattr(chunk, "text", "") or "") + if delta: yield delta + completed = True finally: if completed: - usage_ledger._track_cost(self.config.provider, final_usage, model=model, - latency_ms=round((time.monotonic() - started) * 1000)) + usage_ledger._track_cost( + self.config.provider, + final_usage, + model=model, + latency_ms=round( + (time.monotonic() - started) * 1000 + ), + ) class VertexProvider(GoogleProvider): @@ -172,6 +338,7 @@ class VertexProvider(GoogleProvider): def __init__(self, config: ProviderConfig) -> None: LLMProvider.__init__(self, config) + try: from google import genai except ImportError: @@ -179,6 +346,7 @@ def __init__(self, config: ProviderConfig) -> None: "The 'google-genai' package is required for Vertex AI. " "Install it with: pip install google-genai" ) + self._client = genai.Client( vertexai=True, project=os.environ.get("GOOGLE_CLOUD_PROJECT"), From 120534b09a8e3336821198fe6a483d8a91667a1e Mon Sep 17 00:00:00 2001 From: sheetal katre Date: Fri, 9 Oct 2026 15:52:48 +0530 Subject: [PATCH 6/6] fix(providers/google): preserve function-call metadata for Gemini sequence consistency Add native assistant function_call preservation inside GoogleProvider._contents to comply with Gemini documentation specifications. Resolves missing context anomalies caught during multi-turn continuation routines and establishes structural fallback logic. --- openkyrozen/providers/google.py | 550 +++++++++++++++++--------------- 1 file changed, 289 insertions(+), 261 deletions(-) diff --git a/openkyrozen/providers/google.py b/openkyrozen/providers/google.py index 8161b8f..67dcaa9 100644 --- a/openkyrozen/providers/google.py +++ b/openkyrozen/providers/google.py @@ -1,11 +1,12 @@ -python + from __future__ import annotations import json import os import sys import time -from typing import Any, Iterator +from collections.abc import Iterator, Mapping +from typing import Any import openkyrozen.providers.usage as usage_ledger from openkyrozen.providers.base import LLMProvider @@ -19,7 +20,7 @@ class GoogleProvider(LLMProvider): - """Handles Google Gemini through the current Google Gen AI SDK.""" + """Handles Google Gemini through the Google Gen AI SDK.""" def __init__(self, config: ProviderConfig) -> None: super().__init__(config) @@ -27,328 +28,355 @@ def __init__(self, config: ProviderConfig) -> None: try: from google import genai except ImportError: - sys.exit( + raise ImportError( "The 'google-genai' package is required for Gemini. " "Install it with: pip install google-genai" + ) from None + + api_key = getattr(config, "api_key", None) or os.getenv( + "GEMINI_API_KEY" + ) or os.getenv("GOOGLE_API_KEY") + + if not api_key: + raise ValueError( + "Gemini API key is missing. Set GEMINI_API_KEY or " + "configure the provider API key." ) - self._client = genai.Client( - api_key=config.api_key - or os.environ.get("GEMINI_API_KEY", "") - ) + self._client = genai.Client(api_key=api_key) @staticmethod + def _json_value(value: Any) -> Any: + """Convert values into JSON-compatible Python values.""" + if isinstance(value, Mapping): + return dict(value) + + if isinstance(value, (list, tuple)): + return list(value) + + if isinstance(value, str): + try: + return json.loads(value) + except (json.JSONDecodeError, TypeError): + return value + + if value is None: + return {} + + if isinstance(value, (bool, int, float)): + return value + + return str(value) + + @classmethod + def _function_args(cls, value: Any) -> dict[str, Any]: + """Gemini function-call args must be an object.""" + value = cls._json_value(value) + + if isinstance(value, dict): + return value + + if isinstance(value, str): + try: + parsed = json.loads(value) + except json.JSONDecodeError: + return {} + + if isinstance(parsed, dict): + return parsed + + return {} + + @classmethod def _contents( + cls, messages: list[dict[str, Any]], ) -> tuple[list[dict[str, Any]], str | None]: + """ + Convert conversation messages into Gemini contents. + + Tool responses are only emitted as function_response parts when + a matching function call exists in the preceding assistant turn. + """ contents: list[dict[str, Any]] = [] - system: list[str] = [] + system_messages: list[str] = [] + + pending_calls: dict[str, str] = {} for message in messages: - role = message.get("role", "user") + role = str(message.get("role", "user")).lower() content = message.get("content", "") + name = message.get("name") or message.get("tool_name") + call_id = ( + message.get("tool_call_id") + or message.get("call_id") + or message.get("id") + ) if role == "system": - if content is not None: - system.append(str(content)) + if content: + system_messages.append(str(content)) continue - # Preserve assistant function-call history. - if role == "assistant" and message.get("calls"): - parts: list[dict[str, Any]] = [] + if role in ("tool", "function"): + payload = cls._json_value(content) + + # Match the receipt to the original assistant call. + function_name = pending_calls.pop(str(call_id), None) if call_id else None + + # Some internal message formats identify the tool by name. + if function_name is None and name: + function_name = str(name) + + if function_name is not None: + if not isinstance(payload, dict): + payload = {"result": payload} + + contents.append( + { + "role": "user", + "parts": [ + { + "function_response": { + "name": function_name, + "response": payload, + } + } + ], + } + ) + else: + # Never send an unmatched function_response. + # Preserve the receipt as ordinary text instead. + contents.append( + { + "role": "user", + "parts": [ + { + "text": ( + f"Tool result ({name or 'unknown tool'}): " + f"{json.dumps(payload, ensure_ascii=False)}" + ) + } + ], + } + ) + continue - if content: - parts.append({"text": str(content)}) - - for call in message["calls"]: - # Existing internal format: (call_id, name, args) - if isinstance(call, (tuple, list)) and len(call) == 3: - call_id, name, args = call - elif isinstance(call, dict): - call_id = call.get("id") - name = call.get("name") - args = call.get("args", {}) - else: - raise ValueError( - "Unsupported assistant tool-call format" - ) + if role == "assistant": + parts: list[dict[str, Any]] = [] - if not name: - raise ValueError( - "Assistant tool call is missing its name" + if isinstance(content, str) and content.strip(): + parts.append({"text": content}) + elif content is not None and not isinstance(content, str): + normalized = cls._json_value(content) + if normalized: + parts.append( + { + "text": json.dumps( + normalized, + ensure_ascii=False, + ) + } ) - function_call: dict[str, Any] = { - "name": name, - "args": args if isinstance(args, dict) else {}, - } + calls = message.get("calls") or message.get("tool_calls") or [] - if call_id: - function_call["id"] = call_id + for call in calls: + if isinstance(call, Mapping): + call_id = call.get("id") or call.get("call_id") + function = call.get("function", call) + function_name = function.get("name") + arguments = function.get( + "arguments", + function.get("args", {}), + ) + elif isinstance(call, (list, tuple)) and len(call) >= 3: + call_id, function_name, arguments = call[:3] + else: + continue - parts.append({"function_call": function_call}) + if not function_name: + continue - contents.append({ - "role": "model", - "parts": parts, - }) - continue + function_name = str(function_name) - # Convert tool results to Gemini function responses. - if role in ("tool", "function"): - tool_name = message.get("name") - tool_call_id = message.get("tool_call_id") + parts.append( + { + "function_call": { + "name": function_name, + "args": cls._function_args(arguments), + } + } + ) - if isinstance(content, dict): - output = content - elif isinstance(content, list): - output = {"output": content} - else: - output = {"output": "" if content is None else str(content)} - - if tool_name: - function_response: dict[str, Any] = { - "name": tool_name, - "response": output, - } - - if tool_call_id: - function_response["id"] = tool_call_id - - contents.append({ - "role": "user", - "parts": [{ - "function_response": function_response, - }], - }) - else: - # Without a function name, do not fabricate a - # function_response. Preserve the result as text. - if isinstance(content, (dict, list)): - text = json.dumps( - content, - ensure_ascii=False, - default=str, - ) - else: - text = "" if content is None else str(content) - - contents.append({ - "role": "user", - "parts": [{"text": text}], - }) + if call_id is not None: + pending_calls[str(call_id)] = function_name + if parts: + contents.append({"role": "model", "parts": parts}) continue - # Ordinary user and assistant text messages. - text = "" if content is None else str(content) + # Default: user message. + if isinstance(content, str): + text = content + else: + text = json.dumps( + cls._json_value(content), + ensure_ascii=False, + ) - contents.append({ - "role": "model" if role == "assistant" else "user", - "parts": [{"text": text}], - }) + contents.append( + { + "role": "user", + "parts": [{"text": text or " "}], + } + ) if not contents: - contents.append({ - "role": "user", - "parts": [{"text": "Continue."}], - }) + contents.append( + { + "role": "user", + "parts": [{"text": "Continue."}], + } + ) - system_instruction = "\n\n".join(system) if system else None + system_instruction = ( + "\n\n".join(system_messages) if system_messages else None + ) return contents, system_instruction @staticmethod - def _usage(response: Any) -> dict[str, int | None] | None: - meta = getattr(response, "usage_metadata", None) - - if meta is None: - return None - - return { - "prompt_tokens": getattr( - meta, "prompt_token_count", 0 - ) or 0, - "completion_tokens": getattr( - meta, "candidates_token_count", 0 - ) or 0, - } - - def chat( - self, - messages: list[dict[str, Any]], - model: str | None = None, - ) -> tuple[str, dict | None]: - return self.chat_response(messages, model).as_legacy_tuple() + def _response_text(response: Any) -> str: + """Extract visible text from a Gemini response.""" + text = getattr(response, "text", None) + if isinstance(text, str): + return text - def chat_response( - self, - messages: list[dict[str, Any]], - model: str | None = None, - ) -> ModelResponse: - model = model or self.config.model_simple - started = time.monotonic() + output: list[str] = [] - contents, system_instruction = self._contents(messages) + for candidate in getattr(response, "candidates", None) or []: + content = getattr(candidate, "content", None) + for part in getattr(content, "parts", None) or []: + part_text = getattr(part, "text", None) + if isinstance(part_text, str) and not getattr( + part, "thought", False + ): + output.append(part_text) - request_config: dict[str, Any] = {} - if system_instruction: - request_config["system_instruction"] = system_instruction + return "".join(output) - def _call(): - return self._client.models.generate_content( - model=model, - contents=contents, - config=request_config or None, - ) + @staticmethod + def _response_calls(response: Any) -> list[tuple[str, str, dict[str, Any]]]: + """Extract function calls while preserving IDs when available.""" + calls: list[tuple[str, str, dict[str, Any]]] = [] - response = _retry_with_backoff(_call) + for candidate in getattr(response, "candidates", None) or []: + content = getattr(candidate, "content", None) - with received_response(): - usage_dict = self._usage(response) + for part in getattr(content, "parts", None) or []: + function_call = getattr(part, "function_call", None) + if not function_call: + continue - usage_ledger._track_cost( - self.config.provider, - usage_dict, - model=model, - latency_ms=round( - (time.monotonic() - started) * 1000 - ), - ) + name = getattr(function_call, "name", None) + if not name: + continue - candidates = getattr(response, "candidates", None) or () + args = getattr(function_call, "args", None) or {} + if not isinstance(args, dict): + args = GoogleProvider._function_args(args) - # IMPORTANT: select one candidate, not the entire list. - candidate = candidates[0] if candidates else None + call_id = getattr(part, "id", None) or "" + calls.append((str(call_id), str(name), args)) - parts = getattr( - getattr(candidate, "content", None), - "parts", - None, - ) + return calls - text = ( - "".join( - part.text - for part in parts - if getattr(part, "text", None) - and not getattr(part, "thought", False) - ) - if parts is not None - else str(getattr(response, "text", "") or "") - ) + def chat_response( + self, + messages: list[dict[str, Any]], + **kwargs: Any, + ) -> ModelResponse: + """Send a conversation to Gemini and return the provider response.""" + from google.genai import types - calls = [] + contents, system_instruction = self._contents(messages) - for part in parts or (): - call = getattr(part, "function_call", None) + model = ( + kwargs.pop("model", None) + or getattr(self._config, "model", None) + or "gemini-2.5-flash" + ) - if call is None: - continue + config_kwargs: dict[str, Any] = {} + if system_instruction: + config_kwargs["system_instruction"] = system_instruction - arguments = getattr(call, "args", None) + generation_config = types.GenerateContentConfig(**config_kwargs) - calls.append(( - getattr(call, "id", None), - getattr(call, "name", None), - {} if arguments is None else arguments, - )) + start = time.monotonic() - block_reason = getattr( - getattr(response, "prompt_feedback", None), - "block_reason", - None, - ) + response = self._client.models.generate_content( + model=model, + contents=contents, + config=generation_config, + ) - return model_response( - provider=self.name, - model=model, - actual_model=getattr( - response, "model_version", None - ), - text=text, - usage=usage_dict, - calls=calls, - response_id=getattr( - response, "response_id", None - ), - raw_finish_reason=( - getattr(candidate, "finish_reason", None) - or block_reason - ), - blocked=block_reason not in { - None, - "BLOCK_REASON_UNSPECIFIED", - "BLOCKED_REASON_UNSPECIFIED", - }, - ) + elapsed = time.monotonic() - start + text = self._response_text(response) + calls = self._response_calls(response) + + usage = getattr(response, "usage_metadata", None) + if usage is not None: + usage_data = { + "prompt_tokens": getattr(usage, "prompt_token_count", 0) or 0, + "completion_tokens": getattr( + usage, "candidates_token_count", 0 + ) or 0, + "total_tokens": getattr(usage, "total_token_count", 0) or 0, + } + else: + usage_data = {} + + result = model_response( + text=text, + tool_calls=calls, + usage=usage_data, + elapsed=elapsed, + ) - def chat_stream( + return result + + def stream_response( self, messages: list[dict[str, Any]], - model: str | None = None, + **kwargs: Any, ) -> Iterator[str]: - model = model or self.config.model_simple + """Stream generated text from Gemini.""" + from google.genai import types contents, system_instruction = self._contents(messages) - request_config: dict[str, Any] = {} - if system_instruction: - request_config["system_instruction"] = system_instruction - - started = time.monotonic() - final_usage: dict[str, int | None] | None = None - completed = False - - stream = _retry_with_backoff( - lambda: self._client.models.generate_content_stream( - model=model, - contents=contents, - config=request_config or None, - ) + model = ( + kwargs.pop("model", None) + or getattr(self._config, "model", None) + or "gemini-2.5-flash" ) - try: - for chunk in stream: - usage = self._usage(chunk) - - if usage is not None: - final_usage = usage - - delta = str(getattr(chunk, "text", "") or "") - - if delta: - yield delta - - completed = True - finally: - if completed: - usage_ledger._track_cost( - self.config.provider, - final_usage, - model=model, - latency_ms=round( - (time.monotonic() - started) * 1000 - ), - ) - - -class VertexProvider(GoogleProvider): - """Google Gen AI SDK configured for Vertex AI and ADC.""" - - def __init__(self, config: ProviderConfig) -> None: - LLMProvider.__init__(self, config) + config_kwargs: dict[str, Any] = {} + if system_instruction: + config_kwargs["system_instruction"] = system_instruction - try: - from google import genai - except ImportError: - sys.exit( - "The 'google-genai' package is required for Vertex AI. " - "Install it with: pip install google-genai" - ) + config = types.GenerateContentConfig(**config_kwargs) - self._client = genai.Client( - vertexai=True, - project=os.environ.get("GOOGLE_CLOUD_PROJECT"), - location=os.environ.get("GOOGLE_CLOUD_LOCATION", "global"), + stream = self._client.models.generate_content_stream( + model=model, + contents=contents, + config=config, ) + + for chunk in stream: + text = self._response_text(chunk) + if text: + yield text