悟界·RoboBrain Orca 在跨模态检索服务中的索引构建方法

跨模态检索服务的索引构建，核心是把 RoboBrain Orca 这类多模态模型产出的图像向量与文本向量组织成统一可查的结构。本文给出的流程不绑定具体向量库，覆盖从样本准备、特征提取、集合创建、批量写入到召回验证的整条链路，重点是让你能先跑通一次验证。

判断：适用于以文搜图、图文互检等场景。处理方向：按图文对准备数据、统一调用特征接口、在向量库中建集合并以余弦度量写入、用 count 和 TopK 召回验证。边界：向量维度需与模型输出一致；度量方式应与模型训练损失匹配；若文本与图像编码不共享空间，检索无效。

先从数据源准备成对的图文样本

索引构建的前提是每一条样本同时包含图像和文本标识，缺一不可。建议先维护一张数据表，字段至少包括：

sample_id    -- 全局唯一
image_path   -- 图像文件路径或URL
text         -- 对应文本描述
split        -- train/eval 划分标记

批量导出样本时，直接从数据库或CSV读取，并过滤掉 image_path 为空或 text 为空的记录。示例脚本：

import csv
with open('pairs.csv', 'r') as f:
    reader = csv.DictReader(f)
    samples = [row for row in reader if row['image_path'] and row['text'].strip()]
# 后续处理直接用 samples

这一步的关键是保证样本成对，否则后面生成的向量无法对齐。

调用特征提取接口生成向量

对每个样本，分别调用图像编码器与文本编码器，得到统一维度的向量。RoboBrain Orca 的接口可能封装为以下形式：

from inference import encode_image, encode_text

for s in samples:
    img_vec = encode_image(s['image_path'])   # shape (D,)
    txt_vec = encode_text(s['text'])          # shape (D,)
    # 保存时同时记录 sample_id
    yield {'sample_id': s['sample_id'], 'image_vector': img_vec, 'text_vector': txt_vec}

输出建议保存为 npy 或 jsonl。若用 jsonl，每行一条记录，便于后续流式写入。

import json
with open('vectors.jsonl', 'w') as f:
    for rec in generate_vectors(samples):
        f.write(json.dumps(rec) + '\n')

注意保留 sample_id，且向量维度必须固定，常见为 512、768 或 1024，具体以模型输出为准。

在向量库中创建集合并指定维度与距离函数

向量库选型不同，但集合创建的核心参数一致：字段名、向量维度、距离函数。以伪 SQL 为例：

CREATE COLLECTION sample_vectors (
    id VARCHAR,
    image_vector VECTOR(DIM=512),
    text_vector  VECTOR(DIM=512)
) DISTANCE_METRIC = COSINE;

维度必须与特征接口输出一致，否则写入会被拒绝。度量方式通常优先选 COSINE，因为多数多模态模型训练时使用余弦相似度或对比损失，检索结果与模型语义对齐更好。若模型明确用点积或欧氏距离训练，则对应调整。可以先在一个小集上分别用 COSINE 和 L2 测试召回效果再决定。

以 batch 模式写入向量并验证条数

单条写入太慢，建议按 batch 写入。通用伪代码：

from vector_db import client

batch = []
for rec in vectors_jsonl:
    row = {
        'id': rec['sample_id'],
        'image_vector': rec['image_vector'],
        'text_vector': rec['text_vector']
    }
    batch.append(row)
    if len(batch) == 100:
        client.insert(batch)
        batch = []
if batch:
    client.insert(batch)

入库后立即查询总数，与源样本列表长度比对：

count = client.count('sample_vectors')
print('入库条数:', count, '源样本数:', len(samples))
assert count == len(samples)

如果数量不一致，优先检查是否有重复 sample_id 或向量格式异常。

用测试 query 做召回验证

索引构建完成后，用一条文本 query 验证能否召回对应图像。示例：

query_vec = encode_text('一只橘猫坐在窗台上')
results = client.search(
    collection='sample_vectors',
    vector_field='image_vector',
    query_vector=query_vec,
    top_k=5
)
for r in results:
    print(r['id'], r['similarity'])

判断结果是否合理的标准：与 query 语义相关的图像应排在前面，且相似度明显高于无关项。如果 topK 全是无关样本，需要检查文本向量与图像向量是否在同一向量空间，或度量方式是否与模型训练目标一致。