场景切入：双十一电商大促的 AI 客服危机
凌晨 2 点，你负责的电商 AI 客服系统正在承受前所未有的压力。距离双十一开场还有 3 小时，并发咨询量已经从日常的 500 QPS 暴涨到 8000 QPS。更棘手的是，用户的问题五花八门——"我想买那个 2023 年款的 MacBook Pro，128G 内存的，有优惠吗？"
"iPhone 15 Pro 256G 和 512G 差多少钱？发货到北京要多久？"
"退换货政策是什么？我上周买的耳机包装丢了能退吗？"
这些query混合了精确的商品型号、时间戳、容量规格、地理位置信息。纯向量搜索很难精准匹配"128G"这样的精确关键词；纯 BM25 关键词搜索又无法理解"那个 2023 年款的 MacBook Pro"的语义意图。 这正是 Hybrid Search（混合检索）大显身手的时刻。本文将从电商促销场景出发，详解如何构建一套兼顾语义理解与精确匹配的混合检索系统，并提供完整的 HolySheep AI API 接入方案。为什么需要混合检索？
向量搜索的局限性
向量搜索（Embedding-based Search）通过将文本映射到高维向量空间，计算语义相似度。它的优势在于理解同义词、上下文和模糊意图："我想买一台轻薄本" → 能匹配到 "商务办公笔记本电脑推荐"
"续航给力的办公电脑" → 能匹配到 "MacBook Air 18 小时电池"- 精确数值匹配："128G"、"256GB"、"2023年款"
- 品牌/型号完整匹配："iPhone 15 Pro Max" 需要完整出现
- 长尾关键词："vivo X90 Pro+" 这种专业术语可能被误解
- 搜索量极低的专有名词：冷门商品描述不足导致向量不准
关键词搜索的局限性
传统 BM25 / TF-IDF 关键词搜索擅长精确匹配，但在以下场景力不从心：- 口语化表达："那个拍照好看的手机" 无法匹配"拍照手机推荐"
- 同义词替换："笔记本" vs "笔记本电脑" vs "laptop"
- 上下文理解：用户说"和上次买的一样"需要关联历史记录
- 拼写变体："Iphone" vs "iPhone"
混合检索的核心思路
Hybrid Search 的核心公式可以简化为：final_score = α × semantic_score + (1 - α) × keyword_score- semantic_score：向量搜索的余弦相似度（0~1）
- keyword_score：BM25 关键词匹配得分（归一化后 0~1）
- α（alpha）：混合权重，通常设为 0.3~0.7 之间
实战：构建电商场景的混合检索系统
系统架构设计
┌─────────────────────────────────────────────────────────────────┐
│ 用户 Query 输入 │
│ "2023款 MacBook Pro 14寸 512G 深空灰有优惠吗" │
└─────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ Query 预处理模块 │
│ → 分词 → 纠错 → 实体识别（商品/规格/品牌） │
└─────────────────────────────────────────────────────────────────┘
│
┌───────────────┴───────────────┐
▼ ▼
┌───────────────────┐ ┌───────────────────┐
│ 向量检索引擎 │ │ BM25 关键词索引 │
│ (Milvus/Qdrant) │ │ (Elasticsearch) │
└───────────────────┘ └───────────────────┘
│ │
▼ ▼
┌───────────────────┐ ┌───────────────────┐
│ Top-50 语义候选 │ │ Top-50 关键词候选 │
└───────────────────┘ └───────────────────┘
│ │
└───────────────┬───────────────┘
▼
┌───────────────────────────────┐
│ RRF 融合重排序模块 │
│ RRF_score = Σ 1/(k+rank) │
└───────────────────────────────┘
│
▼
┌───────────────────────────────┐
│ Top-10 结果 → LLM 生成 │
│ HolyShehe AI API │
└───────────────────────────────┘Step 1：商品数据索引构建
首先，我们需要对商品库构建双重索引。以下是使用 Python 构建混合索引的完整代码：import os
import json
from openai import OpenAI
接入 HolyShehe AI API（国内直连，延迟<50ms）
client = OpenAI(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
)
class ProductIndexer:
def __init__(self):
self.client = client
self.products = []
def generate_embedding(self, text: str) -> list[float]:
"""调用 HolyShehe 嵌入接口生成向量"""
response = self.client.embeddings.create(
model="text-embedding-3-large",
input=text
)
return response.data[0].embedding
def index_products(self, products: list[dict]):
"""
批量索引商品数据
products 格式: [{"id": "SKU001", "name": "MacBook Pro 14寸",
"brand": "Apple", "specs": {...}}, ...]
"""
indexed_products = []
for product in products:
# 构建混合文本用于向量生成（包含名称、规格、品牌）
semantic_text = f"""
商品名称: {product['name']}
品牌: {product.get('brand', '')}
分类: {product.get('category', '')}
规格: {json.dumps(product.get('specs', {}), ensure_ascii=False)}
描述: {product.get('description', '')}
卖点: {product.get('highlights', [])}
""".strip()
# 生成语义向量
embedding = self.generate_embedding(semantic_text)
# 同时提取关键词用于 BM25 索引
keywords = self._extract_keywords(product)
indexed_products.append({
"id": product["id"],
"vector": embedding,
"keywords": keywords,
"metadata": {
"name": product["name"],
"price": product.get("price"),
"category": product.get("category"),
"stock": product.get("stock", 0)
}
})
# 打印进度
print(f"✓ 已索引: {product['name']}")
return indexed_products
def _extract_keywords(self, product: dict) -> list[str]:
"""提取关键词用于 BM25 索引"""
keywords = []
# 品牌词
if product.get("brand"):
keywords.append(product["brand"])
# 商品名称分词
name_tokens = product["name"].replace("-", " ").replace("/", " ").split()
keywords.extend(name_tokens)
# 规格参数
specs = product.get("specs", {})
for key, value in specs.items():
keywords.append(str(value))
if isinstance(value, str):
keywords.extend(value.split())
# 去重并返回
return list(set(keywords))
使用示例
indexer = ProductIndexer()
sample_products = [
{
"id": "MACBOOK-PRO-14-2023",
"name": "MacBook Pro 14英寸 M3 Pro 芯片",
"brand": "Apple",
"category": "笔记本电脑",
"specs": {
"chip": "M3 Pro",
"screen": "14.2英寸",
"memory": "18GB",
"storage": "512GB",
"color": "深空灰"
},
"price": 16999,
"stock": 50
}
]
indexed = indexer.index_products(sample_products)
print(f"\n✅ 成功索引 {len(indexed)} 个商品")Step 2：混合检索实现
接下来实现核心的混合检索逻辑，支持 RRF（Reciprocal Rank Fusion）融合算法：import numpy as np
from typing import List, Tuple
class HybridSearcher:
def __init__(self, indexed_products: List[dict], alpha: float = 0.5):
"""
初始化混合检索器
- indexed_products: 已索引的商品列表
- alpha: 混合权重 (0=纯关键词, 1=纯向量)
"""
self.products = indexed_products
self.alpha = alpha
# 初始化本地向量存储（生产环境建议使用 Milvus/Qdrant）
self.vectors = np.array([p["vector"] for p in indexed_products])
self._build_bm25_index()
def _build_bm25_index(self):
"""构建 BM25 关键词索引"""
from rank_bm25 import BM25Okapi
# 提取所有文档的关键词列表
self.corpus_tokens = [p["keywords"] for p in self.products]
# 初始化 BM25
self.bm25 = BM25Okapi(self.corpus_tokens)
print(f"✓ BM25 索引已构建，共 {len(self.corpus_tokens)} 个文档")
def search(
self,
query: str,
top_k: int = 10,
vector_top_k: int = 50,
keyword_top_k: int = 50
) -> List[Tuple[dict, float]]:
"""
执行混合检索
Args:
query: 用户搜索词
top_k: 返回最终结果数量
vector_top_k: 向量搜索候选数量
keyword_top_k: 关键词搜索候选数量
Returns:
[(product, score), ...] 按融合得分降序排列
"""
# Step 1: 向量搜索
vector_results = self._vector_search(query, top_k=vector_top_k)
# Step 2: 关键词搜索
keyword_results = self._keyword_search(query, top_k=keyword_top_k)
# Step 3: RRF 融合重排序
fused_results = self._rrf_fusion(vector_results, keyword_results)
return fused_results[:top_k]
def _vector_search(self, query: str, top_k: int) -> List[Tuple[int, float]]:
"""向量语义搜索"""
# 生成查询向量
query_embedding = self.client.embeddings.create(
model="text-embedding-3-large",
input=query
).data[0].embedding
# 计算余弦相似度
query_vec = np.array(query_embedding)
similarities = np.dot(self.vectors, query_vec) / (
np.linalg.norm(self.vectors, axis=1) * np.linalg.norm(query_vec)
)
# 获取 Top-K
top_indices = np.argsort(similarities)[::-1][:top_k]
return [(idx, float(similarities[idx])) for idx in top_indices]
def _keyword_search(self, query: str, top_k: int) -> List[Tuple[int, float]]:
"""BM25 关键词搜索"""
# 分词
query_tokens = query.lower().split()
# 计算 BM25 得分
scores = self.bm25.get_scores(query_tokens)
# 归一化到 [0, 1]
if scores.max() > 0:
scores = scores / scores.max()
# 获取 Top-K
top_indices = np.argsort(scores)[::-1][:top_k]
return [(idx, float(scores[idx])) for idx in top_indices if scores[idx] > 0]
def _rrf_fusion(
self,
vector_results: List[Tuple[int, float]],
keyword_results: List[Tuple[int, float]],
k: int = 60
) -> List[Tuple[dict, float]]:
"""
RRF (Reciprocal Rank Fusion) 融合算法
RRF_score(d) = Σ 1/(k + rank(d))
这种方法不需要调参，对两个检索结果进行加权融合
"""
rrf_scores = {}
# 向量搜索得分 (权重 alpha)
for rank, (doc_id, vec_score) in enumerate(vector_results):
rrf = 1.0 / (k + rank)
rrf_scores[doc_id] = rrf_scores.get(doc_id, 0) + self.alpha * rrf * vec_score
# 关键词搜索得分 (权重 1-alpha)
for rank, (doc_id, kw_score) in enumerate(keyword_results):
rrf = 1.0 / (k + rank)
rrf_scores[doc_id] = rrf_scores.get(doc_id, 0) + (1 - self.alpha) * rrf * kw_score
# 按融合得分排序
sorted_results = sorted(rrf_scores.items(), key=lambda x: x[1], reverse=True)
return [(self.products[doc_id], score) for doc_id, score in sorted_results]
使用示例
searcher = HybridSearcher(indexed_products=indexed, alpha=0.5)
执行混合搜索
results = searcher.search(
query="2023款 MacBook Pro 14寸 512G 深空灰有优惠吗",
top_k=5
)
print("\n🎯 搜索结果:")
for product, score in results:
print(f" [{score:.4f}] {product['metadata']['name']} - ¥{product['metadata']['price']}")Step 3：结合 LLM 生成智能回复
拿到检索结果后，我们需要让 LLM 基于检索内容生成最终回复：def generate_response(query: str, search_results: List[dict], user_context: dict = None):
"""
基于检索结果生成智能回复
使用 HolyShehe AI API（汇率¥1=$1，节省>85%成本）
"""
# 构建上下文
context_parts = []
for i, product in enumerate(search_results[:3], 1):
meta = product["metadata"]
context_parts.append(f"""
商品 {i}:
- 名称: {meta['name']}
- 价格: ¥{meta['price']}
- 库存: {'有货' if meta['stock'] > 0 else '缺货'}
- 分类: {meta.get('category', '未知')}
""")
context = "\n".join(context_parts)
system_prompt = """你是一位专业的电商客服助手。用户询问商品信息时，你应该：
1. 基于提供的商品信息准确回答
2. 如实告知价格和库存状态
3. 如有优惠活动，主动提及
4. 如果没有完全匹配的商品，推荐类似商品
5. 保持专业、友好的语气"""
user_prompt = f"""用户问题: {query}
参考商品信息:
{context}
{"用户历史: " + str(user_context) if user_context else ""}
请根据以上信息生成回复。"""
# 调用 HolyShehe AI API（GPT-4o 模型，性价比高）
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
],
temperature=0.7,
max_tokens=500
)
return response.choices[0].message.content
生成最终回复
reply = generate_response(
query="2023款 MacBook Pro 14寸 512G 深空灰有优惠吗",
search_results=[p for p, _ in results]
)
print("\n🤖 AI 客服回复:")
print(reply)性能优化：应对高并发场景
异步批量处理
在双十一等大促场景下，需要优化检索吞吐量和响应延迟：import asyncio
from concurrent.futures import ThreadPoolExecutor
import time
class AsyncHybridSearcher(HybridSearcher):
"""异步版本的混合检索器，支持批量请求"""
def __init__(self, indexed_products: List[dict], alpha: float = 0