Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-llm/src/lexigram/ai/llm/clients/_message_utils.py: 12%

1"""Per-provider serializers for ``MessageContent`` → wire-format dicts. 

2 

3Each function converts a ``MessageContent`` value (either a plain ``str`` or 

4a ``list[ContentPart]``) into the format expected by that provider's API. 

5These are pure sync functions; clients that need to fetch image URLs before 

6serializing should call ``fetch_image_as_base64`` from ``lexigram.ai.llm.multimodal`` 

7first and replace ``ImageUrlPart`` entries with the resulting ``ImageBase64Part``. 

8""" 

9 

10from __future__ import annotations 

11 

12from typing import Any 

13 

14from lexigram.contracts.ai.multimodal import ( 

15 ImageBase64Part, 

16 ImageUrlPart, 

17 MessageContent, 

18 TextPart, 

19) 

20 

21 

22def serialize_content_for_openai(content: MessageContent) -> str | list[dict[str, Any]]: 

23 """Serialize ``MessageContent`` to the OpenAI content wire format. 

24 

25 - Plain ``str`` → returned unchanged. 

26 - ``list[ContentPart]`` → list of OpenAI-typed dicts. 

27 - ``ImageUrlPart`` → ``{"type": "image_url", "image_url": {"url": ..., "detail": ...}}``. 

28 - ``ImageBase64Part`` → encoded as ``data:<mime>;base64,<data>`` URL. 

29 

30 Args: 

31 content: Message content to serialize. 

32 

33 Returns: 

34 String or list of OpenAI content dicts. 

35 """ 

36 if isinstance(content, str): 

37 return content 

38 

39 parts: list[dict[str, Any]] = [] 

40 for part in content: 

41 if isinstance(part, TextPart): 

42 parts.append({"type": "text", "text": part.text}) 

43 elif isinstance(part, ImageUrlPart): 

44 parts.append( 

45 { 

46 "type": "image_url", 

47 "image_url": {"url": part.url, "detail": part.detail}, 

48 } 

49 ) 

50 elif isinstance(part, ImageBase64Part): 

51 data_uri = f"data:{part.media_type};base64,{part.data}" 

52 parts.append( 

53 { 

54 "type": "image_url", 

55 "image_url": {"url": data_uri, "detail": "auto"}, 

56 } 

57 ) 

58 return parts 

59 

60 

61def serialize_content_for_anthropic( 

62 content: MessageContent, 

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

64 """Serialize ``MessageContent`` to the Anthropic content block format. 

65 

66 Anthropic always wants a list of typed blocks, even for plain text. 

67 

68 - ``str`` → ``[{"type": "text", "text": ...}]`` 

69 - ``ImageUrlPart`` → ``{"type": "image", "source": {"type": "url", "url": ...}}`` 

70 - ``ImageBase64Part`` → ``{"type": "image", "source": {"type": "base64", ...}}`` 

71 

72 Args: 

73 content: Message content to serialize. 

74 

75 Returns: 

76 List of Anthropic content blocks. 

77 """ 

78 if isinstance(content, str): 

79 return [{"type": "text", "text": content}] 

80 

81 blocks: list[dict[str, Any]] = [] 

82 for part in content: 

83 if isinstance(part, TextPart): 

84 blocks.append({"type": "text", "text": part.text}) 

85 elif isinstance(part, ImageUrlPart): 

86 blocks.append( 

87 { 

88 "type": "image", 

89 "source": {"type": "url", "url": part.url}, 

90 } 

91 ) 

92 elif isinstance(part, ImageBase64Part): 

93 blocks.append( 

94 { 

95 "type": "image", 

96 "source": { 

97 "type": "base64", 

98 "media_type": part.media_type, 

99 "data": part.data, 

100 }, 

101 } 

102 ) 

103 return blocks 

104 

105 

106def serialize_content_for_gemini(content: MessageContent) -> list[dict[str, Any]]: 

107 """Serialize ``MessageContent`` to Gemini content part format. 

108 

109 - ``str`` → ``[{"text": ...}]`` 

110 - ``ImageUrlPart`` → ``{"file_data": {"mime_type": ..., "file_uri": url}}`` 

111 - ``ImageBase64Part`` → ``{"inline_data": {"mime_type": ..., "data": base64}}`` 

112 

113 Gemini supports HTTP/HTTPS URLs directly via ``file_uri`` (Gemini 1.5+). 

114 

115 Args: 

116 content: Message content to serialize. 

117 

118 Returns: 

119 List of Gemini content part dicts. 

120 """ 

121 if isinstance(content, str): 

122 return [{"text": content}] 

123 

124 parts: list[dict[str, Any]] = [] 

125 for part in content: 

126 if isinstance(part, TextPart): 

127 parts.append({"text": part.text}) 

128 elif isinstance(part, ImageUrlPart): 

129 mime = _guess_mime_from_url(part.url) 

130 parts.append({"file_data": {"mime_type": mime, "file_uri": part.url}}) 

131 elif isinstance(part, ImageBase64Part): 

132 parts.append( 

133 {"inline_data": {"mime_type": part.media_type, "data": part.data}} 

134 ) 

135 return parts 

136 

137 

138def serialize_text_for_ollama(content: MessageContent) -> tuple[str, list[str]]: 

139 """Serialize ``MessageContent`` to Ollama's ``(text, images)`` format. 

140 

141 Ollama uses ``content`` (text) + ``images`` (list of raw base64 strings, 

142 no ``data:`` prefix). 

143 

144 - ``str`` → ``(str, [])`` 

145 - ``TextPart`` → text is concatenated 

146 - ``ImageBase64Part`` → appended to ``images`` list 

147 - ``ImageUrlPart`` → callers must pre-fetch via ``fetch_image_as_base64``; 

148 a ``[image: <url>]`` placeholder is inserted into text so the content 

149 is not silently dropped. 

150 

151 Args: 

152 content: Message content to serialize. 

153 

154 Returns: 

155 Tuple of (text_content, base64_images_list). 

156 """ 

157 if isinstance(content, str): 

158 return content, [] 

159 

160 text_parts: list[str] = [] 

161 images: list[str] = [] 

162 for part in content: 

163 if isinstance(part, TextPart): 

164 text_parts.append(part.text) 

165 elif isinstance(part, ImageBase64Part): 

166 images.append(part.data) 

167 elif isinstance(part, ImageUrlPart): 

168 text_parts.append(f"[image: {part.url}]") 

169 return " ".join(text_parts), images 

170 

171 

172def serialize_text_only( 

173 content: MessageContent, 

174 *, 

175 logger: Any, 

176 client_name: str, 

177) -> str: 

178 """Extract text from ``MessageContent``, warning on image parts. 

179 

180 For clients that do not support vision. Image parts produce a structured 

181 warning log and are excluded from the returned text. 

182 

183 Args: 

184 content: MessageContent to extract text from. 

185 logger: structlog logger instance (must have ``.warning()``). 

186 client_name: Client name for warning context (e.g. ``"cloudflare"``). 

187 

188 Returns: 

189 Plain text string with image parts omitted. 

190 """ 

191 if isinstance(content, str): 

192 return content 

193 

194 texts: list[str] = [] 

195 for part in content: 

196 if isinstance(part, TextPart): 

197 texts.append(part.text) 

198 elif isinstance(part, (ImageUrlPart, ImageBase64Part)): 

199 logger.warning( 

200 "multimodal_image_dropped", 

201 client=client_name, 

202 reason="client does not support vision", 

203 part_type=part.type, 

204 ) 

205 return " ".join(texts) 

206 

207 

208def _guess_mime_from_url(url: str) -> str: 

209 """Guess image MIME type from URL file extension. 

210 

211 Args: 

212 url: Image URL. 

213 

214 Returns: 

215 MIME type string, defaulting to ``"image/jpeg"``. 

216 """ 

217 lower = url.lower().split("?")[0] 

218 if lower.endswith(".png"): 

219 return "image/png" 

220 if lower.endswith(".gif"): 

221 return "image/gif" 

222 if lower.endswith(".webp"): 

223 return "image/webp" 

224 return "image/jpeg"