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"RIFF": (MediaType.AUDIO, None), # WAV is RIFF

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

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

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

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

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

184 # Video

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

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

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

188 # Documents

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

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

191 }

192

193 # Audio extensions

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

195

196 # Video extensions

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

198

199 # Image extensions

200 IMAGE_EXTENSIONS = {

201 ".jpg",

202 ".jpeg",

203 ".png",

204 ".gif",

205 ".webp",

206 ".bmp",

207 ".svg",

208 ".tiff",

209 ".heic",

210 ".ico",

211 }

212

213 # Document extensions

214 DOCUMENT_EXTENSIONS = {

215 ".pdf",

216 ".docx",

217 ".doc",

218 ".xlsx",

219 ".xls",

220 ".pptx",

221 ".ppt",

222 ".txt",

223 ".md",

224 ".html",

225 ".epub",

226 }

227

228 @classmethod

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

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

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

232

233 if ext in cls.IMAGE_EXTENSIONS:

234 return MediaType.IMAGE

235 if ext in cls.AUDIO_EXTENSIONS:

236 return MediaType.AUDIO

237 if ext in cls.VIDEO_EXTENSIONS:

238 return MediaType.VIDEO

239 if ext in cls.DOCUMENT_EXTENSIONS:

240 return MediaType.DOCUMENT

241

242 # Fallback to magic bytes

243 try:

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

245 header = f.read(32)

246 except Exception:

247 return MediaType.UNKNOWN

248

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

250 if header.startswith(magic):

251 # RIFF ambiguity resolution

252 if magic == b"RIFF":

253 if b"WEBP" in header:

254 return MediaType.IMAGE

255 if b"WAVE" in header:

256 return MediaType.AUDIO

257 return mtype

258

259 # MIME type fallback

260 mime, _ = mimetypes.guess_type(file_path)

261 if mime:

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

263 return MediaType.IMAGE

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

265 return MediaType.AUDIO

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

267 return MediaType.VIDEO

268

269 return MediaType.UNKNOWN

270

271 @classmethod

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

273 """批量检测。"""

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

275

276

277# ── Media Processors ────────────────────────

278

279

280class MediaProcessor(ABC):

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

282

283 @abstractmethod

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

285

286 @abstractmethod

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

288

289

290class ImageProcessor(MediaProcessor):

291 """图像处理器。

292

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

294

295 Usage:

296 processor = ImageProcessor()

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

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

299 """

300

301 def __init__(

302 self,

303 max_size: int = 2048,

304 quality: int = 85,

305 output_format: str = "JPEG",

306 ):

307 self._max_size = max_size

308 self._quality = quality

309 self._output_format = output_format

310

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

312 ctx = MediaContext(media_type=MediaType.IMAGE)

313 ctx.metadata = self.extract_metadata(file_path)

314 ctx.base64_data = self._encode_base64(file_path)

315 ctx.text_description = self._generate_description(file_path)

316 ctx.thumbnail_path = self._generate_thumbnail(file_path)

317 return ctx

318

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

320 meta = MediaMetadata(

321 file_path=file_path,

322 media_type=MediaType.IMAGE,

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

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

325 )

326

327 # Try to get dimensions using PIL

328 try:

329 from PIL import Image

330

331 with Image.open(file_path) as img:

332 meta.width = img.width

333 meta.height = img.height

334 meta.color_mode = img.mode

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

336

337 # EXIF extraction

338 exif = img.getexif()

339 if exif:

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

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

342 except ImportError:

343 pass

344 except Exception:

345 pass

346

347 return meta

348

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

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

351 try:

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

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

354 except Exception:

355 return ""

356

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

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

359 meta = self.extract_metadata(file_path)

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

361

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

363 """生成缩略图。"""

364 try:

365 from PIL import Image

366

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

368 thumb_dir.mkdir(exist_ok=True)

369

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

371 thumb_path = thumb_dir / thumb_name

372

373 with Image.open(file_path) as img:

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

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

376

377 return str(thumb_path)

378 except Exception:

379 return ""

380

381 def resize(

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

383 ) -> str:

384 """缩放图片。"""

385 try:

386 from PIL import Image

387

388 out = output_path or str(

389 Path(tempfile.gettempdir())

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

391 )

392

393 with Image.open(file_path) as img:

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

395

396 return out

397 except Exception as e:

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

399

400 def compress(

401 self,

402 file_path: str,

403 quality: int = 70,

404 output_path: str | None = None,

405 ) -> str:

406 """压缩图片。"""

407 try:

408 from PIL import Image

409

410 out = output_path or str(

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

412 )

413

414 with Image.open(file_path) as img:

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

416

417 return out

418 except Exception as e:

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

420

421 def convert_format(

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

423 ) -> str:

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

425 try:

426 from PIL import Image

427

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

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

430 out = output_path or str(

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

432 )

433

434 with Image.open(file_path) as img:

435 img.save(out, fmt)

436

437 return out

438 except Exception as e:

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

440

441

442class AudioProcessor(MediaProcessor):

443 """音频处理器。

444

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

446

447 Usage:

448 processor = AudioProcessor()

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

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

451 """

452

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

454 self._model = transcription_model

455

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

457 ctx = MediaContext(media_type=MediaType.AUDIO)

458 ctx.metadata = self.extract_metadata(file_path)

459 ctx.extracted_text = self._transcribe(file_path)

460 return ctx

461

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

463 meta = MediaMetadata(

464 file_path=file_path,

465 media_type=MediaType.AUDIO,

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

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

468 )

469

470 # Extract with ffprobe if available

471 try:

472 result = subprocess.run(

473 [

474 "ffprobe",

475 "-v",

476 "quiet",

477 "-print_format",

478 "json",

479 "-show_format",

480 "-show_streams",

481 file_path,

482 ],

483 capture_output=True,

484 text=True,

485 timeout=10,

486 )

487 if result.returncode == 0:

488 info = json.loads(result.stdout)

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

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

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

492

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

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

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

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

497 break

498 except Exception:

499 pass

500

501 return meta

502

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

504 """音频转录。"""

505 try:

506 import whisper

507

508 model = whisper.load_model(self._model)

509 result = model.transcribe(file_path)

510 return result["text"]

511 except ImportError:

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

513 except Exception as e:

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

515

516

517class VideoProcessor(MediaProcessor):

518 """视频处理器。

519

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

521

522 Usage:

523 processor = VideoProcessor()

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

525 for caption in ctx.captions:

526 print(caption)

527 """

528

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

530 self._keyframe_interval = keyframe_interval_s

531 self._max_keyframes = max_keyframes

532 self._image_processor = ImageProcessor()

533

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

535 ctx = MediaContext(media_type=MediaType.VIDEO)

536 ctx.metadata = self.extract_metadata(file_path)

537 ctx.captions = self._extract_keyframes(file_path)

538 return ctx

539

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

541 meta = MediaMetadata(

542 file_path=file_path,

543 media_type=MediaType.VIDEO,

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

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

546 )

547

548 try:

549 result = subprocess.run(

550 [

551 "ffprobe",

552 "-v",

553 "quiet",

554 "-print_format",

555 "json",

556 "-show_format",

557 "-show_streams",

558 file_path,

559 ],

560 capture_output=True,

561 text=True,

562 timeout=10,

563 )

564 if result.returncode == 0:

565 info = json.loads(result.stdout)

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

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

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

569

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

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

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

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

574 break

575 except Exception:

576 pass

577

578 return meta

579

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

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

582 captions = []

583 meta = self.extract_metadata(file_path)

584 duration = meta.duration_s

585

586 if duration == 0:

587 return captions

588

589 num_frames = min(

590 int(duration / self._keyframe_interval),

591 self._max_keyframes,

592 )

593

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

595 thumb_dir.mkdir(exist_ok=True)

596

597 for i in range(num_frames):

598 timestamp = i * self._keyframe_interval

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

600

601 try:

602 subprocess.run(

603 [

604 "ffmpeg",

605 "-y",

606 "-loglevel",

607 "quiet",

608 "-ss",

609 str(timestamp),

610 "-i",

611 file_path,

612 "-vframes",

613 "1",

614 "-q:v",

615 "2",

616 str(frame_path),

617 ],

618 timeout=30,

619 check=True,

620 )

621

622 if frame_path.exists():

623 # Encode frame as base64

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

625 captions.append(

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

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

628 )

629 # Cleanup frame file

630 frame_path.unlink(missing_ok=True)

631 except Exception:

632 pass

633

634 return captions

635

636 @staticmethod

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

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

639 h, m = divmod(m, 60)

640 if h:

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

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

643

644

645class DocumentProcessor(MediaProcessor):

646 """文档处理器。

647

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

649

650 Usage:

651 processor = DocumentProcessor()

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

653 print(ctx.extracted_text[:500])

654 """

655

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

657 ctx = MediaContext(media_type=MediaType.DOCUMENT)

658 ctx.metadata = self.extract_metadata(file_path)

659 ctx.extracted_text = self._extract_text(file_path)

660 return ctx

661

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

663 return MediaMetadata(

664 file_path=file_path,

665 media_type=MediaType.DOCUMENT,

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

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

668 )

669

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

671 """提取文档文本。"""

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

673

674 if ext == ".pdf":

675 return self._extract_pdf(file_path)

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

677 return self._extract_docx(file_path)

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

679 try:

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

681 except Exception:

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

683 else:

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

685

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

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

688 try:

689 import fitz # PyMuPDF

690

691 doc = fitz.open(file_path)

692 text_parts = []

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

694 page = doc[page_num]

695 text = page.get_text()

696 if text.strip():

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

698 doc.close()

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

700 except ImportError:

701 try:

702 result = subprocess.run(

703 ["pdftotext", file_path, "-"],

704 capture_output=True,

705 text=True,

706 timeout=30,

707 )

708 if result.returncode == 0:

709 return result.stdout

710 except Exception:

711 pass

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

713 except Exception as e:

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

715

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

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

718 try:

719 from docx import Document

720

721 doc = Document(file_path)

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

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

724 except ImportError:

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

726 except Exception as e:

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

728

729

730# ── Multimodal Context Manager ──────────────

731

732

733class MultimodalContextManager:

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

735

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

737

738 Usage:

739 mgr = MultimodalContextManager()

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

741 message = ctx.to_llm_message()

742 """

743

744 def __init__(self):

745 self._detector = MediaDetector()

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

747 MediaType.IMAGE: ImageProcessor(),

748 MediaType.AUDIO: AudioProcessor(),

749 MediaType.VIDEO: VideoProcessor(),

750 MediaType.DOCUMENT: DocumentProcessor(),

751 }

752

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

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

755 mtype = self._detector.detect(file_path)

756 processor = self._processors.get(mtype)

757

758 if not processor:

759 ctx = MediaContext(media_type=MediaType.UNKNOWN)

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

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

762 return ctx

763

764 return processor.process(file_path)

765

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

767 """批量加载。"""

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

769

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

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

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

773

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

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

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

777

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

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

780 self._processors[media_type] = processor

781

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

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

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

785 "images": [],

786 "audio": [],

787 "video": [],

788 "documents": [],

789 "unknown": [],

790 }

791

792 dir_path = Path(directory)

793 if not dir_path.exists():

794 return result

795

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

797 if not file_path.is_file():

798 continue

799

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

801

802 if mtype == MediaType.IMAGE:

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

804 elif mtype == MediaType.AUDIO:

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

806 elif mtype == MediaType.VIDEO:

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

808 elif mtype == MediaType.DOCUMENT:

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

810 else:

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

812

813 return result

814

815

816# ── Quick Start ─────────────────────────────

817

818

819def create_multimodal_manager() -> MultimodalContextManager:

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

821 return MultimodalContextManager()

822

823

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

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

826 return MultimodalContextManager().load(file_path)

827

828

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

830

831MultimodalManager = MultimodalContextManager

832Modality = MediaType

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

373 statements