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

1""" 

2AgentOS v0.40 Multimodal — 多模态输入支持。 

3支持：图片理解、语音转文字、PDF/文档解析。 

4""" 

5 

6from __future__ import annotations 

7 

8import base64 

9import logging 

10from dataclasses import dataclass, field 

11from enum import Enum 

12 

13 

14logger = logging.getLogger(__name__) 

15 

16 

17class Modality(str, Enum): 

18 

19 """模态类型枚举。""" 

20 

21 TEXT = "text" 

22 IMAGE = "image" 

23 AUDIO = "audio" 

24 VIDEO = "video" 

25 DOCUMENT = "document" 

26 

27 

28@dataclass 

29class MultimodalBlock: 

30 """多模态输入块 — 遵循OpenAI/Anthropic content block格式。""" 

31 type: str # text | image_url | audio | image 

32 text: str = "" 

33 source: dict = field(default_factory=dict) 

34 mime_type: str = "" 

35 

36 @classmethod 

37 def text_block(cls, text: str) -> "MultimodalBlock": 

38 return cls(type="text", text=text) 

39 

40 @classmethod 

41 def image_url(cls, url: str, detail: str = "auto") -> "MultimodalBlock": 

42 return cls(type="image_url", source={"type": "image_url", "image_url": {"url": url, "detail": detail}}) 

43 

44 @classmethod 

45 def image_base64(cls, data: bytes, mime: str = "image/jpeg") -> "MultimodalBlock": 

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

47 return cls(type="image_url", source={"type": "image_url", "image_url": {"url": f"data:{mime};base64,{b64}"}}) 

48 

49 @classmethod 

50 def audio(cls, data: bytes, mime: str = "audio/wav") -> "MultimodalBlock": 

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

52 return cls(type="audio", mime_type=mime, source={"data": b64}) 

53 

54 def to_openai_format(self) -> dict: 

55 if self.type == "text": 

56 return {"type": "text", "text": self.text} 

57 if self.type == "image_url": 

58 return {"type": "image_url", "image_url": self.source["image_url"]} 

59 return {"type": self.type, **self.source} 

60 

61 

62class ImageProcessor: 

63 """图片处理器 — 压缩、格式转换、OCR预处理。""" 

64 

65 MAX_SIZE = 2048 

66 JPEG_QUALITY = 85 

67 

68 @staticmethod 

69 def encode_file(path: str) -> tuple[str, str]: 

70 """返回(base64, mime_type)。""" 

71 import mimetypes 

72 mime = mimetypes.guess_type(path)[0] or "image/png" 

73 with open(path, "rb") as f: 

74 data = f.read() 

75 return base64.b64encode(data).decode(), mime 

76 

77 @staticmethod 

78 def encode_bytes(data: bytes, mime: str = "image/jpeg") -> str: 

79 return base64.b64encode(data).decode() 

80 

81 @staticmethod 

82 def estimate_tokens(width: int, height: int, detail: str = "auto") -> int: 

83 """估算图片token消耗（OpenAI定价模型）。""" 

84 if detail == "low": 

85 return 85 

86 # high detail 

87 short_side = min(width, height) 

88 scale = min(768 / short_side, 1.0) if short_side > 768 else 1.0 

89 w = int(width * scale) 

90 h = int(height * scale) 

91 tiles = ((w + 511) // 512) * ((h + 511) // 512) 

92 return 85 + 170 * tiles 

93 

94 @staticmethod 

95 def purge_metadata(data: bytes) -> bytes: 

96 """清除图片EXIF元数据。""" 

97 try: 

98 from PIL import Image 

99 import io 

100 img = Image.open(io.BytesIO(data)) 

101 data_no_exif = list(img.getdata()) 

102 cleaned = Image.new(img.mode, img.size) 

103 cleaned.putdata(data_no_exif) 

104 buf = io.BytesIO() 

105 cleaned.save(buf, format=img.format or "PNG") 

106 return buf.getvalue() 

107 except ImportError: 

108 return data 

109 

110 

111class AudioProcessor: 

112 """音频处理器 — 转文字、格式转换。""" 

113 

114 SUPPORTED_FORMATS = ["wav", "mp3", "ogg", "flac", "m4a"] 

115 

116 @staticmethod 

117 def transcribe(path: str, whisper_model: str = "base") -> str: 

118 """使用whisper转文字。""" 

119 try: 

120 import whisper 

121 model = whisper.load_model(whisper_model) 

122 result = model.transcribe(path) 

123 return result["text"] 

124 except ImportError: 

125 logger.warning("whisper not installed, returning empty") 

126 return "[whisper not available]" 

127 

128 @staticmethod 

129 def encode_file(path: str) -> tuple[str, str]: 

130 import mimetypes 

131 mime = mimetypes.guess_type(path)[0] or "audio/wav" 

132 with open(path, "rb") as f: 

133 data = f.read() 

134 return base64.b64encode(data).decode(), mime 

135 

136 

137class DocumentParser: 

138 """文档解析器 — PDF/Word/Markdown。""" 

139 

140 @staticmethod 

141 def parse_pdf(path: str) -> str: 

142 try: 

143 import PyPDF2 

144 text = [] 

145 with open(path, "rb") as f: 

146 reader = PyPDF2.PdfReader(f) 

147 for page in reader.pages: 

148 page_text = page.extract_text() 

149 if page_text: 

150 text.append(page_text) 

151 return "\n\n".join(text) 

152 except ImportError: 

153 logger.warning("PyPDF2 not installed") 

154 return "[PyPDF2 not available]" 

155 

156 @staticmethod 

157 def parse_docx(path: str) -> str: 

158 try: 

159 from docx import Document 

160 doc = Document(path) 

161 return "\n".join(p.text for p in doc.paragraphs if p.text) 

162 except ImportError: 

163 logger.warning("python-docx not installed") 

164 return "[python-docx not available]" 

165 

166 @staticmethod 

167 def parse_auto(path: str) -> tuple[str, str]: 

168 """自动检测文件类型并解析。返回 (content, format)。""" 

169 ext = path.rsplit(".", 1)[-1].lower() if "." in path else "" 

170 if ext == "pdf": 

171 return DocumentParser.parse_pdf(path), "pdf" 

172 elif ext in ("docx", "doc"): 

173 return DocumentParser.parse_docx(path), "docx" 

174 elif ext in ("md", "markdown", "txt"): 

175 with open(path) as f: 

176 return f.read(), ext 

177 else: 

178 try: 

179 with open(path) as f: 

180 return f.read(), "text" 

181 except Exception: 

182 return "", "unknown" 

183 

184 

185class MultimodalManager: 

186 """多模态管理器 — 统一入口。""" 

187 

188 def __init__(self): 

189 self.image = ImageProcessor() 

190 self.audio = AudioProcessor() 

191 self.document = DocumentParser() 

192 

193 def prepare_input(self, blocks: list[MultimodalBlock]) -> list[dict]: 

194 """转换为OpenAI兼容格式。""" 

195 return [b.to_openai_format() for b in blocks] 

196 

197 def from_files(self, paths: list[str]) -> list[MultimodalBlock]: 

198 """从文件路径自动推断模态。""" 

199 blocks = [] 

200 image_exts = {"png", "jpg", "jpeg", "gif", "webp", "bmp"} 

201 audio_exts = {"wav", "mp3", "ogg", "flac", "m4a"} 

202 doc_exts = {"pdf", "docx", "doc", "md", "txt"} 

203 

204 for p in paths: 

205 ext = p.rsplit(".", 1)[-1].lower() if "." in p else "" 

206 try: 

207 if ext in image_exts: 

208 b64, mime = ImageProcessor.encode_file(p) 

209 blocks.append(MultimodalBlock(type="image_url", 

210 source={"type": "image_url", "image_url": {"url": f"data:{mime};base64,{b64}"}})) 

211 elif ext in audio_exts: 

212 b64, mime = AudioProcessor.encode_file(p) 

213 blocks.append(MultimodalBlock(type="audio", mime_type=mime, source={"data": b64})) 

214 elif ext in doc_exts: 

215 text, fmt = DocumentParser.parse_auto(p) 

216 blocks.append(MultimodalBlock.text_block(text)) 

217 else: 

218 with open(p) as f: 

219 blocks.append(MultimodalBlock.text_block(f.read())) 

220 except Exception as e: 

221 blocks.append(MultimodalBlock.text_block(f"[Error reading {p}: {e}]")) 

222 return blocks 

223 

224 def stats(self) -> dict: 

225 return {"modalities": ["text", "image", "audio", "video", "document"]}