Coverage for src / lexigram / ai / relay / mappers / openai_chat.py: 87%
294 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-08 23:08 +0800
1"""OpenAI Chat Completions request and response mapper.
3Converts the OpenAI Chat Completions wire DTOs
4(:class:`OpenAIChatRequest` / :class:`OpenAIChatResponse`) into the
5canonical relay IR and back. Stream conversion is handled by the shared
6stream lifecycle task and reports ``unsupported_feature`` until then.
7"""
9from __future__ import annotations
11from dataclasses import replace
12from typing import Any, cast
14from lexigram.ai.relay.context import ConversionContext
15from lexigram.ai.relay.errors import translate, unsupported_feature, unsupported_format
16from lexigram.ai.relay.mappers.base import new_uuid, record_loss
17from lexigram.contracts.ai.agents import ToolDefinition
18from lexigram.contracts.ai.exceptions import RelayError
19from lexigram.contracts.ai.llm import ChatMessage, FunctionCall, ToolCall
20from lexigram.contracts.ai.multimodal import ImageBase64Part, ImageUrlPart, TextPart
21from lexigram.contracts.ai.relay.dto import (
22 OpenAIChatChoice,
23 OpenAIChatMessage,
24 OpenAIChatRequest,
25 OpenAIChatResponse,
26)
27from lexigram.contracts.ai.relay.ir import (
28 RelayRequest,
29 RelayResponse,
30 StreamDelta,
31 StreamState,
32 normalize_finish_reason,
33)
34from lexigram.contracts.ai.relay.types import RelayFormat, RelayUsage
35from lexigram.contracts.ai.thinking import ThinkingConfig, ThinkingResult
36from lexigram.contracts.core.result import Err, Ok, Result
37from lexigram.serialization import dumps_str
39__all__ = ["OpenAIChatMapper"]
41_TARGET = RelayFormat.OPENAI_CHAT
42_MESSAGE_METADATA_INTERNAL = {"function_call_item_ids"}
45def _tool_calls_to_ir(
46 wire: list[dict[str, Any]] | None,
47) -> list[ToolCall] | None:
48 """Convert wire tool-call dicts into canonical ``ToolCall`` objects."""
49 if not wire:
50 return None
51 tool_calls: list[ToolCall] = []
52 for item in wire:
53 function = item.get("function")
54 name = function.get("name", "") if isinstance(function, dict) else ""
55 arguments = function.get("arguments", {}) if isinstance(function, dict) else {}
56 tool_calls.append(
57 ToolCall(
58 id=str(item.get("id", "")),
59 type=str(item.get("type", "function")),
60 function=FunctionCall(name=str(name), arguments=arguments),
61 )
62 )
63 return tool_calls
66def _tool_call_to_wire(tool_call: ToolCall) -> dict[str, Any]:
67 """Serialize one canonical ``ToolCall`` as a wire dict."""
68 arguments: Any = tool_call.function.arguments if tool_call.function else {}
69 if isinstance(arguments, dict):
70 arguments = dumps_str(arguments)
71 elif not isinstance(arguments, str):
72 arguments = ""
73 return {
74 "id": tool_call.id,
75 "type": "function",
76 "function": {
77 "name": tool_call.function.name if tool_call.function else "",
78 "arguments": arguments,
79 },
80 }
83def _extract_text(
84 content: str | list[dict[str, Any]] | None,
85 context: ConversionContext,
86 *,
87 field: str,
88) -> str:
89 """Extract the text portion of wire content for flattened fields."""
90 if content is None:
91 return ""
92 if isinstance(content, str):
93 return content
94 texts: list[str] = []
95 lost = False
96 for part in content:
97 if isinstance(part, dict) and part.get("type") == "text":
98 texts.append(str(part.get("text", "")))
99 else:
100 lost = True
101 if lost:
102 record_loss(
103 context, field=field, target=_TARGET, reason="non_text_parts_dropped"
104 )
105 return "".join(texts)
108class OpenAIChatMapper:
109 """Bidirectional OpenAI Chat Completions converter.
111 Attributes:
112 format: The wire format this mapper handles.
113 """
115 format = _TARGET
117 def request_to_ir(
118 self, payload: Any, *, context: ConversionContext
119 ) -> Result[RelayRequest, RelayError]:
120 """Convert an ``OpenAIChatRequest`` into canonical ``RelayRequest``.
122 Args:
123 payload: A wire request DTO.
124 context: Per-conversion context with loss sink.
126 Returns:
127 Ok(request) on success, Err(relay_error) on malformed payload.
128 """
129 if not isinstance(payload, OpenAIChatRequest):
130 return Err(
131 unsupported_format(
132 f"expected OpenAIChatRequest, got {type(payload).__name__}"
133 )
134 )
135 system_parts: list[str] = []
136 messages: list[ChatMessage] = []
137 for position, message in enumerate(payload.messages):
138 if message.role == "system":
139 text = _extract_text(
140 message.content,
141 context,
142 field=f"system_message[{position}].content",
143 )
144 system_parts.append(text)
145 if position > 0:
146 record_loss(
147 context,
148 field="system_message",
149 target=_TARGET,
150 reason="system_message_reordered",
151 )
152 continue
153 content: str | list[Any]
154 if isinstance(message.content, list):
155 content = self._wire_parts_to_ir(message.content, context)
156 elif message.content is None:
157 content = ""
158 else:
159 content = message.content
160 tool_calls = _tool_calls_to_ir(message.tool_calls)
161 messages.append(
162 ChatMessage(
163 role=message.role,
164 content=cast("str | list[Any]", content),
165 name=message.name,
166 tool_call_id=message.tool_call_id,
167 tool_calls=tool_calls,
168 metadata=dict(message.passthrough) or None,
169 )
170 )
171 max_tokens = self._normalize_max_tokens(payload, context)
172 stop_sequences = (
173 [payload.stop]
174 if isinstance(payload.stop, str)
175 else (
176 [s for s in payload.stop if isinstance(s, str)] if payload.stop else []
177 )
178 )
179 include_usage = False
180 if isinstance(payload.stream_options, dict):
181 include_usage = bool(payload.stream_options.get("include_usage", False))
182 thinking: ThinkingConfig | None = None
183 reasoning = payload.reasoning
184 if isinstance(reasoning, dict):
185 thinking = ThinkingConfig(effort=reasoning.get("effort"))
186 metadata: dict[str, Any] = {}
187 if reasoning is not None:
188 metadata["reasoning"] = reasoning
189 if payload.stream_options is not None:
190 metadata["stream_options"] = payload.stream_options
191 if payload.service_tier is not None:
192 metadata["service_tier"] = payload.service_tier
193 return Ok(
194 RelayRequest(
195 model=context.normalize_model(payload.model),
196 messages=messages,
197 system="\n".join(system_parts) if system_parts else None,
198 tools=self._tools_to_ir(payload.tools, context),
199 tool_choice=payload.tool_choice,
200 temperature=payload.temperature,
201 top_p=payload.top_p,
202 max_tokens=max_tokens,
203 stop_sequences=stop_sequences,
204 response_format=payload.response_format,
205 stream=payload.stream,
206 include_usage=include_usage,
207 parallel_tool_calls=payload.parallel_tool_calls,
208 thinking=thinking,
209 metadata=metadata,
210 passthrough=dict(payload.passthrough),
211 )
212 )
214 def ir_to_request(
215 self, request: RelayRequest, *, context: ConversionContext
216 ) -> Result[Any, RelayError]:
217 """Convert canonical ``RelayRequest`` into an ``OpenAIChatRequest``.
219 Args:
220 request: Canonical request IR.
221 context: Per-conversion context with loss sink.
223 Returns:
224 Ok(request) on success, Err(relay_error) on failure.
225 """
226 try:
227 messages: list[OpenAIChatMessage] = []
228 if request.system:
229 messages.append(
230 OpenAIChatMessage(role="system", content=request.system)
231 )
232 for message in request.messages:
233 prepared = message
234 if message.role == "assistant" and message.tool_calls:
235 if any(not tool_call.id for tool_call in message.tool_calls):
236 prepared = replace(
237 message,
238 tool_calls=[
239 tool_call
240 if tool_call.id
241 else replace(tool_call, id=f"call_{index + 1}")
242 for index, tool_call in enumerate(message.tool_calls)
243 ],
244 )
245 elif message.role == "tool" and not message.tool_call_id:
246 prepared = replace(message, tool_call_id="call_0")
247 messages.append(self._message_from_ir(prepared, context))
248 stream_options = self._stream_options_from_ir(request)
249 reasoning = self._reasoning_from_ir(request, context)
250 if request.metadata.get("max_tokens_kind") == "max_completion_tokens":
251 max_completion_tokens: int | None = request.max_tokens
252 max_tokens: int | None = None
253 else:
254 max_completion_tokens = None
255 max_tokens = request.max_tokens
256 return Ok(
257 OpenAIChatRequest(
258 model=context.resolve_model(request.model),
259 messages=messages,
260 temperature=request.temperature,
261 top_p=request.top_p,
262 max_tokens=max_tokens,
263 max_completion_tokens=max_completion_tokens,
264 stream=request.stream,
265 stream_options=stream_options,
266 tools=(
267 [self._tool_from_ir(tool) for tool in request.tools]
268 if request.tools
269 else None
270 ),
271 tool_choice=request.tool_choice,
272 parallel_tool_calls=request.parallel_tool_calls,
273 stop=self._stop_from_ir(request.stop_sequences),
274 response_format=request.response_format,
275 reasoning=reasoning,
276 service_tier=request.metadata.get("service_tier"),
277 passthrough={
278 **request.passthrough,
279 **{
280 key: value
281 for key, value in request.metadata.items()
282 if key
283 not in {
284 "service_tier",
285 "reasoning",
286 "stream_options",
287 "generation_config",
288 "safety_settings",
289 "tool_config",
290 "max_tokens_kind",
291 }
292 },
293 },
294 )
295 )
296 except (RelayError, ValueError, TypeError, KeyError) as exc:
297 return Err(translate(exc, detail="ir_to_request"))
299 def response_to_ir(
300 self, payload: Any, *, context: ConversionContext
301 ) -> Result[RelayResponse, RelayError]:
302 """Convert an ``OpenAIChatResponse`` into canonical ``RelayResponse``.
304 Args:
305 payload: A wire response DTO.
306 context: Per-conversion context with loss sink.
308 Returns:
309 Ok(response) on success, Err(relay_error) on malformed payload.
310 """
311 if not isinstance(payload, OpenAIChatResponse):
312 return Err(
313 unsupported_format(
314 f"expected OpenAIChatResponse, got {type(payload).__name__}"
315 )
316 )
317 try:
318 passthrough = dict(payload.passthrough)
319 if payload.system_fingerprint is not None:
320 passthrough["system_fingerprint"] = payload.system_fingerprint
321 choice = payload.choices[0] if payload.choices else None
322 if len(payload.choices) > 1:
323 record_loss(
324 context,
325 field="choices",
326 target=_TARGET,
327 reason="multiple_choices_collapsed",
328 )
329 message = choice.message if choice is not None else None
330 content = ""
331 tool_calls: list[ToolCall] = []
332 thinking: ThinkingResult | None = None
333 if message is not None:
334 content = self._message_text_to_ir(message, context)
335 tool_calls = list(_tool_calls_to_ir(message.tool_calls) or [])
336 thinking = self._reasoning_from_message(message, payload.usage)
337 return Ok(
338 RelayResponse(
339 model=payload.model,
340 id=payload.id,
341 created=payload.created,
342 content=content,
343 thinking=thinking,
344 tool_calls=tool_calls,
345 finish_reason=normalize_finish_reason(
346 choice.finish_reason if choice is not None else None
347 ),
348 usage=self._usage_from_wire(payload.usage),
349 passthrough=passthrough,
350 )
351 )
352 except (RelayError, ValueError, TypeError, KeyError) as exc:
353 return Err(translate(exc, detail="response_to_ir"))
355 def ir_to_response(
356 self, response: RelayResponse, *, context: ConversionContext
357 ) -> Result[Any, RelayError]:
358 """Convert canonical ``RelayResponse`` into an ``OpenAIChatResponse``.
360 Args:
361 response: Canonical response IR.
362 context: Per-conversion context with loss sink.
364 Returns:
365 Ok(response) on success, Err(relay_error) on failure.
366 """
367 try:
368 passthrough = dict(response.passthrough)
369 system_fingerprint = passthrough.pop("system_fingerprint", None)
370 content: str | None = response.content or None
371 tool_calls: list[dict[str, Any]] = []
372 for tool in response.tool_calls:
373 wire = _tool_call_to_wire(tool)
374 if not wire["id"]:
375 wire["id"] = f"call_{new_uuid()}"
376 tool_calls.append(wire)
377 message = OpenAIChatMessage(
378 role="assistant",
379 content=content,
380 tool_calls=tool_calls or None,
381 )
382 finish_reason = (
383 "tool_calls" if response.tool_calls else response.finish_reason
384 )
385 return Ok(
386 OpenAIChatResponse(
387 id=response.id or f"chatcmpl-{new_uuid()}",
388 model=context.resolve_model(response.model),
389 created=response.created or 0,
390 choices=[
391 OpenAIChatChoice(
392 index=0,
393 message=message,
394 finish_reason=finish_reason,
395 )
396 ],
397 usage=self._usage_to_wire(response.usage),
398 system_fingerprint=(system_fingerprint),
399 passthrough=passthrough,
400 )
401 )
402 except (RelayError, ValueError, TypeError, KeyError) as exc:
403 return Err(translate(exc, detail="ir_to_response"))
405 def stream_to_delta(
406 self, event: Any, *, state: StreamState
407 ) -> Result[tuple[StreamDelta, ...], RelayError]:
408 """Stream conversion is deferred to the shared stream lifecycle task."""
409 return Err(
410 unsupported_feature("openai_chat stream conversion is not implemented yet")
411 )
413 def delta_to_stream(
414 self, delta: StreamDelta, *, state: StreamState
415 ) -> Result[tuple[Any, ...], RelayError]:
416 """Stream conversion is deferred to the shared stream lifecycle task."""
417 return Err(
418 unsupported_feature("openai_chat stream conversion is not implemented yet")
419 )
421 # -- helpers -------------------------------------------------------------
423 @staticmethod
424 def _tools_to_ir(
425 tools: list[dict[str, Any]] | None, context: ConversionContext
426 ) -> list[ToolDefinition]:
427 """Convert wire tool dicts into canonical ``ToolDefinition`` objects."""
428 definitions: list[ToolDefinition] = []
429 if not tools:
430 return definitions
431 for index, tool in enumerate(tools):
432 if not isinstance(tool, dict):
433 record_loss(
434 context,
435 field=f"tools[{index}]",
436 target=_TARGET,
437 reason="non_dict_tool_dropped",
438 )
439 continue
440 if tool.get("type", "function") != "function":
441 record_loss(
442 context,
443 field=f"tools[{index}]",
444 target=_TARGET,
445 reason="non_function_tool_dropped",
446 )
447 continue
448 function = tool.get("function")
449 if not isinstance(function, dict):
450 record_loss(
451 context,
452 field=f"tools[{index}]",
453 target=_TARGET,
454 reason="missing_function",
455 )
456 continue
457 parameters = function.get("parameters", {})
458 definitions.append(
459 ToolDefinition(
460 name=str(function.get("name", "")),
461 description=str(function.get("description", "")),
462 parameters=parameters if isinstance(parameters, dict) else {},
463 )
464 )
465 return definitions
467 @staticmethod
468 def _normalize_max_tokens(
469 payload: OpenAIChatRequest, context: ConversionContext
470 ) -> int | None:
471 """Normalize ``max_tokens``/``max_completion_tokens`` into one value."""
472 max_tokens = payload.max_tokens
473 max_completion_tokens = payload.max_completion_tokens
474 if max_tokens is not None and max_completion_tokens is not None:
475 if max_tokens != max_completion_tokens:
476 record_loss(
477 context,
478 field="max_completion_tokens",
479 target=_TARGET,
480 reason="conflicts_with_max_tokens",
481 )
482 return max_completion_tokens
483 if max_completion_tokens is not None:
484 return max_completion_tokens
485 return max_tokens
487 @staticmethod
488 def _wire_parts_to_ir(
489 parts: list[dict[str, Any]], context: ConversionContext
490 ) -> list[Any]:
491 """Convert wire content parts into canonical content parts."""
492 converted: list[Any] = []
493 for part in parts:
494 if not isinstance(part, dict):
495 converted.append(TextPart(text=str(part)))
496 continue
497 part_type = part.get("type")
498 if part_type == "text":
499 converted.append(TextPart(text=str(part.get("text", ""))))
500 elif part_type == "image_url":
501 image = part.get("image_url")
502 if isinstance(image, dict):
503 converted.append(
504 ImageUrlPart(
505 url=str(image.get("url", "")),
506 detail=cast("Any", image.get("detail", "auto") or "auto"),
507 )
508 )
509 else:
510 converted.append(TextPart(text=str(part)))
511 else:
512 record_loss(
513 context,
514 field=part_type or "part",
515 target=_TARGET,
516 reason="unknown_part_type",
517 )
518 return converted
520 @staticmethod
521 def _message_text_to_ir(
522 message: OpenAIChatMessage, context: ConversionContext
523 ) -> str:
524 """Extract text content from a response message."""
525 content = message.content
526 if isinstance(content, str):
527 return content
528 if isinstance(content, list):
529 return _extract_text(content, context, field="message.content")
530 return ""
532 @staticmethod
533 def _reasoning_from_message(
534 message: OpenAIChatMessage, usage: dict[str, Any] | None
535 ) -> ThinkingResult | None:
536 """Build a ``ThinkingResult`` from message reasoning passthrough."""
537 raw = message.passthrough.get("reasoning") or message.passthrough.get(
538 "reasoning_content"
539 )
540 reasoning_text: str | None = None
541 if isinstance(raw, str) and raw:
542 reasoning_text = raw
543 elif isinstance(raw, dict) and isinstance(raw.get("content"), str):
544 reasoning_text = raw["content"]
545 if reasoning_text is None:
546 return None
547 tokens: int | None = None
548 if isinstance(usage, dict):
549 details = usage.get("completion_tokens_details")
550 if isinstance(details, dict) and isinstance(
551 details.get("reasoning_tokens"), int
552 ):
553 tokens = details["reasoning_tokens"]
554 return ThinkingResult(content=reasoning_text, tokens=tokens)
556 @staticmethod
557 def _usage_from_wire(usage: dict[str, Any] | None) -> RelayUsage | None:
558 """Map a wire usage dict into canonical ``RelayUsage``."""
559 if not isinstance(usage, dict):
560 return None
561 prompt_details = usage.get("prompt_tokens_details")
562 completion_details = usage.get("completion_tokens_details")
563 audio_tokens = usage.get("audio_tokens")
564 prompt_tokens = int(usage.get("prompt_tokens", 0) or 0)
565 completion_tokens = int(usage.get("completion_tokens", 0) or 0)
566 return RelayUsage(
567 prompt_tokens=prompt_tokens,
568 completion_tokens=completion_tokens,
569 cache_read_tokens=(
570 int(prompt_details.get("cached_tokens", 0) or 0)
571 if isinstance(prompt_details, dict)
572 else 0
573 ),
574 cache_creation_tokens=(
575 int(
576 prompt_details.get("cached_creation_tokens", 0)
577 or prompt_details.get("cache_write_tokens", 0)
578 or 0
579 )
580 if isinstance(prompt_details, dict)
581 else 0
582 ),
583 reasoning_tokens=(
584 int(completion_details.get("reasoning_tokens", 0) or 0)
585 if isinstance(completion_details, dict)
586 else 0
587 ),
588 audio_input_tokens=(
589 int(audio_tokens.get("input_tokens", 0) or 0)
590 if isinstance(audio_tokens, dict)
591 else 0
592 ),
593 audio_output_tokens=(
594 int(audio_tokens.get("output_tokens", 0) or 0)
595 if isinstance(audio_tokens, dict)
596 else 0
597 ),
598 input_tokens=int(usage.get("input_tokens", 0) or 0),
599 output_tokens=int(usage.get("output_tokens", 0) or 0),
600 )
602 def _message_from_ir(
603 self, message: ChatMessage, context: ConversionContext
604 ) -> OpenAIChatMessage:
605 """Convert a canonical message into an ``OpenAIChatMessage``."""
606 content: Any
607 if isinstance(message.content, list):
608 parts: list[Any] = []
609 for part in message.content:
610 if isinstance(part, TextPart):
611 parts.append({"type": "text", "text": part.text})
612 elif isinstance(part, ImageUrlPart):
613 parts.append(
614 {
615 "type": "image_url",
616 "image_url": part.url,
617 }
618 )
619 elif isinstance(part, ImageBase64Part):
620 image_url: dict[str, Any] = {
621 "url": f"data:{part.media_type};base64,{part.data}",
622 }
623 if part.detail:
624 image_url["detail"] = part.detail
625 parts.append(
626 {
627 "type": "image_url",
628 "image_url": image_url,
629 }
630 )
631 else:
632 record_loss(
633 context,
634 field="message.content",
635 target=_TARGET,
636 reason="unknown_content_part",
637 )
638 content = parts
639 elif message.content == "":
640 content = None
641 else:
642 content = message.content
643 return OpenAIChatMessage(
644 role=message.role,
645 content=cast("str | None", content),
646 name=message.name,
647 tool_call_id=message.tool_call_id,
648 tool_calls=(
649 [_tool_call_to_wire(tool) for tool in message.tool_calls]
650 if message.tool_calls
651 else None
652 ),
653 passthrough={
654 key: value
655 for key, value in (message.metadata or {}).items()
656 if key not in _MESSAGE_METADATA_INTERNAL
657 },
658 )
660 @staticmethod
661 def _tool_from_ir(tool: ToolDefinition) -> dict[str, Any]:
662 """Serialize a canonical ``ToolDefinition`` as a wire tool dict."""
663 return {
664 "type": "function",
665 "function": {
666 "name": tool.name,
667 "description": tool.description,
668 "parameters": tool.parameters,
669 },
670 }
672 @staticmethod
673 def _stream_options_from_ir(request: RelayRequest) -> dict[str, Any] | None:
674 """Rebuild ``stream_options`` from canonical stream settings."""
675 raw = request.metadata.get("stream_options")
676 options: dict[str, Any] = dict(raw) if isinstance(raw, dict) else {}
677 if request.include_usage:
678 options["include_usage"] = True
679 elif "include_usage" in options:
680 options.pop("include_usage")
681 if not options:
682 return None
683 return options
685 def _reasoning_from_ir(
686 self, request: RelayRequest, context: ConversionContext
687 ) -> dict[str, Any] | None:
688 """Rebuild the OpenAI ``reasoning`` config from canonical thinking."""
689 thinking = request.thinking
690 if thinking is not None:
691 if thinking.effort is not None:
692 return {"effort": thinking.effort}
693 record_loss(
694 context,
695 field="thinking",
696 target=_TARGET,
697 reason="effort_only_supported",
698 )
699 raw = request.metadata.get("reasoning")
700 if isinstance(raw, dict):
701 return dict(raw)
702 return None
704 @staticmethod
705 def _stop_from_ir(stop_sequences: list[str]) -> str | list[str] | None:
706 """Rebuild a wire ``stop`` value from canonical stop sequences."""
707 if not stop_sequences:
708 return None
709 if len(stop_sequences) == 1:
710 return stop_sequences[0]
711 return list(stop_sequences)
713 @staticmethod
714 def _usage_to_wire(usage: RelayUsage | None) -> dict[str, Any] | None:
715 """Serialize canonical ``RelayUsage`` into a wire usage dict.
717 Mirrors relaykit's ``dto.Usage`` serialization: the detail
718 containers and responses-style ``input_tokens``/``output_tokens``
719 are always present (zeros included), and cache-write counters are
720 added only when non-zero.
721 """
722 if usage is None:
723 return None
724 data: dict[str, Any] = {
725 "prompt_tokens": usage.prompt_tokens,
726 "completion_tokens": usage.completion_tokens,
727 "total_tokens": usage.total_tokens,
728 "prompt_tokens_details": {"cached_tokens": usage.cache_read_tokens},
729 "completion_tokens_details": {"reasoning_tokens": usage.reasoning_tokens},
730 "input_tokens": usage.input_tokens,
731 "output_tokens": usage.output_tokens,
732 }
733 if usage.cache_creation_tokens:
734 data["prompt_tokens_details"]["cached_creation_tokens"] = (
735 usage.cache_creation_tokens
736 )
737 data["prompt_tokens_details"]["cache_write_tokens"] = (
738 usage.cache_creation_tokens
739 )
740 if usage.audio_input_tokens or usage.audio_output_tokens:
741 data["audio_tokens"] = {
742 "input_tokens": usage.audio_input_tokens,
743 "output_tokens": usage.audio_output_tokens,
744 }
745 return data