« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 07:19 +0800
1"""Gemini wire request → canonical relay IR conversion."""
3from __future__ import annotations
5from typing import Any
7from lexigram.ai.relay.context import ConversionContext
8from lexigram.ai.relay.errors import translate, unsupported_format
9from lexigram.ai.relay.mappers.base import record_loss
10from lexigram.ai.relay.mappers.gemini_request._shared import (
11 _MIME_KEY,
12 _TARGET,
13 _tool_call_from_part,
14)
15from lexigram.contracts.ai.agents import ToolDefinition
16from lexigram.contracts.ai.exceptions import RelayError
17from lexigram.contracts.ai.llm import ChatMessage, ToolCall
18from lexigram.contracts.ai.multimodal import (
19 ContentPart,
20 ImageBase64Part,
21 TextPart,
22)
23from lexigram.contracts.ai.relay.dto import GeminiContent, GeminiRequest
24from lexigram.contracts.ai.relay.ir import RelayRequest
25from lexigram.contracts.ai.thinking import ThinkingConfig
26from lexigram.contracts.core.result import Err, Ok, Result
27from lexigram.serialization import dumps_str
30def request_to_ir(
31 payload: Any, *, context: ConversionContext
32) -> Result[RelayRequest, RelayError]:
33 """Convert a ``GeminiRequest`` into canonical ``RelayRequest``.
35 Args:
36 payload: A wire request DTO.
37 context: Per-conversion context with loss sink.
39 Returns:
40 Ok(request) on success, Err(relay_error) on malformed payload.
41 """
42 if not isinstance(payload, GeminiRequest):
43 return Err(
44 unsupported_format(f"expected GeminiRequest, got {type(payload).__name__}")
45 )
46 try:
47 messages = [
48 chat_message
49 for index, content in enumerate(payload.contents)
50 for chat_message in _content_to_ir(content, context, index)
51 ]
52 metadata: dict[str, Any] = {}
53 if payload.safety_settings is not None:
54 metadata["safety_settings"] = [
55 dict(item) for item in payload.safety_settings
56 ]
57 if payload.tool_config is not None:
58 metadata["tool_config"] = dict(payload.tool_config)
59 generation_config = dict(payload.generation_config)
60 if generation_config:
61 metadata["generation_config"] = generation_config
62 return Ok(
63 RelayRequest(
64 model=str(payload.passthrough.get("model", "")).strip(),
65 messages=messages,
66 system=_system_to_ir(payload.system_instruction),
67 tools=_tools_to_ir(payload.tools),
68 temperature=_config_number(generation_config, "temperature"),
69 top_p=_config_number(generation_config, "topP"),
70 top_k=_config_int(generation_config, "topK"),
71 max_tokens=_config_int(generation_config, "maxOutputTokens"),
72 stop_sequences=[
73 str(item)
74 for item in generation_config.get("stopSequences", [])
75 if isinstance(item, str)
76 ],
77 response_format=_response_format_to_ir(generation_config),
78 thinking=_thinking_to_ir(generation_config),
79 metadata=metadata,
80 passthrough=dict(payload.passthrough),
81 )
82 )
83 except (RelayError, ValueError, TypeError, KeyError) as exc:
84 return Err(translate(exc, detail="request_to_ir"))
87def _system_to_ir(
88 system_instruction: dict[str, Any] | None,
89) -> str | None:
90 """Extract system text from a Gemini ``systemInstruction`` dict."""
91 if not isinstance(system_instruction, dict):
92 return None
93 parts = system_instruction.get("parts")
94 if not isinstance(parts, list):
95 return None
96 texts: list[str] = []
97 for part in parts:
98 if isinstance(part, dict) and part.get("text") is not None:
99 texts.append(str(part["text"]))
100 return "\n".join(texts)
103def _tools_to_ir(tools: list[dict[str, Any]] | None) -> list[ToolDefinition]:
104 """Convert Gemini wire tools into canonical ``ToolDefinition`` objects."""
105 definitions: list[ToolDefinition] = []
106 for tool in tools or []:
107 if not isinstance(tool, dict):
108 continue
109 declarations = tool.get("functionDeclarations")
110 if not isinstance(declarations, list):
111 continue
112 for declaration in declarations:
113 if not isinstance(declaration, dict):
114 continue
115 parameters = declaration.get("parameters")
116 definitions.append(
117 ToolDefinition(
118 name=str(declaration.get("name", "")),
119 description=str(declaration.get("description", "")),
120 parameters=parameters if isinstance(parameters, dict) else {},
121 )
122 )
123 return definitions
126def _content_to_ir(
127 content: GeminiContent, context: ConversionContext, index: int
128) -> list[ChatMessage]:
129 """Convert one Gemini content turn into canonical messages."""
130 if content.role == "model":
131 return [_assistant_to_ir(content, context)]
132 if content.role == "user":
133 return _user_to_ir(content, context, index)
134 if content.role == "function":
135 return _function_to_ir(content, context, index)
136 record_loss(
137 context,
138 field=f"contents[{index}].role",
139 target=_TARGET,
140 reason="unknown_role_dropped",
141 )
142 return []
145def _assistant_to_ir(content: GeminiContent, context: ConversionContext) -> ChatMessage:
146 """Convert a model content turn, separating thinking/tool parts."""
147 text_parts: list[str] = []
148 thinking_blocks: list[dict[str, Any]] = []
149 tool_calls: list[ToolCall] = []
150 for part in content.parts:
151 if part.thought:
152 thinking_blocks.append(
153 {
154 "thought": True,
155 "text": part.text or "",
156 "thoughtSignature": part.thought_signature or "",
157 }
158 )
159 elif part.text is not None:
160 text_parts.append(part.text)
161 elif part.function_call is not None:
162 tool_calls.append(_tool_call_from_part(part))
163 else:
164 record_loss(
165 context,
166 field="content.part",
167 target=_TARGET,
168 reason="unrepresentable_part_dropped",
169 )
170 return ChatMessage(
171 role="assistant",
172 content="".join(text_parts),
173 tool_calls=tool_calls or None,
174 thinking_blocks=thinking_blocks or None,
175 )
178def _user_to_ir(
179 content: GeminiContent, context: ConversionContext, index: int
180) -> list[ChatMessage]:
181 """Convert a user content turn into canonical parts and tool results."""
182 parts: list[ContentPart] = []
183 tool_results: list[ChatMessage] = []
184 for part in content.parts:
185 if part.text is not None:
186 parts.append(TextPart(text=part.text))
187 elif part.inline_data is not None:
188 inline = part.inline_data
189 parts.append(
190 ImageBase64Part(
191 data=str(inline.get("data", "")),
192 media_type=str(inline.get(_MIME_KEY, "")),
193 detail="auto",
194 )
195 )
196 elif part.function_response is not None:
197 tool_results.append(_function_response_to_ir(part.function_response))
198 else:
199 record_loss(
200 context,
201 field=f"contents[{index}].part",
202 target=_TARGET,
203 reason="unrepresentable_part_dropped",
204 )
205 turns: list[ChatMessage] = []
206 if parts:
207 turns.append(
208 ChatMessage(
209 role="user",
210 content=(
211 parts[0].text
212 if len(parts) == 1 and isinstance(parts[0], TextPart)
213 else list(parts)
214 ),
215 )
216 )
217 turns.extend(tool_results)
218 if not turns:
219 record_loss(
220 context,
221 field=f"contents[{index}]",
222 target=_TARGET,
223 reason="empty_message_dropped",
224 )
225 return turns
228def _function_to_ir(
229 content: GeminiContent, context: ConversionContext, index: int
230) -> list[ChatMessage]:
231 """Convert a function content turn into canonical tool messages."""
232 messages: list[ChatMessage] = []
233 for part in content.parts:
234 if part.function_response is not None:
235 messages.append(_function_response_to_ir(part.function_response))
236 else:
237 record_loss(
238 context,
239 field=f"contents[{index}].part",
240 target=_TARGET,
241 reason="unrepresentable_part_dropped",
242 )
243 return messages
246def _function_response_to_ir(response: dict[str, Any]) -> ChatMessage:
247 """Convert a ``functionResponse`` dict into a canonical tool message."""
248 payload = response.get("response")
249 if isinstance(payload, str):
250 text = payload
251 else:
252 text = dumps_str(payload) if payload is not None else ""
253 return ChatMessage(role="tool", content=text, tool_call_id="")
256def _response_format_to_ir(
257 generation_config: dict[str, Any],
258) -> dict[str, Any] | None:
259 """Derive a canonical response format from the generation config."""
260 mime = generation_config.get("responseMimeType")
261 if not isinstance(mime, str):
262 return None
263 if mime == "application/json":
264 return {"type": "json_object"}
265 return None
268def _thinking_to_ir(generation_config: dict[str, Any]) -> ThinkingConfig | None:
269 """Extract canonical thinking config from ``thinkingConfig``."""
270 config = generation_config.get("thinkingConfig")
271 if not isinstance(config, dict):
272 return None
273 level = config.get("thinkingLevel")
274 budget = config.get("thinkingBudget")
275 if isinstance(level, str) and level:
276 return ThinkingConfig(level=level)
277 if isinstance(budget, int):
278 return ThinkingConfig(budget_tokens=budget)
279 return None
282def _config_number(generation_config: dict[str, Any], key: str) -> int | float | None:
283 """Read a numeric generation config value when well-typed."""
284 value = generation_config.get(key)
285 if isinstance(value, (int, float)) and not isinstance(value, bool):
286 return value
287 return None
290def _config_int(generation_config: dict[str, Any], key: str) -> int | None:
291 """Read an integer generation config value when well-typed."""
292 value = _config_number(generation_config, key)
293 if value is None or isinstance(value, float):
294 return None
295 return value