« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 07:19 +0800
1"""Wire → IR parsing 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 _TARGET,
11 _extract_text,
12)
13from lexigram.contracts.ai.agents import ToolDefinition
14from lexigram.contracts.ai.multimodal import ImageUrlPart, TextPart
15from lexigram.contracts.ai.relay.dto import OpenAIChatMessage, OpenAIChatRequest
16from lexigram.contracts.ai.relay.types import RelayUsage
17from lexigram.contracts.ai.thinking import ThinkingResult
20class WireToIRMixin:
21 """Static parsers turning OpenAI Chat wire shapes into canonical IR."""
23 @staticmethod
24 def _tools_to_ir(
25 tools: list[dict[str, Any]] | None, context: ConversionContext
26 ) -> list[ToolDefinition]:
27 """Convert wire tool dicts into canonical ``ToolDefinition`` objects."""
28 definitions: list[ToolDefinition] = []
29 if not tools:
30 return definitions
31 for index, tool in enumerate(tools):
32 if not isinstance(tool, dict):
33 record_loss(
34 context,
35 field=f"tools[{index}]",
36 target=_TARGET,
37 reason="non_dict_tool_dropped",
38 )
39 continue
40 if tool.get("type", "function") != "function":
41 record_loss(
42 context,
43 field=f"tools[{index}]",
44 target=_TARGET,
45 reason="non_function_tool_dropped",
46 )
47 continue
48 function = tool.get("function")
49 if not isinstance(function, dict):
50 record_loss(
51 context,
52 field=f"tools[{index}]",
53 target=_TARGET,
54 reason="missing_function",
55 )
56 continue
57 parameters = function.get("parameters", {})
58 definitions.append(
59 ToolDefinition(
60 name=str(function.get("name", "")),
61 description=str(function.get("description", "")),
62 parameters=parameters if isinstance(parameters, dict) else {},
63 )
64 )
65 return definitions
67 @staticmethod
68 def _normalize_max_tokens(
69 payload: OpenAIChatRequest, context: ConversionContext
70 ) -> int | None:
71 """Normalize ``max_tokens``/``max_completion_tokens`` into one value."""
72 max_tokens = payload.max_tokens
73 max_completion_tokens = payload.max_completion_tokens
74 if max_tokens is not None and max_completion_tokens is not None:
75 if max_tokens != max_completion_tokens:
76 record_loss(
77 context,
78 field="max_completion_tokens",
79 target=_TARGET,
80 reason="conflicts_with_max_tokens",
81 )
82 return max_completion_tokens
83 if max_completion_tokens is not None:
84 return max_completion_tokens
85 return max_tokens
87 @staticmethod
88 def _wire_parts_to_ir(
89 parts: list[dict[str, Any]], context: ConversionContext
90 ) -> list[Any]:
91 """Convert wire content parts into canonical content parts."""
92 converted: list[Any] = []
93 for part in parts:
94 if not isinstance(part, dict):
95 converted.append(TextPart(text=str(part)))
96 continue
97 part_type = part.get("type")
98 if part_type == "text":
99 converted.append(TextPart(text=str(part.get("text", ""))))
100 elif part_type == "image_url":
101 image = part.get("image_url")
102 if isinstance(image, dict):
103 converted.append(
104 ImageUrlPart(
105 url=str(image.get("url", "")),
106 detail=cast("Any", image.get("detail", "auto") or "auto"),
107 )
108 )
109 else:
110 converted.append(TextPart(text=str(part)))
111 else:
112 record_loss(
113 context,
114 field=part_type or "part",
115 target=_TARGET,
116 reason="unknown_part_type",
117 )
118 return converted
120 @staticmethod
121 def _message_text_to_ir(
122 message: OpenAIChatMessage, context: ConversionContext
123 ) -> str:
124 """Extract text content from a response message."""
125 content = message.content
126 if isinstance(content, str):
127 return content
128 if isinstance(content, list):
129 return _extract_text(content, context, field="message.content")
130 return ""
132 @staticmethod
133 def _reasoning_from_message(
134 message: OpenAIChatMessage, usage: dict[str, Any] | None
135 ) -> ThinkingResult | None:
136 """Build a ``ThinkingResult`` from message reasoning passthrough."""
137 raw = message.passthrough.get("reasoning") or message.passthrough.get(
138 "reasoning_content"
139 )
140 reasoning_text: str | None = None
141 if isinstance(raw, str) and raw:
142 reasoning_text = raw
143 elif isinstance(raw, dict) and isinstance(raw.get("content"), str):
144 reasoning_text = raw["content"]
145 if reasoning_text is None:
146 return None
147 tokens: int | None = None
148 if isinstance(usage, dict):
149 details = usage.get("completion_tokens_details")
150 if isinstance(details, dict) and isinstance(
151 details.get("reasoning_tokens"), int
152 ):
153 tokens = details["reasoning_tokens"]
154 return ThinkingResult(content=reasoning_text, tokens=tokens)
156 @staticmethod
157 def _usage_from_wire(usage: dict[str, Any] | None) -> RelayUsage | None:
158 """Map a wire usage dict into canonical ``RelayUsage``."""
159 if not isinstance(usage, dict):
160 return None
161 prompt_details = usage.get("prompt_tokens_details")
162 completion_details = usage.get("completion_tokens_details")
163 audio_tokens = usage.get("audio_tokens")
164 prompt_tokens = int(usage.get("prompt_tokens", 0) or 0)
165 completion_tokens = int(usage.get("completion_tokens", 0) or 0)
166 return RelayUsage(
167 prompt_tokens=prompt_tokens,
168 completion_tokens=completion_tokens,
169 cache_read_tokens=(
170 int(prompt_details.get("cached_tokens", 0) or 0)
171 if isinstance(prompt_details, dict)
172 else 0
173 ),
174 cache_creation_tokens=(
175 int(
176 prompt_details.get("cached_creation_tokens", 0)
177 or prompt_details.get("cache_write_tokens", 0)
178 or 0
179 )
180 if isinstance(prompt_details, dict)
181 else 0
182 ),
183 reasoning_tokens=(
184 int(completion_details.get("reasoning_tokens", 0) or 0)
185 if isinstance(completion_details, dict)
186 else 0
187 ),
188 audio_input_tokens=(
189 int(audio_tokens.get("input_tokens", 0) or 0)
190 if isinstance(audio_tokens, dict)
191 else 0
192 ),
193 audio_output_tokens=(
194 int(audio_tokens.get("output_tokens", 0) or 0)
195 if isinstance(audio_tokens, dict)
196 else 0
197 ),
198 input_tokens=int(usage.get("input_tokens", 0) or 0),
199 output_tokens=int(usage.get("output_tokens", 0) or 0),
200 )