一、真实场景：一个搜索重构项目差点把我搞崩
2024年Q2，我接手了一个电商搜索系统的重构。老系统用Elasticsearch 7.17做全文检索，用户搜"红色连衣裙"能返回结果，但搜"适合约会穿的裙子"就完全跑偏——返回的是包含"约会"关键词的无关商品。老板拍桌子说："用户搜'约会穿什么'，你给推荐个西装领带？"
更致命的是，新来的产品经理要求支持"图片搜同款"和"模糊语义匹配"。传统ES的倒排索引+BM25算法，对同义词、上下文理解、多模态搜索完全无力。我调研了3周，最终决定用RAG+向量搜索重构。本文记录了这个过程的完整技术细节，包括踩过的坑。
二、问题拆解：传统搜索的三大死穴
先看老系统的架构：
# 老系统架构
search:
engine: elasticsearch
version: 7.17.9
index:
- products (倒排索引)
query:
- match_phrase (精确匹配)
- multi_match (多字段匹配)
- function_score (加权排序)
data_size: 500万商品
qps: 2000
p99_latency: 150ms压测数据暴露了三个问题：
- 语义鸿沟：用户query和商品描述的字面匹配率只有23%。比如"通勤包"和"上班用的手提包"完全不匹配。
- 冷启动困难：新品没有搜索历史，BM25评分极低，永远排在后10页。
- 多模态缺失：图片搜索需要单独训练模型，和文本搜索完全隔离。
我们用2000条真实用户query做了测试：
| 指标 | 传统ES | AI搜索 |
|---|---|---|
| Top5召回率 | 34.2% | 78.6% |
| 语义匹配准确率 | 12.1% | 89.3% |
| 冷启动商品曝光率 | 0.8% | 45.2% |
三、方案对比：三种搜索架构的硬碰硬
我测试了三种方案：
方案A：纯ES + 同义词扩展
在ES里加同义词词典，比如"约会"→"浪漫"、"聚餐"。但维护成本极高，2000个同义词组需要人工标注，且覆盖不全。
// ES同义词配置
{
"settings": {
"analysis": {
"filter": {
"synonym_filter": {
"type": "synonym",
"synonyms": [
"约会, 浪漫, 聚餐, 相亲",
"通勤, 上班, 办公, 商务"
]
}
}
}
}
}效果：Top5召回率提升到41.3%，但语义匹配准确率只有28.7%。而且同义词冲突严重——"苹果"既可以是水果也可以是手机。
方案B：向量搜索 + 双塔模型
用Sentence-BERT（all-MiniLM-L6-v2）把query和商品描述编码成384维向量，用Milvus 2.4做近似最近邻搜索。
# 双塔模型编码
from sentence_transformers import SentenceTransformer
import numpy as np
model = SentenceTransformer('all-MiniLM-L6-v2')
def encode_text(texts):
embeddings = model.encode(texts, normalize_embeddings=True)
return embeddings.astype(np.float32)
# 商品描述编码
products = ["红色连衣裙 雪纺 夏季 约会", "黑色西装 商务 通勤"]
product_embeddings = encode_text(products)
# 用户query编码
query = "适合约会穿的裙子"
query_embedding = encode_text([query])[0]效果：Top5召回率72.1%，语义匹配准确率81.4%。但问题来了——向量搜索对长尾query（比如"那个去年买过的红色带花的裙子"）效果差，因为模型训练数据里没有这种表达。
方案C：RAG + 混合搜索（最终方案）
结合ES的精确匹配和向量搜索的语义匹配，再加一个RAG层做query改写和结果重排。
# 混合搜索架构
def hybrid_search(query, top_k=10):
# 1. Query改写（RAG）
rewritten_query = rag_rewrite(query)
# 2. 向量搜索
query_emb = encode_text([rewritten_query])[0]
vector_results = milvus_search(query_emb, top_k=top_k*2)
# 3. ES精确匹配
es_results = es_search(rewritten_query, top_k=top_k*2)
# 4. 结果融合（RRF算法）
fused_results = reciprocal_rank_fusion(vector_results, es_results)
# 5. 重排（Cross-Encoder）
reranked = cross_encoder_rerank(query, fused_results)
return reranked[:top_k]效果：Top5召回率78.6%，语义匹配准确率89.3%。但延迟从150ms飙升到420ms。
四、完整代码实现：从0搭建AI搜索系统
以下是我最终落地的完整代码，基于Python 3.11 + FastAPI + Milvus 2.4 + Elasticsearch 8.12。
4.1 环境准备
# 安装依赖
pip install fastapi uvicorn pymilvus elasticsearch sentence-transformers
pip install torch transformers langchain chromadb
# 启动Milvus（Docker）
docker run -d --name milvus \
-p 19530:19530 \
-p 9091:9091 \
milvusdb/milvus:2.4.0-latest
# 启动ES
docker run -d --name es \
-p 9200:9200 \
-e "discovery.type=single-node" \
-e "xpack.security.enabled=false" \
docker.elastic.co/elasticsearch/elasticsearch:8.12.04.2 数据索引
# index_data.py
from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType
from elasticsearch import Elasticsearch
from sentence_transformers import SentenceTransformer
import json
# 连接Milvus
connections.connect(host='localhost', port='19530')
# 创建向量集合
fields = [
FieldSchema(name="id", dtype=DataType.INT64, is_primary=True),
FieldSchema(name="product_id", dtype=DataType.INT64),
FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=384),
FieldSchema(name="category", dtype=DataType.VARCHAR, max_length=100)
]
schema = CollectionSchema(fields, "product embeddings")
collection = Collection("products", schema)
# 创建IVF_FLAT索引
index_params = {
"metric_type": "IP",
"index_type": "IVF_FLAT",
"params": {"nlist": 1024}
}
collection.create_index("embedding", index_params)
# 加载数据
model = SentenceTransformer('all-MiniLM-L6-v2')
with open('products.json', 'r') as f:
products = json.load(f)
embeddings = model.encode([p['description'] for p in products], normalize_embeddings=True)
data = [
[i for i in range(len(products))],
[p['id'] for p in products],
embeddings.tolist(),
[p['category'] for p in products]
]
collection.insert(data)
collection.load()
# 同步到ES
es = Elasticsearch(['http://localhost:9200'])
for p in products:
es.index(index='products', id=p['id'], body={
'title': p['title'],
'description': p['description'],
'category': p['category'],
'price': p['price']
})4.3 搜索API
# search_api.py
from fastapi import FastAPI, Query
from pymilvus import connections, Collection
from elasticsearch import Elasticsearch
from sentence_transformers import SentenceTransformer
from transformers import pipeline
import numpy as np
app = FastAPI()
# 初始化
connections.connect(host='localhost', port='19530')
milvus_collection = Collection("products")
milvus_collection.load()
es = Elasticsearch(['http://localhost:9200'])
model = SentenceTransformer('all-MiniLM-L6-v2')
reranker = pipeline('text-classification', model='cross-encoder/ms-marco-MiniLM-L-6-v2')
def rag_rewrite(query):
"""用LLM改写query，增强语义"""
# 这里用简单的规则代替LLM，生产环境用GPT-4或本地模型
rewrite_rules = {
'约会': '浪漫 约会 聚餐 相亲',
'通勤': '上班 办公 商务 通勤',
'运动': '健身 跑步 运动 户外'
}
for key, value in rewrite_rules.items():
if key in query:
query = query.replace(key, value)
return query
def reciprocal_rank_fusion(vector_results, es_results, k=60):
"""RRF融合算法"""
fused_scores = {}
for rank, (id, score) in enumerate(vector_results):
fused_scores[id] = 1.0 / (k + rank + 1)
for rank, (id, score) in enumerate(es_results):
if id in fused_scores:
fused_scores[id] += 1.0 / (k + rank + 1)
else:
fused_scores[id] = 1.0 / (k + rank + 1)
return sorted(fused_scores.items(), key=lambda x: x[1], reverse=True)
@app.get("/search")
async def search(query: str = Query(..., min_length=1), top_k: int = 10):
# 1. Query改写
rewritten_query = rag_rewrite(query)
# 2. 向量搜索
query_emb = model.encode([rewritten_query], normalize_embeddings=True)[0]
search_params = {"metric_type": "IP", "params": {"nprobe": 10}}
vector_results = milvus_collection.search(
data=[query_emb.tolist()],
anns_field="embedding",
param=search_params,
limit=top_k * 2,
output_fields=["product_id"]
)
vector_ids = [(hit.id, hit.score) for hit in vector_results[0]]
# 3. ES搜索
es_body = {
"query": {
"multi_match": {
"query": rewritten_query,
"fields": ["title^3", "description", "category"],
"type": "best_fields"
}
},
"size": top_k * 2
}
es_response = es.search(index="products", body=es_body)
es_ids = [(hit['_id'], hit['_score']) for hit in es_response['hits']['hits']]
# 4. RRF融合
fused = reciprocal_rank_fusion(vector_ids, es_ids)
# 5. Cross-Encoder重排
product_ids = [id for id, _ in fused[:top_k*3]]
products = [es.get(index='products', id=id)['_source'] for id in product_ids]
pairs = [[query, p['description']] for p in products]
scores = reranker(pairs)
reranked = sorted(zip(products, scores), key=lambda x: x[1]['score'], reverse=True)
return {
"query": query,
"rewritten_query": rewritten_query,
"results": [p['title'] for p, _ in reranked[:top_k]]
}4.4 压测脚本
# 用wrk压测
wrk -t12 -c400 -d30s --latency http://localhost:8000/search?query=约会裙子
# 结果
Running 30s test @ http://localhost:8000/search
12 threads and 400 connections
Thread Stats Avg Stdev Max +/- Stdev
Latency 420.23ms 123.45ms 1.23s 68.72%
Req/Sec 78.45 23.12 150.00 72.34%
Latency Distribution
50% 398.12ms
75% 512.34ms
90% 678.90ms
99% 1.02s
2345 requests in 30.00s, 1.23MB read
Requests/sec: 78.17
Transfer/sec: 41.23KB五、效果数据：AI搜索到底强在哪
我们用10000条真实用户query做了A/B测试，对比三个系统：
| 指标 | 传统ES | 纯向量搜索 | 混合搜索（最终） |
|---|---|---|---|
| Top5召回率 | 34.2% | 72.1% | 78.6% |
| Top10召回率 | 51.3% | 85.4% | 91.2% |
| 语义匹配准确率 | 12.1% | 81.4% | 89.3% |
| 冷启动商品曝光率 | 0.8% | 38.7% | 45.2% |
| P99延迟 | 150ms | 280ms | 420ms |
| QPS | 2000 | 850 | 78 |
关键发现：
- 混合搜索的召回率比纯ES高44.4个百分点，但QPS从2000暴跌到78。原因是Cross-Encoder重排太慢。
- 纯向量搜索的语义匹配准确率已经很高（81.4%），但冷启动商品曝光率只有38.7%，因为向量搜索对长尾query不敏感。
- 混合搜索的冷启动曝光率最高（45.2%），因为RAG改写能生成更多相关query。
六、避坑指南：我踩过的6个坑
以下是我在实际部署中遇到的坑，每个都花了至少2天解决：
坑1：向量维度选择
一开始用768维的all-mpnet-base-v2，结果Milvus索引构建花了8小时，搜索延迟1.2秒。换成384维的all-MiniLM-L6-v2后，索引时间降到40分钟，延迟280ms。但召回率只降了3%。
结论：非必要不用高维向量，384维足够。
坑2：IVF_FLAT的nprobe参数
nprobe=1时召回率只有45%，nprobe=100时召回率92%但延迟1.5秒。最终调成nprobe=10，召回率88%，延迟280ms。
# 参数调优
search_params = {"metric_type": "IP", "params": {"nprobe": 10}} # 黄金值坑3：ES和Milvus的数据一致性
商品更新时，ES和Milvus的数据不同步。用户搜"红色连衣裙"，ES返回最新商品，但Milvus还是旧向量。解决方案：用消息队列（RabbitMQ）做异步同步，延迟控制在5秒内。
# 异步同步
def sync_product(product_id, action):
if action == 'update':
# 更新ES
es.update(index='products', id=product_id, body={'doc': new_data})
# 更新Milvus
new_emb = model.encode([new_data['description']])
milvus_collection.delete(f"product_id == {product_id}")
milvus_collection.insert([(product_id, new_emb)])坑4：Cross-Encoder的batch size
默认batch_size=1，重排100个结果要100次推理，耗时3秒。改成batch_size=32后，耗时降到0.4秒。
# 批量推理
scores = reranker(pairs, batch_size=32) # 关键优化坑5：RAG改写导致query膨胀
RAG改写把"约会"变成"浪漫 约会 聚餐 相亲"，结果ES返回了大量不相关的结果（比如"相亲"相关的婚庆用品）。解决方案：限制改写后的query长度不超过5个词，且保留原query的权重。
# 限制改写
def rag_rewrite(query):
rewritten = query
for key, value in rewrite_rules.items():
if key in query:
rewritten = query + ' ' + value.split()[0] # 只加一个词
return rewritten[:50] # 限制长度坑6：生产环境的内存泄漏
SentenceTransformer模型加载后，每次请求都重新编码，导致内存泄漏。解决方案：用单例模式，模型只加载一次。
# 单例模式
class ModelSingleton:
_instance = None
def __new__(cls):
if cls._instance is None:
cls._instance = super().__new__(cls)
cls._instance.model = SentenceTransformer('all-MiniLM-L6-v2')
return cls._instance
model = ModelSingleton().model七、总结：什么时候该用AI搜索
基于我的实战经验，给出决策建议：
- 数据量<100万，query简单：纯ES就够了，别折腾。
- 数据量100万-1000万，需要语义匹配：纯向量搜索（Milvus + Sentence-BERT），不要加RAG和重排，延迟可控在300ms内。
- 数据量>1000万，query复杂，需要冷启动：混合搜索（ES + Milvus + RAG + Cross-Encoder），但要做好缓存和异步处理，否则QPS上不去。
- 需要多模态搜索：必须上向量搜索，传统ES完全做不到。
最后说一句：AI搜索不是银弹。如果你的用户query都是精确匹配（比如搜"iPhone 15"），传统ES的BM25算法反而更快更准。别为了追技术而追技术。