« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 07:19 +0800
1"""IR → wire building helpers for the OpenAI Chat mapper."""
3from __future__ import annotations
5from typing import Any, cast
7from lexigram.ai.relay.context import ConversionContext
8from lexigram.ai.relay.mappers.base import record_loss
9from lexigram.ai.relay.mappers.openai_chat._helpers import (
10 _MESSAGE_METADATA_INTERNAL,
11 _TARGET,
12 _tool_call_to_wire,
13)
14from lexigram.contracts.ai.agents import ToolDefinition
15from lexigram.contracts.ai.llm import ChatMessage
16from lexigram.contracts.ai.multimodal import ImageBase64Part, ImageUrlPart, TextPart
17from lexigram.contracts.ai.relay.dto import OpenAIChatMessage
18from lexigram.contracts.ai.relay.ir import RelayRequest
19from lexigram.contracts.ai.relay.types import RelayUsage
22class IRToWireMixin:
23 """Builders turning canonical IR into OpenAI Chat wire shapes."""
25 def _message_from_ir(
26 self, message: ChatMessage, context: ConversionContext
27 ) -> OpenAIChatMessage:
28 """Convert a canonical message into an ``OpenAIChatMessage``."""
29 content: Any
30 if isinstance(message.content, list):
31 parts: list[Any] = []
32 for part in message.content:
33 if isinstance(part, TextPart):
34 parts.append({"type": "text", "text": part.text})
35 elif isinstance(part, ImageUrlPart):
36 parts.append(
37 {
38 "type": "image_url",
39 "image_url": part.url,
40 }
41 )
42 elif isinstance(part, ImageBase64Part):
43 image_url: dict[str, Any] = {
44 "url": f"data:{part.media_type};base64,{part.data}",
45 }
46 if part.detail:
47 image_url["detail"] = part.detail
48 parts.append(
49 {
50 "type": "image_url",
51 "image_url": image_url,
52 }
53 )
54 else:
55 record_loss(
56 context,
57 field="message.content",
58 target=_TARGET,
59 reason="unknown_content_part",
60 )
61 content = parts
62 elif message.content == "":
63 content = None
64 else:
65 content = message.content
66 return OpenAIChatMessage(
67 role=message.role,
68 content=cast("str | None", content),
69 name=message.name,
70 tool_call_id=message.tool_call_id,
71 tool_calls=(
72 [_tool_call_to_wire(tool) for tool in message.tool_calls]
73 if message.tool_calls
74 else None
75 ),
76 passthrough={
77 key: value
78 for key, value in (message.metadata or {}).items()
79 if key not in _MESSAGE_METADATA_INTERNAL
80 },
81 )
83 @staticmethod
84 def _tool_from_ir(tool: ToolDefinition) -> dict[str, Any]:
85 """Serialize a canonical ``ToolDefinition`` as a wire tool dict."""
86 return {
87 "type": "function",
88 "function": {
89 "name": tool.name,
90 "description": tool.description,
91 "parameters": tool.parameters,
92 },
93 }
95 @staticmethod
96 def _stream_options_from_ir(request: RelayRequest) -> dict[str, Any] | None:
97 """Rebuild ``stream_options`` from canonical stream settings."""
98 raw = request.metadata.get("stream_options")
99 options: dict[str, Any] = dict(raw) if isinstance(raw, dict) else {}
100 if request.include_usage:
101 options["include_usage"] = True
102 elif "include_usage" in options:
103 options.pop("include_usage")
104 if not options:
105 return None
106 return options
108 def _reasoning_from_ir(
109 self, request: RelayRequest, context: ConversionContext
110 ) -> dict[str, Any] | None:
111 """Rebuild the OpenAI ``reasoning`` config from canonical thinking."""
112 thinking = request.thinking
113 if thinking is not None:
114 if thinking.effort is not None:
115 return {"effort": thinking.effort}
116 record_loss(
117 context,
118 field="thinking",
119 target=_TARGET,
120 reason="effort_only_supported",
121 )
122 raw = request.metadata.get("reasoning")
123 if isinstance(raw, dict):
124 return dict(raw)
125 return None
127 @staticmethod
128 def _stop_from_ir(stop_sequences: list[str]) -> str | list[str] | None:
129 """Rebuild a wire ``stop`` value from canonical stop sequences."""
130 if not stop_sequences:
131 return None
132 if len(stop_sequences) == 1:
133 return stop_sequences[0]
134 return list(stop_sequences)
136 @staticmethod
137 def _usage_to_wire(usage: RelayUsage | None) -> dict[str, Any] | None:
138 """Serialize canonical ``RelayUsage`` into a wire usage dict.
140 Mirrors relaykit's ``dto.Usage`` serialization: the detail
141 containers and responses-style ``input_tokens``/``output_tokens``
142 are always present (zeros included), and cache-write counters are
143 added only when non-zero.
144 """
145 if usage is None:
146 return None
147 data: dict[str, Any] = {
148 "prompt_tokens": usage.prompt_tokens,
149 "completion_tokens": usage.completion_tokens,
150 "total_tokens": usage.total_tokens,
151 "prompt_tokens_details": {"cached_tokens": usage.cache_read_tokens},
152 "completion_tokens_details": {"reasoning_tokens": usage.reasoning_tokens},
153 "input_tokens": usage.input_tokens,
154 "output_tokens": usage.output_tokens,
155 }
156 if usage.cache_creation_tokens:
157 data["prompt_tokens_details"]["cached_creation_tokens"] = (
158 usage.cache_creation_tokens
159 )
160 data["prompt_tokens_details"]["cache_write_tokens"] = (
161 usage.cache_creation_tokens
162 )
163 if usage.audio_input_tokens or usage.audio_output_tokens:
164 data["audio_tokens"] = {
165 "input_tokens": usage.audio_input_tokens,
166 "output_tokens": usage.audio_output_tokens,
167 }
168 return data