实战项目三：语义搜索与问答系统

目录

项目概述

语义搜索与问答系统是现代NLP落地的核心场景之一，通过向量检索替代传统关键词匹配，解决“同义不同词、跨语言模糊查询”的痛点。

本项目聚焦轻量级企业FAQ场景，目标实现：

  • ✅ 支持自然语言中文提问
  • ✅ 检索准确率>85%
  • ✅ 单条响应<300ms
  • ✅ 无需复杂硬件也能快速部署

核心架构与技术栈

精简架构

┌─────────────┐
│  Streamlit  │ ← 用户交互
└──────┬──────┘
       │
┌──────▼──────┐
│  FastAPI    │ ← API网关、鉴权
└──────┬──────┘
       │
┌──────▼──────────────────────┐
│  检索增强生成(RAG)引擎       │
│  ┌─────────┐  ┌──────────┐  │
│  │检索模块  │→ │ 生成模块  │  │
│  └─────────┘  └──────────┘  │
└──────┬──────────────────────┘
       │
┌──────▼──────────────┐
│  向量化+FAISS索引   │ ← 存储层
└─────────────────────┘

选型标准与推荐

本项目优先考虑免费开源、轻量易部署的组件 | 模块 | 推荐方案 | 说明 | |---------------|-----------------------------------|----------------------------------------------------------------------| | 嵌入模型 | paraphrase-multilingual-MiniLM-L12-v2 | 多语言轻量版，768维，中文效果中等但够用，加载快，CPU/GPU通用 | | 向量索引 | FAISS IndexFlatIP（归一化后） | 精确检索，适合<10万条FAQ的数据量，内存占用可控 | | 后端框架 | FastAPI + Uvicorn | 异步高性能，自动生成API文档 | | 前端框架 | Streamlit | 10分钟搭好交互界面，无需前端经验 | | 生成备选方案 | 基于检索结果的规则匹配 | 不依赖外部API也能运行，隐私友好 |

数据与向量化索引

1. 数据准备

简单FAQ数据结构

[
  {
    "id": "faq_001",
    "question": "密码忘记了怎么办？",
    "answer": "请访问登录页面→点击'忘记密码'→输入注册邮箱→查收重置链接；或联系客服人工处理。",
    "category": "账户",
    "search_text": "密码忘记了怎么办？忘记密码怎么找回？密码丢失如何重置？"
  }
]

重点加了 search_text 字段：把原问题+常用相似问合并，提高召回率。

2. 向量化与索引构建

核心代码

from sentence_transformers import SentenceTransformer
import faiss
import numpy as np
import pickle

# ----------------------
# 1. 初始化嵌入模型
# ----------------------
embedder = SentenceTransformer("paraphrase-multilingual-MiniLM-L12-v2")

# ----------------------
# 2. 批量编码FAQ
# ----------------------
# 假设processed_faqs是预处理好的列表
search_texts = [faq["search_text"] for faq in processed_faqs]
embeddings = embedder.encode(search_texts, convert_to_numpy=True, show_progress_bar=True)

# ----------------------
# 3. 归一化+构建FAISS索引
# ----------------------
# 归一化用于计算余弦相似度（用内积替代，更快）
norms = np.linalg.norm(embeddings, axis=1, keepdims=True)
embeddings = embeddings / (norms + 1e-8)

# 构建精确内积索引
dimension = embeddings.shape[1]
index = faiss.IndexFlatIP(dimension)
index.add(embeddings.astype("float32"))

# ----------------------
# 4. 保存索引+数据
# ----------------------
faiss.write_index(index, "faq_index.faiss")
with open("faq_data.pkl", "wb") as f:
    pickle.dump(processed_faqs, f)

语义检索与RAG实现

1. 语义检索模块

核心代码

class FAQRetriever:
    def __init__(self, index_path="faq_index.faiss", data_path="faq_data.pkl"):
        self.index = faiss.read_index(index_path)
        with open(data_path, "rb") as f:
            self.faqs = pickle.load(f)
    
    def search(self, query: str, top_k: int = 3, threshold: float = 0.3):
        # 编码查询并归一化
        query_emb = embedder.encode([query], convert_to_numpy=True)
        query_emb = query_emb / (np.linalg.norm(query_emb, axis=1, keepdims=True) + 1e-8)
        
        # 检索
        scores, indices = self.index.search(query_emb.astype("float32"), top_k)
        
        # 过滤低相似度结果
        results = []
        for score, idx in zip(scores[0], indices[0]):
            if idx >= 0 and score >= threshold:
                results.append({**self.faqs[idx], "score": float(score)})
        return results

2. 轻量RAG引擎

同时支持无API规则模式和可选LLM增强模式

class LightRAG:
    def __init__(self, retriever: FAQRetriever, llm_client=None):
        self.retriever = retriever
        self.llm = llm_client  # 如OpenAI()或本地LLM
    
    def query(self, user_question: str, use_llm: bool = False):
        # 1. 检索
        docs = self.retriever.search(user_question)
        if not docs:
            return "抱歉，未找到相关信息，请尝试重新表述问题或联系客服。"
        
        # 2. 构建上下文
        context = "\n\n".join([
            f"【问题】{d['question']}\n【答案】{d['answer']}"
            for d in docs
        ])
        
        # 3. 生成答案
        if use_llm and self.llm:
            prompt = f"""
            你是一个智能客服助手。请根据以下知识库信息准确回答用户问题，不要编造内容，控制在200字以内。
            知识库：
            {context}
            用户问题：{user_question}
            """
            try:
                resp = self.llm.chat.completions.create(
                    model="gpt-4o-mini",
                    messages=[{"role": "user", "content": prompt}],
                    max_tokens=300,
                    temperature=0.3
                )
                return resp.choices[0].message.content
            except Exception as e:
                print(f"LLM调用失败：{e}")
        
        # 无LLM备选方案：返回最相关的第一条答案
        return f"根据知识库信息：{docs[0]['answer']}"

快速部署方案

1. 后端FastAPI核心

省略日志、缓存、健康检查等细节，只保留核心功能

from fastapi import FastAPI
from pydantic import BaseModel
from typing import Optional, List

app = FastAPI(title="FAQ智能问答API")

# 初始化全局对象
retriever = FAQRetriever()
rag = LightRAG(retriever)

class QueryRequest(BaseModel):
    question: str
    use_llm: bool = False
    top_k: int = 3

class QueryResponse(BaseModel):
    answer: str
    docs: Optional[List[dict]] = None

@app.post("/api/query", response_model=QueryResponse)
async def query_endpoint(req: QueryRequest):
    docs = retriever.search(req.question, req.top_k)
    answer = rag.query(req.question, req.use_llm)
    return QueryResponse(answer=answer, docs=docs if req.top_k > 0 else None)

2. 前端Streamlit核心

import streamlit as st
import requests

# ----------------------
# 页面配置
# ----------------------
st.set_page_config(page_title="FAQ智能问答", page_icon="🤖")
st.title("🤖 FAQ 智能问答助手")

# ----------------------
# 侧边栏设置
# ----------------------
with st.sidebar:
    api_url = st.text_input("后端API", "http://localhost:8000/api/query")
    use_llm = st.checkbox("使用LLM增强（可选）", value=False)
    show_docs = st.checkbox("显示参考来源", value=True)

# ----------------------
# 主交互
# ----------------------
user_q = st.text_input("请输入您的问题：")
if st.button("🔍 提问", type="primary") and user_q:
    with st.spinner("思考中..."):
        try:
            resp = requests.post(
                api_url,
                json={"question": user_q, "use_llm": use_llm, "top_k": 3 if show_docs else 0}
            )
            resp_data = resp.json()
            
            st.success(resp_data["answer"])
            
            if show_docs and resp_data.get("docs"):
                with st.expander("📋 查看参考来源"):
                    for i, d in enumerate(resp_data["docs"], 1):
                        st.markdown(f"**来源{i}（相似度：{d['score']:.2f}）**")
                        st.write(f"问题：{d['question']}")
                        st.write(f"答案：{d['answer']}")
                        st.divider()
        except Exception as e:
            st.error(f"请求失败：{e}")

3. 一键启动命令

# 终端1：启动后端（先安装依赖 pip install fastapi uvicorn sentence-transformers faiss-cpu numpy pickle5）
uvicorn backend:app --host 0.0.0.0 --port 8000

# 终端2：启动前端（先安装依赖 pip install streamlit requests）
streamlit run frontend.py

优化要点总结

  1. 向量化优化

    • 把相似问、分类、标签合并到 search_text 字段
    • 优先用批量编码
    • 有GPU的话加 device="cuda" 参数

  2. 检索优化

    • 低数据量用 IndexFlatIP，高数据量换 IndexIVFFlat
    • 加相似度阈值过滤无效结果
    • 可以用混合检索（语义+关键词简单匹配）提高召回

  3. 性能优化

    • 用本地LRU缓存高频查询的嵌入/结果
    • 加Redis作为分布式缓存
    • 生产环境用Uvicorn多worker模式

本项目完整覆盖了从数据到部署的全流程，适合作为企业FAQ、产品咨询、课程问答等场景的入门基础，后续可根据需求扩展功能~