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