Coverage for src/lilbee/server/chat_completions_api/translate.py: 100%
170 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-09-08 09:20 +0000
1"""Translation between OpenAI chat-completions models and the canonical types."""
3from __future__ import annotations
5import json
6import time
7from collections.abc import AsyncIterator
8from typing import Literal, assert_never
10from lilbee.core.config.enums import ReasoningMode
11from lilbee.retrieval.reasoning import StreamToken, TagParser, split_reasoning
12from lilbee.server.chat_completions_api.models import (
13 CompletionsImageContent,
14 CompletionsMessage,
15 CompletionsNamedToolChoice,
16 CompletionsRequest,
17 CompletionsResponse,
18 CompletionsResponseChoice,
19 CompletionsResponseMessage,
20 CompletionsResponseToolCall,
21 CompletionsResponseToolCallFunction,
22 CompletionsStreamChoice,
23 CompletionsStreamChunk,
24 CompletionsStreamDelta,
25 CompletionsStreamToolCall,
26 CompletionsStreamToolCallFunction,
27 CompletionsTextContent,
28 CompletionsTool,
29 CompletionsUsage,
30 FinishReason,
31 ToolChoiceMode,
32)
33from lilbee.server.chat_dispatch.canonical import (
34 CanonicalChatRequest,
35 CanonicalMessage,
36 CanonicalResponse,
37 CanonicalStreamEvent,
38 CanonicalTool,
39 CanonicalToolChoice,
40 CanonicalUsage,
41 ContentBlock,
42 ContentBlockDelta,
43 ContentBlockStart,
44 ContentBlockStop,
45 MessageDelta,
46 MessageStart,
47 MessageStop,
48 StopReason,
49 TextBlock,
50 TextDelta,
51 ToolResultBlock,
52 ToolUseBlock,
53 ToolUseDelta,
54)
55from lilbee.server.chat_dispatch.tool_args import parse_tool_arguments
57_TOOL_CHOICE_MODES: dict[ToolChoiceMode, Literal["auto", "any", "none"]] = {
58 ToolChoiceMode.AUTO: "auto",
59 ToolChoiceMode.NONE: "none",
60 ToolChoiceMode.REQUIRED: "any",
61}
63_STOP_REASON_TO_FINISH: dict[StopReason, FinishReason] = {
64 StopReason.END_TURN: FinishReason.STOP,
65 StopReason.MAX_TOKENS: FinishReason.LENGTH,
66 StopReason.TOOL_USE: FinishReason.TOOL_CALLS,
67}
69_TOOL_CALL_ID_REQUIRED = "A tool message requires tool_call_id naming the call it answers."
70_IMAGE_CONTENT_UNSUPPORTED = (
71 "Image content is not supported by /v1/chat/completions yet. Send a text-only request."
72)
75def completions_to_canonical_request(
76 request: CompletionsRequest, *, mode: ReasoningMode = ReasoningMode.SEPARATE
77) -> CanonicalChatRequest:
78 """Translate a validated ``CompletionsRequest`` to the canonical request."""
79 system_parts: list[str] = []
80 messages: list[CanonicalMessage] = []
81 for msg in request.messages:
82 if msg.role == "system":
83 system_parts.append(_system_text(msg))
84 continue
85 messages.append(_message_from_request(msg))
87 return CanonicalChatRequest(
88 model=request.model,
89 messages=messages,
90 system="\n\n".join(system_parts) if system_parts else None,
91 tools=_tools_from_request(request.tools),
92 tool_choice=_tool_choice_from_request(request.tool_choice),
93 temperature=request.temperature,
94 top_p=request.top_p,
95 top_k=request.top_k,
96 max_tokens=request.max_tokens,
97 seed=request.seed,
98 frequency_penalty=request.frequency_penalty,
99 presence_penalty=request.presence_penalty,
100 stop=_stop_from_request(request.stop),
101 stream=request.stream,
102 # OFF asks the template to skip thinking; the other modes only change
103 # presentation, so the template default stands.
104 think=False if mode is ReasoningMode.OFF else None,
105 )
108def canonical_to_completions_response(
109 resp: CanonicalResponse, *, response_id: str, mode: ReasoningMode = ReasoningMode.SEPARATE
110) -> CompletionsResponse:
111 """Translate a canonical chat response to the OpenAI ``chat.completion`` model."""
112 text_parts = [b.text for b in resp.content if isinstance(b, TextBlock)]
113 tool_calls = [_response_tool_call(b) for b in resp.content if isinstance(b, ToolUseBlock)]
114 # lilbee carries a reasoning model's thinking inline as <think>...</think>; the
115 # OpenAI surface reports it in its own field so agents render a clean answer.
116 # INLINE streams the thinking as ordinary content for clients that never
117 # render reasoning_content -- with the tag markers stripped, because a
118 # client that ignores reasoning_content renders raw <think> text literally.
119 # OFF keeps the split, because a template that ignores enable_thinking
120 # still thinks.
121 if mode is ReasoningMode.INLINE:
122 reasoning, answer = split_reasoning("".join(text_parts))
123 if reasoning:
124 answer = f"{reasoning}\n\n{answer}" if answer else reasoning
125 reasoning = ""
126 else:
127 reasoning, answer = split_reasoning("".join(text_parts))
128 content: str | None = answer if answer or not tool_calls else None
130 total = resp.usage.input_tokens + resp.usage.output_tokens
131 return CompletionsResponse(
132 id=response_id,
133 created=int(time.time()),
134 model=resp.model,
135 choices=[
136 CompletionsResponseChoice(
137 index=0,
138 message=CompletionsResponseMessage(
139 content=content,
140 reasoning_content=reasoning or None,
141 tool_calls=tool_calls or None,
142 ),
143 finish_reason=_STOP_REASON_TO_FINISH[resp.stop_reason],
144 )
145 ],
146 usage=CompletionsUsage(
147 prompt_tokens=resp.usage.input_tokens,
148 completion_tokens=resp.usage.output_tokens,
149 total_tokens=total,
150 ),
151 )
154class _StreamMapper:
155 """Per-stream state for the canonical-to-OpenAI chunk converter."""
157 def __init__(self, *, mode: ReasoningMode = ReasoningMode.SEPARATE) -> None:
158 self._role_emitted = False
159 self._tool_index_for_block: dict[int, int] = {}
160 self._next_tool_index = 0
161 # INLINE routes thinking into content instead of reasoning_content, with
162 # the <think> markers stripped: a client that ignores reasoning_content
163 # renders raw tags literally, which is exactly what INLINE exists to
164 # avoid. The same stateful parse handles a tag split across deltas.
165 self._inline = mode is ReasoningMode.INLINE
166 # Splits lilbee's inline <think> text into its own delta field. Stateful
167 # because a tag can arrive split across deltas.
168 self._reasoning = TagParser(show=True)
170 def block_start(self, event: ContentBlockStart) -> CompletionsStreamDelta | None:
171 # The first delta must carry role:assistant for OpenAI-SDK accumulation,
172 # whether the response opens with text or a tool call.
173 role: Literal["assistant"] | None = None
174 if not self._role_emitted:
175 self._role_emitted = True
176 role = "assistant"
177 if isinstance(event.block, TextBlock):
178 return CompletionsStreamDelta(role=role) if role is not None else None
179 if isinstance(event.block, ToolUseBlock):
180 tool_index = self._next_tool_index
181 self._tool_index_for_block[event.index] = tool_index
182 self._next_tool_index += 1
183 return CompletionsStreamDelta(
184 role=role,
185 tool_calls=[_tool_call_open(tool_index, event.block.id, event.block.name)],
186 )
187 return None
189 def block_delta(self, event: ContentBlockDelta) -> CompletionsStreamDelta | None:
190 if isinstance(event.delta, TextDelta):
191 return self._text_delta(self._reasoning.feed(event.delta.text))
192 if isinstance(event.delta, ToolUseDelta):
193 # A delta for a block we never saw start is a provider quirk, not a
194 # server fault; a bare subscript turned it into a KeyError that
195 # surfaced as a stream-level internal error.
196 tool_index = self._tool_index_for_block.get(event.index)
197 if tool_index is None:
198 return None
199 return CompletionsStreamDelta(
200 tool_calls=[_tool_call_args(tool_index, event.delta.partial_json)],
201 )
202 return None
204 def block_stop(self) -> CompletionsStreamDelta | None:
205 """Emit whatever the reasoning splitter still holds (a partial or unclosed tag)."""
206 remaining = self._reasoning.flush()
207 return self._text_delta([remaining] if remaining else [])
209 def _text_delta(self, tokens: list[StreamToken]) -> CompletionsStreamDelta | None:
210 """One delta carrying the parsed text: split fields, or tag-free content."""
211 if self._inline:
212 # Arrival order preserved; the parser already dropped the markers.
213 text = "".join(t.content for t in tokens)
214 return CompletionsStreamDelta(content=text) if text else None
215 reasoning = "".join(t.content for t in tokens if t.is_reasoning)
216 answer = "".join(t.content for t in tokens if not t.is_reasoning)
217 if not reasoning and not answer:
218 return None
219 return CompletionsStreamDelta(
220 content=answer or None,
221 reasoning_content=reasoning or None,
222 )
225def _tool_call_open(index: int, call_id: str, name: str) -> CompletionsStreamToolCall:
226 return CompletionsStreamToolCall(
227 index=index,
228 id=call_id,
229 type="function",
230 function=CompletionsStreamToolCallFunction(name=name, arguments=""),
231 )
234def _tool_call_args(index: int, partial_json: str) -> CompletionsStreamToolCall:
235 return CompletionsStreamToolCall(
236 index=index,
237 function=CompletionsStreamToolCallFunction(arguments=partial_json),
238 )
241def _finish_reason_for(event: MessageDelta) -> FinishReason:
242 if event.stop_reason is None:
243 return FinishReason.STOP
244 return _STOP_REASON_TO_FINISH[event.stop_reason]
247async def canonical_stream_to_completions_chunks(
248 events: AsyncIterator[CanonicalStreamEvent],
249 *,
250 model: str,
251 response_id: str,
252 include_usage: bool = False,
253 mode: ReasoningMode = ReasoningMode.SEPARATE,
254) -> AsyncIterator[CompletionsStreamChunk]:
255 """Turn canonical stream events into ``CompletionsStreamChunk`` instances.
257 The trailing usage-only chunk is emitted only when *include_usage* is set,
258 matching OpenAI's ``stream_options.include_usage`` contract.
259 """
260 mapper = _StreamMapper(mode=mode)
261 # One timestamp for the whole completion. OpenAI holds created constant
262 # across a stream; recomputing it per chunk reported several creation times
263 # for one completion, and clients order or dedupe on it.
264 created = int(time.time())
265 async for event in events:
266 for chunk in _chunks_for_event(
267 event,
268 mapper,
269 model=model,
270 response_id=response_id,
271 include_usage=include_usage,
272 created=created,
273 ):
274 yield chunk
277def _chunks_for_event(
278 event: CanonicalStreamEvent,
279 mapper: _StreamMapper,
280 *,
281 model: str,
282 response_id: str,
283 include_usage: bool,
284 created: int,
285) -> list[CompletionsStreamChunk]:
286 """The OpenAI chunks one canonical event translates to; empty for a no-op event."""
287 if isinstance(event, ContentBlockStart):
288 return _maybe_chunk(model, response_id, mapper.block_start(event), created=created)
289 if isinstance(event, ContentBlockDelta):
290 return _maybe_chunk(model, response_id, mapper.block_delta(event), created=created)
291 if isinstance(event, ContentBlockStop):
292 # Closing a text block flushes any text the reasoning splitter still
293 # buffers (an unclosed <think>, or a tag that never completed).
294 return _maybe_chunk(model, response_id, mapper.block_stop(), created=created)
295 if isinstance(event, MessageDelta):
296 return _message_delta_chunks(
297 event,
298 model=model,
299 response_id=response_id,
300 include_usage=include_usage,
301 created=created,
302 )
303 if isinstance(event, MessageStart | MessageStop):
304 # OpenAI's wire format has no equivalent: MessageStart carries metadata
305 # already encoded in the chunk header, and MessageStop is replaced by the
306 # final chunk's finish_reason.
307 return []
308 # Unreachable: the branches above exhaust CanonicalStreamEvent. This makes a new
309 # event type a type error rather than a silently dropped frame.
310 assert_never(event) # pragma: no cover
313def _message_delta_chunks(
314 event: MessageDelta, *, model: str, response_id: str, include_usage: bool, created: int
315) -> list[CompletionsStreamChunk]:
316 """The finish chunk, plus the usage-only chunk when the client asked for it."""
317 chunks = [
318 _chunk(
319 model,
320 response_id,
321 CompletionsStreamDelta(),
322 finish_reason=_finish_reason_for(event),
323 created=created,
324 )
325 ]
326 if include_usage:
327 # OpenAI's contract sends the usage-only chunk unconditionally when
328 # include_usage is set; a client blocking on it must not hang because the
329 # provider streamed no usage frame.
330 usage = event.usage or CanonicalUsage(input_tokens=0, output_tokens=0)
331 chunks.append(_usage_chunk(model, response_id, usage, created=created))
332 return chunks
335def _maybe_chunk(
336 model: str, response_id: str, delta: CompletionsStreamDelta | None, *, created: int
337) -> list[CompletionsStreamChunk]:
338 """Wrap a delta in a chunk, or nothing when the event produced no delta."""
339 return [] if delta is None else [_chunk(model, response_id, delta, created=created)]
342def _chunk(
343 model: str,
344 response_id: str,
345 delta: CompletionsStreamDelta,
346 *,
347 finish_reason: FinishReason | None = None,
348 created: int,
349) -> CompletionsStreamChunk:
350 return CompletionsStreamChunk(
351 id=response_id,
352 created=created,
353 model=model,
354 choices=[CompletionsStreamChoice(index=0, delta=delta, finish_reason=finish_reason)],
355 )
358def _usage_chunk(
359 model: str, response_id: str, usage: CanonicalUsage, *, created: int
360) -> CompletionsStreamChunk:
361 """Final include_usage chunk: empty choices, populated usage totals."""
362 total = usage.input_tokens + usage.output_tokens
363 return CompletionsStreamChunk(
364 id=response_id,
365 created=created,
366 model=model,
367 choices=[],
368 usage=CompletionsUsage(
369 prompt_tokens=usage.input_tokens,
370 completion_tokens=usage.output_tokens,
371 total_tokens=total,
372 ),
373 )
376def _system_text(msg: CompletionsMessage) -> str:
377 """Flatten a system message to text, rejecting image parts.
379 The same image part is a 400 in a user message, so silently dropping it
380 here answered as though the request had been honoured.
381 """
382 if isinstance(msg.content, str):
383 return msg.content
384 if isinstance(msg.content, list):
385 if any(isinstance(part, CompletionsImageContent) for part in msg.content):
386 raise ValueError(_IMAGE_CONTENT_UNSUPPORTED)
387 return "".join(
388 part.text for part in msg.content if isinstance(part, CompletionsTextContent)
389 )
390 return ""
393def _message_from_request(msg: CompletionsMessage) -> CanonicalMessage:
394 role = msg.role
395 if role == "system":
396 raise ValueError("system messages should be extracted by the caller")
397 if role == "tool":
398 if not msg.tool_call_id:
399 # Substituting "" produced a tool result no tool call can pair with,
400 # which the provider sees as a result for a call it never made.
401 raise ValueError(_TOOL_CALL_ID_REQUIRED)
402 return CanonicalMessage(
403 role="tool",
404 content=[
405 ToolResultBlock(
406 tool_use_id=msg.tool_call_id,
407 content=_tool_result_content(msg.content),
408 )
409 ],
410 )
412 blocks: list[ContentBlock] = list(_content_blocks(msg.content))
413 for call in msg.tool_calls or []:
414 blocks.append(
415 ToolUseBlock(
416 id=call.id,
417 name=call.function.name,
418 input=parse_tool_arguments(call.function.arguments),
419 )
420 )
421 return CanonicalMessage(role=role, content=blocks)
424def _content_blocks(content: str | list | None) -> list[ContentBlock]:
425 if content is None or content == "":
426 return []
427 if isinstance(content, str):
428 return [TextBlock(text=content)]
429 blocks: list[ContentBlock] = []
430 for part in content:
431 if isinstance(part, CompletionsTextContent):
432 blocks.append(TextBlock(text=part.text))
433 elif isinstance(part, CompletionsImageContent):
434 raise ValueError(_IMAGE_CONTENT_UNSUPPORTED)
435 return blocks
438def _tool_result_content(content: str | list | None) -> list[ContentBlock]:
439 if isinstance(content, str):
440 return [TextBlock(text=content)]
441 if isinstance(content, list):
442 return _content_blocks(content)
443 return [TextBlock(text="" if content is None else str(content))]
446def _response_tool_call(block: ToolUseBlock) -> CompletionsResponseToolCall:
447 return CompletionsResponseToolCall(
448 id=block.id,
449 function=CompletionsResponseToolCallFunction(
450 name=block.name, arguments=json.dumps(block.input)
451 ),
452 )
455def _tools_from_request(tools: list[CompletionsTool] | None) -> list[CanonicalTool] | None:
456 if not tools:
457 return None
458 return [
459 CanonicalTool(
460 name=tool.function.name,
461 description=tool.function.description or "",
462 input_schema=tool.function.parameters,
463 )
464 for tool in tools
465 ]
468def _tool_choice_from_request(
469 choice: ToolChoiceMode | CompletionsNamedToolChoice | None,
470) -> CanonicalToolChoice | None:
471 if choice is None:
472 return None
473 if isinstance(choice, ToolChoiceMode):
474 return CanonicalToolChoice(mode=_TOOL_CHOICE_MODES[choice])
475 return CanonicalToolChoice(mode="tool", tool_name=choice.function.name)
478def _stop_from_request(stop: str | list[str] | None) -> list[str] | None:
479 if stop is None:
480 return None
481 if isinstance(stop, str):
482 return [stop]
483 return list(stop)