Coverage for src/lilbee/providers/backends/liter_llm/events.py: 100%

1"""Map liter-llm replies and stream chunks to lilbee value types.""" 

2 

3from __future__ import annotations 

4 

5from typing import Any 

6 

7from lilbee.providers.backends.liter_llm.client import _require_sdk 

8from lilbee.providers.base import TokenUsage 

9from lilbee.providers.sdk_backend import SdkToolCall, SdkToolCallDelta 

10 

11 

12def _tool_call(call: Any) -> SdkToolCall: 

13 """Build one ``SdkToolCall`` from a liter-llm tool call.""" 

14 return SdkToolCall( 

15 id=str(call.id or ""), 

16 name=str(call.function.name or ""), 

17 arguments=str(call.function.arguments or ""), 

18 ) 

19 

20 

21def _tool_call_delta(call: Any) -> SdkToolCallDelta: 

22 """Build one ``SdkToolCallDelta`` from a streaming tool-call slot. 

23 

24 Empty-string ``name`` / ``arguments`` are normalised to ``None`` so the 

25 SDK stream shape matches the native worker's deltas (the dispatch's 

26 ``_StreamState`` gates on ``is not None``; emitting ``""`` produces a 

27 spurious empty ContentBlockDelta on every opener). 

28 """ 

29 function = call.function 

30 return SdkToolCallDelta( 

31 index=call.index, 

32 id=str(call.id) if call.id else None, 

33 name=str(function.name) if function is not None and function.name else None, 

34 arguments_delta=( 

35 str(function.arguments) if function is not None and function.arguments else None 

36 ), 

37 ) 

38 

39 

40def _text(content: Any) -> str: 

41 """Return a reply's text: ``text`` for a text reply, the text parts joined for a parts reply.""" 

42 if content is None: 

43 return "" 

44 text: str | None = content.text 

45 if text is not None: 

46 return text 

47 if content.parts is not None: 

48 return "".join(part["text"] for part in content.parts if part["type"] == "text") 

49 return "" 

50 

51 

52def _finish_reason(reason: Any) -> str | None: 

53 """Return a ``FinishReason`` member as its wire string, or ``None``.""" 

54 if reason is None: 

55 return None 

56 finish = _require_sdk().FinishReason 

57 wire = { 

58 finish.STOP: "stop", 

59 finish.LENGTH: "length", 

60 finish.TOOL_CALLS: "tool_calls", 

61 finish.CONTENT_FILTER: "content_filter", 

62 finish.FUNCTION_CALL: "function_call", 

63 finish.OTHER: "other", 

64 } 

65 return wire[reason] 

66 

67 

68def _usage(usage: Any) -> TokenUsage: 

69 """Return a ``Usage`` as lilbee's token counts; zero when the reply carries none.""" 

70 if usage is None: 

71 return TokenUsage() 

72 return TokenUsage(prompt_tokens=usage.prompt_tokens, completion_tokens=usage.completion_tokens) 

73 

74 

75async def _next_chunk(stream: Any) -> Any: 

76 """Return the next chunk of an async chunk stream, or ``None`` when it ends.""" 

77 try: 

78 return await stream.__anext__() 

79 except StopAsyncIteration: 

80 return None