【2026年最新】マルチモーダルAIアプリケーション開発ガイド：テキスト・画像・音声の統合処理

import base64
import httpx
from openai import OpenAI

client = OpenAI()

def analyze_image(image_path: str, question: str) -> str:
    """画像を分析してテキストで回答する"""
    # 画像をBase64エンコード
    with open(image_path, "rb") as f:
        image_data = base64.standard_b64encode(f.read()).decode("utf-8")

    # MIMEタイプの判定
    mime_type = "image/png" if image_path.endswith(".png") else "image/jpeg"

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "user",
                "content": [
                    {"type": "text", "text": question},
                    {
                        "type": "image_url",
                        "image_url": {
                            "url": f"data:{mime_type};base64,{image_data}",
                            "detail": "high",  # high/low/auto
                        },
                    },
                ],
            }
        ],
        max_tokens=1024,
    )

    return response.choices[0].message.content


# 使用例
result = analyze_image(
    "screenshot.png",
    "この画面のUIに改善すべき点があれば指摘してください"
)
print(result)

import asyncio
import websockets
import json
import base64

async def realtime_voice_chat():
    """OpenAI Realtime APIを使用した音声対話"""
    url = "wss://api.openai.com/v1/realtime?model=gpt-4o-realtime"
    headers = {
        "Authorization": f"Bearer {OPENAI_API_KEY}",
        "OpenAI-Beta": "realtime=v1",
    }

    async with websockets.connect(url, extra_headers=headers) as ws:
        # セッション設定
        await ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "modalities": ["text", "audio"],
                "instructions": "あなたは親切な日本語アシスタントです。",
                "voice": "alloy",
                "input_audio_format": "pcm16",
                "output_audio_format": "pcm16",
                "input_audio_transcription": {
                    "model": "whisper-1"
                },
                "turn_detection": {
                    "type": "server_vad",
                    "threshold": 0.5,
                    "prefix_padding_ms": 300,
                    "silence_duration_ms": 500,
                },
            }
        }))

        # 音声データの送受信ループ
        async def send_audio():
            """マイクからの音声をストリーミング送信"""
            # 実際にはPyAudioなどで録音した音声データを送信
            pass

        async def receive_response():
            """モデルからの応答を受信"""
            async for message in ws:
                event = json.loads(message)
                if event["type"] == "response.audio.delta":
                    audio_chunk = base64.b64decode(event["delta"])
                    # スピーカーで再生
                    play_audio(audio_chunk)
                elif event["type"] == "response.audio_transcript.delta":
                    print(event["delta"], end="", flush=True)

        await asyncio.gather(send_audio(), receive_response())

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.schema import HumanMessage
import base64

class MultimodalRAG:
    """画像とテキストを統合したRAGシステム"""

    def __init__(self):
        self.llm = ChatOpenAI(model="gpt-4o", temperature=0)
        self.embeddings = OpenAIEmbeddings(model="text-embedding-3-large")
        self.vectorstore = Chroma(
            collection_name="multimodal_docs",
            embedding_function=self.embeddings,
        )

    def index_document(self, text: str, images: list[str], metadata: dict):
        """テキストと画像を含むドキュメントをインデックスに追加"""
        # テキストのインデックス
        self.vectorstore.add_texts(
            texts=[text],
            metadatas=[{**metadata, "type": "text"}],
        )

        # 画像の説明文を生成してインデックス
        for img_path in images:
            description = self._describe_image(img_path)
            self.vectorstore.add_texts(
                texts=[description],
                metadatas=[{**metadata, "type": "image", "image_path": img_path}],
            )

    def query(self, question: str, user_image: str | None = None) -> str:
        """マルチモーダルクエリを処理"""
        # ベクトル検索で関連ドキュメントを取得
        docs = self.vectorstore.similarity_search(question, k=5)

        # コンテキストの構築
        context_parts = []
        image_paths = []
        for doc in docs:
            context_parts.append(doc.page_content)
            if doc.metadata.get("type") == "image":
                image_paths.append(doc.metadata["image_path"])

        context = "\n\n".join(context_parts)

        # マルチモーダルプロンプトの構築
        content = [
            {"type": "text", "text": f"以下のコンテキストに基づいて回答してください。\n\nコンテキスト:\n{context}\n\n質問: {question}"},
        ]

        # ユーザーが画像を添付した場合
        if user_image:
            with open(user_image, "rb") as f:
                img_b64 = base64.standard_b64encode(f.read()).decode()
            content.append({
                "type": "image_url",
                "image_url": {"url": f"data:image/png;base64,{img_b64}"},
            })

        response = self.llm.invoke([HumanMessage(content=content)])
        return response.content

    def _describe_image(self, image_path: str) -> str:
        """画像の詳細な説明文を生成"""
        with open(image_path, "rb") as f:
            img_b64 = base64.standard_b64encode(f.read()).decode()

        response = self.llm.invoke([
            HumanMessage(content=[
                {"type": "text", "text": "この画像の内容を詳細に説明してください。"},
                {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{img_b64}"}},
            ])
        ])
        return response.content

import google.generativeai as genai

genai.configure(api_key="YOUR_API_KEY")

def analyze_video(video_path: str, question: str) -> str:
    """動画を分析して質問に回答する"""
    # 動画ファイルのアップロード
    video_file = genai.upload_file(path=video_path)

    # アップロード完了を待機
    import time
    while video_file.state.name == "PROCESSING":
        time.sleep(2)
        video_file = genai.get_file(video_file.name)

    model = genai.GenerativeModel("gemini-2.0-pro")

    response = model.generate_content(
        [video_file, question],
        generation_config=genai.GenerationConfig(
            temperature=0.2,
            max_output_tokens=2048,
        ),
    )

    return response.text

# 使用例
result = analyze_video(
    "meeting_recording.mp4",
    "この会議の主な議題と決定事項をまとめてください"
)

def process_technical_document(pdf_pages: list[str], images: list[str]) -> dict:
    """技術文書（テキスト+図表）を分析してJSON形式で構造化"""
    model = genai.GenerativeModel("gemini-2.0-pro")

    content_parts = []

    # PDFの各ページをテキストとして追加
    for page in pdf_pages:
        content_parts.append(page)

    # 関連画像を追加
    for img_path in images:
        img = genai.upload_file(path=img_path)
        content_parts.append(img)

    # 分析指示
    content_parts.append("""
    上記の技術文書と図表を分析し、以下のJSON形式で出力してください：
    {
        "title": "文書タイトル",
        "summary": "300字以内の要約",
        "key_findings": ["主要な発見1", "主要な発見2"],
        "technical_specs": {"仕様名": "仕様値"},
        "diagrams_analysis": [{"figure": "図番号", "description": "説明"}]
    }
    """)

    response = model.generate_content(
        content_parts,
        generation_config=genai.GenerationConfig(
            temperature=0,
            response_mime_type="application/json",
        ),
    )

    import json
    return json.loads(response.text)

from PIL import Image
import io

def optimize_image_for_api(image_path: str, max_size: int = 2048) -> bytes:
    """API送信用に画像を最適化"""
    img = Image.open(image_path)

    # EXIF情報に基づく回転の適用
    from PIL import ImageOps
    img = ImageOps.exif_transpose(img)

    # アスペクト比を維持してリサイズ
    if max(img.size) > max_size:
        ratio = max_size / max(img.size)
        new_size = (int(img.size[0] * ratio), int(img.size[1] * ratio))
        img = img.resize(new_size, Image.LANCZOS)

    # JPEG形式で品質85%に圧縮
    buffer = io.BytesIO()
    img.save(buffer, format="JPEG", quality=85, optimize=True)
    return buffer.getvalue()

import asyncio
from typing import Optional

class MultimodalService:
    """フォールバック付きマルチモーダルサービス"""

    def __init__(self):
        self.primary_model = "gpt-4o"
        self.fallback_model = "gpt-4o-mini"
        self.max_retries = 3

    async def process(
        self, text: str, image: Optional[bytes] = None, audio: Optional[bytes] = None
    ) -> str:
        for attempt in range(self.max_retries):
            try:
                model = self.primary_model if attempt == 0 else self.fallback_model
                return await self._call_model(model, text, image, audio)
            except RateLimitError:
                await asyncio.sleep(2 ** attempt)
            except ContentFilterError as e:
                return f"コンテンツポリシーにより処理できません: {e}"
            except InvalidImageError:
                # 画像なしでテキストのみのフォールバック
                return await self._call_model(self.fallback_model, text, None, None)

        raise ServiceUnavailableError("All retry attempts failed")