Coverage for src / lexigram / ai / relay / mappers / gemini.py: 88%
485 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-08 23:08 +0800
1"""Google Gemini ``generateContent`` request and response mapper.
3Converts the Gemini wire DTOs (:class:`GeminiRequest` /
4:class:`GeminiResponse`) into the canonical relay IR and back. Stream
5conversion is handled by the shared stream lifecycle task and reports
6``unsupported_feature`` until then.
7"""
9from __future__ import annotations
11from dataclasses import replace
12from typing import Any
14from lexigram.ai.relay.context import ConversionContext
15from lexigram.ai.relay.errors import (
16 translate,
17 unsupported_feature,
18 unsupported_format,
19)
20from lexigram.ai.relay.finish_reasons import (
21 FINISH_REASON_TO_WIRE,
22 finish_reason_to_wire,
23)
24from lexigram.ai.relay.mappers.base import record_loss
25from lexigram.ai.relay.media import resolve_media
26from lexigram.contracts.ai.agents import ToolDefinition
27from lexigram.contracts.ai.exceptions import RelayError
28from lexigram.contracts.ai.llm import ChatMessage, FunctionCall, ToolCall
29from lexigram.contracts.ai.multimodal import (
30 ContentPart,
31 ImageBase64Part,
32 ImageUrlPart,
33 TextPart,
34)
35from lexigram.contracts.ai.relay.dto import (
36 GeminiCandidate,
37 GeminiContent,
38 GeminiGroundingMetadata,
39 GeminiPart,
40 GeminiPromptFeedback,
41 GeminiRequest,
42 GeminiResponse,
43 GeminiSafetyRating,
44 GeminiUsageMetadata,
45)
46from lexigram.contracts.ai.relay.ir import (
47 RelayRequest,
48 RelayResponse,
49 StreamDelta,
50 StreamState,
51 normalize_finish_reason,
52)
53from lexigram.contracts.ai.relay.types import RelayFormat, RelayUsage
54from lexigram.contracts.ai.thinking import ThinkingConfig, ThinkingResult
55from lexigram.contracts.core.result import Err, Ok, Result
56from lexigram.serialization import dumps_str, loads_str
58__all__ = ["GeminiMapper"]
60_TARGET = RelayFormat.GEMINI
62_SAFETY_CATEGORIES = (
63 "HARM_CATEGORY_HARASSMENT",
64 "HARM_CATEGORY_HATE_SPEECH",
65 "HARM_CATEGORY_SEXUALLY_EXPLICIT",
66 "HARM_CATEGORY_DANGEROUS_CONTENT",
67)
69_MIME_KEY = "mimeType"
71_THOUGHT_SIGNATURE_BYPASS = "context_engineering_is_the_way_to_go"
73_SCHEMA_TYPE_MAP = {
74 "string": "STRING",
75 "object": "OBJECT",
76 "array": "ARRAY",
77 "integer": "INTEGER",
78 "number": "NUMBER",
79 "boolean": "BOOLEAN",
80}
83def _tool_call_from_part(part: GeminiPart) -> ToolCall:
84 """Convert a Gemini ``functionCall`` part into a canonical ``ToolCall``.
86 Gemini function calls carry no stable id; the canonical id stays
87 empty so target writers can generate a dialect-appropriate one.
88 """
89 call = part.function_call or {}
90 name = str(call.get("name", ""))
91 args = call.get("args")
92 return ToolCall(
93 id="",
94 type="custom",
95 function=FunctionCall(
96 name=name,
97 arguments=args if isinstance(args, dict) else {},
98 ),
99 )
102def _tool_call_to_part(tool_call: ToolCall) -> GeminiPart:
103 """Serialize a canonical ``ToolCall`` as a Gemini ``functionCall`` part."""
104 arguments: Any = tool_call.function.arguments if tool_call.function else {}
105 if isinstance(arguments, str):
106 try:
107 arguments = loads_str(arguments)
108 except ValueError:
109 arguments = {}
110 elif not isinstance(arguments, dict):
111 arguments = {}
112 return GeminiPart(
113 function_call={
114 "name": tool_call.function.name if tool_call.function else "",
115 "args": arguments,
116 }
117 )
120class GeminiMapper:
121 """Bidirectional Google Gemini ``generateContent`` converter.
123 Attributes:
124 format: The wire format this mapper handles.
125 """
127 format = _TARGET
129 def request_to_ir(
130 self, payload: Any, *, context: ConversionContext
131 ) -> Result[RelayRequest, RelayError]:
132 """Convert a ``GeminiRequest`` into canonical ``RelayRequest``.
134 Args:
135 payload: A wire request DTO.
136 context: Per-conversion context with loss sink.
138 Returns:
139 Ok(request) on success, Err(relay_error) on malformed payload.
140 """
141 if not isinstance(payload, GeminiRequest):
142 return Err(
143 unsupported_format(
144 f"expected GeminiRequest, got {type(payload).__name__}"
145 )
146 )
147 try:
148 messages = [
149 chat_message
150 for index, content in enumerate(payload.contents)
151 for chat_message in self._content_to_ir(content, context, index)
152 ]
153 metadata: dict[str, Any] = {}
154 if payload.safety_settings is not None:
155 metadata["safety_settings"] = [
156 dict(item) for item in payload.safety_settings
157 ]
158 if payload.tool_config is not None:
159 metadata["tool_config"] = dict(payload.tool_config)
160 generation_config = dict(payload.generation_config)
161 if generation_config:
162 metadata["generation_config"] = generation_config
163 return Ok(
164 RelayRequest(
165 model=str(payload.passthrough.get("model", "")).strip(),
166 messages=messages,
167 system=self._system_to_ir(payload.system_instruction),
168 tools=self._tools_to_ir(payload.tools),
169 temperature=_config_number(generation_config, "temperature"),
170 top_p=_config_number(generation_config, "topP"),
171 top_k=_config_int(generation_config, "topK"),
172 max_tokens=_config_int(generation_config, "maxOutputTokens"),
173 stop_sequences=[
174 str(item)
175 for item in generation_config.get("stopSequences", [])
176 if isinstance(item, str)
177 ],
178 response_format=self._response_format_to_ir(generation_config),
179 thinking=self._thinking_to_ir(generation_config),
180 metadata=metadata,
181 passthrough=dict(payload.passthrough),
182 )
183 )
184 except (RelayError, ValueError, TypeError, KeyError) as exc:
185 return Err(translate(exc, detail="request_to_ir"))
187 def ir_to_request(
188 self, request: RelayRequest, *, context: ConversionContext
189 ) -> Result[Any, RelayError]:
190 """Convert canonical ``RelayRequest`` into a ``GeminiRequest``.
192 Args:
193 request: Canonical request IR.
194 context: Per-conversion context with loss sink.
196 Returns:
197 Ok(request) on success, Err(relay_error) on failure.
198 """
199 try:
200 system_parts: list[str] = []
201 contents: list[GeminiContent] = []
202 tool_names, tool_names_by_id = self._tool_name_resolver(request)
203 for message in request.messages:
204 if message.role == "system":
205 system_parts.append(self._text_from_content(message.content))
206 continue
207 content = self._content_from_ir(
208 message, request.model, tool_names, tool_names_by_id, context
209 )
210 if content.is_err():
211 return content
212 contents.append(content.unwrap())
213 if request.system:
214 system_parts.append(request.system)
215 return Ok(
216 GeminiRequest(
217 contents=contents,
218 system_instruction=(
219 {"parts": [{"text": text} for text in system_parts]}
220 if system_parts
221 else None
222 ),
223 generation_config=self._generation_config_from_ir(request, context),
224 safety_settings=self._safety_settings_from_ir(request, context),
225 tools=self._tools_from_ir(request.tools),
226 tool_config=self._tool_config_from_ir(request),
227 passthrough=self._request_passthrough(request),
228 )
229 )
230 except (RelayError, ValueError, TypeError, KeyError) as exc:
231 return Err(translate(exc, detail="ir_to_request"))
233 def response_to_ir(
234 self, payload: Any, *, context: ConversionContext
235 ) -> Result[RelayResponse, RelayError]:
236 """Convert a ``GeminiResponse`` into canonical ``RelayResponse``.
238 Args:
239 payload: A wire response DTO.
240 context: Per-conversion context with loss sink.
242 Returns:
243 Ok(response) on success, Err(relay_error) on malformed payload.
244 """
245 if not isinstance(payload, GeminiResponse):
246 return Err(
247 unsupported_format(
248 f"expected GeminiResponse, got {type(payload).__name__}"
249 )
250 )
251 try:
252 candidates = payload.candidates or []
253 if len(candidates) > 1:
254 record_loss(
255 context,
256 field="candidates",
257 target=_TARGET,
258 reason="multiple_candidates_collapsed",
259 )
260 candidate = candidates[0] if candidates else None
261 passthrough: dict[str, Any] = dict(payload.passthrough)
262 if payload.model_version is not None:
263 passthrough["model_version"] = payload.model_version
264 if payload.create_time is not None:
265 passthrough["create_time"] = payload.create_time
266 if payload.prompt_feedback is not None:
267 passthrough["prompt_feedback"] = payload.prompt_feedback.to_dict()
268 content = ""
269 thinking: ThinkingResult | None = None
270 tool_calls: list[ToolCall] = []
271 if candidate is not None and candidate.content is not None:
272 text_parts: list[str] = []
273 think_parts: list[str] = []
274 think_signature: str | None = None
275 for part in candidate.content.parts:
276 if part.thought:
277 think_parts.append(part.text or "")
278 think_signature = part.thought_signature
279 elif part.text is not None:
280 text_parts.append(part.text)
281 elif part.function_call is not None:
282 tool_calls.append(_tool_call_from_part(part))
283 elif (
284 part.inline_data is not None
285 or part.function_response is not None
286 ):
287 record_loss(
288 context,
289 field="content.part",
290 target=_TARGET,
291 reason="unrepresentable_part_dropped",
292 )
293 content = "".join(text_parts)
294 if think_parts:
295 thinking = ThinkingResult(
296 content="".join(think_parts),
297 signature=think_signature,
298 tokens=self._thought_tokens(payload),
299 )
300 self._preserve_candidate_metadata(candidate, passthrough)
301 return Ok(
302 RelayResponse(
303 model=payload.model_version or "",
304 id=payload.response_id,
305 content=content,
306 thinking=thinking,
307 tool_calls=tool_calls,
308 finish_reason=normalize_finish_reason(
309 candidate.finish_reason if candidate else None
310 ),
311 usage=self._usage_from_wire(payload.usage_metadata),
312 passthrough=passthrough,
313 )
314 )
315 except (RelayError, ValueError, TypeError, KeyError) as exc:
316 return Err(translate(exc, detail="response_to_ir"))
318 def ir_to_response(
319 self, response: RelayResponse, *, context: ConversionContext
320 ) -> Result[Any, RelayError]:
321 """Convert canonical ``RelayResponse`` into a ``GeminiResponse``.
323 Args:
324 response: Canonical response IR.
325 context: Per-conversion context with loss sink.
327 Returns:
328 Ok(response) on success, Err(relay_error) on failure.
329 """
330 try:
331 passthrough = dict(response.passthrough)
332 model_version = passthrough.pop("model_version", None)
333 prompt_feedback = passthrough.pop("prompt_feedback", None)
334 create_time = passthrough.pop("create_time", None)
335 safety_ratings = passthrough.pop("safety_ratings", None)
336 grounding_metadata = passthrough.pop("grounding_metadata", None)
337 citation_metadata = passthrough.pop("citation_metadata", None)
338 token_count = passthrough.pop("token_count", None)
339 avg_logprobs = passthrough.pop("avg_logprobs", None)
340 parts: list[GeminiPart] = []
341 if response.content:
342 parts.append(GeminiPart(text=response.content))
343 for tool_call in response.tool_calls:
344 parts.append(_tool_call_to_part(tool_call))
345 candidate = GeminiCandidate(
346 content=GeminiContent(role="model", parts=parts),
347 finish_reason=self._finish_reason_from_ir(
348 response.finish_reason, context
349 ),
350 index=0,
351 safety_ratings=self._safety_ratings_from_passthrough(safety_ratings)
352 or [],
353 grounding_metadata=self._grounding_from_passthrough(grounding_metadata),
354 citation_metadata=(
355 citation_metadata if isinstance(citation_metadata, dict) else None
356 ),
357 token_count=token_count if isinstance(token_count, int) else None,
358 avg_logprobs=(
359 avg_logprobs if isinstance(avg_logprobs, (int, float)) else None
360 ),
361 passthrough=dict(passthrough),
362 )
363 return Ok(
364 GeminiResponse(
365 candidates=[candidate],
366 prompt_feedback=self._prompt_feedback_from_passthrough(
367 prompt_feedback
368 ),
369 usage_metadata=self._usage_to_wire(response.usage),
370 model_version=model_version
371 if isinstance(model_version, str)
372 else None,
373 create_time=create_time if isinstance(create_time, str) else None,
374 passthrough=passthrough,
375 )
376 )
377 except (RelayError, ValueError, TypeError, KeyError) as exc:
378 return Err(translate(exc, detail="ir_to_response"))
380 def stream_to_delta(
381 self, event: Any, *, state: StreamState
382 ) -> Result[tuple[StreamDelta, ...], RelayError]:
383 """Stream conversion is deferred to the shared stream lifecycle task."""
384 return Err(
385 unsupported_feature("gemini stream conversion is not implemented yet")
386 )
388 def delta_to_stream(
389 self, delta: StreamDelta, *, state: StreamState
390 ) -> Result[tuple[Any, ...], RelayError]:
391 """Stream conversion is deferred to the shared stream lifecycle task."""
392 return Err(
393 unsupported_feature("gemini stream conversion is not implemented yet")
394 )
396 # -- helpers -------------------------------------------------------------
398 @staticmethod
399 def _system_to_ir(
400 system_instruction: dict[str, Any] | None,
401 ) -> str | None:
402 """Extract system text from a Gemini ``systemInstruction`` dict."""
403 if not isinstance(system_instruction, dict):
404 return None
405 parts = system_instruction.get("parts")
406 if not isinstance(parts, list):
407 return None
408 texts: list[str] = []
409 for part in parts:
410 if isinstance(part, dict) and part.get("text") is not None:
411 texts.append(str(part["text"]))
412 return "\n".join(texts)
414 @staticmethod
415 def _tools_to_ir(tools: list[dict[str, Any]] | None) -> list[ToolDefinition]:
416 """Convert Gemini wire tools into canonical ``ToolDefinition`` objects."""
417 definitions: list[ToolDefinition] = []
418 for tool in tools or []:
419 if not isinstance(tool, dict):
420 continue
421 declarations = tool.get("functionDeclarations")
422 if not isinstance(declarations, list):
423 continue
424 for declaration in declarations:
425 if not isinstance(declaration, dict):
426 continue
427 parameters = declaration.get("parameters")
428 definitions.append(
429 ToolDefinition(
430 name=str(declaration.get("name", "")),
431 description=str(declaration.get("description", "")),
432 parameters=parameters if isinstance(parameters, dict) else {},
433 )
434 )
435 return definitions
437 def _content_to_ir(
438 self, content: GeminiContent, context: ConversionContext, index: int
439 ) -> list[ChatMessage]:
440 """Convert one Gemini content turn into canonical messages."""
441 if content.role == "model":
442 return [self._assistant_to_ir(content, context)]
443 if content.role == "user":
444 return self._user_to_ir(content, context, index)
445 if content.role == "function":
446 return self._function_to_ir(content, context, index)
447 record_loss(
448 context,
449 field=f"contents[{index}].role",
450 target=_TARGET,
451 reason="unknown_role_dropped",
452 )
453 return []
455 def _assistant_to_ir(
456 self, content: GeminiContent, context: ConversionContext
457 ) -> ChatMessage:
458 """Convert a model content turn, separating thinking/tool parts."""
459 text_parts: list[str] = []
460 thinking_blocks: list[dict[str, Any]] = []
461 tool_calls: list[ToolCall] = []
462 for part in content.parts:
463 if part.thought:
464 thinking_blocks.append(
465 {
466 "thought": True,
467 "text": part.text or "",
468 "thoughtSignature": part.thought_signature or "",
469 }
470 )
471 elif part.text is not None:
472 text_parts.append(part.text)
473 elif part.function_call is not None:
474 tool_calls.append(_tool_call_from_part(part))
475 else:
476 record_loss(
477 context,
478 field="content.part",
479 target=_TARGET,
480 reason="unrepresentable_part_dropped",
481 )
482 return ChatMessage(
483 role="assistant",
484 content="".join(text_parts),
485 tool_calls=tool_calls or None,
486 thinking_blocks=thinking_blocks or None,
487 )
489 def _user_to_ir(
490 self, content: GeminiContent, context: ConversionContext, index: int
491 ) -> list[ChatMessage]:
492 """Convert a user content turn into canonical parts and tool results."""
493 parts: list[ContentPart] = []
494 tool_results: list[ChatMessage] = []
495 for part in content.parts:
496 if part.text is not None:
497 parts.append(TextPart(text=part.text))
498 elif part.inline_data is not None:
499 inline = part.inline_data
500 parts.append(
501 ImageBase64Part(
502 data=str(inline.get("data", "")),
503 media_type=str(inline.get(_MIME_KEY, "")),
504 detail="auto",
505 )
506 )
507 elif part.function_response is not None:
508 tool_results.append(
509 self._function_response_to_ir(part.function_response)
510 )
511 else:
512 record_loss(
513 context,
514 field=f"contents[{index}].part",
515 target=_TARGET,
516 reason="unrepresentable_part_dropped",
517 )
518 turns: list[ChatMessage] = []
519 if parts:
520 turns.append(
521 ChatMessage(
522 role="user",
523 content=(
524 parts[0].text
525 if len(parts) == 1 and isinstance(parts[0], TextPart)
526 else list(parts)
527 ),
528 )
529 )
530 turns.extend(tool_results)
531 if not turns:
532 record_loss(
533 context,
534 field=f"contents[{index}]",
535 target=_TARGET,
536 reason="empty_message_dropped",
537 )
538 return turns
540 def _function_to_ir(
541 self, content: GeminiContent, context: ConversionContext, index: int
542 ) -> list[ChatMessage]:
543 """Convert a function content turn into canonical tool messages."""
544 messages: list[ChatMessage] = []
545 for part in content.parts:
546 if part.function_response is not None:
547 messages.append(self._function_response_to_ir(part.function_response))
548 else:
549 record_loss(
550 context,
551 field=f"contents[{index}].part",
552 target=_TARGET,
553 reason="unrepresentable_part_dropped",
554 )
555 return messages
557 @staticmethod
558 def _function_response_to_ir(response: dict[str, Any]) -> ChatMessage:
559 """Convert a ``functionResponse`` dict into a canonical tool message."""
560 payload = response.get("response")
561 if isinstance(payload, str):
562 text = payload
563 else:
564 text = dumps_str(payload) if payload is not None else ""
565 return ChatMessage(role="tool", content=text, tool_call_id="")
567 @staticmethod
568 def _response_format_to_ir(
569 generation_config: dict[str, Any],
570 ) -> dict[str, Any] | None:
571 """Derive a canonical response format from the generation config."""
572 mime = generation_config.get("responseMimeType")
573 if not isinstance(mime, str):
574 return None
575 if mime == "application/json":
576 return {"type": "json_object"}
577 return None
579 def _thinking_to_ir(
580 self, generation_config: dict[str, Any]
581 ) -> ThinkingConfig | None:
582 """Extract canonical thinking config from ``thinkingConfig``."""
583 config = generation_config.get("thinkingConfig")
584 if not isinstance(config, dict):
585 return None
586 level = config.get("thinkingLevel")
587 budget = config.get("thinkingBudget")
588 if isinstance(level, str) and level:
589 return ThinkingConfig(level=level)
590 if isinstance(budget, int):
591 return ThinkingConfig(budget_tokens=budget)
592 return None
594 def _content_from_ir(
595 self,
596 message: ChatMessage,
597 model: str,
598 tool_names: list[str],
599 tool_names_by_id: dict[str, str],
600 context: ConversionContext,
601 ) -> Result[GeminiContent, RelayError]:
602 """Convert one canonical message into a Gemini content turn.
604 Args:
605 message: Canonical message to serialize.
606 model: The selected model, for capability lookups.
607 tool_names: Positional resolver for tool-message function
608 names when the canonical ``tool_call_id`` carries no
609 stable link.
610 tool_names_by_id: Resolver keyed by canonical tool-call id.
611 context: Per-conversion context with loss sink.
613 Returns:
614 Ok(content) on success, Err(relay_error) on failure.
615 """
616 if message.role == "tool":
617 response: Any = message.content
618 if isinstance(response, list):
619 response = self._text_from_content(message.content)
620 if isinstance(response, str):
621 try:
622 response = loads_str(response)
623 except ValueError:
624 pass
625 name = tool_names_by_id.get(message.tool_call_id or "", "")
626 if not name and tool_names:
627 name = tool_names.pop(0)
628 if isinstance(response, str):
629 response = {"content": response}
630 return Ok(
631 GeminiContent(
632 role="user",
633 parts=[
634 GeminiPart(
635 function_response={
636 "name": name,
637 "response": response,
638 }
639 )
640 ],
641 )
642 )
643 if message.role == "assistant":
644 parts = self._assistant_parts_from_ir(message, model, context)
645 if parts.is_err():
646 return Err(parts.unwrap_err())
647 return Ok(GeminiContent(role="model", parts=parts.unwrap()))
648 if message.role == "user":
649 parts = self._user_parts_from_ir(message.content, context)
650 if parts.is_err():
651 return Err(parts.unwrap_err())
652 return Ok(GeminiContent(role="user", parts=parts.unwrap()))
653 record_loss(
654 context,
655 field="messages",
656 target=_TARGET,
657 reason=f"unknown_role_{message.role}_dropped",
658 )
659 return Ok(GeminiContent(role="user", parts=[GeminiPart(text="")]))
661 @staticmethod
662 def _tool_name_resolver(
663 request: RelayRequest,
664 ) -> tuple[list[str], dict[str, str]]:
665 """Return tool-message name resolvers for a request.
667 Gemini ``functionResponse`` blocks name the function, not the
668 call id. Tool messages resolve their function name by canonical
669 ``tool_call_id`` first; unresolved messages fall back to
670 positional order against the assistant tool calls that preceded
671 them.
672 """
673 names: list[str] = []
674 names_by_id: dict[str, str] = {}
675 for message in request.messages:
676 if message.role == "assistant":
677 for tool_call in message.tool_calls or []:
678 if tool_call.function:
679 names.append(tool_call.function.name)
680 if tool_call.id:
681 names_by_id[tool_call.id] = tool_call.function.name
682 return names, names_by_id
684 def _assistant_parts_from_ir(
685 self, message: ChatMessage, model: str, context: ConversionContext
686 ) -> Result[list[GeminiPart], RelayError]:
687 """Rebuild Gemini model parts from an assistant message.
689 When the thought-signature bypass policy is enabled the thinking
690 blocks are folded away and a bypass ``thoughtSignature`` is
691 attached to the first function-call part (relaykit's
692 ``FunctionCallThoughtSignatureEnabled`` behavior). Otherwise
693 thinking blocks are re-emitted as native Gemini thought parts.
694 """
695 attach_signature = context.options.gemini.thought_signature_bypass
696 parts: list[GeminiPart] = []
697 if not attach_signature:
698 for block in message.thinking_blocks or []:
699 if not isinstance(block, dict):
700 continue
701 signature = block.get("thoughtSignature")
702 parts.append(
703 GeminiPart(
704 text=str(block.get("text", "")),
705 thought=True,
706 thought_signature=str(signature) if signature else None,
707 )
708 )
709 content_parts = self._user_parts_from_ir(message.content, context)
710 if content_parts.is_err():
711 return Err(content_parts.unwrap_err())
712 parts.extend(content_parts.unwrap())
713 for tool_call in message.tool_calls or []:
714 parts.append(_tool_call_to_part(tool_call))
715 if attach_signature:
716 parts = self._attach_thought_signature(parts)
717 return Ok(parts)
719 @staticmethod
720 def _attach_thought_signature(parts: list[GeminiPart]) -> list[GeminiPart]:
721 """Attach the relaykit thought-signature bypass value to model parts.
723 The signature lands on the first function-call part, or on the
724 first non-empty text part when the message carries no tool calls.
725 """
726 rebuilt: list[GeminiPart] = []
727 attached = False
728 for part in parts:
729 current = part
730 if not attached and part.function_call is not None:
731 current = replace(part, thought_signature=_THOUGHT_SIGNATURE_BYPASS)
732 attached = True
733 rebuilt.append(current)
734 if not attached:
735 for index, part in enumerate(rebuilt):
736 if part.text:
737 rebuilt[index] = replace(
738 part, thought_signature=_THOUGHT_SIGNATURE_BYPASS
739 )
740 break
741 return rebuilt
743 def _user_parts_from_ir(
744 self, content: str | list[ContentPart], context: ConversionContext
745 ) -> Result[list[GeminiPart], RelayError]:
746 """Convert canonical content into Gemini parts."""
747 if isinstance(content, str):
748 return Ok([GeminiPart(text=content)] if content else [])
749 parts: list[GeminiPart] = []
750 for part in content:
751 if isinstance(part, TextPart):
752 parts.append(GeminiPart(text=part.text))
753 elif isinstance(part, ImageBase64Part):
754 parts.append(
755 GeminiPart(
756 inline_data={
757 _MIME_KEY: part.media_type,
758 "data": part.data,
759 }
760 )
761 )
762 elif isinstance(part, ImageUrlPart):
763 resolved = self._resolve_image(part, context)
764 if resolved.is_err():
765 return Err(resolved.unwrap_err())
766 media_type, data = resolved.unwrap()
767 parts.append(
768 GeminiPart(inline_data={_MIME_KEY: media_type, "data": data})
769 )
770 else:
771 record_loss(
772 context,
773 field="message.content",
774 target=_TARGET,
775 reason="unknown_content_part",
776 )
777 return Ok(parts)
779 @staticmethod
780 def _resolve_image(
781 part: ImageUrlPart, context: ConversionContext
782 ) -> Result[tuple[str, str], RelayError]:
783 """Resolve a URL or data-URI image for Gemini.
785 Data URIs decode locally; URLs go through the context resolver.
786 """
787 resolved = resolve_media(
788 part.url,
789 context,
790 field="message.content",
791 target=_TARGET,
792 lossy=False,
793 )
794 if resolved.is_err():
795 return Err(resolved.unwrap_err())
796 image = resolved.unwrap()
797 assert image is not None # lossy=False never drops media
798 return Ok(image)
800 @staticmethod
801 def _text_from_content(content: str | list[ContentPart]) -> str:
802 """Extract plain text from canonical content."""
803 if isinstance(content, str):
804 return content
805 return "".join(part.text for part in content if isinstance(part, TextPart))
807 def _generation_config_from_ir(
808 self, request: RelayRequest, context: ConversionContext
809 ) -> dict[str, Any]:
810 """Rebuild ``generationConfig`` from protocol metadata and canonical fields."""
811 raw = request.metadata.get("generation_config")
812 config: dict[str, Any] = dict(raw) if isinstance(raw, dict) else {}
813 for key in (
814 "temperature",
815 "topP",
816 "topK",
817 "maxOutputTokens",
818 "stopSequences",
819 "responseMimeType",
820 "responseSchema",
821 "thinkingConfig",
822 ):
823 config.pop(key, None)
824 if request.temperature is not None:
825 config["temperature"] = request.temperature
826 if request.top_p is not None:
827 config["topP"] = request.top_p
828 if request.top_k is not None:
829 config["topK"] = request.top_k
830 if request.max_tokens is not None:
831 config["maxOutputTokens"] = request.max_tokens
832 if request.stop_sequences:
833 config["stopSequences"] = list(request.stop_sequences)
834 if request.response_format is not None:
835 if request.response_format.get("type") == "json_object":
836 config["responseMimeType"] = "application/json"
837 if isinstance(request.response_format.get("schema"), dict):
838 config["responseSchema"] = request.response_format["schema"]
839 thinking_config = self._thinking_config_from_ir(request, context)
840 if thinking_config is not None:
841 config["thinkingConfig"] = thinking_config
842 if "responseModalities" not in config and context.supports_image_generation(
843 request.model
844 ):
845 config["responseModalities"] = ["TEXT", "IMAGE"]
846 return config
848 def _thinking_config_from_ir(
849 self, request: RelayRequest, context: ConversionContext
850 ) -> dict[str, Any] | None:
851 """Build a Gemini ``thinkingConfig`` from canonical thinking."""
852 if request.thinking is not None:
853 record_loss(
854 context,
855 field="thinking",
856 target=_TARGET,
857 reason="thinking_not_supported",
858 )
859 if (
860 context.options.gemini.thinking_adapter_enabled
861 and context.options.gemini.thinking_budget
862 ):
863 return {"thinkingBudget": context.options.gemini.thinking_budget}
864 return None
866 @staticmethod
867 def _safety_settings_from_ir(
868 request: RelayRequest, context: ConversionContext
869 ) -> list[dict[str, Any]] | None:
870 """Rebuild Gemini safety settings from metadata or the callback."""
871 raw = request.metadata.get("safety_settings")
872 if isinstance(raw, list):
873 preserved = [dict(item) for item in raw if isinstance(item, dict)]
874 return preserved or None
875 collected: list[dict[str, Any]] = []
876 for category in _SAFETY_CATEGORIES:
877 threshold = context.safety_setting(category)
878 if threshold and isinstance(threshold, str):
879 collected.append({"category": category, "threshold": threshold})
880 return collected or None
882 @staticmethod
883 def _tools_from_ir(tools: list[ToolDefinition]) -> list[dict[str, Any]] | None:
884 """Serialize canonical tools as Gemini function declarations."""
885 if not tools:
886 return None
887 return [
888 {
889 "functionDeclarations": [
890 {
891 "name": tool.name,
892 "description": tool.description,
893 "parameters": GeminiMapper._upper_schema_types(tool.parameters),
894 }
895 for tool in tools
896 ]
897 }
898 ]
900 @staticmethod
901 def _request_passthrough(request: RelayRequest) -> dict[str, Any]:
902 """Carry canonical passthrough state into the request wire payload."""
903 return dict(request.passthrough)
905 @staticmethod
906 def _upper_schema_types(parameters: dict[str, Any]) -> dict[str, Any]:
907 """Uppercase Gemini schema type markers recursively.
909 Gemini function declarations require ``STRING``/``OBJECT`` type
910 values; canonical schemas carry the lowercase JSON-Schema form.
911 """
912 out: dict[str, Any] = {}
913 for key, value in parameters.items():
914 if key == "type" and isinstance(value, str):
915 out[key] = _SCHEMA_TYPE_MAP.get(value, value)
916 elif isinstance(value, dict):
917 out[key] = GeminiMapper._upper_schema_types(value)
918 elif isinstance(value, list):
919 out[key] = [
920 GeminiMapper._upper_schema_types(item)
921 if isinstance(item, dict)
922 else item
923 for item in value
924 ]
925 else:
926 out[key] = value
927 return out
929 @staticmethod
930 def _tool_config_from_ir(request: RelayRequest) -> dict[str, Any] | None:
931 """Rebuild a Gemini ``toolConfig`` from canonical tool choice."""
932 raw = request.metadata.get("tool_config")
933 if isinstance(raw, dict):
934 return dict(raw)
935 choice = request.tool_choice
936 if isinstance(choice, dict):
937 name = choice.get("function", {})
938 if isinstance(name, dict):
939 name = name.get("name")
940 if isinstance(name, str) and name:
941 return {
942 "functionCallingConfig": {
943 "mode": "ANY",
944 "allowedFunctionNames": [name],
945 }
946 }
947 return {"functionCallingConfig": {"mode": "ANY"}}
948 if isinstance(choice, str):
949 mode = {"auto": "AUTO", "none": "NONE", "required": "ANY"}.get(
950 choice, "AUTO"
951 )
952 return {"functionCallingConfig": {"mode": mode}}
953 return None
955 def _usage_from_wire(self, usage: GeminiUsageMetadata | None) -> RelayUsage | None:
956 """Map a wire ``GeminiUsageMetadata`` into canonical ``RelayUsage``.
958 Mirrors relaykit's ``UsageFromGeminiMetadata``: completion counts
959 thinking tokens, the prompt adds tool-use tokens, and the explicit
960 total is preserved because Gemini counts thoughts within both.
961 """
962 if usage is None:
963 return None
964 return RelayUsage(
965 prompt_tokens=(
966 usage.prompt_token_count + usage.tool_use_prompt_token_count
967 ),
968 completion_tokens=(
969 usage.candidates_token_count + (usage.thoughts_token_count or 0)
970 ),
971 cache_read_tokens=usage.cached_content_token_count or 0,
972 reasoning_tokens=usage.thoughts_token_count or 0,
973 total_tokens_override=usage.total_token_count or None,
974 )
976 @staticmethod
977 def _usage_to_wire(usage: RelayUsage | None) -> GeminiUsageMetadata | None:
978 """Serialize canonical ``RelayUsage`` into a ``GeminiUsageMetadata``.
980 Gemini reports thinking tokens as a subset of the candidate
981 tokens and does not surface cache or reasoning fields in the
982 generated payload, so those counters are emitted as zeros.
983 """
984 if usage is None:
985 return None
986 return GeminiUsageMetadata(
987 prompt_token_count=usage.prompt_tokens,
988 candidates_token_count=usage.completion_tokens,
989 total_token_count=usage.total_tokens,
990 cached_content_token_count=0,
991 thoughts_token_count=0,
992 tool_use_prompt_token_count=0,
993 )
995 @staticmethod
996 def _thought_tokens(payload: GeminiResponse) -> int | None:
997 """Read thinking tokens from the usage metadata."""
998 if (
999 payload.usage_metadata is None
1000 or not payload.usage_metadata.thoughts_token_count
1001 ):
1002 return None
1003 return payload.usage_metadata.thoughts_token_count
1005 @staticmethod
1006 def _preserve_candidate_metadata(
1007 candidate: GeminiCandidate, passthrough: dict[str, Any]
1008 ) -> None:
1009 """Preserve candidate-level provider metadata as passthrough."""
1010 if candidate.safety_ratings:
1011 passthrough["safety_ratings"] = [
1012 rating.to_dict() for rating in candidate.safety_ratings
1013 ]
1014 if candidate.grounding_metadata is not None:
1015 passthrough["grounding_metadata"] = candidate.grounding_metadata.to_dict()
1016 if candidate.citation_metadata is not None:
1017 passthrough["citation_metadata"] = candidate.citation_metadata
1018 if candidate.token_count is not None:
1019 passthrough["token_count"] = candidate.token_count
1020 if candidate.avg_logprobs is not None:
1021 passthrough["avg_logprobs"] = candidate.avg_logprobs
1022 passthrough.update(candidate.passthrough)
1024 def _finish_reason_from_ir(
1025 self, finish_reason: str | None, context: ConversionContext
1026 ) -> str | None:
1027 """Map a canonical finish reason back to a Gemini value."""
1028 if finish_reason is None:
1029 return None
1030 if finish_reason == "function_call":
1031 record_loss(
1032 context,
1033 field="finish_reason",
1034 target=_TARGET,
1035 reason="function_call_adapted",
1036 )
1037 elif finish_reason not in FINISH_REASON_TO_WIRE:
1038 record_loss(
1039 context,
1040 field="finish_reason",
1041 target=_TARGET,
1042 reason="finish_reason_adapted",
1043 )
1044 return finish_reason_to_wire(finish_reason, _TARGET)
1046 @staticmethod
1047 def _safety_ratings_from_passthrough(
1048 raw: Any,
1049 ) -> list[GeminiSafetyRating] | None:
1050 """Rebuild safety ratings from passthrough dicts."""
1051 if not isinstance(raw, list):
1052 return None
1053 ratings = [
1054 GeminiSafetyRating.from_dict(item) for item in raw if isinstance(item, dict)
1055 ]
1056 return ratings or None
1058 @classmethod
1059 def _grounding_from_passthrough(cls, raw: Any) -> GeminiGroundingMetadata | None:
1060 """Rebuild grounding metadata from a passthrough dict."""
1061 if not isinstance(raw, dict):
1062 return None
1063 return GeminiGroundingMetadata.from_dict(raw)
1065 @staticmethod
1066 def _prompt_feedback_from_passthrough(
1067 raw: Any,
1068 ) -> GeminiPromptFeedback | None:
1069 """Rebuild prompt feedback from a passthrough dict."""
1070 if not isinstance(raw, dict):
1071 return None
1072 return GeminiPromptFeedback.from_dict(raw)
1075def _config_number(generation_config: dict[str, Any], key: str) -> int | float | None:
1076 """Read a numeric generation config value when well-typed."""
1077 value = generation_config.get(key)
1078 if isinstance(value, (int, float)) and not isinstance(value, bool):
1079 return value
1080 return None
1083def _config_int(generation_config: dict[str, Any], key: str) -> int | None:
1084 """Read an integer generation config value when well-typed."""
1085 value = _config_number(generation_config, key)
1086 if value is None or isinstance(value, float):
1087 return None
1088 return value