Реализация Multimodal AI обработка текста изображений аудио

Реализация Multimodal AI: обработка текста, изображений и аудио

Мультимодальные системы объединяют несколько типов данных в единый pipeline. Реальные задачи: анализ медицинских снимков с голосовым вводом врача, обработка накладных (сканы + распознавание), мониторинг производства (видео + сенсорные данные). Ключевые компоненты: Vision LLM, Whisper/Speech-to-Text, Text-to-Speech, Document Understanding.

Архитектура мультимодального pipeline

from anthropic import AsyncAnthropic
from openai import AsyncOpenAI
import base64
import aiohttp
from pathlib import Path
from typing import Union
from pydantic import BaseModel

anthropic_client = AsyncAnthropic()
openai_client = AsyncOpenAI()

class MultimodalInput(BaseModel):
    text: str | None = None
    image_paths: list[str] = []
    audio_path: str | None = None
    document_path: str | None = None

class MultimodalPipeline:

    async def process(self, input_data: MultimodalInput, task: str) -> str:
        """Обрабатывает мультимодальный ввод"""
        content_blocks = []

        # 1. Аудио → текст
        if input_data.audio_path:
            transcript = await self.transcribe_audio(input_data.audio_path)
            content_blocks.append({"type": "text", "text": f"[Аудио транскрипция]: {transcript}"})

        # 2. Изображения
        for image_path in input_data.image_paths:
            image_block = await self.prepare_image(image_path)
            content_blocks.append(image_block)

        # 3. Документы (PDF)
        if input_data.document_path:
            doc_text = await self.extract_document(input_data.document_path)
            content_blocks.append({"type": "text", "text": f"[Документ]:\n{doc_text}"})

        # 4. Текстовый запрос
        if input_data.text:
            content_blocks.append({"type": "text", "text": input_data.text})

        content_blocks.append({"type": "text", "text": f"\nЗадача: {task}"})

        response = await anthropic_client.messages.create(
            model="claude-sonnet-4-5",
            max_tokens=4096,
            messages=[{"role": "user", "content": content_blocks}],
        )

        return response.content[0].text

    async def transcribe_audio(self, audio_path: str) -> str:
        """Транскрипция аудио через Whisper"""
        with open(audio_path, "rb") as f:
            transcription = await openai_client.audio.transcriptions.create(
                file=(Path(audio_path).name, f.read()),
                model="whisper-1",
                language="ru",
            )
        return transcription.text

    async def prepare_image(self, image_path: str) -> dict:
        """Подготавливает изображение для Claude Vision"""
        if image_path.startswith("http"):
            # URL изображения
            async with aiohttp.ClientSession() as session:
                async with session.get(image_path) as resp:
                    image_data = base64.b64encode(await resp.read()).decode()
            media_type = resp.content_type or "image/jpeg"
        else:
            # Локальный файл
            with open(image_path, "rb") as f:
                image_data = base64.b64encode(f.read()).decode()
            ext = Path(image_path).suffix.lower()
            media_type = {"jpg": "image/jpeg", "jpeg": "image/jpeg",
                         "png": "image/png", "gif": "image/gif",
                         "webp": "image/webp"}.get(ext[1:], "image/jpeg")

        return {
            "type": "image",
            "source": {"type": "base64", "media_type": media_type, "data": image_data}
        }

    async def extract_document(self, pdf_path: str) -> str:
        """Извлекает текст из PDF документа"""
        import pdfplumber

        text_parts = []
        with pdfplumber.open(pdf_path) as pdf:
            for page in pdf.pages:
                text = page.extract_text()
                if text:
                    text_parts.append(text)

        return "\n\n".join(text_parts)[:10000]

Практический кейс: система обработки накладных

class InvoiceProcessor:
    """Обрабатывает сканы накладных: изображение + извлечение данных"""

    def __init__(self):
        self.pipeline = MultimodalPipeline()

    async def process_invoice_scan(self, image_path: str) -> dict:
        """Извлекает данные из скана накладной"""

        input_data = MultimodalInput(
            image_paths=[image_path],
            text="Извлеки данные накладной"
        )

        response = await anthropic_client.messages.create(
            model="claude-sonnet-4-5",
            max_tokens=2048,
            messages=[{
                "role": "user",
                "content": [
                    await self.pipeline.prepare_image(image_path),
                    {
                        "type": "text",
                        "text": """Извлеки данные из накладной. Верни JSON:
{
  "vendor": "поставщик",
  "invoice_number": "номер",
  "date": "дата",
  "items": [{"name": "...", "qty": 0, "price": 0}],
  "total": 0,
  "vat": 0,
  "currency": "RUB"
}"""
                    }
                ]
            }],
        )

        import json
        text = response.content[0].text
        return json.loads(text[text.find("{"):text.rfind("}") + 1])

Speech-to-Text + AI + Text-to-Speech pipeline

async def voice_assistant_pipeline(audio_input: bytes) -> bytes:
    """Голосовой ввод → AI ответ → голосовой вывод"""

    # 1. Speech-to-Text
    transcription = await openai_client.audio.transcriptions.create(
        file=("audio.webm", audio_input),
        model="whisper-1",
        language="ru",
    )
    user_text = transcription.text

    # 2. LLM обработка
    response = await anthropic_client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=512,
        messages=[{"role": "user", "content": user_text}]
    )
    answer_text = response.content[0].text

    # 3. Text-to-Speech
    tts_response = await openai_client.audio.speech.create(
        model="tts-1",
        voice="alloy",
        input=answer_text,
        response_format="mp3",
    )

    return tts_response.content

Анализ видео через покадровую обработку

import cv2

async def analyze_video(video_path: str, task: str, sample_fps: int = 1) -> str:
    """Анализирует видео, семплируя кадры"""
    cap = cv2.VideoCapture(video_path)
    fps = cap.get(cv2.CAP_PROP_FPS)
    frame_interval = int(fps / sample_fps)

    frames = []
    frame_count = 0

    while cap.isOpened() and len(frames) < 10:  # Максимум 10 кадров
        ret, frame = cap.read()
        if not ret:
            break

        if frame_count % frame_interval == 0:
            _, buffer = cv2.imencode(".jpg", frame, [cv2.IMWRITE_JPEG_QUALITY, 60])
            frames.append(base64.b64encode(buffer).decode())

        frame_count += 1

    cap.release()

    content = [{"type": "text", "text": f"Проанализируй видео ({len(frames)} кадров):"}]
    for i, frame_data in enumerate(frames):
        content.append({
            "type": "image",
            "source": {"type": "base64", "media_type": "image/jpeg", "data": frame_data}
        })
    content.append({"type": "text", "text": task})

    response = await anthropic_client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=2048,
        messages=[{"role": "user", "content": content}],
    )
    return response.content[0].text

Практический кейс: контроль качества на производстве

Задача: завод по производству электроники, 100+ продуктов в час, ручной осмотр = 15% пропущенных дефектов.

Система:

• Камера делает снимок каждого продукта
• Claude Vision анализирует на наличие дефектов (царапины, неправильная пайка)
• Звуковой сигнал при обнаружении дефекта
• Статистика по типам дефектов за смену

Результаты:

• Обнаружение дефектов: 15% пропуска → 3% пропуска
• Скорость обработки: 2 сек/продукт (vs 30 сек ручной осмотр)
• Экономия: -4 оператора ОТК на линии

Сроки

• Vision анализ изображений: 2–3 дня
• Whisper транскрипция: 1–2 дня
• Голосовой pipeline (STT + TTS): 1 неделя
• Видео анализ: 1–2 недели
• Production-система с очередью: 2–3 недели