Python构建多模态AI应用的开发指南

# 核心依赖
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      # 多模态融合核心
└── 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:
    """
    多模态图像理解：结合图片内容回答问题

    Args:
        image_path: 图片路径
        question: 要提问的问题

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

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": "你是一个专业的图像分析助手，请详细描述图片内容并回答用户问题。",
            },
            {
                "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",
    "请分析这张数据看板截图，告诉我哪个指标表现最好，哪个需要关注。",
)
print(result)

from dataclasses import dataclass
from typing import Optional


@dataclass
class ImageAnalysisResult:
    """图像分析结果"""
    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"],
    "请分析图片内容，返回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:
    """音频处理器：语音转文字"""

    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:
        """
        语音转文字

        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"检测到语言: {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"时长: {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: 语音转文字
        transcript = self.audio_processor.transcribe(audio_path)
        question = transcript["full_text"]
        print(f"识别到的问题: {question}")

        # Step 2: 基于文本生成回答
        messages = [
            {
                "role": "system",
                "content": "你是一个智能助手，请根据用户的问题给出准确、详细的回答。",
            },
        ]
        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="这是一个关于Python多模态AI开发的技术讨论会。",
)
print(f"回答: {answer}")

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


@dataclass
class MultimodalInput:
    """多模态输入数据"""
    text: Optional[str] = None
    image_path: Optional[str] = None
    audio_path: Optional[str] = None


class MultimodalAssistant:
    """
    多模态智能助手
    同时处理文字、图片、语音三种输入
    """

    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": (
                    "你是一个专业的多模态AI助手。"
                    "你可以同时理解文字描述、图片内容和语音信息。"
                    "请综合所有输入信息给出准确、有帮助的回答。"
                ),
            }
        ]

    def _build_content(self, user_input: MultimodalInput) -> list[dict]:
        """构建多模态消息内容"""
        content = []

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

        # 处理语音（先转文字）
        if user_input.audio_path:
            transcript = self.audio_processor.transcribe(user_input.audio_path)
            voice_text = f"[语音转文字] {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:
        """
        多模态对话

        Args:
            user_input: 多模态输入，可包含文字/图片/语音

        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):
        """重置对话历史"""
        self.conversation_history = [self.conversation_history[0]]


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

# 场景1: 纯文本问答
answer = assistant.chat(MultimodalInput(text="什么是多模态AI？"))
print(f"文本问答: {answer}\n")

# 场景2: 图片 + 文字
answer = assistant.chat(MultimodalInput(
    text="这张图片里有什么问题？请帮我分析一下。",
    image_path="screenshot.png",
))
print(f"图文分析: {answer}\n")

# 场景3: 语音 + 图片 + 文字（全模态）
answer = assistant.chat(MultimodalInput(
    text="结合我发的语音和图片，给我一个完整的解决方案。",
    image_path="error_log.png",
    audio_path="bug_description.mp3",
))
print(f"多模态回答: {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调用开销

    Args:
        image_path: 原图路径
        max_size: 最大尺寸（宽, 高）
        quality: JPEG压缩质量 (1-100)

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

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

    # 转为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:
    """
    音频预处理：统一格式和采样率

    Args:
        audio_path: 原始音频路径
        target_format: 目标格式
        sample_rate: 目标采样率

    Returns:
        处理后的音频文件路径
    """
    audio = AudioSegment.from_file(audio_path)

    # 转为单声道
    audio = audio.set_channels(1)

    # 设置采样率
    audio = audio.set_frame_rate(sample_rate)

    # 导出
    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: 统一提示词
        concurrency: 最大并发数

    Returns:
        处理结果列表
    """
    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)


# 运行批量处理
results = asyncio.run(process_batch(
    ["img1.jpg", "img2.jpg", "img3.jpg", "img4.jpg", "img5.jpg"],
    "请描述这张图片的主要内容，用一句话概括。",
))
for r in results:
    print(f"{r['image']}: {r['result']}")