Coverage for src/lilbee/server/anthropic_api/translate.py: 100%
170 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-14 11:46 +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.retrieval.reasoning import StreamToken, TagParser, split_reasoning
11from lilbee.server.anthropic_api.models import (
12 AnthropicEventType,
13 AnthropicMessage,
14 AnthropicTool,
15 AnthropicToolChoice,
16 AnthropicUsage,
17 ContentBlockParam,
18 ImageBlockParam,
19 MessagesRequest,
20 MessagesResponse,
21 SystemTextBlock,
22 TextBlockParam,
23 ToolResultBlockParam,
24 ToolUseBlockParam,
25 UnknownBlockParam,
26)
27from lilbee.server.chat_dispatch.canonical import (
28 CanonicalChatRequest,
29 CanonicalMessage,
30 CanonicalResponse,
31 CanonicalStreamEvent,
32 CanonicalTool,
33 CanonicalToolChoice,
34 CanonicalUsage,
35 ContentBlock,
36 ContentBlockDelta,
37 ContentBlockStart,
38 ContentBlockStop,
39 MessageDelta,
40 MessageStart,
41 MessageStop,
42 StopReason,
43 TextBlock,
44 TextDelta,
45 ToolResultBlock,
46 ToolUseBlock,
47)
49_IMAGE_CONTENT_UNSUPPORTED = (
50 "Image content is not supported by /v1/messages yet. Send a text-only request."
51)
52_TOOL_CHOICE_NAME_REQUIRED = 'tool_choice type "tool" requires a name.'
55class _BlockKind(StrEnum):
56 """Kind of the mapper's open output block."""
58 THINKING = "thinking"
59 TEXT = "text"
60 TOOL = "tool"
63_ANTHROPIC_CHOICE_MODES: dict[str, str] = {
64 "auto": "auto",
65 "any": "any",
66 "none": "none",
67}
70def messages_to_canonical_request(request: MessagesRequest) -> CanonicalChatRequest:
71 """Translate a validated ``MessagesRequest`` to the canonical request."""
72 messages: list[CanonicalMessage] = []
73 for msg in request.messages:
74 messages.extend(_canonical_messages_for(msg))
75 return CanonicalChatRequest(
76 model=request.model,
77 messages=messages,
78 system=_system_text(request.system),
79 tools=_tools_from_request(request.tools),
80 tool_choice=_tool_choice_from_request(request.tool_choice),
81 temperature=request.temperature,
82 top_p=request.top_p,
83 top_k=request.top_k,
84 max_tokens=request.max_tokens,
85 stop=list(request.stop_sequences) if request.stop_sequences else None,
86 stream=request.stream,
87 )
90def _system_text(system: str | list[SystemTextBlock] | None) -> str | None:
91 if system is None:
92 return None
93 if isinstance(system, str):
94 return system or None
95 joined = "\n\n".join(block.text for block in system)
96 return joined or None
99def _canonical_messages_for(msg: AnthropicMessage) -> list[CanonicalMessage]:
100 """Fan one Anthropic message out to canonical messages.
102 Tool results become their own ``role: "tool"`` messages, emitted before
103 the user's text so the provider sees results adjacent to the calls they
104 answer. Unknown blocks (replayed thinking) are dropped. A mid-conversation
105 ``system`` message becomes a system-reminder user turn -- Anthropic's own
106 documented degradation for models without the operator channel, and it
107 keeps the canonical layer's role set unchanged.
108 """
109 if msg.role == "system":
110 text = _message_text(msg)
111 if not text:
112 return []
113 return [
114 CanonicalMessage.from_string(
115 role="user", text=f"<system-reminder>\n{text}\n</system-reminder>"
116 )
117 ]
118 if isinstance(msg.content, str):
119 if not msg.content:
120 return []
121 return [CanonicalMessage.from_string(role=msg.role, text=msg.content)]
122 return _block_messages(msg.role, msg.content)
125def _block_messages(
126 role: Literal["user", "assistant"], content: list[ContentBlockParam]
127) -> list[CanonicalMessage]:
128 """Canonical messages for a block-form user or assistant message."""
129 tool_messages: list[CanonicalMessage] = []
130 blocks: list[ContentBlock] = []
131 for block in content:
132 if isinstance(block, TextBlockParam):
133 blocks.append(TextBlock(text=block.text))
134 elif isinstance(block, ToolUseBlockParam):
135 blocks.append(ToolUseBlock(id=block.id, name=block.name, input=block.input))
136 elif isinstance(block, ToolResultBlockParam):
137 tool_messages.append(_tool_result_message(block))
138 elif isinstance(block, ImageBlockParam):
139 raise ValueError(_IMAGE_CONTENT_UNSUPPORTED)
140 elif isinstance(block, UnknownBlockParam):
141 continue
143 out = tool_messages
144 if blocks:
145 out = [*tool_messages, CanonicalMessage(role=role, content=blocks)]
146 return out
149def _tool_result_message(block: ToolResultBlockParam) -> CanonicalMessage:
150 return CanonicalMessage(
151 role="tool",
152 content=[
153 ToolResultBlock(
154 tool_use_id=block.tool_use_id,
155 content=_tool_result_content(block),
156 is_error=block.is_error,
157 )
158 ],
159 )
162def _message_text(msg: AnthropicMessage) -> str:
163 """The concatenated text of a message, ignoring non-text blocks."""
164 if isinstance(msg.content, str):
165 return msg.content
166 return "".join(b.text for b in msg.content if isinstance(b, TextBlockParam))
169def _tool_result_content(block: ToolResultBlockParam) -> list[ContentBlock]:
170 if block.content is None:
171 return []
172 if isinstance(block.content, str):
173 return [TextBlock(text=block.content)]
174 parts: list[ContentBlock] = []
175 for part in block.content:
176 if isinstance(part, TextBlockParam):
177 parts.append(TextBlock(text=part.text))
178 elif isinstance(part, ImageBlockParam):
179 raise ValueError(_IMAGE_CONTENT_UNSUPPORTED)
180 # UnknownBlockParam: dropped
181 return parts
184def _tools_from_request(tools: list[AnthropicTool] | None) -> list[CanonicalTool] | None:
185 if not tools:
186 return None
187 return [
188 CanonicalTool(
189 name=tool.name,
190 description=tool.description or "",
191 input_schema=tool.input_schema,
192 )
193 for tool in tools
194 ]
197def _tool_choice_from_request(
198 choice: AnthropicToolChoice | None,
199) -> CanonicalToolChoice | None:
200 if choice is None:
201 return None
202 if choice.type == "tool":
203 if not choice.name:
204 raise ValueError(_TOOL_CHOICE_NAME_REQUIRED)
205 return CanonicalToolChoice(mode="tool", tool_name=choice.name)
206 mode = _ANTHROPIC_CHOICE_MODES[choice.type]
207 return CanonicalToolChoice(mode=mode) # type: ignore[arg-type]
210def canonical_to_messages_response(
211 resp: CanonicalResponse, *, response_id: str
212) -> MessagesResponse:
213 """Translate a canonical chat response to the Anthropic message shape.
215 lilbee carries a reasoning model's thinking inline as ``<think>...</think>``;
216 the Anthropic surface reports it as a leading ``thinking`` block so clients
217 render a clean answer.
218 """
219 text_parts = [b.text for b in resp.content if isinstance(b, TextBlock)]
220 reasoning, answer = split_reasoning("".join(text_parts))
221 content: list[dict[str, Any]] = []
222 if reasoning:
223 content.append({"type": "thinking", "thinking": reasoning})
224 tool_uses = [b for b in resp.content if isinstance(b, ToolUseBlock)]
225 if answer or not (reasoning or tool_uses):
226 content.append({"type": "text", "text": answer})
227 content.extend(
228 {"type": "tool_use", "id": b.id, "name": b.name, "input": b.input} for b in tool_uses
229 )
230 return MessagesResponse(
231 id=response_id,
232 model=resp.model,
233 content=content,
234 stop_reason=str(resp.stop_reason),
235 usage=AnthropicUsage(
236 input_tokens=resp.usage.input_tokens,
237 output_tokens=resp.usage.output_tokens,
238 ),
239 )
242class _AnthropicStreamMapper:
243 """Per-stream state for the canonical-to-Anthropic event converter.
245 lilbee streams reasoning inline as ``<think>`` text, and Anthropic's wire
246 format wants thinking and answer text in separate indexed blocks; the
247 mapper re-blocks the stream, closing the open block whenever the token
248 kind (thinking / text / tool_use) changes.
249 """
251 def __init__(self) -> None:
252 self._reasoning = TagParser(show=True)
253 self._next_index = 0
254 self._open: _BlockKind | None = None
256 def _close_open(self) -> list[tuple[AnthropicEventType, dict[str, Any]]]:
257 if self._open is None:
258 return []
259 index = self._next_index - 1
260 self._open = None
261 return [
262 (AnthropicEventType.CONTENT_BLOCK_STOP, {"type": "content_block_stop", "index": index})
263 ]
265 def _ensure_block(self, kind: _BlockKind) -> list[tuple[AnthropicEventType, dict[str, Any]]]:
266 if self._open == kind:
267 return []
268 events = self._close_open()
269 shell = (
270 {"type": "thinking", "thinking": ""}
271 if kind is _BlockKind.THINKING
272 else {"type": "text", "text": ""}
273 )
274 events.append(
275 (
276 AnthropicEventType.CONTENT_BLOCK_START,
277 {
278 "type": "content_block_start",
279 "index": self._next_index,
280 "content_block": shell,
281 },
282 )
283 )
284 self._open = kind
285 self._next_index += 1
286 return events
288 def _text_events(
289 self, tokens: list[StreamToken]
290 ) -> list[tuple[AnthropicEventType, dict[str, Any]]]:
291 events: list[tuple[AnthropicEventType, dict[str, Any]]] = []
292 for token in tokens:
293 if not token.content:
294 continue
295 kind = _BlockKind.THINKING if token.is_reasoning else _BlockKind.TEXT
296 events.extend(self._ensure_block(kind))
297 index = self._next_index - 1
298 delta = (
299 {"type": "thinking_delta", "thinking": token.content}
300 if token.is_reasoning
301 else {"type": "text_delta", "text": token.content}
302 )
303 events.append(
304 (
305 AnthropicEventType.CONTENT_BLOCK_DELTA,
306 {"type": "content_block_delta", "index": index, "delta": delta},
307 )
308 )
309 return events
311 def block_start(
312 self, event: ContentBlockStart
313 ) -> list[tuple[AnthropicEventType, dict[str, Any]]]:
314 if isinstance(event.block, ToolUseBlock):
315 events = self._close_open()
316 index = self._next_index
317 self._next_index += 1
318 self._open = _BlockKind.TOOL
319 events.append(
320 (
321 AnthropicEventType.CONTENT_BLOCK_START,
322 {
323 "type": "content_block_start",
324 "index": index,
325 "content_block": {
326 "type": "tool_use",
327 "id": event.block.id,
328 "name": event.block.name,
329 "input": {},
330 },
331 },
332 )
333 )
334 if event.block.input:
335 # A provider that announces a whole call up front carries the
336 # parsed input on the start block; forward it as one delta so
337 # SDK accumulation still sees arguments.
338 events.append(
339 (
340 AnthropicEventType.CONTENT_BLOCK_DELTA,
341 {
342 "type": "content_block_delta",
343 "index": index,
344 "delta": {
345 "type": "input_json_delta",
346 "partial_json": json.dumps(event.block.input),
347 },
348 },
349 )
350 )
351 return events
352 # Text blocks open lazily on the first delta so an empty block never
353 # emits a start/stop pair.
354 return []
356 def block_delta(
357 self, event: ContentBlockDelta
358 ) -> list[tuple[AnthropicEventType, dict[str, Any]]]:
359 if isinstance(event.delta, TextDelta):
360 return self._text_events(self._reasoning.feed(event.delta.text))
361 if self._open is not _BlockKind.TOOL:
362 # A tool delta for a block that never started is a provider quirk,
363 # not a stream error; dropping beats crashing the stream.
364 return []
365 return [
366 (
367 AnthropicEventType.CONTENT_BLOCK_DELTA,
368 {
369 "type": "content_block_delta",
370 "index": self._next_index - 1,
371 "delta": {
372 "type": "input_json_delta",
373 "partial_json": event.delta.partial_json,
374 },
375 },
376 )
377 ]
379 def block_stop(self) -> list[tuple[AnthropicEventType, dict[str, Any]]]:
380 remaining = self._reasoning.flush()
381 events = self._text_events([remaining] if remaining else [])
382 events.extend(self._close_open())
383 return events
386async def canonical_stream_to_anthropic_events(
387 events: AsyncIterator[CanonicalStreamEvent],
388 *,
389 model: str,
390 response_id: str,
391) -> AsyncIterator[tuple[AnthropicEventType, dict[str, Any]]]:
392 """Turn canonical stream events into Anthropic SSE ``(type, payload)`` pairs."""
393 mapper = _AnthropicStreamMapper()
394 yield (
395 AnthropicEventType.MESSAGE_START,
396 {
397 "type": "message_start",
398 "message": {
399 "id": response_id,
400 "type": "message",
401 "role": "assistant",
402 "model": model,
403 "content": [],
404 "stop_reason": None,
405 "stop_sequence": None,
406 "usage": {"input_tokens": 0, "output_tokens": 0},
407 },
408 },
409 )
410 async for event in events:
411 if isinstance(event, ContentBlockStart):
412 for out in mapper.block_start(event):
413 yield out
414 elif isinstance(event, ContentBlockDelta):
415 for out in mapper.block_delta(event):
416 yield out
417 elif isinstance(event, ContentBlockStop):
418 for out in mapper.block_stop():
419 yield out
420 elif isinstance(event, MessageDelta):
421 usage = event.usage or CanonicalUsage(input_tokens=0, output_tokens=0)
422 yield (
423 AnthropicEventType.MESSAGE_DELTA,
424 {
425 "type": "message_delta",
426 "delta": {
427 "stop_reason": str(event.stop_reason or StopReason.END_TURN),
428 "stop_sequence": None,
429 },
430 "usage": {
431 "input_tokens": usage.input_tokens,
432 "output_tokens": usage.output_tokens,
433 },
434 },
435 )
436 elif isinstance(event, MessageStart | MessageStop):
437 # The Anthropic message_start is emitted eagerly above (it needs no
438 # canonical data), and message_stop follows the loop so it stays
439 # last even if the provider never sends one.
440 continue
441 yield (AnthropicEventType.MESSAGE_STOP, {"type": "message_stop"})