多模态AI应用开发完全教程

教程简介

本教程全面讲解多模态AI应用开发核心技术,涵盖视觉语言模型(VLM)原理、GPT-4o/Claude/Gemini等主流模型对比、图像理解、OCR文档处理、视频分析、音频转录、多模态RAG系统构建、多模态Agent开发、API调用实战及性能优化等内容。

多模态 AI 应用开发完全教程

适用读者:有基础编程经验的开发者,希望掌握多模态 AI 应用开发的工程师、产品经理和技术决策者。

预计阅读时间:35-45 分钟


目录

  1. 多模态 AI 概述
  2. 主流多模态模型对比
  3. 图像理解与描述生成
  4. OCR 与文档智能处理
  5. 视频理解与分析
  6. 音频转录与理解
  7. 多模态 RAG 系统构建
  8. 多模态 Agent 开发
  9. API 调用实战代码示例
  10. 性能优化与成本控制

1. 多模态 AI 概述

1.1 什么是多模态 AI

多模态 AI(Multimodal AI)是指能够同时理解和处理多种数据类型(模态)的人工智能系统。这些模态包括:

  • 文本(Text):自然语言、代码、结构化数据
  • 图像(Image):照片、截图、图表、手绘草图
  • 视频(Video):连续的图像序列加上音频
  • 音频(Audio):语音、音乐、环境声音

传统的 AI 模型通常是"单模态"的——语言模型只处理文本,图像分类模型只处理图片。而多模态模型的突破在于,它能在一个统一的框架内理解不同模态之间的语义关联

1.2 视觉-语言模型(VLM)

视觉-语言模型(Vision-Language Model, VLM)是当前多模态 AI 的核心范式。VLM 的基本架构如下:

输入:图像 + 文本提示
      │         │
      ▼         ▼
┌──────────┐ ┌──────────┐
│ 视觉编码器 │ │ 文本编码器 │
│ (ViT等)   │ │ (Tokenizer)│
└────┬─────┘ └────┬─────┘
     │            │
     ▼            ▼
┌─────────────────────┐
│    跨模态融合层       │
│  (Cross-Attention)   │
└──────────┬──────────┘
           │
           ▼
┌─────────────────────┐
│    语言模型解码器     │
│  (Transformer)       │
└──────────┬──────────┘
           │
           ▼
输出:文本回答

关键组件

  1. 视觉编码器:将图像转换为特征向量。常用架构包括 ViT(Vision Transformer)、CLIP 视觉编码器、SigLIP 等。

  2. 跨模态投影层:将视觉特征映射到语言模型的嵌入空间。常见的方法有线性投影、MLP 投影、Q-Former 等。

  3. 语言模型骨干:基于 Transformer 的大语言模型,负责理解融合后的多模态信息并生成回答。

1.3 多模态 AI 的发展历程

时间 里程碑 意义
2021 CLIP 发布 首次证明视觉-语言对齐的大规模可行性
2022 Flamingo 首个强大的少样本视觉语言模型
2023.3 GPT-4V 发布 商用多模态模型的标志性产品
2023.10 LLaVA 开源 开启开源多模态模型的繁荣
2024.5 GPT-4o 发布 原生多模态,支持实时音视频
2024-2025 Claude 3.5、Gemini、Qwen-VL 多模态能力成为主流标配

2. 主流多模态模型对比

2.1 模型概览

模型 厂商 图像理解 视频理解 音频理解 上下文窗口 特点
GPT-4o OpenAI ✅ 优秀 ✅ 支持 ✅ 原生 128K 原生多模态,实时交互
Claude 3.5 Sonnet Anthropic ✅ 优秀 ❌ 暂不支持 ❌ 暂不支持 200K 长上下文,精准理解
Gemini Pro Vision Google ✅ 优秀 ✅ 支持 ✅ 支持 1M+ 超长上下文,原生多模态
Qwen-VL 阿里巴巴 ✅ 优秀 ✅ 支持 ❌ 暂不支持 32K+ 开源可部署,中文优化

2.2 选型建议

选择 GPT-4o 当

  • 需要实时音视频交互
  • 已有 OpenAI 生态集成
  • 追求最广泛的通用能力

选择 Claude 3.5 Sonnet 当

  • 需要处理超长文档(200K 上下文)
  • 对输出格式的精确控制要求高
  • 需要高质量的代码生成和分析

选择 Gemini Pro Vision 当

  • 需要处理超长视频或多文档
  • 已有 Google Cloud 生态
  • 对上下文窗口有极高要求

选择 Qwen-VL 当

  • 需要私有化部署
  • 中文场景为主
  • 需要开源模型进行微调

2.3 关键能力维度详解

图像理解能力对比

场景识别    ████████████████████  所有主流模型均优秀
OCR 文字    █████████████████░░░  GPT-4o ≈ Claude > Gemini ≈ Qwen
图表理解    █████████████████░░░  GPT-4o ≈ Claude > Gemini
细节描述    ████████████████░░░░  Claude 细节更精准
空间推理    ███████████████░░░░░  Gemini 和 GPT-4o 略优
多图对比    ████████████████████  所有模型均支持

3. 图像理解与描述生成

3.1 基础图像理解

图像理解是多模态 AI 最基础也最常用的能力。通过向模型发送图像并附带文字提示,可以实现:

  • 场景描述与分类
  • 物体检测与计数
  • 人脸表情分析
  • 图像情感判断
  • 技术图表解读

3.2 实战:使用 OpenAI API 进行图像理解

"""
图像理解实战:使用 GPT-4o 分析图像内容。
支持本地图片(Base64)和远程图片(URL)。
"""

import base64
import httpx
from openai import OpenAI

client = OpenAI()


def analyze_image_with_url(image_url: str, prompt: str) -> str:
    """
    分析远程图片。
    
    Args:
        image_url: 图片的 URL 地址
        prompt: 分析提示词
    Returns:
        模型的回答
    """
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "type": "image_url",
                        "image_url": {
                            "url": image_url,
                            "detail": "high"  # high / low / auto
                        }
                    },
                    {
                        "type": "text",
                        "text": prompt
                    }
                ]
            }
        ],
        max_tokens=1000
    )
    
    return response.choices[0].message.content


def analyze_image_with_base64(image_path: str, prompt: str) -> str:
    """
    分析本地图片(Base64 编码)。
    
    Args:
        image_path: 本地图片路径
        prompt: 分析提示词
    Returns:
        模型的回答
    """
    # 读取图片并编码为 Base64
    with open(image_path, "rb") as f:
        image_data = base64.standard_b64encode(f.read()).decode("utf-8")
    
    # 根据文件扩展名确定 MIME 类型
    ext = image_path.rsplit(".", 1)[-1].lower()
    mime_map = {
        "jpg": "image/jpeg", "jpeg": "image/jpeg",
        "png": "image/png", "gif": "image/gif",
        "webp": "image/webp"
    }
    mime_type = mime_map.get(ext, "image/png")
    
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "type": "image_url",
                        "image_url": {
                            "url": f"data:{mime_type};base64,{image_data}"
                        }
                    },
                    {
                        "type": "text",
                        "text": prompt
                    }
                ]
            }
        ],
        max_tokens=1000
    )
    
    return response.choices[0].message.content


# ==================== 使用示例 ====================

if __name__ == "__main__":
    # 示例 1:分析网络图片
    result = analyze_image_with_url(
        "https://upload.wikimedia.org/wikipedia/commons/thumb/3/3a/Cat03.jpg/1200px-Cat03.jpg",
        "请详细描述这张图片中的内容,包括动物的品种特征、姿态和环境。"
    )
    print("图片分析结果:")
    print(result)
    
    # 示例 2:分析本地图片
    # result = analyze_image_with_base64(
    #     "./photo.jpg",
    #     "这张图片中有哪些文字?请逐一列出。"
    # )

3.3 多图对比分析

多模态模型支持同时输入多张图片进行对比分析:

def compare_images(image_urls: list[str], question: str) -> str:
    """
    对比多张图片。
    
    Args:
        image_urls: 图片 URL 列表
        question: 对比问题
    """
    content = []
    
    # 添加每张图片
    for i, url in enumerate(image_urls):
        content.append({
            "type": "image_url",
            "image_url": {"url": url}
        })
    
    # 添加文字问题
    content.append({
        "type": "text",
        "text": question
    })
    
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": content}],
        max_tokens=1500
    )
    
    return response.choices[0].message.content


# 使用示例
result = compare_images(
    [
        "https://example.com/product_a.jpg",
        "https://example.com/product_b.jpg"
    ],
    "请对比这两款产品的外观设计差异,从颜色、形状、材质三个方面分析。"
)

3.4 结构化图像分析

让模型输出结构化的分析结果,便于后续处理:

import json

def structured_image_analysis(image_url: str) -> dict:
    """
    对图像进行结构化分析,返回 JSON 格式结果。
    """
    prompt = """请分析这张图片并以 JSON 格式返回结果,包含以下字段:
{
    "scene": "场景描述",
    "objects": ["检测到的物体列表"],
    "colors": ["主要颜色"],
    "mood": "整体氛围/情绪",
    "text_content": "图片中的文字(如有)",
    "technical_quality": {
        "resolution": "高/中/低",
        "lighting": "光线条件",
        "composition": "构图评价"
    },
    "tags": ["标签列表"]
}

请只返回 JSON,不要添加其他说明。"""
    
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "user",
                "content": [
                    {"type": "image_url", "image_url": {"url": image_url}},
                    {"type": "text", "text": prompt}
                ]
            }
        ],
        max_tokens=800,
        temperature=0.1  # 低温度以获得更稳定的输出
    )
    
    result_text = response.choices[0].message.content
    # 清理可能的 markdown 代码块标记
    if result_text.startswith("```"):
        result_text = result_text.split("\n", 1)[1].rsplit("```", 1)[0]
    
    return json.loads(result_text)

4. OCR 与文档智能处理

4.1 多模态 OCR 的优势

传统 OCR(如 Tesseract)只能提取文字,而多模态 AI 的 OCR 能力远不止于此:

能力 传统 OCR 多模态 AI
文字提取
格式保持 ❌ 需后处理 ✅ 理解布局
表格识别 ⚠️ 有限 ✅ 结构化提取
手写体 ⚠️ 效果一般 ✅ 较好
理解语义 ✅ 理解内容含义
多语言混合 ⚠️

4.2 实战:文档 OCR 与结构化提取

"""
文档 OCR 实战:使用多模态模型提取文档内容并结构化。
"""

import base64
import json
from openai import OpenAI

client = OpenAI()


def extract_document_content(image_path: str, document_type: str = "auto") -> dict:
    """
    从文档图片中提取结构化内容。
    
    Args:
        image_path: 文档图片路径
        document_type: 文档类型(auto/invoice/receipt/id_card/business_card/form)
    Returns:
        结构化的文档内容
    """
    with open(image_path, "rb") as f:
        image_data = base64.standard_b64encode(f.read()).decode("utf-8")
    
    # 根据文档类型定制提示词
    prompts = {
        "auto": """请分析这张文档图片,识别文档类型并提取所有内容。
返回 JSON 格式:
{
    "document_type": "识别出的文档类型",
    "language": "主要语言",
    "content": {
        // 根据文档类型动态生成的字段
    },
    "raw_text": "完整的原始文字内容(保持原始格式)"
}""",
        
        "invoice": """请提取这张发票中的所有信息。
返回 JSON 格式:
{
    "document_type": "发票",
    "invoice_number": "发票号码",
    "date": "开票日期",
    "seller": {
        "name": "销售方名称",
        "tax_id": "税号"
    },
    "buyer": {
        "name": "购买方名称",
        "tax_id": "税号"
    },
    "items": [
        {
            "name": "项目名称",
            "quantity": 数量,
            "unit_price": 单价,
            "amount": 金额
        }
    ],
    "total": 总金额,
    "tax": 税额,
    "raw_text": "原始文字"
}""",
        
        "business_card": """请提取这张名片中的所有联系信息。
返回 JSON 格式:
{
    "document_type": "名片",
    "name": "姓名",
    "title": "职位",
    "company": "公司名称",
    "phone": "电话",
    "email": "邮箱",
    "address": "地址",
    "website": "网站",
    "social": {
        "wechat": "微信",
        "linkedin": "LinkedIn"
    },
    "raw_text": "原始文字"
}""",
        
        "table": """请提取这张表格图片中的所有数据,保持表格结构。
返回 JSON 格式:
{
    "document_type": "表格",
    "headers": ["列标题1", "列标题2", ...],
    "rows": [
        ["单元格1", "单元格2", ...],
        ...
    ],
    "raw_text": "原始文字"
}"""
    }
    
    prompt = prompts.get(document_type, prompts["auto"])
    
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "type": "image_url",
                        "image_url": {
                            "url": f"data:image/png;base64,{image_data}",
                            "detail": "high"
                        }
                    },
                    {"type": "text", "text": prompt}
                ]
            }
        ],
        max_tokens=2000,
        temperature=0.1
    )
    
    result_text = response.choices[0].message.content
    if result_text.startswith("```"):
        result_text = result_text.split("\n", 1)[1].rsplit("```", 1)[0]
    
    return json.loads(result_text)


def batch_extract_text(image_paths: list[str]) -> list[dict]:
    """
    批量提取多张图片中的文字内容。
    
    Args:
        image_paths: 图片路径列表
    Returns:
        提取结果列表
    """
    results = []
    
    for path in image_paths:
        try:
            with open(path, "rb") as f:
                image_data = base64.standard_b64encode(f.read()).decode("utf-8")
            
            response = client.chat.completions.create(
                model="gpt-4o",
                messages=[
                    {
                        "role": "user",
                        "content": [
                            {
                                "type": "image_url",
                                "image_url": {
                                    "url": f"data:image/png;base64,{image_data}"
                                }
                            },
                            {
                                "type": "text",
                                "text": "请提取图片中的所有文字内容,保持原始格式和排版。只返回文字内容,不要添加额外说明。"
                            }
                        ]
                    }
                ],
                max_tokens=1500
            )
            
            results.append({
                "file": path,
                "status": "success",
                "text": response.choices[0].message.content
            })
        except Exception as e:
            results.append({
                "file": path,
                "status": "error",
                "error": str(e)
            })
    
    return results

4.3 PDF 文档处理

处理 PDF 文档的常见策略是先将页面转换为图片,再用多模态模型分析:

"""
PDF 文档多模态处理:将 PDF 页面转为图片后进行智能分析。
"""

import base64
from pathlib import Path


def pdf_to_images(pdf_path: str, output_dir: str = "./pdf_pages") -> list[str]:
    """
    将 PDF 每页转换为图片。
    
    Args:
        pdf_path: PDF 文件路径
        output_dir: 输出目录
    Returns:
        生成的图片路径列表
    """
    # 使用 pdf2image 或 PyMuPDF
    import fitz  # PyMuPDF
    
    output_path = Path(output_dir)
    output_path.mkdir(parents=True, exist_ok=True)
    
    doc = fitz.open(pdf_path)
    image_paths = []
    
    for page_num in range(len(doc)):
        page = doc.load_page(page_num)
        # 渲染为高分辨率图片
        pix = page.get_pixmap(matrix=fitz.Matrix(2, 2))  # 2x 缩放
        image_path = output_path / f"page_{page_num + 1:03d}.png"
        pix.save(str(image_path))
        image_paths.append(str(image_path))
    
    doc.close()
    return image_paths


def analyze_pdf_document(pdf_path: str) -> dict:
    """
    分析 PDF 文档,提取结构化内容。
    
    Args:
        pdf_path: PDF 文件路径
    Returns:
        文档分析结果
    """
    # 1. 将 PDF 转换为图片
    image_paths = pdf_to_images(pdf_path)
    
    # 2. 逐页分析
    pages_content = []
    
    for i, img_path in enumerate(image_paths):
        with open(img_path, "rb") as f:
            image_data = base64.standard_b64encode(f.read()).decode("utf-8")
        
        from openai import OpenAI
        client = OpenAI()
        
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {
                    "role": "user",
                    "content": [
                        {
                            "type": "image_url",
                            "image_url": {
                                "url": f"data:image/png;base64,{image_data}",
                                "detail": "high"
                            }
                        },
                        {
                            "type": "text",
                            "text": f"这是 PDF 文档的第 {i + 1} 页。请提取此页的完整内容,包括文字、表格、图表描述。保持原始结构。"
                        }
                    ]
                }
            ],
            max_tokens=2000
        )
        
        pages_content.append({
            "page": i + 1,
            "content": response.choices[0].message.content
        })
    
    # 3. 汇总分析
    all_content = "\n\n".join(
        f"--- 第 {p['page']} 页 ---\n{p['content']}" for p in pages_content
    )
    
    return {
        "total_pages": len(image_paths),
        "pages": pages_content,
        "full_text": all_content
    }

5. 视频理解与分析

5.1 视频理解的技术挑战

视频理解比图像理解复杂得多,主要挑战包括:

  • 时间维度:需要理解事件的时序关系
  • 信息密度:视频包含大量冗余信息
  • 计算成本:处理长视频需要大量计算资源
  • 上下文窗口:需要在有限的上下文中表示视频内容

5.2 视频处理策略

策略一:关键帧抽取

从视频中抽取关键帧,将视频理解转化为多图理解:

"""
视频关键帧抽取与分析。
"""

import subprocess
import base64
import json
from pathlib import Path


def extract_keyframes(video_path: str, fps: float = 1.0, output_dir: str = "./frames") -> list[str]:
    """
    从视频中按固定帧率抽取关键帧。
    
    Args:
        video_path: 视频文件路径
        fps: 每秒抽取帧数(1.0 = 每秒1帧)
        output_dir: 输出目录
    Returns:
        抽取的帧图片路径列表
    """
    output_path = Path(output_dir)
    output_path.mkdir(parents=True, exist_ok=True)
    
    output_pattern = str(output_path / "frame_%04d.png")
    
    cmd = [
        "ffmpeg", "-i", video_path,
        "-vf", f"fps={fps}",
        "-q:v", "2",  # 高质量
        output_pattern,
        "-y", "-loglevel", "error"
    ]
    
    subprocess.run(cmd, check=True)
    
    frames = sorted(output_path.glob("frame_*.png"))
    return [str(f) for f in frames]


def get_video_duration(video_path: str) -> float:
    """获取视频时长(秒)。"""
    cmd = [
        "ffprobe", "-v", "quiet",
        "-print_format", "json",
        "-show_format",
        video_path
    ]
    result = subprocess.run(cmd, capture_output=True, text=True)
    info = json.loads(result.stdout)
    return float(info["format"]["duration"])


def analyze_video_scenes(video_path: str, sample_fps: float = 0.5) -> dict:
    """
    分析视频内容,生成场景描述。
    
    Args:
        video_path: 视频文件路径
        sample_fps: 采样帧率
    Returns:
        视频分析结果
    """
    from openai import OpenAI
    client = OpenAI()
    
    # 1. 获取视频基本信息
    duration = get_video_duration(video_path)
    
    # 2. 抽取关键帧
    frames = extract_keyframes(video_path, fps=sample_fps)
    print(f"视频时长: {duration:.1f}秒, 抽取 {len(frames)} 帧")
    
    # 3. 分批分析(每批最多 10 帧)
    batch_size = 10
    scene_descriptions = []
    
    for batch_start in range(0, len(frames), batch_size):
        batch_frames = frames[batch_start:batch_start + batch_size]
        
        content = []
        for i, frame_path in enumerate(batch_frames):
            with open(frame_path, "rb") as f:
                img_data = base64.standard_b64encode(f.read()).decode("utf-8")
            
            frame_idx = batch_start + i
            timestamp = frame_idx / sample_fps
            
            content.append({
                "type": "image_url",
                "image_url": {
                    "url": f"data:image/png;base64,{img_data}",
                    "detail": "low"  # 使用低细节以节省 token
                }
            })
        
        content.append({
            "type": "text",
            "text": f"""这是视频的第 {batch_start + 1} 到 {batch_start + len(batch_frames)} 帧(每秒 {sample_fps} 帧采样)。
请分析这些帧,描述视频在这段时间内发生的事情。
按时间顺序描述主要事件和场景变化。
返回 JSON 格式:
{{
    "scenes": [
        {{
            "start_frame": 起始帧号,
            "end_frame": 结束帧号,
            "timestamp": "大致时间",
            "description": "场景描述",
            "key_objects": ["关键物体"],
            "actions": ["发生的动作"]
        }}
    ]
}}"""
        })
        
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": content}],
            max_tokens=2000,
            temperature=0.2
        )
        
        result_text = response.choices[0].message.content
        if result_text.startswith("```"):
            result_text = result_text.split("\n", 1)[1].rsplit("```", 1)[0]
        
        scene_descriptions.append(json.loads(result_text))
    
    # 4. 汇总结果
    return {
        "duration": duration,
        "total_frames_sampled": len(frames),
        "sample_fps": sample_fps,
        "analysis": scene_descriptions
    }

策略二:视频帧分组 + 逐段分析

对于长视频,先按时间段分组,再逐段分析:

def analyze_long_video(video_path: str, segment_duration: float = 30.0) -> dict:
    """
    分段分析长视频。
    
    Args:
        video_path: 视频文件路径
        segment_duration: 每段时长(秒)
    Returns:
        分段分析结果
    """
    duration = get_video_duration(video_path)
    segments = []
    
    for start_time in range(0, int(duration), int(segment_duration)):
        end_time = min(start_time + segment_duration, duration)
        
        # 提取该段的关键帧
        segment_frames = extract_segment_frames(
            video_path, start_time, end_time, fps=1.0
        )
        
        # 分析该段
        segment_analysis = analyze_frame_batch(
            segment_frames,
            f"这是视频 {format_time(start_time)} 到 {format_time(end_time)} 的画面。请描述这段时间内发生的主要事件。"
        )
        
        segments.append({
            "start": format_time(start_time),
            "end": format_time(end_time),
            "analysis": segment_analysis
        })
    
    return {
        "duration": format_time(duration),
        "segments": segments,
        "summary": generate_video_summary(segments)
    }


def format_time(seconds: float) -> str:
    """将秒数格式化为 HH:MM:SS。"""
    h = int(seconds // 3600)
    m = int((seconds % 3600) // 60)
    s = int(seconds % 60)
    return f"{h:02d}:{m:02d}:{s:02d}"

6. 音频转录与理解

6.1 音频处理能力概览

多模态 AI 在音频领域的应用包括:

  • 语音转录(ASR):将语音转换为文字
  • 语音翻译:跨语言语音转换
  • 音频理解:理解音频中的语义、情感、意图
  • 声音事件检测:识别环境声音(如门铃、汽车喇叭)
  • 说话人分离:区分不同说话人

6.2 使用 Whisper 进行语音转录

"""
音频转录实战:使用 OpenAI Whisper API 进行语音转文字。
"""

from openai import OpenAI
from pathlib import Path

client = OpenAI()


def transcribe_audio(
    audio_path: str,
    language: str = None,
    response_format: str = "verbose_json"
) -> dict:
    """
    转录音频文件。
    
    Args:
        audio_path: 音频文件路径(支持 mp3, mp4, mpeg, mpga, m4a, wav, webm)
        language: 语言代码(如 "zh", "en"),None 表示自动检测
        response_format: 输出格式(json, text, srt, verbose_json, vtt)
    Returns:
        转录结果
    """
    with open(audio_path, "rb") as audio_file:
        result = client.audio.transcriptions.create(
            model="whisper-1",
            file=audio_file,
            language=language,
            response_format=response_format,
            timestamp_granularities=["segment", "word"]  # 获取时间戳
        )
    
    if response_format == "verbose_json":
        return {
            "text": result.text,
            "language": result.language,
            "duration": result.duration,
            "segments": [
                {
                    "id": seg.id,
                    "start": seg.start,
                    "end": seg.end,
                    "text": seg.text
                }
                for seg in result.segments
            ]
        }
    else:
        return {"text": result.text}


def translate_audio(audio_path: str) -> str:
    """
    将音频翻译为英文。
    
    Args:
        audio_path: 音频文件路径
    Returns:
        英文翻译文本
    """
    with open(audio_path, "rb") as audio_file:
        result = client.audio.translations.create(
            model="whisper-1",
            file=audio_file,
            response_format="text"
        )
    
    return result


def transcribe_with_speaker_diarization(audio_path: str) -> list[dict]:
    """
    转录并区分说话人(简化版实现)。
    
    实际的说话人分离通常需要 pyannote.audio 等专用工具。
    此处演示将 Whisper 与简单的分段策略结合。
    """
    # 1. 获取带时间戳的转录结果
    result = transcribe_audio(audio_path, response_format="verbose_json")
    
    # 2. 基于静音间隔分段(简化策略)
    segments = result["segments"]
    conversation = []
    current_speaker = 0
    
    for i, seg in enumerate(segments):
        # 如果与上一段间隔超过 2 秒,认为换了一个说话人
        if i > 0 and seg["start"] - segments[i-1]["end"] > 2.0:
            current_speaker = 1 - current_speaker  # 在 0 和 1 之间切换
        
        conversation.append({
            "speaker": f"说话人{current_speaker + 1}",
            "start": seg["start"],
            "end": seg["end"],
            "text": seg["text"]
        })
    
    return conversation

6.3 音频内容理解

不仅仅是转录,还可以让多模态模型理解音频的深层含义:

def understand_audio(audio_path: str, question: str) -> str:
    """
    理解音频内容并回答问题。
    
    先转录,再结合转录文本进行分析。
    """
    # 1. 转录音频
    transcript = transcribe_audio(audio_path)
    
    # 2. 使用 LLM 分析转录内容
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": "你是一个音频内容分析专家。根据转录文本分析音频内容。"
            },
            {
                "role": "user",
                "content": f"""以下是音频转录内容:

---
{transcript['text']}
---

语言:{transcript.get('language', '未知')}
时长:{transcript.get('duration', '未知')}秒

问题:{question}"""
            }
        ],
        max_tokens=1000
    )
    
    return response.choices[0].message.content


# 使用示例
# result = understand_audio(
#     "meeting.mp3",
#     "这段会议录音的主要议题是什么?有哪些决策?"
# )

7. 多模态 RAG 系统构建

7.1 什么是多模态 RAG

RAG(Retrieval-Augmented Generation,检索增强生成)是让 LLM 基于外部知识回答问题的技术。多模态 RAG 将这一能力扩展到图像、表格、图表等非文本内容。

用户问题(文本)
      │
      ▼
┌──────────────┐
│  多模态嵌入   │ ← 将问题编码为向量
└──────┬───────┘
       │
       ▼
┌──────────────┐     ┌──────────────────┐
│  向量数据库   │────►│ 检索相关文档片段   │
│              │     │ (文本 + 图片)     │
└──────────────┘     └────────┬─────────┘
                              │
                              ▼
                     ┌─────────────────┐
                     │  多模态 LLM      │
                     │  生成最终回答     │
                     └─────────────────┘

7.2 实战:构建多模态 RAG 系统

"""
多模态 RAG 系统:支持文本和图片的检索增强生成。
"""

import base64
import hashlib
from pathlib import Path
from dataclasses import dataclass
from typing import Optional
from openai import OpenAI

client = OpenAI()


@dataclass
class DocumentChunk:
    """文档片段。"""
    id: str
    content: str  # 文本内容
    image_path: Optional[str] = None  # 关联的图片路径
    metadata: Optional[dict] = None  # 元数据(来源、页码等)
    embedding: Optional[list[float]] = None


class MultimodalRAG:
    """多模态 RAG 系统。"""
    
    def __init__(self, collection_name: str = "default"):
        self.chunks: list[DocumentChunk] = []
        self.collection_name = collection_name
    
    def add_text_document(self, text: str, metadata: dict = None):
        """
        添加文本文档。
        自动分块并生成嵌入向量。
        """
        # 简单的分块策略:按段落分割
        paragraphs = [p.strip() for p in text.split("\n\n") if p.strip()]
        
        for i, para in enumerate(paragraphs):
            chunk = DocumentChunk(
                id=hashlib.md5(f"{para}_{i}".encode()).hexdigest()[:12],
                content=para,
                metadata={**(metadata or {}), "chunk_index": i}
            )
            chunk.embedding = self._get_text_embedding(para)
            self.chunks.append(chunk)
    
    def add_image_document(self, image_path: str, description: str = None, metadata: dict = None):
        """
        添加图片文档。
        使用多模态模型生成描述,然后嵌入。
        """
        if description is None:
            description = self._describe_image(image_path)
        
        chunk = DocumentChunk(
            id=hashlib.md5(image_path.encode()).hexdigest()[:12],
            content=description,
            image_path=image_path,
            metadata={**(metadata or {}), "type": "image"}
        )
        chunk.embedding = self._get_text_embedding(description)
        self.chunks.append(chunk)
    
    def _describe_image(self, image_path: str) -> str:
        """使用多模态模型描述图片。"""
        with open(image_path, "rb") as f:
            img_data = base64.standard_b64encode(f.read()).decode("utf-8")
        
        ext = image_path.rsplit(".", 1)[-1].lower()
        mime = f"image/{'jpeg' if ext in ('jpg', 'jpeg') else ext}"
        
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[{
                "role": "user",
                "content": [
                    {"type": "image_url", "image_url": {"url": f"data:{mime};base64,{img_data}"}},
                    {"type": "text", "text": "请详细描述这张图片的内容,包括所有可见的文字、图表、数据和关键信息。"}
                ]
            }],
            max_tokens=500
        )
        
        return response.choices[0].message.content
    
    def _get_text_embedding(self, text: str) -> list[float]:
        """获取文本嵌入向量。"""
        response = client.embeddings.create(
            model="text-embedding-3-small",
            input=text
        )
        return response.data[0].embedding
    
    def _cosine_similarity(self, a: list[float], b: list[float]) -> float:
        """计算余弦相似度。"""
        import math
        dot = sum(x * y for x, y in zip(a, b))
        norm_a = math.sqrt(sum(x * x for x in a))
        norm_b = math.sqrt(sum(x * x for x in b))
        return dot / (norm_a * norm_b) if norm_a * norm_b > 0 else 0
    
    def retrieve(self, query: str, top_k: int = 5) -> list[DocumentChunk]:
        """检索与查询最相关的文档片段。"""
        query_embedding = self._get_text_embedding(query)
        
        scored = []
        for chunk in self.chunks:
            if chunk.embedding:
                score = self._cosine_similarity(query_embedding, chunk.embedding)
                scored.append((score, chunk))
        
        scored.sort(key=lambda x: x[0], reverse=True)
        return [chunk for _, chunk in scored[:top_k]]
    
    def query(self, question: str, top_k: int = 5) -> str:
        """
        回答用户问题(检索 + 生成)。
        """
        # 1. 检索相关文档
        relevant_chunks = self.retrieve(question, top_k=top_k)
        
        # 2. 构建上下文
        context_parts = []
        for i, chunk in enumerate(relevant_chunks):
            source = chunk.metadata.get("source", "未知") if chunk.metadata else "未知"
            context_parts.append(f"[来源 {i+1}: {source}]\n{chunk.content}")
        
        context = "\n\n---\n\n".join(context_parts)
        
        # 3. 调用 LLM 生成回答
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {
                    "role": "system",
                    "content": """你是一个知识库问答助手。根据提供的参考资料回答用户问题。
要求:
- 只基于参考资料回答,不要编造信息
- 如果参考资料不足以回答问题,明确说明
- 引用信息时标注来源编号"""
                },
                {
                    "role": "user",
                    "content": f"""参考资料:
{context}

问题:{question}"""
                }
            ],
            max_tokens=1500
        )
        
        return response.choices[0].message.content


# ==================== 使用示例 ====================

if __name__ == "__main__":
    rag = MultimodalRAG("my_knowledge_base")
    
    # 添加文本知识
    rag.add_text_document("""
    MCP(Model Context Protocol)是一种开放协议,用于标准化 AI 模型与外部工具的连接。
    它采用 JSON-RPC 2.0 通信协议,支持三种核心能力:Resources、Tools 和 Prompts。
    MCP 的架构由 Host、Client 和 Server 三层组成。
    """, metadata={"source": "MCP 教程"})
    
    # 添加图片知识
    # rag.add_image_document("./architecture.png", metadata={"source": "架构图"})
    
    # 提问
    answer = rag.query("MCP 协议有哪些核心能力?")
    print(answer)

8. 多模态 Agent 开发

8.1 多模态 Agent 架构

多模态 Agent 是能够感知多种输入模态并执行复杂任务的智能体。其核心架构包括:

用户输入(文本 + 图片 + 音频)
              │
              ▼
    ┌───────────────────┐
    │    感知层          │
    │  (Perception)     │
    │  图像理解/音频转录  │
    └────────┬──────────┘
             │
             ▼
    ┌───────────────────┐
    │    认知层          │
    │  (Cognition)      │
    │  推理/规划/决策     │
    └────────┬──────────┘
             │
             ▼
    ┌───────────────────┐
    │    行动层          │
    │  (Action)         │
    │  工具调用/代码执行  │
    └────────┬──────────┘
             │
             ▼
    ┌───────────────────┐
    │    记忆层          │
    │  (Memory)         │
    │  短期/长期记忆     │
    └───────────────────┘

8.2 实战:多模态任务 Agent

"""
多模态 Agent:能够处理图片、音频输入并执行复杂任务。
"""

import base64
import json
from typing import Any
from openai import OpenAI

client = OpenAI()


class MultimodalAgent:
    """多模态任务 Agent。"""
    
    def __init__(self):
        self.tools = self._define_tools()
        self.conversation_history = []
        self.memory = []  # 长期记忆
    
    def _define_tools(self) -> list[dict]:
        """定义 Agent 可用的工具。"""
        return [
            {
                "type": "function",
                "function": {
                    "name": "analyze_image",
                    "description": "详细分析图片内容",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "image_path": {"type": "string", "description": "图片路径"},
                            "focus": {"type": "string", "description": "分析重点"}
                        },
                        "required": ["image_path"]
                    }
                }
            },
            {
                "type": "function",
                "function": {
                    "name": "transcribe_audio",
                    "description": "转录音频文件为文字",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "audio_path": {"type": "string", "description": "音频文件路径"}
                        },
                        "required": ["audio_path"]
                    }
                }
            },
            {
                "type": "function",
                "function": {
                    "name": "search_knowledge",
                    "description": "搜索知识库获取相关信息",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "query": {"type": "string", "description": "搜索查询"}
                        },
                        "required": ["query"]
                    }
                }
            },
            {
                "type": "function",
                "function": {
                    "name": "generate_report",
                    "description": "生成分析报告",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "title": {"type": "string", "description": "报告标题"},
                            "content": {"type": "string", "description": "报告内容"},
                            "format": {
                                "type": "string",
                                "enum": ["markdown", "html", "plain"],
                                "description": "输出格式"
                            }
                        },
                        "required": ["title", "content"]
                    }
                }
            }
        ]
    
    def _execute_tool(self, tool_name: str, arguments: dict) -> str:
        """执行工具调用。"""
        if tool_name == "analyze_image":
            return self._tool_analyze_image(arguments)
        elif tool_name == "transcribe_audio":
            return self._tool_transcribe(arguments)
        elif tool_name == "search_knowledge":
            return self._tool_search(arguments)
        elif tool_name == "generate_report":
            return self._tool_report(arguments)
        else:
            return f"未知工具:{tool_name}"
    
    def _tool_analyze_image(self, args: dict) -> str:
        """分析图片。"""
        image_path = args["image_path"]
        focus = args.get("focus", "全面分析")
        
        with open(image_path, "rb") as f:
            img_data = base64.standard_b64encode(f.read()).decode("utf-8")
        
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[{
                "role": "user",
                "content": [
                    {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{img_data}"}},
                    {"type": "text", "text": f"请{focus}这张图片。"}
                ]
            }],
            max_tokens=1000
        )
        
        return response.choices[0].message.content
    
    def _tool_transcribe(self, args: dict) -> str:
        """转录音频。"""
        with open(args["audio_path"], "rb") as f:
            result = client.audio.transcriptions.create(
                model="whisper-1",
                file=f,
                response_format="text"
            )
        return result
    
    def _tool_search(self, args: dict) -> str:
        """搜索知识库(模拟实现)。"""
        return f"知识库搜索结果:关于 '{args['query']}' 的相关信息..."
    
    def _tool_report(self, args: dict) -> str:
        """生成报告。"""
        title = args["title"]
        content = args["content"]
        fmt = args.get("format", "markdown")
        
        if fmt == "markdown":
            return f"# {title}\n\n{content}"
        elif fmt == "html":
            return f"<html><body><h1>{title}</h1><p>{content}</p></body></html>"
        else:
            return f"{title}\n{'='*len(title)}\n\n{content}"
    
    def process_message(self, user_message: str, images: list[str] = None) -> str:
        """
        处理用户消息,支持文本和图片输入。
        
        Args:
            user_message: 用户的文字消息
            images: 图片路径列表(可选)
        Returns:
            Agent 的回答
        """
        # 构建消息内容
        content = []
        
        # 添加图片
        if images:
            for img_path in images:
                with open(img_path, "rb") as f:
                    img_data = base64.standard_b64encode(f.read()).decode("utf-8")
                content.append({
                    "type": "image_url",
                    "image_url": {"url": f"data:image/png;base64,{img_data}"}
                })
        
        # 添加文字
        content.append({"type": "text", "text": user_message})
        
        # 添加到对话历史
        self.conversation_history.append({"role": "user", "content": content})
        
        # 调用 LLM(支持工具调用)
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": """你是一个多模态 AI 助手。
你可以分析图片、转录音频、搜索知识库并生成报告。
请根据用户的需求,合理使用提供的工具来完成任务。
回答时使用中文。"""},
                *self.conversation_history
            ],
            tools=self.tools,
            tool_choice="auto"
        )
        
        assistant_message = response.choices[0].message
        
        # 处理工具调用
        if assistant_message.tool_calls:
            # 记录助手的工具调用决策
            self.conversation_history.append(assistant_message.model_dump())
            
            # 执行每个工具调用
            for tool_call in assistant_message.tool_calls:
                result = self._execute_tool(
                    tool_call.function.name,
                    json.loads(tool_call.function.arguments)
                )
                
                self.conversation_history.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": result
                })
            
            # 基于工具结果生成最终回答
            final_response = client.chat.completions.create(
                model="gpt-4o",
                messages=[
                    {"role": "system", "content": "根据工具执行结果,生成清晰的最终回答。"},
                    *self.conversation_history
                ]
            )
            
            final_text = final_response.choices[0].message.content
            self.conversation_history.append({"role": "assistant", "content": final_text})
            return final_text
        
        else:
            # 直接回答
            self.conversation_history.append({"role": "assistant", "content": assistant_message.content})
            return assistant_message.content


# ==================== 使用示例 ====================

if __name__ == "__main__":
    agent = MultimodalAgent()
    
    # 纯文本交互
    result = agent.process_message("你好,请介绍一下你的能力。")
    print(result)
    
    # 带图片的交互
    # result = agent.process_message(
    #     "请分析这张图表,提取关键数据并生成报告。",
    #     images=["./chart.png"]
    # )
    # print(result)

9. API 调用实战代码示例

9.1 统一的多模态 API 封装

"""
统一的多模态 API 封装:支持多个模型提供商。
"""

import base64
from typing import Optional, Literal
from dataclasses import dataclass


@dataclass
class MultimodalMessage:
    """多模态消息。"""
    text: str
    images: Optional[list[str]] = None  # 图片路径或 URL
    audio: Optional[str] = None  # 音频路径


class MultimodalClient:
    """统一的多模态客户端。"""
    
    def __init__(self, provider: Literal["openai", "anthropic", "qwen"] = "openai"):
        self.provider = provider
        self._init_client(provider)
    
    def _init_client(self, provider: str):
        if provider == "openai":
            from openai import OpenAI
            self.client = OpenAI()
            self.model = "gpt-4o"
        elif provider == "anthropic":
            import anthropic
            self.client = anthropic.Anthropic()
            self.model = "claude-3-5-sonnet-20241022"
        elif provider == "qwen":
            from openai import OpenAI
            self.client = OpenAI(
                base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
            )
            self.model = "qwen-vl-max"
    
    def chat(self, message: MultimodalMessage, max_tokens: int = 1000) -> str:
        """发送多模态消息并获取回答。"""
        if self.provider == "anthropic":
            return self._chat_anthropic(message, max_tokens)
        else:
            return self._chat_openai_format(message, max_tokens)
    
    def _chat_openai_format(self, message: MultimodalMessage, max_tokens: int) -> str:
        """OpenAI 兼容格式的调用。"""
        content = []
        
        # 添加图片
        if message.images:
            for img in message.images:
                if img.startswith(("http://", "https://")):
                    content.append({
                        "type": "image_url",
                        "image_url": {"url": img}
                    })
                else:
                    with open(img, "rb") as f:
                        b64 = base64.standard_b64encode(f.read()).decode()
                    ext = img.rsplit(".", 1)[-1].lower()
                    mime = f"image/{'jpeg' if ext in ('jpg', 'jpeg') else ext}"
                    content.append({
                        "type": "image_url",
                        "image_url": {"url": f"data:{mime};base64,{b64}"}
                    })
        
        # 添加文本
        content.append({"type": "text", "text": message.text})
        
        response = self.client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": content}],
            max_tokens=max_tokens
        )
        
        return response.choices[0].message.content
    
    def _chat_anthropic(self, message: MultimodalMessage, max_tokens: int) -> str:
        """Anthropic 格式的调用。"""
        content = []
        
        if message.images:
            for img in message.images:
                if img.startswith(("http://", "https://")):
                    content.append({
                        "type": "image",
                        "source": {
                            "type": "url",
                            "url": img
                        }
                    })
                else:
                    with open(img, "rb") as f:
                        b64 = base64.standard_b64encode(f.read()).decode()
                    ext = img.rsplit(".", 1)[-1].lower()
                    media = f"image/{'jpeg' if ext in ('jpg', 'jpeg') else ext}"
                    content.append({
                        "type": "image",
                        "source": {
                            "type": "base64",
                            "media_type": media,
                            "data": b64
                        }
                    })
        
        content.append({"type": "text", "text": message.text})
        
        response = self.client.messages.create(
            model=self.model,
            max_tokens=max_tokens,
            messages=[{"role": "user", "content": content}]
        )
        
        return response.content[0].text


# ==================== 使用示例 ====================

if __name__ == "__main__":
    # 切换不同提供商
    for provider in ["openai", "qwen"]:
        print(f"\n--- 使用 {provider} ---")
        client = MultimodalClient(provider)
        result = client.chat(MultimodalMessage(
            text="用一句话解释什么是多模态 AI。",
        ))
        print(result)

9.2 流式响应处理

"""
多模态流式响应:实时输出模型回答。
"""

from openai import OpenAI

client = OpenAI()


def stream_multimodal_response(image_url: str, prompt: str):
    """
    流式获取多模态模型的回答。
    
    Args:
        image_url: 图片 URL
        prompt: 提示词
    Yields:
        逐个文本片段
    """
    stream = client.chat.completions.create(
        model="gpt-4o",
        messages=[{
            "role": "user",
            "content": [
                {"type": "image_url", "image_url": {"url": image_url}},
                {"type": "text", "text": prompt}
            ]
        }],
        max_tokens=1500,
        stream=True  # 启用流式输出
    )
    
    for chunk in stream:
        if chunk.choices[0].delta.content:
            yield chunk.choices[0].delta.content


# 使用示例
if __name__ == "__main__":
    for text_chunk in stream_multimodal_response(
        "https://example.com/image.jpg",
        "描述这张图片的内容"
    ):
        print(text_chunk, end="", flush=True)
    print()  # 换行

10. 性能优化与成本控制

10.1 图像优化策略

图像处理是多模态 API 的主要成本来源。以下是关键优化策略:

"""
图像预处理优化:在不影响分析质量的前提下降低 API 成本。
"""

from PIL import Image
import io
import base64


def optimize_image_for_api(
    image_path: str,
    max_size: tuple[int, int] = (1024, 1024),
    quality: int = 85,
    format: str = "JPEG"
) -> str:
    """
    优化图片用于 API 调用。
    
    策略:
    1. 缩放到合理尺寸(不超过 max_size)
    2. 压缩质量(85% 通常足够)
    3. 转换为 JPEG(更小的文件大小)
    
    Args:
        image_path: 原始图片路径
        max_size: 最大尺寸
        quality: JPEG 质量(1-100)
        format: 输出格式
    Returns:
        Base64 编码的优化后图片
    """
    img = Image.open(image_path)
    
    # 转换为 RGB(移除 alpha 通道)
    if img.mode in ("RGBA", "P"):
        img = img.convert("RGB")
    
    # 计算等比缩放
    ratio = min(max_size[0] / img.width, max_size[1] / img.height)
    if ratio < 1:
        new_size = (int(img.width * ratio), int(img.height * ratio))
        img = img.resize(new_size, Image.LANCZOS)
    
    # 压缩为字节流
    buffer = io.BytesIO()
    img.save(buffer, format=format, quality=quality, optimize=True)
    
    original_size = len(open(image_path, "rb").read())
    optimized_size = buffer.tell()
    
    print(f"图片优化: {original_size / 1024:.1f}KB → {optimized_size / 1024:.1f}KB "
          f"(节省 {(1 - optimized_size/original_size)*100:.1f}%)")
    
    return base64.standard_b64encode(buffer.getvalue()).decode("utf-8")


def estimate_image_tokens(width: int, height: int, detail: str = "auto") -> int:
    """
    估算图片消耗的 token 数量(基于 OpenAI 的计算方式)。
    
    Args:
        width: 图片宽度
        height: 图片高度
        detail: 细节级别(low/high/auto)
    Returns:
        估算的 token 数
    """
    if detail == "low":
        return 85  # low 模式固定 85 tokens
    
    # high 模式的计算
    if width > 2048 or height > 2048:
        ratio = 2048 / max(width, height)
        width = int(width * ratio)
        height = int(height * ratio)
    
    if width > 768 or height > 768:
        ratio = 768 / min(width, height)
        width = int(width * ratio)
        height = int(height * ratio)
    
    # 按 512x512 的瓦片计算
    tiles_w = -(-width // 512)  # 向上取整
    tiles_h = -(-height // 512)
    total_tiles = tiles_w * tiles_h
    
    return 85 + 170 * total_tokens

10.2 缓存策略

"""
多模态 API 结果缓存:避免重复调用相同内容。
"""

import hashlib
import json
from pathlib import Path
from datetime import datetime, timedelta


class MultimodalCache:
    """多模态 API 结果缓存。"""
    
    def __init__(self, cache_dir: str = "./cache", ttl_hours: int = 24):
        self.cache_dir = Path(cache_dir)
        self.cache_dir.mkdir(parents=True, exist_ok=True)
        self.ttl = timedelta(hours=ttl_hours)
    
    def _make_key(self, image_path: str, prompt: str, model: str) -> str:
        """生成缓存键。"""
        # 对图片内容计算哈希
        with open(image_path, "rb") as f:
            img_hash = hashlib.md5(f.read()).hexdigest()[:12]
        
        prompt_hash = hashlib.md5(prompt.encode()).hexdigest()[:8]
        return f"{model}_{img_hash}_{prompt_hash}"
    
    def get(self, image_path: str, prompt: str, model: str) -> str | None:
        """获取缓存结果。"""
        key = self._make_key(image_path, prompt, model)
        cache_file = self.cache_dir / f"{key}.json"
        
        if not cache_file.exists():
            return None
        
        data = json.loads(cache_file.read_text())
        
        # 检查是否过期
        cached_time = datetime.fromisoformat(data["timestamp"])
        if datetime.now() - cached_time > self.ttl:
            cache_file.unlink()
            return None
        
        return data["result"]
    
    def set(self, image_path: str, prompt: str, model: str, result: str):
        """保存缓存结果。"""
        key = self._make_key(image_path, prompt, model)
        cache_file = self.cache_dir / f"{key}.json"
        
        data = {
            "image": image_path,
            "prompt": prompt,
            "model": model,
            "result": result,
            "timestamp": datetime.now().isoformat()
        }
        
        cache_file.write_text(json.dumps(data, ensure_ascii=False, indent=2))


# 带缓存的分析函数
def analyze_with_cache(image_path: str, prompt: str, model: str = "gpt-4o") -> str:
    """带缓存的图像分析。"""
    cache = MultimodalCache()
    
    # 尝试从缓存获取
    cached = cache.get(image_path, prompt, model)
    if cached:
        print("命中缓存,跳过 API 调用")
        return cached
    
    # 调用 API
    from openai import OpenAI
    client = OpenAI()
    
    with open(image_path, "rb") as f:
        img_data = base64.standard_b64encode(f.read()).decode()
    
    response = client.chat.completions.create(
        model=model,
        messages=[{
            "role": "user",
            "content": [
                {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{img_data}"}},
                {"type": "text", "text": prompt}
            ]
        }],
        max_tokens=1000
    )
    
    result = response.choices[0].message.content
    
    # 保存到缓存
    cache.set(image_path, prompt, model, result)
    
    return result

10.3 成本估算与监控

"""
多模态 API 成本估算工具。
"""

# 各模型的价格(2025 年参考价格,实际价格请查阅官方)
PRICING = {
    "gpt-4o": {
        "input_text": 2.50,    # 每百万 token
        "input_image": 3.825,  # 每百万 token(high detail 平均)
        "output": 10.00,       # 每百万 token
    },
    "gpt-4o-mini": {
        "input_text": 0.15,
        "input_image": 0.15,
        "output": 0.60,
    },
    "claude-3-5-sonnet": {
        "input_text": 3.00,
        "input_image": 3.00,
        "output": 15.00,
    }
}


def estimate_cost(
    model: str,
    text_tokens: int,
    image_tokens: int,
    output_tokens: int
) -> dict:
    """
    估算 API 调用成本。
    
    Args:
        model: 模型名称
        text_tokens: 文本输入 token 数
        image_tokens: 图片输入 token 数
        output_tokens: 输出 token 数
    Returns:
        成本明细
    """
    pricing = PRICING.get(model, PRICING["gpt-4o"])
    
    text_cost = (text_tokens / 1_000_000) * pricing["input_text"]
    image_cost = (image_tokens / 1_000_000) * pricing["input_image"]
    output_cost = (output_tokens / 1_000_000) * pricing["output"]
    total_cost = text_cost + image_cost + output_cost
    
    return {
        "model": model,
        "breakdown": {
            "text_input": f"${text_cost:.6f} ({text_tokens} tokens)",
            "image_input": f"${image_cost:.6f} ({image_tokens} tokens)",
            "output": f"${output_cost:.6f} ({output_tokens} tokens)",
        },
        "total": f"${total_cost:.6f}",
        "per_1000_calls": f"${total_cost * 1000:.4f}"
    }


# 使用示例
if __name__ == "__main__":
    # 估算一次典型的图片分析调用
    cost = estimate_cost(
        model="gpt-4o",
        text_tokens=200,      # 提示词约 200 tokens
        image_tokens=1105,    # 一张 1024x1024 的图片约 1105 tokens
        output_tokens=500     # 回答约 500 tokens
    )
    
    print("成本估算:")
    print(json.dumps(cost, indent=2, ensure_ascii=False))

10.4 最佳实践总结

性能优化

策略 效果 实现难度
图片压缩与缩减 降低 50-80% 图片 token ⭐ 简单
使用 detail: low 固定 85 tokens/图 ⭐ 简单
结果缓存 重复调用零成本 ⭐⭐ 中等
批量请求 减少网络开销 ⭐⭐ 中等
关键帧抽取(视频) 减少 90%+ 帧数 ⭐⭐ 中等
异步并发处理 提升吞吐量 ⭐⭐⭐ 较高

成本控制

  1. 选对模型:简单任务用小模型(GPT-4o-mini),复杂任务用大模型
  2. 控制图片质量:大多数场景不需要 detail: high
  3. 限制输出长度:合理设置 max_tokens
  4. 监控用量:建立成本监控仪表板
  5. 预估预算:上线前进行成本预估和压力测试
  6. 利用缓存:相同输入不重复调用
  7. 降级策略:API 限流时自动切换到更便宜的模型

总结

多模态 AI 应用开发正在从"锦上添花"变为"刚需能力"。通过本文的学习,你应该能够:

  1. ✅ 理解多模态 AI 的技术架构和发展趋势
  2. ✅ 根据需求选择合适的多模态模型
  3. ✅ 实现图像理解、OCR、视频分析、音频转录等功能
  4. ✅ 构建多模态 RAG 系统
  5. ✅ 开发多模态 Agent
  6. ✅ 进行性能优化和成本控制

多模态 AI 的发展日新月异,建议持续关注以下方向:

  • 原生多模态模型:从"视觉编码器 + LLM"走向端到端的原生多模态架构
  • 实时交互:GPT-4o 展示的实时音视频对话能力将成为标配
  • 具身智能:多模态 AI 与机器人、自动驾驶的结合
  • 多模态生成:从"理解"走向"生成"(图像、视频、音频生成)

下一步行动

  1. 注册一个支持多模态的 API(OpenAI / Anthropic / 阿里云)
  2. 运行本文的代码示例,体验图像理解能力
  3. 尝试为你的业务场景构建一个多模态 RAG 原型
  4. 评估成本和效果,制定落地计划

内容声明

本文内容为AI技术学习教程,仅供学习参考。如涉及技术问题,欢迎通过 xurj005@163.com 与我们交流。

目录