Gemini 的多模态架构设计如何统一文本、图像、视频的表示？

Gemini 的多模态架构设计如何统一文本、图像、视频的表示？

文档概述

本文核心价值

学习目标

一、Gemini 多模态技术概述

1.1 什么是 Gemini 多模态模型？

Gemini 是 Google DeepMind 推出的新一代多模态大模型，核心定位是原生支持文本、图像、音频、视频等多模态输入，并在统一的张量空间中完成表示与推理。与传统 “单模态模型 + 模态转换模块” 的拼接式架构不同，Gemini 从底层设计上实现了多模态的深度融合，真正做到 “一个模型处理所有模态”。

1.2 传统多模态方案的局限性（Gemini 解决的核心痛点）

1.3 Gemini 多模态表示统一的核心设计原则

1.3 Gemini 多模态表示统一的核心设计原则

Gemini 的核心突破在于摒弃了传统多模态模型“拼接式”的异构设计，转而采用原生（Native）多模态架构。这意味着模型不再需要针对不同模态训练独立的编码器（如 CLIP 中的 Image Encoder 和 Text Encoder），而是通过以下五大原则，在底层物理空间和高层语义空间上实现了彻底的统一。

1.3.1 原理架构树形图 (Conceptual Tree)

Gemini_Unified_Principles/ # 核心设计原则全景
│
├── 1. Universal_Input (万物皆序列：物理层的统一)
│ ├── Text_Stream ──> [Tokenization] ──> 1D ID序列
│ ├── Image_Stream ──> [Patch Partition] ──> 线性投影 ──> 伪1D序列
│ └── Video_Stream ──> [3D Tubelet] ──> 时空压平 ──> 带有时间戳的序列
│
├── 2. Shared_Backbone (共享大脑：计算层的统一)
│ ├── Single_Transformer # 拒绝双塔结构，所有模态进入同一个网络
│ ├── Shared_Parameters # 文本和图像共用同一组权重进行推理
│ └── Early_Fusion # 模态融合发生在第一层，而非最后输出层
│
├── 3. Attention_Mechanism (动态对齐：交互层的统一)
│ ├── Self_Attention # 模态内部建模 (如：理解句子语法 / 图像纹理)
│ ├── Cross_Attention # 跨模态桥接 (如：单词"猫" <-> 图像中猫的像素)
│ └── Temporal_Attention # (视频专用) 沿时间轴捕捉动作变化
│
└── 4. Unified_Metric (度量一致：语义层的统一)
├── Projection_Layer # 强制维度对齐 (如统一映射到 2048 维)
└── L2_Normalization # 归一化到超球面上，确保 Cosine 距离可比

1.3.2 核心原则深度解析

1. 模态无关的输入编码 (Modality-Agnostic Input Encoding)

• 原理：Gemini 建立了一个“通用词汇表”。它不直接处理原始的像素或音频波形，而是将它们视为一种“外语”，翻译成模型能理解的 Token。
• 扩充细节：
  • 文本：使用 SentencePiece 模型将自然语言切分为子词 Token。
  • 图像：采用 Patch Embedding 技术，将 $H\times W$
  • 视频：不只是图像的堆叠，而是采用 3D Tubelet 或时空采样，将时间维度 $T$

2. 共享的 Transformer 骨干网络 (Shared Transformer Backbone)

• 原理：这是“原生多模态”的标志。所有模态的数据流经同一个神经网络，使用同一套参数（权重/偏置）进行计算。
• 扩充细节：
  • 参数效率：与“双塔结构”（独立视觉塔 + 独立文本塔）相比，Gemini 的参数利用率极高。模型学会的“逻辑推理能力”可以同时作用于文本分析和图像理解。
  • 早期融合 (Early Fusion)：模态间的交互从网络的第一层就开始了，而不是像传统模型那样等到最后才进行特征拼接。这使得模型能捕捉到更深层次的图文隐喻关系。

3. 跨模态注意力对齐 (Cross-Modal Attention Alignment)

• 原理：利用 Attention 机制动态建立不同模态 Token 之间的语义链接。
• 扩充细节：
  • 细粒度对齐：传统模型只能判断“整张图”和“整句话”是否匹配。Gemini 通过交叉注意力（Cross-Attention），可以让文本序列中的 Token（Query）去查询视觉序列中的 Patch（Key/Value）。例如，当处理文本“戴红帽子的男孩”时，模型的注意力头会高亮图像中“红色”和“帽子”对应的 Patch 区域。

4. 时空联合建模 (Joint Spatiotemporal Modeling)

• 原理：针对视频模态，模型必须理解“变化”。
• 扩充细节：
  • 3D 位置编码：除了 x, y 空间坐标，每个视频 Token 还被赋予了 $t$
  • 因果注意力：在理解视频时，模型不仅关注当前帧，还会通过时序注意力机制（Temporal Attention）回顾历史帧，从而理解动作的连续性（如分辨“人坐下”和“人站起”的区别）。

5. 统一的向量归一化 (Unified Vector Normalization)

• 原理：消除不同模态在特征分布上的“量纲”差异，确保它们在同一个几何空间中可比较。
• 扩充细节：
  • 超球面投影：所有输出向量经过 LayerNorm 和 L2 归一化后，都被投射到同一个单位超球面上。
  • 数学意义：这保证了我们可以直接使用余弦相似度 (Cosine Similarity) 来衡量任何两个对象的距离。如果文本“狗”和图像“狗”的向量夹角为 0 度，说明模型真正实现了语义上的统一，而不仅仅是数值上的接近。

二、Gemini 本地部署与环境配置

2.1 部署环境要求

2.2 环境配置步骤

2.2.1 创建独立 Conda 环境

运行

# 创建 Gemini 专属环境
conda create -n gemini-multimodal python=3.10
# 激活环境
conda activate gemini-multimodal

2.2.2 安装核心依赖库

运行

# 基础依赖
pip install numpy==1.26.4 pandas==2.2.2 pillow==10.3.0 opencv-python==4.9.0.80
# PyTorch (GPU版本，匹配CUDA 12.1)
pip install torch==2.3.1 torchvision==0.18.1 torchaudio==2.3.1 –index-url https://download.pytorch.org/whl/cu121
# Hugging Face 生态（模型加载/处理）
pip install transformers==4.41.2 datasets==2.20.0 accelerate==0.31.0
# Google Gemini API/本地部署依赖
pip install google-generativeai==0.7.2 sentence-transformers==2.7.0
# 视频处理依赖
pip install decord==0.6.0 ffmpeg-python==0.2.0
# 向量存储与检索
pip install chromadb==0.5.17 faiss-gpu==1.7.4
# 日志与配置
pip install python-dotenv==1.0.1 loguru==0.7.2

2.2.3 模型下载（本地部署）

Gemini 提供不同量级的模型版本，可根据硬件条件选择：

通过 Hugging Face 下载开源适配版：

运行

from huggingface_hub import snapshot_download

# 下载 Gemini 1.5 Flash 多模态模型（开源适配版）
model_dir = snapshot_download(
repo_id="google/gemma-2-9b-it", # Gemini 开源适配版本
cache_dir="/data/models/gemini",
ignore_patterns=["*.bin.index.json"] # 跳过不必要文件
)
print(f"模型下载完成，路径：{model_dir}")

三、Gemini 多模态表示统一的核心原理

3.1 输入层：模态的统一序列化

Gemini 首先将不同模态转化为统一的序列 Token，为后续的共享编码奠定基础：

在进入神经网络之前，Gemini 必须消除不同数据在物理形态上的差异（像素矩阵 vs 字符编码）。它通过 通用序列化 (Universal Serialization) 将所有模态转化为统一的 Token 序列。

• 文本 (1D 序列)：
  • 机制：使用 SentencePiece/BPE 分词。
  • 转化："一只猫" $\to$
• 图像 (2D –> 1D 序列)：
  • 机制：Patch Partition。将高清图像切分为固定大小（如 $16\times 16$
  • 转化：每个 Patch 被线性投影 (Linear Projection) 展平为一个向量，这就相当于图像的一个“单词”。
• 视频 (3D –>1D 序列)：
  • 机制：Spatiotemporal Patching (时空分块)。不仅在空间上切分 Patch，还在时间轴上采样。
  • 转化：视频被视为一系列随时间变化的图像 Patch，最终也被拉平为一条长长的 Token 序列。

统一结果：无论输入是文字、JPG 还是 MP4，在模型入口处，它们都变成了形状相同的张量 [Batch_Size, Sequence_Length, Dimension]。

示例：视频序列化过程

运行

import cv2
import numpy as np

def video_to_patches(video_path, frame_rate=1, patch_size=16):
"""将视频转化为时空Patch序列"""
# 1. 读取视频并按帧率抽帧
cap = cv2.VideoCapture(video_path)
frames = []
frame_idx = 0
while cap.isOpened():
ret, frame = cap.read()
if not ret:
break
# 按帧率采样
if frame_idx % int(cap.get(cv2.CAP_PROP_FPS) / frame_rate) == 0:
# 转为RGB并Resize到固定尺寸
frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
frame = cv2.resize(frame, (256, 256)) # 16×16 Patch × 16
frames.append(frame)
frame_idx += 1
cap.release()

# 2. 生成时空Patch
patches = []
for t, frame in enumerate(frames):
# 空间维度切分Patch
for y in range(0, frame.shape[0], patch_size):
for x in range(0, frame.shape[1], patch_size):
patch = frame[y:y+patch_size, x:x+patch_size]
# 记录时空坐标 (t, y, x)
patches.append({
"patch": patch.flatten(),
"coords": (t, y//patch_size, x//patch_size)
})

# 3. 转化为统一序列（时间→空间展开）
patch_sequence = np.array([p["patch"] for p in patches])
return patch_sequence, frames

# 测试视频序列化
video_path = "sample_video.mp4"
patch_seq, frames = video_to_patches(video_path)
print(f"视频抽帧数量：{len(frames)}")
print(f"生成Patch序列长度：{len(patch_seq)}")
print(f"单个Patch维度：{patch_seq[0].shape}")

3.2 编码层：共享 Transformer 与跨模态注意力

Gemini 的核心是共享的 Transformer 编码器，所有模态的 Token 序列输入同一套 Transformer 网络，并通过跨模态注意力机制实现语义对齐：

graph TD
A[文本 Token] —> D{共享 Transformer}
B[图像 Patch Token] —> D
C[视频时空 Token] —> D

subgraph "Gemini 统一编码器"
D –自注意力机制 (Self-Attention)—> E[交互与融合]
E –前馈网络 (FFN)—> F[语义特征提取]
end

F —> G[统一向量空间]

核心代码：跨模态注意力实现

运行

import torch
import torch.nn as nn
import torch.nn.functional as F

class CrossModalAttention(nn.Module):
def __init__(self, d_model=2048, n_heads=16):
super().__init__()
self.d_model = d_model
self.n_heads = n_heads
self.head_dim = d_model // n_heads

# 共享的注意力投影层
self.q_proj = nn.Linear(d_model, d_model)
self.k_proj = nn.Linear(d_model, d_model)
self.v_proj = nn.Linear(d_model, d_model)
self.out_proj = nn.Linear(d_model, d_model)

def forward(self, text_embeds, visual_embeds):
"""
跨模态注意力计算：以文本为Query，视觉（图像/视频）为Key/Value
Args:
text_embeds: [batch, text_len, d_model] 文本嵌入
visual_embeds: [batch, visual_len, d_model] 视觉嵌入
Returns:
cross_embeds: [batch, text_len + visual_len, d_model] 融合后的嵌入
"""
batch_size = text_embeds.shape[0]

# 1. 投影为Query/Key/Value
q = self.q_proj(text_embeds) # 文本作为Query
k = self.k_proj(visual_embeds) # 视觉作为Key
v = self.v_proj(visual_embeds) # 视觉作为Value

# 2. 分拆注意力头
q = q.view(batch_size, –1, self.n_heads, self.head_dim).transpose(1, 2)
k = k.view(batch_size, –1, self.n_heads, self.head_dim).transpose(1, 2)
v = v.view(batch_size, –1, self.n_heads, self.head_dim).transpose(1, 2)

# 3. 计算注意力分数（缩放点积）
scores = torch.matmul(q, k.transpose(–2, –1)) / torch.sqrt(torch.tensor(self.head_dim, dtype=torch.float32))
attn_weights = F.softmax(scores, dim=–1)

# 4. 注意力加权求和
cross_output = torch.matmul(attn_weights, v)
cross_output = cross_output.transpose(1, 2).contiguous().view(batch_size, –1, self.d_model)
cross_output = self.out_proj(cross_output)

# 5. 拼接文本与视觉嵌入（保持序列完整性）
cross_embeds = torch.cat([text_embeds + cross_output, visual_embeds], dim=1)
return cross_embeds

# 测试跨模态注意力
if __name__ == "__main__":
# 模拟文本和视觉嵌入
batch_size = 2
text_len = 32
visual_len = 256
d_model = 2048

text_embeds = torch.randn(batch_size, text_len, d_model)
visual_embeds = torch.randn(batch_size, visual_len, d_model)

# 初始化跨模态注意力层
cross_attn = CrossModalAttention(d_model=d_model, n_heads=16)
cross_embeds = cross_attn(text_embeds, visual_embeds)

print(f"融合后嵌入维度：{cross_embeds.shape}") # [2, 32+256=288, 2048]

3.3 输出层：统一的向量表示空间

Gemini 通过以下方式保证所有模态输出向量的统一可比较性：

示例：多模态向量归一化

运行

def normalize_embeddings(embeddings):
"""L2归一化，保证向量长度为1"""
norm = torch.norm(embeddings, p=2, dim=–1, keepdim=True)
return embeddings / (norm + 1e-8)

# 模拟不同模态的原始嵌入
text_emb = torch.randn(1, 2048) # 文本嵌入
image_emb = torch.randn(1, 2048) # 图像嵌入
video_emb = torch.randn(1, 2048) # 视频嵌入

# 归一化
text_emb_norm = normalize_embeddings(text_emb)
image_emb_norm = normalize_embeddings(image_emb)
video_emb_norm = normalize_embeddings(video_emb)

# 验证归一化结果
print(f"文本向量长度：{torch.norm(text_emb_norm).item():.4f}") # 应为 1.0
print(f"图像向量长度：{torch.norm(image_emb_norm).item():.4f}") # 应为 1.0
print(f"视频向量长度：{torch.norm(video_emb_norm).item():.4f}") # 应为 1.0

# 计算跨模态相似度
text_image_sim = torch.cosine_similarity(text_emb_norm, image_emb_norm)
text_video_sim = torch.cosine_similarity(text_emb_norm, video_emb_norm)
print(f"文本-图像相似度：{text_image_sim.item():.4f}")
print(f"文本-视频相似度：{text_video_sim.item():.4f}")

3.4 总结：传统架构 vs Gemini 架构

为了更直观地理解，我们可以对比两种架构：

四、实战：Gemini 统一文本 / 图像 / 视频表示的完整实现

4.1 项目结构设计

gemini-multimodal/
├── .env # [密钥金库] 存放 API Key、本地模型绝对路径，隔离敏感环境配置
├── requirements.txt # [依赖清单] 定义 PyTorch、Transformers、ChromaDB 等核心计算库版本
├── config.py # [统一标准] 定义跨模态共享的维度(D_MODEL)、最大序列长度及硬件参数
├── main.py # [中控台] 系统入口，负责调度预处理、编码与检索模块的流水线交互
│
├── core/ # [多模态引擎] 核心算法实现库（本系统的“心脏”）
│ ├── __init__.py # 包初始化文件
│ ├── model_setup.py # [共享骨架] 加载 Gemini 统一 Transformer 编码器与跨模态适配层
│ ├── modal_processor.py# [序列化工厂] 将文本/图像/视频转化为统一格式的 Token 序列
│ ├── embedding.py # [统一映射] 执行前向传播，输出并归一化多模态向量（语义对齐核心）
│ └── retriever.py # [混合记忆] 管理向量数据库，执行跨模态（如文搜图、图搜视频）检索
│
├── data/ # [异构数据源] 存放待处理的多模态原始文件
│ ├── text/ # [文本语料] .txt / .json 格式的文本描述
│ ├── images/ # [静态视觉] .jpg / .png 格式的图像文件
│ └── videos/ # [动态视觉] .mp4 / .avi 格式的视频片段
│
├── embeddings_db/ # [语义空间] 持久化的向量存储
│ └── chroma/ # [ChromaDB] 存储已对齐的高维向量及其元数据（Metadata）
│
└── logs/ # [运行记录]
└── multimodal.log # 记录编码延迟、模态转换错误及检索匹配度日志

4.1.2 核心模块深度解析 (The Components)

这些文件共同构成了一个能够理解多种模态的“统一大脑”。与传统多模态项目不同，这里的核心逻辑在于如何抹平不同模态之间的差异。

1. 基础设施与标准定义 (The Infrastructure)

• config.py (统一标准)
  • 用途：这是系统的“度量衡”。
  • 深度解析：它定义了 Gemini 统一表示空间 的物理法则。最关键的参数是 D_MODEL = 2048（或根据模型变动），这强制要求无论是文本、图像还是视频，最终输出的向量必须是这个维度。同时，它定义了 MAX_SEQ_LEN，确保不同模态在序列化后能适配同一个 Transformer 的上下文窗口。
• core/model_setup.py (共享骨架)
  • 用途：加载 Gemini 的“单一模型权重”。
  • 深度解析：传统架构可能需要分别加载 BERT（处理文本）和 ResNet（处理图像）。而此文件通过 AutoModel.from_pretrained 加载 同一个 Transformer 骨干网络。它确保文本和视觉信号流向同一个神经网络，这是实现“原生多模态”的物理基础。

2. 统一表示引擎 (The Unification Engine)

这是项目中最核心的部分，负责执行“输入对齐”和“输出对齐”。

• core/modal_processor.py (序列化工厂 / Input Layer)
  • 用途：模态的“粉碎机”与“包装机”。
  • 协作逻辑：
    • 文本：调用 Tokenizer 进行 BPE 编码，生成 1D ID 序列。
    • 图像：调用 ImageProcessor 将图片切分为 16×16 的 Patch，拉平成序列。
    • 视频：这是关键。它执行“时空采样”，先按时间抽帧，再按空间切块，最终生成带有时间戳信息的 3D Token 序列。
  • 核心价值：它向模型屏蔽了原始数据的格式差异，让模型只看到“Token 序列”。
• core/embedding.py (统一映射 / Output Layer)
  • 用途：向量生成器与归一化器。
  • 深度解析：
    • 投影 (Projection)：如果原始模型的视觉头输出维度（如 1024）与文本头（如 4096）不一致，此模块会通过线性层 (nn.Linear) 将它们强制投影到 config.py 定义的 D_MODEL 维度。
    • 归一化 (Normalization)：执行 L2 归一化。这是跨模态检索精度的保证，它确保“一只猫”的图片向量和“cat”的文本向量落在同一个超球面上，使其余弦相似度可计算。

3. 记忆与检索 (The Memory & Retrieval)

• core/retriever.py (混合记忆)
  • 用途：跨模态检索引擎。
  • 深度解析：它不区分数据来源。在 ChromaDB 中，它将文本向量、图像向量和视频向量存储在同一个集合 (Collection) 中。
  • 工作流：当你输入一段文字“海边日落”时，此模块将其转化为向量，并在同一个数学空间中同时寻找距离最近的图片和视频。这证明了不同模态的数据已经在数学上实现了“统一”。
• embeddings_db/ (语义空间)
  • 用途：持久化存储。
  • 核心特点：这里的二进制文件不仅存储了向量，还存储了 modal_type (模态类型) 等元数据，允许我们在统一搜索的基础上进行特定模态的过滤（例如：只搜视频）。

4.2 核心配置文件（config.py）

运行

import os
from pathlib import Path

# 项目路径配置
PROJECT_ROOT = Path(__file__).parent
DATA_DIR = PROJECT_ROOT / "data"
EMBEDDINGS_DB_DIR = PROJECT_ROOT / "embeddings_db" / "chroma"
LOGS_DIR = PROJECT_ROOT / "logs"

# 创建必要目录
for dir_path in [DATA_DIR, EMBEDDINGS_DB_DIR, LOGS_DIR]:
dir_path.mkdir(parents=True, exist_ok=True)

# 模型配置
MODEL_DIR = Path("/data/models/gemini")
MODEL_NAME = "google/gemma-2-9b-it"
DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
D_MODEL = 2048 # 统一嵌入维度
MAX_SEQ_LEN = 4096 # 最大序列长度

# 多模态处理配置
VIDEO_FRAME_RATE = 1 # 视频抽帧帧率
IMAGE_SIZE = (256, 256) # 图像尺寸
PATCH_SIZE = 16 # 视觉Patch尺寸
NORMALIZE_EMBEDDINGS = True # 是否归一化嵌入向量

# 向量数据库配置
CHROMA_PERSIST_DIRECTORY = str(EMBEDDINGS_DB_DIR)
CHROMA_COLLECTION_NAME = "gemini_multimodal"
RETRIEVE_TOP_K = 5 # 跨模态检索返回数量

# Gemini API配置（备用，本地部署优先）
GEMINI_API_KEY = os.getenv("GEMINI_API_KEY", "")
GEMINI_API_ENDPOINT = "https://generativelanguage.googleapis.com/v1/models/gemini-1.5-flash:generateContent"

4.3 模型初始化（core/model_setup.py）

运行

import torch
from transformers import AutoModel, AutoTokenizer, AutoImageProcessor
from config import MODEL_DIR, MODEL_NAME, DEVICE, D_MODEL, MAX_SEQ_LEN

def setup_gemini_model():
"""初始化Gemini多模态模型"""
# 1. 加载文本Tokenizer
tokenizer = AutoTokenizer.from_pretrained(
MODEL_NAME,
cache_dir=str(MODEL_DIR),
padding_side="right",
truncation_side="right"
)

# 2. 加载视觉处理器（图像/视频）
image_processor = AutoImageProcessor.from_pretrained(
MODEL_NAME,
cache_dir=str(MODEL_DIR),
image_size=IMAGE_SIZE # 匹配config中的图像尺寸
)

# 3. 加载多模态模型
model = AutoModel.from_pretrained(
MODEL_NAME,
cache_dir=str(MODEL_DIR),
torch_dtype=torch.float16 if DEVICE == "cuda" else torch.float32,
device_map=DEVICE,
trust_remote_code=True # 加载自定义模型代码
)

# 4. 适配统一嵌入维度（若模型输出维度不符）
if model.config.hidden_size != D_MODEL:
model.embed_proj = torch.nn.Linear(model.config.hidden_size, D_MODEL).to(DEVICE)

model.eval() # 推理模式
print(f"Gemini模型初始化完成，设备：{DEVICE}")
return model, tokenizer, image_processor

# 单例模式加载模型（避免重复初始化）
_model, _tokenizer, _image_processor = None, None, None
def get_gemini_model():
global _model, _tokenizer, _image_processor
if _model is None:
_model, _tokenizer, _image_processor = setup_gemini_model()
return _model, _tokenizer, _image_processor

4.4 多模态预处理（core/modal_processor.py）

运行

import cv2
import numpy as np
import torch
from PIL import Image
from config import VIDEO_FRAME_RATE, IMAGE_SIZE, PATCH_SIZE, MAX_SEQ_LEN

class ModalProcessor:
def __init__(self, tokenizer, image_processor):
self.tokenizer = tokenizer
self.image_processor = image_processor

def process_text(self, text):
"""处理文本：Tokenize并截断"""
inputs = self.tokenizer(
text,
max_length=MAX_SEQ_LEN,
padding="max_length",
truncation=True,
return_tensors="pt"
)
return inputs

def process_image(self, image_path):
"""处理图像：加载→Resize→Patch化"""
# 加载图像
image = Image.open(image_path).convert("RGB")
# 图像预处理
inputs = self.image_processor(
images=image,
return_tensors="pt",
do_resize=True,
size=IMAGE_SIZE
)
return inputs

def process_video(self, video_path):
"""处理视频：抽帧→逐帧处理→拼接"""
# 1. 抽帧
cap = cv2.VideoCapture(video_path)
frames = []
frame_idx = 0
fps = cap.get(cv2.CAP_PROP_FPS)
sample_interval = int(fps / VIDEO_FRAME_RATE)

while cap.isOpened():
ret, frame = cap.read()
if not ret:
break
if frame_idx % sample_interval == 0:
# 转换为RGB并Resize
frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
frame = cv2.resize(frame, IMAGE_SIZE)
frames.append(Image.fromarray(frame))
frame_idx += 1
cap.release()

# 2. 预处理帧（限制最大帧数）
max_frames = MAX_SEQ_LEN // (IMAGE_SIZE[0] // PATCH_SIZE) **2
frames = frames[:max_frames]

# 3. 视频帧预处理
inputs = self.image_processor(
images=frames,
return_tensors="pt",
do_resize=True,
size=IMAGE_SIZE
)
# 添加时间维度标记
inputs["time_ids"] = torch.arange(len(frames)).unsqueeze(0)
return inputs

# 测试处理器
if __name__ == "__main__":
from core.model_setup import get_gemini_model
_, tokenizer, image_processor = get_gemini_model()
processor = ModalProcessor(tokenizer, image_processor)

# 测试文本处理
text_inputs = processor.process_text("红色跑车在公路上行驶")
print(f"文本Token形状：{text_inputs['input_ids'].shape}")

# 测试图像处理
image_inputs = processor.process_image("data/images/car.jpg")
print(f"图像Pixel值形状：{image_inputs['pixel_values'].shape}")

# 测试视频处理
video_inputs = processor.process_video("data/videos/car_driving.mp4")
print(f"视频Pixel值形状：{video_inputs['pixel_values'].shape}")
print(f"视频时间ID形状：{video_inputs['time_ids'].shape}")

4.5 生成统一嵌入（core/embedding.py）

运行

import torch
from config import DEVICE, D_MODEL, NORMALIZE_EMBEDDINGS
from core.modal_processor import ModalProcessor

class MultimodalEmbedding:
def __init__(self):
from core.model_setup import get_gemini_model
self.model, self.tokenizer, self.image_processor = get_gemini_model()
self.processor = ModalProcessor(self.tokenizer, self.image_processor)

@torch.no_grad()
def get_text_embedding(self, text):
"""生成文本的统一嵌入"""
inputs = self.processor.process_text(text).to(DEVICE)
# 前向传播获取隐藏状态
outputs = self.model(**inputs)
# 取<CLS> token的嵌入（或均值池化）
embedding = outputs.last_hidden_state[:, 0, :]
# 适配统一维度
if hasattr(self.model, "embed_proj"):
embedding = self.model.embed_proj(embedding)
# 归一化
if NORMALIZE_EMBEDDINGS:
embedding = torch.nn.functional.normalize(embedding, p=2, dim=–1)
return embedding.cpu().numpy()[0]

@torch.no_grad()
def get_image_embedding(self, image_path):
"""生成图像的统一嵌入"""
inputs = self.processor.process_image(image_path).to(DEVICE)
# 视觉前向传播
outputs = self.model.visual_model(**inputs)
# 均值池化获取图像嵌入
embedding = outputs.last_hidden_state.mean(dim=1)
# 适配统一维度
if hasattr(self.model, "embed_proj"):
embedding = self.model.embed_proj(embedding)
# 归一化
if NORMALIZE_EMBEDDINGS:
embedding = torch.nn.functional.normalize(embedding, p=2, dim=–1)
return embedding.cpu().numpy()[0]

@torch.no_grad()
def get_video_embedding(self, video_path):
"""生成视频的统一嵌入"""
inputs = self.processor.process_video(video_path).to(DEVICE)
# 视频前向传播（时空编码）
visual_outputs = self.model.visual_model(**inputs)
# 时空均值池化（时间+空间维度）
embedding = visual_outputs.last_hidden_state.mean(dim=[1, 2])
# 适配统一维度
if hasattr(self.model, "embed_proj"):
embedding = self.model.embed_proj(embedding)
# 归一化
if NORMALIZE_EMBEDDINGS:
embedding = torch.nn.functional.normalize(embedding, p=2, dim=–1)
return embedding.cpu().numpy()[0]

# 测试嵌入生成
if __name__ == "__main__":
embedder = MultimodalEmbedding()

# 生成文本嵌入
text_emb = embedder.get_text_embedding("红色跑车在公路上行驶")
print(f"文本嵌入维度：{text_emb.shape}") # 应为 (2048,)

# 生成图像嵌入
image_emb = embedder.get_image_embedding("data/images/car.jpg")
print(f"图像嵌入维度：{image_emb.shape}") # 应为 (2048,)

# 生成视频嵌入
video_emb = embedder.get_video_embedding("data/videos/car_driving.mp4")
print(f"视频嵌入维度：{video_emb.shape}") # 应为 (2048,)

# 计算跨模态相似度
text_image_sim = np.dot(text_emb, image_emb)
text_video_sim = np.dot(text_emb, video_emb)
print(f"文本-图像相似度：{text_image_sim:.4f}")
print(f"文本-视频相似度：{text_video_sim:.4f}")

4.6 跨模态检索（core/retriever.py）

运行

import chromadb
import numpy as np
from chromadb.config import Settings
from config import CHROMA_PERSIST_DIRECTORY, CHROMA_COLLECTION_NAME, RETRIEVE_TOP_K

class MultimodalRetriever:
def __init__(self):
# 初始化Chroma向量数据库
self.client = chromadb.Client(Settings(
persist_directory=CHROMA_PERSIST_DIRECTORY,
anonymized_telemetry=False
))
# 获取或创建多模态集合
self.collection = self.client.get_or_create_collection(
name=CHROMA_COLLECTION_NAME,
metadata={"description": "Gemini多模态统一嵌入集合"}
)

def add_embedding(self, id, embedding, modal_type, metadata=None):
"""添加嵌入到向量库"""
self.collection.add(
ids=[id],
embeddings=[embedding],
metadatas=[{"modal_type": modal_type, **(metadata or {})}]
)
self.client.persist()

def retrieve(self, query_embedding, filter_modal_type=None):
"""跨模态检索相似内容"""
# 构建过滤条件
where_clause = None
if filter_modal_type:
where_clause = {"modal_type": filter_modal_type}

# 检索
results = self.collection.query(
query_embeddings=[query_embedding],
n_results=RETRIEVE_TOP_K,
where=where_clause
)
# 格式化结果
formatted_results = []
for i in range(len(results["ids"][0])):
formatted_results.append({
"id": results["ids"][0][i],
"score": results["distances"][0][i],
"modal_type": results["metadatas"][0][i]["modal_type"],
"metadata": results["metadatas"][0][i]
})
return formatted_results

# 测试跨模态检索
if __name__ == "__main__":
from core.embedding import MultimodalEmbedding

# 初始化嵌入器和检索器
embedder = MultimodalEmbedding()
retriever = MultimodalRetriever()

# 1. 添加测试数据到向量库
# 文本嵌入
text_id = "text_1"
text = "红色跑车在公路上行驶"
text_emb = embedder.get_text_embedding(text)
retriever.add_embedding(
id=text_id,
embedding=text_emb,
modal_type="text",
metadata={"content": text}
)

# 图像嵌入
image_id = "image_1"
image_path = "data/images/car.jpg"
image_emb = embedder.get_image_embedding(image_path)
retriever.add_embedding(
id=image_id,
embedding=image_emb,
modal_type="image",
metadata={"path": image_path}
)

# 视频嵌入
video_id = "video_1"
video_path = "data/videos/car_driving.mp4"
video_emb = embedder.get_video_embedding(video_path)
retriever.add_embedding(
id=video_id,
embedding=video_emb,
modal_type="video",
metadata={"path": video_path}
)

# 2. 跨模态检索（文本查询→找相似图像/视频）
query_text = "红色跑车在路上开"
query_emb = embedder.get_text_embedding(query_text)
results = retriever.retrieve(query_emb)

# 3. 打印检索结果
print(f"查询文本：{query_text}")
print("跨模态检索结果：")
for res in results:
print(f"- ID: {res['id']}, 类型: {res['modal_type']}, 相似度: {1–res['score']:.4f}")

4.7 主程序（main.py）

运行

import logging
from loguru import logger
from config import LOGS_DIR
from core.embedding import MultimodalEmbedding
from core.retriever import MultimodalRetriever

# 配置日志
logger.add(
LOGS_DIR / "multimodal.log",
rotation="100MB",
retention="7 days",
encoding="utf-8",
level="INFO"
)

def main():
logger.info("启动Gemini多模态统一表示系统")

# 初始化组件
embedder = MultimodalEmbedding()
retriever = MultimodalRetriever()

# 交互菜单
print("=== Gemini多模态统一表示系统 ===")
print("1. 生成文本嵌入")
print("2. 生成图像嵌入")
print("3. 生成视频嵌入")
print("4. 跨模态检索")
print("5. 退出")

while True:
choice = input("\\n请选择功能（1-5）：")
try:
if choice == "1":
text = input("请输入文本：")
emb = embedder.get_text_embedding(text)
logger.info(f"生成文本嵌入：{text[:20]}… 维度：{emb.shape}")
print(f"文本嵌入生成成功，维度：{emb.shape}")

elif choice == "2":
image_path = input("请输入图像路径：")
emb = embedder.get_image_embedding(image_path)
logger.info(f"生成图像嵌入：{image_path} 维度：{emb.shape}")
print(f"图像嵌入生成成功，维度：{emb.shape}")

elif choice == "3":
video_path = input("请输入视频路径：")
emb = embedder.get_video_embedding(video_path)
logger.info(f"生成视频嵌入：{video_path} 维度：{emb.shape}")
print(f"视频嵌入生成成功，维度：{emb.shape}")

elif choice == "4":
query_type = input("请选择查询类型（text/image/video）：")
if query_type == "text":
query = input("请输入查询文本：")
query_emb = embedder.get_text_embedding(query)
elif query_type == "image":
query = input("请输入查询图像路径：")
query_emb = embedder.get_image_embedding(query)
elif query_type == "video":
query = input("请输入查询视频路径：")
query_emb = embedder.get_video_embedding(query)
else:
print("无效的查询类型")
continue

# 检索
results = retriever.retrieve(query_emb)
print("\\n检索结果（按相似度排序）：")
for i, res in enumerate(results, 1):
similarity = 1 – res["score"] # Chroma返回的是距离，转换为相似度
print(f"{i}. ID: {res['id']}, 类型: {res['modal_type']}, 相似度: {similarity:.4f}")

elif choice == "5":
logger.info("退出系统")
print("再见！")
break

else:
print("无效的选择，请输入1-5")

except Exception as e:
logger.error(f"操作失败：{str(e)}", exc_info=True)
print(f"错误：{str(e)}")

if __name__ == "__main__":
main()

4.8 这些文件是如何协作的？

Gemini 多模态系统的运行逻辑是：“输入多模态，转化皆序列，输出皆向量”。

│
├── 【用户输入】
│ ├── 待处理数据: 文本 "夕阳下的海滩" / 图片 [img.jpg] / 视频 [vid.mp4]
│ └── 目的: 生成向量 (Embedding) 或 执行检索 (Retrieval)
│
▼
[1. 感知与序列化阶段 (Perception & Serialization)] ───────────┐
│ │
├── <调用模块>: core/modal_processor.py (序列化工厂) │
│ ├── 依赖配置: config.py (定义 PATCH_SIZE, MAX_SEQ_LEN) │
│ │ │
│ ├── A. 文本流 (Text Stream) │
│ │ ├── <工具>: Tokenizer (来自 model_setup.py) │
│ │ └── > 输出: 1D Token IDs [101, 2345, 889...] │
│ │ │
│ ├── B. 视觉流 (Visual Stream – 图像/视频) │
│ │ ├── <工具>: ImageProcessor (来自 model_setup.py) │
│ │ ├── <动作>: Resize –> Crop –> Patch化 │
│ │ │ ├── 图像: 16×16 Patch 展开 │
│ │ │ └── 视频: 时空采样 (Time-Space Sampling) │
│ │ └── > 输出: Visual Tokens (Pixel Values + Time IDs) │
│ │
└── > 合并状态: 统一的 PyTorch Tensor (适配 Transformer 输入) ┘
│
▼
[2. 统一编码阶段 (Unified Encoding)] <★ 核心/The Brain> ──────┐
│ │
├── <调用模块>: core/embedding.py (向量生成器) │
│ ├── 依赖模型: core/model_setup.py (加载共享 Transformer) │
│ │ └── 权重: google/gemma-2-9b-it │
│ │ │
│ ├── ⚙ 前向传播 (Forward Pass) │
│ │ ├── 输入: 异构的 Token 序列 │
│ │ ├── 机制: 共享自注意力 (Self-Attention) │
│ │ │ (文本Token与视觉Token在同一层进行交互) │
│ │ └── > 原始输出: [Batch, Seq_Len, Hidden_Dim] │
│ │ │
│ └── 📏 维度对齐 (Projection & Normalization) │
│ ├── 投影: Linear层将维度强制转为 config.D_MODEL │
│ ├── 归一化: L2 Normalize (确保模态间距离可比) │
│ └── > 最终产物: 统一向量 (Unified Vector, 2048维) │
│ │
└── > 输出结果: Numpy Array [0.12, -0.56, 0.99, ...] ────────┘
│
▼
[3. 存储与检索阶段 (Storage & Retrieval)] ────────────────────┐
│ │
├── <调用模块>: core/retriever.py (混合记忆管家) │
│ ├── 依赖存储: embeddings_db/chroma (向量数据库) │
│ │ │
│ ├── A. 写入模式 (Indexing) │
│ │ ├── 动作: 将向量 + 元数据 (Metadata) 存入集合 │
│ │ └── 结果: 持久化到磁盘 (Chroma Persist) │
│ │ │
│ └── B. 查询模式 (Searching) │
│ ├── 输入: 查询向量 (Query Vector) │
│ ├── 计算: 余弦相似度 (Cosine Similarity) │
│ │ (在同一空间内计算 文本-图 / 图-视频 的距离) │
│ └── > 最终结果: Top-K 匹配项 (如: 用文字搜到的视频) │
└────────────────────────────────────────────────────────────┘

4.2.2 协作细节深度解析

以下是这些代码文件在一次“跨模态检索”任务（例如：用文字搜视频）中的具体握手过程：

1. 预处理协作 (Processor & Config)

• 输入：用户输入文字 query = “红色跑车在公路上”。
• 协作逻辑：
  • main.py 接收指令，调用 core/modal_processor.py。
  • modal_processor.py 读取 config.py 中的 MAX_SEQ_LEN=4096，确保生成的序列不会爆显存。
  • 它向 core/model_setup.py 借用初始化好的 Tokenizer，将文字切分为数字 ID。
• 产物：一个标准的 PyTorch Tensor，准备送入显卡。

2. 核心编码协作 (Embedding & Model) 这是 Gemini 架构的灵魂，也是**“多模态统一”**发生的物理场所。

• 协作逻辑：
  • core/embedding.py 获取处理好的 Tensor，将其喂给 core/model_setup.py 加载的 共享 Transformer 模型。
  • 关键点：此时模型并不区分这是“文字”还是“视频”。在模型眼中，它们都是需要进行注意力计算的 Token。
  • 如果输入是视频，embedding.py 会特别处理输出的维度（对时间轴和空间轴取平均池化），并根据 config.py 中的 D_MODEL 强行将向量长度压制为 2048 维。
  • 最后，执行 L2 归一化。这一步至关重要，它把所有向量拉伸到单位长度，使得计算相似度（夹角余弦）成为可能。

3. 记忆检索协作 (Retriever & DB)

• 协作逻辑：
  • core/retriever.py 唤醒 embeddings_db/chroma 目录下的向量数据库。
  • 它接收 embedding.py 吐出的那个 2048 维向量。
  • 它在数据库中快速扫描几百万个已存储的向量（可能是图片生成的，也可能是视频生成的）。
  • 魔法时刻：因为它在数学上找到了距离最近的向量，而该向量的元数据指向一个 .mp4 文件，系统成功实现了“用文字搜索视频”。

总结

• config.py 是宪法：规定了所有模态必须遵守的尺寸和标准。
• model_setup.py 是大脑：提供了理解所有数据的共享神经网络权重。
• modal_processor.py 是翻译官：把人类能看懂的图/文/视，翻译成大脑能懂的序列。
• embedding.py 是压缩机：把庞大的序列压缩成一个精简的数学向量。
• retriever.py 是图书管理员：负责把这些向量分门别类地存好，并在需要时快速找到。

五、多模态表示的验证与优化

5.1 表示统一的验证方法

5.1.1 定性验证：跨模态语义一致性

通过人工标注的 “文本 – 图像 – 视频” 三元组，验证同一语义内容的向量相似度是否高于不同语义：

运行

def validate_semantic_consistency(embedder):
"""验证跨模态语义一致性"""
# 正例：同一语义的不同模态
positive_samples = {
"text": "小猫追着毛线球跑",
"image": "data/images/cat_ball.jpg",
"video": "data/videos/cat_ball.mp4"
}

# 负例：不同语义的模态
negative_samples = {
"text": "飞机在天空中飞行",
"image": "data/images/plane.jpg",
"video": "data/videos/plane_fly.mp4"
}

# 生成正例嵌入
pos_text_emb = embedder.get_text_embedding(positive_samples["text"])
pos_image_emb = embedder.get_image_embedding(positive_samples["image"])
pos_video_emb = embedder.get_video_embedding(positive_samples["video"])

# 生成负例嵌入
neg_text_emb = embedder.get_text_embedding(negative_samples["text"])

# 计算相似度
pos_text_image = np.dot(pos_text_emb, pos_image_emb)
pos_text_video = np.dot(pos_text_emb, pos_video_emb)
neg_text_image = np.dot(neg_text_emb, pos_image_emb)

print("=== 语义一致性验证 ===")
print(f"正例：文本-图像相似度 = {pos_text_image:.4f}")
print(f"正例：文本-视频相似度 = {pos_text_video:.4f}")
print(f"负例：文本-图像相似度 = {neg_text_image:.4f}")

# 验证阈值（理想情况下正例>0.7，负例<0.3）
assert pos_text_image > 0.7, "正例文本-图像相似度过低"
assert pos_text_video > 0.7, "正例文本-视频相似度过低"
assert neg_text_image < 0.3, "负例相似度过高"
print("验证通过！")

5.1.2 定量验证：检索准确率

使用标准多模态数据集（如 MSCOCO、Flickr30k）计算跨模态检索的 Top-K 准确率：

运行

def calculate_retrieval_accuracy(retriever, test_queries, test_ground_truth):
"""计算跨模态检索准确率"""
top1_correct = 0
top5_correct = 0
total = len(test_queries)

for query_id, query_emb in test_queries.items():
# 检索
results = retriever.retrieve(query_emb)
retrieved_ids = [res["id"] for res in results]

# 检查Top-1
if retrieved_ids[0] in test_ground_truth[query_id]:
top1_correct += 1

# 检查Top-5
if any(id in test_ground_truth[query_id] for id in retrieved_ids[:5]):
top5_correct += 1

# 计算准确率
top1_acc = top1_correct / total
top5_acc = top5_correct / total

print(f"Top-1 准确率：{top1_acc:.4f}")
print(f"Top-5 准确率：{top5_acc:.4f}")
return top1_acc, top5_acc

5.2 性能优化技巧

5.2.1 模型优化

5.2.2 推理优化

5.2.3 存储优化

六、生产级部署最佳实践

6.1 服务化封装（FastAPI）

运行

from fastapi import FastAPI, UploadFile, File
from fastapi.middleware.cors import CORSMiddleware
import uvicorn
import numpy as np
from core.embedding import MultimodalEmbedding
from core.retriever import MultimodalRetriever

# 初始化FastAPI
app = FastAPI(title="Gemini Multimodal API")
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)

# 预加载组件
embedder = MultimodalEmbedding()
retriever = MultimodalRetriever()

# 健康检查接口
@app.get("/health")
async def health_check():
return {"status": "healthy"}

# 文本嵌入接口
@app.post("/embed/text")
async def embed_text(text: str):
emb = embedder.get_text_embedding(text)
return {
"embedding": emb.tolist(),
"dim": emb.shape[0],
"modal_type": "text"
}

# 图像嵌入接口（支持文件上传）
@app.post("/embed/image")
async def embed_image(file: UploadFile = File(...)):
# 保存上传文件
file_path = f"temp/{file.filename}"
with open(file_path, "wb") as f:
f.write(await file.read())
# 生成嵌入
emb = embedder.get_image_embedding(file_path)
return {
"embedding": emb.tolist(),
"dim": emb.shape[0],
"modal_type": "image",
"filename": file.filename
}

# 跨模态检索接口
@app.post("/retrieve")
async def retrieve(embedding: list[float], modal_type: str = None):
query_emb = np.array(embedding)
results = retriever.retrieve(query_emb, filter_modal_type=modal_type)
return {"results": results}

if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8000)

6.2 监控与日志

6.3 扩展性设计

七、总结

Gemini 通过原生多模态设计、统一的序列化输入、共享的 Transformer 编码器、跨模态注意力机制，真正实现了文本、图像、视频的统一表示。其核心价值在于：

通过本文的实战代码，开发者可以快速搭建基于 Gemini 的多模态表示系统，并通过优化和部署技巧，将其落地为生产级应用。未来，随着 Gemini 模型的持续迭代，多模态表示的统一程度和应用场景还将进一步扩展。