Coverage for src/lilbee/server/anthropic_api/translate.py: 100%
192 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-09-04 17:08 +0000
1"""Translation between Anthropic Messages models and the canonical types."""
3from __future__ import annotations
5import json
6from collections.abc import AsyncIterator
7from enum import StrEnum
8from typing import Any, Literal
10from lilbee.core.config.enums import ReasoningMode
11from lilbee.retrieval.reasoning import (
12 PseudoThinkingNormalizer,
13 StreamToken,
14 TagParser,
15 normalize_pseudo_thinking,
16 split_reasoning,
17)
18from lilbee.server.anthropic_api.models import (
19 _THINKING_DISABLED,
20 AnthropicEventType,
21 AnthropicMessage,
22 AnthropicThinking,
23 AnthropicTool,
24 AnthropicToolChoice,
25 AnthropicUsage,
26 ContentBlockParam,
27 ImageBlockParam,
28 MessagesRequest,
29 MessagesResponse,
30 SystemTextBlock,
31 TextBlockParam,
32 ToolResultBlockParam,
33 ToolUseBlockParam,
34 UnknownBlockParam,
35)
36from lilbee.server.chat_dispatch.canonical import (
37 CanonicalChatRequest,
38 CanonicalMessage,
39 CanonicalResponse,
40 CanonicalStreamEvent,
41 CanonicalTool,
42 CanonicalToolChoice,
43 CanonicalUsage,
44 ContentBlock,
45 ContentBlockDelta,
46 ContentBlockStart,
47 ContentBlockStop,
48 MessageDelta,
49 MessageStart,
50 MessageStop,
51 StopReason,
52 TextBlock,
53 TextDelta,
54 ToolResultBlock,
55 ToolUseBlock,
56)
58_IMAGE_CONTENT_UNSUPPORTED = (
59 "Image content is not supported by /v1/messages yet. Send a text-only request."
60)
61_TOOL_CHOICE_NAME_REQUIRED = 'tool_choice type "tool" requires a name.'
64class _BlockKind(StrEnum):
65 """Kind of the mapper's open output block."""
67 THINKING = "thinking"
68 TEXT = "text"
69 TOOL = "tool"
72_ANTHROPIC_CHOICE_MODES: dict[str, str] = {
73 "auto": "auto",
74 "any": "any",
75 "none": "none",
76}
79def resolve_reasoning_mode(
80 thinking: AnthropicThinking | None, *, default: ReasoningMode
81) -> ReasoningMode:
82 """Pick the reasoning mode for one call from the request and the setting.
84 Thinking is opt-in per request, as on the Anthropic API: a body with no
85 ``thinking`` gets none, whatever the setting presents it as. The setting
86 says how to present thinking a request asked for, and ``off`` refuses it
87 outright, so a request can only tighten.
88 """
89 if default is ReasoningMode.OFF:
90 return ReasoningMode.OFF
91 if thinking is None or thinking.type == _THINKING_DISABLED:
92 return ReasoningMode.OFF
93 return default
96def messages_to_canonical_request(
97 request: MessagesRequest, *, mode: ReasoningMode = ReasoningMode.SEPARATE
98) -> CanonicalChatRequest:
99 """Translate a validated ``MessagesRequest`` to the canonical request."""
100 messages: list[CanonicalMessage] = []
101 for msg in request.messages:
102 messages.extend(_canonical_messages_for(msg))
103 return CanonicalChatRequest(
104 model=request.model,
105 messages=messages,
106 system=_system_text(request.system),
107 tools=_tools_from_request(request.tools),
108 tool_choice=_tool_choice_from_request(request.tool_choice),
109 temperature=request.temperature,
110 top_p=request.top_p,
111 top_k=request.top_k,
112 max_tokens=request.max_tokens,
113 stop=list(request.stop_sequences) if request.stop_sequences else None,
114 stream=request.stream,
115 # OFF asks the template to skip thinking; the other modes only change
116 # presentation, so the template default stands.
117 think=False if mode is ReasoningMode.OFF else None,
118 )
121def _system_text(system: str | list[SystemTextBlock] | None) -> str | None:
122 if system is None:
123 return None
124 if isinstance(system, str):
125 return system or None
126 joined = "\n\n".join(block.text for block in system)
127 return joined or None
130def _canonical_messages_for(msg: AnthropicMessage) -> list[CanonicalMessage]:
131 """Fan one Anthropic message out to canonical messages.
133 Tool results become their own ``role: "tool"`` messages, emitted before
134 the user's text so the provider sees results adjacent to the calls they
135 answer. Unknown blocks (replayed thinking) are dropped. A mid-conversation
136 ``system`` message becomes a system-reminder user turn -- Anthropic's own
137 documented degradation for models without the operator channel, and it
138 keeps the canonical layer's role set unchanged.
139 """
140 if msg.role == "system":
141 text = _message_text(msg)
142 if not text:
143 return []
144 return [
145 CanonicalMessage.from_string(
146 role="user", text=f"<system-reminder>\n{text}\n</system-reminder>"
147 )
148 ]
149 if isinstance(msg.content, str):
150 if not msg.content:
151 return []
152 return [CanonicalMessage.from_string(role=msg.role, text=msg.content)]
153 return _block_messages(msg.role, msg.content)
156def _block_messages(
157 role: Literal["user", "assistant"], content: list[ContentBlockParam]
158) -> list[CanonicalMessage]:
159 """Canonical messages for a block-form user or assistant message."""
160 tool_messages: list[CanonicalMessage] = []
161 blocks: list[ContentBlock] = []
162 for block in content:
163 if isinstance(block, TextBlockParam):
164 blocks.append(TextBlock(text=block.text))
165 elif isinstance(block, ToolUseBlockParam):
166 blocks.append(ToolUseBlock(id=block.id, name=block.name, input=block.input))
167 elif isinstance(block, ToolResultBlockParam):
168 tool_messages.append(_tool_result_message(block))
169 elif isinstance(block, ImageBlockParam):
170 raise ValueError(_IMAGE_CONTENT_UNSUPPORTED)
171 elif isinstance(block, UnknownBlockParam):
172 continue
174 out = tool_messages
175 if blocks:
176 out = [*tool_messages, CanonicalMessage(role=role, content=blocks)]
177 return out
180def _tool_result_message(block: ToolResultBlockParam) -> CanonicalMessage:
181 return CanonicalMessage(
182 role="tool",
183 content=[
184 ToolResultBlock(
185 tool_use_id=block.tool_use_id,
186 content=_tool_result_content(block),
187 is_error=block.is_error,
188 )
189 ],
190 )
193def _message_text(msg: AnthropicMessage) -> str:
194 """The concatenated text of a message, ignoring non-text blocks."""
195 if isinstance(msg.content, str):
196 return msg.content
197 return "".join(b.text for b in msg.content if isinstance(b, TextBlockParam))
200def _tool_result_content(block: ToolResultBlockParam) -> list[ContentBlock]:
201 if block.content is None:
202 return []
203 if isinstance(block.content, str):
204 return [TextBlock(text=block.content)]
205 parts: list[ContentBlock] = []
206 for part in block.content:
207 if isinstance(part, TextBlockParam):
208 parts.append(TextBlock(text=part.text))
209 elif isinstance(part, ImageBlockParam):
210 raise ValueError(_IMAGE_CONTENT_UNSUPPORTED)
211 # UnknownBlockParam: dropped
212 return parts
215def _tools_from_request(tools: list[AnthropicTool] | None) -> list[CanonicalTool] | None:
216 if not tools:
217 return None
218 return [
219 CanonicalTool(
220 name=tool.name,
221 description=tool.description or "",
222 input_schema=tool.input_schema,
223 )
224 for tool in tools
225 ]
228def _tool_choice_from_request(
229 choice: AnthropicToolChoice | None,
230) -> CanonicalToolChoice | None:
231 if choice is None:
232 return None
233 if choice.type == "tool":
234 if not choice.name:
235 raise ValueError(_TOOL_CHOICE_NAME_REQUIRED)
236 return CanonicalToolChoice(mode="tool", tool_name=choice.name)
237 mode = _ANTHROPIC_CHOICE_MODES[choice.type]
238 return CanonicalToolChoice(mode=mode) # type: ignore[arg-type]
241def canonical_to_messages_response(
242 resp: CanonicalResponse, *, response_id: str, mode: ReasoningMode = ReasoningMode.SEPARATE
243) -> MessagesResponse:
244 """Translate a canonical chat response to the Anthropic message shape.
246 lilbee carries a reasoning model's thinking inline as ``<think>...</think>``;
247 SEPARATE reports it as a leading ``thinking`` block so clients render a
248 clean answer. INLINE folds it into the answer text with the markers
249 stripped, for clients that never render thinking blocks. OFF drops it: the
250 caller asked for no thinking, and a template that ignores the request still
251 thinks, so the block would contradict the answer the caller asked for. OFF
252 also drops a reply-initial pseudo-thinking block a model emits as plain text.
253 """
254 text = "".join(b.text for b in resp.content if isinstance(b, TextBlock))
255 if mode is ReasoningMode.OFF:
256 text = normalize_pseudo_thinking(text)
257 reasoning, answer = split_reasoning(text)
258 if mode is ReasoningMode.INLINE and reasoning:
259 answer = f"{reasoning}\n\n{answer}" if answer else reasoning
260 if mode is not ReasoningMode.SEPARATE:
261 reasoning = ""
262 content: list[dict[str, Any]] = []
263 if reasoning:
264 content.append({"type": "thinking", "thinking": reasoning})
265 tool_uses = [b for b in resp.content if isinstance(b, ToolUseBlock)]
266 if answer or not (reasoning or tool_uses):
267 content.append({"type": "text", "text": answer})
268 content.extend(
269 {"type": "tool_use", "id": b.id, "name": b.name, "input": b.input} for b in tool_uses
270 )
271 return MessagesResponse(
272 id=response_id,
273 model=resp.model,
274 content=content,
275 stop_reason=str(resp.stop_reason),
276 usage=AnthropicUsage(
277 input_tokens=resp.usage.input_tokens,
278 output_tokens=resp.usage.output_tokens,
279 ),
280 )
283class _AnthropicStreamMapper:
284 """Per-stream state for the canonical-to-Anthropic event converter.
286 lilbee streams reasoning inline as ``<think>`` text, and Anthropic's wire
287 format wants thinking and answer text in separate indexed blocks; the
288 mapper re-blocks the stream, closing the open block whenever the token
289 kind (thinking / text / tool_use) changes.
291 INLINE routes reasoning into the text block instead, and OFF drops it: a
292 parser built with ``show=False`` reports reasoning tokens with empty
293 content, which never opens a block or emits a delta. OFF also rewrites a
294 reply-initial pseudo-thinking tag to the ``<think>`` tags before parsing,
295 so a planning block a model emits as plain text is dropped too.
296 """
298 def __init__(self, *, mode: ReasoningMode = ReasoningMode.SEPARATE) -> None:
299 self._reasoning = TagParser(show=mode is not ReasoningMode.OFF)
300 self._pseudo: PseudoThinkingNormalizer | None = (
301 PseudoThinkingNormalizer() if mode is ReasoningMode.OFF else None
302 )
303 self._inline = mode is ReasoningMode.INLINE
304 self._next_index = 0
305 self._open: _BlockKind | None = None
307 def _close_open(self) -> list[tuple[AnthropicEventType, dict[str, Any]]]:
308 if self._open is None:
309 return []
310 index = self._next_index - 1
311 self._open = None
312 return [
313 (AnthropicEventType.CONTENT_BLOCK_STOP, {"type": "content_block_stop", "index": index})
314 ]
316 def _ensure_block(self, kind: _BlockKind) -> list[tuple[AnthropicEventType, dict[str, Any]]]:
317 if self._open == kind:
318 return []
319 events = self._close_open()
320 shell = (
321 {"type": "thinking", "thinking": ""}
322 if kind is _BlockKind.THINKING
323 else {"type": "text", "text": ""}
324 )
325 events.append(
326 (
327 AnthropicEventType.CONTENT_BLOCK_START,
328 {
329 "type": "content_block_start",
330 "index": self._next_index,
331 "content_block": shell,
332 },
333 )
334 )
335 self._open = kind
336 self._next_index += 1
337 return events
339 def _text_events(
340 self, tokens: list[StreamToken]
341 ) -> list[tuple[AnthropicEventType, dict[str, Any]]]:
342 events: list[tuple[AnthropicEventType, dict[str, Any]]] = []
343 for token in tokens:
344 if not token.content:
345 continue
346 thinking = token.is_reasoning and not self._inline
347 kind = _BlockKind.THINKING if thinking else _BlockKind.TEXT
348 events.extend(self._ensure_block(kind))
349 index = self._next_index - 1
350 delta = (
351 {"type": "thinking_delta", "thinking": token.content}
352 if thinking
353 else {"type": "text_delta", "text": token.content}
354 )
355 events.append(
356 (
357 AnthropicEventType.CONTENT_BLOCK_DELTA,
358 {"type": "content_block_delta", "index": index, "delta": delta},
359 )
360 )
361 return events
363 def block_start(
364 self, event: ContentBlockStart
365 ) -> list[tuple[AnthropicEventType, dict[str, Any]]]:
366 if isinstance(event.block, ToolUseBlock):
367 events = self._close_open()
368 index = self._next_index
369 self._next_index += 1
370 self._open = _BlockKind.TOOL
371 events.append(
372 (
373 AnthropicEventType.CONTENT_BLOCK_START,
374 {
375 "type": "content_block_start",
376 "index": index,
377 "content_block": {
378 "type": "tool_use",
379 "id": event.block.id,
380 "name": event.block.name,
381 "input": {},
382 },
383 },
384 )
385 )
386 if event.block.input:
387 # A provider that announces a whole call up front carries the
388 # parsed input on the start block; forward it as one delta so
389 # SDK accumulation still sees arguments.
390 events.append(
391 (
392 AnthropicEventType.CONTENT_BLOCK_DELTA,
393 {
394 "type": "content_block_delta",
395 "index": index,
396 "delta": {
397 "type": "input_json_delta",
398 "partial_json": json.dumps(event.block.input),
399 },
400 },
401 )
402 )
403 return events
404 # Text blocks open lazily on the first delta so an empty block never
405 # emits a start/stop pair.
406 return []
408 def block_delta(
409 self, event: ContentBlockDelta
410 ) -> list[tuple[AnthropicEventType, dict[str, Any]]]:
411 if isinstance(event.delta, TextDelta):
412 text = event.delta.text
413 if self._pseudo is not None:
414 text = self._pseudo.feed(text)
415 return self._text_events(self._reasoning.feed(text))
416 if self._open is not _BlockKind.TOOL:
417 # A tool delta for a block that never started is a provider quirk,
418 # not a stream error; dropping beats crashing the stream.
419 return []
420 return [
421 (
422 AnthropicEventType.CONTENT_BLOCK_DELTA,
423 {
424 "type": "content_block_delta",
425 "index": self._next_index - 1,
426 "delta": {
427 "type": "input_json_delta",
428 "partial_json": event.delta.partial_json,
429 },
430 },
431 )
432 ]
434 def block_stop(self) -> list[tuple[AnthropicEventType, dict[str, Any]]]:
435 tokens = self._reasoning.feed(self._pseudo.flush()) if self._pseudo is not None else []
436 remaining = self._reasoning.flush()
437 if remaining is not None:
438 tokens.append(remaining)
439 events = self._text_events(tokens)
440 events.extend(self._close_open())
441 return events
444async def canonical_stream_to_anthropic_events(
445 events: AsyncIterator[CanonicalStreamEvent],
446 *,
447 model: str,
448 response_id: str,
449 mode: ReasoningMode = ReasoningMode.SEPARATE,
450) -> AsyncIterator[tuple[AnthropicEventType, dict[str, Any]]]:
451 """Turn canonical stream events into Anthropic SSE ``(type, payload)`` pairs."""
452 mapper = _AnthropicStreamMapper(mode=mode)
453 yield (
454 AnthropicEventType.MESSAGE_START,
455 {
456 "type": "message_start",
457 "message": {
458 "id": response_id,
459 "type": "message",
460 "role": "assistant",
461 "model": model,
462 "content": [],
463 "stop_reason": None,
464 "stop_sequence": None,
465 "usage": {"input_tokens": 0, "output_tokens": 0},
466 },
467 },
468 )
469 async for event in events:
470 if isinstance(event, ContentBlockStart):
471 for out in mapper.block_start(event):
472 yield out
473 elif isinstance(event, ContentBlockDelta):
474 for out in mapper.block_delta(event):
475 yield out
476 elif isinstance(event, ContentBlockStop):
477 for out in mapper.block_stop():
478 yield out
479 elif isinstance(event, MessageDelta):
480 usage = event.usage or CanonicalUsage(input_tokens=0, output_tokens=0)
481 yield (
482 AnthropicEventType.MESSAGE_DELTA,
483 {
484 "type": "message_delta",
485 "delta": {
486 "stop_reason": str(event.stop_reason or StopReason.END_TURN),
487 "stop_sequence": None,
488 },
489 "usage": {
490 "input_tokens": usage.input_tokens,
491 "output_tokens": usage.output_tokens,
492 },
493 },
494 )
495 elif isinstance(event, MessageStart | MessageStop):
496 # The Anthropic message_start is emitted eagerly above (it needs no
497 # canonical data), and message_stop follows the loop so it stays
498 # last even if the provider never sends one.
499 continue
500 yield (AnthropicEventType.MESSAGE_STOP, {"type": "message_stop"})