L2-1:第5关|AI怎么＂懂＂你的问题?语义向量化核心解密

上一关我们搞定了向量索引，但问题来了：用户提问的方式千奇百怪，和知识库里的文档表达完全不同，AI怎么找到它们之间的关联？

答案就藏在语义检索里。

一、开篇：为什么你搜不到想要的东西？

先来个灵魂拷问——

你的知识库文档写的是：“本产品自签收之日起7日内支持无理由退货”

用户搜的是：“不想要了能退吗”

从字面上看，这两个表达没有一个字是一样的。传统关键词检索大概率翻车。

但语义检索会说：小场面，看我的。

它能把”7日内无理由退货”和”不想要了能退吗”变成两个向量，在向量空间里一算——哎，相关度0.92，命中了。

这就是语义检索的魔力：让AI真正”理解”你的意思，而不是机械地匹配文字。

二、核心概念：语义检索的三大关键技术

2.1 语义检索 vs 关键词检索

关键词检索的典型场景：

• • 搜索”苹果” → 返回所有包含”苹果”的文档
• • 但用户可能想问水果，也可能想问手机

语义检索的优势：

• • 搜索”水果” → 能找到”苹果””香蕉””橙子”
• • 搜索”买完不想要了” → 能找到退货相关政策

2.2 查询理解：AI在搜索前做了什么？

用户输入一句话，语义检索系统会先”消化”一下：

用户输入："我想问一下就是那个退货的流程是咋回事儿"         ↓查询预处理         ↓结构化查询："退货流程"

查询预处理的常见步骤：

2.3 Query改写：让查询更精准

Query改写是语义检索的核心环节，主要解决三个问题：

问题1：表达多样性

用户说："电脑开不了机"文档写："笔记本无法启动"改写后：["电脑开不了机", "笔记本无法启动", "笔记本无法开机"]

问题2：口语化表达

用户说："咋退啊"文档写："退货流程"改写后：["咋退啊", "退货流程", "如何退货"]

问题3：拼写错误

用户输入："手几充不进电"改写后：["手机充不进电"]

Query改写的三大策略：

# 1. 同义词扩展query = "电脑很卡"expanded = ["电脑很卡", "计算机卡顿", "PC运行慢", "电脑响应慢"]# 2. HyDE（假设性文档）query = "怎么退换货"# 先生成一个"理想答案"的向量hypothetical_doc = "退货流程：请在7天内联系客服，提供订单号..."# 用假设文档去检索# 3. Query分解query = "苹果公司创始人是谁"decomposed = ["苹果公司", "创始人", "Tim Cook", "乔布斯"]

三、避坑指南：Query处理的五大陷阱

坑1：Query太短，信息不足

反面案例：

用户搜："电脑"（只有两个字）系统：返回10万条关于电脑的文档用户：???

解决方案：Query扩展

short_query = "电脑"# 使用关联词扩展expanded_query = expand_query(short_query, top_k=5)# 扩展后：["电脑", "笔记本电脑", "台式机", "计算机", "PC"]

坑2：歧义未消除

反面案例：

用户搜："苹果"系统：返回水果和手机混杂的结果用户：我问的是手机！

解决方案：意图识别 + 上下文

# 结合上下文判断意图user_history = ["我最近想换手机", "苹果15怎么样"]current_query = "苹果"# 识别为：苹果手机disambiguated = disambiguate(current_query, context=user_history)# 输出：苹果手机

坑3：口语噪声太多

反面案例：

用户搜："就是那个啥我想退一下就是买的东西"系统：搜"就是那个啥我想退一下就是买的东西" → 无结果

解决方案：关键信息提取

messy_query = "就是那个啥我想退一下就是买的东西"cleaned = extract_key_info(messy_query)# 输出："退货"

坑4：专业术语与口语混用

反面案例：

用户搜："显示器颜色发黄"文档写："屏幕色偏暖色调"

解决方案：术语对齐

user_query = "显示器颜色发黄"aligned = align_term(user_query, domain="电子产品")# 输出："显示器颜色发黄 OR 屏幕色偏暖色调"

坑5：否定词处理不当

反面案例：

用户搜："不要苹果手机"系统：返回包含"苹果"的结果，忽略了"不要"用户：!!!

解决方案：否定词识别

query = "不要苹果手机"parsed = parse_negation(query)# 输出：{"include": ["手机"], "exclude": ["苹果"]}# 检索时：找包含"手机"但不包含"苹果"的文档

四、代码实战：智能Query预处理系统

下面来实现一个完整的Query预处理Pipeline：

"""RAG第5关：语义Query预处理系统实现：查询清洗、意图识别、Query改写、扩展检索依赖：pip install sentence-transformers torch rank_bm25"""from typing import List, Dict, Tupleimport reimport json# ==================== 第一部分：查询清洗模块 ====================class QueryCleaner:    """    查询清洗：去除口语噪声，提取核心信息    典型问题：    - "就是那个啥我想退货" → "退货"    - "请问一下就是..." → "..."    - 拼写错误、标点混乱    """    # 口语填充词库    FILLER_WORDS = [        '就是', '那个', '啥', '这个', '嗯', '啊', '吧', '嘛',        '请问', '我想问一下', '打扰一下', '你好', '请问一下',        '就是想问', '就是问一下', '就是想知道', '问一下',        '有个问题', '我想问一下就是'    ]    # 标点符号    PUNCTUATION = r'[,，.。!！?？;；:：""''（）()【】[\]\\]'    def clean(self, query: str) -> str:        """        清洗查询语句        Args:            query: 原始查询，如 "就是那个啥我想退货"        Returns:            清洗后查询，如 "退货"        """        text = query        # 1. 转小写        text = text.lower()        # 2. 去除标点        text = re.sub(self.PUNCTUATION, ' ', text)        # 3. 去除口语填充词        for filler in self.FILLER_WORDS:            text = text.replace(filler, ' ')        # 4. 去除多余空格        text = re.sub(r'\s+', ' ', text).strip()        # 5. 长度检查：如果太短，保留原查询        if len(text) < 2:            return query        return text    def remove_negations(self, query: str) -> Tuple[str, List[str]]:        """        分离否定词        Args:            query: "不要苹果手机"        Returns:            (肯定部分, 否定词列表)            ("手机", ["苹果"])        """        negation_words = ['不', '不是', '不要', '别', '非', '无']        positive_parts = []        negative_words = []        # 简单实现：检测否定词后面的名词        for neg in negation_words:            if neg in query:                idx = query.index(neg)                # 假设否定词后面5个字是目标词                end_idx = min(idx + len(neg) + 5, len(query))                negative_words.append(query[idx:end_idx])        # 提取肯定部分（去除否定词）        positive = query        for neg in negation_words:            positive = positive.replace(neg, '')        return positive.strip(), negative_words    def extract_keywords(self, query: str) -> List[str]:        """        提取关键词（简化版）        实际生产中可使用：LAC、jieba、HanLP等分词工具        """        # 去除停用词        stop_words = ['的', '了', '是', '在', '和', '与', '或', '有', '我', '你', '他', '她']        words = self.clean(query).split()        keywords = [w for w in words if w not in stop_words and len(w) > 1]        return keywords# ==================== 第二部分：Query改写模块 ====================class QueryRewriter:    """    Query改写：同义词扩展、HyDE、Query分解    三大策略：    1. 同义词扩展：用多种表达方式检索    2. HyDE：用"理想答案"引导检索    3. Query分解：复杂问题拆成简单问题    """    def __init__(self, embedder=None):        self.embedder = embedder        # 常见同义词映射（简化版）        self.synonym_dict = {            '电脑': ['计算机', 'PC', '笔记本', '笔记本电脑'],            '手机': ['移动电话', '智能手机', '移动端'],            '退货': ['退换货', '退款', '退货退款'],            '激活': ['启用', '开通', '启动使用'],            '卡': ['慢', '卡顿', '运行慢', '响应慢'],        }    def expand_with_synonyms(self, query: str, top_k: int = 5) -> List[str]:        """        同义词扩展        "电脑很卡" → ["电脑很卡", "计算机卡顿", "PC运行慢", ...]        """        expanded = [query]  # 保留原始查询        # 检查是否有可扩展的词        for word, synonyms in self.synonym_dict.items():            if word in query:                for syn in synonyms[:2]:  # 每个词取2个同义词                    new_query = query.replace(word, syn)                    if new_query not in expanded:                        expanded.append(new_query)        # 如果没有匹配，使用原始query        if len(expanded) == 1:            expanded.append(query + "的解决方法")        return expanded[:top_k]    def generate_hypothetical_answer(self, query: str) -> str:        """        HyDE：生成假设性答案        思路：        1. 让LLM根据query生成一个"理想答案"        2. 用这个理想答案去检索        3. 理想答案包含了可能的文档表达方式        """        # 实际使用时需要调用LLM API        # 这里用模板模拟        templates = {            '怎么': '需要按照以下步骤操作：第一步...第二步...',            '为什么': '原因可能有以下几点：1...2...3...',            '如何': '解决这个问题的方法是...',            '是什么': '这个概念指的是...具体包括...'        }        for key, template in templates.items():            if key in query:                return f"根据{query}，{template}"        return f"关于{query}的详细说明如下："    def decompose_query(self, query: str) -> List[str]:        """        Query分解：复杂问题拆成子问题        "苹果公司创始人是谁" → ["苹果公司", "创始人", "Tim Cook"]        """        # 简化实现：基于规则的分解        # 实际可用命名实体识别(NER)        decomposed = []        # 检测实体类型        entities = self._extract_entities(query)        decomposed.extend(entities)        # 添加关键词        keywords = self._extract_keywords(query)        decomposed.extend(keywords)        return list(set(decomposed))[:5]  # 去重，最多5个    def _extract_entities(self, text: str) -> List[str]:        """提取实体（简化版）"""        # 实际使用NER模型，如: bert-base-chinese-ner        entities = []        # 公司名检测（简化规则）        companies = ['苹果', '谷歌', '微软', '亚马逊', '阿里巴巴', '腾讯', '华为']        for company in companies:            if company in text:                entities.append(company)        # 人名检测（简化规则）        names = ['Tim Cook', '库克', '乔布斯', '马斯克', '马云']        for name in names:            if name in text:                entities.append(name)        return entities    def _extract_keywords(self, text: str) -> List[str]:        """提取关键词"""        keywords = []        keyword_patterns = ['是什么', '怎么做', '怎么', '为什么', '如何', '流程', '步骤']        for pattern in keyword_patterns:            if pattern in text:                keywords.append(pattern)                text = text.replace(pattern, '')        # 剩余部分作为关键词        remaining = text.strip()        if remaining:            keywords.append(remaining)        return keywords# ==================== 第三部分：语义检索模块 ====================class SemanticSearchPreprocessor:    """    语义检索预处理Pipeline    整合所有预处理步骤，形成完整的查询处理流程    """    def __init__(self, embedder=None):        self.cleaner = QueryCleaner()        self.rewriter = QueryRewriter(embedder)        self.embedder = embedder    def preprocess(self, query: str) -> Dict:        """        完整预处理流程        输入："就是那个我想退一下货"        输出：{            "original": "就是那个我想退一下货",            "cleaned": "退货",            "expanded_queries": ["退货", "退换货", "退款"],            "hypothetical_answer": "...",            "decomposed": ["退货"],            "final_query": "退货"        }        """        result = {            "original": query,            "cleaned": None,            "expanded_queries": [],            "hypothetical_answer": None,            "decomposed": [],            "final_query": None,        }        # 1. 清洗        result["cleaned"] = self.cleaner.clean(query)        # 2. 同义词扩展        result["expanded_queries"] = self.rewriter.expand_with_synonyms(            result["cleaned"]        )        # 3. HyDE生成        result["hypothetical_answer"] = self.rewriter.generate_hypothetical_answer(            result["cleaned"]        )        # 4. Query分解        result["decomposed"] = self.rewriter.decompose_query(result["cleaned"])        # 5. 最终查询：综合所有策略        result["final_query"] = result["cleaned"]        return result    def build_multi_strategy_query(self, query: str) -> List[str]:        """        构建多策略查询列表        用于多路召回：同时用多个查询去检索，结果合并        """        preprocessed = self.preprocess(query)        queries = []        # 策略1：原始查询        queries.append(query)        # 策略2：清洗后的查询        if preprocessed["cleaned"] != query:            queries.append(preprocessed["cleaned"])        # 策略3：同义词扩展        queries.extend(preprocessed["expanded_queries"][:3])        # 去重        return list(set(queries))[:5]    def debug_query(self, query: str) -> str:        """        调试模式：打印完整预处理过程        """        result = self.preprocess(query)        debug_info = f"""========== Query预处理调试 ==========原始查询: {result['original']}----------------------------------------清洗后: {result['cleaned']}----------------------------------------同义词扩展: {result['expanded_queries']}----------------------------------------假设答案: {result['hypothetical_answer']}----------------------------------------Query分解: {result['decomposed']}----------------------------------------最终查询: {result['final_query']}========================================"""        return debug_info# ==================== 第四部分：完整示例 ====================def demo():    """演示完整流程"""    print("=" * 60)    print("RAG第5关：语义Query预处理系统演示")    print("=" * 60)    # 初始化    preprocessor = SemanticSearchPreprocessor()    # 测试用例    test_queries = [        "就是那个我想退一下货",        "电脑很卡怎么办",        "苹果公司是干什么的",        "不要苹果手机",        "请问一下激活流程是什么"    ]    for query in test_queries:        print(preprocessor.debug_query(query))        # 多策略查询        multi_queries = preprocessor.build_multi_strategy_query(query)        print(f"多路召回查询: {multi_queries}")        print("-" * 60)def advanced_demo():    """    进阶演示：结合向量检索的完整流程    """    try:        from sentence_transformers import SentenceTransformer        import numpy as np        print("\n" + "=" * 60)        print("进阶演示：语义Query + 向量检索")        print("=" * 60)        # 加载Embedding模型        model = SentenceTransformer('BAAI/bge-base-zh-v1.5')        # 初始化预处理器        preprocessor = SemanticSearchPreprocessor(embedder=model)        # 示例查询        query = "不想要了怎么退"        # 预处理        result = preprocessor.preprocess(query)        # 生成多路查询        queries = preprocessor.build_multi_strategy_query(query)        # 向量化        query_embeddings = model.encode(queries)        print(f"原始查询: {query}")        print(f"多路召回查询: {queries}")        print(f"生成查询向量维度: {query_embeddings.shape}")        print("\n各查询的向量：")        for i, (q, emb) in enumerate(zip(queries, query_embeddings)):            print(f"  [{i}] {q}: {emb[:5]}...")    except ImportError:        print("提示：运行进阶演示需要安装：pip install sentence-transformers torch")if __name__ == "__main__":    demo()    advanced_demo()

代码结构总结

Query预处理Pipeline├── QueryCleaner：清洗模块│   ├── clean()：去除口语噪声│   ├── remove_negations()：分离否定词│   └── extract_keywords()：提取关键词│├── QueryRewriter：改写模块│   ├── expand_with_synonyms()：同义词扩展│   ├── generate_hypothetical_answer()：HyDE生成│   └── decompose_query()：Query分解│└── SemanticSearchPreprocessor：整合Pipeline    ├── preprocess()：完整预处理    ├── build_multi_strategy_query()：多路召回查询    └── debug_query()：调试模式

五、最佳实践

实践1：Query预处理是”必需品”

不管你用多好的Embedding模型，Query预处理都能显著提升效果。数据证明，好的预处理能提升10%-30%的召回率。

实践2：多策略并行

不要只依赖单一策略。同义词扩展 + HyDE + Query分解 组合使用，效果远优于任何单一方法。

实践3：分场景配置

实践4：持续优化

• • 定期分析用户搜索日志
• • 找出高频无结果/低相关性的查询
• • 针对性优化预处理规则

实践5：结合重排序

Query预处理提升召回，后续用重排序（Rerank） 提升精度。这两个环节配合使用，是业界最佳实践。

六、思考题

1. 1. 场景分析：用户搜”苹果”，知识库里有水果和手机两类文档。如何设计预处理策略，让系统能根据上下文自动判断用户意图？
2. 2. 代码实践：尝试为QueryRewriter添加一个”拼写纠错”功能，使用编辑距离算法（Levenshtein Distance）找出并修正query中的错别字。
3. 3. 进阶思考：如果用户的问题是”不需要了”，但系统文档写的是”退款”，预处理如何处理这种”否定+需求”的组合表达？

下节预告

第6关我们将进入向量库选型环节：Milvus、Chroma、Weaviate…这么多向量库，到底选哪个？