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 StrEnum 

15from pathlib import Path 

16from typing import Any, Protocol, runtime_checkable 

17 

18# ── Enums & Data Classes ────────────────────────────────────────── 

19 

20 

21class Modality(StrEnum): 

22 TEXT = "text" 

23 IMAGE = "image" 

24 AUDIO = "audio" 

25 VIDEO = "video" 

26 

27 

28class ImageFormat(StrEnum): 

29 PNG = "png" 

30 JPEG = "jpeg" 

31 WEBP = "webp" 

32 GIF = "gif" 

33 SVG = "svg" 

34 

35 

36class AudioFormat(StrEnum): 

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 

48 type: Modality 

49 text: str = "" 

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

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

52 mime_type: str = "" 

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

54 

55 @staticmethod 

56 def from_text(content: str) -> MultiModalContent: 

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

58 

59 @staticmethod 

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

61 path = Path(path) 

62 data = path.read_bytes() 

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

64 fmt_map = { 

65 "png": "image/png", 

66 "jpg": "image/jpeg", 

67 "jpeg": "image/jpeg", 

68 "webp": "image/webp", 

69 "gif": "image/gif", 

70 "mp3": "audio/mpeg", 

71 "wav": "audio/wav", 

72 "ogg": "audio/ogg", 

73 "flac": "audio/flac", 

74 } 

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

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

77 modality = ( 

78 Modality.IMAGE 

79 if mime.startswith("image/") 

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

81 ) 

82 return MultiModalContent( 

83 type=modality, 

84 data=data, 

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

86 mime_type=mime, 

87 ) 

88 

89 @staticmethod 

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

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

92 modality = ( 

93 Modality.IMAGE 

94 if "image" in mime_type 

95 else (Modality.AUDIO if "audio" in mime_type else Modality.TEXT) 

96 ) 

97 return MultiModalContent( 

98 type=modality, 

99 data=data, 

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

101 mime_type=mime_type, 

102 ) 

103 

104 

105@dataclass 

106class MultiModalMessage: 

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

108 

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

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

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

112 

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

114 self.content.append(MultiModalContent.from_text(text)) 

115 return self 

116 

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

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

119 return self 

120 

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

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

123 return self 

124 

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

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

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

128 for c in self.content: 

129 if c.type == Modality.TEXT: 

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

131 elif c.type == Modality.IMAGE: 

132 blocks.append( 

133 { 

134 "type": "image_url", 

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

136 } 

137 ) 

138 elif c.type == Modality.AUDIO: 

139 blocks.append( 

140 { 

141 "type": "input_audio", 

142 "input_audio": { 

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

144 "format": c.mime_type.split("/")[-1] if c.mime_type else "wav", 

145 }, 

146 } 

147 ) 

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

149 

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

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

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

153 for c in self.content: 

154 if c.type == Modality.TEXT: 

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

156 elif c.type == Modality.IMAGE: 

157 parts.append( 

158 { 

159 "inline_data": { 

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

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

162 } 

163 } 

164 ) 

165 elif c.type == Modality.AUDIO: 

166 parts.append( 

167 { 

168 "inline_data": { 

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

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

171 } 

172 } 

173 ) 

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

175 

176 

177# ── Vision Provider ─────────────────────────────────────────────── 

178 

179 

180@runtime_checkable 

181class VisionProvider(Protocol): 

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

183 

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

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

186 ... 

187 

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

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

190 ... 

191 

192 

193class OpenAIVisionProvider: 

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

195 

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

197 self.api_key = api_key 

198 self.model = model 

199 self.base_url = base_url 

200 

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

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

203 

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

205 import aiohttp 

206 

207 message = MultiModalMessage(role="user") 

208 for img in images: 

209 message.content.append(img) 

210 message.add_text(question) 

211 body = message.to_openai_format() 

212 

213 payload = { 

214 "model": self.model, 

215 "messages": [ 

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

217 body, 

218 ], 

219 "max_tokens": 1024, 

220 } 

221 

222 url = ( 

223 f"{self.base_url}/chat/completions" 

224 if self.base_url 

225 else "https://api.openai.com/v1/chat/completions" 

226 ) 

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

228 

229 async with aiohttp.ClientSession() as session: 

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

231 result = await resp.json() 

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

233 

234 

235class LocalVisionProvider: 

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

237 

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

239 self.endpoint = endpoint 

240 self.model = model 

241 

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

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

244 

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

246 import aiohttp 

247 

248 async with aiohttp.ClientSession() as session: 

249 async with session.post( 

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

251 json={ 

252 "model": self.model, 

253 "prompt": question, 

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

255 "stream": False, 

256 }, 

257 ) as resp: 

258 result = await resp.json() 

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

260 

261 

262# ── Audio Provider ───────────────────────────────────────────────── 

263 

264 

265@runtime_checkable 

266class AudioProvider(Protocol): 

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

268 

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

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

271 ... 

272 

273 async def synthesize( 

274 self, text: str, voice: str = "alloy", speed: float = 1.0 

275 ) -> MultiModalContent: 

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

277 ... 

278 

279 

280class OpenAIAudioProvider: 

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

282 

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

284 self.api_key = api_key 

285 self.tts_model = tts_model 

286 self.stt_model = stt_model 

287 

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

289 import aiohttp 

290 

291 form = aiohttp.FormData() 

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

293 form.add_field( 

294 "file", 

295 audio.data, 

296 filename=f"audio.{audio.mime_type.split('/')[-1] or 'wav'}", 

297 content_type=audio.mime_type or "audio/wav", 

298 ) 

299 if language: 

300 form.add_field("language", language) 

301 

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

303 async with aiohttp.ClientSession() as session: 

304 async with session.post( 

305 "https://api.openai.com/v1/audio/transcriptions", data=form, headers=headers 

306 ) as resp: 

307 result = await resp.json() 

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

309 

310 async def synthesize( 

311 self, text: str, voice: str = "alloy", speed: float = 1.0 

312 ) -> MultiModalContent: 

313 import aiohttp 

314 

315 payload = { 

316 "model": self.tts_model, 

317 "input": text, 

318 "voice": voice, 

319 "speed": speed, 

320 "response_format": "mp3", 

321 } 

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

323 async with aiohttp.ClientSession() as session: 

324 async with session.post( 

325 "https://api.openai.com/v1/audio/speech", json=payload, headers=headers 

326 ) as resp: 

327 audio_data = await resp.read() 

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

329 

330 

331class EdgeTTSProvider: 

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

333 

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

335 self.voice = voice 

336 

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

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

339 

340 voice_name = voice or self.voice 

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

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

343 audio_chunks = [] 

344 async for chunk in communicate.stream(): 

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

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

347 audio_data = b"".join(audio_chunks) 

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

349 

350 

351# ── MultiModal Client ───────────────────────────────────────────── 

352 

353 

354class MultiModalClient: 

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

356 

357 def __init__( 

358 self, 

359 vision: VisionProvider | None = None, 

360 audio: AudioProvider | None = None, 

361 ): 

362 self.vision = vision or LocalVisionProvider() 

363 self.audio = audio 

364 

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

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

367 img = MultiModalContent.from_path(image_path) 

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

369 

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

371 """Transcribe audio to text.""" 

372 if not self.audio: 

373 raise RuntimeError("No audio provider configured") 

374 audio = MultiModalContent.from_path(audio_path) 

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

376 

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

378 """Generate speech from text.""" 

379 if not self.audio: 

380 raise RuntimeError("No audio provider configured") 

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