AI 搜索引擎开发教程

SEO 信息

名称：AI搜索引擎开发教程

描述：零基础AI搜索引擎开发教程，涵盖RAG搜索架构、网页爬取、向量检索、混合检索、重排序、查询改写、搜索结果生成等核心技能，适合AI开发者系统学习。

关键词：AI搜索引擎, RAG搜索, 向量检索, 混合检索, Perplexity

长尾关键词：AI搜索引擎开发教程, RAG搜索引擎实战, Perplexity风格搜索引擎开发, 混合检索系统开发教程

一、AI 搜索引擎与传统搜索的区别

传统搜索引擎（如 Google、百度）基于关键词匹配和链接分析（PageRank）返回结果列表，用户需要自行浏览多个网页找到答案。AI 搜索引擎则更进一步：

维度	传统搜索	AI 搜索
核心技术	关键词匹配、PageRank	向量语义检索 + LLM 生成
返回形式	网页链接列表	直接回答 + 引用来源
理解能力	关键词级别	语义级别
多轮对话	不支持	支持上下文理解
信息整合	用户自行完成	AI 自动整合多个来源
典型产品	Google、Bing	Perplexity AI、秘塔搜索

AI 搜索引擎的核心架构是 RAG（Retrieval-Augmented Generation）：先检索相关文档，再由 LLM 基于检索结果生成回答。本教程将手把手构建一个 Perplexity AI 风格的搜索引擎。

二、RAG 搜索架构设计

2.1 整体架构

一个完整的 AI 搜索引擎包含以下模块：

用户查询 → 查询理解 → 查询改写 → 检索（BM25 + 向量）→ 重排序 → LLM 生成 → 引用标注 → 返回结果
                                          ↑
                                    网页爬取 → 内容提取 → 文档分块 → 向量化 → 索引存储

2.2 技术选型

模块	推荐方案
网页爬取	requests + BeautifulSoup / Playwright
文档分块	LangChain TextSplitter / 自定义分块
向量模型	BGE-M3 / text-embedding-3-small
向量数据库	Chroma / Milvus / Qdrant
BM25 检索	rank_bm25 / Elasticsearch
重排序	BGE-Reranker / Cohere Rerank
LLM 生成	Qwen / DeepSeek / GPT-4o

2.3 核心数据流

from dataclasses import dataclass

@dataclass
class SearchResult:
    """单条搜索结果"""
    title: str
    url: str
    snippet: str
    score: float
    content: str = ""

@dataclass
class SearchResponse:
    """搜索响应"""
    answer: str           # AI 生成的回答
    sources: list[SearchResult]  # 引用来源
    query: str            # 原始查询
    rewritten_query: str  # 改写后的查询

三、网页爬取与内容提取

3.1 基础爬虫实现

import requests
from bs4 import BeautifulSoup
from urllib.parse import urljoin, urlparse
import time
import hashlib

class WebCrawler:
    """网页爬取器"""
    
    def __init__(self, timeout: int = 10, delay: float = 1.0):
        self.session = requests.Session()
        self.session.headers.update({
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
        })
        self.timeout = timeout
        self.delay = delay
        self.visited = set()
    
    def fetch_page(self, url: str) -> dict | None:
        """爬取单个页面，提取标题和正文"""
        if url in self.visited:
            return None
        
        try:
            time.sleep(self.delay)
            resp = self.session.get(url, timeout=self.timeout)
            resp.raise_for_status()
            resp.encoding = resp.apparent_encoding
            
            soup = BeautifulSoup(resp.text, "html.parser")
            
            # 移除无关标签
            for tag in soup.find_all(["script", "style", "nav", "footer", "header", "aside"]):
                tag.decompose()
            
            title = soup.title.string.strip() if soup.title and soup.title.string else url
            content = self._extract_content(soup)
            
            self.visited.add(url)
            
            return {
                "url": url,
                "title": title,
                "content": content,
                "content_hash": hashlib.md5(content.encode()).hexdigest()
            }
        except Exception as e:
            print(f"爬取失败 {url}: {e}")
            return None
    
    def _extract_content(self, soup: BeautifulSoup) -> str:
        """提取页面正文内容"""
        # 优先查找 article 或 main 标签
        main = soup.find("article") or soup.find("main") or soup.find("body")
        if not main:
            return ""
        
        paragraphs = []
        for p in main.find_all(["p", "h1", "h2", "h3", "h4", "li"]):
            text = p.get_text(strip=True)
            if len(text) > 20:  # 过滤太短的段落
                paragraphs.append(text)
        
        return "\n".join(paragraphs)
    
    def crawl_search_results(self, query: str, num_results: int = 10) -> list[dict]:
        """爬取搜索结果页面（以 DuckDuckGo 为例）"""
        search_url = f"https://html.duckduckgo.com/html/?q={query}"
        try:
            resp = self.session.get(search_url, timeout=self.timeout)
            soup = BeautifulSoup(resp.text, "html.parser")
            
            results = []
            for link in soup.select(".result__a")[:num_results]:
                href = link.get("href", "")
                if href.startswith("//duckduckgo.com/l/?uddg="):
                    from urllib.parse import unquote, parse_qs, urlparse
                    parsed = urlparse(href)
                    actual_url = parse_qs(parsed.query).get("uddg", [href])[0]
                else:
                    actual_url = href
                
                page = self.fetch_page(actual_url)
                if page:
                    results.append(page)
            
            return results
        except Exception as e:
            print(f"搜索失败: {e}")
            return []

3.2 使用 Playwright 处理动态页面

对于 JavaScript 渲染的页面，需要使用无头浏览器：

from playwright.sync_api import sync_playwright

class DynamicCrawler:
    """动态页面爬取器（处理 JS 渲染）"""
    
    def __init__(self):
        self.playwright = sync_playwright().start()
        self.browser = self.playwright.chromium.launch(headless=True)
    
    def fetch_page(self, url: str) -> dict | None:
        page = self.browser.new_page()
        try:
            page.goto(url, timeout=15000, wait_until="networkidle")
            page.wait_for_timeout(2000)  # 等待动态内容加载
            
            title = page.title()
            content = page.evaluate("""
                () => {
                    const article = document.querySelector('article') || 
                                    document.querySelector('main') || 
                                    document.body;
                    // 移除无关元素
                    article.querySelectorAll('script,style,nav,footer').forEach(e => e.remove());
                    return article.innerText;
                }
            """)
            
            return {"url": url, "title": title, "content": content[:5000]}
        except Exception as e:
            print(f"动态爬取失败 {url}: {e}")
            return None
        finally:
            page.close()
    
    def close(self):
        self.browser.close()
        self.playwright.stop()

3.3 网页正文提取优化

使用 readability 算法可以更精准地提取正文：

from readability import Document
import requests

def extract_article(url: str) -> dict:
    """使用 readability 提取文章正文"""
    resp = requests.get(url, timeout=10, headers={
        "User-Agent": "Mozilla/5.0 (compatible; AIsearchBot/1.0)"
    })
    doc = Document(resp.text)
    
    return {
        "title": doc.title(),
        "content": doc.summary(),  # HTML 格式的正文
        "text": BeautifulSoup(doc.summary(), "html.parser").get_text(separator="\n")
    }

四、文档分块与向量化

4.1 文档分块策略

将长文档切分为适合检索的小块（chunk）是 RAG 的关键步骤：

import re
from dataclasses import dataclass

@dataclass
class Chunk:
    """文档块"""
    text: str
    metadata: dict  # 来源 URL、标题、位置等
    chunk_id: str = ""

class DocumentChunker:
    """文档分块器"""
    
    def __init__(self, chunk_size: int = 512, overlap: int = 64):
        self.chunk_size = chunk_size
        self.overlap = overlap
    
    def chunk_by_sentence(self, text: str, metadata: dict) -> list[Chunk]:
        """按句子边界分块（推荐方法）"""
        # 按句子切分
        sentences = re.split(r'(?<=[。！？.!?\n])', text)
        sentences = [s.strip() for s in sentences if s.strip()]
        
        chunks = []
        current_chunk = ""
        
        for sentence in sentences:
            # 如果加上这个句子不会超长
            if len(current_chunk) + len(sentence) <= self.chunk_size:
                current_chunk += sentence
            else:
                # 保存当前块
                if current_chunk:
                    chunks.append(Chunk(
                        text=current_chunk,
                        metadata={**metadata, "chunk_index": len(chunks)}
                    ))
                # 新块包含 overlap（上一块的尾部）
                if self.overlap > 0 and current_chunk:
                    overlap_text = current_chunk[-self.overlap:]
                    current_chunk = overlap_text + sentence
                else:
                    current_chunk = sentence
        
        # 最后一块
        if current_chunk:
            chunks.append(Chunk(
                text=current_chunk,
                metadata={**metadata, "chunk_index": len(chunks)}
            ))
        
        return chunks
    
    def chunk_by_semantic(self, text: str, metadata: dict) -> list[Chunk]:
        """按语义段落分块（适合结构化文档）"""
        # 按标题切分
        sections = re.split(r'\n(?=#{1,3}\s)', text)
        chunks = []
        
        for section in sections:
            section = section.strip()
            if not section:
                continue
            
            # 提取标题
            title_match = re.match(r'^#{1,3}\s+(.+)', section)
            section_title = title_match.group(1) if title_match else ""
            
            if len(section) <= self.chunk_size:
                chunks.append(Chunk(
                    text=section,
                    metadata={**metadata, "section_title": section_title}
                ))
            else:
                # 超长段落再按句子切分
                sub_chunks = self.chunk_by_sentence(section, metadata)
                for c in sub_chunks:
                    c.metadata["section_title"] = section_title
                chunks.extend(sub_chunks)
        
        return chunks
    
    def chunk_document(self, doc: dict) -> list[Chunk]:
        """分块一篇文档"""
        metadata = {
            "url": doc.get("url", ""),
            "title": doc.get("title", ""),
        }
        content = doc.get("content", "")
        
        chunks = self.chunk_by_sentence(content, metadata)
        
        # 生成 chunk_id
        for i, chunk in enumerate(chunks):
            chunk.chunk_id = f"{hashlib.md5(metadata['url'].encode()).hexdigest()[:8]}_{i}"
        
        return chunks

4.2 向量化

使用 BGE 系列模型生成高质量的中文向量：

from sentence_transformers import SentenceTransformer
import numpy as np

class VectorEncoder:
    """文本向量化器"""
    
    def __init__(self, model_name: str = "BAAI/bge-m3"):
        self.model = SentenceTransformer(model_name)
        self.dimension = self.model.get_sentence_embedding_dimension()
    
    def encode(self, texts: list[str], batch_size: int = 32) -> np.ndarray:
        """批量编码文本为向量"""
        embeddings = self.model.encode(
            texts,
            batch_size=batch_size,
            normalize_embeddings=True,  # 归一化，便于余弦相似度计算
            show_progress_bar=True
        )
        return embeddings
    
    def encode_query(self, query: str) -> np.ndarray:
        """编码查询（BGE 模型建议加前缀）"""
        # BGE 系列模型对查询建议加前缀
        prefixed_query = f"Represent this sentence for searching relevant passages: {query}"
        return self.model.encode([prefixed_query], normalize_embeddings=True)[0]

# 使用示例
encoder = VectorEncoder()
chunks_text = ["人工智能的发展历程", "机器学习的基本原理"]
vectors = encoder.encode(chunks_text)
print(f"向量维度: {vectors.shape}")  # (2, 1024)

4.3 向量数据库存储

使用 Chroma 作为向量数据库（轻量级，适合入门）：

import chromadb

class VectorStore:
    """向量存储"""
    
    def __init__(self, collection_name: str = "documents", persist_dir: str = "./chroma_db"):
        self.client = chromadb.PersistentClient(path=persist_dir)
        self.collection = self.client.get_or_create_collection(
            name=collection_name,
            metadata={"hnsw:space": "cosine"}  # 使用余弦相似度
        )
    
    def add_chunks(self, chunks: list[Chunk], vectors: np.ndarray):
        """添加文档块到向量数据库"""
        ids = [c.chunk_id for c in chunks]
        documents = [c.text for c in chunks]
        metadatas = [c.metadata for c in chunks]
        embeddings = vectors.tolist()
        
        # Chroma 有批量限制，分批写入
        batch_size = 500
        for i in range(0, len(ids), batch_size):
            self.collection.add(
                ids=ids[i:i+batch_size],
                embeddings=embeddings[i:i+batch_size],
                documents=documents[i:i+batch_size],
                metadatas=metadatas[i:i+batch_size]
            )
    
    def search(self, query_vector: np.ndarray, top_k: int = 10) -> list[dict]:
        """向量检索"""
        results = self.collection.query(
            query_embeddings=[query_vector.tolist()],
            n_results=top_k,
            include=["documents", "metadatas", "distances"]
        )
        
        search_results = []
        for i in range(len(results["ids"][0])):
            search_results.append({
                "id": results["ids"][0][i],
                "text": results["documents"][0][i],
                "metadata": results["metadatas"][0][i],
                "distance": results["distances"][0][i],
                "score": 1 - results["distances"][0][i]  # 余弦相似度
            })
        
        return search_results
    
    def count(self) -> int:
        return self.collection.count()

五、混合检索：BM25 + 向量检索

5.1 为什么需要混合检索

单一检索方式各有局限：

向量检索：擅长语义匹配（"怎么减肥"能匹配到"瘦身方法"），但对精确关键词匹配较弱
BM25 检索：擅长精确匹配（人名、产品型号、专业术语），但无法理解语义

混合检索将两者结合，取长补短。

5.2 BM25 检索实现

from rank_bm25 import BM25Okapi
import jieba
import numpy as np

class BM25Index:
    """BM25 检索索引"""
    
    def __init__(self):
        self.corpus = []
        self.metadata = []
        self.bm25 = None
    
    def _tokenize(self, text: str) -> list[str]:
        """中文分词"""
        return list(jieba.cut(text))
    
    def build_index(self, chunks: list[Chunk]):
        """构建 BM25 索引"""
        self.corpus = [self._tokenize(c.text) for c in chunks]
        self.metadata = [c.metadata for c in chunks]
        self.bm25 = BM25Okapi(self.corpus)
    
    def search(self, query: str, top_k: int = 10) -> list[dict]:
        """BM25 检索"""
        if self.bm25 is None:
            return []
        
        query_tokens = self._tokenize(query)
        scores = self.bm25.get_scores(query_tokens)
        
        # 获取 top_k
        top_indices = np.argsort(scores)[::-1][:top_k]
        
        results = []
        for idx in top_indices:
            if scores[idx] > 0:
                results.append({
                    "index": int(idx),
                    "text": " ".join(self.corpus[idx]),
                    "metadata": self.metadata[idx],
                    "bm25_score": float(scores[idx])
                })
        
        return results

5.3 混合检索融合

from dataclasses import dataclass

@dataclass
class HybridSearchConfig:
    vector_weight: float = 0.6    # 向量检索权重
    bm25_weight: float = 0.4      # BM25 检索权重
    vector_top_k: int = 20
    bm25_top_k: int = 20
    final_top_k: int = 10

class HybridRetriever:
    """混合检索器"""
    
    def __init__(self, vector_store: VectorStore, bm25_index: BM25Index, 
                 encoder: VectorEncoder, config: HybridSearchConfig = None):
        self.vector_store = vector_store
        self.bm25_index = bm25_index
        self.encoder = encoder
        self.config = config or HybridSearchConfig()
    
    def search(self, query: str) -> list[dict]:
        """执行混合检索"""
        # 1. 向量检索
        query_vector = self.encoder.encode_query(query)
        vector_results = self.vector_store.search(query_vector, self.config.vector_top_k)
        
        # 2. BM25 检索
        bm25_results = self.bm25_index.search(query, self.config.bm25_top_k)
        
        # 3. 分数归一化并融合
        merged = self._merge_results(vector_results, bm25_results)
        
        return merged[:self.config.final_top_k]
    
    def _merge_results(self, vector_results: list[dict], bm25_results: list[dict]) -> list[dict]:
        """归一化并合并结果（RRF 方法）"""
        # 使用 Reciprocal Rank Fusion (RRF)
        k = 60  # RRF 常数
        scores = {}
        text_map = {}
        metadata_map = {}
        
        # 向量检索排名
        for rank, result in enumerate(vector_results):
            key = result["id"]
            rrf_score = 1.0 / (k + rank + 1)
            scores[key] = scores.get(key, 0) + rrf_score * self.config.vector_weight
            text_map[key] = result["text"]
            metadata_map[key] = result["metadata"]
        
        # BM25 检索排名
        for rank, result in enumerate(bm25_results):
            key = result["metadata"].get("url", "") + "_" + str(result["index"])
            rrf_score = 1.0 / (k + rank + 1)
            scores[key] = scores.get(key, 0) + rrf_score * self.config.bm25_weight
            text_map[key] = result["text"]
            metadata_map[key] = result["metadata"]
        
        # 按融合分数排序
        sorted_keys = sorted(scores.keys(), key=lambda k: scores[k], reverse=True)
        
        return [
            {
                "id": key,
                "text": text_map[key],
                "metadata": metadata_map[key],
                "hybrid_score": scores[key]
            }
            for key in sorted_keys
        ]

六、重排序与相关性优化

6.1 为什么需要重排序

初筛（BM25 + 向量检索）速度快但精度有限。重排序模型（Reranker）会对查询和每个候选文档进行更精细的交叉编码，显著提升排序质量。

6.2 使用 BGE-Reranker

from sentence_transformers import CrossEncoder

class Reranker:
    """重排序器"""
    
    def __init__(self, model_name: str = "BAAI/bge-reranker-v2-m3"):
        self.model = CrossEncoder(model_name, max_length=512)
    
    def rerank(self, query: str, documents: list[dict], top_k: int = 5) -> list[dict]:
        """对检索结果进行重排序"""
        if not documents:
            return []
        
        # 构建 query-document 对
        pairs = [(query, doc["text"]) for doc in documents]
        
        # 计算相关性分数
        scores = self.model.predict(pairs)
        
        # 按重排序分数排序
        for i, doc in enumerate(documents):
            doc["rerank_score"] = float(scores[i])
        
        reranked = sorted(documents, key=lambda x: x["rerank_score"], reverse=True)
        return reranked[:top_k]

6.3 多阶段检索流水线

class SearchPipeline:
    """多阶段检索流水线"""
    
    def __init__(self, retriever: HybridRetriever, reranker: Reranker):
        self.retriever = retriever
        self.reranker = reranker
    
    def search(self, query: str) -> list[dict]:
        # 第一阶段：混合检索（召回 20 条候选）
        candidates = self.retriever.search(query)
        
        # 第二阶段：重排序（精选 top 5）
        reranked = self.reranker.rerank(query, candidates, top_k=5)
        
        return reranked

七、查询理解与改写

7.1 查询理解

用户输入的查询往往不够精确，需要进行理解与扩展：

class QueryProcessor:
    """查询处理器"""
    
    def __init__(self, llm_client):
        self.llm = llm_client
    
    async def rewrite_query(self, query: str, history: list = None) -> str:
        """查询改写：将口语化查询转为更适合检索的形式"""
        context = ""
        if history:
            recent = history[-3:]  # 最近 3 轮对话
            context = "\n".join([f"用户: {h['q']}\n助手: {h['a']}" for h in recent])
        
        prompt = f"""你是一个搜索查询优化器。请将用户的口语化查询改写为更适合搜索引擎的形式。

{f"对话历史：{context}" if context else ""}

用户查询：{query}

请直接返回改写后的查询，不要解释。改写规则：
1. 去除口语化表达（"那个啥"、"就是"等）
2. 补充隐含的上下文
3. 保持核心意图不变
4. 如果是追问，结合对话历史补充完整"""

        result = await self.llm.generate(prompt)
        return result.strip()
    
    async def decompose_query(self, query: str) -> list[str]:
        """查询分解：将复杂查询拆分为多个子查询"""
        prompt = f"""将以下复杂查询分解为 2-4 个简单的子查询，每个子查询聚焦一个方面。

原始查询：{query}

请以 JSON 数组格式返回子查询列表，例如：["子查询1", "子查询2"]"""

        result = await self.llm.generate(prompt)
        try:
            import json
            return json.loads(result)
        except:
            return [query]
    
    async def detect_intent(self, query: str) -> str:
        """意图检测"""
        prompt = f"""判断以下查询的意图类型，只返回类别名称。

类别：事实查询 | 操作指南 | 观点分析 | 最新资讯 | 比较对比

查询：{query}"""
        
        result = await self.llm.generate(prompt)
        return result.strip()

7.2 查询扩展

async def expand_query(self, query: str) -> list[str]:
    """查询扩展：生成相关搜索词"""
    prompt = f"""基于以下查询，生成 3-5 个相关的搜索关键词或短语，用于扩大搜索范围。

原始查询：{query}

以 JSON 数组格式返回。"""
    
    result = await self.llm.generate(prompt)
    try:
        import json
        expansions = json.loads(result)
        return [query] + expansions  # 原始查询 + 扩展
    except:
        return [query]

八、Perplexity AI 风格搜索实现

8.1 Perplexity 的核心体验

Perplexity AI 的核心特点是：

输入查询后，实时展示搜索过程
生成结构化的回答，包含内联引用标注
底部列出所有引用来源
支持追问和多轮对话

8.2 流式搜索实现

import asyncio
import json
from typing import AsyncGenerator

class PerplexityStyleEngine:
    """Perplexity AI 风格搜索引擎"""
    
    def __init__(self, pipeline: SearchPipeline, llm_client):
        self.pipeline = pipeline
        self.llm = llm_client
    
    async def search_stream(self, query: str, history: list = None) -> AsyncGenerator[dict, None]:
        """流式搜索，yield 各阶段结果"""
        
        # 阶段 1：查询理解
        yield {"type": "status", "text": "正在理解您的问题..."}
        query_processor = QueryProcessor(self.llm)
        rewritten = await query_processor.rewrite_query(query, history)
        yield {"type": "query_rewrite", "original": query, "rewritten": rewritten}
        
        # 阶段 2：检索
        yield {"type": "status", "text": "正在搜索相关信息..."}
        results = self.pipeline.search(rewritten)
        yield {"type": "search_results", "count": len(results)}
        
        # 阶段 3：生成回答
        yield {"type": "status", "text": "正在生成回答..."}
        
        # 构建上下文
        context = self._build_context(results)
        system_prompt = self._build_system_prompt()
        
        messages = [{"role": "system", "content": system_prompt}]
        if history:
            for h in history[-3:]:
                messages.append({"role": "user", "content": h["q"]})
                messages.append({"role": "assistant", "content": h["a"]})
        messages.append({"role": "user", "content": f"根据以下搜索结果回答问题。\n\n{context}\n\n问题：{query}"})
        
        # 流式生成
        full_answer = ""
        async for token in self.llm.stream_generate(messages):
            full_answer += token
            yield {"type": "answer_token", "text": token}
        
        # 阶段 4：返回引用
        sources = []
        for i, result in enumerate(results):
            sources.append({
                "index": i + 1,
                "title": result["metadata"].get("title", ""),
                "url": result["metadata"].get("url", ""),
                "snippet": result["text"][:200]
            })
        yield {"type": "sources", "data": sources}
        yield {"type": "done", "answer": full_answer}
    
    def _build_context(self, results: list[dict]) -> str:
        """构建检索上下文"""
        context_parts = []
        for i, result in enumerate(results):
            title = result["metadata"].get("title", "")
            url = result["metadata"].get("url", "")
            text = result["text"]
            context_parts.append(f"[{i+1}] 来源：{title}\n链接：{url}\n内容：{text}\n")
        return "\n---\n".join(context_parts)
    
    def _build_system_prompt(self) -> str:
        return """你是一个 AI 搜索助手。根据提供的搜索结果回答用户问题。

规则：
1. 回答准确、全面、有条理
2. 使用 [1] [2] [3] 格式标注引用来源
3. 如果搜索结果不足以回答，坦诚说明
4. 回答使用 markdown 格式
5. 先给出核心答案，再展开详细说明
6. 列出关键要点时使用列表格式"""

九、搜索结果生成与引用

9.1 引用标注系统

import re

class CitationManager:
    """引用管理器"""
    
    def __init__(self):
        self.sources = {}
    
    def register_sources(self, results: list[dict]):
        """注册可用的引用来源"""
        self.sources = {}
        for i, result in enumerate(results):
            idx = i + 1
            self.sources[idx] = {
                "title": result["metadata"].get("title", ""),
                "url": result["metadata"].get("url", ""),
                "snippet": result["text"][:300]
            }
    
    def validate_citations(self, answer: str) -> tuple[str, list[int]]:
        """验证回答中的引用是否有效，返回清理后的文本和有效引用列表"""
        # 提取所有引用标记 [1] [2] 等
        citations = re.findall(r'\[(\d+)\]', answer)
        valid_citations = []
        
        for c in citations:
            idx = int(c)
            if idx in self.sources:
                valid_citations.append(idx)
        
        valid_citations = sorted(set(valid_citations))
        return answer, valid_citations
    
    def format_sources(self, valid_citations: list[int]) -> str:
        """格式化引用来源列表"""
        lines = []
        for idx in valid_citations:
            src = self.sources.get(idx, {})
            lines.append(f"[{idx}] {src.get('title', '')} - {src.get('url', '')}")
        return "\n".join(lines)

9.2 结构化回答生成

class AnswerGenerator:
    """回答生成器"""
    
    def __init__(self, llm_client):
        self.llm = llm_client
        self.citation_mgr = CitationManager()
    
    async def generate(self, query: str, search_results: list[dict], 
                       style: str = "detailed") -> dict:
        """生成结构化回答"""
        self.citation_mgr.register_sources(search_results)
        
        # 根据风格选择 prompt
        style_prompts = {
            "detailed": "请给出详细全面的回答，包含背景信息和具体细节。",
            "concise": "请给出简洁明了的回答，突出核心要点。",
            "step_by_step": "请分步骤回答，每步清晰说明。"
        }
        
        context = self._build_context(search_results)
        style_instruction = style_prompts.get(style, style_prompts["detailed"])
        
        prompt = f"""基于以下搜索结果回答用户问题。{style_instruction}

搜索结果：
{context}

用户问题：{query}

要求：
1. 用 [序号] 格式标注引用来源
2. 回答使用 markdown 格式
3. 保持客观准确"""

        answer = await self.llm.generate(prompt)
        answer, valid_citations = self.citation_mgr.validate_citations(answer)
        sources_text = self.citation_mgr.format_sources(valid_citations)
        
        return {
            "answer": answer,
            "citations": valid_citations,
            "sources": sources_text,
            "query": query
        }
    
    def _build_context(self, results: list[dict]) -> str:
        parts = []
        for i, r in enumerate(results):
            parts.append(f"[{i+1}] {r['metadata'].get('title', '')}\n{r['text']}")
        return "\n\n".join(parts)

十、实战：构建个人 AI 搜索引擎

10.1 项目结构

ai-search-engine/
├── search_engine/
│   ├── __init__.py
│   ├── crawler.py          # 网页爬取
│   ├── chunker.py          # 文档分块
│   ├── encoder.py          # 向量化
│   ├── vector_store.py     # 向量存储
│   ├── bm25_index.py       # BM25 索引
│   ├── hybrid_retriever.py # 混合检索
│   ├── reranker.py         # 重排序
│   ├── query_processor.py  # 查询处理
│   ├── answer_generator.py # 回答生成
│   └── llm_client.py       # LLM 客户端
├── web/
│   ├── app.py              # FastAPI 服务
│   ├── static/
│   │   └── style.css
│   └── templates/
│       └── index.html
├── data/                   # 索引数据
├── requirements.txt
└── main.py                 # 入口

10.2 LLM 客户端封装

# search_engine/llm_client.py
import httpx
import json

class LLMClient:
    """LLM 调用客户端"""
    
    def __init__(self, api_base: str = "http://localhost:8000/v1", 
                 model: str = "qwen2.5-7b-instruct", api_key: str = "EMPTY"):
        self.api_base = api_base
        self.model = model
        self.api_key = api_key
    
    async def generate(self, prompt: str, temperature: float = 0.3) -> str:
        """单轮生成"""
        async with httpx.AsyncClient(timeout=60) as client:
            resp = await client.post(
                f"{self.api_base}/chat/completions",
                json={
                    "model": self.model,
                    "messages": [{"role": "user", "content": prompt}],
                    "temperature": temperature,
                    "max_tokens": 2048
                },
                headers={"Authorization": f"Bearer {self.api_key}"}
            )
            return resp.json()["choices"][0]["message"]["content"]
    
    async def stream_generate(self, messages: list, temperature: float = 0.3):
        """流式生成"""
        async with httpx.AsyncClient(timeout=60) as client:
            async with client.stream(
                "POST",
                f"{self.api_base}/chat/completions",
                json={
                    "model": self.model,
                    "messages": messages,
                    "stream": True,
                    "temperature": temperature,
                    "max_tokens": 2048
                },
                headers={"Authorization": f"Bearer {self.api_key}"}
            ) as resp:
                async for line in resp.aiter_lines():
                    if line.startswith("data: ") and line != "data: [DONE]":
                        chunk = json.loads(line[6:])
                        delta = chunk["choices"][0].get("delta", {})
                        if "content" in delta:
                            yield delta["content"]

10.3 索引构建脚本

# main.py - 构建索引
import asyncio
import json
from search_engine.crawler import WebCrawler
from search_engine.chunker import DocumentChunker
from search_engine.encoder import VectorEncoder
from search_engine.vector_store import VectorStore
from search_engine.bm25_index import BM25Index

async def build_index(seed_urls: list[str], collection_name: str = "my_search"):
    """构建搜索索引"""
    print("=== 开始构建索引 ===")
    
    # 1. 爬取网页
    print("1. 爬取网页...")
    crawler = WebCrawler(delay=1.0)
    documents = []
    for url in seed_urls:
        page = crawler.fetch_page(url)
        if page and len(page["content"]) > 100:
            documents.append(page)
            print(f"   ✓ {page['title'][:50]}")
    print(f"   共爬取 {len(documents)} 个页面")
    
    # 2. 文档分块
    print("2. 文档分块...")
    chunker = DocumentChunker(chunk_size=512, overlap=64)
    all_chunks = []
    for doc in documents:
        chunks = chunker.chunk_document(doc)
        all_chunks.extend(chunks)
    print(f"   共生成 {len(all_chunks)} 个文档块")
    
    # 3. 向量化
    print("3. 向量化...")
    encoder = VectorEncoder()
    texts = [c.text for c in all_chunks]
    vectors = encoder.encode(texts)
    print(f"   向量维度: {vectors.shape}")
    
    # 4. 存储到向量数据库
    print("4. 存储向量...")
    vector_store = VectorStore(collection_name=collection_name)
    vector_store.add_chunks(all_chunks, vectors)
    print(f"   已存储 {vector_store.count()} 条记录")
    
    # 5. 构建 BM25 索引
    print("5. 构建 BM25 索引...")
    bm25_index = BM25Index()
    bm25_index.build_index(all_chunks)
    
    # 保存 BM25 索引
    import pickle
    with open(f"data/{collection_name}_bm25.pkl", "wb") as f:
        pickle.dump(bm25_index, f)
    
    print("=== 索引构建完成 ===")
    return vector_store, bm25_index

# 运行
if __name__ == "__main__":
    # 示例：爬取一些技术博客
    urls = [
        "https://docs.python.org/3/tutorial/",
        "https://fastapi.tiangolo.com/tutorial/",
        "https://pytorch.org/tutorials/",
    ]
    asyncio.run(build_index(urls))

10.4 搜索服务

# web/app.py
from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse, StreamingResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
import json
import asyncio

from search_engine.encoder import VectorEncoder
from search_engine.vector_store import VectorStore
from search_engine.bm25_index import BM25Index
from search_engine.hybrid_retriever import HybridRetriever, HybridSearchConfig
from search_engine.reranker import Reranker
from search_engine.query_processor import QueryProcessor
from search_engine.answer_generator import AnswerGenerator
from search_engine.llm_client import LLMClient

app = FastAPI()
app.mount("/static", StaticFiles(directory="web/static"), name="static")
templates = Jinja2Templates(directory="web/templates")

# 初始化组件
encoder = VectorEncoder()
vector_store = VectorStore(collection_name="my_search")
import pickle
with open("data/my_search_bm25.pkl", "rb") as f:
    bm25_index = pickle.load(f)

retriever = HybridRetriever(vector_store, bm25_index, encoder, HybridSearchConfig())
reranker = Reranker()
llm_client = LLMClient()
query_processor = QueryProcessor(llm_client)
answer_generator = AnswerGenerator(llm_client)

# 对话历史（内存存储，生产环境应使用 Redis）
sessions = {}

@app.get("/", response_class=HTMLResponse)
async def home(request: Request):
    return templates.TemplateResponse("index.html", {"request": request})

@app.get("/api/search")
async def search(q: str, session_id: str = "default"):
    """流式搜索接口"""
    history = sessions.get(session_id, [])
    
    async def generate():
        # 查询改写
        yield f"data: {json.dumps({'type': 'status', 'text': '正在理解您的问题...'})}\n\n"
        rewritten = await query_processor.rewrite_query(q, history)
        yield f"data: {json.dumps({'type': 'rewrite', 'text': rewritten})}\n\n"
        
        # 检索
        yield f"data: {json.dumps({'type': 'status', 'text': '正在搜索...'})}\n\n"
        candidates = retriever.search(rewritten)
        
        # 重排序
        results = reranker.rerank(rewritten, candidates, top_k=5)
        yield f"data: {json.dumps({'type': 'results', 'count': len(results)})}\n\n"
        
        # 生成回答
        yield f"data: {json.dumps({'type': 'status', 'text': '正在生成回答...'})}\n\n"
        
        context = answer_generator._build_context(results)
        system_prompt = answer_generator._build_system_prompt() if hasattr(answer_generator, '_build_system_prompt') else "根据搜索结果回答问题，使用 [序号] 标注引用。"
        
        messages = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": f"搜索结果：\n{context}\n\n问题：{rewritten}"}
        ]
        
        full_answer = ""
        async for token in llm_client.stream_generate(messages):
            full_answer += token
            yield f"data: {json.dumps({'type': 'token', 'text': token})}\n\n"
        
        # 引用来源
        sources = []
        for i, r in enumerate(results):
            sources.append({
                "index": i + 1,
                "title": r["metadata"].get("title", ""),
                "url": r["metadata"].get("url", ""),
                "snippet": r["text"][:200]
            })
        yield f"data: {json.dumps({'type': 'sources', 'data': sources})}\n\n"
        yield f"data: {json.dumps({'type': 'done'})}\n\n"
        
        # 保存对话历史
        if session_id not in sessions:
            sessions[session_id] = []
        sessions[session_id].append({"q": q, "a": full_answer})
        if len(sessions[session_id]) > 10:
            sessions[session_id] = sessions[session_id][-10:]
    
    return StreamingResponse(generate(), media_type="text/event-stream")

10.5 前端界面

<!-- web/templates/index.html -->
<!DOCTYPE html>
<html lang="zh-CN">
<head>
    <meta charset="UTF-8">
    <title>AI 搜索引擎</title>
    <style>
        * { margin: 0; padding: 0; box-sizing: border-box; }
        body {
            font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", sans-serif;
            background: #fafafa;
            color: #1a1a2e;
            line-height: 1.6;
        }
        .header {
            background: #fff;
            border-bottom: 1px solid #e0e0e0;
            padding: 1rem 2rem;
            position: sticky;
            top: 0;
            z-index: 100;
        }
        .logo { font-size: 1.5rem; font-weight: 700; color: #2563eb; }
        .search-container {
            max-width: 720px;
            margin: 3rem auto 0;
            padding: 0 1rem;
        }
        .search-box {
            display: flex;
            gap: 0.5rem;
            margin-bottom: 2rem;
        }
        .search-input {
            flex: 1;
            padding: 0.8rem 1.2rem;
            border: 2px solid #e0e0e0;
            border-radius: 12px;
            font-size: 1rem;
            outline: none;
            transition: border-color 0.2s;
        }
        .search-input:focus { border-color: #2563eb; }
        .search-btn {
            padding: 0.8rem 1.5rem;
            background: #2563eb;
            color: #fff;
            border: none;
            border-radius: 12px;
            font-size: 1rem;
            cursor: pointer;
        }
        .search-btn:hover { background: #1d4ed8; }
        .result-container {
            max-width: 720px;
            margin: 0 auto;
            padding: 0 1rem;
        }
        .status-bar {
            color: #6b7280;
            font-size: 0.9rem;
            margin-bottom: 1rem;
        }
        .answer-card {
            background: #fff;
            border-radius: 12px;
            padding: 1.5rem;
            margin-bottom: 1.5rem;
            box-shadow: 0 1px 3px rgba(0,0,0,0.1);
        }
        .answer-content { 
            font-size: 1.05rem; 
            line-height: 1.8;
        }
        .answer-content strong { color: #1a1a2e; }
        .answer-content code {
            background: #f3f4f6;
            padding: 0.1rem 0.4rem;
            border-radius: 4px;
            font-size: 0.9rem;
        }
        .sources-card {
            background: #fff;
            border-radius: 12px;
            padding: 1.5rem;
            box-shadow: 0 1px 3px rgba(0,0,0,0.1);
        }
        .sources-title {
            font-size: 1rem;
            font-weight: 600;
            margin-bottom: 1rem;
            color: #4b5563;
        }
        .source-item {
            display: flex;
            gap: 1rem;
            padding: 0.8rem 0;
            border-bottom: 1px solid #f3f4f6;
        }
        .source-item:last-child { border-bottom: none; }
        .source-index {
            width: 28px;
            height: 28px;
            background: #eff6ff;
            color: #2563eb;
            border-radius: 50%;
            display: flex;
            align-items: center;
            justify-content: center;
            font-size: 0.8rem;
            font-weight: 600;
            flex-shrink: 0;
        }
        .source-info { flex: 1; }
        .source-title { 
            font-weight: 500; 
            color: #1a1a2e;
            text-decoration: none;
        }
        .source-title:hover { color: #2563eb; }
        .source-snippet {
            font-size: 0.85rem;
            color: #6b7280;
            margin-top: 0.3rem;
            display: -webkit-box;
            -webkit-line-clamp: 2;
            -webkit-box-orient: vertical;
            overflow: hidden;
        }
        .hidden { display: none; }
    </style>
</head>
<body>
    <div class="header">
        <span class="logo">🔍 AI Search</span>
    </div>
    
    <div class="search-container">
        <div class="search-box">
            <input type="text" class="search-input" id="queryInput" 
                   placeholder="输入你的问题..." autofocus>
            <button class="search-btn" id="searchBtn">搜索</button>
        </div>
    </div>
    
    <div class="result-container hidden" id="results">
        <div class="status-bar" id="status"></div>
        <div class="answer-card">
            <div class="answer-content" id="answerContent"></div>
        </div>
        <div class="sources-card hidden" id="sourcesCard">
            <div class="sources-title">📚 参考来源</div>
            <div id="sourcesList"></div>
        </div>
    </div>

    <script>
        const queryInput = document.getElementById('queryInput');
        const searchBtn = document.getElementById('searchBtn');
        const results = document.getElementById('results');
        const status = document.getElementById('status');
        const answerContent = document.getElementById('answerContent');
        const sourcesCard = document.getElementById('sourcesCard');
        const sourcesList = document.getElementById('sourcesList');

        let currentAnswer = '';

        searchBtn.addEventListener('click', doSearch);
        queryInput.addEventListener('keydown', (e) => {
            if (e.key === 'Enter') doSearch();
        });

        async function doSearch() {
            const query = queryInput.value.trim();
            if (!query) return;

            results.classList.remove('hidden');
            answerContent.innerHTML = '';
            sourcesCard.classList.add('hidden');
            sourcesList.innerHTML = '';
            currentAnswer = '';
            status.textContent = '搜索中...';

            const eventSource = new EventSource(`/api/search?q=${encodeURIComponent(query)}`);
            
            eventSource.onmessage = (event) => {
                const data = JSON.parse(event.data);
                
                switch (data.type) {
                    case 'status':
                        status.textContent = data.text;
                        break;
                    case 'rewrite':
                        if (data.text !== query) {
                            status.textContent = `已优化查询: ${data.text}`;
                        }
                        break;
                    case 'results':
                        status.textContent = `找到 ${data.count} 条相关结果`;
                        break;
                    case 'token':
                        currentAnswer += data.text;
                        answerContent.innerHTML = renderMarkdown(currentAnswer);
                        break;
                    case 'sources':
                        renderSources(data.data);
                        break;
                    case 'done':
                        status.textContent = '搜索完成';
                        eventSource.close();
                        break;
                }
            };

            eventSource.onerror = () => {
                status.textContent = '搜索出错，请重试';
                eventSource.close();
            };
        }

        function renderMarkdown(text) {
            // 简单 markdown 渲染
            return text
                .replace(/\*\*(.*?)\*\*/g, '<strong>$1</strong>')
                .replace(/\*(.*?)\*/g, '<em>$1</em>')
                .replace(/`(.*?)`/g, '<code>$1</code>')
                .replace(/\[(\d+)\]/g, '<sup style="color:#2563eb;cursor:pointer" title="查看来源">[$1]</sup>')
                .replace(/\n/g, '<br>');
        }

        function renderSources(sources) {
            sourcesCard.classList.remove('hidden');
            sourcesList.innerHTML = sources.map(s => `
                <div class="source-item">
                    <div class="source-index">${s.index}</div>
                    <div class="source-info">
                        <a href="${s.url}" target="_blank" class="source-title">${s.title}</a>
                        <div class="source-snippet">${s.snippet}</div>
                    </div>
                </div>
            `).join('');
        }
    </script>
</body>
</html>

10.6 启动运行

# 安装依赖
pip install fastapi uvicorn chromadb sentence-transformers \
    rank_bm25 jieba beautifulsoup4 httpx readability-lxml jinja2

# 1. 构建索引
python main.py

# 2. 启动搜索服务
uvicorn web.app:app --host 0.0.0.0 --port 8080

# 3. 浏览器访问 http://localhost:8080

10.7 进阶优化建议

1. 增量索引更新

class IncrementalIndexer:
    """增量索引更新"""
    
    def __init__(self, vector_store: VectorStore, bm25_index: BM25Index):
        self.vector_store = vector_store
        self.bm25_index = bm25_index
        self.crawled_urls = set()
    
    async def update(self, urls: list[str]):
        """增量更新指定 URL 的索引"""
        new_urls = [u for u in urls if u not in self.crawled_urls]
        if not new_urls:
            print("没有新 URL 需要索引")
            return
        
        # 爬取、分块、向量化、存储
        crawler = WebCrawler()
        chunker = DocumentChunker()
        encoder = VectorEncoder()
        
        for url in new_urls:
            page = crawler.fetch_page(url)
            if page:
                chunks = chunker.chunk_document(page)
                vectors = encoder.encode([c.text for c in chunks])
                self.vector_store.add_chunks(chunks, vectors)
                self.crawled_urls.add(url)
                print(f"已索引: {url}")

2. 缓存策略

import hashlib
import json
from functools import lru_cache

class SearchCache:
    """搜索结果缓存"""
    
    def __init__(self, ttl: int = 3600):
        self.cache = {}
        self.ttl = ttl
    
    def _key(self, query: str) -> str:
        return hashlib.md5(query.encode()).hexdigest()
    
    def get(self, query: str) -> dict | None:
        key = self._key(query)
        if key in self.cache:
            entry = self.cache[key]
            if time.time() - entry["time"] < self.ttl:
                return entry["data"]
            del self.cache[key]
        return None
    
    def set(self, query: str, data: dict):
        key = self._key(query)
        self.cache[key] = {"data": data, "time": time.time()}

3. 多源搜索整合

class MultiSourceRetriever:
    """多源检索器：同时检索本地索引和实时网络"""
    
    def __init__(self, local_retriever: HybridRetriever, web_crawler: WebCrawler):
        self.local = local_retriever
        self.web = web_crawler
    
    async def search(self, query: str) -> list[dict]:
        # 并行执行本地检索和网络搜索
        import asyncio
        
        local_task = asyncio.create_task(
            asyncio.to_thread(self.local.search, query)
        )
        web_task = asyncio.create_task(
            asyncio.to_thread(self.web.crawl_search_results, query, num_results=5)
        )
        
        local_results, web_results = await asyncio.gather(local_task, web_task)
        
        # 合并结果（本地优先，网络补充）
        combined = []
        seen_urls = set()
        
        for r in local_results:
            url = r["metadata"].get("url", "")
            if url not in seen_urls:
                combined.append(r)
                seen_urls.add(url)
        
        for page in web_results:
            if page["url"] not in seen_urls:
                combined.append({
                    "text": page["content"][:1000],
                    "metadata": {"title": page["title"], "url": page["url"]},
                    "hybrid_score": 0.3  # 网络结果给较低初始分
                })
                seen_urls.add(page["url"])
        
        return combined

十一、总结

本教程从零开始构建了一个 Perplexity AI 风格的搜索引擎，涵盖了完整的 RAG 搜索链路：

网页爬取：从互联网获取原始内容
文档分块：将长文档切分为可检索的小块
向量化：使用 BGE 模型生成语义向量
混合检索：BM25 + 向量检索取长补短
重排序：使用交叉编码器精排
查询改写：LLM 优化用户查询
回答生成：基于检索结果生成有引用的回答
流式输出：实时展示搜索过程

进阶方向：

多模态搜索：支持图片、视频搜索
个性化排序：基于用户历史调整排序
实时索引：持续爬取更新索引
分布式架构：Milvus + Elasticsearch 支撑大规模数据
意图路由：不同查询类型走不同的处理流程

AI 搜索引擎是 RAG 技术最典型的应用场景，掌握这套技术栈将为你打开大模型应用开发的大门。

本教程内容原创，仅供学习交流使用。

AI搜索引擎开发教程