Amazon Nova 多模态嵌入模型实战指南

request_body = {
    "schemaVersion": "nova-multimodal-embed-v1", # 模型请求架构版本（默认 nova-multimodal-embed-v1）
    "taskType": "SINGLE_EMBEDDING", # 嵌入任务类型：同步调用时必须为 SINGLE_EMBEDDING             
    "singleEmbeddingParams": {
        # embeddingPurpose: 嵌入用途（根据使用场景优化效果）
        # 可选值:
        #   - GENERIC_INDEX        建立索引（index）
        #   - GENERIC_RETRIEVAL    搜索混合模态存储库
        #   - TEXT_RETRIEVAL       搜索仅含文本嵌入
        #   - IMAGE_RETRIEVAL      搜索仅含图像嵌入（STANDARD_IMAGE）
        #   - VIDEO_RETRIEVAL      搜索视频或音视频联合嵌入
        #   - DOCUMENT_RETRIEVAL   搜索文档（DOCUMENT_IMAGE）嵌入
        #   - AUDIO_RETRIEVAL      搜索音频嵌入
        #   - CLASSIFICATION       分类任务
        #   - CLUSTERING           聚类任务
        "embeddingPurpose": "GENERIC_INDEX",
        
        # embeddingDimension: 向量维度（默认 3072）
        # 可选值: 256 | 384 | 1024 | 3072
        "embeddingDimension": 3072,

        # text: 文本内容输入（text、image、video、audio 中必须且只能有一个）
        "text": {
            # truncationMode: 当文本超出最大长度时的截断策略
            #   START - 从开头截断
            #   END   - 从结尾截断
            #   NONE  - 超出则报错
            "truncationMode": "END",
            # value: 要生成嵌入的文本内容（最大 8192 字符）
            # 可与 source 互斥（必须二选一）
            "value": "string",
            # source: S3 文件引用（SourceObject 类型，参见通用对象定义）
            "source": "SourceObject"
        },
        # image: 图像内容输入
        "image": {
            # detailLevel: 图像分辨率层级（默认 STANDARD_IMAGE）
            #   STANDARD_IMAGE 低分辨率
            #   DOCUMENT_IMAGE 高分辨率，更适合含文本图像
            "detailLevel": "STANDARD_IMAGE",
            # format: 图像格式
            # 可选值: png | jpeg | gif | webp
            "format": "png",
            # source: 图像文件来源（SourceObject 类型）
            "source": "SourceObject"
        },
        # audio: 音频内容输入（最大长度 30 秒）
        "audio": {
            # format: 音频格式
            # 可选值: mp3 | wav | ogg
            "format": "mp3",
            # source: 音频文件来源（SourceObject 类型）
            "source": "SourceObject"
        },
        # video: 视频内容输入（最大长度 30 秒）
        "video": {
            # format: 视频格式
            # 可选值: mp4 | mov | mkv | webm | flv | mpeg | mpg | wmv | 3gp
            "format": "mp4",
            # source: 视频文件来源（SourceObject 类型）
            "source": "SourceObject",
            # embeddingMode: 视频嵌入模式
            #   AUDIO_VIDEO_COMBINED - 生成一个结合音频与视觉内容的嵌入
            #   AUDIO_VIDEO_SEPARATE - 分别生成音频与视觉的两个嵌入
            "embeddingMode": "AUDIO_VIDEO_COMBINED"
        }
    }
}

import json
import boto3
# 创建Bedrock Runtime客户端
bedrock_runtime = boto3.client(
    service_name="bedrock-runtime",
    region_name="us-east-1",
)
try:
    # 调用Nova Embeddings模型
    response = bedrock_runtime.invoke_model(
        body=json.dumps(request_body, indent=2), # 上面定义的payload
        modelId="amazon.nova-embedding-v1:0",
        accept="application/json",
        contentType="application/json",
    )
except Exception as e:
    print(e)

print("Request ID:", response.get("ResponseMetadata").get("RequestId"))
response_body = json.loads(response.get("body").read())
print(json.dumps(response_body, indent=2))

{
    "embeddings": [
        {
            # embeddingType: 嵌入向量类型
            # 可选值: TEXT | IMAGE | VIDEO | AUDIO | AUDIO_VIDEO_COMBINED
            "embeddingType": "TEXT",
            # embedding: 生成的向量（浮点数数组）
            "embedding": [0.123, -0.456, ...],
            # truncatedCharLength: （可选）若文本被截断，则返回被截断的字符位置
            "truncatedCharLength": 8120
        }
    ]
}

request_body = {
    "taskType": "SEGMENTED_EMBEDDING",
    "segmentedEmbeddingParams": {
        "embeddingPurpose": "GENERIC_INDEX",
        "embeddingDimension": 3072,
        "video": {
            "format": "mp4",
            "embeddingMode": "AUDIO_VIDEO_COMBINED",
            "source": {
                "s3Location": {"uri": "s3://<source-bucket>/path/to/video.mp4"}
            },
            "segmentationConfig": {
                "durationSeconds": 15  # 分段为15秒块
            },
        },
    },
}

import boto3
# 创建Bedrock Runtime客户端
bedrock_runtime = boto3.client(
    service_name="bedrock-runtime",
    region_name="us-east-1",
)
try:
    # 调用Nova Embeddings模型
    response = bedrock_runtime.start_async_invoke(
        modelId="amazon.nova-embedding-v1:0",
        modelInput=request_body,
        outputDataConfig={
            "s3OutputDataConfig": {
                "s3Uri": "s3://<destination-bucket>"
            }
        },
    )
except Exception as e:
    print(e)

print("Request ID:", response.get("ResponseMetadata").get("RequestId"))
print("Invocation ARN:", response.get("invocationArn")) # 打印调用ARN

Request ID：07681e80-5ce0-4723-cf52-68bf699cd23e
Invocation ARN：arn:aws:bedrock:us-east-1:<account-id>:async-invoke/g7ur3b32a10n

<destination-bucket>/
<job-id>/
segmented-embedding-result.json
embedding-audio.jsonl
embedding-image.json
embedding-text.jsonl
embedding-video.jsonl
manifest.json

# 仅展示核心代码，省略 0. 创建OpenSearch集群和索引
# 1. 用Nova MME生成文本嵌入
def get_nova_embedding(text):
    
    request = {
        "taskType": "SINGLE_EMBEDDING",
        "singleEmbeddingParams": {
            "embeddingPurpose": "TEXT_RETRIEVAL",
            "text": {"truncationMode": "END", "value": text}
        }
    } 
    response = bedrock.invoke_model(
        modelId=MODEL_ID,
        body=json.dumps(request)
    )
    return json.loads(response['body'].read())['embeddings'][0]['embedding']
    
# 2. 索引文档，存储到AWS OpenSearch
docs = [{"title": "机器学习基础", "content": "机器学习是人工智能的核心技术，包括监督学习、无监督学习和强化学习等方法。"}]
for i, doc in enumerate(docs):
    text = f"{doc['title']} {doc['content']}"
    embedding = get_nova_embedding(text)
    opensearch_client.index(
        index=INDEX_NAME,
        id=i+1,
        body={
            "title": doc["title"],
            "content": doc["content"],
            "embedding": embedding
        },
        refresh=True
    )
# 3. query文字用Nova MME生成嵌入，基于Opensearch的K-NN算法基于向量相似度排序返回topK结果     
def search_text(query, top_k=3):
    query_embedding = get_nova_embedding(query)
    search_query = {
        "size": top_k,
        "query": {
            "script_score": {
                "query": {"match_all": {}},
                "script": {
                    "source": "knn_score",
                    "lang": "knn",
                    "params": {
                        "field": "embedding",
                        "query_value": query_embedding,
                        "space_type": "cosinesimil"
                    }
                }
            }
        },
        "_source": ["title", "content"]
    }
    response = opensearch_client.search(body=search_query, index=INDEX_NAME)
    for i, hit in enumerate(response["hits"]["hits"]):
        print(f"{i+1}. {hit['_source']['title']} (相似度: {hit['_score']:.4f})")
        print(f"   {hit['_source']['content']}")
    return response["hits"]["hits"]
    
# demo：演示搜索链路
search_text("人工智能算法") # 返回："机器学习基础 (相似度: 1.6373)  机器学习是人工智能的核心技术...."

# 1. 用Nova MME 读取视频并生成 embedding
bedrock = boto3.client('bedrock-runtime', region_name='us-east-1')
with open('game_highlight.mp4', 'rb') as f:
    video_bytes = base64.b64encode(f.read()).decode('utf-8')
video_request = {
    "taskType": "SINGLE_EMBEDDING",
    "singleEmbeddingParams": {
        "embeddingPurpose": "GENERIC_INDEX",
        "embeddingDimension": 1024,
        "video": {
            "format": "mp4",
            "embeddingMode": "AUDIO_VIDEO_COMBINED",
            "source": {"bytes": video_bytes}
        }
    }
}
response = bedrock.invoke_model(
    modelId='amazon.nova-2-multimodal-embeddings-v1:0',
    body=json.dumps(video_request)
)
video_embedding = json.loads(response['body'].read())['embeddings'][0]['embedding']

# 2. 生成查询文本的 embedding
query_text = "新英格兰爱国者队多名球员对踢球手施压并阻挡，球滚出界外，进攻未果，失去球权"
text_request = {
    "taskType": "SINGLE_EMBEDDING",
    "singleEmbeddingParams": {
        "embeddingPurpose": "GENERIC_RETRIEVAL",
        "embeddingDimension": 1024,
        "text": {"truncationMode": "END", "value": query_text}
    }
}
response = bedrock.invoke_model(
    modelId='amazon.nova-2-multimodal-embeddings-v1:0',
    body=json.dumps(text_request)
)
text_embedding = json.loads(response['body'].read())['embeddings'][0]['embedding']

# 3. 计算余弦相似度
similarity = np.dot(text_embedding, video_embedding) / (
    np.linalg.norm(text_embedding) * np.linalg.norm(video_embedding)
)
print(f"相似度分数: {similarity:.4f}") #相似度分数: 0.6295

# 1. 用Nova MME生成人脸嵌入
def get_nova_embedding(image_path):
    with open(image_path, 'rb') as f:
        image_bytes = base64.b64encode(f.read()).decode('utf-8')
    img_format = "jpeg" if image_path.lower().endswith(('.jpg', '.jpeg')) else "png"
    request = {
        "taskType": "SINGLE_EMBEDDING",
        "singleEmbeddingParams": {
            "embeddingDimension": EMBEDDING_DIMENSION,
            "embeddingPurpose": "GENERIC_INDEX",
            "image": {
                "format": img_format,
                "detailLevel": "STANDARD_IMAGE",
                "source": {"bytes": image_bytes}
            }
        }
    }
    response = bedrock.invoke_model(
        modelId=MODEL_ID,
        body=json.dumps(request)
    )
    return json.loads(response['body'].read())['embeddings'][0]['embedding']

# 2. 索引人脸图像，存储到FAISS
gallery_images = [
    "sample_assets/images/andy_portrait_resized.jpg",
    "sample_assets/images/jeff_portrait.jpg"
]
embeddings_list = []
metadata = []
for i, img_path in enumerate(gallery_images):
    embedding = get_nova_embedding(img_path)
    embeddings_list.append(embedding)
    metadata.append({
        'filename': os.path.basename(img_path),
        'index': i
    })
index = faiss.IndexFlatL2(EMBEDDING_DIMENSION) # 构建FAISS索引
embeddings_array = np.array(embeddings_list, dtype=np.float32)
index.add(embeddings_array)

# 3. query人脸图像用Nova MME生成嵌入，基于FAISS的L2距离算法进行人脸检索
def search_portrait(query_image_path, top_k=2):
    query_embedding = get_nova_embedding(query_image_path)
    query_vector = np.array([query_embedding], dtype=np.float32)
    k = min(len(metadata), top_k)
    distances, indices = index.search(query_vector, k)
    for i, (distance, idx) in enumerate(zip(distances[0], indices[0])):
        print(f"{i+1}. {metadata[idx]['filename']} (距离: {distance:.4f})")   
    return distances, indices

# demo：演示人脸检索链路
distances, indices = search_portrait("sample_assets/images/andy_portrait_2.jpg")  # 返回：Andy最相似

# 1. 用 Nova MME 生成文档页面的 embedding
def get_document_page_embedding(image_base64):
    request = {
        "taskType": "SINGLE_EMBEDDING",
        "singleEmbeddingParams": {
            "embeddingPurpose": "GENERIC_INDEX",
            "embeddingDimension": 3072,
            "image": {
                "format": "png",
                "detailLevel": "DOCUMENT_IMAGE",  # 文档图像专用模式
                "source": {"bytes": image_base64}
            }
        }
    }
    response = bedrock_runtime.invoke_model(
        modelId="amazon.nova-2-multimodal-embeddings-v1:0",
        body=json.dumps(request)
    )
    return json.loads(response['body'].read())['embeddings'][0]['embedding']
    
# 2. 索引文档页面，存储到向量数据库
pages = convert_from_path("AMZN-Q2-2025-Earnings-Release-1.pdf", dpi=200)
for page_num, page_image in enumerate(pages, start=1):
    buffered = BytesIO()
    page_image.save(buffered, format="PNG")
    image_base64 = base64.b64encode(buffered.getvalue()).decode("utf-8")
    embedding = get_document_page_embedding(image_base64)
    collection.add(
        ids=[f"page_{page_num}"],
        embeddings=[embedding],
        metadatas=[{"page_number": page_num, "document": "AMZN-Q2-2025"}]
    )
    
# 3. 查询文本用 Nova MME 生成 embedding，基于向量相似度检索文档页面
def search_document(query_text, top_k=3):
    query_request = {
        "taskType": "SINGLE_EMBEDDING",
        "singleEmbeddingParams": {
            "embeddingPurpose": "DOCUMENT_RETRIEVAL",  # 文档检索专用
            "embeddingDimension": 3072,
            "text": {"truncationMode": "END", "value": query_text}
        }
    }
    response = bedrock_runtime.invoke_model(
        modelId="amazon.nova-2-multimodal-embeddings-v1:0",
        body=json.dumps(query_request)
    )
    query_embedding = json.loads(response['body'].read())['embeddings'][0]['embedding']
    results = collection.query(
        query_embeddings=[query_embedding],
        n_results=top_k
    )
    for i, (page_id, distance) in enumerate(zip(results["ids"][0], results["distances"][0])):
        metadata = results["metadatas"][0][i]
        similarity = 1 - (distance / 2)  # 余弦距离转相似度
        print(f"{i+1}. 页码: {metadata['page_number']} (相似度: {similarity:.4f})")  
    return results
    
# Demo：演示文档检索链路
search_document("AWS Q2的运营利润贡献了多少，相比去年变化如何？")
# 返回： 1. 页码: 11 (相似度: 0.6829)  2. 页码: 10 (相似度: 0.6725)  3. 页码: 1 (相似度: 0.6617)

# 技术文档数据库
technical_docs = [
    "反应堆退役程序包括系统隔离、燃料移除、设备拆除和场地清理四个主要阶段",
    "反应堆退役后的下一步骤是进行详细的环境影响评估和制定废料处理计划",
    "核设施退役需要获得监管部门批准并建立长期监测体系",
    "反应堆拆除工作需要专业团队使用远程操作设备进行辐射防护",
    "退役完成后必须进行土壤修复和地下水监测工作",
    "操作流程要求建立应急响应机制和人员培训计划"
]

Query: '反应堆退役后操作流程的下一步骤' 
Knowledge Retrieval Results: 
================================================== 
Rank 1 (Relevance: 0.7829):  反应堆退役后的下一步骤是进行详细的环境影响评估和制定废料处理计划 
Rank 2 (Relevance: 0.7444):  反应堆退役程序包括系统隔离、燃料移除、设备拆除和场地清理四个主要阶段 
Rank 3 (Relevance: 0.6781):  核设施退役需要获得监管部门批准并建立长期监测体系

# 0. 配置（略，参考前述章节“图像分类”或“图像检索”）

# 1. 用Nova MME生成图像嵌入与文本嵌入
def get_image_embedding(image_path):
    #（略，参考前述章节“图像分类”或“图像检索”）
    pass
 
def get_text_embedding(query_text):
    #（略，参考前述章节“知识检索增强”）
    pass
 
# 2. 索引风景图像，构建图像库
gallery_images = [
 "sample_assets/images/scenery1.webp",
 "sample_assets/images/scenery2.jpg", 
 "sample_assets/images/scenery3.webp",
 "sample_assets/images/scenery4.webp",
 "sample_assets/images/scenery5.jpg"
]

image_embeddings = []
image_metadata = []

for img_path in gallery_images:
 embedding = get_image_embedding(img_path)
 image_embeddings.append(embedding)
 image_metadata.append(img_path.split('/')[-1])

print(f"Generated {len(image_embeddings)} image embeddings")

# 3. 跨模态文本查询，基于文本描述检索相似图像
def search_images_by_text(query_text, top_k=3):
    # 生成文本嵌入，优化图像检索
    text_embedding = get_text_embedding(query_text)
    
    # 计算跨模态相似度
    similarities = []
    for i, img_embedding in enumerate(image_embeddings):
    similarity = np.dot(text_embedding, img_embedding) / (
    np.linalg.norm(text_embedding) * np.linalg.norm(img_embedding)
    )
    similarities.append((similarity, i))
    
    # 排序并返回top-k结果
    similarities.sort(reverse=True)
    
    for i, (similarity, idx) in enumerate(similarities[:top_k]):
    print(f"{i+1}. {image_metadata[idx]} (相似度: {similarity:.4f})")
    
    return similarities[:top_k]

# demo：演示跨模态检索链路
results = search_images_by_text("日落海滩") # 返回：最相似的风景图像

# 0. 配置
MODEL_ID = "amazon.nova-2-multimodal-embeddings-v1:0"
REGION = "us-east-1"
EMBEDDING_DIMENSION = 1024
MAX_AUDIO_DURATION_MS = 30000  # 30秒限制
bedrock = boto3.client('bedrock-runtime', region_name=REGION)

# 1. 用Nova MME生成音频嵌入
def get_audio_embedding(audio_path, purpose="GENERIC_INDEX"):
    with open(audio_path, 'rb') as f:
        audio_bytes = base64.b64encode(f.read()).decode('utf-8')
    audio_format = os.path.splitext(audio_path)[1][1:].lower()
    request = {
        "taskType": "SINGLE_EMBEDDING",
        "singleEmbeddingParams": {
            "embeddingPurpose": purpose,
            "embeddingDimension": EMBEDDING_DIMENSION,
            "audio": {
                "format": audio_format,
                "source": {"bytes": audio_bytes}
            }
        }
    }
    response = bedrock.invoke_model(modelId=MODEL_ID, body=json.dumps(request))
    return json.loads(response['body'].read())['embeddings'][0]['embedding']

# 2. 索引音频文件，构建音频库
audio_library = [
    "sample_assets/audio/audio1_beat.mp3",
    "sample_assets/audio/audio2_positive1.mp3",
    "sample_assets/audio/audio3_rock.mp3",
    "sample_assets/audio/audio4_positive2.mp3",
    "sample_assets/audio/audio5_sad1.mp3",
    "sample_assets/audio/audio5_sad2.mp3"
]
audio_embeddings = []
audio_metadata = []
for audio_path in audio_library:
    embedding = get_audio_embedding(audio_path, "GENERIC_INDEX")
    audio_embeddings.append(embedding)
    audio_metadata.append({
        'filename': os.path.basename(audio_path),
        'genre': audio_path.split('_')[1] if '_' in audio_path else 'unknown'
    })
print(f"Generated {len(audio_embeddings)} audio embeddings")

# 3. 音频片段检索，基于音频片段查找相似音频
def search_audio_by_segment(source_audio_path, start_ms=10000, duration_ms=10000, top_k=3):
    if duration_ms > MAX_AUDIO_DURATION_MS:  # 创建音频片段
        duration_ms = MAX_AUDIO_DURATION_MS
    audio = AudioSegment.from_file(source_audio_path)
    segment = audio[start_ms:start_ms + duration_ms]
    temp_file = tempfile.NamedTemporaryFile(suffix='.wav', delete=False)
    segment.export(temp_file.name, format='wav')
    segment_embedding = get_audio_embedding(temp_file.name)  # 生成片段嵌入
    similarities = []  # 计算相似度
    for i, audio_emb in enumerate(audio_embeddings):
        similarity = np.dot(segment_embedding, audio_emb) / (
            np.linalg.norm(segment_embedding) * np.linalg.norm(audio_emb)
        )
        similarities.append((similarity, i))
    similarities.sort(reverse=True) # 排序并返回top-k结果
    print(f"Audio segment from {os.path.basename(source_audio_path)}:")
    for rank, (sim, idx) in enumerate(similarities[:top_k], 1):
        meta = audio_metadata[idx]
        print(f"  {rank}. {meta['filename']} - {sim:.4f}")
    os.unlink(temp_file.name)
    return similarities[:top_k]

# demo：演示音频片段检索链路
for audio_path in audio_library[:3]:  # 测试前三个音频
    search_audio_by_segment(audio_path)  # 返回：最相似的音频文件

# 0. 配置
#（略，参考“音频/语音检索-1.听音识曲”部分)

# 1. 用Nova MME生成音频嵌入和文本嵌入
def get_audio_embedding(audio_path, purpose="GENERIC_INDEX"):
    #（略，参考“音频/语音检索-1.听音识曲”部分)
    pass
def get_text_embedding(query_text):
    #（略，参考前述章节“检索链路构建-Nova MME集成的检索链路----创建AWS OpenSearch索引，Nova MME生成嵌入，用OpenSearch支持的k-NN算法实现文本到文本检索”
    pass

# 2. 索引音频文件，构建音频库
#（略，参考“音频/语音检索-1.听音识曲”部分)

# 3. 跨模态文本查询，基于中文描述检索相似音频
def search_audio_by_text(query_text, top_k=3):
    text_embedding = get_text_embedding(query_text)
    similarities = []  # 计算跨模态相似度
    for i, audio_emb in enumerate(audio_embeddings):
        similarity = np.dot(text_embedding, audio_emb) / (
            np.linalg.norm(text_embedding) * np.linalg.norm(audio_emb)
        )
        similarities.append((similarity, i))
    similarities.sort(reverse=True)  # 排序并返回top-k结果
    print(f"查询: {query_text}")
    for rank, (sim, idx) in enumerate(similarities[:top_k], 1):
        print(f"  {rank}. {audio_metadata[idx]['filename']} - {sim:.4f}")
    return similarities[:top_k]

# demo：演示跨模态文本-音频检索链路
queries = ["节拍感强的电子音乐", "积极快乐的歌曲", "悲伤的音乐"]
for query in queries:
    search_audio_by_text(query)