« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 07:19 +0800
1"""IR-to-wire item building for the OpenAI Responses mapper."""
3from __future__ import annotations
5from typing import TYPE_CHECKING, Any
7from lexigram.ai.relay.context import ConversionContext
8from lexigram.ai.relay.mappers.base import record_loss
9from lexigram.ai.relay.mappers.openai_responses.utils import (
10 _TARGET,
11 _arguments_to_wire,
12)
13from lexigram.contracts.ai.llm import ChatMessage
14from lexigram.contracts.ai.multimodal import ImageBase64Part, ImageUrlPart, TextPart
15from lexigram.contracts.ai.relay.dto import ResponsesItem
16from lexigram.contracts.ai.relay.ir import RelayRequest
18if TYPE_CHECKING:
19 from lexigram.ai.relay.mappers.openai_responses import OpenAIResponsesMapper
22class ItemsMixin:
23 """Serializes canonical messages into wire response items."""
25 def _message_to_item(
26 self: OpenAIResponsesMapper,
27 message: ChatMessage,
28 context: ConversionContext,
29 ) -> ResponsesItem:
30 """Convert a canonical message into a wire message item."""
31 data: dict[str, Any] = {"role": message.role}
32 if message.metadata and message.metadata.get("item_id"):
33 data["id"] = message.metadata["item_id"]
34 files = (message.metadata or {}).get("input_files")
35 has_files = isinstance(files, list) and any(
36 isinstance(item, dict) for item in files
37 )
38 if isinstance(message.content, str):
39 if message.content and not has_files:
40 data["content"] = message.content
41 elif message.content or has_files:
42 parts = self._message_content_parts(message, context)
43 if parts:
44 data["content"] = parts
45 else:
46 parts = self._message_content_parts(message, context)
47 if parts:
48 data["content"] = parts
49 return ResponsesItem(**data)
51 @staticmethod
52 def _message_content_parts(
53 message: ChatMessage, context: ConversionContext
54 ) -> list[dict[str, Any]]:
55 """Serialize canonical content into wire message parts."""
56 parts: list[dict[str, Any]] = []
57 content = message.content
58 if isinstance(content, str):
59 if content:
60 parts.append({"type": "input_text", "text": content})
61 elif isinstance(content, list):
62 for part in content:
63 if isinstance(part, TextPart):
64 parts.append({"type": "input_text", "text": part.text})
65 elif isinstance(part, ImageUrlPart):
66 parts.append(
67 {
68 "type": "input_image",
69 "image_url": part.url,
70 }
71 )
72 elif isinstance(part, ImageBase64Part):
73 parts.append(
74 {
75 "type": "input_image",
76 "image_url": f"data:{part.media_type};base64,{part.data}",
77 }
78 )
79 else:
80 record_loss(
81 context,
82 field="message.content",
83 target=_TARGET,
84 reason="unknown_content_part",
85 )
86 files = (message.metadata or {}).get("input_files")
87 if isinstance(files, list):
88 for file_part in files:
89 if isinstance(file_part, dict):
90 parts.append(file_part)
91 return parts
93 def _tool_calls_to_items(
94 self: OpenAIResponsesMapper,
95 message: ChatMessage,
96 context: ConversionContext,
97 ) -> list[ResponsesItem]:
98 """Convert a tool-calling assistant turn into wire items."""
99 items: list[ResponsesItem] = []
100 if self._message_content_parts(message, context):
101 items.append(self._message_to_item(message, context))
102 else:
103 items.append(ResponsesItem(role="assistant", content=""))
104 item_ids = (message.metadata or {}).get("function_call_item_ids")
105 for index, tool in enumerate(message.tool_calls or []):
106 data: dict[str, Any] = {
107 "type": "function_call",
108 "call_id": tool.id or f"call_{index + 1}",
109 "name": tool.function.name if tool.function else "",
110 "arguments": _arguments_to_wire(
111 tool.function.arguments if tool.function else {}
112 ),
113 }
114 if isinstance(item_ids, list) and index < len(item_ids) and item_ids[index]:
115 data["id"] = item_ids[index]
116 items.append(ResponsesItem(**data))
117 return items
119 def _thinking_to_items(
120 self: OpenAIResponsesMapper, message: ChatMessage
121 ) -> list[ResponsesItem]:
122 """Convert canonical thinking blocks into a reasoning item."""
123 summary = list(message.thinking_blocks or [])
124 if not summary:
125 return []
126 data: dict[str, Any] = {"type": "reasoning", "summary": summary}
127 if message.metadata and message.metadata.get("item_id"):
128 data["id"] = message.metadata["item_id"]
129 return [ResponsesItem(**data)]
131 @staticmethod
132 def _tool_result_to_item(message: ChatMessage) -> ResponsesItem:
133 """Convert a canonical tool message into a wire output item."""
134 content = message.content
135 if isinstance(content, list):
136 output = "".join(
137 part.text for part in content if isinstance(part, TextPart)
138 )
139 else:
140 output = content or ""
141 data: dict[str, Any] = {
142 "type": "function_call_output",
143 "output": str(output),
144 "call_id": message.tool_call_id or "call_0",
145 }
146 if message.metadata and message.metadata.get("item_id"):
147 data["id"] = message.metadata["item_id"]
148 return ResponsesItem(**data)
150 @staticmethod
151 def _system_text(message: ChatMessage) -> str:
152 """Extract text from a system-role canonical message."""
153 content = message.content
154 if isinstance(content, str):
155 return content
156 return "".join(part.text for part in content if isinstance(part, TextPart))
158 @staticmethod
159 def _web_search_items(request: RelayRequest) -> list[ResponsesItem]:
160 """Restore preserved web_search_call input items."""
161 raw = request.metadata.get("input_web_search_calls")
162 items: list[ResponsesItem] = []
163 if isinstance(raw, list):
164 for entry in raw:
165 if not isinstance(entry, dict):
166 continue
167 data = dict(entry)
168 items.append(
169 ResponsesItem(
170 type=str(data.pop("type", "web_search_call")),
171 id=data.pop("id", None),
172 passthrough=data,
173 )
174 )
175 return items