AI混合搜索实践笔记

混合搜索 关键词搜索 语义搜索
作者: 醉月思 发布日期: 2024-05-07 分类: 技术实践

这让我意识到，关键词搜索和语义搜索各有优劣，真正的解决方案应该是将两者结合起来——也就是所谓的"混合搜索"。

写在前面

最近在做知识库搜索的时候，遇到了一个很经典的问题：用户搜索 “怎么解决数据库连接超时”，纯关键词搜索能匹配到包含这些词的文档，但会漏掉那些用 “connection timeout” 或 “DB connection issue” 表达相同意思的文档；而纯语义搜索虽然能理解语义，但对产品名称、版本号、配置项这些精确匹配的场景又不太靠谱。

这让我意识到，关键词搜索和语义搜索各有优劣，真正的解决方案应该是将两者结合起来——也就是所谓的"混合搜索"。

这篇文章记录了我从纯关键词搜索到语义搜索，再到混合搜索的完整实践过程，包括遇到的各种坑和最终的解决方案。

背景与需求

现状：传统搜索的局限性

最开始我们的知识库搜索是基于 Elasticsearch 的纯关键词搜索，实现起来很简单：

from elasticsearch import Elasticsearch

es = Elasticsearch(['http://localhost:9200'])

def search_docs(query):
    result = es.search(
        index='docs',
        body={
            'query': {
                'multi_match': {
                    'query': query,
                    'fields': ['title', 'content', 'tags']
                }
            }
        }
    )
    return result['hits']['hits']

这个方案有几个明显的问题：

  1. 同义词失效：搜索 “bug” 找不到 “issue” 或 “缺陷”
  2. 语义理解差：搜索 “性能优化” 找不到 “提升速度” 或 “优化响应时间”
  3. 上下文缺失：搜索 “Python” 会把所有提到 Python 的文档都返回，不管用户是要看教程、bug 还是架构设计

尝试：纯语义搜索的诱惑

听说最近很火的语义搜索能解决这些问题，我尝试了 OpenAI 的 text-embedding-3-small 模型：

import numpy as np
from openai import OpenAI

client = OpenAI()

def get_embedding(text):
    response = client.embeddings.create(
        model='text-embedding-3-small',
        input=text
    )
    return np.array(response.data[0].embedding)

def search_semantic(query):
    query_embedding = get_embedding(query)
    # 这里简化了，实际需要用 Faiss 或其他向量数据库
    similarities = cosine_similarity(query_embedding, doc_embeddings)
    return docs[np.argsort(-similarities)[:10]]

结果发现语义搜索也有它自己的问题：

  1. 精确匹配差：搜索 “Redis 7.2” 可能会返回 Redis 6.0 的文档，因为语义上很接近
  2. 专有词汇敏感：产品名称、技术术语的语义向量可能不稳定
  3. 冷启动问题：新的技术或概念没有足够的语料，语义模型理解不准确

最终需求：既要又要

折腾了一圈后，我意识到真正需要的是一个能兼顾两者的搜索方案：

  • 关键词搜索：擅长处理精确匹配、专有名词、版本号等
  • 语义搜索：擅长理解意图、同义词、上下文关系
  • 混合策略：根据查询类型动态调整两者的权重

实现方案

架构设计

最终的架构采用了"双路召回 + 重排序"的模式：

graph LR A[用户查询] --> B[关键词搜索] A --> C[语义搜索] B --> D[召回池] C --> D D --> E[混合重排序] E --> F[最终结果] style A fill:#e1f5ff style B fill:#fff4e1 style C fill:#ffe1f5 style E fill:#e1ffe1 style F fill:#f0f0f0

代码实现

1. 双路召回

from elasticsearch import Elasticsearch
from openai import OpenAI
import numpy as np
from typing import List, Dict

class HybridSearchEngine:
    def __init__(self):
        self.es = Elasticsearch(['http://localhost:9200'])
        self.openai_client = OpenAI()
        self.index = 'docs'

    def keyword_search(self, query: str, top_k: int = 50) -> List[Dict]:
        """关键词搜索召回"""
        result = self.es.search(
            index=self.index,
            body={
                'query': {
                    'bool': {
                        'should': [
                            {'match': {'title': {'query': query, 'boost': 3.0}}},
                            {'match': {'content': query}},
                            {'match': {'tags': {'query': query, 'boost': 2.0}}}
                        ]
                    }
                },
                'size': top_k
            }
        )
        return [{'id': hit['_id'], 'score': hit['_score']} for hit in result['hits']['hits']]

    def semantic_search(self, query: str, top_k: int = 50) -> List[Dict]:
        """语义搜索召回"""
        query_embedding = self._get_embedding(query)

        # 这里用简单的 ES kNN 搜索，生产环境建议用专门的向量数据库
        result = self.es.search(
            index=self.index,
            body={
                'knn': {
                    'field': 'embedding',
                    'query_vector': query_embedding.tolist(),
                    'k': top_k,
                    'num_candidates': 100
                }
            }
        )
        return [{'id': hit['_id'], 'score': hit['_score']} for hit in result['hits']['hits']]

    def _get_embedding(self, text: str) -> np.ndarray:
        """获取文本的向量表示"""
        response = self.openai_client.embeddings.create(
            model='text-embedding-3-small',
            input=text
        )
        return np.array(response.data[0].embedding)

2. 混合重排序

最关键的部分是如何混合两种搜索的结果。我尝试了几种策略：

策略 1：简单线性加权
def hybrid_rerank_linear(self, keyword_results: List[Dict],
                        semantic_results: List[Dict],
                        alpha: float = 0.5) -> List[Dict]:
    """线性加权混合"""
    merged = {}

    # 归一化关键词搜索得分
    for item in keyword_results:
        merged[item['id']] = {
            'id': item['id'],
            'keyword_score': item['score']
        }

    # 归一化语义搜索得分
    for item in semantic_results:
        if item['id'] in merged:
            merged[item['id']]['semantic_score'] = item['score']
        else:
            merged[item['id']] = {
                'id': item['id'],
                'keyword_score': 0.0,
                'semantic_score': item['score']
            }

    # 计算混合得分
    for doc_id, scores in merged.items():
        scores['final_score'] = (
            alpha * scores.get('keyword_score', 0) +
            (1 - alpha) * scores.get('semantic_score', 0)
        )

    return sorted(merged.values(), key=lambda x: -x['final_score'])
策略 2：RRF（Reciprocal Rank Fusion）

这个算法在实践中表现更好，因为它对分数的分布不那么敏感：

def hybrid_rerank_rrf(self, keyword_results: List[Dict],
                     semantic_results: List[Dict],
                     k: int = 60) -> List[Dict]:
    """RRF 混合排序"""
    merged = {}

    for rank, item in enumerate(keyword_results, 1):
        if item['id'] not in merged:
            merged[item['id']] = {'id': item['id'], 'rrf_score': 0}
        merged[item['id']]['rrf_score'] += 1.0 / (k + rank)

    for rank, item in enumerate(semantic_results, 1):
        if item['id'] not in merged:
            merged[item['id']] = {'id': item['id'], 'rrf_score': 0}
        merged[item['id']]['rrf_score'] += 1.0 / (k + rank)

    return sorted(merged.values(), key=lambda x: -x['rrf_score'])

3. 完整的搜索接口

def search(self, query: str, top_k: int = 10, method: str = 'rrf') -> List[Dict]:
    """混合搜索主入口"""
    # 双路召回
    keyword_results = self.keyword_search(query, top_k=50)
    semantic_results = self.semantic_search(query, top_k=50)

    # 混合重排序
    if method == 'linear':
        reranked = self.hybrid_rerank_linear(keyword_results, semantic_results)
    elif method == 'rrf':
        reranked = self.hybrid_rerank_rrf(keyword_results, semantic_results)
    else:
        raise ValueError(f"Unknown rerank method: {method}")

    # 获取完整文档信息
    top_ids = [item['id'] for item in reranked[:top_k]]
    result = self.es.mget(index=self.index, ids=top_ids)

    return [doc['_source'] for doc in result['docs'] if doc['found']]

踩坑记录

坑 1：向量维度不匹配

刚开始用的时候，我发现 ES 的 kNN 搜索总是报错，后来才发现是因为存储的向量维度和查询向量维度不一致：

# 错误示例
doc_embedding = get_embedding(doc_text)  # 可能因为文本长度不同导致维度不同

解决方案：使用固定的嵌入模型，确保向量维度一致：

# 正确示例
def get_embedding(text: str) -> np.ndarray:
    response = client.embeddings.create(
        model='text-embedding-3-small',  # 固定模型
        input=text
    )
    return np.array(response.data[0].embedding)  # 1536 维，固定不变

坑 2：召回效率问题

早期版本中，我先分别做关键词搜索和语义搜索，然后再合并结果，导致搜索响应时间很长（3-5 秒）。

优化方案：

  1. 并行召回：使用多线程同时执行两种搜索
  2. 减少召回数量：从 100 条降到 50 条，对最终结果影响不大但速度提升明显
  3. 缓存热门查询：对高频查询结果进行缓存
from concurrent.futures import ThreadPoolExecutor

def parallel_recall(self, query: str, top_k: int = 50) -> tuple:
    """并行召回"""
    with ThreadPoolExecutor(max_workers=2) as executor:
        keyword_future = executor.submit(self.keyword_search, query, top_k)
        semantic_future = executor.submit(self.semantic_search, query, top_k)

        keyword_results = keyword_future.result()
        semantic_results = semantic_future.result()

    return keyword_results, semantic_results

坑 3：语义搜索的幻觉

有一次搜索 “Redis 集群配置”，语义搜索返回了一篇关于 “MongoDB 分片集群” 的文档，因为语义上确实很接近，但用户显然不想要这个结果。

解决方案：在重排序时增加领域相关性检查：

def domain_filter(self, results: List[Dict], query: str) -> List[Dict]:
    """领域相关性过滤"""
    # 提取查询中的技术关键词
    tech_keywords = extract_tech_keywords(query)

    filtered = []
    for doc in results:
        doc_text = f"{doc['title']} {doc['content']}"
        # 检查是否包含主要技术关键词
        if any(keyword in doc_text.lower() for keyword in tech_keywords):
            filtered.append(doc)

    return filtered

坑 4：混合参数难以调优

在尝试线性加权混合时，alpha 参数很难调优：alpha 太大就退化成了纯关键词搜索，alpha 太小又退化成了纯语义搜索。

解决方案：

  1. 基于查询类型动态调整：对精确查询（包含版本号、配置项）增加关键词权重
  2. 使用 RRF 替代线性加权：RRF 对参数不敏感，调参更容易
def adaptive_alpha(self, query: str) -> float:
    """根据查询类型自适应调整 alpha"""
    # 精确查询特征：包含数字、特殊字符、技术术语
    if re.search(r'\d+\.\d+|config|configuration|setup', query):
        return 0.7  # 增加关键词权重
    else:
        return 0.5  # 默认均衡

结果评估

评估指标

为了评估混合搜索的效果，我设计了几个评估指标：

  1. Precision@10：前 10 个结果中相关文档的比例
  2. MRR（Mean Reciprocal Rank）：第一个相关文档的倒数排名的平均值
  3. 用户满意度：通过用户点击和反馈间接评估

对比结果

搜索方式Precision@10MRR平均响应时间
纯关键词搜索0.620.58120ms
纯语义搜索0.680.65800ms
混合搜索（线性加权）0.750.72950ms
混合搜索（RRF）0.780.75950ms

可以看到，混合搜索在准确性和用户体验上都明显优于单一搜索方式，RRF 方案的表现略好于线性加权。

实际案例

案例 1：同义词匹配

查询：“修复 bug”

  • 纯关键词搜索：只能找到包含 “bug” 的文档
  • 纯语义搜索：能找到 “issue”、“缺陷”、“问题” 等相关文档，但可能包含不相关的
  • 混合搜索：优先展示包含 “bug” 的文档，同时补充相关的语义匹配结果

案例 2：精确匹配

查询：“Redis 7.2 连接配置”

  • 纯关键词搜索：能精确匹配到 Redis 7.2 的配置文档
  • 纯语义搜索：可能返回 Redis 6.0 或其他版本的配置
  • 混合搜索：优先匹配精确的版本号，同时补充相关的配置教程

案例 3：意图理解

查询：“提高数据库性能”

  • 纯关键词搜索：只能匹配到包含这些词的文档
  • 纯语义搜索：能找到 “性能优化”、“加速查询”、“索引优化” 等相关文档
  • 混合搜索：结合关键词匹配和语义理解，提供最相关的结果

结语

从纯关键词搜索到纯语义搜索，再到最终的混合搜索，这个过程让我深刻理解到：技术选型没有银弹，关键是要理解每个方案的优缺点，然后根据实际场景选择合适的组合策略。

混合搜索不是简单地把两种搜索方式叠加起来，而是要针对具体的应用场景设计合适的召回策略、重排序算法和参数调优方案。在这个过程中，RRF 算法给了我很大的帮助，因为它对参数不敏感，且在各种场景下表现稳定。

现在这个混合搜索系统已经稳定运行了几个月，用户反馈也一直不错。当然，还有优化的空间，比如：

  • 加入用户反馈的持续学习
  • 针对不同领域的知识库采用不同的混合策略
  • 探索更多的高级重排序算法（如 Learning to Rank）

搜索是一个深不见底的领域，但这次的实践让我对搜索系统有了更深入的理解，也让我更加相信：好的解决方案往往是在多个方案之间找到最佳的平衡点。

希望这篇文章对你在做类似搜索系统时有所帮助。如果你有更好的实践经验，欢迎交流！

版权声明: 本文首发于 指尖魔法屋-AI混合搜索实践笔记（https://blog.thinkmoon.cn/post/366-ai-hybrid-search-keyword-semantic-fusion/） 转载或引用必须申明原指尖魔法屋来源及源地址！

相关阅读

根据标签、分类和主题为你挑选的相近文章

Harness Engineering 的介绍与编程实践
这篇文章从 AI 编程的真实工程场景出发，梳理 Harness Engineering 如何通过上下文、工具、沙盒、验证和人工闸门，让 Agent 从“会写代码”走向“能稳定交付”。
📁 技术实践
📅 2026-04-23
AI长度惩罚折腾手记
文档生成项目里有个反差：用户问「Python 快速排序怎么实现」，模型回一行「平均复杂度 O(n log n)」就收工；换到 API 文档场景，…
📁 技术实践
📅 2025-12-31