对话式搜索：当搜索引擎学会了“追问”与“澄清”

引言

传统搜索引擎依赖用户一次性输入精确的关键词，而对话式搜索（Conversational Search） 通过多轮交互主动“追问”与“澄清”，将搜索从“单次检索”升级为“渐进式信息获取”。2026年的技术核心在于意图模糊识别与动态检索策略，让搜索引擎具备“追问”能力，解决用户意图不明确、信息过载等痛点。

技术背景

  • 痛点：用户查询往往简短、模糊（如“苹果”指水果还是公司？），传统搜索只能返回宽泛结果，用户需手动筛选。
  • 趋势：基于LLM + RAG的混合架构，通过Query Rewrite（查询重写）和Clarification Question（澄清提问）动态调整搜索策略。
  • 关键：系统需判断“何时该直接搜索”与“何时该反问用户”，平衡搜索效率与准确性。

应用使用场景

  1. 电商导购：用户输入“想买手机”，系统追问“预算范围？”、“用途？（游戏/拍照）”，再返回精准商品列表。
  2. 知识库检索：员工提问“报销流程”，系统识别需澄清“是国内差旅还是海外项目？”，再定位具体文档。
  3. 垂直搜索（法律/医疗）：用户描述症状“头痛发热”，系统追问“持续时间？”、“有无咳嗽？”，再提供初步诊断建议。

核心架构与原理

原理流程图

完整

不完整

用户输入Query

意图识别 & 槽位填充

关键槽位是否完整?

查询重写 + 向量检索

生成澄清问题

用户补充信息

生成最终答案

输出并更新对话状态

核心特性

  • 主动澄清：当查询置信度低或槽位缺失时，自动生成选择题或填空式提问。
  • 上下文感知：利用对话历史（History）进行指代消解（如“它”指代上轮提到的商品）。
  • 混合检索：结合关键词（BM25）与向量检索（Dense Retrieval），提升召回率。

环境准备（Python 3.11+）

# 核心框架
pip install fastapi uvicorn pydantic
# 检索与向量库
pip install faiss-cpu sentence-transformers
# LLM集成（以Ollama本地模型为例）
pip install ollama langchain
# 对话状态管理
pip install redis

场景一：电商导购对话搜索（FastAPI + 槽位填充）

场景描述

用户输入模糊查询（如“买笔记本”），系统通过多轮追问补全预算、品牌、用途等槽位，最终返回精准商品。

代码实现

# conversational_search_ecommerce.py
import json
import re
from enum import Enum
from typing import Dict, List, Optional, Tuple
from pydantic import BaseModel
from fastapi import FastAPI, HTTPException

app = FastAPI()

# 商品数据库模拟（实际应为MySQL/ES）
PRODUCTS = [
    {"id": 1, "name": "MacBook Pro 16", "category": "laptop", "brand": "Apple", "price": 18999, "tags": ["办公", "设计"]},
    {"id": 2, "name": "ThinkPad X1", "category": "laptop", "brand": "Lenovo", "price": 12999, "tags": ["商务", "便携"]},
    {"id": 3, "name": "ROG枪神7", "category": "laptop", "brand": "ASUS", "price": 14999, "tags": ["游戏", "高性能"]},
]

class Intent(str, Enum):
    BUY_LAPTOP = "buy_laptop"
    UNKNOWN = "unknown"

class SlotStatus(str, Enum):
    UNKNOWN = "unknown"
    FILLED = "filled"

class ConversationState(BaseModel):
    intent: Intent = Intent.UNKNOWN
    slots: Dict[str, str] = {}  # 槽位：budget, brand, usage
    missing_slots: List[str] = ["budget", "brand", "usage"]
    history: List[Dict] = []

class EcommerceSearchEngine:
    def __init__(self):
        self.state_manager = {}

    def get_user_state(self, user_id: str) -> ConversationState:
        return self.state_manager.get(user_id, ConversationState())

    def update_state(self, user_id: str, state: ConversationState):
        self.state_manager[user_id] = state

    def nlu_parse(self, text: str, current_state: ConversationState) -> Tuple[Intent, Dict]:
        """自然语言理解：提取意图与槽位"""
        intent = Intent.UNKNOWN
        slots = {}

        # 关键词+规则匹配（实际可用NER模型）
        if re.search(r"(笔记本|电脑|MacBook|ThinkPad)", text):
            intent = Intent.BUY_LAPTOP
            # 提取预算
            budget_match = re.search(r"(\d+)元", text)
            if budget_match:
                slots["budget"] = budget_match.group(1)
            # 提取品牌
            if "苹果" in text or "Mac" in text:
                slots["brand"] = "Apple"
            elif "联想" in text or "ThinkPad" in text:
                slots["brand"] = "Lenovo"
            # 提取用途
            if "游戏" in text:
                slots["usage"] = "游戏"
            elif "办公" in text or "商务" in text:
                slots["usage"] = "办公"

        return intent, slots

    def generate_clarification(self, missing_slots: List[str]) -> str:
        """生成澄清问题"""
        slot_questions = {
            "budget": "请问您的预算范围是？（例如：5000-10000元）",
            "brand": "您偏好哪个品牌？（苹果/联想/华硕/其他）",
            "usage": "主要用途是？（游戏/办公/设计/日常使用）"
        }
        # 优先询问第一个缺失的槽位
        slot_to_ask = missing_slots[0] if missing_slots else None
        return slot_questions.get(slot_to_ask, "请提供更多细节。")

    def search_products(self, slots: Dict) -> List[Dict]:
        """根据槽位检索商品"""
        filtered = PRODUCTS
        # 预算过滤
        if "budget" in slots:
            budget = int(slots["budget"])
            filtered = [p for p in filtered if p["price"] <= budget]
        # 品牌过滤
        if "brand" in slots:
            filtered = [p for p in filtered if p["brand"] == slots["brand"]]
        # 用途过滤（标签匹配）
        if "usage" in slots:
            filtered = [p for p in filtered if slots["usage"] in p["tags"]]
        return filtered

    def process_query(self, user_id: str, query: str) -> Dict:
        # 1. 获取当前对话状态
        state = self.get_user_state(user_id)
        
        # 2. NLU解析
        intent, new_slots = self.nlu_parse(query, state)
        
        # 3. 更新槽位状态
        if intent != Intent.UNKNOWN:
            state.intent = intent
        for slot, value in new_slots.items():
            if slot in state.missing_slots:
                state.missing_slots.remove(slot)
            state.slots[slot] = value
        
        # 4. 决策：继续澄清 or 返回结果
        if state.missing_slots:
            # 生成澄清问题
            reply = self.generate_clarification(state.missing_slots)
            action = "clarify"
        else:
            # 执行搜索
            products = self.search_products(state.slots)
            if products:
                product_names = [p["name"] for p in products]
                reply = f"为您找到以下商品：{', '.join(product_names)}"
            else:
                reply = "未找到匹配商品，请调整条件。"
            action = "answer"
            # 重置状态（可选）
            state = ConversationState()
        
        # 5. 保存状态
        state.history.append({"query": query, "reply": reply})
        self.update_state(user_id, state)
        
        return {"reply": reply, "action": action, "missing_slots": state.missing_slots}

# API端点
search_engine = EcommerceSearchEngine()

class ChatRequest(BaseModel):
    user_id: str
    query: str

class ChatResponse(BaseModel):
    reply: str
    action: str

@app.post("/search", response_model=ChatResponse)
async def conversational_search(request: ChatRequest):
    try:
        result = search_engine.process_query(request.user_id, request.query)
        return ChatResponse(**result)
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

运行结果与测试

# 测试命令序列
curl -X POST "http://localhost:8000/search" -H "Content-Type: application/json" -d '{"user_id": "u123", "query": "想买笔记本"}'
# 输出: {"reply": "请问您的预算范围是？（例如：5000-10000元）", "action": "clarify"}

curl -X POST "http://localhost:8000/search" -H "Content-Type: application/json" -d '{"user_id": "u123", "query": "8000元左右"}'
# 输出: {"reply": "主要用途是？（游戏/办公/设计/日常使用）", "action": "clarify"}

curl -X POST "http://localhost:8000/search" -H "Content-Type: application/json" -d '{"user_id": "u123", "query": "办公"}'
# 输出: {"reply": "为您找到以下商品：ThinkPad X1", "action": "answer"}

场景二：知识库多轮RAG追问（向量检索 + 置信度检测）

场景描述

用户提问“报销流程”，系统先检索知识库，若发现存在“国内/国际”不同流程且用户未指定，主动追问澄清，再返回精准答案。

代码实现

# conversational_rag.py
import numpy as np
from sentence_transformers import SentenceTransformer
import faiss
from typing import List, Tuple

class KnowledgeBase:
    def __init__(self):
        self.docs = [
            "国内差旅报销流程：1. 填写申请单 2. 主管审批 3. 财务审核",
            "国际差旅报销流程：1. 填写申请单 2. 外事处备案 3. 财务审核",
            "办公用品采购流程：1. 提交采购申请 2. 行政部确认"
        ]
        self.embedder = SentenceTransformer('BAAI/bge-small-zh-v1.5')
        self.index = faiss.IndexFlatIP(384)  # 假设维度384
        self._build_index()

    def _build_index(self):
        vectors = self.embedder.encode(self.docs, normalize_embeddings=True)
        self.index.add(vectors.astype('float32'))

    def search(self, query: str, top_k: int = 3) -> List[Tuple[str, float]]:
        query_vec = self.embedder.encode([query], normalize_embeddings=True)
        scores, indices = self.index.search(query_vec.astype('float32'), top_k)
        return [(self.docs[i], scores[0][j]) for j, i in enumerate(indices[0])]

class ClarificationRAG:
    def __init__(self):
        self.kb = KnowledgeBase()
        self.clarify_threshold = 0.7  # 置信度阈值

    def needs_clarification(self, query: str, results: List[Tuple[str, float]]) -> Tuple[bool, str]:
        """判断是否需要澄清：若top结果相似度低或存在歧义，则追问"""
        if not results:
            return True, "未找到相关文档，请重新描述您的问题。"
        
        top_score = results[0][1]
        if top_score < self.clarify_threshold:
            return True, "您的问题比较宽泛，请问您是指国内差旅还是国际差旅？"
        
        # 检查歧义：如果前两个结果属于不同类别且分数接近
        if len(results) >= 2:
            doc1, score1 = results[0]
            doc2, score2 = results[1]
            if abs(score1 - score2) < 0.1 and ("国内" in doc1) != ("国内" in doc2):
                return True, "请问您是指国内差旅还是国际差旅？"
        
        return False, ""

    def answer(self, query: str, context: str = "") -> Dict:
        # 1. 检索
        search_query = f"{context} {query}".strip()
        results = self.kb.search(search_query)
        
        # 2. 澄清判断
        need_clarify, clarify_msg = self.needs_clarification(query, results)
        if need_clarify:
            return {"reply": clarify_msg, "action": "clarify", "context": query}
        
        # 3. 生成答案
        top_doc = results[0][0]
        answer = f"根据知识库：{top_doc}"
        return {"reply": answer, "action": "answer", "context": ""}

# 测试
if __name__ == "__main__":
    rag = ClarificationRAG()
    
    # 第一轮：模糊查询触发澄清
    result1 = rag.answer("报销流程")
    print("Round 1:", result1)
    # 输出: {"reply": "您的问题比较宽泛，请问您是指国内差旅还是国际差旅？", "action": "clarify", "context": "报销流程"}
    
    # 第二轮：补充信息后精准回答
    result2 = rag.answer("国内", context="报销流程")
    print("Round 2:", result2)
    # 输出: {"reply": "根据知识库：国内差旅报销流程：1. 填写申请单 2. 主管审批 3. 财务审核", "action": "answer", "context": ""}

场景三：生产级部署（Redis状态管理 + 异步检索）

场景描述

高并发场景下，使用Redis持久化对话状态，并利用异步IO提升检索效率。

代码实现

# production_conversational_search.py
import asyncio
import json
import redis.asyncio as redis
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()
redis_client = redis.Redis(host='localhost', port=6379, decode_responses=True)

class RedisStateManager:
    async def get_state(self, user_id: str) -> Optional[ConversationState]:
        state_json = await redis_client.get(f"conv_state:{user_id}")
        if state_json:
            return ConversationState(**json.loads(state_json))
        return ConversationState()

    async def save_state(self, user_id: str, state: ConversationState):
        await redis_client.setex(
            f"conv_state:{user_id}", 
            3600,  # 1小时过期
            json.dumps(state.dict())
        )

@app.post("/v2/search")
async def search_v2(request: ChatRequest):
    state_manager = RedisStateManager()
    engine = EcommerceSearchEngine()  # 复用之前的引擎
    
    # 异步获取状态
    state = await state_manager.get_state(request.user_id)
    
    # 处理查询（假设process_query已适配异步）
    result = await asyncio.to_thread(engine.process_query, request.user_id, request.query, state)
    
    # 异步保存状态
    await state_manager.save_state(request.user_id, state)
    
    return result

疑难解答

  1. 澄清死循环：设置最大澄清轮次（如3次），超时后强制返回当前最佳结果或转人工。
  2. 状态混乱：用户突然切换话题时，通过意图变化检测自动重置对话状态（如检测到“查天气”后清空之前的“买电脑”槽位）。
  3. 性能瓶颈：向量检索耗时较长，可对澄清阶段的查询使用关键词检索（BM25） 快速过滤，仅在最终回答时使用向量检索。

未来展望与技术趋势

  • LLM原生控制：不再依赖规则判断，由LLM直接决策“是否追问”及“问什么”，实现更自然的对话流。
  • 多模态澄清：支持图片+文本混合输入（如用户上传商品图片，系统追问具体型号）。
  • 个性化追问：基于用户历史行为预测缺失信息，减少提问次数（如老用户默认使用上次的预算范围）。

总结

对话式搜索的“追问”能力本质是信息熵的降低过程。通过槽位填充（结构化信息）与检索置信度（语义匹配）的双重判断，系统在“猜测”与“确认”间找到平衡。2026年的最佳实践是**“规则兜底 + LLM驱动”**：用规则保证可控性，用LLM提升对话的自然度，最终实现搜索从“工具”到“助手”的进化。

# 搜索引擎 # microsoft
Logo
2048 AI社区

有“AI”的1024 = 2048，欢迎大家加入2048 AI社区

更多推荐

cover

从收录到引用：企业官网GEO监测表怎么设计

avatar 2048 AI社区
cover

AI个人时尚风格管家OPC项目

avatar 2048 AI社区

AI漫剧应用的核心判断依据与资料准备指南

本文梳理AI漫剧应用的三类核心判断依据与准备资料，明确适用主体与边界，为有品牌AI曝光需求的主体提供可落地的决策参考

avatar 2048 AI社区