当视频库增长到 14 万条以上，仅靠标题和描述检索，很快就会变成一项体力活。Condé Nast 的编辑团队平均需要约 250 分钟才能完成一次素材查找；在与 AWS Generative AI Innovation Center 合作，引入 Amazon Bedrock 和 Amazon OpenSearch Service 后，这一过程被压缩到 2 分钟以内。
关键变化不是把关键词搜索替换成聊天框，而是让检索系统同时理解画面、语义和视频时间点，并把返回结果定位到可直接预览的片段。
传统元数据为什么不够用
标题和描述只能表达上传者当时愿意记录的信息，无法完整覆盖视频中的视觉内容。例如，编辑可能需要寻找：
- 夜间城市街道上的黄色出租车；
- 厨师把酱汁浇到成品上的特写；
- 有红毯、闪光灯和人群的活动现场；
- 某位人物出现但没有被写进标题的历史镜头。
如果这些信息没有进入标题、标签或描述，传统倒排索引就无从匹配。编辑只能尝试多个关键词、打开候选视频，再拖动时间轴人工确认。
多模态检索改变了索引对象。系统不再只为“整条视频的一段文字”建立索引，而是可以把视频拆成多个时间片，为每个片段保存：
- 关键帧或代表性画面；
- 语音转写文本；
- 标题、栏目、人物、版权状态等业务元数据；
- 画面或文本对应的向量；
- 视频 ID、开始时间和预览地址。
这样，“找一段雨天街头的人群镜头”就能与画面语义匹配，而不必要求原始描述中恰好出现同样的词。
Bedrock 与 OpenSearch 各自承担什么
根据公开摘要，可以确定该方案使用了 Amazon Bedrock 和 Amazon OpenSearch Service。具体生产实现并未在摘要中展开，但工程上可以按下面的职责边界来理解和实践：
- 视频预处理：按固定间隔或镜头切换提取关键帧，并生成音频转写。
- 多模态向量化：调用 Bedrock 上支持图像与文本的嵌入模型，把关键帧、说明或搜索词映射到同一语义空间。
- 片段级索引：将向量和视频元数据写入 OpenSearch 的
knn_vector字段。 - 查询与重排：把自然语言查询转换为向量，执行近邻搜索，再结合版权、日期、栏目和地域等条件过滤。
- 编辑工作台：展示缩略图、时间码和预览片段，让编辑在搜索结果页直接判断素材是否可用。
这种设计中，向量搜索负责“含义相近”，结构化字段负责“业务上可用”。二者不能互相替代。例如，一个画面非常匹配，但授权已经到期，就不应该排在可用结果前面。
一个可改造的最小实现
下面的例子演示一条简化链路：从本地视频每 10 秒抽取一帧，调用 Bedrock 生成图像向量，将片段写入 OpenSearch，再使用自然语言搜索。
这只是可运行、可改造的参考实现，并不代表 Condé Nast 的完整生产架构。运行前需要准备 AWS 凭证、已获准调用的多模态嵌入模型，以及开启向量检索的 OpenSearch 域或集合。
1. 提取视频帧
安装 FFmpeg 后执行：
mkdir -p frames
ffmpeg -i input.mp4 \
-vf "fps=1/10,scale=640:-2" \
-q:v 2 \
"frames/demo__%06d.jpg"
这里每 10 秒保存一张图片。生产环境更适合使用镜头边界检测，否则静止场景会产生大量重复向量，快速剪辑的视频又可能漏掉重要画面。
2. 安装 Python 依赖
python -m venv .venv
source .venv/bin/activate
pip install boto3 opensearch-py
设置环境变量。托管 OpenSearch 域通常使用 es，OpenSearch Serverless 集合通常使用 aoss：
export AWS_REGION=us-east-1
export OPENSEARCH_URL=https://search-example.us-east-1.es.amazonaws.com
export OPENSEARCH_SERVICE=es
export VIDEO_ID=demo
3. 建立索引、写入帧并执行搜索
将下面内容保存为 video_search.py。示例使用 Amazon Titan Multimodal Embeddings G1；模型可用性、模型 ID 和向量维度可能因区域与账户配置而异，修改模型时必须同步调整 DIMENSION。
import base64
import glob
import json
import os
from pathlib import Path
import boto3
from opensearchpy import AWSV4SignerAuth, OpenSearch, RequestsHttpConnection
REGION = os.getenv("AWS_REGION", "us-east-1")
HOST = os.environ["OPENSEARCH_URL"].replace("https://", "").rstrip("/")
SERVICE = os.getenv("OPENSEARCH_SERVICE", "es")
VIDEO_ID = os.getenv("VIDEO_ID", "demo")
INDEX = "video-segments"
MODEL_ID = "amazon.titan-embed-image-v1"
DIMENSION = 1024
FRAME_INTERVAL_SECONDS = 10
session = boto3.Session(region_name=REGION)
credentials = session.get_credentials()
auth = AWSV4SignerAuth(credentials, REGION, SERVICE)
search = OpenSearch(
hosts=[{"host": HOST, "port": 443}],
http_auth=auth,
use_ssl=True,
verify_certs=True,
connection_class=RequestsHttpConnection,
timeout=60,
)
bedrock = session.client("bedrock-runtime")
def embed(*, text=None, image_path=None):
body = {"embeddingConfig": {"outputEmbeddingLength": DIMENSION}}
if text:
body["inputText"] = text
if image_path:
body["inputImage"] = base64.b64encode(
Path(image_path).read_bytes()
).decode("utf-8")
response = bedrock.invoke_model(
modelId=MODEL_ID,
contentType="application/json",
accept="application/json",
body=json.dumps(body),
)
payload = json.loads(response["body"].read())
return payload["embedding"]
def ensure_index():
if search.indices.exists(index=INDEX):
return
search.indices.create(
index=INDEX,
body={
"settings": {"index": {"knn": True}},
"mappings": {
"properties": {
"video_id": {"type": "keyword"},
"frame_path": {"type": "keyword"},
"offset_sec": {"type": "integer"},
"rights_status": {"type": "keyword"},
"embedding": {
"type": "knn_vector",
"dimension": DIMENSION,
"method": {
"name": "hnsw",
"space_type": "cosinesimil",
"engine": "faiss"
}
}
}
}
},
)
def index_frames():
files = sorted(glob.glob("frames/*.jpg"))
if not files:
raise SystemExit("frames 目录中没有 JPG 文件")
for position, filename in enumerate(files):
document = {
"video_id": VIDEO_ID,
"frame_path": filename,
"offset_sec": position * FRAME_INTERVAL_SECONDS,
"rights_status": "approved",
"embedding": embed(image_path=filename),
}
search.index(
index=INDEX,
id=f"{VIDEO_ID}-{position}",
body=document,
refresh=False,
)
print("indexed", filename)
search.indices.refresh(index=INDEX)
def query(text):
vector = embed(text=text)
response = search.search(
index=INDEX,
body={
"size": 5,
"_source": [
"video_id",
"frame_path",
"offset_sec",
"rights_status"
],
"query": {
"bool": {
"filter": [
{"term": {"rights_status": "approved"}}
],
"must": [
{
"knn": {
"embedding": {
"vector": vector,
"k": 20
}
}
}
]
}
}
},
)
for hit in response["hits"]["hits"]:
source = hit["_source"]
print(
f"score={hit['_score']:.4f} "
f"video={source['video_id']} "
f"time={source['offset_sec']}s "
f"frame={source['frame_path']}"
)
if __name__ == "__main__":
ensure_index()
index_frames()
query("夜晚街道上的黄色出租车和行人")
运行：
python video_search.py
实际部署时不要在每次查询前重复写入索引。更合理的方式是将视频入库与在线搜索拆成两个服务：前者通过队列异步处理，后者只负责查询和权限过滤。
从演示到生产，难点在检索质量
把向量写入 OpenSearch 并不等于完成了视频发现系统。真正影响编辑效率的通常是以下几项。
片段粒度
每 10 秒抽一帧实现简单，但未必准确。过密会增加 Bedrock 调用量、索引体积和重复结果；过疏则可能错过只出现两秒的重要镜头。可以结合镜头切换检测，并对长镜头设置最大采样间隔。
多路召回
只搜索画面会漏掉无法从像素判断的信息，例如人物身份、节目名称或采访内容。可以同时建立视觉向量、转写文本向量和关键词索引，再用加权分数或重排模型合并结果。
一个常见的评分思路是：
最终分数 = 0.55 × 视觉相似度
+ 0.30 × 转写文本相似度
+ 0.15 × 关键词匹配分数
这些权重只是起点，应通过真实编辑任务调优，而不是凭感觉固定。
权限与数据边界
媒体资产可能受到地域、时间、品牌和人物授权限制。索引文档应携带可过滤的权限字段，并在服务端强制执行；不能依赖前端隐藏按钮。还要避免把未授权预览地址、敏感转写或内部标签直接暴露给模型和终端用户。
可衡量的质量
“搜索看起来更智能”不是验收指标。可以从历史任务中构建测试集，持续跟踪：
- Top 5 或 Top 10 结果中是否包含目标片段；
- 编辑找到第一个可用片段所需的时间；
- 零结果率和重复结果率；
- 单条视频的处理成本；
- 从视频上传到可搜索的延迟。
Condé Nast 案例中最有价值的数字正是任务时间：从平均约 250 分钟降到 2 分钟以内。它直接反映系统是否改变了编辑工作流，而不仅仅是离线相似度指标是否变高。
落地时的检查清单
开始构建之前，可以先确认以下问题：
- 用户要找的是整条视频，还是某个精确时间片？
- 查询主要描述画面、对白、人物，还是版权和栏目属性？
- 是否需要把视觉、文本和关键词结果混合排序？
- 权限过滤是在向量检索前执行，还是在结果返回后补做？
- 抽帧和转写成本能否随视频库持续增长？
- 搜索结果能否直接跳转到正确时间码并快速预览？
- 是否有真实编辑任务组成的评测集？
多模态视频发现的价值不在于“给视频接入大模型”，而在于重新定义可搜索的最小单位，并把语义召回、业务过滤和片段预览串成完整工作流。Bedrock 提供多模态模型能力，OpenSearch 承担向量与元数据检索；只有两者与媒体处理、权限治理和编辑界面协同起来，分钟级发现才可能稳定进入日常生产。