Coverage for src / lexigram / contracts / ai / relay / dto / openai_chat.py: 44%
186 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-19 05:41 +0800
1"""OpenAI Chat Completions wire DTO family."""
3from __future__ import annotations
5from dataclasses import dataclass, field
6from typing import Any
8from lexigram.contracts.ai.relay.dto.common import require_field
10__all__ = [
11 "OpenAIChatChoice",
12 "OpenAIChatMessage",
13 "OpenAIChatRequest",
14 "OpenAIChatResponse",
15 "OpenAIChatStreamChoice",
16 "OpenAIChatStreamChunk",
17 "OpenAIChatStreamDelta",
18]
21@dataclass(frozen=True)
22class OpenAIChatMessage:
23 """A message in OpenAI Chat Completions format.
25 Attributes:
26 role: ``system``, ``user``, ``assistant``, ``tool``, ``function``.
27 content: String content, or ``None`` for tool-call turns.
28 name: Optional author name.
29 tool_call_id: Id of the tool call this message answers.
30 tool_calls: Tool calls on an assistant message, or ``None``.
31 passthrough: Unknown fields preserved verbatim.
32 """
34 role: str
35 content: str | None = None
36 name: str | None = None
37 tool_call_id: str | None = None
38 tool_calls: list[dict[str, Any]] | None = None
39 passthrough: dict[str, Any] = field(default_factory=dict)
41 def to_dict(self) -> dict[str, Any]:
42 """Serialize to wire dict, omitting ``None`` optional fields."""
43 data: dict[str, Any] = {**self.passthrough, "role": self.role}
45 data["content"] = self.content
46 if self.name is not None:
47 data["name"] = self.name
48 if self.tool_call_id is not None:
49 data["tool_call_id"] = self.tool_call_id
50 if self.tool_calls is not None:
51 data["tool_calls"] = self.tool_calls
52 return data
54 @classmethod
55 def from_dict(cls, data: dict[str, Any]) -> OpenAIChatMessage:
56 """Build a message from a wire dict, capturing unknown keys."""
57 known = {"role", "content", "name", "tool_call_id", "tool_calls"}
58 return cls(
59 role=data.get("role", "user"),
60 content=data.get("content"),
61 name=data.get("name"),
62 tool_call_id=data.get("tool_call_id"),
63 tool_calls=data.get("tool_calls"),
64 passthrough={k: v for k, v in data.items() if k not in known},
65 )
68@dataclass(frozen=True)
69class OpenAIChatRequest:
70 """OpenAI Chat Completions request body.
72 Attributes:
73 model: Model name.
74 messages: Message list.
75 temperature: Sampling temperature (``None`` = omitted upstream).
76 top_p: Nucleus sampling (``None`` = omitted upstream).
77 max_tokens: Max output tokens (``None`` = omitted upstream).
78 max_completion_tokens: Max completion tokens (``None`` = omitted
79 upstream; normalized against ``max_tokens`` by the mapper).
80 stream: Whether the caller wants a stream.
81 stream_options: ``{"include_usage": bool}`` or ``None``.
82 tools: Raw tool definitions, or ``None``.
83 tool_choice: Tool choice directive, or ``None``.
84 parallel_tool_calls: Parallel tool-call flag, or ``None``.
85 stop: Stop string or list of strings, or ``None``.
86 response_format: JSON-mode config, or ``None``.
87 reasoning: Reasoning config (e.g. ``{"effort": ...}``), or ``None``.
88 service_tier: Service tier, or ``None``.
89 passthrough: Unknown fields preserved verbatim.
90 """
92 model: str
93 messages: list[OpenAIChatMessage]
94 temperature: float | None = None
95 top_p: float | None = None
96 max_tokens: int | None = None
97 max_completion_tokens: int | None = None
98 stream: bool = False
99 stream_options: dict[str, Any] | None = None
100 tools: list[dict[str, Any]] | None = None
101 tool_choice: Any | None = None
102 parallel_tool_calls: bool | None = None
103 stop: str | list[str] | None = None
104 response_format: dict[str, Any] | None = None
105 reasoning: dict[str, Any] | None = None
106 service_tier: str | None = None
107 passthrough: dict[str, Any] = field(default_factory=dict)
109 def to_dict(self) -> dict[str, Any]:
110 """Serialize to wire dict, omitting ``None`` optional fields."""
111 data: dict[str, Any] = {
112 **self.passthrough,
113 "model": self.model,
114 "messages": [m.to_dict() for m in self.messages],
115 }
116 if self.temperature is not None:
117 data["temperature"] = self.temperature
118 if self.top_p is not None:
119 data["top_p"] = self.top_p
120 if self.max_tokens is not None:
121 data["max_tokens"] = self.max_tokens
122 if self.max_completion_tokens is not None:
123 data["max_completion_tokens"] = self.max_completion_tokens
124 data["stream"] = self.stream
125 if self.stream_options is not None:
126 data["stream_options"] = self.stream_options
127 if self.tools is not None:
128 data["tools"] = self.tools
129 if self.tool_choice is not None:
130 data["tool_choice"] = self.tool_choice
131 if self.parallel_tool_calls is not None:
132 data["parallel_tool_calls"] = self.parallel_tool_calls
133 if self.stop is not None:
134 data["stop"] = self.stop
135 if self.response_format is not None:
136 data["response_format"] = self.response_format
137 if self.reasoning is not None:
138 data["reasoning"] = self.reasoning
139 if self.service_tier is not None:
140 data["service_tier"] = self.service_tier
141 return data
143 @classmethod
144 def from_dict(cls, data: dict[str, Any]) -> OpenAIChatRequest:
145 """Build a request from a wire dict, capturing unknown keys.
147 Raises:
148 RelayError: With code ``malformed_payload`` when ``model``
149 is absent.
150 """
151 known = {
152 "model",
153 "messages",
154 "temperature",
155 "top_p",
156 "max_tokens",
157 "max_completion_tokens",
158 "stream",
159 "stream_options",
160 "tools",
161 "tool_choice",
162 "parallel_tool_calls",
163 "stop",
164 "response_format",
165 "reasoning",
166 "service_tier",
167 }
168 return cls(
169 model=require_field(data, "model"),
170 messages=[OpenAIChatMessage.from_dict(m) for m in data.get("messages", [])],
171 temperature=data.get("temperature"),
172 top_p=data.get("top_p"),
173 max_tokens=data.get("max_tokens"),
174 max_completion_tokens=data.get("max_completion_tokens"),
175 stream=bool(data.get("stream", False)),
176 stream_options=data.get("stream_options"),
177 tools=data.get("tools"),
178 tool_choice=data.get("tool_choice"),
179 parallel_tool_calls=data.get("parallel_tool_calls"),
180 stop=data.get("stop"),
181 response_format=data.get("response_format"),
182 reasoning=data.get("reasoning"),
183 service_tier=data.get("service_tier"),
184 passthrough={k: v for k, v in data.items() if k not in known},
185 )
188@dataclass(frozen=True)
189class OpenAIChatChoice:
190 """One choice in a non-streamed completion response.
192 Attributes:
193 index: Choice index.
194 message: Assistant message, or ``None``.
195 finish_reason: ``stop``, ``length``, ``tool_calls``, etc.
196 logprobs: Token log-probability info, or ``None``.
197 passthrough: Unknown fields preserved verbatim.
198 """
200 index: int = 0
201 message: OpenAIChatMessage | None = None
202 finish_reason: str | None = None
203 logprobs: Any | None = None
204 passthrough: dict[str, Any] = field(default_factory=dict)
206 def to_dict(self) -> dict[str, Any]:
207 """Serialize to wire dict, omitting ``None`` optional fields."""
208 data: dict[str, Any] = {**self.passthrough, "index": self.index}
209 if self.message is not None:
210 data["message"] = self.message.to_dict()
211 if self.finish_reason is not None:
212 data["finish_reason"] = self.finish_reason
213 if self.logprobs is not None:
214 data["logprobs"] = self.logprobs
215 return data
217 @classmethod
218 def from_dict(cls, data: dict[str, Any]) -> OpenAIChatChoice:
219 """Build a choice from a wire dict, capturing unknown keys."""
220 known = {"index", "message", "finish_reason", "logprobs"}
221 message = data.get("message")
222 return cls(
223 index=data.get("index", 0),
224 message=OpenAIChatMessage.from_dict(message)
225 if isinstance(message, dict)
226 else None,
227 finish_reason=data.get("finish_reason"),
228 logprobs=data.get("logprobs"),
229 passthrough={k: v for k, v in data.items() if k not in known},
230 )
233@dataclass(frozen=True)
234class OpenAIChatResponse:
235 """Non-streamed Chat Completions response body.
237 Attributes:
238 id: Completion id.
239 model: Model name.
240 choices: Completion choices.
241 object: Object type (``chat.completion``).
242 created: Unix timestamp.
243 usage: Raw usage dict, or ``None``.
244 system_fingerprint: System fingerprint, or ``None``.
245 passthrough: Unknown fields preserved verbatim.
246 """
248 id: str
249 model: str
250 choices: list[OpenAIChatChoice]
251 object: str = "chat.completion"
252 created: int = 0
253 usage: dict[str, Any] | None = None
254 system_fingerprint: str | None = None
255 passthrough: dict[str, Any] = field(default_factory=dict)
257 def to_dict(self) -> dict[str, Any]:
258 """Serialize to wire dict, omitting ``None`` optional fields."""
259 data: dict[str, Any] = {
260 **self.passthrough,
261 "id": self.id,
262 "object": self.object,
263 "created": self.created,
264 "model": self.model,
265 "choices": [c.to_dict() for c in self.choices],
266 }
267 if self.usage is not None:
268 data["usage"] = self.usage
269 if self.system_fingerprint is not None:
270 data["system_fingerprint"] = self.system_fingerprint
271 return data
273 @classmethod
274 def from_dict(cls, data: dict[str, Any]) -> OpenAIChatResponse:
275 """Build a response from a wire dict, capturing unknown keys.
277 Raises:
278 RelayError: With code ``malformed_payload`` when ``id`` or
279 ``model`` is absent.
280 """
281 known = {
282 "id",
283 "object",
284 "created",
285 "model",
286 "choices",
287 "usage",
288 "system_fingerprint",
289 }
290 return cls(
291 id=require_field(data, "id"),
292 object=data.get("object", "chat.completion"),
293 created=data.get("created", 0),
294 model=require_field(data, "model"),
295 choices=[OpenAIChatChoice.from_dict(c) for c in data.get("choices", [])],
296 usage=data.get("usage"),
297 system_fingerprint=data.get("system_fingerprint"),
298 passthrough={k: v for k, v in data.items() if k not in known},
299 )
302@dataclass(frozen=True)
303class OpenAIChatStreamDelta:
304 """Delta payload inside one stream choice.
306 Attributes:
307 role: Role announcement (``assistant``), or ``None``.
308 content: Text delta, or ``None``.
309 reasoning_content: Reasoning text delta, or ``None``.
310 tool_calls: Partial tool-call fragments (raw wire shape), or ``None``.
311 refusal: Refusal text delta, or ``None``.
312 passthrough: Unknown fields preserved verbatim.
313 """
315 role: str | None = None
316 content: str | None = None
317 reasoning_content: str | None = None
318 tool_calls: list[dict[str, Any]] | None = None
319 refusal: str | None = None
320 passthrough: dict[str, Any] = field(default_factory=dict)
322 def to_dict(self) -> dict[str, Any]:
323 """Serialize to wire dict, omitting ``None`` optional fields."""
324 data: dict[str, Any] = {**self.passthrough}
325 if self.role is not None:
326 data["role"] = self.role
327 if self.content is not None:
328 data["content"] = self.content
329 if self.reasoning_content is not None:
330 data["reasoning_content"] = self.reasoning_content
331 if self.tool_calls is not None:
332 data["tool_calls"] = self.tool_calls
333 if self.refusal is not None:
334 data["refusal"] = self.refusal
335 return data
337 @classmethod
338 def from_dict(cls, data: dict[str, Any]) -> OpenAIChatStreamDelta:
339 """Build a delta from a wire dict, capturing unknown keys."""
340 known = {"role", "content", "reasoning_content", "tool_calls", "refusal"}
341 return cls(
342 role=data.get("role"),
343 content=data.get("content"),
344 reasoning_content=data.get("reasoning_content"),
345 tool_calls=data.get("tool_calls"),
346 refusal=data.get("refusal"),
347 passthrough={k: v for k, v in data.items() if k not in known},
348 )
351@dataclass(frozen=True)
352class OpenAIChatStreamChoice:
353 """One choice inside a stream chunk.
355 Attributes:
356 index: Choice index.
357 delta: Delta payload, or ``None``.
358 finish_reason: Terminal finish reason, or ``None``.
359 logprobs: Token log-probability info, or ``None``.
360 passthrough: Unknown fields preserved verbatim.
361 """
363 index: int = 0
364 delta: OpenAIChatStreamDelta | None = None
365 finish_reason: str | None = None
366 logprobs: Any | None = None
367 passthrough: dict[str, Any] = field(default_factory=dict)
369 def to_dict(self) -> dict[str, Any]:
370 """Serialize to wire dict, omitting ``None`` optional fields."""
371 data: dict[str, Any] = {**self.passthrough, "index": self.index}
372 if self.delta is not None:
373 data["delta"] = self.delta.to_dict()
374 if self.finish_reason is not None:
375 data["finish_reason"] = self.finish_reason
376 if self.logprobs is not None:
377 data["logprobs"] = self.logprobs
378 return data
380 @classmethod
381 def from_dict(cls, data: dict[str, Any]) -> OpenAIChatStreamChoice:
382 """Build a choice from a wire dict, capturing unknown keys."""
383 known = {"index", "delta", "finish_reason", "logprobs"}
384 delta = data.get("delta")
385 return cls(
386 index=data.get("index", 0),
387 delta=OpenAIChatStreamDelta.from_dict(delta)
388 if isinstance(delta, dict)
389 else None,
390 finish_reason=data.get("finish_reason"),
391 logprobs=data.get("logprobs"),
392 passthrough={k: v for k, v in data.items() if k not in known},
393 )
396@dataclass(frozen=True)
397class OpenAIChatStreamChunk:
398 """One SSE chunk of a streamed Chat Completions response.
400 Attributes:
401 id: Completion id.
402 model: Model name.
403 choices: Stream choices.
404 object: Object type (``chat.completion.chunk``).
405 created: Unix timestamp.
406 usage: Final usage dict (usage-only terminal chunk), or ``None``.
407 system_fingerprint: System fingerprint, or ``None``.
408 passthrough: Unknown fields preserved verbatim.
409 """
411 id: str
412 model: str
413 choices: list[OpenAIChatStreamChoice]
414 object: str = "chat.completion.chunk"
415 created: int = 0
416 usage: dict[str, Any] | None = None
417 system_fingerprint: str | None = None
418 passthrough: dict[str, Any] = field(default_factory=dict)
420 def to_dict(self) -> dict[str, Any]:
421 """Serialize to wire dict, omitting ``None`` optional fields."""
422 data: dict[str, Any] = {
423 **self.passthrough,
424 "id": self.id,
425 "object": self.object,
426 "created": self.created,
427 "model": self.model,
428 "choices": [c.to_dict() for c in self.choices],
429 }
430 if self.usage is not None:
431 data["usage"] = self.usage
432 if self.system_fingerprint is not None:
433 data["system_fingerprint"] = self.system_fingerprint
434 return data
436 @classmethod
437 def from_dict(cls, data: dict[str, Any]) -> OpenAIChatStreamChunk:
438 """Build a chunk from a wire dict, capturing unknown keys.
440 Raises:
441 RelayError: With code ``malformed_payload`` when ``id`` or
442 ``model`` is absent.
443 """
444 known = {
445 "id",
446 "object",
447 "created",
448 "model",
449 "choices",
450 "usage",
451 "system_fingerprint",
452 }
453 return cls(
454 id=require_field(data, "id"),
455 object=data.get("object", "chat.completion.chunk"),
456 created=data.get("created", 0),
457 model=require_field(data, "model"),
458 choices=[
459 OpenAIChatStreamChoice.from_dict(c) for c in data.get("choices", [])
460 ],
461 usage=data.get("usage"),
462 system_fingerprint=data.get("system_fingerprint"),
463 passthrough={k: v for k, v in data.items() if k not in known},
464 )