AI原生搜索引擎开发完全教程

从零构建下一代AI搜索体验：语义检索、多源融合、答案生成与引用溯源

前言

传统搜索引擎（如Google、百度）的核心逻辑是"关键词匹配→链接排序→用户自行浏览"。用户需要输入精确的关键词，然后从一堆蓝色链接中自己寻找答案。而AI原生搜索引擎（如Perplexity、秘塔搜索、SearchGPT）彻底改变了这一范式——它们直接理解用户的自然语言问题，实时检索互联网信息，然后生成结构化的、带引用来源的综合答案。

本教程将带你从零开始，系统掌握AI原生搜索引擎的核心架构与开发技术，最终构建一个类似Perplexity的AI搜索引擎原型。

第一章：AI原生搜索引擎架构总览

1.1 与传统搜索引擎的核心区别

1.2 核心架构模块

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

用户查询 (Query)
    │
    ▼
┌─────────────────────┐
│  Query理解与改写     │  ← 意图识别、查询扩展、多轮理解
└─────────┬───────────┘
          │
          ▼
┌─────────────────────┐
│  混合检索引擎        │  ← 关键词检索 + 语义检索 + 知识图谱
└─────────┬───────────┘
          │
          ▼
┌─────────────────────┐
│  实时信息获取        │  ← 网页爬取、API调用、实时数据
└─────────┬───────────┘
          │
          ▼
┌─────────────────────┐
│  信息融合与去重      │  ← 多源合并、冲突消解、可信度评估
└─────────┬───────────┘
          │
          ▼
┌─────────────────────┐
│  结果排序与重排      │  ← 语义相关性、时效性、权威性
└─────────┬───────────┘
          │
          ▼
┌─────────────────────┐
│  答案生成与引用溯源  │  ← LLM生成、引用标注、事实校验
└─────────┬───────────┘
          │
          ▼
    结构化答案输出

第二章：Query理解与改写

2.1 为什么需要Query理解

用户的搜索查询往往是模糊的、口语化的、甚至有歧义的。Query理解模块的任务是将用户的原始查询转化为更适合检索的形式。

核心任务包括：

• 意图分类：判断查询类型（事实性问题、导航性查询、交易性查询、建议性问题）
• 实体识别：提取查询中的关键实体（人名、地名、时间、事件）
• 查询扩展：添加同义词、相关词，提高召回率
• 多轮理解：结合对话历史，理解代词指代和上下文

2.2 基于LLM的Query理解实现

import openai
from dataclasses import dataclass
from typing import List, Optional
from enum import Enum

class QueryIntent(Enum):
    FACTUAL = "factual"           # 事实性问题："Python是什么时候发布的？"
    NAVIGATIONAL = "navigational" # 导航性查询："GitHub官网"
    INFORMATIONAL = "informational"  # 信息查询："如何学习机器学习"
    TRANSACTIONAL = "transactional"  # 交易性查询："购买iPhone 16"
    SUGGESTION = "suggestion"     # 建议性问题："推荐一些Python框架"

@dataclass
class QueryUnderstandingResult:
    original_query: str
    intent: QueryIntent
    entities: List[str]
    rewritten_queries: List[str]  # 改写后的多个查询
    expanded_terms: List[str]     # 扩展关键词
    time_constraint: Optional[str]  # 时间约束，如"最近"、"2024年"

class QueryUnderstandingEngine:
    """基于LLM的Query理解引擎"""
    
    def __init__(self, api_key: str, model: str = "gpt-4o-mini"):
        self.client = openai.OpenAI(api_key=api_key)
        self.model = model
    
    async def understand(self, query: str, 
                         chat_history: List[dict] = None) -> QueryUnderstandingResult:
        """对用户查询进行深度理解"""
        
        history_context = ""
        if chat_history:
            history_context = "\n对话历史：\n" + "\n".join(
                f"{m['role']}: {m['content']}" for m in chat_history[-5:]
            )
        
        prompt = f"""分析以下搜索查询，返回JSON格式结果：
        
查询：{query}
{history_context}

请返回以下JSON格式：
{{
    "intent": "factual|navigational|informational|transactional|suggestion",
    "entities": ["实体1", "实体2"],
    "rewritten_queries": ["改写查询1", "改写查询2", "改写查询3"],
    "expanded_terms": ["扩展词1", "扩展词2", "扩展词3"],
    "time_constraint": "时间约束或null"
}}

改写规则：
1. 将口语化表达转为更精确的搜索查询
2. 消除代词指代（结合对话历史）
3. 生成2-3个不同角度的改写查询
4. 扩展词应包含同义词和相关概念"""

        response = self.client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
            response_format={"type": "json_object"},
            temperature=0.1
        )
        
        result = json.loads(response.choices[0].message.content)
        
        return QueryUnderstandingResult(
            original_query=query,
            intent=QueryIntent(result["intent"]),
            entities=result.get("entities", []),
            rewritten_queries=result.get("rewritten_queries", [query]),
            expanded_terms=result.get("expanded_terms", []),
            time_constraint=result.get("time_constraint")
        )
    
    def generate_search_queries(self, understanding: QueryUnderstandingResult) -> List[str]:
        """生成用于检索的查询组合"""
        queries = [understanding.original_query]
        queries.extend(understanding.rewritten_queries)
        
        # 将扩展词组合成额外查询
        if understanding.expanded_terms:
            expanded_query = understanding.original_query + " " + " ".join(
                understanding.expanded_terms[:3]
            )
            queries.append(expanded_query)
        
        # 去重并限制数量
        seen = set()
        unique_queries = []
        for q in queries:
            normalized = q.strip().lower()
            if normalized not in seen:
                seen.add(normalized)
                unique_queries.append(q)
        
        return unique_queries[:5]  # 最多5个查询

2.3 查询改写的几种策略

class QueryRewriter:
    """多种查询改写策略"""
    
    @staticmethod
    def hyde_rewrite(query: str, llm_client) -> str:
        """
        HyDE (Hypothetical Document Embeddings) 改写策略
        生成一个假设性的答案文档，用它来检索
        """
        prompt = f"请针对以下问题，写一段可能包含答案的简短文档：\n问题：{query}"
        response = llm_client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
            max_tokens=200
        )
        return response.choices[0].message.content
    
    @staticmethod
    def step_back_rewrite(query: str, llm_client) -> str:
        """
        Step-back改写策略
        将具体问题抽象为更通用的问题
        """
        prompt = f"""将以下具体问题改写为更通用、更抽象的问题：
        
原始问题：{query}
        
示例：
- "Python 3.12有什么新特性？" → "Python最新版本有哪些重要更新？"
- "特斯拉Q3财报怎么样？" → "特斯拉最近的财务表现如何？"

只返回改写后的问题，不要解释。"""
        
        response = llm_client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
            max_tokens=100
        )
        return response.choices[0].message.content
    
    @staticmethod
    def sub_query_decomposition(query: str, llm_client) -> List[str]:
        """
        子查询分解策略
        将复杂问题分解为多个简单子问题
        """
        prompt = f"""将以下复杂问题分解为2-4个简单的子问题：

问题：{query}

返回JSON格式：{{"sub_queries": ["子问题1", "子问题2", ...]}}"""
        
        response = llm_client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
            response_format={"type": "json_object"},
            temperature=0.1
        )
        result = json.loads(response.choices[0].message.content)
        return result.get("sub_queries", [query])

第三章：混合检索引擎

3.1 为什么需要混合检索

单一检索方式各有局限：

• 关键词检索（BM25）：精确匹配能力强，但无法理解语义（"汽车"搜不到"轿车"）
• 语义检索（向量搜索）：理解语义相似性，但对精确实体匹配较弱
• 知识图谱检索：结构化知识推理，但覆盖范围有限

混合检索将三者结合，取长补短，是AI搜索引擎的核心能力。

3.2 向量语义检索实现

import numpy as np
from sentence_transformers import SentenceTransformer
import faiss
from typing import List, Tuple
import hashlib
import json

class VectorSearchEngine:
    """基于FAISS的向量语义检索引擎"""
    
    def __init__(self, model_name: str = "BAAI/bge-large-zh-v1.5"):
        self.encoder = SentenceTransformer(model_name)
        self.dimension = self.encoder.get_sentence_embedding_dimension()
        self.index = faiss.IndexFlatIP(self.dimension)  # 内积相似度
        self.documents: List[dict] = []  # 存储原始文档
    
    def add_documents(self, documents: List[dict]):
        """添加文档到索引"""
        texts = [doc["content"] for doc in documents]
        embeddings = self.encoder.encode(texts, normalize_embeddings=True)
        self.index.add(embeddings.astype('float32'))
        self.documents.extend(documents)
    
    def search(self, query: str, top_k: int = 10) -> List[Tuple[dict, float]]:
        """语义检索"""
        query_embedding = self.encoder.encode([query], normalize_embeddings=True)
        scores, indices = self.index.search(query_embedding.astype('float32'), top_k)
        
        results = []
        for score, idx in zip(scores[0], indices[0]):
            if idx < len(self.documents):
                results.append((self.documents[idx], float(score)))
        return results
    
    def batch_search(self, queries: List[str], top_k: int = 10) -> List[List[Tuple[dict, float]]]:
        """批量检索"""
        query_embeddings = self.encoder.encode(queries, normalize_embeddings=True)
        scores, indices = self.index.search(query_embeddings.astype('float32'), top_k)
        
        all_results = []
        for query_scores, query_indices in zip(scores, indices):
            results = []
            for score, idx in zip(query_scores, query_indices):
                if idx < len(self.documents):
                    results.append((self.documents[idx], float(score)))
            all_results.append(results)
        return all_results

3.3 BM25关键词检索

import math
from collections import Counter, defaultdict
from typing import List, Tuple
import jieba

class BM25SearchEngine:
    """BM25关键词检索引擎"""
    
    def __init__(self, k1: float = 1.5, b: float = 0.75):
        self.k1 = k1
        self.b = b
        self.documents: List[List[str]] = []
        self.doc_metadata: List[dict] = []
        self.avg_doc_len = 0
        self.doc_freq: dict = defaultdict(int)  # 包含term的文档数
        self.doc_len: List[int] = []
        self.idf_cache: dict = {}
    
    def add_documents(self, documents: List[dict]):
        """添加文档"""
        for doc in documents:
            tokens = list(jieba.cut(doc["content"]))
            self.documents.append(tokens)
            self.doc_metadata.append(doc)
            self.doc_len.append(len(tokens))
            
            # 更新文档频率
            unique_tokens = set(tokens)
            for token in unique_tokens:
                self.doc_freq[token] += 1
        
        self.avg_doc_len = sum(self.doc_len) / len(self.doc_len) if self.doc_len else 0
        
        # 预计算IDF
        n = len(self.documents)
        for term, freq in self.doc_freq.items():
            self.idf_cache[term] = math.log((n - freq + 0.5) / (freq + 0.5) + 1)
    
    def _bm25_score(self, query_tokens: List[str], doc_idx: int) -> float:
        """计算单个文档的BM25分数"""
        doc = self.documents[doc_idx]
        doc_len = self.doc_len[doc_idx]
        term_freq = Counter(doc)
        
        score = 0.0
        for token in query_tokens:
            if token not in term_freq:
                continue
            
            tf = term_freq[token]
            idf = self.idf_cache.get(token, 0)
            
            numerator = tf * (self.k1 + 1)
            denominator = tf + self.k1 * (1 - self.b + self.b * doc_len / self.avg_doc_len)
            score += idf * (numerator / denominator)
        
        return score
    
    def search(self, query: str, top_k: int = 10) -> List[Tuple[dict, float]]:
        """BM25检索"""
        query_tokens = list(jieba.cut(query))
        scores = []
        
        for idx in range(len(self.documents)):
            score = self._bm25_score(query_tokens, idx)
            scores.append((idx, score))
        
        scores.sort(key=lambda x: x[1], reverse=True)
        
        results = []
        for idx, score in scores[:top_k]:
            if score > 0:
                results.append((self.doc_metadata[idx], score))
        return results

3.4 混合检索融合

class HybridSearchEngine:
    """混合检索引擎：融合BM25 + 向量语义检索"""
    
    def __init__(self, vector_weight: float = 0.6, bm25_weight: float = 0.4):
        self.vector_engine = VectorSearchEngine()
        self.bm25_engine = BM25SearchEngine()
        self.vector_weight = vector_weight
        self.bm25_weight = bm25_weight
    
    def add_documents(self, documents: List[dict]):
        """同时添加到两个引擎"""
        self.vector_engine.add_documents(documents)
        self.bm25_engine.add_documents(documents)
    
    def search(self, query: str, top_k: int = 10) -> List[Tuple[dict, float]]:
        """混合检索，使用RRF (Reciprocal Rank Fusion) 融合排序"""
        # 分别检索
        vector_results = self.vector_engine.search(query, top_k=top_k * 2)
        bm25_results = self.bm25_engine.search(query, top_k=top_k * 2)
        
        # RRF融合
        k = 60  # RRF常数
        doc_scores: dict = {}
        doc_map: dict = {}
        
        for rank, (doc, _) in enumerate(vector_results):
            doc_id = hashlib.md5(doc["content"].encode()).hexdigest()
            rrf_score = 1.0 / (k + rank + 1)
            doc_scores[doc_id] = doc_scores.get(doc_id, 0) + rrf_score * self.vector_weight
            doc_map[doc_id] = doc
        
        for rank, (doc, _) in enumerate(bm25_results):
            doc_id = hashlib.md5(doc["content"].encode()).hexdigest()
            rrf_score = 1.0 / (k + rank + 1)
            doc_scores[doc_id] = doc_scores.get(doc_id, 0) + rrf_score * self.bm25_weight
            doc_map[doc_id] = doc
        
        # 排序
        sorted_docs = sorted(doc_scores.items(), key=lambda x: x[1], reverse=True)
        
        results = []
        for doc_id, score in sorted_docs[:top_k]:
            results.append((doc_map[doc_id], score))
        return results

第四章：实时信息获取与网页解析

4.1 搜索引擎API集成

import httpx
from bs4 import BeautifulSoup
from typing import List, Optional
from dataclasses import dataclass
import asyncio
import hashlib
from datetime import datetime

@dataclass
class SearchResult:
    title: str
    url: str
    snippet: str
    content: Optional[str] = None
    published_date: Optional[str] = None
    source_domain: str = ""
    credibility_score: float = 0.5

class WebSearchEngine:
    """实时网络搜索引擎"""
    
    def __init__(self, serper_api_key: str = None, bing_api_key: str = None):
        self.serper_api_key = serper_api_key
        self.bing_api_key = bing_api_key
        self.client = httpx.AsyncClient(timeout=15)
    
    async def search(self, query: str, num_results: int = 10, 
                     time_range: str = None) -> List[SearchResult]:
        """执行网络搜索"""
        if self.serper_api_key:
            return await self._search_serper(query, num_results, time_range)
        elif self.bing_api_key:
            return await self._search_bing(query, num_results)
        else:
            raise ValueError("需要配置至少一个搜索API密钥")
    
    async def _search_serper(self, query: str, num: int, 
                              time_range: str = None) -> List[SearchResult]:
        """使用Serper API搜索"""
        payload = {"q": query, "num": num}
        if time_range:
            payload["tbs"] = time_range  # 如 "qdr:d"(过去一天), "qdr:w"(过去一周)
        
        resp = await self.client.post(
            "https://google.serper.dev/search",
            json=payload,
            headers={"X-API-KEY": self.serper_api_key}
        )
        data = resp.json()
        
        results = []
        for item in data.get("organic", []):
            results.append(SearchResult(
                title=item.get("title", ""),
                url=item.get("link", ""),
                snippet=item.get("snippet", ""),
                source_domain=item.get("link", "").split("/")[2] if "/" in item.get("link", "") else "",
                credibility_score=self._estimate_credibility(item.get("link", ""))
            ))
        return results
    
    async def _search_bing(self, query: str, num: int) -> List[SearchResult]:
        """使用Bing API搜索"""
        resp = await self.client.get(
            "https://api.bing.microsoft.com/v7.0/search",
            params={"q": query, "count": num, "mkt": "zh-CN"},
            headers={"Ocp-Apim-Subscription-Key": self.bing_api_key}
        )
        data = resp.json()
        
        results = []
        for item in data.get("webPages", {}).get("value", []):
            results.append(SearchResult(
                title=item.get("name", ""),
                url=item.get("url", ""),
                snippet=item.get("snippet", ""),
                source_domain=item.get("url", "").split("/")[2] if "/" in item.get("url", "") else ""
            ))
        return results
    
    def _estimate_credibility(self, url: str) -> float:
        """估算来源可信度"""
        high_credibility = [
            "wikipedia.org", "github.com", "arxiv.org",
            "stackoverflow.com", "nature.com", "science.org",
            "gov.cn", "edu.cn", "bbc.com", "reuters.com"
        ]
        medium_credibility = [
            "medium.com", "zhihu.com", "csdn.net",
            "jianshu.com", "segmentfault.com"
        ]
        
        domain = url.split("/")[2] if len(url.split("/")) > 2 else ""
        
        for hd in high_credibility:
            if hd in domain:
                return 0.9
        for md in medium_credibility:
            if md in domain:
                return 0.7
        return 0.5

4.2 网页内容提取与解析

class WebContentExtractor:
    """网页内容提取器"""
    
    def __init__(self):
        self.client = httpx.AsyncClient(
            timeout=10,
            headers={"User-Agent": "Mozilla/5.0 (compatible; AISearchBot/1.0)"}
        )
    
    async def extract_content(self, url: str, max_length: int = 5000) -> Optional[str]:
        """提取网页正文内容"""
        try:
            resp = await self.client.get(url, follow_redirects=True)
            resp.raise_for_status()
            
            soup = BeautifulSoup(resp.text, 'html.parser')
            
            # 移除无用标签
            for tag in soup.find_all(['script', 'style', 'nav', 'footer', 
                                       'header', 'aside', 'iframe']):
                tag.decompose()
            
            # 尝试找正文容器
            content = self._find_main_content(soup)
            
            if not content:
                content = soup.get_text(separator='\n', strip=True)
            
            # 清理和截断
            lines = [line.strip() for line in content.split('\n') if line.strip()]
            content = '\n'.join(lines)
            
            return content[:max_length]
            
        except Exception as e:
            print(f"提取失败 {url}: {e}")
            return None
    
    def _find_main_content(self, soup: BeautifulSoup) -> Optional[str]:
        """智能识别正文区域"""
        # 尝试常见的正文容器选择器
        selectors = [
            'article', 'main', '.content', '.article-content',
            '.post-content', '#content', '.entry-content',
            '[role="main"]'
        ]
        
        for selector in selectors:
            element = soup.select_one(selector)
            if element:
                text = element.get_text(separator='\n', strip=True)
                if len(text) > 100:  # 确保是有效内容
                    return text
        return None
    
    async def batch_extract(self, urls: List[str], 
                           max_concurrent: int = 5) -> List[Optional[str]]:
        """并发批量提取"""
        semaphore = asyncio.Semaphore(max_concurrent)
        
        async def extract_with_limit(url):
            async with semaphore:
                return await self.extract_content(url)
        
        tasks = [extract_with_limit(url) for url in urls]
        return await asyncio.gather(*tasks)

第五章：结果排序与重排（Reranking）

5.1 为什么要重排

初步检索（召回阶段）追求的是高召回率，会返回大量可能相关的文档。重排阶段则使用更精确的模型对候选文档进行精细排序，提升准确率。

5.2 Cross-Encoder重排器

from sentence_transformers import CrossEncoder
from typing import List, Tuple

class SemanticReranker:
    """基于Cross-Encoder的语义重排器"""
    
    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[Tuple[dict, float]]:
        """对检索结果进行重排"""
        pairs = [(query, doc["content"][:512]) for doc in documents]
        scores = self.model.predict(pairs)
        
        scored_docs = list(zip(documents, scores))
        scored_docs.sort(key=lambda x: x[1], reverse=True)
        
        return scored_docs[:top_k]


class MultiSignalRanker:
    """多信号融合排序器"""
    
    def __init__(self, reranker: SemanticReranker):
        self.reranker = reranker
    
    def rank(self, query: str, candidates: List[dict],
             weights: dict = None) -> List[Tuple[dict, float]]:
        """
        多维度综合排序
        
        weights参数控制各维度权重：
        - semantic_score: 语义相关性
        - credibility: 来源可信度
        - freshness: 时效性
        - content_quality: 内容质量
        """
        if weights is None:
            weights = {
                "semantic_score": 0.45,
                "credibility": 0.2,
                "freshness": 0.2,
                "content_quality": 0.15
            }
        
        # 计算各维度分数
        reranked = self.reranker.rerank(query, candidates, top_k=len(candidates))
        
        scored_results = []
        for doc, semantic_score in reranked:
            final_score = (
                weights["semantic_score"] * self._normalize(semantic_score, -10, 10) +
                weights["credibility"] * doc.get("credibility_score", 0.5) +
                weights["freshness"] * self._freshness_score(doc.get("published_date")) +
                weights["content_quality"] * self._content_quality_score(doc.get("content", ""))
            )
            scored_results.append((doc, final_score))
        
        scored_results.sort(key=lambda x: x[1], reverse=True)
        return scored_results
    
    def _normalize(self, value: float, min_val: float, max_val: float) -> float:
        """归一化到0-1"""
        return max(0, min(1, (value - min_val) / (max_val - min_val)))
    
    def _freshness_score(self, date_str: Optional[str]) -> float:
        """时效性评分：越新越高"""
        if not date_str:
            return 0.3  # 无日期信息，给默认分
        try:
            pub_date = datetime.fromisoformat(date_str.replace('Z', '+00:00'))
            days_old = (datetime.now() - pub_date.replace(tzinfo=None)).days
            if days_old <= 1:
                return 1.0
            elif days_old <= 7:
                return 0.9
            elif days_old <= 30:
                return 0.7
            elif days_old <= 365:
                return 0.5
            else:
                return 0.3
        except:
            return 0.3
    
    def _content_quality_score(self, content: str) -> float:
        """内容质量评分"""
        if not content:
            return 0.0
        
        score = 0.5
        
        # 长度评分
        if len(content) > 500:
            score += 0.1
        if len(content) > 2000:
            score += 0.1
        
        # 结构化内容加分
        if '```' in content or '<code>' in content:
            score += 0.1
        if any(marker in content for marker in ['##', '**', '- ', '1.']):
            score += 0.1
        
        return min(1.0, score)

第六章：答案生成与引用溯源

6.1 带引用的答案生成

这是AI搜索引擎最核心的功能——基于检索到的信息，生成结构化答案并标注引用来源。

from typing import List
import re
import json

@dataclass
class Citation:
    """引用信息"""
    index: int           # 引用编号 [1], [2] 等
    title: str
    url: str
    snippet: str         # 被引用的具体片段
    source_domain: str

@dataclass
class SearchAnswer:
    """搜索答案"""
    answer: str          # 生成的答案
    citations: List[Citation]  # 引用列表
    confidence: float    # 答案置信度
    follow_up_questions: List[str]  # 推荐追问

class AnswerGenerator:
    """带引用溯源的答案生成器"""
    
    def __init__(self, api_key: str, model: str = "gpt-4o"):
        self.client = openai.OpenAI(api_key=api_key)
        self.model = model
    
    async def generate(self, query: str, 
                       search_results: List[SearchResult],
                       max_citations: int = 8) -> SearchAnswer:
        """基于检索结果生成带引用的答案"""
        
        # 准备上下文
        context_parts = []
        citation_map = {}
        
        for i, result in enumerate(search_results[:max_citations], 1):
            citation_map[i] = Citation(
                index=i,
                title=result.title,
                url=result.url,
                snippet=result.snippet,
                source_domain=result.source_domain
            )
            context_parts.append(
                f"[来源{i}] {result.title}\n"
                f"URL: {result.url}\n"
                f"内容: {result.snippet}"
            )
        
        context = "\n\n".join(context_parts)
        
        prompt = f"""基于以下检索到的信息，回答用户的问题。

要求：
1. 综合多个来源的信息，生成全面、准确的答案
2. 在答案中使用 [数字] 标注信息来源（如 [1]、[2]）
3. 如果不同来源的信息有冲突，请指出并分析
4. 如果信息不足以回答问题，诚实说明
5. 保持客观、准确，不要编造信息
6. 使用Markdown格式，适当使用标题、列表等结构

检索到的信息：
{context}

用户问题：{query}

请按以下JSON格式返回：
{{
    "answer": "你的答案（Markdown格式，包含[数字]引用）",
    "confidence": 0.0到1.0的置信度,
    "follow_up_questions": ["追问1", "追问2", "追问3"]
}}"""

        response = self.client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
            response_format={"type": "json_object"},
            temperature=0.3
        )
        
        result = json.loads(response.choices[0].message.content)
        
        return SearchAnswer(
            answer=result["answer"],
            citations=list(citation_map.values()),
            confidence=result.get("confidence", 0.7),
            follow_up_questions=result.get("follow_up_questions", [])
        )

6.2 流式答案生成（Streaming）

import asyncio
from typing import AsyncGenerator

class StreamingAnswerGenerator(AnswerGenerator):
    """支持流式输出的答案生成器"""
    
    async def generate_stream(self, query: str, 
                               search_results: List[SearchResult]) -> AsyncGenerator[str, None]:
        """流式生成答案，实现打字机效果"""
        
        context_parts = []
        citations = []
        for i, result in enumerate(search_results[:8], 1):
            citations.append(Citation(
                index=i, title=result.title, url=result.url,
                snippet=result.snippet, source_domain=result.source_domain
            ))
            context_parts.append(f"[来源{i}] {result.title}\n{result.snippet}")
        
        context = "\n\n".join(context_parts)
        
        messages = [
            {"role": "system", "content": "你是一个AI搜索引擎的助手。基于检索到的信息回答问题，使用[数字]标注引用来源。"},
            {"role": "user", "content": f"检索信息：\n{context}\n\n问题：{query}"}
        ]
        
        # 流式调用
        stream = self.client.chat.completions.create(
            model=self.model,
            messages=messages,
            stream=True,
            temperature=0.3,
            max_tokens=2000
        )
        
        for chunk in stream:
            if chunk.choices[0].delta.content:
                yield chunk.choices[0].delta.content

第七章：多源信息融合与去重

7.1 内容去重与合并

from difflib import SequenceMatcher
from collections import defaultdict

class ContentFusion:
    """多源信息融合器"""
    
    def __init__(self, similarity_threshold: float = 0.7):
        self.similarity_threshold = similarity_threshold
    
    def deduplicate(self, results: List[SearchResult]) -> List[SearchResult]:
        """基于内容相似度去重"""
        unique_results = []
        
        for result in results:
            is_duplicate = False
            for existing in unique_results:
                similarity = self._content_similarity(
                    result.snippet, existing.snippet
                )
                if similarity > self.similarity_threshold:
                    is_duplicate = True
                    # 保留可信度更高的版本
                    if result.credibility_score > existing.credibility_score:
                        unique_results.remove(existing)
                        unique_results.append(result)
                    break
            
            if not is_duplicate:
                unique_results.append(result)
        
        return unique_results
    
    def _content_similarity(self, text1: str, text2: str) -> float:
        """计算两段文本的相似度"""
        return SequenceMatcher(None, text1, text2).ratio()
    
    def extract_key_facts(self, results: List[SearchResult], 
                          llm_client, query: str) -> dict:
        """从多个来源提取关键事实并融合"""
        sources_text = "\n\n".join([
            f"[来源{i+1}] {r.title}\n{r.snippet}" 
            for i, r in enumerate(results[:5])
        ])
        
        prompt = f"""从以下多个来源中提取关于"{query}"的关键事实。
        
对于每个事实：
1. 标注支持该事实的来源编号
2. 如果不同来源的信息冲突，请标注"存在争议"
3. 按重要性排序

来源信息：
{sources_text}

返回JSON格式：
{{
    "facts": [
        {{"fact": "事实描述", "sources": [1, 2], "confidence": "high/medium/low"}},
        ...
    ],
    "conflicts": [
        {{"topic": "争议点", "perspectives": [
            {{"view": "观点1", "sources": [1]}},
            {{"view": "观点2", "sources": [3]}}
        ]}}
    ]
}}"""
        
        response = llm_client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
            response_format={"type": "json_object"},
            temperature=0.1
        )
        
        return json.loads(response.choices[0].message.content)

第八章：搜索质量评估

8.1 评估指标体系

from dataclasses import dataclass
from typing import List, Dict
import numpy as np

@dataclass
class SearchEvaluationMetrics:
    """搜索质量评估指标"""
    
    # 检索质量
    mrr: float = 0.0          # Mean Reciprocal Rank
    ndcg: float = 0.0         # Normalized Discounted Cumulative Gain
    recall_at_k: float = 0.0  # Recall@K
    precision_at_k: float = 0.0  # Precision@K
    
    # 答案质量
    faithfulness: float = 0.0    # 答案忠实度（是否基于检索结果）
    relevance: float = 0.0       # 答案相关性
    completeness: float = 0.0    # 答案完整性
    citation_accuracy: float = 0.0  # 引用准确性

class SearchQualityEvaluator:
    """搜索质量评估器"""
    
    def __init__(self, llm_client):
        self.llm_client = llm_client
    
    def compute_mrr(self, ranked_results: List[List[str]], 
                     relevant_docs: List[List[str]]) -> float:
        """计算MRR（平均倒数排名）"""
        reciprocal_ranks = []
        
        for results, relevant in zip(ranked_results, relevant_docs):
            rr = 0
            for rank, doc in enumerate(results, 1):
                if doc in relevant:
                    rr = 1.0 / rank
                    break
            reciprocal_ranks.append(rr)
        
        return np.mean(reciprocal_ranks)
    
    def compute_ndcg(self, ranked_scores: List[float], 
                      ideal_scores: List[float], k: int = 10) -> float:
        """计算NDCG@K"""
        def dcg(scores, k):
            return sum(score / np.log2(i + 2) for i, score in enumerate(scores[:k]))
        
        actual_dcg = dcg(ranked_scores, k)
        ideal_dcg = dcg(sorted(ideal_scores, reverse=True), k)
        
        return actual_dcg / ideal_dcg if ideal_dcg > 0 else 0.0
    
    async def evaluate_answer_quality(self, query: str, answer: str,
                                       sources: List[SearchResult]) -> Dict[str, float]:
        """使用LLM评估答案质量"""
        sources_text = "\n".join([f"- {s.snippet}" for s in sources[:5]])
        
        prompt = f"""评估以下AI搜索答案的质量（0-1分）：

用户查询：{query}

检索来源：
{sources_text}

生成的答案：
{answer}

请评估以下维度（返回JSON）：
{{
    "faithfulness": 答案是否忠实于检索来源（不编造信息）,
    "relevance": 答案是否直接回答了用户问题,
    "completeness": 答案是否全面完整,
    "citation_accuracy": 引用标注是否准确
}}"""
        
        response = self.llm_client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
            response_format={"type": "json_object"},
            temperature=0.1
        )
        
        return json.loads(response.choices[0].message.content)

第九章：Perplexity-like搜索引擎实战

9.1 完整搜索引擎集成

现在，我们将前面所有模块整合为一个完整的AI搜索引擎：

import asyncio
from typing import Optional

class AISearchEngine:
    """
    AI原生搜索引擎主类
    整合所有模块，提供端到端的搜索体验
    """
    
    def __init__(self, config: dict):
        self.config = config
        
        # 初始化各模块
        self.query_engine = QueryUnderstandingEngine(
            api_key=config["openai_api_key"]
        )
        self.hybrid_search = HybridSearchEngine(
            vector_weight=config.get("vector_weight", 0.6),
            bm25_weight=config.get("bm25_weight", 0.4)
        )
        self.web_search = WebSearchEngine(
            serper_api_key=config.get("serper_api_key"),
            bing_api_key=config.get("bing_api_key")
        )
        self.content_extractor = WebContentExtractor()
        self.reranker = SemanticReranker()
        self.ranker = MultiSignalRanker(self.reranker)
        self.answer_generator = StreamingAnswerGenerator(
            api_key=config["openai_api_key"]
        )
        self.fusion = ContentFusion()
    
    async def search(self, query: str, 
                     chat_history: List[dict] = None) -> SearchAnswer:
        """
        执行完整的AI搜索流程
        """
        # Step 1: Query理解与改写
        understanding = await self.query_engine.understand(query, chat_history)
        search_queries = self.query_engine.generate_search_queries(understanding)
        
        print(f"[Query理解] 意图: {understanding.intent.value}")
        print(f"[Query改写] 生成了 {len(search_queries)} 个查询")
        
        # Step 2: 并行执行多路检索
        all_results = []
        
        # 网络实时搜索
        web_tasks = [
            self.web_search.search(q, num_results=5) 
            for q in search_queries[:3]
        ]
        web_results_lists = await asyncio.gather(*web_tasks, return_exceptions=True)
        
        for results in web_results_lists:
            if isinstance(results, list):
                all_results.extend(results)
        
        print(f"[检索] 获取到 {len(all_results)} 条原始结果")
        
        # Step 3: 去重
        unique_results = self.fusion.deduplicate(all_results)
        print(f"[去重] 剩余 {len(unique_results)} 条结果")
        
        # Step 4: 提取网页内容（并发）
        urls = [r.url for r in unique_results[:8]]
        contents = await self.content_extractor.batch_extract(urls)
        
        for result, content in zip(unique_results[:8], contents):
            if content:
                result.content = content
                result.snippet = content[:500]  # 用完整内容更新snippet
        
        # Step 5: 重排
        candidates = [
            {"content": r.snippet, "credibility_score": r.credibility_score,
             "published_date": r.published_date, "title": r.title, "url": r.url}
            for r in unique_results if r.snippet
        ]
        
        if candidates:
            ranked = self.ranker.rank(query, candidates)
        else:
            ranked = [(r.__dict__, 0.5) for r in unique_results[:5]]
        
        top_results = [
            SearchResult(
                title=doc.get("title", ""),
                url=doc.get("url", ""),
                snippet=doc.get("content", ""),
                credibility_score=doc.get("credibility_score", 0.5)
            )
            for doc, score in ranked[:8]
        ]
        
        print(f"[重排] 选出 {len(top_results)} 条最优结果")
        
        # Step 6: 生成答案
        answer = await self.answer_generator.generate(query, top_results)
        
        return answer
    
    async def search_stream(self, query: str,
                             chat_history: List[dict] = None):
        """流式搜索，逐步返回结果"""
        # 同样的流程，但答案生成部分使用流式输出
        understanding = await self.query_engine.understand(query, chat_history)
        search_queries = self.query_engine.generate_search_queries(understanding)
        
        # 检索
        all_results = []
        for q in search_queries[:2]:
            try:
                results = await self.web_search.search(q, num_results=5)
                all_results.extend(results)
            except Exception:
                pass
        
        unique_results = self.fusion.deduplicate(all_results)
        
        # 流式生成
        async for chunk in self.answer_generator.generate_stream(query, unique_results[:8]):
            yield chunk


# 使用示例
async def main():
    config = {
        "openai_api_key": "your-api-key",
        "serper_api_key": "your-serper-key",
        "vector_weight": 0.6,
        "bm25_weight": 0.4
    }
    
    engine = AISearchEngine(config)
    
    # 同步搜索
    answer = await engine.search("2024年AI领域最重要的突破有哪些？")
    print(f"\n答案：\n{answer.answer}")
    print(f"\n置信度：{answer.confidence}")
    print(f"\n引用：")
    for cite in answer.citations:
        print(f"  [{cite.index}] {cite.title} - {cite.url}")
    
    # 流式搜索
    print("\n--- 流式搜索 ---")
    async for chunk in engine.search_stream("Python异步编程最佳实践"):
        print(chunk, end="", flush=True)

if __name__ == "__main__":
    asyncio.run(main())

第十章：进阶优化与最佳实践

10.1 性能优化

# 1. 缓存层
import redis
import hashlib
import json

class SearchCache:
    """搜索结果缓存"""
    
    def __init__(self, redis_url: str = "redis://localhost:6379"):
        self.redis = redis.from_url(redis_url)
        self.default_ttl = 3600  # 1小时
    
    def _cache_key(self, query: str) -> str:
        return f"search:{hashlib.md5(query.encode()).hexdigest()}"
    
    async def get(self, query: str) -> Optional[dict]:
        key = self._cache_key(query)
        cached = self.redis.get(key)
        return json.loads(cached) if cached else None
    
    async def set(self, query: str, result: dict, ttl: int = None):
        key = self._cache_key(query)
        self.redis.setex(key, ttl or self.default_ttl, json.dumps(result))

# 2. 查询分类器：判断是否需要实时搜索
class QueryClassifier:
    """判断查询是否需要实时网络搜索"""
    
    STATIC_TOPICS = [  # 不需要实时搜索的静态知识
        "什么是", "定义", "历史", "原理", "教程"
    ]
    
    REALTIME_TOPICS = [  # 需要实时搜索的动态信息
        "最新", "今天", "新闻", "价格", "天气", "股价"
    ]
    
    @staticmethod
    def needs_web_search(query: str) -> bool:
        for keyword in QueryClassifier.REALTIME_TOPICS:
            if keyword in query:
                return True
        for keyword in QueryClassifier.STATIC_TOPICS:
            if keyword in query:
                return False
        return True  # 默认需要搜索

10.2 生产环境部署建议

1. 异步架构：使用FastAPI + asyncio处理并发请求
2. 向量数据库：大规模场景用Milvus/Qdrant替代FAISS
3. 缓存策略：热点查询缓存，相似查询去重
4. 监控告警：搜索延迟、答案质量、API调用量
5. A/B测试：对比不同检索策略和排序算法的效果
6. 用户反馈：收集点赞/踩数据，持续优化

总结

本教程系统讲解了AI原生搜索引擎的完整技术栈：

构建AI搜索引擎是一个系统工程，需要不断迭代优化。建议从核心检索+答案生成开始，逐步添加Query理解、重排、缓存等高级功能。

下一步学习建议：

• 尝试用RAGAS框架评估RAG系统质量
• 研究多模态搜索（图片、视频检索）
• 探索Agentic RAG（让Agent自主决定检索策略）
• 学习搜索引擎的在线学习与个性化

本教程内容约5000字，涵盖AI原生搜索引擎开发的核心技术与实战代码。希望对你的AI搜索项目有所帮助！