多模态 AI 应用开发教程
面向零基础用户的完整实战指南
目录
- 第一章：多模态AI概述
- 第二章：视觉语言模型 VLM 原理
- 第三章：GPT-4V/GPT-4o 多模态API实战
- 第四章：LLaVA 本地部署与使用
- 第五章：Qwen-VL 通义千问视觉模型实战
- 第六章：图像理解应用开发
- 第七章：语音多模态
- 第八章：视频理解与分析
- 第九章：多模态RAG系统
- 第十章：多模态Agent开发
- 第十一章：综合实战项目
- 第十二章：性能优化与部署策略
- 第十三章：未来趋势与学习路径
第一章：多模态AI概述
1.1 什么是多模态
在人工智能领域，「模态」（Modality）指的是一种信息的表达形式。人类通过五感——视觉、听觉、触觉、嗅觉、味觉来感知世界，每一种感官就是一种模态。在计算机和AI领域，常见的模态包括：
- 文本模态：自然语言文字，如文章、对话、代码等
- 视觉模态：图像、视频、图表、文档扫描件等
- 音频模态：语音、音乐、环境声音等
- 结构化数据模态：表格、数据库记录、知识图谱等
- 传感器模态：激光雷达点云、红外图像、深度图等
多模态（Multimodal） 指的是同时处理和理解两种或两种以上模态信息的能力。举一个生活中的例子：当你看到一张朋友发来的美食照片时，你同时在处理视觉信息（食物的外观）和可能的文本信息（配文"今天的晚餐"），并综合理解这是一个朋友分享晚餐的社交行为。这就是多模态理解。
从技术角度看，多模态AI需要解决以下核心挑战：
异构数据对齐：不同模态的数据形式完全不同——图像是像素矩阵，文本是离散符号序列，音频是时间序列波形。如何将它们放在同一个语义空间中进行比较和推理？
信息互补与融合：不同模态提供互补信息。比如一张图片中有人在弹吉他，加上文字说明"我的第一堂吉他课"，两个模态的信息结合起来才能完整理解场景。
跨模态生成：不仅要理解，还要能生成。比如根据文字描述生成图像，或者根据图像生成文字描述。
模态缺失处理：实际应用中，某些模态可能缺失。系统需要在部分模态缺失的情况下仍能正常工作。
1.2 多模态AI的发展历程
多模态AI的发展可以追溯到很早，但真正爆发是在2023年之后：
早期阶段（2015-2019）：这个阶段主要是图像描述（Image Captioning）和视觉问答（VQA）的探索。代表性工作包括Show-and-Tell模型，它能为图像生成简单的文字描述。这个时期的模型通常将CNN（卷积神经网络）作为图像编码器，将RNN（循环神经网络）作为文本解码器，通过简单的拼接或注意力机制来融合两种模态。
发展阶段（2020-2022）：Transformer架构的兴起带来了重大突破。CLIP（Contrastive Language-Image Pre-training）模型通过对比学习，将图像和文本映射到同一个语义空间，实现了零样本图像分类。DALL-E系列展示了文本到图像生成的可能性。这个阶段，多模态模型开始展现出强大的跨模态理解能力。
爆发阶段（2023-至今）：GPT-4V的发布标志着多模态大语言模型（Multimodal LLM）时代的到来。随后，开源社区也快速跟进，LLaVA、Qwen-VL、InternVL等模型相继发布。多模态AI从"能看图"进化到"能看视频、能听声音、能操作工具"，应用场景极大丰富。
1.3 多模态AI的核心技术
多模态AI涉及多项核心技术：
1. 模态编码（Modal Encoding）
不同模态的数据需要被转换成统一的数值表示。图像通常使用Vision Transformer（ViT）或CNN进行编码，文本使用Tokenizer和Embedding层编码，音频则使用Whisper等模型编码。
# 简化的模态编码示例
from transformers import ViTModel, AutoTokenizer
# 图像编码
vit_model = ViTModel.from_pretrained('google/vit-base-patch16-224')
image_features = vit_model(pixel_values).last_hidden_state
# 文本编码
tokenizer = AutoTokenizer.from_pretrained('bert-base-chinese')
text_tokens = tokenizer("这是一张图片", return_tensors="pt")
2. 跨模态对齐（Cross-modal Alignment）
将不同模态的特征映射到共享的语义空间。常用的方法包括对比学习（如CLIP）、投影层（Projection Layer）等。
# 简化的跨模态投影层
import torch.nn as nn
class ProjectionLayer(nn.Module):
def __init__(self, vision_dim, text_dim):
super().__init__()
self.projector = nn.Linear(vision_dim, text_dim)
def forward(self, vision_features):
return self.projector(vision_features)
3. 多模态融合（Multimodal Fusion）
融合策略决定了不同模态信息如何交互：
- 早期融合（Early Fusion）：在输入阶段就将不同模态拼接在一起
- 晚期融合（Late Fusion）：各模态独立处理后在决策层融合
- 交叉注意力融合（Cross-attention Fusion）：通过注意力机制让不同模态相互关注
4. 多模态生成（Multimodal Generation）
不仅仅是理解，还能生成多种模态的输出。例如，根据文字描述生成图像（Stable Diffusion），根据图像生成文字描述，或根据语音生成文本（ASR）。
1.4 多模态AI的典型应用场景
场景一：智能客服系统
传统客服只能处理文字，多模态客服可以同时处理用户发送的截图、语音消息和文字描述，更准确地理解问题。
# 多模态客服处理流程示例
def handle_customer_request(text, image=None, audio=None):
context = []
if text:
context.append(f"用户文字描述：{text}")
if image:
image_desc = describe_image(image)
context.append(f"用户发送的图片内容：{image_desc}")
if audio:
transcript = speech_to_text(audio)
context.append(f"用户语音内容：{transcript}")
combined_context = "\n".join(context)
response = llm.generate(combined_context)
return response
场景二：文档智能处理
OCR识别扫描文档中的文字，同时理解文档中的图表、表格和布局信息，实现智能文档分析。在金融领域，银行需要处理大量的贷款申请表、发票、合同等文档。多模态AI可以自动识别文档类型，提取关键信息，进行合规性检查。
场景三：自动驾驶感知
融合摄像头图像、激光雷达点云、GPS定位等多模态传感器数据，实现对驾驶环境的全面理解。每个传感器提供不同维度的信息：摄像头提供颜色和纹理，激光雷达提供精确的距离信息，GPS提供全局位置。
场景四：医学影像分析
结合医学影像（X光、CT、MRI）和病历文本，辅助医生进行诊断。例如，AI可以分析X光片中的异常区域，同时参考患者的病史和症状描述，提供更准确的诊断建议。
场景五：教育领域
多模态AI可以为学生提供更丰富的学习体验。比如，学生拍下数学题的照片，AI不仅能识别题目内容，还能理解题目中的图表，给出详细的解题步骤和讲解视频。
1.5 开发环境准备
在开始学习之前，我们需要准备好开发环境：
# 创建Python虚拟环境
python3 -m venv multimodal-env
source multimodal-env/bin/activate
# 安装基础依赖
pip install torch torchvision torchaudio
pip install transformers
pip install pillow
pip install opencv-python
pip install numpy
pip install requests
pip install gradio
# 验证环境
import torch
import transformers
from PIL import Image
print(f"PyTorch版本：{torch.__version__}")
print(f"Transformers版本：{transformers.__version__}")
print(f"CUDA可用：{torch.cuda.is_available()}")
if torch.cuda.is_available():
print(f"GPU设备：{torch.cuda.get_device_name(0)}")
1.6 本教程的学习路线
本教程将从基础概念出发，逐步深入到实际应用开发：
- 基础篇（第1-3章）：理解多模态AI的核心概念，学会使用商业API
- 实践篇（第4-5章）：部署和使用开源多模态模型
- 应用篇（第6-8章）：开发图像、语音、视频相关的多模态应用
- 进阶篇（第9-10章）：构建多模态RAG和Agent系统
- 项目篇（第11-13章）：综合实战、性能优化和未来展望
第二章：视觉语言模型 VLM 原理
2.1 什么是视觉语言模型
视觉语言模型（Vision-Language Model，简称VLM）是一类能够同时理解图像和文本的AI模型。它是多模态AI最核心的技术之一，让AI不仅能"读懂"文字，还能"看懂"图片。
VLM的核心能力包括：
- 图像描述：为图片生成自然语言描述
- 视觉问答：根据图片回答问题
- 图像推理：理解图像中的逻辑关系并进行推理
- OCR与文档理解：识别图像中的文字并理解其含义
- 视觉对话：围绕图像进行多轮对话
VLM与传统计算机视觉模型的区别在于：传统CV模型只能输出分类标签或检测框，而VLM能用自然语言自由地描述和讨论图像内容，就像人类看图说话一样。
2.2 VLM的架构演进
VLM的架构经历了几个重要阶段：
2.2.1 双塔架构（Dual-Encoder）
最简单的VLM架构是双塔模型，以CLIP为代表：
[图像] -> Image Encoder -> 图像特征向量
[文本] -> Text Encoder -> 文本特征向量
|
计算相似度（对比学习）
CLIP通过对比学习训练，将匹配的图文对拉近，不匹配的推远。这种架构擅长图文匹配和检索，但不能生成文本。
# CLIP模型的基本使用
import torch
from transformers import CLIPProcessor, CLIPModel
model = CLIPModel.from_pretrained("openai/clip-vit-base-patch32")
processor = CLIPProcessor.from_pretrained("openai/clip-vit-base-patch32")
# 计算图像和文本的相似度
from PIL import Image
image = Image.open("example.jpg")
inputs = processor(
text=["一只猫", "一只狗", "一辆汽车"],
images=image,
return_tensors="pt",
padding=True
)
outputs = model(**inputs)
logits_per_image = outputs.logits_per_image
probs = logits_per_image.softmax(dim=1)
print(f"匹配概率：{probs}")
2.2.2 编码器-解码器架构（Encoder-Decoder）
以BLIP-2为代表，引入了Q-Former作为桥梁：
[图像] -> Frozen Image Encoder -> 视觉特征
|
Q-Former（可学习的查询）
|
[文本] -> Frozen LLM -> 文本生成
Q-Former包含一组可学习的查询向量，通过交叉注意力机制从图像特征中提取与语言相关的信息，然后传递给大语言模型。
2.2.3 视觉编码器+投影层+LLM架构
这是当前最主流的VLM架构，以LLaVA为代表：
[图像] -> Vision Encoder (ViT) -> 视觉Token
|
Projection Layer (MLP)
|
[文本] -> Tokenizer -> 文本Token
|
[视觉Token + 文本Token] -> LLM -> 生成回答
这种架构的核心思想是：将图像通过视觉编码器转换为一系列"视觉Token"，再通过投影层将这些Token映射到LLM的嵌入空间，然后与文本Token拼接后输入LLM进行处理。
# 简化的VLM前向传播
import torch
import torch.nn as nn
class SimpleVLM(nn.Module):
def __init__(self, vision_encoder, projection, llm):
super().__init__()
self.vision_encoder = vision_encoder # 如ViT
self.projection = projection # MLP投影层
self.llm = llm # 大语言模型
def forward(self, image, text_tokens):
# 1. 图像编码
vision_features = self.vision_encoder(image)
# 2. 投影到LLM空间
vision_tokens = self.projection(vision_features)
# 3. 拼接视觉和文本Token
combined = torch.cat([vision_tokens, text_tokens], dim=1)
# 4. LLM处理
output = self.llm(combined)
return output
2.3 Vision Transformer（ViT）详解
ViT是当前VLM中最常用的视觉编码器。它的核心思想是将图像分割成固定大小的Patch，然后像处理文本Token一样处理这些Patch。
2.3.1 图像分Patch
一张224×224的图像，按16×16的Patch大小分割，会得到(224/16)×(224/16) = 14×14 = 196个Patch。每个Patch被展平为一个向量。
import torch
import torch.nn as nn
class PatchEmbedding(nn.Module):
def __init__(self, img_size=224, patch_size=16, in_channels=3, embed_dim=768):
super().__init__()
self.patch_size = patch_size
self.num_patches = (img_size // patch_size) ** 2
# 使用卷积实现Patch分割和嵌入
self.projection = nn.Conv2d(
in_channels, embed_dim,
kernel_size=patch_size,
stride=patch_size
)
def forward(self, x):
# x: (batch, channels, height, width)
x = self.projection(x) # (batch, embed_dim, h/p, w/p)
x = x.flatten(2) # (batch, embed_dim, num_patches)
x = x.transpose(1, 2) # (batch, num_patches, embed_dim)
return x
# 测试
patch_embed = PatchEmbedding()
dummy_image = torch.randn(1, 3, 224, 224)
patches = patch_embed(dummy_image)
print(f"Patch形状：{patches.shape}") # (1, 196, 768)
2.3.2 位置编码
由于Transformer本身不包含位置信息，需要为每个Patch添加位置编码：
class ViTWithPosition(nn.Module):
def __init__(self, num_patches=196, embed_dim=768):
super().__init__()
# 可学习的位置编码
self.position_embedding = nn.Parameter(
torch.randn(1, num_patches + 1, embed_dim) # +1 for CLS token
)
self.cls_token = nn.Parameter(torch.randn(1, 1, embed_dim))
def forward(self, patch_embeddings):
batch_size = patch_embeddings.shape[0]
# 添加CLS token
cls_tokens = self.cls_token.expand(batch_size, -1, -1)
x = torch.cat([cls_tokens, patch_embeddings], dim=1)
# 添加位置编码
x = x + self.position_embedding
return x
2.3.3 Transformer编码器
经过位置编码后的Patch序列输入标准的Transformer编码器：
class TransformerBlock(nn.Module):
def __init__(self, embed_dim=768, num_heads=12, mlp_ratio=4.0):
super().__init__()
self.norm1 = nn.LayerNorm(embed_dim)
self.attn = nn.MultiheadAttention(embed_dim, num_heads, batch_first=True)
self.norm2 = nn.LayerNorm(embed_dim)
self.mlp = nn.Sequential(
nn.Linear(embed_dim, int(embed_dim * mlp_ratio)),
nn.GELU(),
nn.Linear(int(embed_dim * mlp_ratio), embed_dim)
)
def forward(self, x):
# 多头自注意力
residual = x
x = self.norm1(x)
x, _ = self.attn(x, x, x)
x = x + residual
# 前馈网络
residual = x
x = self.norm2(x)
x = self.mlp(x)
x = x + residual
return x
2.4 训练策略：三阶段训练
以LLaVA为代表的VLM通常采用三阶段训练策略：
第一阶段：预训练视觉编码器
使用大规模图文对数据训练ViT，通常采用CLIP的预训练权重。这一步让模型学会提取有意义的视觉特征。数据集通常包含数亿个图文对，如LAION-5B数据集。
第二阶段：特征对齐预训练
冻结视觉编码器和LLM，只训练中间的投影层。使用大量图文对数据，让投影层学会将视觉特征映射到LLM能理解的空间。
# 第二阶段训练的关键代码
def train_alignment(vlm, dataloader, epochs=1):
# 冻结视觉编码器和LLM
for param in vlm.vision_encoder.parameters():
param.requires_grad = False
for param in vlm.llm.parameters():
param.requires_grad = False
# 只训练投影层
optimizer = torch.optim.AdamW(
vlm.projection.parameters(),
lr=1e-3,
weight_decay=0.01
)
for epoch in range(epochs):
for images, texts in dataloader:
vision_tokens = vlm.vision_encoder(images)
projected = vlm.projection(vision_tokens)
text_tokens = vlm.tokenizer(texts)
combined = torch.cat([projected, text_tokens], dim=1)
loss = vlm.llm(combined, labels=text_tokens)
optimizer.zero_grad()
loss.backward()
optimizer.step()
第三阶段：指令微调
解冻LLM（可选解冻视觉编码器），使用高质量的指令跟随数据进行微调。这一步让模型学会按照指令格式回答关于图像的问题。
# 第三阶段训练的关键代码
def train_instruction_tuning(vlm, dataloader, epochs=3):
# 解冻LLM
for param in vlm.llm.parameters():
param.requires_grad = True
# 视觉编码器可以选择性解冻
for param in vlm.vision_encoder.parameters():
param.requires_grad = False # 或True进行全量微调
optimizer = torch.optim.AdamW(
filter(lambda p: p.requires_grad, vlm.parameters()),
lr=2e-5,
weight_decay=0.01
)
for epoch in range(epochs):
for batch in dataloader:
loss = vlm(
images=batch['images'],
input_ids=batch['input_ids'],
labels=batch['labels']
)
optimizer.zero_grad()
loss.backward()
optimizer.step()
2.5 关键VLM模型对比
| 模型 | 发布时间 | 架构特点 | 参数量 | 开源 |
|---|---|---|---|---|
| CLIP | 2021.01 | 双塔对比学习 | 400M | 是 |
| BLIP-2 | 2023.01 | Q-Former桥接 | 12B | 是 |
| LLaVA | 2023.04 | MLP投影层 | 7B/13B | 是 |
| GPT-4V | 2023.09 | 未公开 | 未公开 | 否 |
| Qwen-VL | 2023.08 | 交叉注意力 | 10B | 是 |
| InternVL | 2024.01 | 动态分辨率 | 6B-26B | 是 |
2.6 本章小结
本章我们学习了VLM的基本概念、架构演进、核心技术原理。关键要点：
- VLM的核心是让模型同时理解视觉和语言信息
- 当前主流架构是"视觉编码器+投影层+LLM"的三段式结构
- ViT将图像分割为Patch，像处理Token一样处理
- 训练通常分为预训练、对齐、指令微调三个阶段
- 理解这些原理有助于我们更好地使用和调优VLM
第三章：GPT-4V/GPT-4o 多模态API实战
3.1 GPT-4V 与 GPT-4o 简介
GPT-4V（GPT-4 with Vision）是OpenAI在2023年9月发布的多模态大语言模型，它在GPT-4的基础上增加了视觉理解能力。用户可以向模型发送图片，并就图片内容进行对话和推理。
GPT-4o（"o"代表"omni"，全能）是2024年5月发布的升级版本，支持文本、图像、音频三种模态的输入和输出，实现了真正的多模态交互。相比GPT-4V，GPT-4o具有以下优势：
- 更快的响应速度：API延迟降低约50%
- 更低的成本：价格比GPT-4V便宜约50%
- 原生多模态：不再需要将图像转换为特定格式
- 音频支持：可以直接处理语音输入和输出
3.2 API接入准备
使用GPT-4V/GPT-4o的多模态能力，需要通过OpenAI的Chat Completions API。以下是接入步骤：
步骤一：获取API Key
前往OpenAI官网注册账号，在API Keys页面创建一个新的API Key。
步骤二：安装SDK
pip install openai
步骤三：基础配置
from openai import OpenAI
client = OpenAI(
api_key="your-api-key-here",
base_url="https://api.openai.com/v1" # 可替换为兼容的第三方服务
)
3.3 图像输入方式
GPT-4V/GPT-4o支持两种图像输入方式：
3.3.1 URL方式
import base64
from openai import OpenAI
client = OpenAI(api_key="your-api-key")
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{
"role": "user",
"content": [
{
"type": "text",
"text": "请描述这张图片中的内容"
},
{
"type": "image_url",
"image_url": {
"url": "https://example.com/photo.jpg",
"detail": "high" # low, high, auto
}
}
]
}
],
max_tokens=1000
)
print(response.choices[0].message.content)
3.3.2 Base64编码方式
import base64
from pathlib import Path
def encode_image_to_base64(image_path):
"""将本地图片编码为base64字符串"""
with open(image_path, "rb") as f:
return base64.b64encode(f.read()).decode("utf-8")
def analyze_local_image(image_path, question):
"""分析本地图片"""
base64_image = encode_image_to_base64(image_path)
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{
"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_local_image(
"my_photo.jpg",
"这张图片中有什么？请详细描述。"
)
print(result)
3.4 多图分析
GPT-4o支持在一次请求中发送多张图片，进行对比分析：
def compare_images(image_paths, question):
"""对比多张图片"""
content = [{"type": "text", "text": question}]
for i, path in enumerate(image_paths):
base64_image = encode_image_to_base64(path)
content.append({
"type": "text",
"text": f"图片{i+1}："
})
content.append({
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{base64_image}",
"detail": "high"
}
})
response = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": content}],
max_tokens=2000
)
return response.choices[0].message.content
# 使用示例
result = compare_images(
["before.jpg", "after.jpg"],
"请对比这两张图片，说明它们之间的差异。"
)
print(result)
3.5 图文混合对话
在实际应用中，往往需要在多轮对话中交替发送图片和文本：
class MultimodalChatbot:
def __init__(self, model="gpt-4o"):
self.client = OpenAI(api_key="your-api-key")
self.model = model
self.history = []
def chat(self, text, image_path=None):
"""发送消息，可选附带图片"""
content = []
if image_path:
base64_image = encode_image_to_base64(image_path)
content.append({
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{base64_image}",
"detail": "high"
}
})
content.append({"type": "text", "text": text})
self.history.append({"role": "user", "content": content})
response = self.client.chat.completions.create(
model=self.model,
messages=self.history,
max_tokens=2000
)
assistant_message = response.choices[0].message.content
self.history.append({"role": "assistant", "content": assistant_message})
return assistant_message
def reset(self):
"""重置对话历史"""
self.history = []
# 使用示例
bot = MultimodalChatbot()
print(bot.chat("请描述这张图片", "photo1.jpg"))
print(bot.chat("图片中的人在做什么？"))
print(bot.chat("这个场景可能发生在哪里？"))
3.6 结构化输出
GPT-4o支持返回结构化的JSON数据，非常适合数据提取场景：
import json
def extract_info_from_image(image_path, schema_description):
"""从图片中提取结构化信息"""
base64_image = encode_image_to_base64(image_path)
prompt = f"""请分析这张图片，按照以下JSON格式提取信息：
{schema_description}
请只返回JSON格式的结果，不要添加其他文字。"""
response = 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_image}",
"detail": "high"
}
}
]
}
],
response_format={"type": "json_object"},
max_tokens=1000
)
return json.loads(response.choices[0].message.content)
# 使用示例：提取名片信息
schema = """{
"name": "姓名",
"company": "公司名称",
"title": "职位",
"phone": "电话号码",
"email": "邮箱地址",
"address": "地址"
}"""
result = extract_info_from_image("business_card.jpg", schema)
print(json.dumps(result, ensure_ascii=False, indent=2))
3.7 错误处理与最佳实践
import time
from openai import OpenAI, APIError, RateLimitError, APIConnectionError
class RobustMultimodalClient:
def __init__(self, api_key, model="gpt-4o", max_retries=3):
self.client = OpenAI(api_key=api_key)
self.model = model
self.max_retries = max_retries
def analyze_image(self, image_path, prompt, detail="high"):
"""带重试机制的图片分析"""
base64_image = encode_image_to_base64(image_path)
for attempt in range(self.max_retries):
try:
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_image}",
"detail": detail
}
}
]
}],
max_tokens=2000,
temperature=0.7
)
return response.choices[0].message.content
except RateLimitError:
wait_time = 2 ** attempt # 指数退避
print(f"请求频率超限，等待{wait_time}秒后重试...")
time.sleep(wait_time)
except APIConnectionError:
print(f"网络连接失败，重试中...")
time.sleep(2)
except APIError as e:
print(f"API错误：{e}")
raise
raise Exception(f"重试{self.max_retries}次后仍然失败")
def batch_analyze(self, image_paths, prompt):
"""批量分析多张图片"""
results = []
for i, path in enumerate(image_paths):
print(f"正在处理第{i+1}/{len(image_paths)}张图片...")
result = self.analyze_image(path, prompt)
results.append({
"path": path,
"analysis": result
})
time.sleep(1) # 避免触发频率限制
return results
# 使用示例
client = RobustMultimodalClient("your-api-key")
result = client.analyze_image("photo.jpg", "请描述图片内容")
print(result)
3.8 成本控制
使用GPT-4o进行图像分析时，需要注意成本控制：
def estimate_cost(num_images, avg_tokens_per_image=1000, output_tokens=500):
"""估算API调用成本"""
# GPT-4o 定价（2024年）
input_price_per_1k = 0.0025 # $2.50 / 1M tokens
output_price_per_1k = 0.01 # $10.00 / 1M tokens
# 图像Token计算：high detail约85 tokens基础 + 每个512x512块170 tokens
# 一张典型图片约 85 + 170*4 = 765 tokens（low detail固定85 tokens）
input_cost = num_images * avg_tokens_per_image / 1000 * input_price_per_1k
output_cost = num_images * output_tokens / 1000 * output_price_per_1k
total_cost = input_cost + output_cost
print(f"图片数量：{num_images}")
print(f"预计输入成本：${input_cost:.4f}")
print(f"预计输出成本：${output_cost:.4f}")
print(f"总成本：${total_cost:.4f}")
return total_cost
# 估算100张图片的成本
estimate_cost(100)
3.9 本章小结
本章我们学习了如何使用GPT-4V/GPT-4o的多模态API：
- 支持URL和Base64两种图像输入方式
- 可以在单次请求中发送多张图片
- 支持多轮图文混合对话
- 可以返回结构化的JSON数据
- 需要注意错误处理、重试机制和成本控制
第四章：LLaVA 本地部署与使用
4.1 LLaVA简介
LLaVA（Large Language and Vision Assistant）是由威斯康星大学麦迪逊分校和微软研究院在2023年开源的多模态大语言模型。它的核心特点是：
- 简单高效的架构：使用MLP投影层连接视觉编码器和LLM
- 完全开源：模型权重、训练代码、数据集全部开放
- 高性能：在多个基准测试上表现优异
- 易于部署：支持多种推理框架
LLaVA的最新版本包括LLaVA-1.5、LLaVA-1.6（LLaVA-NeXT）和LLaVA-OneVision，在性能上不断提升。
4.2 环境准备
4.2.1 硬件要求
| 模型版本 | 最低GPU显存 | 推荐GPU | 说明 |
|---|---|---|---|
| LLaVA-7B | 10GB | RTX 3090/4090 | INT4量化 |
| LLaVA-7B | 16GB | RTX 4090 | FP16精度 |
| LLaVA-13B | 20GB | A5000 | INT4量化 |
| LLaVA-13B | 32GB | A6000 | FP16精度 |
4.2.2 安装依赖
# 创建虚拟环境
conda create -n llava python=3.10 -y
conda activate llava
# 克隆LLaVA仓库
git clone https://github.com/haotian-liu/LLaVA.git
cd LLaVA
# 安装依赖
pip install --upgrade pip
pip install -e .
pip install -e ".[train]"
# 或者使用pip直接安装核心依赖
pip install torch torchvision
pip install transformers>=4.36.0
pip install accelerate
pip install bitsandbytes
pip install peft
pip install Pillow
4.3 使用Transformers快速上手
最简单的方式是通过Hugging Face Transformers库直接加载LLaVA模型：
from transformers import LlavaForConditionalGeneration, AutoProcessor
from PIL import Image
import torch
# 加载模型和处理器
model_id = "llava-hf/llava-1.5-7b-hf"
processor = AutoProcessor.from_pretrained(model_id)
model = LlavaForConditionalGeneration.from_pretrained(
model_id,
torch_dtype=torch.float16,
device_map="auto"
)
# 准备输入
image = Image.open("test_image.jpg")
conversation = [
{
"role": "user",
"content": [
{"type": "image"},
{"type": "text", "text": "请描述这张图片的内容。"},
],
},
]
prompt = processor.apply_chat_template(conversation, add_generation_prompt=True)
inputs = processor(images=image, text=prompt, return_tensors="pt").to(model.device)
# 生成回答
output = model.generate(
**inputs,
max_new_tokens=512,
do_sample=True,
temperature=0.7
)
result = processor.decode(output[0], skip_special_tokens=True)
print(result)
4.4 使用LLaVA原生代码
LLaVA项目提供了自己的推理代码，功能更丰富：
# 使用LLaVA原生API
from llava.model.builder import load_pretrained_model
from llava.mm_utils import get_model_name_from_path
from llava.eval.run_llava import eval_model
# 加载模型
model_path = "liuhaotian/llava-v1.5-7b"
tokenizer, model, image_processor, context_len = load_pretrained_model(
model_path=model_path,
model_base=None,
model_name=get_model_name_from_path(model_path)
)
# 评估图片
args = type('Args', (), {
"model_path": model_path,
"model_base": None,
"model_name": get_model_name_from_path(model_path),
"query": "请详细描述这张图片中发生了什么",
"image_file": "test_image.jpg",
"sep": ",",
"temperature": 0,
"top_p": None,
"num_beams": 1,
"max_new_tokens": 512
})()
eval_model(args)
4.5 量化部署
对于显存有限的设备，可以使用量化技术降低显存需求：
4.5.1 INT8量化
from transformers import LlavaForConditionalGeneration, AutoProcessor, BitsAndBytesConfig
import torch
# INT8量化配置
quantization_config = BitsAndBytesConfig(
load_in_8bit=True,
bnb_8bit_compute_dtype=torch.float16
)
model = LlavaForConditionalGeneration.from_pretrained(
"llava-hf/llava-1.5-7b-hf",
quantization_config=quantization_config,
device_map="auto"
)
4.5.2 INT4量化
# INT4量化配置（GPTQ）
quantization_config = BitsAndBytesConfig(
load_in_4bit=True,
bnb_4bit_compute_dtype=torch.float16,
bnb_4bit_use_double_quant=True,
bnb_4bit_quant_type="nf4"
)
model = LlavaForConditionalGeneration.from_pretrained(
"llava-hf/llava-1.5-7b-hf",
quantization_config=quantization_config,
device_map="auto"
)
4.6 搭建Web界面
使用Gradio快速搭建一个交互式的Web界面：
import gradio as gr
from transformers import LlavaForConditionalGeneration, AutoProcessor
from PIL import Image
import torch
# 加载模型
model_id = "llava-hf/llava-1.5-7b-hf"
processor = AutoProcessor.from_pretrained(model_id)
model = LlavaForConditionalGeneration.from_pretrained(
model_id,
torch_dtype=torch.float16,
device_map="auto"
)
def chat_with_image(image, question, history):
"""与图片进行对话"""
if image is None:
return "请先上传一张图片"
conversation = [
{
"role": "user",
"content": [
{"type": "image"},
{"type": "text", "text": question},
],
},
]
prompt = processor.apply_chat_template(conversation, add_generation_prompt=True)
inputs = processor(images=image, text=prompt, return_tensors="pt").to(model.device)
output = model.generate(
**inputs,
max_new_tokens=512,
do_sample=True,
temperature=0.7
)
result = processor.decode(output[0], skip_special_tokens=True)
# 提取助手的回答
if "ASSISTANT:" in result:
result = result.split("ASSISTANT:")[-1].strip()
return result
# 创建Gradio界面
demo = gr.Interface(
fn=chat_with_image,
inputs=[
gr.Image(type="pil", label="上传图片"),
gr.Textbox(label="请输入问题", value="请描述这张图片"),
gr.State([])
],
outputs=gr.Textbox(label="AI回答"),
title="LLaVA 多模态对话",
description="上传图片，然后向AI提问关于图片的问题"
)
demo.launch(server_name="0.0.0.0", server_port=7860)
4.7 批量处理
对于需要处理大量图片的场景：
import os
import json
from tqdm import tqdm
def batch_process_images(image_dir, questions, output_file):
"""批量处理图片并保存结果"""
results = []
image_files = [
f for f in os.listdir(image_dir)
if f.lower().endswith(('.jpg', '.jpeg', '.png', '.bmp'))
]
for image_file in tqdm(image_files, desc="处理图片"):
image_path = os.path.join(image_dir, image_file)
image = Image.open(image_path)
for question in questions:
conversation = [
{
"role": "user",
"content": [
{"type": "image"},
{"type": "text", "text": question},
],
},
]
prompt = processor.apply_chat_template(conversation, add_generation_prompt=True)
inputs = processor(images=image, text=prompt, return_tensors="pt").to(model.device)
output = model.generate(**inputs, max_new_tokens=256)
result = processor.decode(output[0], skip_special_tokens=True)
results.append({
"image": image_file,
"question": question,
"answer": result
})
# 保存结果
with open(output_file, 'w', encoding='utf-8') as f:
json.dump(results, f, ensure_ascii=False, indent=2)
print(f"处理完成，结果已保存到 {output_file}")
# 使用示例
batch_process_images(
image_dir="./images",
questions=["请描述这张图片", "图片中有几个人？"],
output_file="results.json"
)
4.8 常见问题排查
问题1：CUDA显存不足
# 解决方案1：使用量化
# 解决方案2：减小图片分辨率
# 解决方案3：使用device_map自动分配
# 在代码中监控显存
import torch
print(f"已分配显存：{torch.cuda.memory_allocated() / 1024**3:.2f} GB")
print(f"缓存显存：{torch.cuda.memory_reserved() / 1024**3:.2f} GB")
问题2：模型下载失败
# 使用镜像源
export HF_ENDPOINT=https://hf-mirror.com
# 或者使用huggingface-cli下载
pip install huggingface_hub
huggingface-cli download llava-hf/llava-1.5-7b-hf --local-dir ./llava-1.5-7b
问题3：推理速度慢
# 使用flash attention加速
pip install flash-attn --no-build-isolation
# 在加载模型时启用
model = LlavaForConditionalGeneration.from_pretrained(
model_id,
torch_dtype=torch.float16,
device_map="auto",
attn_implementation="flash_attention_2"
)
4.9 本章小结
本章我们学习了LLaVA的本地部署和使用：
- LLaVA是简单高效的开源VLM，使用MLP投影层架构
- 支持通过Transformers和原生代码两种方式使用
- 可以通过INT4/INT8量化降低显存需求
- 使用Gradio可以快速搭建Web交互界面
- 支持批量处理和结果导出
第五章：Qwen-VL 通义千问视觉模型实战
5.1 Qwen-VL简介
Qwen-VL（通义千问视觉语言模型）是阿里巴巴达摩院开发的多模态大语言模型。它是基于Qwen-7B语言模型，通过视觉编码器和跨模态融合模块实现图像理解能力。
Qwen-VL的主要特点包括：
- 中文原生支持：针对中文场景进行了大量优化
- 高分辨率输入：支持448×448分辨率的图像输入
- 多图理解：支持单轮对话中处理多张图片
- 视觉定位：可以在图片中定位特定物体
- OCR能力强：对中文文档和文字的识别能力出色
Qwen-VL的版本演进：
- Qwen-VL（2023年8月）：首个版本，7B参数
- Qwen-VL-Chat（2023年8月）：对话微调版本
- Qwen2-VL（2024年8月）：大幅升级版本，支持视频理解
- Qwen2.5-VL（2025年1月）：最新版本，性能进一步提升
5.2 通过API使用Qwen-VL
最简单的方式是通过阿里云的DashScope API使用：
pip install dashscope
import dashscope
from dashscope import MultiModalConversation
# 设置API Key
dashscope.api_key = "your-dashscope-api-key"
# 单图分析
def analyze_image_with_qwen(image_path, question):
"""使用Qwen-VL分析图片"""
messages = [
{
"role": "user",
"content": [
{"image": image_path},
{"text": question}
]
}
]
response = MultiModalConversation.call(
model="qwen-vl-plus",
messages=messages
)
return response.output.choices[0].message.content[0]["text"]
# 使用示例
result = analyze_image_with_qwen(
"https://example.com/photo.jpg",
"请描述这张图片的内容"
)
print(result)
5.3 多图对话
Qwen-VL支持在一次对话中处理多张图片：
def multi_image_conversation(image_paths, question):
"""多图对话"""
content = []
for path in image_paths:
content.append({"image": path})
content.append({"text": question})
messages = [
{"role": "user", "content": content}
]
response = MultiModalConversation.call(
model="qwen-vl-plus",
messages=messages
)
return response.output.choices[0].message.content[0]["text"]
# 使用示例
result = multi_image_conversation(
["image1.jpg", "image2.jpg", "image3.jpg"],
"这三张图片有什么共同点和不同点？"
)
print(result)
5.4 本地部署Qwen-VL
对于需要本地运行的场景，可以使用Transformers加载Qwen-VL：
from transformers import AutoModelForCausalLM, AutoTokenizer
from PIL import Image
import torch
# 加载模型
model_path = "Qwen/Qwen-VL-Chat"
tokenizer = AutoTokenizer.from_pretrained(model_path, trust_remote_code=True)
model = AutoModelForCausalLM.from_pretrained(
model_path,
device_map="auto",
torch_dtype=torch.bfloat16,
trust_remote_code=True
).eval()
# 单图对话
image = Image.open("test.jpg")
query = tokenizer.from_list_format([
{"image": "test.jpg"},
{"text": "请描述这张图片"},
])
response, history = model.chat(tokenizer, query, history=None)
print(response)
5.5 多轮视觉对话
def multi_turn_visual_chat(model, tokenizer, image_path):
"""多轮视觉对话"""
# 第一轮：带图片的对话
query1 = tokenizer.from_list_format([
{"image": image_path},
{"text": "请详细描述这张图片的内容"},
])
response1, history = model.chat(tokenizer, query1, history=None)
print(f"问：请详细描述这张图片的内容")
print(f"答：{response1}")
# 第二轮：追问（无需重复发送图片）
query2 = "图片中有几个人？他们在做什么？"
response2, history = model.chat(tokenizer, query2, history=history)
print(f"问：{query2}")
print(f"答：{response2}")
# 第三轮：继续追问
query3 = "这个场景可能是在什么地方？"
response3, history = model.chat(tokenizer, query3, history=history)
print(f"问：{query3}")
print(f"答：{response3}")
return history
# 使用示例
history = multi_turn_visual_chat(model, tokenizer, "street_photo.jpg")
5.6 OCR与文档理解
Qwen-VL在OCR方面表现突出，特别适合中文文档处理：
def ocr_with_qwen(image_path):
"""使用Qwen-VL进行OCR识别"""
query = tokenizer.from_list_format([
{"image": image_path},
{"text": "请识别图片中的所有文字，按照原始排版输出"},
])
response, _ = model.chat(tokenizer, query, history=None)
return response
def extract_table_from_image(image_path):
"""从图片中提取表格数据"""
query = tokenizer.from_list_format([
{"image": image_path},
{"text": "请识别图片中的表格，以Markdown格式输出"},
])
response, _ = model.chat(tokenizer, query, history=None)
return response
def understand_receipt(image_path):
"""理解发票/收据内容"""
query = tokenizer.from_list_format([
{"image": image_path},
{"text": "请识别这张发票/收据上的所有信息，包括：商户名称、日期、商品明细、金额等，以JSON格式输出"},
])
response, _ = model.chat(tokenizer, query, history=None)
return response
# 使用示例
print(ocr_with_qwen("document.jpg"))
print(extract_table_from_image("table.jpg"))
print(understand_receipt("receipt.jpg"))
5.7 视觉定位功能
Qwen-VL支持在图片中定位特定物体：
def locate_objects(image_path, target_object):
"""在图片中定位指定物体"""
query = tokenizer.from_list_format([
{"image": image_path},
{"text": f"请在图片中找到所有的{target_object}，并用边界框标注它们的位置"},
])
response, _ = model.chat(tokenizer, query, history=None)
return response
# 使用示例
result = locate_objects("street.jpg", "汽车")
print(result)
5.8 Qwen-VL vs GPT-4o 对比
在中文场景下，Qwen-VL有独特的优势：
| 特性 | Qwen-VL | GPT-4o |
|---|---|---|
| 中文OCR | 优秀 | 良好 |
| 英文OCR | 良好 | 优秀 |
| 价格 | 较低 | 较高 |
| 本地部署 | 支持 | 不支持 |
| 视频理解 | Qwen2-VL支持 | 支持 |
| 视觉定位 | 支持 | 有限支持 |
5.9 本章小结
本章我们学习了Qwen-VL的使用方法：
- Qwen-VL是中文友好的开源VLM，OCR能力突出
- 可以通过DashScope API快速使用
- 支持本地部署和多轮视觉对话
- 特别适合中文文档理解和OCR场景
- Qwen2-VL进一步支持了视频理解能力
第六章：图像理解应用开发
6.1 图像理解的核心任务
图像理解是多模态AI最基础也最广泛的应用方向。核心任务包括：
- 图像分类：判断图像属于哪个类别
- 目标检测：找出图像中的物体及其位置
- 图像描述：用自然语言描述图像内容
- 视觉问答：回答关于图像的问题
- 场景理解：理解图像中的场景、关系和语义
- OCR：识别图像中的文字
6.2 构建图像描述应用
import gradio as gr
from openai import OpenAI
import base64
client = OpenAI(api_key="your-api-key")
def generate_image_caption(image, style="详细"):
"""生成图片描述"""
# 将图片转为base64
import io
buffer = io.BytesIO()
image.save(buffer, format="JPEG")
base64_image = base64.b64encode(buffer.getvalue()).decode("utf-8")
style_prompts = {
"详细": "请详细描述这张图片的内容，包括场景、物体、颜色、动作等所有细节。",
"简洁": "用一句话简洁地描述这张图片。",
"文艺": "用富有诗意和文学性的语言描述这张图片。",
"技术": "从技术角度分析这张图片，包括构图、光线、色彩等。",
"儿童": "用简单易懂的语言，像给小朋友讲故事一样描述这张图片。"
}
prompt = style_prompts.get(style, style_prompts["详细"])
response = 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_image}",
"detail": "high"
}
}
]
}],
max_tokens=500,
temperature=0.7
)
return response.choices[0].message.content
# Gradio界面
with gr.Blocks(title="智能图像描述生成器") as demo:
gr.Markdown("# 智能图像描述生成器")
gr.Markdown("上传图片，选择描述风格，AI将为你生成图片描述。")
with gr.Row():
with gr.Column():
image_input = gr.Image(type="pil", label="上传图片")
style_input = gr.Radio(
choices=["详细", "简洁", "文艺", "技术", "儿童"],
value="详细",
label="描述风格"
)
generate_btn = gr.Button("生成描述", variant="primary")
with gr.Column():
output = gr.Textbox(label="生成的描述", lines=8)
generate_btn.click(
fn=generate_image_caption,
inputs=[image_input, style_input],
outputs=output
)
demo.launch()
6.3 构建视觉问答系统
class VisualQASystem:
"""视觉问答系统"""
def __init__(self, api_key):
self.client = OpenAI(api_key=api_key)
self.conversation_history = []
def ask(self, image, question):
"""关于图片提问"""
import io
import base64
buffer = io.BytesIO()
image.save(buffer, format="JPEG")
base64_image = base64.b64encode(buffer.getvalue()).decode("utf-8")
# 如果是新图片，重置历史
if not self.conversation_history:
self.conversation_history.append({
"role": "user",
"content": [
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{base64_image}",
"detail": "high"
}
},
{"type": "text", "text": question}
]
})
else:
self.conversation_history.append({
"role": "user",
"content": question
})
response = self.client.chat.completions.create(
model="gpt-4o",
messages=self.conversation_history,
max_tokens=1000,
temperature=0.5
)
answer = response.choices[0].message.content
self.conversation_history.append({
"role": "assistant",
"content": answer
})
return answer
def reset(self):
"""重置对话"""
self.conversation_history = []
# 使用示例
vqa = VisualQASystem("your-api-key")
image = Image.open("scene.jpg")
print(vqa.ask(image, "这张图片中有什么？"))
print(vqa.ask(image, "有几个人？"))
print(vqa.ask(image, "他们可能在做什么？"))
6.4 构建智能相册分类系统
import os
import json
import shutil
from datetime import datetime
class SmartPhotoOrganizer:
"""智能相册分类系统"""
CATEGORIES = [
"风景", "人物", "美食", "动物", "建筑",
"交通工具", "文档/截图", "艺术/设计", "运动/户外", "其他"
]
def __init__(self, api_key):
self.client = OpenAI(api_key=api_key)
def classify_photo(self, image_path):
"""对单张照片进行分类"""
with open(image_path, "rb") as f:
base64_image = base64.b64encode(f.read()).decode("utf-8")
categories_str = "、".join(self.CATEGORIES)
response = self.client.chat.completions.create(
model="gpt-4o",
messages=[{
"role": "user",
"content": [
{
"type": "text",
"text": f"请将这张图片分类到以下类别之一：{categories_str}\n只返回类别名称，不要添加其他文字。"
},
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{base64_image}",
"detail": "low"
}
}
]
}],
max_tokens=50,
temperature=0
)
category = response.choices[0].message.content.strip()
return category
def organize_photos(self, source_dir, output_dir):
"""整理照片到分类文件夹"""
# 创建输出目录
for category in self.CATEGORIES:
os.makedirs(os.path.join(output_dir, category), exist_ok=True)
# 遍历所有图片
image_extensions = ('.jpg', '.jpeg', '.png', '.bmp', '.webp')
image_files = [
f for f in os.listdir(source_dir)
if f.lower().endswith(image_extensions)
]
results = []
for image_file in image_files:
source_path = os.path.join(source_dir, image_file)
try:
category = self.classify_photo(source_path)
# 如果分类不在预设类别中，归为"其他"
if category not in self.CATEGORIES:
category = "其他"
# 复制文件到对应目录
dest_path = os.path.join(output_dir, category, image_file)
shutil.copy2(source_path, dest_path)
results.append({
"file": image_file,
"category": category
})
print(f"已分类：{image_file} -> {category}")
except Exception as e:
print(f"处理失败：{image_file}, 错误：{e}")
# 保存分类结果
with open(os.path.join(output_dir, "classification.json"), 'w', encoding='utf-8') as f:
json.dump(results, f, ensure_ascii=False, indent=2)
return results
# 使用示例
organizer = SmartPhotoOrganizer("your-api-key")
results = organizer.organize_photos("./my_photos", "./organized_photos")
# 统计分类结果
from collections import Counter
category_counts = Counter(r["category"] for r in results)
for category, count in category_counts.most_common():
print(f"{category}: {count}张")
6.5 构建图片内容审核系统
class ImageContentModerator:
"""图片内容审核系统"""
def __init__(self, api_key):
self.client = OpenAI(api_key=api_key)
def moderate_image(self, image_path):
"""审核图片内容"""
with open(image_path, "rb") as f:
base64_image = base64.b64encode(f.read()).decode("utf-8")
prompt = """请分析这张图片的内容，判断是否存在以下问题：
1. 暴力/血腥内容
2. 色情/低俗内容
3. 仇恨/歧视内容
4. 危险行为
5. 个人信息泄露（如身份证、银行卡号等）
6. 虚假信息/误导性内容
请返回JSON格式的分析结果：
{
"safe": true/false,
"issues": ["存在的问题列表"],
"confidence": 0.0-1.0,
"description": "简要描述图片内容"
}"""
response = self.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_image}",
"detail": "high"
}
}
]
}],
response_format={"type": "json_object"},
max_tokens=500,
temperature=0
)
return json.loads(response.choices[0].message.content)
def batch_moderate(self, image_paths):
"""批量审核"""
results = []
for path in image_paths:
try:
result = self.moderate_image(path)
result["file"] = path
results.append(result)
except Exception as e:
results.append({
"file": path,
"safe": None,
"error": str(e)
})
return results
# 使用示例
moderator = ImageContentModerator("your-api-key")
result = moderator.moderate_image("test_image.jpg")
print(json.dumps(result, ensure_ascii=False, indent=2))
6.6 构建图片相似度搜索系统
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity
class ImageSearchEngine:
"""基于语义的图片搜索引擎"""
def __init__(self, api_key):
self.client = OpenAI(api_key=api_key)
self.image_descriptions = {}
self.image_embeddings = {}
def index_image(self, image_path):
"""为图片建立索引"""
with open(image_path, "rb") as f:
base64_image = base64.b64encode(f.read()).decode("utf-8")
# 生成图片描述
response = self.client.chat.completions.create(
model="gpt-4o",
messages=[{
"role": "user",
"content": [
{"type": "text", "text": "请用5-10个关键词描述这张图片的主要内容，用逗号分隔"},
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{base64_image}",
"detail": "low"
}
}
]
}],
max_tokens=100
)
description = response.choices[0].message.content
self.image_descriptions[image_path] = description
# 生成文本嵌入
embedding_response = self.client.embeddings.create(
model="text-embedding-3-small",
input=description
)
self.image_embeddings[image_path] = np.array(
embedding_response.data[0].embedding
)
return description
def search(self, query, top_k=5):
"""根据文本查询搜索图片"""
# 生成查询嵌入
embedding_response = self.client.embeddings.create(
model="text-embedding-3-small",
input=query
)
query_embedding = np.array(embedding_response.data[0].embedding)
# 计算相似度
similarities = {}
for path, emb in self.image_embeddings.items():
sim = cosine_similarity(
query_embedding.reshape(1, -1),
emb.reshape(1, -1)
)[0][0]
similarities[path] = sim
# 排序返回
sorted_results = sorted(
similarities.items(),
key=lambda x: x[1],
reverse=True
)[:top_k]
return [
{
"path": path,
"score": score,
"description": self.image_descriptions.get(path, "")
}
for path, score in sorted_results
]
# 使用示例
search_engine = ImageSearchEngine("your-api-key")
# 索引图片
image_dir = "./my_photos"
for filename in os.listdir(image_dir):
if filename.lower().endswith(('.jpg', '.png')):
path = os.path.join(image_dir, filename)
desc = search_engine.index_image(path)
print(f"已索引：{filename} - {desc}")
# 搜索图片
results = search_engine.search("日落时分的海边风景")
for r in results:
print(f"文件：{r['path']}, 相似度：{r['score']:.3f}")
6.7 本章小结
本章我们学习了多种图像理解应用的开发方法：
- 图像描述生成：支持多种风格的图片描述
- 视觉问答系统：支持多轮对话式的图片理解
- 智能相册分类：自动将照片按内容分类
- 内容审核系统：自动检测图片中的不当内容
- 图片搜索系统：基于语义的图片检索
第七章：语音多模态
7.1 语音多模态概述
语音是人类最自然的交流方式之一。在多模态AI中，语音模态的加入使得AI系统能够通过"听"和"说"与用户交互，大大提升了用户体验。
语音多模态的核心技术包括：
- 语音识别（ASR）：将语音转换为文字，也称为Speech-to-Text（STT）
- 语音合成（TTS）：将文字转换为语音，也称为Text-to-Speech
- 语音理解：理解语音中的情感、意图、语调等信息
- 语音生成：生成自然、富有表现力的语音
- 端到端语音模型：直接从语音输入到语音输出，如GPT-4o的语音模式
7.2 语音识别（ASR）
7.2.1 使用OpenAI Whisper
Whisper是OpenAI开源的语音识别模型，支持多种语言：
pip install openai-whisper
import whisper
# 加载模型（可选：tiny, base, small, medium, large）
model = whisper.load_model("medium")
# 识别音频文件
result = model.transcribe("audio.mp3", language="zh")
print(f"识别结果：{result['text']}")
print(f"语言：{result['language']}")
7.2.2 使用Whisper API
from openai import OpenAI
client = OpenAI(api_key="your-api-key")
def transcribe_audio(audio_path, language="zh"):
"""使用Whisper API转录音频"""
with open(audio_path, "rb") as audio_file:
result = client.audio.transcriptions.create(
model="whisper-1",
file=audio_file,
language=language,
response_format="verbose_json"
)
return {
"text": result.text,
"language": result.language,
"duration": result.duration,
"segments": [
{
"start": seg.start,
"end": seg.end,
"text": seg.text
}
for seg in result.segments
]
}
# 使用示例
result = transcribe_audio("meeting_recording.mp3")
print(f"转录文本：{result['text']}")
print(f"时长：{result['duration']:.1f}秒")
7.2.3 实时语音识别
import pyaudio
import wave
import threading
import tempfile
import os
class RealtimeTranscriber:
"""实时语音识别系统"""
def __init__(self, api_key, chunk_duration=5):
self.client = OpenAI(api_key=api_key)
self.chunk_duration = chunk_duration # 每段音频时长（秒）
self.sample_rate = 16000
self.is_recording = False
self.transcript = []
def start_recording(self):
"""开始录音和识别"""
self.is_recording = True
# 启动录音线程
record_thread = threading.Thread(target=self._record_and_transcribe)
record_thread.start()
def _record_and_transcribe(self):
"""录音并实时转录"""
p = pyaudio.PyAudio()
stream = p.open(
format=pyaudio.paInt16,
channels=1,
rate=self.sample_rate,
input=True,
frames_per_buffer=1024
)
while self.is_recording:
# 录制一段音频
frames = []
for _ in range(0, int(self.sample_rate / 1024 * self.chunk_duration)):
data = stream.read(1024)
frames.append(data)
# 保存为临时文件
with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmp:
wf = wave.open(tmp.name, 'wb')
wf.setnchannels(1)
wf.setsampwidth(p.get_sample_size(pyaudio.paInt16))
wf.setframerate(self.sample_rate)
wf.writeframes(b''.join(frames))
wf.close()
# 转录
try:
with open(tmp.name, "rb") as audio_file:
result = self.client.audio.transcriptions.create(
model="whisper-1",
file=audio_file,
language="zh"
)
text = result.text.strip()
if text:
self.transcript.append(text)
print(f"[识别] {text}")
except Exception as e:
print(f"识别错误：{e}")
os.unlink(tmp.name)
stream.stop_stream()
stream.close()
p.terminate()
def stop_recording(self):
"""停止录音"""
self.is_recording = False
def get_full_transcript(self):
"""获取完整转录文本"""
return " ".join(self.transcript)
7.3 语音合成（TTS）
7.3.1 使用OpenAI TTS API
from openai import OpenAI
client = OpenAI(api_key="your-api-key")
def text_to_speech(text, output_path, voice="alloy"):
"""将文本转换为语音"""
# 可选声音：alloy, echo, fable, onyx, nova, shimmer
response = client.audio.speech.create(
model="tts-1", # 或 tts-1-hd（高质量）
voice=voice,
input=text,
speed=1.0 # 语速：0.25-4.0
)
response.stream_to_file(output_path)
print(f"语音已保存到：{output_path}")
# 使用示例
text_to_speech(
"你好，欢迎学习多模态AI应用开发教程。本章我们将学习语音相关的技术。",
"output.mp3",
voice="nova"
)
7.3.2 多语言语音合成
def multilingual_tts(text, output_path, language_hint=None):
"""多语言语音合成"""
# OpenAI TTS能自动检测语言
# 也可以通过文本内容暗示语言
response = client.audio.speech.create(
model="tts-1-hd",
voice="nova",
input=text,
speed=1.0
)
response.stream_to_file(output_path)
# 中文
multilingual_tts("今天天气真好！", "chinese.mp3")
# 英文
multilingual_tts("What a beautiful day!", "english.mp3")
# 混合语言
multilingual_tts("这个API的performance很不错", "mixed.mp3")
7.4 语音对话系统
构建一个完整的语音对话系统：
import pyaudio
import wave
import tempfile
import os
from openai import OpenAI
class VoiceAssistant:
"""语音助手系统"""
def __init__(self, api_key):
self.client = OpenAI(api_key=api_key)
self.conversation_history = []
def listen(self, duration=5, output_path=None):
"""录音"""
if output_path is None:
output_path = tempfile.mktemp(suffix=".wav")
CHUNK = 1024
FORMAT = pyaudio.paInt16
CHANNELS = 1
RATE = 16000
p = pyaudio.PyAudio()
stream = p.open(
format=FORMAT,
channels=CHANNELS,
rate=RATE,
input=True,
frames_per_buffer=CHUNK
)
print("正在录音...")
frames = []
for _ in range(0, int(RATE / CHUNK * duration)):
data = stream.read(CHUNK)
frames.append(data)
print("录音结束")
stream.stop_stream()
stream.close()
p.terminate()
wf = wave.open(output_path, 'wb')
wf.setnchannels(CHANNELS)
wf.setsampwidth(p.get_sample_size(FORMAT))
wf.setframerate(RATE)
wf.writeframes(b''.join(frames))
wf.close()
return output_path
def transcribe(self, audio_path):
"""语音转文字"""
with open(audio_path, "rb") as audio_file:
result = self.client.audio.transcriptions.create(
model="whisper-1",
file=audio_file,
language="zh"
)
return result.text
def think(self, user_text):
"""AI思考并生成回答"""
self.conversation_history.append({
"role": "user",
"content": user_text
})
response = self.client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": "你是一个友好的语音助手，请简洁明了地回答问题。"},
*self.conversation_history
],
max_tokens=500
)
assistant_text = response.choices[0].message.content
self.conversation_history.append({
"role": "assistant",
"content": assistant_text
})
return assistant_text
def speak(self, text, output_path="response.mp3"):
"""文字转语音"""
response = self.client.audio.speech.create(
model="tts-1",
voice="nova",
input=text,
speed=1.0
)
response.stream_to_file(output_path)
return output_path
def chat_once(self, listen_duration=5):
"""完成一轮对话：听->想->说"""
# 听
audio_path = self.listen(duration=listen_duration)
# 转录
user_text = self.transcribe(audio_path)
print(f"用户说：{user_text}")
# 思考
response_text = self.think(user_text)
print(f"AI回答：{response_text}")
# 说
output_audio = self.speak(response_text)
return user_text, response_text, output_audio
def reset(self):
"""重置对话历史"""
self.conversation_history = []
# 使用示例
assistant = VoiceAssistant("your-api-key")
# 进行一轮对话
user_text, response_text, audio_path = assistant.chat_once()
print(f"对话完成，音频已保存到：{audio_path}")
7.5 音频情感分析
def analyze_audio_emotion(audio_path):
"""分析音频中的情感"""
with open(audio_path, "rb") as audio_file:
# 先转录
transcript = client.audio.transcriptions.create(
model="whisper-1",
file=audio_file,
language="zh"
)
# 分析文本情感
response = client.chat.completions.create(
model="gpt-4o",
messages=[{
"role": "user",
"content": f"""请分析以下文本的情感：
"{transcript.text}"
返回JSON格式：
{{
"emotion": "主要情感（开心/悲伤/愤怒/惊讶/恐惧/中性）",
"confidence": 0.0-1.0,
"sentiment": "positive/negative/neutral",
"keywords": ["关键词列表"]
}}"""
}],
response_format={"type": "json_object"},
temperature=0
)
return json.loads(response.choices[0].message.content)
7.6 会议纪要自动生成
class MeetingMinutesGenerator:
"""会议纪要自动生成器"""
def __init__(self, api_key):
self.client = OpenAI(api_key=api_key)
def transcribe_meeting(self, audio_path):
"""转录会议录音"""
with open(audio_path, "rb") as audio_file:
result = self.client.audio.transcriptions.create(
model="whisper-1",
file=audio_file,
language="zh",
response_format="verbose_json"
)
return result
def generate_minutes(self, audio_path):
"""生成会议纪要"""
# 第一步：转录
transcription = self.transcribe_meeting(audio_path)
# 第二步：生成纪要
response = self.client.chat.completions.create(
model="gpt-4o",
messages=[{
"role": "user",
"content": f"""请根据以下会议转录文本，生成一份结构化的会议纪要：
转录文本：
{transcription.text}
请按以下格式输出：
# 会议纪要
## 基本信息
- 日期：
- 时长：{transcription.duration:.0f}秒
- 参与者：（从内容推断）
## 主要议题
（列出讨论的主要话题）
## 关键决策
（列出做出的决定）
## 行动项
（列出需要执行的任务和负责人）
## 详细记录
（按时间顺序记录讨论内容）"""
}],
max_tokens=3000
)
return response.choices[0].message.content
# 使用示例
generator = MeetingMinutesGenerator("your-api-key")
minutes = generator.generate_minutes("meeting.mp3")
print(minutes)
7.7 本章小结
本章我们学习了语音多模态的核心技术：
- 语音识别（ASR）：使用Whisper进行语音转文字
- 语音合成（TTS）：使用OpenAI TTS API生成自然语音
- 语音对话系统：整合ASR和TTS构建完整的语音助手
- 音频情感分析：分析语音中的情感信息
- 会议纪要生成：自动转录和总结会议内容
第八章：视频理解与分析
8.1 视频理解的挑战
视频理解比图像理解更具挑战性，因为视频包含时间维度的信息。主要挑战包括：
- 数据量大：一段1分钟的视频可能包含1800帧（30fps），处理开销巨大
- 时序信息：需要理解动作的先后顺序和因果关系
- 长视频理解：需要理解跨越数分钟甚至数小时的内容
- 多模态融合：视频同时包含视觉、音频、字幕等多种模态
8.2 视频帧提取
视频理解的第一步是从视频中提取关键帧：
import cv2
import os
from PIL import Image
import numpy as np
class VideoFrameExtractor:
"""视频帧提取器"""
def __init__(self, video_path):
self.video_path = video_path
self.cap = cv2.VideoCapture(video_path)
self.fps = self.cap.get(cv2.CAP_PROP_FPS)
self.total_frames = int(self.cap.get(cv2.CAP_PROP_FRAME_COUNT))
self.duration = self.total_frames / self.fps
def extract_by_interval(self, interval=1.0):
"""按时间间隔提取帧"""
frames = []
timestamps = []
frame_interval = int(self.fps * interval)
for frame_idx in range(0, self.total_frames, frame_interval):
self.cap.set(cv2.CAP_PROP_POS_FRAMES, frame_idx)
ret, frame = self.cap.read()
if ret:
# 转换颜色空间（BGR -> RGB）
frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
frames.append(Image.fromarray(frame_rgb))
timestamps.append(frame_idx / self.fps)
return frames, timestamps
def extract_keyframes(self, threshold=30.0):
"""基于场景变化提取关键帧"""
frames = []
timestamps = []
prev_frame = None
for frame_idx in range(self.total_frames):
ret, frame = self.cap.read()
if not ret:
break
gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
if prev_frame is not None:
# 计算帧差
diff = cv2.absdiff(gray, prev_frame)
mean_diff = np.mean(diff)
if mean_diff > threshold:
frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
frames.append(Image.fromarray(frame_rgb))
timestamps.append(frame_idx / self.fps)
prev_frame = gray
return frames, timestamps
def extract_uniform(self, num_frames=10):
"""均匀提取指定数量的帧"""
frame_indices = np.linspace(0, self.total_frames - 1, num_frames, dtype=int)
frames = []
timestamps = []
for idx in frame_indices:
self.cap.set(cv2.CAP_PROP_POS_FRAMES, idx)
ret, frame = self.cap.read()
if ret:
frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
frames.append(Image.fromarray(frame_rgb))
timestamps.append(idx / self.fps)
return frames, timestamps
def close(self):
"""释放资源"""
self.cap.release()
# 使用示例
extractor = VideoFrameExtractor("video.mp4")
print(f"视频时长：{extractor.duration:.1f}秒")
print(f"帧率：{extractor.fps}fps")
print(f"总帧数：{extractor.total_frames}")
# 均匀提取10帧
frames, timestamps = extractor.extract_uniform(num_frames=10)
for i, (frame, ts) in enumerate(zip(frames, timestamps)):
frame.save(f"frame_{i:03d}_{ts:.1f}s.jpg")
print(f"已保存帧 {i+1}，时间戳：{ts:.1f}秒")
extractor.close()
8.3 使用GPT-4o分析视频
将视频分解为多帧图像，然后用GPT-4o进行分析：
import base64
from io import BytesIO
class VideoAnalyzer:
"""视频分析器"""
def __init__(self, api_key):
self.client = OpenAI(api_key=api_key)
def frame_to_base64(self, frame):
"""将PIL图像转为base64"""
buffer = BytesIO()
frame.save(buffer, format="JPEG", quality=85)
return base64.b64encode(buffer.getvalue()).decode("utf-8")
def analyze_video(self, video_path, question, num_frames=8):
"""分析视频内容"""
# 提取帧
extractor = VideoFrameExtractor(video_path)
frames, timestamps = extractor.extract_uniform(num_frames)
extractor.close()
# 构建多图请求
content = [
{"type": "text", "text": f"以下是从一段视频中均匀提取的{num_frames}帧画面。"},
]
for i, (frame, ts) in enumerate(zip(frames, timestamps)):
content.append({
"type": "text",
"text": f"帧{i+1}（时间 {ts:.1f}秒）："
})
content.append({
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{self.frame_to_base64(frame)}",
"detail": "low"
}
})
content.append({
"type": "text",
"text": f"
请根据这些画面回答以下问题：{question}"
})
response = self.client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": content}],
max_tokens=2000
)
return response.choices[0].message.content
def generate_video_summary(self, video_path, num_frames=12):
"""生成视频摘要"""
return self.analyze_video(
video_path,
"请对这个视频进行详细的内容摘要，包括：\n1. 视频主题\n2. 主要内容\n3. 关键事件\n4. 出现的人物/物体",
num_frames
)
def extract_video_highlights(self, video_path, num_frames=16):
"""提取视频精彩片段"""
return self.analyze_video(
video_path,
"请分析这个视频，指出最精彩的3-5个时刻，并说明每个时刻的时间段和内容。",
num_frames
)
# 使用示例
analyzer = VideoAnalyzer("your-api-key")
# 分析视频
result = analyzer.analyze_video(
"cooking_video.mp4",
"这个视频中在做什么菜？请列出主要步骤。",
num_frames=10
)
print(result)
# 生成摘要
summary = analyzer.generate_video_summary("vlog.mp4")
print(summary)
8.4 视频字幕生成
class VideoCaptionGenerator:
"""视频字幕生成器"""
def __init__(self, api_key):
self.client = OpenAI(api_key=api_key)
def generate_captions(self, video_path, output_srt_path):
"""生成SRT字幕文件"""
# 第一步：提取音频并转录
# 使用ffmpeg提取音频
import subprocess
audio_path = video_path.rsplit('.', 1)[0] + '.wav'
subprocess.run([
'ffmpeg', '-i', video_path,
'-vn', '-acodec', 'pcm_s16le',
'-ar', '16000', '-ac', '1',
audio_path, '-y'
], capture_output=True)
# 转录
with open(audio_path, "rb") as audio_file:
result = self.client.audio.transcriptions.create(
model="whisper-1",
file=audio_file,
language="zh",
response_format="verbose_json",
timestamp_granularities=["segment"]
)
# 生成SRT格式字幕
srt_content = ""
for i, segment in enumerate(result.segments, 1):
start_time = self._format_timestamp(segment.start)
end_time = self._format_timestamp(segment.end)
text = segment.text.strip()
srt_content += f"{i}\n{start_time} --> {end_time}\n{text}\n\n"
# 保存SRT文件
with open(output_srt_path, 'w', encoding='utf-8') as f:
f.write(srt_content)
# 清理临时音频文件
os.remove(audio_path)
return output_srt_path
def _format_timestamp(self, seconds):
"""格式化时间戳为SRT格式"""
hours = int(seconds // 3600)
minutes = int((seconds % 3600) // 60)
secs = int(seconds % 60)
millis = int((seconds % 1) * 1000)
return f"{hours:02d}:{minutes:02d}:{secs:02d},{millis:03d}"
# 使用示例
caption_gen = VideoCaptionGenerator("your-api-key")
srt_path = caption_gen.generate_captions("video.mp4", "video.srt")
print(f"字幕已保存到：{srt_path}")
8.5 视频内容检索系统
class VideoSearchEngine:
"""视频内容检索系统"""
def __init__(self, api_key):
self.client = OpenAI(api_key=api_key)
self.video_index = [] # 存储视频片段的索引
def index_video(self, video_path, segment_duration=30):
"""为视频建立索引"""
extractor = VideoFrameExtractor(video_path)
video_name = os.path.basename(video_path)
# 按片段处理视频
num_segments = int(extractor.duration / segment_duration)
for seg_idx in range(num_segments):
start_time = seg_idx * segment_duration
end_time = min((seg_idx + 1) * segment_duration, extractor.duration)
# 提取该片段的关键帧
frames, timestamps = extractor.extract_uniform(num_frames=3)
# 生成片段描述
content = [
{"type": "text", "text": "请用3-5个关键词描述这个视频片段的内容："}
]
for frame in frames:
buffer = BytesIO()
frame.save(buffer, format="JPEG")
b64 = base64.b64encode(buffer.getvalue()).decode("utf-8")
content.append({
"type": "image_url",
"image_url": {"url": f"data:image/jpeg;base64,{b64}", "detail": "low"}
})
response = self.client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": content}],
max_tokens=100
)
description = response.choices[0].message.content
# 生成嵌入
emb_response = self.client.embeddings.create(
model="text-embedding-3-small",
input=description
)
self.video_index.append({
"video": video_path,
"video_name": video_name,
"segment_idx": seg_idx,
"start_time": start_time,
"end_time": end_time,
"description": description,
"embedding": emb_response.data[0].embedding
})
extractor.close()
print(f"已为 {video_name} 建立索引，共 {num_segments} 个片段")
def search(self, query, top_k=5):
"""搜索视频内容"""
# 生成查询嵌入
emb_response = self.client.embeddings.create(
model="text-embedding-3-small",
input=query
)
query_emb = np.array(emb_response.data[0].embedding)
# 计算相似度
results = []
for item in self.video_index:
item_emb = np.array(item["embedding"])
sim = cosine_similarity(
query_emb.reshape(1, -1),
item_emb.reshape(1, -1)
)[0][0]
results.append({**item, "score": sim})
results.sort(key=lambda x: x["score"], reverse=True)
return results[:top_k]
# 使用示例
search_engine = VideoSearchEngine("your-api-key")
search_engine.index_video("lecture.mp4", segment_duration=60)
# 搜索
results = search_engine.search("机器学习的基本概念")
for r in results:
print(f"视频：{r['video_name']}, 时间：{r['start_time']:.0f}-{r['end_time']:.0f}秒")
print(f"描述：{r['description']}")
print(f"相似度：{r['score']:.3f}")
print("---")
8.6 本章小结
本章我们学习了视频理解与分析的核心技术：
- 视频帧提取：支持按时间间隔、场景变化和均匀分布三种方式
- 使用GPT-4o进行视频内容分析
- 自动生成视频字幕（SRT格式）
- 构建视频内容检索系统
- 视频摘要和精彩片段提取
第九章：多模态RAG系统
9.1 什么是多模态RAG
RAG（Retrieval-Augmented Generation，检索增强生成）是一种结合信息检索和文本生成的技术。传统的RAG系统只处理文本，而多模态RAG扩展了这一概念，能够同时检索和利用文本、图像、表格、图表等多种模态的信息。
多模态RAG的典型应用场景：
- 企业文档分析：处理包含文字、图表、表格的PDF文档
- 产品知识库：结合产品图片和文字说明回答用户问题
- 医学诊断辅助：检索相关的医学影像和病历
- 教育领域：根据教材中的图片和文字回答学生问题
9.2 多模态RAG架构
多模态RAG系统的核心流程如下：
用户查询（文本/图片）
|
v
+------------------+
| 查询理解与编码 |
+------------------+
|
v
+------------------+
| 多模态向量检索 |
+------------------+
|
v
+------------------+
| 检索结果排序 |
+------------------+
|
v
+------------------+
| 上下文构建 |
+------------------+
|
v
+------------------+
| LLM生成回答 |
+------------------+
9.3 文档解析与处理
9.3.1 PDF文档解析
import fitz # PyMuPDF
from PIL import Image
import io
import os
class PDFProcessor:
"""PDF文档处理器"""
def __init__(self, pdf_path):
self.pdf_path = pdf_path
self.doc = fitz.open(pdf_path)
def extract_text(self):
"""提取文本内容"""
pages_text = []
for page_num in range(len(self.doc)):
page = self.doc[page_num]
text = page.get_text()
pages_text.append({
"page": page_num + 1,
"text": text.strip()
})
return pages_text
def extract_images(self, output_dir="./extracted_images"):
"""提取图片"""
os.makedirs(output_dir, exist_ok=True)
images = []
for page_num in range(len(self.doc)):
page = self.doc[page_num]
image_list = page.get_images()
for img_idx, img in enumerate(image_list):
xref = img[0]
base_image = self.doc.extract_image(xref)
image_bytes = base_image["image"]
image = Image.open(io.BytesIO(image_bytes))
image_path = os.path.join(
output_dir,
f"page{page_num+1}_img{img_idx+1}.png"
)
image.save(image_path)
images.append({
"page": page_num + 1,
"path": image_path,
"size": image.size
})
return images
def get_page_screenshots(self, output_dir="./page_screenshots"):
"""将每页渲染为图片"""
os.makedirs(output_dir, exist_ok=True)
screenshots = []
for page_num in range(len(self.doc)):
page = self.doc[page_num]
pix = page.get_pixmap(dpi=150)
image_path = os.path.join(output_dir, f"page{page_num+1}.png")
pix.save(image_path)
screenshots.append(image_path)
return screenshots
# 使用示例
processor = PDFProcessor("document.pdf")
text_pages = processor.extract_text()
images = processor.extract_images()
screenshots = processor.get_page_screenshots()
9.3.2 使用多模态模型解析文档页面
class MultimodalDocumentParser:
"""使用多模态模型解析文档"""
def __init__(self, api_key):
self.client = OpenAI(api_key=api_key)
def parse_page(self, page_image_path):
"""解析单页文档"""
with open(page_image_path, "rb") as f:
base64_image = base64.b64encode(f.read()).decode("utf-8")
prompt = """请分析这个文档页面，提取：
1. 文本内容
2. 表格数据（Markdown格式）
3. 图表描述
4. 结构信息（标题、段落等）
返回JSON格式。"""
response = self.client.chat.completions.create(
model="gpt-4o",
messages=[{
"role": "user",
"content": [
{"type": "text", "text": prompt},
{
"type": "image_url",
"image_url": {
"url": f"data:image/png;base64,{base64_image}",
"detail": "high"
}
}
]
}],
response_format={"type": "json_object"},
max_tokens=3000
)
return json.loads(response.choices[0].message.content)
def parse_document(self, page_images):
"""解析整个文档"""
document_content = []
for i, page_image in enumerate(page_images):
print(f"解析第{i+1}/{len(page_images)}页...")
page_content = self.parse_page(page_image)
page_content["page_number"] = i + 1
document_content.append(page_content)
return document_content
9.4 多模态向量存储
import numpy as np
class MultimodalVectorStore:
"""多模态向量存储"""
def __init__(self, api_key):
self.client = OpenAI(api_key=api_key)
self.documents = []
self.vectors = []
def add_text(self, text, metadata=None):
"""添加文本文档"""
response = self.client.embeddings.create(
model="text-embedding-3-small",
input=text
)
embedding = response.data[0].embedding
self.documents.append({
"type": "text", "content": text,
"metadata": metadata or {}
})
self.vectors.append(embedding)
def add_image(self, image_path, description=None, metadata=None):
"""添加图片文档"""
if description is None:
with open(image_path, "rb") as f:
base64_image = base64.b64encode(f.read()).decode("utf-8")
response = self.client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": [
{"type": "text", "text": "请描述这张图片"},
{"type": "image_url", "image_url": {
"url": f"data:image/jpeg;base64,{base64_image}",
"detail": "low"
}}
]}],
max_tokens=200
)
description = response.choices[0].message.content
emb_response = self.client.embeddings.create(
model="text-embedding-3-small", input=description
)
self.documents.append({
"type": "image", "content": description,
"path": image_path, "metadata": metadata or {}
})
self.vectors.append(emb_response.data[0].embedding)
def search(self, query, top_k=5):
"""语义检索"""
response = self.client.embeddings.create(
model="text-embedding-3-small", input=query
)
query_vector = np.array(response.data[0].embedding)
similarities = []
for i, doc_vector in enumerate(self.vectors):
doc_vec = np.array(doc_vector)
sim = np.dot(query_vector, doc_vec) / (
np.linalg.norm(query_vector) * np.linalg.norm(doc_vec)
)
similarities.append((i, sim))
similarities.sort(key=lambda x: x[1], reverse=True)
return [
{"document": self.documents[idx], "score": float(score)}
for idx, score in similarities[:top_k]
]
9.5 完整的多模态RAG系统
class MultimodalRAGSystem:
"""完整的多模态RAG系统"""
def __init__(self, api_key):
self.client = OpenAI(api_key=api_key)
self.vector_store = MultimodalVectorStore(api_key)
def index_document(self, document_path):
"""索引文档"""
if document_path.endswith(".pdf"):
processor = PDFProcessor(document_path)
for page in processor.extract_text():
if page["text"].strip():
self.vector_store.add_text(
page["text"],
metadata={"source": document_path, "page": page["page"]}
)
screenshots = processor.get_page_screenshots()
for i, screenshot in enumerate(screenshots):
self.vector_store.add_image(screenshot,
metadata={"source": document_path, "page": i + 1})
elif document_path.endswith((".jpg", ".png")):
self.vector_store.add_image(document_path)
def _split_text(self, text, chunk_size=500, overlap=50):
"""文本分块"""
chunks = []
start = 0
while start < len(text):
chunks.append(text[start:start + chunk_size])
start += chunk_size - overlap
return chunks
def query(self, question, top_k=5):
"""查询RAG系统"""
results = self.vector_store.search(question, top_k=top_k)
context_parts = []
for i, result in enumerate(results):
doc = result["document"]
score = result["score"]
if doc["type"] == "text":
source = doc.get("metadata", {}).get("source", "未知")
context_parts.append(f"[文档{i+1}] 来源：{source}, 相关度：{score:.2f}\n{doc["content"]}")
elif doc["type"] == "image":
context_parts.append(f"[图片{i+1}] 相关度：{score:.2f}\n描述：{doc["content"]}")
context = "\n\n".join(context_parts)
response = self.client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": "你是一个知识库助手。根据上下文回答问题。"},
{"role": "user", "content": f"上下文：\n{context}\n\n问题：{question}"}
],
max_tokens=2000
)
return {
"answer": response.choices[0].message.content,
"sources": [{"type": r["document"]["type"], "score": r["score"]} for r in results]
}
# 使用示例
rag = MultimodalRAGSystem("your-api-key")
rag.index_document("company_report.pdf")
result = rag.query("公司的主要业务是什么？")
print(result["answer"])
9.6 本章小结
本章我们学习了多模态RAG系统的构建：
- PDF文档解析：提取文本、图片和表格
- 使用多模态模型理解文档页面
- 多模态向量存储：支持文本、图片和表格的索引
- 完整的RAG流程：从文档索引到智能问答
- 多模态RAG适用于包含丰富视觉信息的企业文档场景
第十章：多模态Agent开发
10.1 什么是多模态Agent
多模态Agent是能够感知和操作多种模态信息的智能代理系统。它不仅能"看"和"说"，还能调用工具、执行操作、与环境交互。
与普通的多模态对话系统不同，Agent具有以下特点：
- 自主决策：根据任务目标自主规划行动步骤
- 工具使用：能够调用外部工具（搜索引擎、计算器、代码执行等）
- 环境感知：通过多种模态感知环境状态
- 记忆系统：维护短期和长期记忆
- 反思能力：能够评估自己的行为并调整策略
10.2 Agent架构设计
from abc import ABC, abstractmethod
from typing import List, Dict, Any
import json
class Tool(ABC):
"""工具基类"""
@property
@abstractmethod
def name(self) -> str:
pass
@property
@abstractmethod
def description(self) -> str:
pass
@property
@abstractmethod
def parameters(self) -> Dict:
pass
@abstractmethod
def execute(self, **kwargs) -> str:
pass
class MultimodalAgent:
"""多模态Agent"""
def __init__(self, api_key, model="gpt-4o"):
self.client = OpenAI(api_key=api_key)
self.model = model
self.tools: Dict[str, Tool] = {}
self.memory: List[Dict] = []
self.system_prompt = ""
def register_tool(self, tool: Tool):
"""注册工具"""
self.tools[tool.name] = tool
def set_system_prompt(self, prompt: str):
"""设置系统提示"""
self.system_prompt = prompt
def _build_tools_schema(self) -> List[Dict]:
"""构建工具描述"""
schemas = []
for tool in self.tools.values():
schemas.append({
"type": "function",
"function": {
"name": tool.name,
"description": tool.description,
"parameters": tool.parameters
}
})
return schemas
def observe(self, image=None, text=None, audio=None):
"""观察环境"""
observation = {"role": "user", "content": []}
if image:
with open(image, "rb") as f:
base64_image = base64.b64encode(f.read()).decode("utf-8")
observation["content"].append({
"type": "image_url",
"image_url": {"url": f"data:image/jpeg;base64,{base64_image}"}
})
if text:
observation["content"].append({"type": "text", "text": text})
if audio:
with open(audio, "rb") as audio_file:
transcript = self.client.audio.transcriptions.create(
model="whisper-1", file=audio_file
)
observation["content"].append({
"type": "text",
"text": f"[语音输入] {transcript.text}"
})
self.memory.append(observation)
def think(self, max_rounds=5):
"""Agent思考和行动循环"""
messages = []
if self.system_prompt:
messages.append({"role": "system", "content": self.system_prompt})
messages.extend(self.memory)
tools_schema = self._build_tools_schema()
for round_idx in range(max_rounds):
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
tools=tools_schema if tools_schema else None,
max_tokens=2000
)
assistant_message = response.choices[0].message
if assistant_message.tool_calls:
messages.append(assistant_message)
for tool_call in assistant_message.tool_calls:
tool_name = tool_call.function.name
tool_args = json.loads(tool_call.function.arguments)
print(f"[Agent] 调用工具：{tool_name}")
if tool_name in self.tools:
result = self.tools[tool_name].execute(**tool_args)
else:
result = f"错误：工具 {tool_name} 不存在"
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": str(result)
})
else:
final_answer = assistant_message.content
self.memory.append({"role": "assistant", "content": final_answer})
return final_answer
return "达到最大思考轮数，未能完成任务"
10.3 实现具体工具
class ImageAnalysisTool(Tool):
"""图像分析工具"""
def __init__(self, api_key):
self.client = OpenAI(api_key=api_key)
@property
def name(self):
return "analyze_image"
@property
def description(self):
return "分析图片内容，回答关于图片的问题"
@property
def parameters(self):
return {
"type": "object",
"properties": {
"image_path": {"type": "string", "description": "图片文件路径"},
"question": {"type": "string", "description": "关于图片的问题"}
},
"required": ["image_path", "question"]
}
def execute(self, image_path, question):
with open(image_path, "rb") as f:
base64_image = base64.b64encode(f.read()).decode("utf-8")
response = self.client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": [
{"type": "text", "text": question},
{"type": "image_url", "image_url": {
"url": f"data:image/jpeg;base64,{base64_image}",
"detail": "high"
}}
]}],
max_tokens=500
)
return response.choices[0].message.content
class CalculatorTool(Tool):
"""计算器工具"""
@property
def name(self):
return "calculator"
@property
def description(self):
return "执行数学计算"
@property
def parameters(self):
return {
"type": "object",
"properties": {
"expression": {"type": "string", "description": "数学表达式"}
},
"required": ["expression"]
}
def execute(self, expression):
try:
result = eval(expression)
return f"计算结果：{result}"
except Exception as e:
return f"计算错误：{str(e)}"
class CodeExecutionTool(Tool):
"""代码执行工具"""
@property
def name(self):
return "execute_code"
@property
def description(self):
return "执行Python代码并返回结果"
@property
def parameters(self):
return {
"type": "object",
"properties": {
"code": {"type": "string", "description": "要执行的Python代码"}
},
"required": ["code"]
}
def execute(self, code):
import subprocess, tempfile
with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f:
f.write(code)
temp_path = f.name
try:
result = subprocess.run(
["python3", temp_path],
capture_output=True, text=True, timeout=30
)
return result.stdout + ("\n错误：" + result.stderr if result.stderr else "")
except subprocess.TimeoutExpired:
return "错误：代码执行超时"
10.4 构建完整的Agent应用
class ImageAssistantAgent(MultimodalAgent):
"""图像助手Agent"""
def __init__(self, api_key):
super().__init__(api_key)
self.set_system_prompt(
"你是一个智能图像助手Agent。你可以分析图片、使用工具处理任务、回答问题。"
)
self.register_tool(ImageAnalysisTool(api_key))
self.register_tool(CalculatorTool())
self.register_tool(CodeExecutionTool())
def process_image_task(self, image_path, task):
"""处理图片相关任务"""
self.observe(image=image_path, text=task)
return self.think(max_rounds=5)
class ConversationalMultimodalAgent:
"""多模态对话Agent"""
def __init__(self, api_key):
self.client = OpenAI(api_key=api_key)
self.conversation_history = []
self.system_prompt = "你是一个友好的多模态AI助手。"
def process_message(self, text=None, image_path=None, audio_path=None):
"""处理用户消息"""
user_content = []
if image_path:
with open(image_path, "rb") as f:
base64_image = base64.b64encode(f.read()).decode("utf-8")
user_content.append({
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{base64_image}",
"detail": "high"
}
})
if audio_path:
with open(audio_path, "rb") as audio_file:
transcript = self.client.audio.transcriptions.create(
model="whisper-1", file=audio_file
)
user_content.append({
"type": "text",
"text": f"[语音消息] {transcript.text}"
})
if text:
user_content.append({"type": "text", "text": text})
self.conversation_history.append({"role": "user", "content": user_content})
messages = [
{"role": "system", "content": self.system_prompt},
*self.conversation_history
]
response = self.client.chat.completions.create(
model="gpt-4o", messages=messages, max_tokens=2000
)
assistant_text = response.choices[0].message.content
self.conversation_history.append({"role": "assistant", "content": assistant_text})
return assistant_text
def reset(self):
self.conversation_history = []
# 使用示例
agent = ConversationalMultimodalAgent("your-api-key")
print(agent.process_message(text="你好！"))
print(agent.process_message(text="请描述这张图片", image_path="photo.jpg"))
print(agent.process_message(audio_path="voice.mp3"))
10.5 Agent记忆系统
class AgentMemory:
"""Agent记忆系统"""
def __init__(self, api_key, max_short_term=20):
self.client = OpenAI(api_key=api_key)
self.short_term_memory = []
self.long_term_memory = []
self.max_short_term = max_short_term
def add_to_short_term(self, role, content):
"""添加到短期记忆"""
self.short_term_memory.append({
"role": role, "content": content
})
if len(self.short_term_memory) > self.max_short_term:
self._compress_memory()
def add_to_long_term(self, content, importance=0.5):
"""添加到长期记忆"""
self.long_term_memory.append({
"content": content, "importance": importance
})
def _compress_memory(self):
"""压缩短期记忆"""
recent = self.short_term_memory[-10:]
old = self.short_term_memory[:-10]
# 将旧对话总结为长期记忆
conversation_text = "\n".join([
f"{m['role']}: {m['content']}" for m in old
])
response = self.client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content":
f"请总结以下对话的关键信息：\n{conversation_text}"}],
max_tokens=200
)
summary = response.choices[0].message.content
self.add_to_long_term(summary, importance=0.7)
self.short_term_memory = recent
def get_context(self):
"""获取记忆上下文"""
context = []
if self.long_term_memory:
long_term_text = "\n".join([
m["content"] for m in self.long_term_memory[-5:]
])
context.append({
"role": "system",
"content": f"历史记忆：\n{long_term_text}"
})
context.extend([
{"role": m["role"], "content": m["content"]}
for m in self.short_term_memory
])
return context
10.6 本章小结
本章我们学习了多模态Agent的开发：
- Agent架构设计：工具注册、观察、思考、行动的循环
- 工具实现：图像分析、代码执行、计算等工具
- 完整的Agent应用：图像助手和对话Agent
- 多模态对话Agent：支持文字、图片、语音的综合交互
- 记忆系统：短期记忆和长期记忆的管理
