# 核心依賴
pip install openai>=1.30.0
pip install pillow>=10.0.0
pip install pydub>=0.25.1
pip install requests>=2.31.0
# 語音處理
pip install faster-whisper>=1.0.0
pip install pyttsx3>=2.90
# 圖像處理
pip install opencv-python>=4.9.0
pip install ultralytics>=8.2.0

multimodal_app/
├── config.py          # 配置文件
├── text_handler.py    # 文本處理模塊
├── vision_handler.py  # 圖像處理模塊
├── audio_handler.py   # 音頻處理模塊
├── multimodal.py      # 多模態(tài)融合核心
└── main.py            # 主入口

import base64
from pathlib import Path
from openai import OpenAI

client = OpenAI(api_key="your-api-key")


def encode_image(image_path: str) -> str:
    """將圖片編碼為base64字符串"""
    with open(image_path, "rb") as f:
        return base64.b64encode(f.read()).decode("utf-8")


def analyze_image(image_path: str, question: str) -> str:
    """
    多模態(tài)圖像理解：結(jié)合圖片內(nèi)容回答問題

    Args:
        image_path: 圖片路徑
        question: 要提問的問題

    Returns:
        模型的回答文本
    """
    base64_image = encode_image(image_path)

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": "你是一個(gè)專業(yè)的圖像分析助手，請(qǐng)?jiān)敿?xì)描述圖片內(nèi)容并回答用戶問題。",
            },
            {
                "role": "user",
                "content": [
                    {"type": "text", "text": question},
                    {
                        "type": "image_url",
                        "image_url": {
                            "url": f"data:image/jpeg;base64,{base64_image}",
                            "detail": "high",
                        },
                    },
                ],
            },
        ],
        max_tokens=1000,
    )
    return response.choices[0].message.content


# 使用示例
result = analyze_image(
    "dashboard.png",
    "請(qǐng)分析這張數(shù)據(jù)看板截圖，告訴我哪個(gè)指標(biāo)表現(xiàn)最好，哪個(gè)需要關(guān)注。",
)
print(result)

from dataclasses import dataclass
from typing import Optional


@dataclass
class ImageAnalysisResult:
    """圖像分析結(jié)果"""
    image_path: str
    description: str
    labels: list[str]
    confidence: float


class MultiImageAnalyzer:
    """批量多圖分析器"""

    def __init__(self, model: str = "gpt-4o"):
        self.client = OpenAI()
        self.model = model

    def analyze_single(
        self, image_path: str, prompt: str
    ) -> ImageAnalysisResult:
        """分析單張圖片"""
        base64_img = encode_image(image_path)

        response = self.client.chat.completions.create(
            model=self.model,
            messages=[
                {
                    "role": "user",
                    "content": [
                        {"type": "text", "text": prompt},
                        {
                            "type": "image_url",
                            "image_url": {
                                "url": f"data:image/jpeg;base64,{base64_img}",
                            },
                        },
                    ],
                }
            ],
            response_format={"type": "json_object"},
        )

        import json
        data = json.loads(response.choices[0].message.content)

        return ImageAnalysisResult(
            image_path=image_path,
            description=data.get("description", ""),
            labels=data.get("labels", []),
            confidence=data.get("confidence", 0.0),
        )

    def analyze_batch(
        self, image_paths: list[str], prompt: str
    ) -> list[ImageAnalysisResult]:
        """批量分析多張圖片"""
        results = []
        for path in image_paths:
            result = self.analyze_single(path, prompt)
            results.append(result)
            print(f"已分析: {path} -> {result.labels}")
        return results


# 使用示例
analyzer = MultiImageAnalyzer()
results = analyzer.analyze_batch(
    ["img1.jpg", "img2.jpg", "img3.jpg"],
    "請(qǐng)分析圖片內(nèi)容，返回JSON格式：{description, labels, confidence}",
)
for r in results:
    print(f"{r.image_path}: {r.description[:50]}... (置信度: {r.confidence})")

from faster_whisper import WhisperModel
from pathlib import Path


class AudioProcessor:
    """音頻處理器：語音轉(zhuǎn)文字"""

    def __init__(self, model_size: str = "large-v3"):
        # 使用GPU加速（device="cuda"），無GPU則用"cpu"
        self.model = WhisperModel(
            model_size,
            device="cpu",
            compute_type="int8",
        )

    def transcribe(
        self, audio_path: str, language: str = "zh"
    ) -> dict:
        """
        語音轉(zhuǎn)文字

        Args:
            audio_path: 音頻文件路徑（支持mp3/wav/m4a等）
            language: 語言代碼，zh=中文, en=英文

        Returns:
            包含完整文本和分段信息的字典
        """
        segments, info = self.model.transcribe(
            audio_path,
            language=language,
            beam_size=5,
            vad_filter=True,           # 啟用VAD過濾靜音
            vad_parameters=dict(
                min_silence_duration_ms=500,
            ),
        )

        print(f"檢測(cè)到語言: {info.language} (概率: {info.language_probability:.2%})")

        transcript_segments = []
        full_text_parts = []

        for segment in segments:
            entry = {
                "start": round(segment.start, 2),
                "end": round(segment.end, 2),
                "text": segment.text.strip(),
            }
            transcript_segments.append(entry)
            full_text_parts.append(segment.text.strip())

        return {
            "full_text": "".join(full_text_parts),
            "segments": transcript_segments,
            "language": info.language,
            "duration": round(info.duration, 2),
        }


# 使用示例
processor = AudioProcessor()
result = processor.transcribe("meeting_recording.mp3", language="zh")
print(f"時(shí)長(zhǎng): {result['duration']}s")
print(f"全文: {result['full_text']}")
for seg in result["segments"]:
    print(f"[{seg['start']}s - {seg['end']}s] {seg['text']}")

from openai import OpenAI


class VoiceQA:
    """語音問答：聽懂問題，生成回答"""

    def __init__(self):
        self.client = OpenAI()
        self.audio_processor = AudioProcessor()

    def ask(self, audio_path: str, context: str = "") -> str:
        """
        聽取語音問題并回答

        Args:
            audio_path: 語音問題文件路徑
            context: 可選的上下文信息

        Returns:
            AI的回答文本
        """
        # Step 1: 語音轉(zhuǎn)文字
        transcript = self.audio_processor.transcribe(audio_path)
        question = transcript["full_text"]
        print(f"識(shí)別到的問題: {question}")

        # Step 2: 基于文本生成回答
        messages = [
            {
                "role": "system",
                "content": "你是一個(gè)智能助手，請(qǐng)根據(jù)用戶的問題給出準(zhǔn)確、詳細(xì)的回答。",
            },
        ]
        if context:
            messages.append(
                {"role": "system", "content": f"參考上下文：{context}"}
            )
        messages.append({"role": "user", "content": question})

        response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            max_tokens=1000,
        )
        return response.choices[0].message.content


# 使用示例
qa = VoiceQA()
answer = qa.ask(
    "question.mp3",
    context="這是一個(gè)關(guān)于Python多模態(tài)AI開發(fā)的技術(shù)討論會(huì)。",
)
print(f"回答: {answer}")

import json
from dataclasses import dataclass
from pathlib import Path
from openai import OpenAI
from typing import Optional


@dataclass
class MultimodalInput:
    """多模態(tài)輸入數(shù)據(jù)"""
    text: Optional[str] = None
    image_path: Optional[str] = None
    audio_path: Optional[str] = None


class MultimodalAssistant:
    """
    多模態(tài)智能助手
    同時(shí)處理文字、圖片、語音三種輸入
    """

    def __init__(self, api_key: str = None):
        self.client = OpenAI(api_key=api_key)
        self.audio_processor = AudioProcessor()
        self.conversation_history: list[dict] = [
            {
                "role": "system",
                "content": (
                    "你是一個(gè)專業(yè)的多模態(tài)AI助手。"
                    "你可以同時(shí)理解文字描述、圖片內(nèi)容和語音信息。"
                    "請(qǐng)綜合所有輸入信息給出準(zhǔn)確、有幫助的回答。"
                ),
            }
        ]

    def _build_content(self, user_input: MultimodalInput) -> list[dict]:
        """構(gòu)建多模態(tài)消息內(nèi)容"""
        content = []

        # 處理文本
        if user_input.text:
            content.append({"type": "text", "text": user_input.text})

        # 處理語音（先轉(zhuǎn)文字）
        if user_input.audio_path:
            transcript = self.audio_processor.transcribe(user_input.audio_path)
            voice_text = f"[語音轉(zhuǎn)文字] {transcript['full_text']}"
            content.append({"type": "text", "text": voice_text})

        # 處理圖像
        if user_input.image_path:
            base64_img = encode_image(user_input.image_path)
            content.append({
                "type": "image_url",
                "image_url": {
                    "url": f"data:image/jpeg;base64,{base64_img}",
                    "detail": "auto",
                },
            })

        return content

    def chat(self, user_input: MultimodalInput) -> str:
        """
        多模態(tài)對(duì)話

        Args:
            user_input: 多模態(tài)輸入，可包含文字/圖片/語音

        Returns:
            AI助手回答
        """
        content = self._build_content(user_input)

        self.conversation_history.append({
            "role": "user",
            "content": content,
        })

        response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=self.conversation_history,
            max_tokens=1500,
            temperature=0.7,
        )

        answer = response.choices[0].message.content
        self.conversation_history.append({
            "role": "assistant",
            "content": answer,
        })
        return answer

    def reset(self):
        """重置對(duì)話歷史"""
        self.conversation_history = [self.conversation_history[0]]


# ===== 使用示例 =====
assistant = MultimodalAssistant()

# 場(chǎng)景1: 純文本問答
answer = assistant.chat(MultimodalInput(text="什么是多模態(tài)AI？"))
print(f"文本問答: {answer}\n")

# 場(chǎng)景2: 圖片 + 文字
answer = assistant.chat(MultimodalInput(
    text="這張圖片里有什么問題？請(qǐng)幫我分析一下。",
    image_path="screenshot.png",
))
print(f"圖文分析: {answer}\n")

# 場(chǎng)景3: 語音 + 圖片 + 文字（全模態(tài)）
answer = assistant.chat(MultimodalInput(
    text="結(jié)合我發(fā)的語音和圖片，給我一個(gè)完整的解決方案。",
    image_path="error_log.png",
    audio_path="bug_description.mp3",
))
print(f"多模態(tài)回答: {answer}\n")

from PIL import Image
import io


def compress_image(
    image_path: str,
    max_size: tuple[int, int] = (1024, 1024),
    quality: int = 85,
) -> str:
    """
    壓縮圖片以減少API調(diào)用開銷

    Args:
        image_path: 原圖路徑
        max_size: 最大尺寸（寬, 高）
        quality: JPEG壓縮質(zhì)量 (1-100)

    Returns:
        壓縮后的base64字符串
    """
    img = Image.open(image_path)

    # 保持比例縮放
    img.thumbnail(max_size, Image.Resampling.LANCZOS)

    # 轉(zhuǎn)為RGB（處理RGBA/P模式圖片）
    if img.mode in ("RGBA", "P"):
        img = img.convert("RGB")

    # 壓縮為JPEG
    buffer = io.BytesIO()
    img.save(buffer, format="JPEG", quality=quality)
    return base64.b64encode(buffer.getvalue()).decode("utf-8")

from pydub import AudioSegment


def preprocess_audio(
    audio_path: str,
    target_format: str = "wav",
    sample_rate: int = 16000,
) -> str:
    """
    音頻預(yù)處理：統(tǒng)一格式和采樣率

    Args:
        audio_path: 原始音頻路徑
        target_format: 目標(biāo)格式
        sample_rate: 目標(biāo)采樣率

    Returns:
        處理后的音頻文件路徑
    """
    audio = AudioSegment.from_file(audio_path)

    # 轉(zhuǎn)為單聲道
    audio = audio.set_channels(1)

    # 設(shè)置采樣率
    audio = audio.set_frame_rate(sample_rate)

    # 導(dǎo)出
    output_path = Path(audio_path).stem + "_processed." + target_format
    audio.export(output_path, format=target_format)
    return output_path

import asyncio
from openai import AsyncOpenAI


async_client = AsyncOpenAI()


async def process_single(
    image_path: str, prompt: str
) -> dict:
    """異步處理單張圖片"""
    base64_img = encode_image(image_path)
    response = await async_client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "user",
                "content": [
                    {"type": "text", "text": prompt},
                    {
                        "type": "image_url",
                        "image_url": {
                            "url": f"data:image/jpeg;base64,{base64_img}",
                        },
                    },
                ],
            }
        ],
        max_tokens=500,
    )
    return {
        "image": image_path,
        "result": response.choices[0].message.content,
    }


async def process_batch(
    image_paths: list[str], prompt: str, concurrency: int = 5
) -> list[dict]:
    """
    異步批量處理圖片

    Args:
        image_paths: 圖片路徑列表
        prompt: 統(tǒng)一提示詞
        concurrency: 最大并發(fā)數(shù)

    Returns:
        處理結(jié)果列表
    """
    semaphore = asyncio.Semaphore(concurrency)

    async def limited_process(path: str):
        async with semaphore:
            return await process_single(path, prompt)

    tasks = [limited_process(p) for p in image_paths]
    return await asyncio.gather(*tasks)


# 運(yùn)行批量處理
results = asyncio.run(process_batch(
    ["img1.jpg", "img2.jpg", "img3.jpg", "img4.jpg", "img5.jpg"],
    "請(qǐng)描述這張圖片的主要內(nèi)容，用一句話概括。",
))
for r in results:
    print(f"{r['image']}: {r['result']}")