Coverage for agentos/multimodal/__init_

1"""

2AgentOS v1.14.3 — Multimodal Context Manager.

4Unified multimodal context layer for AgentOS agents. Handles images,

5audio, video, and structured documents as first-class context objects.

7Features:

8- Multi-format image processing (PNG, JPEG, GIF, WebP, SVG, HEIC)

9- Audio transcription & processing (WAV, MP3, FLAC, M4A)

10- Video keyframe extraction & captioning

11- PDF/DOCX document text extraction with layout awareness

12- File type auto-detection (magic bytes)

13- Image preprocessing pipeline (resize, compress, format convert)

14- Vision LLM adapter for base64 images

15- Thumbnail generation

16- Metadata extraction (EXIF, duration, dimensions)

18Architecture:

19 File Input

20 ├── MediaDetector (magic bytes identification)

21 ├── MediaProcessor (format-specific pipeline)

22 │ ├── ImageProcessor (resize/compress/convert/base64)

23 │ ├── AudioProcessor (transcription via whisper)

24 │ ├── VideoProcessor (keyframe extraction)

25 │ └── DocumentProcessor (PDF/DOCX extraction)

26 └── MediaContext (unified context object)

28Inspired by: GPT-4V multimodal API, Claude Vision, Gemini 1.5 Pro

29"""

31from __future__ import annotations

33import base64

34import json

35import mimetypes

36import os

37import subprocess

38import tempfile

39import uuid

40from abc import ABC, abstractmethod

41from dataclasses import dataclass, field

42from enum import StrEnum

43from pathlib import Path

44from typing import (

45 Any,

46)

48# ── Types ───────────────────────────────────

51class MediaType(StrEnum):

52 IMAGE = "image"

53 AUDIO = "audio"

54 VIDEO = "video"

55 DOCUMENT = "document"

56 UNKNOWN = "unknown"

59class ImageFormat(StrEnum):

60 PNG = "png"

61 JPEG = "jpeg"

62 GIF = "gif"

63 WEBP = "webp"

64 SVG = "svg"

65 BMP = "bmp"

66 HEIC = "heic"

67 TIFF = "tiff"

70@dataclass

71class MediaMetadata:

72 """媒体文件元数据。"""

74 file_path: str = ""

75 media_type: MediaType = MediaType.UNKNOWN

76 mime_type: str = ""

77 file_size_bytes: int = 0

79 # Image

80 width: int = 0

81 height: int = 0

82 color_mode: str = ""

84 # Audio/Video

85 duration_s: float = 0.0

86 sample_rate: int = 0

87 channels: int = 0

88 bitrate_kbps: int = 0

90 # General

91 has_alpha: bool = False

92 page_count: int = 0

93 extra: dict[str, Any] = field(default_factory=dict)

95 def to_dict(self) -> dict:

96 return {

97 "file_path": self.file_path,

98 "media_type": self.media_type.value,

99 "mime_type": self.mime_type,

100 "file_size_bytes": self.file_size_bytes,

101 "width": self.width,

102 "height": self.height,

103 "duration_s": self.duration_s,

104 "page_count": self.page_count,

105 }

106

107

108@dataclass

109class MediaContext:

110 """统一的多模态上下文对象。

111

112 This is what gets passed to LLM context windows.

113 """

114

115 context_id: str = field(default_factory=lambda: f"mctx-{uuid.uuid4().hex[:12]}")

116 media_type: MediaType = MediaType.UNKNOWN

117 metadata: MediaMetadata = field(default_factory=MediaMetadata)

118

119 # Processed representations

120 text_description: str = "" # 自然语言描述

121 base64_data: str = "" # Base64 编码（用于视觉 LLM）

122 extracted_text: str = "" # OCR/转录文本

123 thumbnail_path: str = "" # 缩略图路径

124

125 # Structured

126 entities: list[dict[str, Any]] = field(default_factory=list)

127 captions: list[str] = field(default_factory=list)

128

129 def to_llm_message(self) -> dict:

130 """转换为 LLM API 消息格式。"""

131 if self.media_type == MediaType.IMAGE and self.base64_data:

132 return {

133 "role": "user",

134 "content": [

135 {

136 "type": "image_url",

137 "image_url": {

138 "url": f"data:{self.metadata.mime_type};base64,{self.base64_data}",

139 "detail": "auto",

140 },

141 },

142 {

143 "type": "text",

144 "text": self.text_description or "Describe this image.",

145 },

146 ],

147 }

148 return {

149 "role": "user",

150 "content": self.text_description or self.extracted_text or "",

151 }

152

153

154# ── Media Detector ──────────────────────────

155

156

157class MediaDetector:

158 """通过文件魔数 (magic bytes) 检测媒体类型。

159

160 Usage:

161 detector = MediaDetector()

162 mt = detector.detect("photo.jpg") # MediaType.IMAGE

163 """

164

165 # Magic bytes signatures

166 MAGIC_SIGNATURES = {

167 b"\xff\xd8\xff": (MediaType.IMAGE, ImageFormat.JPEG),

168 b"\x89PNG\r\n\x1a\n": (MediaType.IMAGE, ImageFormat.PNG),

169 b"GIF87a": (MediaType.IMAGE, ImageFormat.GIF),

170 b"GIF89a": (MediaType.IMAGE, ImageFormat.GIF),

171 b"RIFF": (MediaType.IMAGE, ImageFormat.WEBP), # WEBP is RIFF{size}WEBP

172 b"\x42\x4d": (MediaType.IMAGE, ImageFormat.BMP),

173 b"<?xml": (MediaType.IMAGE, ImageFormat.SVG),

174 b"<svg": (MediaType.IMAGE, ImageFormat.SVG),

175 b"II*\x00": (MediaType.IMAGE, ImageFormat.TIFF),

176 b"MM\x00*": (MediaType.IMAGE, ImageFormat.TIFF),

177 # Audio

178 b"ID3": (MediaType.AUDIO, None), # MP3 with ID3

179 b"\xff\xfb": (MediaType.AUDIO, None), # MP3

180 b"\xff\xf3": (MediaType.AUDIO, None), # MP3

181 b"fLaC": (MediaType.AUDIO, None), # FLAC

182 b"OggS": (MediaType.AUDIO, None), # OGG

183 # Video

184 b"\x00\x00\x00\x18ftyp": (MediaType.VIDEO, None), # MP4

185 b"\x00\x00\x00\x20ftyp": (MediaType.VIDEO, None),

186 b"\x1a\x45\xdf\xa3": (MediaType.VIDEO, None), # WebM/MKV

187 # Documents

188 b"%PDF": (MediaType.DOCUMENT, None),

189 b"PK\x03\x04": (MediaType.DOCUMENT, None), # DOCX/XLSX/PPTX (ZIP)

190 }

191

192 # Audio extensions

193 AUDIO_EXTENSIONS = {".mp3", ".wav", ".flac", ".m4a", ".ogg", ".aac", ".wma", ".opus"}

194

195 # Video extensions

196 VIDEO_EXTENSIONS = {".mp4", ".avi", ".mkv", ".mov", ".wmv", ".webm", ".flv", ".m4v", ".3gp"}

197

198 # Image extensions

199 IMAGE_EXTENSIONS = {

200 ".jpg",

201 ".jpeg",

202 ".png",

203 ".gif",

204 ".webp",

205 ".bmp",

206 ".svg",

207 ".tiff",

208 ".heic",

209 ".ico",

210 }

211

212 # Document extensions

213 DOCUMENT_EXTENSIONS = {

214 ".pdf",

215 ".docx",

216 ".doc",

217 ".xlsx",

218 ".xls",

219 ".pptx",

220 ".ppt",

221 ".txt",

222 ".md",

223 ".html",

224 ".epub",

225 }

226

227 @classmethod

228 def detect(cls, file_path: str) -> MediaType:

229 """检测文件媒体类型。"""

230 ext = Path(file_path).suffix.lower()

231

232 if ext in cls.IMAGE_EXTENSIONS:

233 return MediaType.IMAGE

234 if ext in cls.AUDIO_EXTENSIONS:

235 return MediaType.AUDIO

236 if ext in cls.VIDEO_EXTENSIONS:

237 return MediaType.VIDEO

238 if ext in cls.DOCUMENT_EXTENSIONS:

239 return MediaType.DOCUMENT

240

241 # Fallback to magic bytes

242 try:

243 with open(file_path, "rb") as f:

244 header = f.read(32)

245 except Exception:

246 return MediaType.UNKNOWN

247

248 for magic, (mtype, _) in cls.MAGIC_SIGNATURES.items():

249 if header.startswith(magic):

250 # RIFF ambiguity resolution

251 if magic == b"RIFF":

252 if b"WEBP" in header:

253 return MediaType.IMAGE

254 if b"WAVE" in header:

255 return MediaType.AUDIO

256 return mtype

257

258 # MIME type fallback

259 mime, _ = mimetypes.guess_type(file_path)

260 if mime:

261 if mime.startswith("image/"):

262 return MediaType.IMAGE

263 if mime.startswith("audio/"):

264 return MediaType.AUDIO

265 if mime.startswith("video/"):

266 return MediaType.VIDEO

267

268 return MediaType.UNKNOWN

269

270 @classmethod

271 def batch_detect(cls, file_paths: list[str]) -> dict[str, MediaType]:

272 """批量检测。"""

273 return {fp: cls.detect(fp) for fp in file_paths}

274

275

276# ── Media Processors ────────────────────────

277

278

279class MediaProcessor(ABC):

280 """媒体处理器基类。"""

281

282 @abstractmethod

283 def process(self, file_path: str) -> MediaContext: ...

284

285 @abstractmethod

286 def extract_metadata(self, file_path: str) -> MediaMetadata: ...

287

288

289class ImageProcessor(MediaProcessor):

290 """图像处理器。

291

292 支持格式转换、缩放、压缩、Base64 编码。

293

294 Usage:

295 processor = ImageProcessor()

296 ctx = processor.process("photo.jpg")

297 base64_str = ctx.base64_data # 可直接用于 LLM API

298 """

299

300 def __init__(

301 self,

302 max_size: int = 2048,

303 quality: int = 85,

304 output_format: str = "JPEG",

305 ):

306 self._max_size = max_size

307 self._quality = quality

308 self._output_format = output_format

309

310 def process(self, file_path: str) -> MediaContext:

311 ctx = MediaContext(media_type=MediaType.IMAGE)

312 ctx.metadata = self.extract_metadata(file_path)

313 ctx.base64_data = self._encode_base64(file_path)

314 ctx.text_description = self._generate_description(file_path)

315 ctx.thumbnail_path = self._generate_thumbnail(file_path)

316 return ctx

317

318 def extract_metadata(self, file_path: str) -> MediaMetadata:

319 meta = MediaMetadata(

320 file_path=file_path,

321 media_type=MediaType.IMAGE,

322 mime_type=mimetypes.guess_type(file_path)[0] or "application/octet-stream",

323 file_size_bytes=os.path.getsize(file_path) if os.path.exists(file_path) else 0,

324 )

325

326 # Try to get dimensions using PIL

327 try:

328 from PIL import Image

329

330 with Image.open(file_path) as img:

331 meta.width = img.width

332 meta.height = img.height

333 meta.color_mode = img.mode

334 meta.has_alpha = img.mode in ("RGBA", "LA", "PA")

335

336 # EXIF extraction

337 exif = img.getexif()

338 if exif:

339 for tag_id, value in exif.items():

340 meta.extra[str(tag_id)] = str(value)

341 except ImportError:

342 pass

343 except Exception:

344 pass

345

346 return meta

347

348 def _encode_base64(self, file_path: str) -> str:

349 """将图片编码为 Base64。"""

350 try:

351 with open(file_path, "rb") as f:

352 return base64.b64encode(f.read()).decode("utf-8")

353 except Exception:

354 return ""

355

356 def _generate_description(self, file_path: str) -> str:

357 """生成图片自然语言描述（应由视觉 LLM 生成）。"""

358 meta = self.extract_metadata(file_path)

359 return f"Image: {meta.width}x{meta.height}, format: {Path(file_path).suffix}"

360

361 def _generate_thumbnail(self, file_path: str) -> str:

362 """生成缩略图。"""

363 try:

364 from PIL import Image

365

366 thumb_dir = Path(tempfile.gettempdir()) / "agentos_thumbnails"

367 thumb_dir.mkdir(exist_ok=True)

368

369 thumb_name = f"thumb_{uuid.uuid4().hex[:8]}.jpg"

370 thumb_path = thumb_dir / thumb_name

371

372 with Image.open(file_path) as img:

373 img.thumbnail((self._max_size, self._max_size))

374 img.convert("RGB").save(thumb_path, self._output_format, quality=self._quality)

375

376 return str(thumb_path)

377 except Exception:

378 return ""

379

380 def resize(

381 self, file_path: str, width: int, height: int, output_path: str | None = None

382 ) -> str:

383 """缩放图片。"""

384 try:

385 from PIL import Image

386

387 out = output_path or str(

388 Path(tempfile.gettempdir())

389 / f"resized_{uuid.uuid4().hex[:8]}{Path(file_path).suffix}"

390 )

391

392 with Image.open(file_path) as img:

393 img.resize((width, height), Image.LANCZOS).save(out)

394

395 return out

396 except Exception as e:

397 raise RuntimeError(f"Image resize failed: {e}")

398

399 def compress(

400 self,

401 file_path: str,

402 quality: int = 70,

403 output_path: str | None = None,

404 ) -> str:

405 """压缩图片。"""

406 try:

407 from PIL import Image

408

409 out = output_path or str(

410 Path(tempfile.gettempdir()) / f"compressed_{uuid.uuid4().hex[:8]}.jpg"

411 )

412

413 with Image.open(file_path) as img:

414 img.convert("RGB").save(out, "JPEG", quality=quality, optimize=True)

415

416 return out

417 except Exception as e:

418 raise RuntimeError(f"Image compression failed: {e}")

419

420 def convert_format(

421 self, file_path: str, target_format: str, output_path: str | None = None

422 ) -> str:

423 """转换图片格式。"""

424 try:

425 from PIL import Image

426

427 fmt = target_format.upper().replace(".", "")

428 ext = f".{target_format.lower().lstrip('.')}"

429 out = output_path or str(

430 Path(tempfile.gettempdir()) / f"converted_{uuid.uuid4().hex[:8]}{ext}"

431 )

432

433 with Image.open(file_path) as img:

434 img.save(out, fmt)

435

436 return out

437 except Exception as e:

438 raise RuntimeError(f"Format conversion failed: {e}")

439

440

441class AudioProcessor(MediaProcessor):

442 """音频处理器。

443

444 支持转录（需 whisper）、格式转换、元数据提取。

445

446 Usage:

447 processor = AudioProcessor()

448 ctx = processor.process("recording.mp3")

449 print(ctx.extracted_text) # 转录文本

450 """

451

452 def __init__(self, transcription_model: str = "base"):

453 self._model = transcription_model

454

455 def process(self, file_path: str) -> MediaContext:

456 ctx = MediaContext(media_type=MediaType.AUDIO)

457 ctx.metadata = self.extract_metadata(file_path)

458 ctx.extracted_text = self._transcribe(file_path)

459 return ctx

460

461 def extract_metadata(self, file_path: str) -> MediaMetadata:

462 meta = MediaMetadata(

463 file_path=file_path,

464 media_type=MediaType.AUDIO,

465 mime_type=mimetypes.guess_type(file_path)[0] or "application/octet-stream",

466 file_size_bytes=os.path.getsize(file_path) if os.path.exists(file_path) else 0,

467 )

468

469 # Extract with ffprobe if available

470 try:

471 result = subprocess.run(

472 [

473 "ffprobe",

474 "-v",

475 "quiet",

476 "-print_format",

477 "json",

478 "-show_format",

479 "-show_streams",

480 file_path,

481 ],

482 capture_output=True,

483 text=True,

484 timeout=10,

485 )

486 if result.returncode == 0:

487 info = json.loads(result.stdout)

488 fmt = info.get("format", {})

489 meta.duration_s = float(fmt.get("duration", 0))

490 meta.bitrate_kbps = int(int(fmt.get("bit_rate", 0)) / 1000)

491

492 for stream in info.get("streams", []):

493 if stream.get("codec_type") == "audio":

494 meta.sample_rate = int(stream.get("sample_rate", 0))

495 meta.channels = int(stream.get("channels", 0))

496 break

497 except Exception:

498 pass

499

500 return meta

501

502 def _transcribe(self, file_path: str) -> str:

503 """音频转录。"""

504 try:

505 import whisper

506

507 model = whisper.load_model(self._model)

508 result = model.transcribe(file_path)

509 return result["text"]

510 except ImportError:

511 return "[Transcription requires: pip install openai-whisper]"

512 except Exception as e:

513 return f"[Transcription error: {e}]"

514

515

516class VideoProcessor(MediaProcessor):

517 """视频处理器。

518

519 提取关键帧、生成描述。

520

521 Usage:

522 processor = VideoProcessor()

523 ctx = processor.process("demo.mp4")

524 for caption in ctx.captions:

525 print(caption)

526 """

527

528 def __init__(self, keyframe_interval_s: float = 2.0, max_keyframes: int = 10):

529 self._keyframe_interval = keyframe_interval_s

530 self._max_keyframes = max_keyframes

531 self._image_processor = ImageProcessor()

532

533 def process(self, file_path: str) -> MediaContext:

534 ctx = MediaContext(media_type=MediaType.VIDEO)

535 ctx.metadata = self.extract_metadata(file_path)

536 ctx.captions = self._extract_keyframes(file_path)

537 return ctx

538

539 def extract_metadata(self, file_path: str) -> MediaMetadata:

540 meta = MediaMetadata(

541 file_path=file_path,

542 media_type=MediaType.VIDEO,

543 mime_type=mimetypes.guess_type(file_path)[0] or "application/octet-stream",

544 file_size_bytes=os.path.getsize(file_path) if os.path.exists(file_path) else 0,

545 )

546

547 try:

548 result = subprocess.run(

549 [

550 "ffprobe",

551 "-v",

552 "quiet",

553 "-print_format",

554 "json",

555 "-show_format",

556 "-show_streams",

557 file_path,

558 ],

559 capture_output=True,

560 text=True,

561 timeout=10,

562 )

563 if result.returncode == 0:

564 info = json.loads(result.stdout)

565 fmt = info.get("format", {})

566 meta.duration_s = float(fmt.get("duration", 0))

567 meta.bitrate_kbps = int(int(fmt.get("bit_rate", 0)) / 1000)

568

569 for stream in info.get("streams", []):

570 if stream.get("codec_type") == "video":

571 meta.width = int(stream.get("width", 0))

572 meta.height = int(stream.get("height", 0))

573 break

574 except Exception:

575 pass

576

577 return meta

578

579 def _extract_keyframes(self, file_path: str) -> list[str]:

580 """提取视频关键帧。"""

581 captions = []

582 meta = self.extract_metadata(file_path)

583 duration = meta.duration_s

584

585 if duration == 0:

586 return captions

587

588 num_frames = min(

589 int(duration / self._keyframe_interval),

590 self._max_keyframes,

591 )

592

593 thumb_dir = Path(tempfile.gettempdir()) / "agentos_video_frames"

594 thumb_dir.mkdir(exist_ok=True)

595

596 for i in range(num_frames):

597 timestamp = i * self._keyframe_interval

598 frame_path = thumb_dir / f"frame_{uuid.uuid4().hex[:8]}.jpg"

599

600 try:

601 subprocess.run(

602 [

603 "ffmpeg",

604 "-y",

605 "-loglevel",

606 "quiet",

607 "-ss",

608 str(timestamp),

609 "-i",

610 file_path,

611 "-vframes",

612 "1",

613 "-q:v",

614 "2",

615 str(frame_path),

616 ],

617 timeout=30,

618 check=True,

619 )

620

621 if frame_path.exists():

622 # Encode frame as base64

623 ctx = self._image_processor.process(str(frame_path))

624 captions.append(

625 f"[{self._format_time(timestamp)}] {ctx.text_description} "

626 f"base64:{ctx.base64_data[:50]}..."

627 )

628 # Cleanup frame file

629 frame_path.unlink(missing_ok=True)

630 except Exception:

631 pass

632

633 return captions

634

635 @staticmethod

636 def _format_time(seconds: float) -> str:

637 m, s = divmod(int(seconds), 60)

638 h, m = divmod(m, 60)

639 if h:

640 return f"{h}:{m:02d}:{s:02d}"

641 return f"{m}:{s:02d}"

642

643

644class DocumentProcessor(MediaProcessor):

645 """文档处理器。

646

647 从 PDF/DOCX 等文档中提取文本。

648

649 Usage:

650 processor = DocumentProcessor()

651 ctx = processor.process("report.pdf")

652 print(ctx.extracted_text[:500])

653 """

654

655 def process(self, file_path: str) -> MediaContext:

656 ctx = MediaContext(media_type=MediaType.DOCUMENT)

657 ctx.metadata = self.extract_metadata(file_path)

658 ctx.extracted_text = self._extract_text(file_path)

659 return ctx

660

661 def extract_metadata(self, file_path: str) -> MediaMetadata:

662 return MediaMetadata(

663 file_path=file_path,

664 media_type=MediaType.DOCUMENT,

665 mime_type=mimetypes.guess_type(file_path)[0] or "application/octet-stream",

666 file_size_bytes=os.path.getsize(file_path) if os.path.exists(file_path) else 0,

667 )

668

669 def _extract_text(self, file_path: str) -> str:

670 """提取文档文本。"""

671 ext = Path(file_path).suffix.lower()

672

673 if ext == ".pdf":

674 return self._extract_pdf(file_path)

675 elif ext in (".docx", ".doc"):

676 return self._extract_docx(file_path)

677 elif ext in (".txt", ".md", ".py", ".json", ".yaml", ".xml", ".html", ".csv"):

678 try:

679 return Path(file_path).read_text(encoding="utf-8")

680 except Exception:

681 return Path(file_path).read_text(encoding="latin-1")

682 else:

683 return f"[Unsupported document format: {ext}]"

684

685 def _extract_pdf(self, file_path: str) -> str:

686 """从 PDF 中提取文本。"""

687 try:

688 import fitz # PyMuPDF

689

690 doc = fitz.open(file_path)

691 text_parts = []

692 for page_num in range(len(doc)):

693 page = doc[page_num]

694 text = page.get_text()

695 if text.strip():

696 text_parts.append(f"--- Page {page_num + 1} ---\n{text}")

697 doc.close()

698 return "\n\n".join(text_parts) if text_parts else "[No extractable text in PDF]"

699 except ImportError:

700 try:

701 result = subprocess.run(

702 ["pdftotext", file_path, "-"],

703 capture_output=True,

704 text=True,

705 timeout=30,

706 )

707 if result.returncode == 0:

708 return result.stdout

709 except Exception:

710 pass

711 return "[PDF extraction requires: pip install PyMuPDF]"

712 except Exception as e:

713 return f"[PDF extraction error: {e}]"

714

715 def _extract_docx(self, file_path: str) -> str:

716 """从 DOCX 中提取文本。"""

717 try:

718 from docx import Document

719

720 doc = Document(file_path)

721 paragraphs = [p.text for p in doc.paragraphs if p.text.strip()]

722 return "\n\n".join(paragraphs) if paragraphs else "[No text in document]"

723 except ImportError:

724 return "[DOCX extraction requires: pip install python-docx]"

725 except Exception as e:

726 return f"[DOCX extraction error: {e}]"

727

728

729# ── Multimodal Context Manager ──────────────

730

731

732class MultimodalContextManager:

733 """多模态上下文管理器。

734

735 统一入口：接收文件路径，返回 MediaContext。

736

737 Usage:

738 mgr = MultimodalContextManager()

739 ctx = mgr.load("photo.jpg")

740 message = ctx.to_llm_message()

741 """

742

743 def __init__(self):

744 self._detector = MediaDetector()

745 self._processors: dict[MediaType, MediaProcessor] = {

746 MediaType.IMAGE: ImageProcessor(),

747 MediaType.AUDIO: AudioProcessor(),

748 MediaType.VIDEO: VideoProcessor(),

749 MediaType.DOCUMENT: DocumentProcessor(),

750 }

751

752 def load(self, file_path: str) -> MediaContext:

753 """加载并处理单个媒体文件。"""

754 mtype = self._detector.detect(file_path)

755 processor = self._processors.get(mtype)

756

757 if not processor:

758 ctx = MediaContext(media_type=MediaType.UNKNOWN)

759 ctx.metadata = MediaMetadata(file_path=file_path, media_type=MediaType.UNKNOWN)

760 ctx.extracted_text = f"[Unsupported media type: {mtype}]"

761 return ctx

762

763 return processor.process(file_path)

764

765 def load_batch(self, file_paths: list[str]) -> list[MediaContext]:

766 """批量加载。"""

767 return [self.load(fp) for fp in file_paths]

768

769 def load_as_message(self, file_path: str) -> dict:

770 """加载并转换为 LLM 消息格式。"""

771 return self.load(file_path).to_llm_message()

772

773 def load_batch_as_messages(self, file_paths: list[str]) -> list[dict]:

774 """批量加载为 LLM 消息。"""

775 return [self.load_as_message(fp) for fp in file_paths]

776

777 def register_processor(self, media_type: MediaType, processor: MediaProcessor) -> None:

778 """注册自定义处理器。"""

779 self._processors[media_type] = processor

780

781 def analyze_directory(self, directory: str) -> dict[str, list[str]]:

782 """分析目录中的媒体文件分布。"""

783 result: dict[str, list[str]] = {

784 "images": [],

785 "audio": [],

786 "video": [],

787 "documents": [],

788 "unknown": [],

789 }

790

791 dir_path = Path(directory)

792 if not dir_path.exists():

793 return result

794

795 for file_path in dir_path.rglob("*"):

796 if not file_path.is_file():

797 continue

798

799 mtype = self._detector.detect(str(file_path))

800

801 if mtype == MediaType.IMAGE:

802 result["images"].append(str(file_path))

803 elif mtype == MediaType.AUDIO:

804 result["audio"].append(str(file_path))

805 elif mtype == MediaType.VIDEO:

806 result["video"].append(str(file_path))

807 elif mtype == MediaType.DOCUMENT:

808 result["documents"].append(str(file_path))

809 else:

810 result["unknown"].append(str(file_path))

811

812 return result

813

814

815# ── Quick Start ─────────────────────────────

816

817

818def create_multimodal_manager() -> MultimodalContextManager:

819 """快速创建多模态上下文管理器。"""

820 return MultimodalContextManager()

821

822

823def quick_load(file_path: str) -> MediaContext:

824 """快速加载单个文件。"""

825 return MultimodalContextManager().load(file_path)

826

827

828# ── Compatibility aliases (required by agentos/__init__.py) ──

829

830MultimodalManager = MultimodalContextManager

831Modality = MediaType

Coverage for agentos/multimodal/init.py: 29%

373 statements