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

1"""Map lilbee requests to liter-llm request fields.""" 

2 

3from __future__ import annotations 

4 

5import base64 

6import json 

7from typing import Any 

8 

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

10from lilbee.providers.local_servers import OLLAMA, detect_local_server, local_server_for_key 

11from lilbee.providers.model_ref import ProviderModelRef 

12from lilbee.providers.sdk_backend import CompletionRequest 

13 

14 

15def _route_model(ref: ProviderModelRef, api_base: str | None) -> str: 

16 """Format *ref* for liter-llm using the ``provider/model`` convention. 

17 

18 API and local-server refs already carry their canonical prefix. A bare 

19 ``local`` ref forced through the SDK (``llm_provider=remote``) gets the 

20 prefix of whichever local server its ``api_base`` points at. 

21 """ 

22 if ref.is_api or local_server_for_key(ref.provider) is not None: 

23 return ref.for_openai_prefix() 

24 if api_base and (spec := detect_local_server(api_base)) is not None: 

25 return spec.qualify(ref.name) 

26 return ref.name 

27 

28 

29def _wire_base_url(model: str, api_base: str | None) -> str | None: 

30 """Return the base URL to give liter-llm for a routed *model*, or ``None``. 

31 

32 liter-llm drops one trailing ``/v1`` from an Ollama base and appends 

33 ``/api/...``, so Ollama gets the base as configured. An OpenAI-compatible 

34 server such as LM Studio serves ``/chat/completions`` under ``/v1``, so it 

35 gets the base with exactly one ``/v1``. 

36 """ 

37 if not api_base: 

38 return None 

39 trimmed = api_base.rstrip("/") 

40 if model.startswith(OLLAMA.wire_prefix): 

41 return trimmed 

42 return f"{trimmed.removesuffix('/v1')}/v1" 

43 

44 

45EMPTY_MESSAGE_PLACEHOLDER = "[System: Empty message content sanitised to satisfy protocol]" 

46 

47 

48def _is_blank(text: Any) -> bool: 

49 """Whether *text* is not a string, or is empty or only whitespace.""" 

50 return not isinstance(text, str) or not text.strip() 

51 

52 

53def _fill_blank_text(msg: dict[str, Any]) -> dict[str, Any]: 

54 """Give a blank text of a user or assistant turn the placeholder, as Anthropic needs. 

55 

56 A string content and each text part of a list content are checked. Other parts and 

57 other roles pass through. 

58 """ 

59 if msg.get("role") not in ("user", "assistant"): 

60 return msg 

61 content = msg.get("content") 

62 if isinstance(content, str): 

63 return {**msg, "content": EMPTY_MESSAGE_PLACEHOLDER} if _is_blank(content) else msg 

64 if isinstance(content, list): 

65 parts = [ 

66 {**part, "text": EMPTY_MESSAGE_PLACEHOLDER} 

67 if isinstance(part, dict) and part.get("type") == "text" and _is_blank(part.get("text")) 

68 else part 

69 for part in content 

70 ] 

71 return {**msg, "content": parts} 

72 return msg 

73 

74 

75def _format_messages( 

76 messages: list[dict[str, Any]], *, fill_blank_turns: bool = False 

77) -> list[dict[str, Any]]: 

78 """Convert messages with inline image bytes into OpenAI content parts. 

79 

80 Messages without ``images`` pass through untouched. With *fill_blank_turns*, 

81 blank text in a user or assistant turn carries a placeholder, as Anthropic rejects it. 

82 """ 

83 formatted: list[dict[str, Any]] = [] 

84 for msg in messages: 

85 if "images" in msg: 

86 content_parts: list[dict[str, Any]] = [{"type": "text", "text": msg.get("content", "")}] 

87 for img in msg["images"]: 

88 if isinstance(img, bytes): 

89 b64 = base64.b64encode(img).decode() 

90 content_parts.append( 

91 { 

92 "type": "image_url", 

93 "image_url": {"url": f"data:image/png;base64,{b64}"}, 

94 } 

95 ) 

96 formatted.append({"role": msg["role"], "content": content_parts}) 

97 else: 

98 formatted.append(msg) 

99 return [_fill_blank_text(msg) for msg in formatted] if fill_blank_turns else formatted 

100 

101 

102# lilbee option -> chat request field, for the options liter-llm takes as typed fields. 

103_REQUEST_FIELDS = frozenset( 

104 { 

105 "temperature", 

106 "top_p", 

107 "seed", 

108 "max_tokens", 

109 "frequency_penalty", 

110 "presence_penalty", 

111 "stop", 

112 "response_format", 

113 "tool_choice", 

114 } 

115) 

116 

117 

118def _chat_fields(request: CompletionRequest, *, stream: bool) -> dict[str, Any]: 

119 """Translate a ``CompletionRequest`` into liter-llm chat request fields.""" 

120 liter_llm = _require_sdk() 

121 fields: dict[str, Any] = { 

122 "model": _route_model(request.ref, request.api_base), 

123 "messages": _format_messages( 

124 request.messages, fill_blank_turns=request.ref.provider == "anthropic" 

125 ), 

126 "stream": stream, 

127 } 

128 extra_body: dict[str, Any] = {} 

129 for key, value in request.options.items(): 

130 if key == "num_predict": 

131 fields["max_tokens"] = value 

132 elif key == "tools": 

133 fields["tools"] = [liter_llm.ChatCompletionTool.from_json(json.dumps(t)) for t in value] 

134 elif key in _REQUEST_FIELDS: 

135 fields[key] = value 

136 elif key == "reasoning_effort": 

137 fields[key] = liter_llm.ReasoningEffort(value) 

138 else: 

139 extra_body[key] = value 

140 if extra_body: 

141 fields["extra_body"] = extra_body 

142 return fields