« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 07:19 +0800
1"""Canonical relay IR → Gemini wire request conversion."""
3from __future__ import annotations
5from dataclasses import replace
6from typing import Any
8from lexigram.ai.relay.context import ConversionContext
9from lexigram.ai.relay.errors import translate
10from lexigram.ai.relay.mappers.base import record_loss
11from lexigram.ai.relay.mappers.gemini_request._shared import (
12 _MIME_KEY,
13 _SAFETY_CATEGORIES,
14 _SCHEMA_TYPE_MAP,
15 _TARGET,
16 _THOUGHT_SIGNATURE_BYPASS,
17 _tool_call_to_part,
18)
19from lexigram.ai.relay.media import resolve_media
20from lexigram.contracts.ai.agents import ToolDefinition
21from lexigram.contracts.ai.exceptions import RelayError
22from lexigram.contracts.ai.llm import ChatMessage
23from lexigram.contracts.ai.multimodal import (
24 ContentPart,
25 ImageBase64Part,
26 ImageUrlPart,
27 TextPart,
28)
29from lexigram.contracts.ai.relay.dto import GeminiContent, GeminiPart, GeminiRequest
30from lexigram.contracts.ai.relay.ir import RelayRequest
31from lexigram.contracts.core.result import Err, Ok, Result
32from lexigram.serialization import loads_str
35def ir_to_request(
36 request: RelayRequest, *, context: ConversionContext
37) -> Result[Any, RelayError]:
38 """Convert canonical ``RelayRequest`` into a ``GeminiRequest``.
40 Args:
41 request: Canonical request IR.
42 context: Per-conversion context with loss sink.
44 Returns:
45 Ok(request) on success, Err(relay_error) on failure.
46 """
47 try:
48 system_parts: list[str] = []
49 contents: list[GeminiContent] = []
50 tool_names, tool_names_by_id = _tool_name_resolver(request)
51 for message in request.messages:
52 if message.role == "system":
53 system_parts.append(_text_from_content(message.content))
54 continue
55 content = _content_from_ir(
56 message, request.model, tool_names, tool_names_by_id, context
57 )
58 if content.is_err():
59 return content
60 contents.append(content.unwrap())
61 if request.system:
62 system_parts.append(request.system)
63 return Ok(
64 GeminiRequest(
65 contents=contents,
66 system_instruction=(
67 {"parts": [{"text": text} for text in system_parts]}
68 if system_parts
69 else None
70 ),
71 generation_config=_generation_config_from_ir(request, context),
72 safety_settings=_safety_settings_from_ir(request, context),
73 tools=_tools_from_ir(request.tools),
74 tool_config=_tool_config_from_ir(request),
75 passthrough=_request_passthrough(request),
76 )
77 )
78 except (RelayError, ValueError, TypeError, KeyError) as exc:
79 return Err(translate(exc, detail="ir_to_request"))
82def _content_from_ir(
83 message: ChatMessage,
84 model: str,
85 tool_names: list[str],
86 tool_names_by_id: dict[str, str],
87 context: ConversionContext,
88) -> Result[GeminiContent, RelayError]:
89 """Convert one canonical message into a Gemini content turn.
91 Args:
92 message: Canonical message to serialize.
93 model: The selected model, for capability lookups.
94 tool_names: Positional resolver for tool-message function
95 names when the canonical ``tool_call_id`` carries no
96 stable link.
97 tool_names_by_id: Resolver keyed by canonical tool-call id.
98 context: Per-conversion context with loss sink.
100 Returns:
101 Ok(content) on success, Err(relay_error) on failure.
102 """
103 if message.role == "tool":
104 response: Any = message.content
105 if isinstance(response, list):
106 response = _text_from_content(message.content)
107 if isinstance(response, str):
108 try:
109 response = loads_str(response)
110 except ValueError:
111 pass
112 name = tool_names_by_id.get(message.tool_call_id or "", "")
113 if not name and tool_names:
114 name = tool_names.pop(0)
115 if isinstance(response, str):
116 response = {"content": response}
117 return Ok(
118 GeminiContent(
119 role="user",
120 parts=[
121 GeminiPart(
122 function_response={
123 "name": name,
124 "response": response,
125 }
126 )
127 ],
128 )
129 )
130 if message.role == "assistant":
131 parts = _assistant_parts_from_ir(message, model, context)
132 if parts.is_err():
133 return Err(parts.unwrap_err())
134 return Ok(GeminiContent(role="model", parts=parts.unwrap()))
135 if message.role == "user":
136 parts = _user_parts_from_ir(message.content, context)
137 if parts.is_err():
138 return Err(parts.unwrap_err())
139 return Ok(GeminiContent(role="user", parts=parts.unwrap()))
140 record_loss(
141 context,
142 field="messages",
143 target=_TARGET,
144 reason=f"unknown_role_{message.role}_dropped",
145 )
146 return Ok(GeminiContent(role="user", parts=[GeminiPart(text="")]))
149def _tool_name_resolver(
150 request: RelayRequest,
151) -> tuple[list[str], dict[str, str]]:
152 """Return tool-message name resolvers for a request.
154 Gemini ``functionResponse`` blocks name the function, not the
155 call id. Tool messages resolve their function name by canonical
156 ``tool_call_id`` first; unresolved messages fall back to
157 positional order against the assistant tool calls that preceded
158 them.
159 """
160 names: list[str] = []
161 names_by_id: dict[str, str] = {}
162 for message in request.messages:
163 if message.role == "assistant":
164 for tool_call in message.tool_calls or []:
165 if tool_call.function:
166 names.append(tool_call.function.name)
167 if tool_call.id:
168 names_by_id[tool_call.id] = tool_call.function.name
169 return names, names_by_id
172def _assistant_parts_from_ir(
173 message: ChatMessage, model: str, context: ConversionContext
174) -> Result[list[GeminiPart], RelayError]:
175 """Rebuild Gemini model parts from an assistant message.
177 When the thought-signature bypass policy is enabled the thinking
178 blocks are folded away and a bypass ``thoughtSignature`` is
179 attached to the first function-call part (relaykit's
180 ``FunctionCallThoughtSignatureEnabled`` behavior). Otherwise
181 thinking blocks are re-emitted as native Gemini thought parts.
182 """
183 attach_signature = context.options.gemini.thought_signature_bypass
184 parts: list[GeminiPart] = []
185 if not attach_signature:
186 for block in message.thinking_blocks or []:
187 if not isinstance(block, dict):
188 continue
189 signature = block.get("thoughtSignature")
190 parts.append(
191 GeminiPart(
192 text=str(block.get("text", "")),
193 thought=True,
194 thought_signature=str(signature) if signature else None,
195 )
196 )
197 content_parts = _user_parts_from_ir(message.content, context)
198 if content_parts.is_err():
199 return Err(content_parts.unwrap_err())
200 parts.extend(content_parts.unwrap())
201 for tool_call in message.tool_calls or []:
202 parts.append(_tool_call_to_part(tool_call))
203 if attach_signature:
204 parts = _attach_thought_signature(parts)
205 return Ok(parts)
208def _attach_thought_signature(parts: list[GeminiPart]) -> list[GeminiPart]:
209 """Attach the relaykit thought-signature bypass value to model parts.
211 The signature lands on the first function-call part, or on the
212 first non-empty text part when the message carries no tool calls.
213 """
214 rebuilt: list[GeminiPart] = []
215 attached = False
216 for part in parts:
217 current = part
218 if not attached and part.function_call is not None:
219 current = replace(part, thought_signature=_THOUGHT_SIGNATURE_BYPASS)
220 attached = True
221 rebuilt.append(current)
222 if not attached:
223 for index, part in enumerate(rebuilt):
224 if part.text:
225 rebuilt[index] = replace(
226 part, thought_signature=_THOUGHT_SIGNATURE_BYPASS
227 )
228 break
229 return rebuilt
232def _user_parts_from_ir(
233 content: str | list[ContentPart], context: ConversionContext
234) -> Result[list[GeminiPart], RelayError]:
235 """Convert canonical content into Gemini parts."""
236 if isinstance(content, str):
237 return Ok([GeminiPart(text=content)] if content else [])
238 parts: list[GeminiPart] = []
239 for part in content:
240 if isinstance(part, TextPart):
241 parts.append(GeminiPart(text=part.text))
242 elif isinstance(part, ImageBase64Part):
243 parts.append(
244 GeminiPart(
245 inline_data={
246 _MIME_KEY: part.media_type,
247 "data": part.data,
248 }
249 )
250 )
251 elif isinstance(part, ImageUrlPart):
252 resolved = _resolve_image(part, context)
253 if resolved.is_err():
254 return Err(resolved.unwrap_err())
255 media_type, data = resolved.unwrap()
256 parts.append(GeminiPart(inline_data={_MIME_KEY: media_type, "data": data}))
257 else:
258 record_loss(
259 context,
260 field="message.content",
261 target=_TARGET,
262 reason="unknown_content_part",
263 )
264 return Ok(parts)
267def _resolve_image(
268 part: ImageUrlPart, context: ConversionContext
269) -> Result[tuple[str, str], RelayError]:
270 """Resolve a URL or data-URI image for Gemini.
272 Data URIs decode locally; URLs go through the context resolver.
273 """
274 resolved = resolve_media(
275 part.url,
276 context,
277 field="message.content",
278 target=_TARGET,
279 lossy=False,
280 )
281 if resolved.is_err():
282 return Err(resolved.unwrap_err())
283 image = resolved.unwrap()
284 assert image is not None # lossy=False never drops media # noqa: S101
285 return Ok(image)
288def _text_from_content(content: str | list[ContentPart]) -> str:
289 """Extract plain text from canonical content."""
290 if isinstance(content, str):
291 return content
292 return "".join(part.text for part in content if isinstance(part, TextPart))
295def _generation_config_from_ir(
296 request: RelayRequest, context: ConversionContext
297) -> dict[str, Any]:
298 """Rebuild ``generationConfig`` from protocol metadata and canonical fields."""
299 raw = request.metadata.get("generation_config")
300 config: dict[str, Any] = dict(raw) if isinstance(raw, dict) else {}
301 for key in (
302 "temperature",
303 "topP",
304 "topK",
305 "maxOutputTokens",
306 "stopSequences",
307 "responseMimeType",
308 "responseSchema",
309 "thinkingConfig",
310 ):
311 config.pop(key, None)
312 if request.temperature is not None:
313 config["temperature"] = request.temperature
314 if request.top_p is not None:
315 config["topP"] = request.top_p
316 if request.top_k is not None:
317 config["topK"] = request.top_k
318 if request.max_tokens is not None:
319 config["maxOutputTokens"] = request.max_tokens
320 if request.stop_sequences:
321 config["stopSequences"] = list(request.stop_sequences)
322 if request.response_format is not None:
323 if request.response_format.get("type") == "json_object":
324 config["responseMimeType"] = "application/json"
325 if isinstance(request.response_format.get("schema"), dict):
326 config["responseSchema"] = request.response_format["schema"]
327 thinking_config = _thinking_config_from_ir(request, context)
328 if thinking_config is not None:
329 config["thinkingConfig"] = thinking_config
330 if "responseModalities" not in config and context.supports_image_generation(
331 request.model
332 ):
333 config["responseModalities"] = ["TEXT", "IMAGE"]
334 return config
337def _thinking_config_from_ir(
338 request: RelayRequest, context: ConversionContext
339) -> dict[str, Any] | None:
340 """Build a Gemini ``thinkingConfig`` from canonical thinking."""
341 if request.thinking is not None:
342 record_loss(
343 context,
344 field="thinking",
345 target=_TARGET,
346 reason="thinking_not_supported",
347 )
348 if (
349 context.options.gemini.thinking_adapter_enabled
350 and context.options.gemini.thinking_budget
351 ):
352 return {"thinkingBudget": context.options.gemini.thinking_budget}
353 return None
356def _safety_settings_from_ir(
357 request: RelayRequest, context: ConversionContext
358) -> list[dict[str, Any]] | None:
359 """Rebuild Gemini safety settings from metadata or the callback."""
360 raw = request.metadata.get("safety_settings")
361 if isinstance(raw, list):
362 preserved = [dict(item) for item in raw if isinstance(item, dict)]
363 return preserved or None
364 collected: list[dict[str, Any]] = []
365 for category in _SAFETY_CATEGORIES:
366 threshold = context.safety_setting(category)
367 if threshold and isinstance(threshold, str):
368 collected.append({"category": category, "threshold": threshold})
369 return collected or None
372def _tools_from_ir(tools: list[ToolDefinition]) -> list[dict[str, Any]] | None:
373 """Serialize canonical tools as Gemini function declarations."""
374 if not tools:
375 return None
376 return [
377 {
378 "functionDeclarations": [
379 {
380 "name": tool.name,
381 "description": tool.description,
382 "parameters": _upper_schema_types(tool.parameters),
383 }
384 for tool in tools
385 ]
386 }
387 ]
390def _request_passthrough(request: RelayRequest) -> dict[str, Any]:
391 """Carry canonical passthrough state into the request wire payload."""
392 return dict(request.passthrough)
395def _upper_schema_types(parameters: dict[str, Any]) -> dict[str, Any]:
396 """Uppercase Gemini schema type markers recursively.
398 Gemini function declarations require ``STRING``/``OBJECT`` type
399 values; canonical schemas carry the lowercase JSON-Schema form.
400 """
401 out: dict[str, Any] = {}
402 for key, value in parameters.items():
403 if key == "type" and isinstance(value, str):
404 out[key] = _SCHEMA_TYPE_MAP.get(value, value)
405 elif isinstance(value, dict):
406 out[key] = _upper_schema_types(value)
407 elif isinstance(value, list):
408 out[key] = [
409 _upper_schema_types(item) if isinstance(item, dict) else item
410 for item in value
411 ]
412 else:
413 out[key] = value
414 return out
417def _tool_config_from_ir(request: RelayRequest) -> dict[str, Any] | None:
418 """Rebuild a Gemini ``toolConfig`` from canonical tool choice."""
419 raw = request.metadata.get("tool_config")
420 if isinstance(raw, dict):
421 return dict(raw)
422 choice = request.tool_choice
423 if isinstance(choice, dict):
424 name = choice.get("function", {})
425 if isinstance(name, dict):
426 name = name.get("name")
427 if isinstance(name, str) and name:
428 return {
429 "functionCallingConfig": {
430 "mode": "ANY",
431 "allowedFunctionNames": [name],
432 }
433 }
434 return {"functionCallingConfig": {"mode": "ANY"}}
435 if isinstance(choice, str):
436 mode = {"auto": "AUTO", "none": "NONE", "required": "ANY"}.get(choice, "AUTO")
437 return {"functionCallingConfig": {"mode": mode}}
438 return None