Coverage for agentos/multimodal/provider.py: 0%

1""" 

2v1.10.0: Multimodal Provider — Vision & Audio abstraction layer. 

3 

4Supports: 

5- VisionProvider: image→text (base class + OpenAI/VLLM adapters) 

6- AudioProvider: TTS + STT (base class + adapters) 

7- MultiModalMessage: unified multimodal message format 

8""" 

9 

10from __future__ import annotations 

11 

12import base64 

13from dataclasses import dataclass, field 

14from enum import Enum 

15from pathlib import Path 

16from typing import Any, Protocol, runtime_checkable 

17 

18 

19# ── Enums & Data Classes ────────────────────────────────────────── 

20 

21class Modality(str, Enum): 

22 TEXT = "text" 

23 IMAGE = "image" 

24 AUDIO = "audio" 

25 VIDEO = "video" 

26 

27 

28class ImageFormat(str, Enum): 

29 PNG = "png" 

30 JPEG = "jpeg" 

31 WEBP = "webp" 

32 GIF = "gif" 

33 SVG = "svg" 

34 

35 

36class AudioFormat(str, Enum): 

37 MP3 = "mp3" 

38 WAV = "wav" 

39 OGG = "ogg" 

40 FLAC = "flac" 

41 AAC = "aac" 

42 

43 

44@dataclass 

45class MultiModalContent: 

46 """A piece of multimodal content.""" 

47 type: Modality 

48 text: str = "" 

49 data: bytes = field(default=b"", repr=False) 

50 data_url: str = "" # data:image/png;base64,... 

51 mime_type: str = "" 

52 metadata: dict[str, Any] = field(default_factory=dict) 

53 

54 @staticmethod 

55 def text(content: str) -> "MultiModalContent": 

56 return MultiModalContent(type=Modality.TEXT, text=content) 

57 

58 @staticmethod 

59 def from_path(path: str | Path) -> "MultiModalContent": 

60 path = Path(path) 

61 data = path.read_bytes() 

62 ext = path.suffix.lower().lstrip(".") 

63 fmt_map = {"png": "image/png", "jpg": "image/jpeg", "jpeg": "image/jpeg", 

64 "webp": "image/webp", "gif": "image/gif", "mp3": "audio/mpeg", 

65 "wav": "audio/wav", "ogg": "audio/ogg", "flac": "audio/flac"} 

66 mime = fmt_map.get(ext, "application/octet-stream") 

67 b64 = base64.b64encode(data).decode() 

68 modality = Modality.IMAGE if mime.startswith("image/") else ( 

69 Modality.AUDIO if mime.startswith("audio/") else Modality.TEXT 

70 ) 

71 return MultiModalContent( 

72 type=modality, data=data, 

73 data_url=f"data:{mime};base64,{b64}", 

74 mime_type=mime, 

75 ) 

76 

77 @staticmethod 

78 def from_bytes(data: bytes, mime_type: str = "image/png") -> "MultiModalContent": 

79 b64 = base64.b64encode(data).decode() 

80 modality = Modality.IMAGE if "image" in mime_type else ( 

81 Modality.AUDIO if "audio" in mime_type else Modality.TEXT 

82 ) 

83 return MultiModalContent( 

84 type=modality, data=data, 

85 data_url=f"data:{mime_type};base64,{b64}", 

86 mime_type=mime_type, 

87 ) 

88 

89 

90@dataclass 

91class MultiModalMessage: 

92 """A multimodal message with mixed content blocks.""" 

93 role: str = "user" # system / user / assistant 

94 content: list[MultiModalContent] = field(default_factory=list) 

95 metadata: dict[str, Any] = field(default_factory=dict) 

96 

97 def add_text(self, text: str) -> "MultiModalMessage": 

98 self.content.append(MultiModalContent.text(text)) 

99 return self 

100 

101 def add_image_path(self, path: str | Path) -> "MultiModalMessage": 

102 self.content.append(MultiModalContent.from_path(path)) 

103 return self 

104 

105 def add_audio_path(self, path: str | Path) -> "MultiModalMessage": 

106 self.content.append(MultiModalContent.from_path(path)) 

107 return self 

108 

109 def to_openai_format(self) -> dict[str, Any]: 

110 """Convert to OpenAI chat completion message format.""" 

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

112 for c in self.content: 

113 if c.type == Modality.TEXT: 

114 blocks.append({"type": "text", "text": c.text}) 

115 elif c.type == Modality.IMAGE: 

116 blocks.append({ 

117 "type": "image_url", 

118 "image_url": {"url": c.data_url, "detail": "auto"}, 

119 }) 

120 elif c.type == Modality.AUDIO: 

121 blocks.append({ 

122 "type": "input_audio", 

123 "input_audio": {"data": base64.b64encode(c.data).decode(), "format": c.mime_type.split("/")[-1] if c.mime_type else "wav"}, 

124 }) 

125 return {"role": self.role, "content": blocks} 

126 

127 def to_gemini_format(self) -> dict[str, Any]: 

128 """Convert to Gemini API message format.""" 

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

130 for c in self.content: 

131 if c.type == Modality.TEXT: 

132 parts.append({"text": c.text}) 

133 elif c.type == Modality.IMAGE: 

134 parts.append({ 

135 "inline_data": { 

136 "mime_type": c.mime_type or "image/png", 

137 "data": base64.b64encode(c.data).decode(), 

138 } 

139 }) 

140 elif c.type == Modality.AUDIO: 

141 parts.append({ 

142 "inline_data": { 

143 "mime_type": c.mime_type or "audio/wav", 

144 "data": base64.b64encode(c.data).decode(), 

145 } 

146 }) 

147 return {"role": "user" if self.role == "user" else "model", "parts": parts} 

148 

149 

150# ── Vision Provider ─────────────────────────────────────────────── 

151 

152@runtime_checkable 

153class VisionProvider(Protocol): 

154 """Protocol for vision providers (image → text).""" 

155 

156 async def describe(self, image: MultiModalContent, prompt: str = "") -> str: 

157 """Describe an image. Returns text description.""" 

158 ... 

159 

160 async def ask(self, images: list[MultiModalContent], question: str) -> str: 

161 """Ask a question about one or more images.""" 

162 ... 

163 

164 

165class OpenAIVisionProvider: 

166 """OpenAI GPT-4V / GPT-4o vision provider.""" 

167 

168 def __init__(self, api_key: str = "", model: str = "gpt-4o", base_url: str = ""): 

169 self.api_key = api_key 

170 self.model = model 

171 self.base_url = base_url 

172 

173 async def describe(self, image: MultiModalContent, prompt: str = "") -> str: 

174 return await self.ask([image], prompt or "Describe this image in detail.") 

175 

176 async def ask(self, images: list[MultiModalContent], question: str) -> str: 

177 import aiohttp 

178 

179 message = MultiModalMessage(role="user") 

180 for img in images: 

181 message.content.append(img) 

182 message.add_text(question) 

183 body = message.to_openai_format() 

184 

185 payload = { 

186 "model": self.model, 

187 "messages": [ 

188 {"role": "system", "content": "You are a helpful vision assistant."}, 

189 body, 

190 ], 

191 "max_tokens": 1024, 

192 } 

193 

194 url = f"{self.base_url}/chat/completions" if self.base_url else "https://api.openai.com/v1/chat/completions" 

195 headers = {"Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json"} 

196 

197 async with aiohttp.ClientSession() as session: 

198 async with session.post(url, json=payload, headers=headers) as resp: 

199 result = await resp.json() 

200 return result["choices"][0]["message"]["content"] 

201 

202 

203class LocalVisionProvider: 

204 """Local vision provider (placeholder for vLLM/Ollama).""" 

205 

206 def __init__(self, endpoint: str = "http://localhost:11434", model: str = "llava"): 

207 self.endpoint = endpoint 

208 self.model = model 

209 

210 async def describe(self, image: MultiModalContent, prompt: str = "") -> str: 

211 return await self.ask([image], prompt or "Describe this image.") 

212 

213 async def ask(self, images: list[MultiModalContent], question: str) -> str: 

214 import aiohttp 

215 

216 async with aiohttp.ClientSession() as session: 

217 async with session.post( 

218 f"{self.endpoint}/api/generate", 

219 json={ 

220 "model": self.model, 

221 "prompt": question, 

222 "images": [img.data_url.split(",", 1)[1] for img in images if img.data_url], 

223 "stream": False, 

224 }, 

225 ) as resp: 

226 result = await resp.json() 

227 return result.get("response", "") 

228 

229 

230# ── Audio Provider ───────────────────────────────────────────────── 

231 

232@runtime_checkable 

233class AudioProvider(Protocol): 

234 """Protocol for audio providers (TTS + STT).""" 

235 

236 async def transcribe(self, audio: MultiModalContent, language: str = "") -> str: 

237 """Speech-to-text: transcribe audio to text.""" 

238 ... 

239 

240 async def synthesize(self, text: str, voice: str = "alloy", speed: float = 1.0) -> MultiModalContent: 

241 """Text-to-speech: generate audio from text.""" 

242 ... 

243 

244 

245class OpenAIAudioProvider: 

246 """OpenAI Whisper + TTS audio provider.""" 

247 

248 def __init__(self, api_key: str = "", tts_model: str = "tts-1", stt_model: str = "whisper-1"): 

249 self.api_key = api_key 

250 self.tts_model = tts_model 

251 self.stt_model = stt_model 

252 

253 async def transcribe(self, audio: MultiModalContent, language: str = "") -> str: 

254 import aiohttp 

255 

256 form = aiohttp.FormData() 

257 form.add_field("model", self.stt_model) 

258 form.add_field("file", audio.data, filename=f"audio.{audio.mime_type.split('/')[-1] or 'wav'}", 

259 content_type=audio.mime_type or "audio/wav") 

260 if language: 

261 form.add_field("language", language) 

262 

263 headers = {"Authorization": f"Bearer {self.api_key}"} 

264 async with aiohttp.ClientSession() as session: 

265 async with session.post("https://api.openai.com/v1/audio/transcriptions", 

266 data=form, headers=headers) as resp: 

267 result = await resp.json() 

268 return result.get("text", "") 

269 

270 async def synthesize(self, text: str, voice: str = "alloy", speed: float = 1.0) -> MultiModalContent: 

271 import aiohttp 

272 

273 payload = { 

274 "model": self.tts_model, "input": text, 

275 "voice": voice, "speed": speed, 

276 "response_format": "mp3", 

277 } 

278 headers = {"Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json"} 

279 async with aiohttp.ClientSession() as session: 

280 async with session.post("https://api.openai.com/v1/audio/speech", 

281 json=payload, headers=headers) as resp: 

282 audio_data = await resp.read() 

283 return MultiModalContent.from_bytes(audio_data, "audio/mpeg") 

284 

285 

286class EdgeTTSProvider: 

287 """Microsoft Edge TTS (free, local).""" 

288 

289 def __init__(self, voice: str = "zh-CN-XiaoxiaoNeural"): 

290 self.voice = voice 

291 

292 async def synthesize(self, text: str, voice: str = "", speed: float = 1.0) -> MultiModalContent: 

293 import edge_tts # type: ignore[import-untyped] 

294 

295 voice_name = voice or self.voice 

296 rate = f"{int((speed - 1.0) * 100):+d}%" 

297 communicate = edge_tts.Communicate(text, voice_name, rate=rate) 

298 audio_chunks = [] 

299 async for chunk in communicate.stream(): 

300 if chunk["type"] == "audio": 

301 audio_chunks.append(chunk["data"]) 

302 audio_data = b"".join(audio_chunks) 

303 return MultiModalContent.from_bytes(audio_data, "audio/mpeg") 

304 

305 

306# ── MultiModal Client ───────────────────────────────────────────── 

307 

308class MultiModalClient: 

309 """Unified multimodal client: vision + audio in one interface.""" 

310 

311 def __init__( 

312 self, 

313 vision: VisionProvider | None = None, 

314 audio: AudioProvider | None = None, 

315 ): 

316 self.vision = vision or LocalVisionProvider() 

317 self.audio = audio 

318 

319 async def see(self, image_path: str | Path, question: str = "What's in this image?") -> str: 

320 """Look at an image and answer a question about it.""" 

321 img = MultiModalContent.from_path(image_path) 

322 return await self.vision.ask([img], question) 

323 

324 async def hear(self, audio_path: str | Path, language: str = "") -> str: 

325 """Transcribe audio to text.""" 

326 if not self.audio: 

327 raise RuntimeError("No audio provider configured") 

328 audio = MultiModalContent.from_path(audio_path) 

329 return await self.audio.transcribe(audio, language) 

330 

331 async def speak(self, text: str, voice: str = "alloy", speed: float = 1.0) -> MultiModalContent: 

332 """Generate speech from text.""" 

333 if not self.audio: 

334 raise RuntimeError("No audio provider configured") 

335 return await self.audio.synthesize(text, voice, speed)