如何为Vertex AI Search生成字典（非文本）的向量嵌入？

一、将字典数据转换为可嵌入的文本 #

嵌入模型仅接受文本输入，因此需要先把结构化的字典字段转换成连贯的文本字符串。比如针对产品类字典数据：

product = {
    "name": "无线蓝牙耳机",
    "brand": "XX科技",
    "price": 299,
    "description": "续航24小时，支持主动降噪，适配全平台设备"
}

可以通过自定义模板拼接文本：

def dict_to_text(data_dict: dict) -> str:
    # 按需选择要嵌入的字段，自定义拼接逻辑
    text_parts = [
        f"产品名称：{data_dict['name']}",
        f"品牌：{data_dict['brand']}",
        f"价格：{data_dict['price']}元",
        f"描述：{data_dict['description']}"
    ]
    return "\n".join(text_parts)

再将生成的文本传入嵌入模型：

from vertexai.language_models import TextEmbeddingModel

def generate_embedding_from_dict(data_dict: dict) -> list:
    model = TextEmbeddingModel.from_pretrained("textembedding-gecko@001")
    input_text = dict_to_text(data_dict)
    embeddings = model.get_embeddings([input_text])
    return embeddings[0].values

二、实现属性加权/降权 #

如果需要调整不同属性对最终嵌入向量的影响力，有两种可行方式：

1. 文本层面加权 #

通过重复高权重属性的文本内容，或在模板中强化该属性的呈现：

def dict_to_text_with_weight(data_dict: dict, weights: dict) -> str:
    # weights示例：{"description": 3, "name": 1, "brand": 0.5}
    text_parts = []
    for key, weight in weights.items():
        if weight <= 0:
            continue  # 跳过权重为0的属性
        content = f"{key}：{data_dict[key]}"
        # 按权重重复文本，权重小于1时仅保留原内容
        text_parts.append(content * int(weight) if weight >= 1 else content)
    return "\n".join(text_parts)

比如对产品描述加权3倍，生成的文本会包含3次描述内容，模型生成嵌入时会更侧重该属性的信息。

2. 向量层面加权 #

先为每个单独属性生成嵌入向量，再按设定权重加权合并：

def generate_weighted_embedding(data_dict: dict, weights: dict) -> list:
    model = TextEmbeddingModel.from_pretrained("textembedding-gecko@001")
    total_embedding = None
    total_weight = 0
    
    for key, weight in weights.items():
        if weight <= 0:
            continue
        # 生成单个属性的嵌入向量
        attr_text = f"{key}：{data_dict[key]}"
        embedding = model.get_embeddings([attr_text])[0].values
        # 对向量做加权处理并累加
        weighted_embedding = [val * weight for val in embedding]
        if total_embedding is None:
            total_embedding = weighted_embedding
        else:
            total_embedding = [a + b for a, b in zip(total_embedding, weighted_embedding)]
        total_weight += weight
    
    # 归一化处理，避免向量长度差异过大
    if total_weight > 0:
        total_embedding = [val / total_weight for val in total_embedding]
    return total_embedding

这种方式能更精准地控制每个属性在最终向量中的占比。

三、批量处理Spanner行数据 #

如果是从Spanner读取的多行字典数据，可以直接批量生成嵌入：

def batch_generate_embeddings(spanner_rows: list[dict]) -> list[list]:
    model = TextEmbeddingModel.from_pretrained("textembedding-gecko@001")
    # 先将所有行数据转换为文本
    input_texts = [dict_to_text(row) for row in spanner_rows]
    # 批量生成嵌入（注意模型有批量大小限制，需根据实际情况调整）
    embeddings = model.get_embeddings(input_texts)
    return [emb.values for emb in embeddings]

内容的提问来源于stack exchange，提问作者WJA