RAG 在跨境电商选品中的应用：多语言产品描述的向量化与竞品分析

RAG 在跨境电商选品中的应用多语言产品描述的向量化与竞品分析一、深度引言与场景痛点大家好我是赵咕咕。做过跨境电商的朋友都知道选品是最重要也最痛苦的事。你要做的事情本质是在全球几百个品类、几千个竞品中找到需求在涨、竞争还小、利润空间够的产品。传统选品的做法是靠人肉刷——看 Amazon Best Seller、翻 AliExpress 热销榜、读 1688 的供应商页面、扒独立站的评论。一个选品经理一天最多看 300-500 个产品大量信息因为语言障碍、格式不统一而被忽略。这个场景跟 RAG 天然契合。产品描述有中文、英文、日文、德文……语言不统一。竞品分析要跨平台比较数据格式千奇百怪。靠关键词搜索完全不够用——户外便携式折叠椅和Outdoor Folding Camping Chair在关键词系统里是两个东西但在语义上是同一个品类。这篇文章我分享怎么用多语言 Embedding 向量检索构建一个跨境电商选品和竞品分析的 RAG 系统。二、底层机制与原理深度剖析2.1 选品 RAG 的特殊挑战跟普通文档 RAG 相比电商选品 RAG 有三个独特挑战多语言一份产品数据可能包含中文标题、日文描述、英文评论。如果用单语言 Embedding 模型如text-embedding-ada-002跨语言检索效果很差——中文防水运动手表跟英文Waterproof Sport Watch在向量空间里可能相距很远。多模态产品信息不只有文字——主图、细节图、视频里的视觉信息也是选品判断的重要依据。图文多模态 Embedding 是必须的。结构化 非结构化混合产品有价格、销量、评分这些结构化字段也有描述、评论、QA 这些非结构化文本。检索既要语义匹配长文本描述相似度也要精确过滤价格在 $10-$30、评分 4.0。2.2 多语言选品 RAG 架构关键设计点BGE-M3 做多语言 Embedding这是 BAAI 开发的多语言 Embedding 模型支持 100 语言且在同一向量空间中对齐。中英文同一个语义会落到相近的位置。不需要翻译就能做跨语言检索。混合表征向量不只是文本 Embedding而是文本 Embedding 图片 Embedding 拼接 结构化特征编码价格分桶、评分归一化。这样检索时间同时考虑语义、视觉和商业数据。Payload 过滤Qdrant 的 payload 过滤用来做精确条件筛选价格范围、评分门槛向量检索做语义匹配两者并行而非串行。2.3 多语言 Embedding 的对齐原理多语言 Embedding 模型如 BGE-M3、LaBSE在训练时会同时输入中文-英文平行句对强迫模型把同语义的不同语言文本映射到相近的向量位置。训练完成后中文蓝牙降噪耳机的向量和英文Bluetooth Noise Cancelling Earbuds的向量在余弦距离上是接近的。这意味着检索时不需要翻译——直接用中文查询去搜英文产品描述就能拿到语义相关的结果。三、生产级代码实现import asyncio import hashlib import logging from dataclasses import dataclass, field from enum import Enum from typing import Any from qdrant_client import QdrantClient from qdrant_client.http import models as qdrant_models from qdrant_client.models import Distance, VectorParams, Filter, FieldCondition, Range import numpy as np logger logging.getLogger(__name__) # ── 数据模型 ─────────────────────────────────────────── class ProductSource(Enum): AMAZON amazon ALIEXPRESS aliexpress SHOPIFY shopify dataclass class ProductInfo: 跨平台统一产品信息模型。 product_id: str source: ProductSource title: str description: str # 多语言字段 title_en: str description_en: str # 结构化数据 price_usd: float 0.0 rating: float 0.0 review_count: int 0 category: str # 图片 image_urls: list[str] field(default_factorylist) # 计算字段 embedding: list[float] | None None chunk_id: str def __post_init__(self): if not self.chunk_id: raw f{self.source.value}:{self.product_id} self.chunk_id hashlib.sha256(raw.encode()).hexdigest()[:16] dataclass class SearchResult: 检索结果。 product: ProductInfo score: float highlights: list[str] field(default_factorylist) # ── 多语言 Embedding ─────────────────────────────────── class MultiLingualEmbedder: 多语言产品 Embedding 生成器。 def __init__(self, model_name: str BAAI/bge-m3): self._model_name model_name self._model: Any None self._initialized False async def initialize(self) - None: 延迟加载模型避免首次调用阻塞。 if self._initialized: return # BGE-M3 通过 FlagEmbedding 库使用这里用 sentence-transformers 做 compat 层 try: from FlagEmbedding import BGEM3FlagModel self._model await asyncio.to_thread( BGEM3FlagModel, self._model_name, use_fp16True ) except ImportError: from sentence_transformers import SentenceTransformer self._model await asyncio.to_thread( SentenceTransformer, self._model_name ) self._initialized True logger.info(多语言 Embedding 模型 %s 加载完成, self._model_name) async def encode_product(self, product: ProductInfo) - list[float]: 为产品生成多语言向量。 await self.initialize() # 构建多语言拼接文本中文 英文 结构化特征 text ( fTitle: {product.title}\n fTitle(EN): {product.title_en}\n fDescription: {product.description[:500]}\n fDescription(EN): {product.description_en[:500]}\n fCategory: {product.category}\n fPrice: ${product.price_usd:.2f}\n fRating: {product.rating}/5.0 ({product.review_count} reviews) ) if hasattr(self._model, encode): # BGE-M3 原生 API embedding await asyncio.to_thread( self._model.encode, [text], return_denseTrue, return_sparseFalse ) vec embedding[dense_vecs][0] else: # SentenceTransformer vec await asyncio.to_thread( self._model.encode, text, normalize_embeddingsTrue ) return vec.tolist() # ── 向量数据库管理 ───────────────────────────────────── class ProductVectorStore: 产品向量存储与检索。 COLLECTION_NAME cross_border_products def __init__(self, embedder: MultiLingualEmbedder | None None): self._client QdrantClient(path./qdrant_ecommerce) self._embedder embedder or MultiLingualEmbedder() async def initialize_collection(self, vector_size: int 1024) - None: 初始化/重建向量集合。 if self._client.collection_exists(self.COLLECTION_NAME): logger.info(向量集合 %s 已存在, self.COLLECTION_NAME) return self._client.create_collection( collection_nameself.COLLECTION_NAME, vectors_configVectorParams( sizevector_size, distanceDistance.COSINE, ), ) # 创建 payload 索引用于过滤查询 self._client.create_payload_index( collection_nameself.COLLECTION_NAME, field_nameprice_usd, field_schemaqdrant_models.FloatIndexParams( typeqdrant_models.FloatIndexType.FLOAT, ), ) self._client.create_payload_index( collection_nameself.COLLECTION_NAME, field_namerating, field_schemaqdrant_models.FloatIndexParams( typeqdrant_models.FloatIndexType.FLOAT, ), ) self._client.create_payload_index( collection_nameself.COLLECTION_NAME, field_namesource, field_schemaqdrant_models.KeywordIndexParams( typeqdrant_models.KeywordIndexType.KEYWORD, ), ) logger.info(向量集合 %s 创建完成, self.COLLECTION_NAME) async def index_products(self, products: list[ProductInfo]) - None: 批量索引产品。 await self._embedder.initialize() batch_size 50 for i in range(0, len(products), batch_size): batch products[i:i batch_size] points [] for product in batch: embedding await self._embedder.encode_product(product) product.embedding embedding points.append(qdrant_models.PointStruct( idproduct.chunk_id, vectorembedding, payload{ product_id: product.product_id, source: product.source.value, title: product.title, title_en: product.title_en, description: product.description[:1000], description_en: product.description_en[:1000], price_usd: product.price_usd, rating: product.rating, review_count: product.review_count, category: product.category, image_urls: product.image_urls, }, )) self._client.upsert( collection_nameself.COLLECTION_NAME, pointspoints, ) logger.info(已索引 %d/%d 个产品, min(i batch_size, len(products)), len(products)) async def semantic_search( self, query: str, price_min: float | None None, price_max: float | None None, rating_min: float | None None, source: ProductSource | None None, top_k: int 10, ) - list[SearchResult]: 混合搜索语义匹配 结构化过滤。 await self._embedder.initialize() # 生成查询向量 query_embedding await self._embedder.encode_product( ProductInfo( product_id__query__, sourceProductSource.AMAZON, titlequery, descriptionquery, ) ) # 构建过滤条件 must_conditions [] if price_min is not None or price_max is not None: price_range {} if price_min is not None: price_range[gte] price_min if price_max is not None: price_range[lte] price_max must_conditions.append( FieldCondition(keyprice_usd, rangeRange(**price_range)) ) if rating_min is not None: must_conditions.append( FieldCondition(keyrating, rangeRange(gterating_min)) ) if source is not None: must_conditions.append( FieldCondition(keysource, match{value: source.value}) ) query_filter Filter(mustmust_conditions) if must_conditions else None # 检索 results self._client.search( collection_nameself.COLLECTION_NAME, query_vectorquery_embedding, query_filterquery_filter, limittop_k, with_payloadTrue, ) search_results [] for hit in results: payload hit.payload or {} product ProductInfo( product_idpayload.get(product_id, ), sourceProductSource(payload.get(source, amazon)), titlepayload.get(title, ), descriptionpayload.get(description, ), title_enpayload.get(title_en, ), description_enpayload.get(description_en, ), price_usdpayload.get(price_usd, 0), ratingpayload.get(rating, 0), review_countpayload.get(review_count, 0), categorypayload.get(category, ), image_urlspayload.get(image_urls, []), ) search_results.append(SearchResult( productproduct, scorehit.score, )) return search_results async def find_competitors( self, product: ProductInfo, max_price_diff_pct: float 0.3, top_k: int 5 ) - list[SearchResult]: 竞品分析找到相似且价格接近的产品。 price_min product.price_usd * (1 - max_price_diff_pct) price_max product.price_usd * (1 max_price_diff_pct) search_text f{product.title} {product.title_en} {product.description[:200]} results await self.semantic_search( querysearch_text, price_minprice_min, price_maxprice_max, rating_min3.5, top_ktop_k * 2, # 多取一些排除自身 ) # 排除自身 competitors [ r for r in results if r.product.product_id ! product.product_id ] return competitors[:top_k] async def find_trending( self, category: str, days: int 30, top_k: int 10 ) - list[SearchResult]: 发现某品类下增长最快的产品。 # 设计思路按评论增长率review_count 近期增长排序 # 简化版取评分最高且 review 最多的产品 results self._client.scroll( collection_nameself.COLLECTION_NAME, scroll_filterFilter(must[ FieldCondition(keycategory, match{value: category}), FieldCondition(keyrating, rangeRange(gte4.0)), ]), limittop_k, with_payloadTrue, with_vectorsFalse, )[0] # 按 review 数量排序粗略代表受欢迎度 results.sort( keylambda x: (x.payload or {}).get(review_count, 0), reverseTrue, ) search_results [] for hit in results: payload hit.payload or {} search_results.append(SearchResult( productProductInfo( product_idpayload.get(product_id, ), sourceProductSource(payload.get(source, amazon)), titlepayload.get(title, ), descriptionpayload.get(description, ), price_usdpayload.get(price_usd, 0), ratingpayload.get(rating, 0), review_countpayload.get(review_count, 0), categorypayload.get(category, ), ), score1.0, )) return search_results # ── 使用示例 ──────────────────────────────────────────── async def main(): embedder MultiLingualEmbedder() store ProductVectorStore(embedder) # 初始化集合 await store.initialize_collection(vector_size1024) # 索引导入的产品 sample_products [ ProductInfo( product_idB09XYZ0001, sourceProductSource.AMAZON, title便携式折叠露营椅 户外超轻铝合金, title_enPortable Folding Camping Chair Ultralight Aluminum, description铝合金框架承重150kg收纳仅35cm适合户外徒步, description_enAluminum frame, 150kg capacity, packs to 35cm, ideal for hiking, price_usd39.99, rating4.5, review_count2340, categoryOutdoor Camping, ), ProductInfo( product_idAE20240002, sourceProductSource.ALIEXPRESS, title蓝牙5.3降噪耳机 无线TWS入耳式运动防水, title_enBluetooth 5.3 Noise Cancelling Earbuds TWS Sport Waterproof, description蓝牙5.3芯片ANC主动降噪IPX7防水续航8小时, description_enBT5.3 chip, ANC, IPX7 waterproof, 8hr battery, price_usd24.99, rating4.2, review_count15200, categoryElectronics Audio, ), ] await store.index_products(sample_products) # 中文搜索英文产品 results await store.semantic_search( query户外折叠椅 轻便 徒步, price_min10, price_max60, rating_min4.0, ) for r in results: print(f[{r.score:.3f}] {r.product.title} (${r.product.price_usd}) - {r.product.rating}/5) # 竞品分析 target sample_products[0] competitors await store.find_competitors(target) print(f\n竞品分析 for {target.title}:) for r in competitors: print(f - {r.product.title} (${r.product.price_usd})) if __name__ __main__: asyncio.run(main())代码设计的核心考量ProductInfo是跨平台统一数据模型不管产品来自 Amazon、AliExpress、Shopify都归一化到这个模型。这是做跨平台检索的前提。中文搜索英文产品encode_product把中英文 title/description 都拼进编码文本BGE-M3 在同一向量空间中处理多语言。不需要翻译中间层。结构化过滤不走向量价格区间、评分门槛、平台来源这些精确条件通过 Qdrant 的Filter机制处理。向量检索负责语义匹配Filter 负责精确约束两个维度各司其职。find_competitors用价格区间限制语义搜索范围语义相似还不够竞品必须价格相近。同时排除自身避免自己跟自己比。四、边界分析与架构权衡4.1 翻译 vs 多语言 Embedding一个常见的纠结是要不要先把所有产品描述翻译成英文再做 Embedding翻译方案的优点可以用性能更好的单语言 Embedding 模型如text-embedding-3-large。多语言 Embedding 的优点无翻译延迟保留原始语言的细微语义差异。翻译总是有损耗——锦纶四面弹翻译成Nylon four-way stretch可能丢失了一些面料行业特有的信息。对于选品场景我倾向多语言 Embedding。选品分析需要保留原始语言的细节特别是行业术语翻译引入的误差可能比 Embedding 模型的跨语言误差更大。4.2 图片特征要不要纳入 EmbeddingCLIP 这类多模态模型可以给产品图生成向量。在时尚、家居、设计类产品中图片向量非常有价值——两把椅子文字描述可以相近但视觉风格完全不同。纳入图片特征会带来两个成本向量维度翻倍文本 1024 图片 512图片下载和处理时间每个产品 3-5 张图建议做法对时尚、家居、消费电子品类启用图片 Embedding对日用百货、工具配件等品类只做文本。用 ProductInfo 的image_urls字段可选填来控制。4.3 数据时效性电商产品数据变化很快——价格三天一变促销活动每周更新。向量索引如果是一个月前建的搜出来的价格信息已经不准了。推荐的分层策略Embedding 索引产品描述、标题这类慢变信息每周更新一次。Payload 价格/评分通过 Qdrant 的set_payload按 product_id 增量更新每小时刷一次。库存/促销状态不做向量检索直接走 Redis 缓存 API 实时查询。4.4 选品分析 vs 推荐系统有人可能会问这不就是个推荐系统吗是的但有本质区别。推荐系统是用户 A 买了 X跟他相似的用户 B 可能也喜欢 X——基于协同过滤。选品 RAG 是给我找跟这个产品在功能、价格、目标人群上高度相似的竞品——基于语义理解。前者依赖用户行为数据后者依赖产品内容理解。一个新品类没有用户行为数据推荐系统做不了但 RAG 选品可以做——因为只需要产品描述。五、总结跨境电商选品天生适合 RAG。信息量大、多语言、跨平台、混合搜索——这些都是 RAG 的强项场景。把这条路走通的关键三件事选对 Embedding 模型BGE-M3 是目前多语言场景下的最佳选择。不需要翻译层中文直接搜英文产品。向量维度和质量平衡得不错。结构化 语义混合搜索向量检索做语义匹配Payload Filter 做价格/评分/平台筛选。两者并行不要让精确过滤变成检索后再筛——那样会丢结果。分层更新策略慢变的 Embedding 周更快变的价格日更实时状态的库存走 Redis。不要试图让一个索引承担所有职责。把这个系统跑起来之后一个选品经理的工作方式会发生质变从翻几百个产品页面变成输入选品方向系统返回最相关的 20 个候选 竞品分析。从劳动力密集型变成智力密集型。下一篇预告Redis JSON Search 联合使用实现结构化与非结构化混合搜索的终极方案。