AI驱动的智能搜索引擎开发完全教程
从零到一构建生产级AI搜索引擎：语义理解、混合检索、智能排序与Perplexity风格搜索的完整实现
目录
- 教程简介
- 搜索引擎架构设计
- 语义搜索与向量检索
- 查询理解与意图识别
- 多源数据融合
- 搜索结果排序与重排(Reranking)
- 实时索引与增量更新
- 混合搜索策略：BM25+向量
- Perplexity风格AI搜索实现
- 搜索Agent设计
- 搜索质量评估
- 最佳实践与生产部署
- 总结
教程简介
传统搜索引擎依赖关键词匹配（如BM25算法），在处理语义相似但词汇不同的查询时表现欠佳。随着大语言模型（LLM）和向量检索技术的成熟，AI搜索引擎正在彻底改变信息检索的方式。
本教程将带你从零构建一个完整的AI搜索引擎系统，涵盖：
- 语义搜索：利用Embedding模型将文本转化为向量，实现语义级别的检索
- 混合检索：结合传统BM25与向量检索的优势，提升召回质量
- 查询理解：通过NLU技术理解用户真实意图，支持多轮对话式搜索
- 智能排序：使用Reranking模型对搜索结果进行精排
- AI生成式回答：参考Perplexity的设计理念，基于检索结果生成带引用的综合回答
- 搜索Agent：设计能够自主规划搜索策略的智能Agent
技术栈概览：
| 组件 | 技术选型 |
|---|---|
| Embedding模型 | text-embedding-3-small / BGE / GTE |
| 向量数据库 | Milvus / Qdrant / Chroma |
| 全文检索 | Elasticsearch / Meilisearch |
| Reranking | Cohere Rerank / BGE-Reranker / Cross-Encoder |
| LLM | OpenAI GPT-4 / Claude / 开源LLM |
| Agent框架 | LangChain / LlamaIndex / 自研 |
搜索引擎架构设计
整体架构
一个生产级AI搜索引擎通常包含以下核心层次：
┌─────────────────────────────────────────────────────┐
│ 用户交互层 │
│ Web UI / API / 对话式界面 │
├─────────────────────────────────────────────────────┤
│ 查询处理层 │
│ 查询理解 → 意图识别 → 查询改写 → 查询扩展 │
├─────────────────────────────────────────────────────┤
│ 检索层 │
│ BM25全文检索 ←→ 向量语义检索 ←→ 知识图谱检索 │
├─────────────────────────────────────────────────────┤
│ 排序与融合层 │
│ 结果融合 → Reranking → 去重 → 多样性控制 │
├─────────────────────────────────────────────────────┤
│ 生成与呈现层 │
│ LLM总结生成 → 引用标注 → 摘要展示 │
├─────────────────────────────────────────────────────┤
│ 数据与索引层 │
│ 文档解析 → Chunking → Embedding → 索引构建 │
└─────────────────────────────────────────────────────┘
核心模块设计
"""
AI搜索引擎核心架构 - 主入口模块
"""
from dataclasses import dataclass, field
from typing import List, Optional, Dict, Any
from enum import Enum
import asyncio
class SearchMode(Enum):
"""搜索模式枚举"""
KEYWORD = "keyword" # 纯关键词搜索
SEMANTIC = "semantic" # 纯语义搜索
HYBRID = "hybrid" # 混合搜索
AI_ANSWER = "ai_answer" # AI生成式回答
@dataclass
class SearchQuery:
"""搜索查询对象"""
text: str
mode: SearchMode = SearchMode.HYBRID
top_k: int = 10
filters: Dict[str, Any] = field(default_factory=dict)
session_id: Optional[str] = None
context: List[str] = field(default_factory=list) # 多轮对话上下文
@dataclass
class SearchDocument:
"""搜索文档对象"""
id: str
title: str
content: str
url: Optional[str] = None
source: Optional[str] = None
metadata: Dict[str, Any] = field(default_factory=dict)
score: float = 0.0
@dataclass
class SearchResponse:
"""搜索响应对象"""
query: str
rewritten_query: Optional[str] = None
intent: Optional[str] = None
documents: List[SearchDocument] = field(default_factory=list)
ai_answer: Optional[str] = None
citations: List[Dict[str, str]] = field(default_factory=list)
search_time_ms: float = 0.0
class AISearchEngine:
"""AI搜索引擎主类"""
def __init__(self, config: Dict[str, Any]):
self.config = config
self.query_processor = QueryProcessor(config)
self.retriever = HybridRetriever(config)
self.reranker = Reranker(config)
self.answer_generator = AnswerGenerator(config)
self.session_manager = SessionManager()
async def search(self, query: SearchQuery) -> SearchResponse:
"""执行搜索的主流程"""
import time
start = time.time()
# 1. 查询处理：理解意图、改写查询
processed = await self.query_processor.process(query)
# 2. 混合检索
documents = await self.retriever.retrieve(
query=processed.rewritten_query or query.text,
mode=query.mode,
top_k=query.top_k * 3, # 多召回，后续重排
filters=query.filters
)
# 3. 重排序
reranked = await self.reranker.rerank(
query=query.text,
documents=documents,
top_k=query.top_k
)
# 4. 生成式回答（如果是AI模式）
ai_answer = None
citations = []
if query.mode == SearchMode.AI_ANSWER:
ai_answer, citations = await self.answer_generator.generate(
query=query.text,
context_docs=reranked
)
elapsed_ms = (time.time() - start) * 1000
return SearchResponse(
query=query.text,
rewritten_query=processed.rewritten_query,
intent=processed.intent,
documents=reranked,
ai_answer=ai_answer,
citations=citations,
search_time_ms=elapsed_ms
)
语义搜索与向量检索
Embedding模型选型
语义搜索的核心是将文本转化为高维向量，使语义相似的文本在向量空间中距离更近。选择合适的Embedding模型至关重要。
主流Embedding模型对比：
| 模型 | 维度 | 特点 | 适用场景 |
|---|---|---|---|
| text-embedding-3-small | 1536 | OpenAI出品，性价比高 | 通用场景 |
| text-embedding-3-large | 3072 | 高精度，成本较高 | 高精度需求 |
| BGE-large-zh | 1024 | 中文优化，开源 | 中文场景 |
| GTE-large-zh | 1024 | 阿里出品，中文强 | 中文企业场景 |
| E5-mistral-7b | 4096 | 基于LLM的Embedding | 超高精度需求 |
向量检索实现
"""
语义搜索与向量检索模块
"""
import numpy as np
from typing import List, Tuple, Optional
import hashlib
class EmbeddingService:
"""Embedding服务封装"""
def __init__(self, model_name: str = "text-embedding-3-small",
api_key: Optional[str] = None):
self.model_name = model_name
self.api_key = api_key
self._cache = {} # 简单缓存
async def embed_text(self, text: str) -> List[float]:
"""单条文本Embedding"""
cache_key = hashlib.md5(text.encode()).hexdigest()
if cache_key in self._cache:
return self._cache[cache_key]
if "text-embedding" in self.model_name:
embedding = await self._openai_embed(text)
else:
embedding = await self._local_embed(text)
self._cache[cache_key] = embedding
return embedding
async def embed_batch(self, texts: List[str]) -> List[List[float]]:
"""批量Embedding"""
if "text-embedding" in self.model_name:
return await self._openai_embed_batch(texts)
else:
return [await self._local_embed(t) for t in texts]
async def _openai_embed(self, text: str) -> List[float]:
"""调用OpenAI Embedding API"""
from openai import AsyncOpenAI
client = AsyncOpenAI(api_key=self.api_key)
response = await client.embeddings.create(
model=self.model_name,
input=text
)
return response.data[0].embedding
async def _openai_embed_batch(self, texts: List[str]) -> List[List[float]]:
"""批量调用OpenAI Embedding API（支持分批）"""
from openai import AsyncOpenAI
client = AsyncOpenAI(api_key=self.api_key)
all_embeddings = []
batch_size = 100 # OpenAI限制每次最多处理的文本数
for i in range(0, len(texts), batch_size):
batch = texts[i:i + batch_size]
response = await client.embeddings.create(
model=self.model_name,
input=batch
)
all_embeddings.extend([d.embedding for d in response.data])
return all_embeddings
async def _local_embed(self, text: str) -> List[float]:
"""本地模型Embedding（以sentence-transformers为例）"""
from sentence_transformers import SentenceTransformer
if not hasattr(self, '_local_model'):
self._local_model = SentenceTransformer(self.model_name)
embedding = self._local_model.encode(text, normalize_embeddings=True)
return embedding.tolist()
class VectorStore:
"""向量存储抽象层 - 支持多种后端"""
def __init__(self, backend: str = "milvus", **kwargs):
self.backend = backend
self.config = kwargs
async def add_documents(self, ids: List[str],
vectors: List[List[float]],
metadatas: List[dict]):
"""添加文档到向量库"""
if self.backend == "milvus":
await self._milvus_insert(ids, vectors, metadatas)
elif self.backend == "qdrant":
await self._qdrant_insert(ids, vectors, metadatas)
elif self.backend == "chroma":
await self._chroma_insert(ids, vectors, metadatas)
async def search(self, query_vector: List[float],
top_k: int = 10,
filters: Optional[dict] = None) -> List[Tuple[str, float, dict]]:
"""向量相似度搜索，返回 (id, score, metadata)"""
if self.backend == "milvus":
return await self._milvus_search(query_vector, top_k, filters)
elif self.backend == "qdrant":
return await self._qdrant_search(query_vector, top_k, filters)
elif self.backend == "chroma":
return await self._chroma_search(query_vector, top_k, filters)
async def _milvus_insert(self, ids, vectors, metadatas):
"""Milvus插入实现"""
from pymilvus import Collection
collection = Collection(self.config.get("collection_name", "documents"))
entities = [
ids,
vectors,
[m.get("title", "") for m in metadatas],
[m.get("content", "") for m in metadatas],
]
collection.insert(entities)
collection.flush()
async def _milvus_search(self, query_vector, top_k, filters):
"""Milvus搜索实现"""
from pymilvus import Collection
collection = Collection(self.config.get("collection_name", "documents"))
collection.load()
search_params = {
"metric_type": "COSINE",
"params": {"nprobe": 16}
}
results = collection.search(
data=[query_vector],
anns_field="embedding",
param=search_params,
limit=top_k,
output_fields=["title", "content", "source"]
)
return [
(hit.id, hit.score, hit.entity)
for hit in results[0]
]
async def _qdrant_insert(self, ids, vectors, metadatas):
"""Qdrant插入实现"""
from qdrant_client import QdrantClient
from qdrant_client.models import PointStruct
client = QdrantClient(**self.config.get("connection", {}))
points = [
PointStruct(id=i, vector=v, payload=m)
for i, v, m in zip(ids, vectors, metadatas)
]
client.upsert(
collection_name=self.config.get("collection_name", "documents"),
points=points
)
async def _qdrant_search(self, query_vector, top_k, filters):
"""Qdrant搜索实现"""
from qdrant_client import QdrantClient
from qdrant_client.models import Filter, FieldCondition, MatchValue
client = QdrantClient(**self.config.get("connection", {}))
query_filter = None
if filters:
conditions = [
FieldCondition(key=k, match=MatchValue(value=v))
for k, v in filters.items()
]
query_filter = Filter(must=conditions)
results = client.search(
collection_name=self.config.get("collection_name", "documents"),
query_vector=query_vector,
limit=top_k,
query_filter=query_filter
)
return [(r.id, r.score, r.payload) for r in results]
async def _chroma_insert(self, ids, vectors, metadatas):
"""Chroma插入实现"""
import chromadb
client = chromadb.Client()
collection = client.get_or_create_collection(
name=self.config.get("collection_name", "documents")
)
collection.add(
ids=ids,
embeddings=vectors,
metadatas=metadatas,
documents=[m.get("content", "") for m in metadatas]
)
async def _chroma_search(self, query_vector, top_k, filters):
"""Chroma搜索实现"""
import chromadb
client = chromadb.Client()
collection = client.get_or_create_collection(
name=self.config.get("collection_name", "documents")
)
results = collection.query(
query_embeddings=[query_vector],
n_results=top_k,
where=filters
)
return list(zip(
results["ids"][0],
results["distances"][0],
results["metadatas"][0]
))
余弦相似度与近似最近邻(ANN)
向量检索的核心算法是近似最近邻搜索（ANN），常用算法包括：
- HNSW（Hierarchical Navigable Small World）：基于图结构的ANN算法，查询速度快，精度高
- IVF（Inverted File Index）：基于聚类的索引方式，适合大规模数据
- PQ（Product Quantization）：向量压缩技术，节省存储空间
"""
HNSW索引构建示例（使用hnswlib）
"""
import hnswlib
import numpy as np
def build_hnsw_index(vectors: np.ndarray, dim: int,
ef_construction: int = 200,
m: int = 16) -> hnswlib.Index:
"""
构建HNSW索引
参数:
vectors: 形状为 (n, dim) 的向量矩阵
dim: 向量维度
ef_construction: 构建时的搜索宽度，越大越精确但越慢
m: 每个节点的最大连接数
"""
index = hnswlib.Index(space='cosine', dim=dim)
# 初始化索引
index.init_index(
max_elements=len(vectors),
ef_construction=ef_construction,
m=m
)
# 添加向量
labels = np.arange(len(vectors))
index.add_items(vectors, labels)
# 设置查询时的搜索精度
index.set_ef(50)
return index
def search_hnsw(index: hnswlib.Index, query_vector: np.ndarray,
top_k: int = 10) -> tuple:
"""使用HNSW索引进行搜索"""
labels, distances = index.knn_query(query_vector, k=top_k)
return labels[0], distances[0]
查询理解与意图识别
意图识别分类
用户搜索意图通常分为以下几类：
- 信息查询（Informational）：用户想要了解某个知识或信息
- 导航查询（Navigational）：用户想要到达某个特定网站或页面
- 交易查询（Transactional）：用户想要完成某个操作或购买
- 本地查询（Local）：用户寻找本地服务或位置信息
"""
查询理解与意图识别模块
"""
from typing import List, Dict, Optional, Tuple
from dataclasses import dataclass
from enum import Enum
class QueryIntent(Enum):
INFORMATIONAL = "informational"
NAVIGATIONAL = "navigational"
TRANSACTIONAL = "transactional"
LOCAL = "local"
CONVERSATIONAL = "conversational"
@dataclass
class ProcessedQuery:
"""处理后的查询"""
original: str
rewritten_query: Optional[str] = None
intent: QueryIntent = QueryIntent.INFORMATIONAL
entities: List[Dict[str, str]] = None
keywords: List[str] = None
language: str = "zh"
confidence: float = 0.0
def __post_init__(self):
if self.entities is None:
self.entities = []
if self.keywords is None:
self.keywords = []
class QueryProcessor:
"""查询处理器"""
def __init__(self, config: dict):
self.config = config
self.llm_client = self._init_llm(config)
def _init_llm(self, config):
"""初始化LLM客户端"""
from openai import AsyncOpenAI
return AsyncOpenAI(api_key=config.get("api_key"))
async def process(self, query) -> ProcessedQuery:
"""
完整的查询处理流程:
1. 语言检测
2. 查询纠错
3. 意图识别
4. 实体抽取
5. 查询改写
"""
text = query.text
# 并行执行多个处理步骤
import asyncio
intent_task = asyncio.create_task(self._classify_intent(text))
entities_task = asyncio.create_task(self._extract_entities(text))
rewrite_task = asyncio.create_task(self._rewrite_query(text, query.context))
intent, intent_conf = await intent_task
entities = await entities_task
rewritten = await rewrite_task
# 提取关键词
keywords = self._extract_keywords(text)
return ProcessedQuery(
original=text,
rewritten_query=rewritten if rewritten != text else None,
intent=intent,
entities=entities,
keywords=keywords,
confidence=intent_conf
)
async def _classify_intent(self, query: str) -> Tuple[QueryIntent, float]:
"""使用LLM进行意图分类"""
prompt = f"""请分析以下搜索查询的用户意图，返回JSON格式:
查询: "{query}"
分类选项:
- informational: 用户想了解知识或信息
- navigational: 用户想找到特定网站或页面
- transactional: 用户想完成操作或购买
- local: 用户寻找本地服务
- conversational: 对话式查询，需要上下文
返回格式: {{"intent": "分类", "confidence": 0.0-1.0, "reason": "原因"}}"""
response = await self.llm_client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
temperature=0,
response_format={"type": "json_object"}
)
import json
result = json.loads(response.choices[0].message.content)
intent_map = {e.value: e for e in QueryIntent}
intent = intent_map.get(result["intent"], QueryIntent.INFORMATIONAL)
return intent, result.get("confidence", 0.5)
async def _extract_entities(self, query: str) -> List[Dict[str, str]]:
"""实体抽取"""
prompt = f"""从以下查询中提取实体，返回JSON数组:
查询: "{query}"
实体类型: 人名(PER)、地名(LOC)、组织(ORG)、产品(PROD)、时间(TIME)、技术(TECH)
返回格式: [{{"text": "实体文本", "type": "实体类型"}}]"""
response = await self.llm_client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
temperature=0,
response_format={"type": "json_object"}
)
import json
result = json.loads(response.choices[0].message.content)
return result.get("entities", [])
async def _rewrite_query(self, query: str,
context: List[str] = None) -> str:
"""
查询改写：
- 多轮对话上下文融合
- 同义词扩展
- 查询补全
"""
context_str = ""
if context:
context_str = "\n对话历史:\n" + "\n".join(
f"- {c}" for c in context[-5:] # 最近5轮
)
prompt = f"""请改写以下搜索查询，使其更适合搜索引擎检索。
要求:
1. 如果有对话历史，融合上下文信息
2. 补充必要的同义词或相关词
3. 保持查询的核心语义不变
4. 如果查询已经很清晰，直接返回原文
{context_str}
当前查询: "{query}"
只返回改写后的查询文本，不要解释。"""
response = await self.llm_client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
temperature=0,
max_tokens=200
)
return response.choices[0].message.content.strip().strip('"')
def _extract_keywords(self, query: str) -> List[str]:
"""基于规则的关键词提取（可替换为更高级的方法）"""
# 简单实现：按空格和标点分词
import re
# 移除常见停用词
stopwords = {"的", "了", "是", "在", "我", "有", "和", "就",
"不", "人", "都", "一", "一个", "上", "也", "很",
"到", "说", "要", "去", "你", "会", "着", "没有",
"看", "好", "自己", "这", "他", "吗", "什么", "怎么",
"the", "a", "an", "is", "are", "was", "were", "in",
"on", "at", "to", "for", "of", "with", "by"}
tokens = re.findall(r'[\w\u4e00-\u9fff]+', query.lower())
keywords = [t for t in tokens if t not in stopwords and len(t) > 1]
return keywords
多源数据融合
数据源接入
AI搜索引擎通常需要融合多种数据源，包括网页、文档、数据库、API等。
"""
多源数据融合模块
"""
from typing import List, Dict, Any, AsyncIterator
from dataclasses import dataclass
from abc import ABC, abstractmethod
import asyncio
@dataclass
class RawDocument:
"""原始文档"""
content: str
title: str = ""
url: str = ""
source_type: str = ""
metadata: Dict[str, Any] = None
def __post_init__(self):
if self.metadata is None:
self.metadata = {}
class DataSource(ABC):
"""数据源抽象基类"""
@abstractmethod
async def fetch(self, query: str, limit: int = 10) -> List[RawDocument]:
"""根据查询获取数据"""
pass
@abstractmethod
async def index_all(self) -> AsyncIterator[RawDocument]:
"""全量索引，返回文档流"""
pass
class WebDataSource(DataSource):
"""网页数据源"""
def __init__(self, api_key: str, search_engine_id: str):
self.api_key = api_key
self.search_engine_id = search_engine_id
async def fetch(self, query: str, limit: int = 10) -> List[RawDocument]:
"""通过Google Custom Search API获取网页"""
import aiohttp
url = "https://www.googleapis.com/customsearch/v1"
params = {
"key": self.api_key,
"cx": self.search_engine_id,
"q": query,
"num": min(limit, 10)
}
async with aiohttp.ClientSession() as session:
async with session.get(url, params=params) as resp:
data = await resp.json()
documents = []
for item in data.get("items", []):
doc = RawDocument(
content=item.get("snippet", ""),
title=item.get("title", ""),
url=item.get("link", ""),
source_type="web",
metadata={
"displayLink": item.get("displayLink", ""),
"pageMap": item.get("pagemap", {})
}
)
documents.append(doc)
return documents
async def index_all(self) -> AsyncIterator[RawDocument]:
"""网页数据源通常不支持全量索引"""
return
yield # Make it a generator
class DatabaseDataSource(DataSource):
"""数据库数据源"""
def __init__(self, connection_string: str, table: str,
content_columns: List[str],
title_column: str = "title"):
self.connection_string = connection_string
self.table = table
self.content_columns = content_columns
self.title_column = title_column
async def fetch(self, query: str, limit: int = 10) -> List[RawDocument]:
"""数据库全文搜索"""
import asyncpg
conn = await asyncpg.connect(self.connection_string)
# 构建全文搜索查询
search_expr = " || ' ' || ".join(
f"COALESCE({col}, '')" for col in self.content_columns
)
sql = f"""
SELECT {self.title_column}, {', '.join(self.content_columns)},
ts_rank(to_tsvector('simple', {search_expr}),
plainto_tsquery('simple', $1)) as rank
FROM {self.table}
WHERE to_tsvector('simple', {search_expr}) @@
plainto_tsquery('simple', $1)
ORDER BY rank DESC
LIMIT $2
"""
rows = await conn.fetch(sql, query, limit)
await conn.close()
return [
RawDocument(
content=" ".join(str(row[col]) for col in self.content_columns),
title=str(row[self.title_column]),
source_type="database",
metadata={"rank": row["rank"]}
)
for row in rows
]
async def index_all(self) -> AsyncIterator[RawDocument]:
"""全量导出数据库内容"""
import asyncpg
conn = await asyncpg.connect(self.connection_string)
sql = f"SELECT {self.title_column}, {', '.join(self.content_columns)} FROM {self.table}"
rows = await conn.fetch(sql)
await conn.close()
for row in rows:
yield RawDocument(
content=" ".join(str(row[col]) for col in self.content_columns),
title=str(row[self.title_column]),
source_type="database"
)
class DataFusionPipeline:
"""数据融合管线"""
def __init__(self, sources: List[DataSource]):
self.sources = sources
async def search_all(self, query: str,
limit_per_source: int = 10) -> List[RawDocument]:
"""并行搜索所有数据源"""
tasks = [
source.fetch(query, limit_per_source)
for source in self.sources
]
results = await asyncio.gather(*tasks, return_exceptions=True)
all_docs = []
for result in results:
if isinstance(result, list):
all_docs.extend(result)
elif isinstance(result, Exception):
print(f"数据源查询失败: {result}")
return all_docs
def deduplicate(self, documents: List[RawDocument]) -> List[RawDocument]:
"""基于内容相似度去重"""
seen = set()
unique = []
for doc in documents:
# 使用内容哈希作为去重键
content_hash = hash(doc.content[:200])
if content_hash not in seen:
seen.add(content_hash)
unique.append(doc)
return unique
def normalize(self, documents: List[RawDocument]) -> List[RawDocument]:
"""统一不同数据源的文档格式"""
for doc in documents:
# 清理HTML标签
if "<" in doc.content and ">" in doc.content:
import re
doc.content = re.sub(r'<[^>]+>', '', doc.content)
# 标准化空白字符
doc.content = " ".join(doc.content.split())
return documents
搜索结果排序与重排(Reranking)
排序架构
搜索排序通常分为两个阶段：
- 粗排（First-Stage Retrieval）：从海量文档中快速召回候选集
- 精排（Reranking）：使用更精确的模型对候选集进行重排序
"""
搜索结果排序与重排模块
"""
from typing import List, Dict, Tuple, Optional
import numpy as np
class BM25Scorer:
"""BM25评分器"""
def __init__(self, k1: float = 1.5, b: float = 0.75):
self.k1 = k1
self.b = b
self.doc_freqs = {} # 词 -> 出现该词的文档数
self.doc_lens = [] # 每篇文档的长度
self.avg_dl = 0 # 平均文档长度
self.n_docs = 0 # 文档总数
self.index = {} # 倒排索引: term -> [(doc_id, term_freq)]
def build_index(self, documents: List[str]):
"""构建BM25倒排索引"""
self.n_docs = len(documents)
self.doc_lens = []
self.index = {}
self.doc_freqs = {}
for doc_id, doc in enumerate(documents):
terms = self._tokenize(doc)
self.doc_lens.append(len(terms))
term_counts = {}
for term in terms:
term_counts[term] = term_counts.get(term, 0) + 1
for term, count in term_counts.items():
if term not in self.index:
self.index[term] = []
self.index[term].append((doc_id, count))
self.doc_freqs[term] = self.doc_freqs.get(term, 0) + 1
self.avg_dl = sum(self.doc_lens) / self.n_docs if self.n_docs > 0 else 0
def score(self, query: str, top_k: int = 10) -> List[Tuple[int, float]]:
"""计算BM25分数"""
query_terms = self._tokenize(query)
scores = np.zeros(self.n_docs)
for term in query_terms:
if term not in self.index:
continue
df = self.doc_freqs[term]
idf = np.log((self.n_docs - df + 0.5) / (df + 0.5) + 1)
for doc_id, tf in self.index[term]:
dl = self.doc_lens[doc_id]
tf_norm = (tf * (self.k1 + 1)) / (
tf + self.k1 * (1 - self.b + self.b * dl / self.avg_dl)
)
scores[doc_id] += idf * tf_norm
# 返回top_k
top_indices = np.argsort(scores)[::-1][:top_k]
return [(int(idx), float(scores[idx])) for idx in top_indices if scores[idx] > 0]
def _tokenize(self, text: str) -> List[str]:
"""中文分词（简化实现，生产环境建议使用jieba）"""
try:
import jieba
return list(jieba.cut(text))
except ImportError:
# 回退到简单字符级分词
import re
tokens = re.findall(r'[\u4e00-\u9fff]+|[a-zA-Z]+|\d+', text.lower())
return tokens
class CrossEncoderReranker:
"""基于Cross-Encoder的重排序器"""
def __init__(self, model_name: str = "BAAI/bge-reranker-v2-m3"):
self.model_name = model_name
self.model = None
def _load_model(self):
"""延迟加载模型"""
if self.model is None:
from sentence_transformers import CrossEncoder
self.model = CrossEncoder(self.model_name, max_length=512)
async def rerank(self, query: str, documents: List,
top_k: int = 10) -> List:
"""对文档进行重排序"""
self._load_model()
if not documents:
return []
# 构建 query-document 对
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)
# 返回top_k并更新分数
result = []
for doc, score in scored_docs[:top_k]:
doc.score = float(score)
result.append(doc)
return result
class CohereReranker:
"""使用Cohere Rerank API的重排序器"""
def __init__(self, api_key: str, model: str = "rerank-multilingual-v3.0"):
self.api_key = api_key
self.model = model
async def rerank(self, query: str, documents: List,
top_k: int = 10) -> List:
"""调用Cohere Rerank API"""
import cohere
co = cohere.Client(self.api_key)
doc_texts = [doc.content[:4000] for doc in documents] # Cohere限制
response = co.rerank(
query=query,
documents=doc_texts,
top_n=top_k,
model=self.model
)
result = []
for item in response.results:
doc = documents[item.index]
doc.score = item.relevance_score
result.append(doc)
return result
class HybridRanker:
"""混合排序器：融合多种排序信号"""
def __init__(self, weights: Dict[str, float] = None):
self.weights = weights or {
"bm25": 0.3,
"vector": 0.4,
"rerank": 0.3
}
def fuse_scores(self, score_maps: Dict[str, Dict[int, float]]) -> List[Tuple[int, float]]:
"""
融合多种排序信号
参数:
score_maps: {"bm25": {doc_id: score}, "vector": {doc_id: score}, ...}
"""
all_doc_ids = set()
for scores in score_maps.values():
all_doc_ids.update(scores.keys())
fused_scores = {}
for doc_id in all_doc_ids:
score = 0.0
for method, weight in self.weights.items():
if method in score_maps and doc_id in score_maps[method]:
# Min-Max归一化
method_scores = list(score_maps[method].values())
if method_scores:
min_s = min(method_scores)
max_s = max(method_scores)
raw = score_maps[method][doc_id]
normalized = (raw - min_s) / (max_s - min_s + 1e-10)
score += weight * normalized
fused_scores[doc_id] = score
ranked = sorted(fused_scores.items(), key=lambda x: x[1], reverse=True)
return ranked
@staticmethod
def reciprocal_rank_fusion(
rankings: List[List[int]], k: int = 60
) -> List[Tuple[int, float]]:
"""
Reciprocal Rank Fusion (RRF)算法
参数:
rankings: 多个排序列表，每个列表是按相关性排序的文档ID
k: 平滑参数，通常取60
"""
scores = {}
for ranking in rankings:
for rank, doc_id in enumerate(ranking):
scores[doc_id] = scores.get(doc_id, 0) + 1 / (k + rank + 1)
return sorted(scores.items(), key=lambda x: x[1], reverse=True)
实时索引与增量更新
索引管理
生产环境中，搜索引擎需要支持文档的实时增删改和索引的增量更新。
"""
实时索引与增量更新模块
"""
from typing import List, Dict, Optional
from datetime import datetime
import asyncio
import logging
logger = logging.getLogger(__name__)
class IndexManager:
"""索引管理器"""
def __init__(self, vector_store, embedding_service, config: dict):
self.vector_store = vector_store
self.embedding_service = embedding_service
self.config = config
self._pending_ops = [] # 待处理的索引操作
self._batch_size = config.get("batch_size", 100)
self._flush_interval = config.get("flush_interval_seconds", 5)
async def add_document(self, doc_id: str, title: str,
content: str, metadata: dict = None):
"""添加单个文档到索引"""
self._pending_ops.append({
"op": "add",
"id": doc_id,
"title": title,
"content": content,
"metadata": metadata or {},
"timestamp": datetime.utcnow().isoformat()
})
if len(self._pending_ops) >= self._batch_size:
await self.flush()
async def update_document(self, doc_id: str, title: str = None,
content: str = None, metadata: dict = None):
"""更新文档"""
self._pending_ops.append({
"op": "update",
"id": doc_id,
"title": title,
"content": content,
"metadata": metadata,
"timestamp": datetime.utcnow().isoformat()
})
if len(self._pending_ops) >= self._batch_size:
await self.flush()
async def delete_document(self, doc_id: str):
"""删除文档"""
self._pending_ops.append({
"op": "delete",
"id": doc_id,
"timestamp": datetime.utcnow().isoformat()
})
async def flush(self):
"""批量处理待处理的索引操作"""
if not self._pending_ops:
return
ops = self._pending_ops.copy()
self._pending_ops.clear()
add_ops = [op for op in ops if op["op"] == "add"]
update_ops = [op for op in ops if op["op"] == "update"]
delete_ops = [op for op in ops if op["op"] == "delete"]
# 批量添加
if add_ops:
await self._batch_add(add_ops)
# 批量更新（删除旧的 + 添加新的）
if update_ops:
await self._batch_update(update_ops)
# 批量删除
if delete_ops:
await self._batch_delete(delete_ops)
logger.info(f"索引刷新完成: 添加{len(add_ops)}, "
f"更新{len(update_ops)}, 删除{len(delete_ops)}")
async def _batch_add(self, ops: List[dict]):
"""批量添加文档"""
texts = [op["content"] for op in ops]
ids = [op["id"] for op in ops]
# 批量生成Embedding
embeddings = await self.embedding_service.embed_batch(texts)
metadatas = [
{
"title": op["title"],
"content": op["content"],
**op["metadata"],
"indexed_at": op["timestamp"]
}
for op in ops
]
await self.vector_store.add_documents(ids, embeddings, metadatas)
async def _batch_update(self, ops: List[dict]):
"""批量更新文档"""
# 先删除旧文档
ids = [op["id"] for op in ops]
await self.vector_store.delete(ids)
# 重新添加
add_ops = [
{
"id": op["id"],
"title": op.get("title", ""),
"content": op.get("content", ""),
"metadata": op.get("metadata", {}),
"timestamp": op["timestamp"]
}
for op in ops if op.get("content")
]
if add_ops:
await self._batch_add(add_ops)
async def _batch_delete(self, ops: List[dict]):
"""批量删除文档"""
ids = [op["id"] for op in ops]
await self.vector_store.delete(ids)
class IncrementalIndexer:
"""增量索引器 - 定期检查数据源变化并更新索引"""
def __init__(self, data_source, index_manager: IndexManager,
check_interval: int = 300):
self.data_source = data_source
self.index_manager = index_manager
self.check_interval = check_interval
self._running = False
async def start(self):
"""启动增量索引任务"""
self._running = True
logger.info(f"增量索引器启动，检查间隔: {self.check_interval}秒")
while self._running:
try:
await self._check_and_update()
except Exception as e:
logger.error(f"增量索引检查失败: {e}")
await asyncio.sleep(self.check_interval)
async def stop(self):
"""停止增量索引任务"""
self._running = False
async def _check_and_update(self):
"""检查数据源变化并更新索引"""
# 获取自上次检查以来变化的文档
changed_docs = await self.data_source.get_changed_documents()
for doc in changed_docs:
if doc.get("deleted"):
await self.index_manager.delete_document(doc["id"])
else:
await self.index_manager.add_document(
doc_id=doc["id"],
title=doc["title"],
content=doc["content"],
metadata=doc.get("metadata", {})
)
# 批量刷新
await self.index_manager.flush()
logger.info(f"增量更新完成: {len(changed_docs)} 个文档变化")
混合搜索策略：BM25+向量
混合检索实现
混合搜索结合了BM25的精确匹配能力和向量检索的语义理解能力，是当前最主流的搜索策略。
"""
混合搜索策略实现
"""
from typing import List, Dict, Any, Optional
import asyncio
class HybridRetriever:
"""混合检索器"""
def __init__(self, config: dict):
self.config = config
self.bm25_scorer = BM25Scorer(
k1=config.get("bm25_k1", 1.5),
b=config.get("bm25_b", 0.75)
)
self.vector_store = VectorStore(
backend=config.get("vector_backend", "milvus"),
**config.get("vector_config", {})
)
self.embedding_service = EmbeddingService(
model_name=config.get("embedding_model", "text-embedding-3-small"),
api_key=config.get("api_key")
)
self.ranker = HybridRanker(weights=config.get("fusion_weights", {
"bm25": 0.3,
"vector": 0.4,
"rerank": 0.3
}))
async def retrieve(self, query: str, mode: str = "hybrid",
top_k: int = 30,
filters: Optional[dict] = None) -> List:
"""
混合检索
参数:
query: 查询文本
mode: 检索模式 (keyword/semantic/hybrid)
top_k: 返回数量
filters: 过滤条件
"""
if mode == "keyword":
return await self._keyword_search(query, top_k)
elif mode == "semantic":
return await self._semantic_search(query, top_k, filters)
else:
return await self._hybrid_search(query, top_k, filters)
async def _keyword_search(self, query: str, top_k: int) -> List:
"""BM25关键词搜索"""
results = self.bm25_scorer.score(query, top_k)
return [SearchDocument(id=str(doc_id), title="", content="",
score=score) for doc_id, score in results]
async def _semantic_search(self, query: str, top_k: int,
filters: Optional[dict]) -> List:
"""向量语义搜索"""
query_embedding = await self.embedding_service.embed_text(query)
results = await self.vector_store.search(
query_vector=query_embedding,
top_k=top_k,
filters=filters
)
documents = []
for doc_id, score, metadata in results:
doc = SearchDocument(
id=str(doc_id),
title=metadata.get("title", ""),
content=metadata.get("content", ""),
url=metadata.get("url", ""),
source=metadata.get("source", ""),
score=score
)
documents.append(doc)
return documents
async def _hybrid_search(self, query: str, top_k: int,
filters: Optional[dict]) -> List:
"""混合搜索：并行执行BM25和向量搜索"""
# 并行执行两种搜索
keyword_task = asyncio.create_task(self._keyword_search(query, top_k))
semantic_task = asyncio.create_task(
self._semantic_search(query, top_k, filters)
)
keyword_results, semantic_results = await asyncio.gather(
keyword_task, semantic_task
)
# 使用RRF融合排序
keyword_ranking = [int(doc.id) for doc in keyword_results]
semantic_ranking = [int(doc.id) for doc in semantic_results]
fused = HybridRanker.reciprocal_rank_fusion(
[keyword_ranking, semantic_ranking], k=60
)
# 构建文档映射
doc_map = {}
for doc in keyword_results + semantic_results:
doc_map[doc.id] = doc
# 按融合分数排序
result = []
for doc_id, score in fused[:top_k]:
if str(doc_id) in doc_map:
doc = doc_map[str(doc_id)]
doc.score = score
result.append(doc)
return result
def index_document(self, doc_id: str, title: str, content: str,
metadata: dict = None):
"""将文档添加到BM25索引"""
# 这里简化处理，实际应该维护一个完整的文档存储
if not hasattr(self, '_documents'):
self._documents = {}
self._documents[doc_id] = {"title": title, "content": content,
"metadata": metadata or {}}
# 重建BM25索引
docs = [f"{d['title']} {d['content']}" for d in self._documents.values()]
self.bm25_scorer.build_index(docs)
Perplexity风格AI搜索实现
设计理念
Perplexity AI的核心创新在于将搜索引擎与大语言模型结合，不仅返回搜索结果，还生成带有引用来源的综合回答。本节我们将实现类似的功能。
"""
Perplexity风格AI搜索实现
"""
from typing import List, Dict, Tuple, Optional
from dataclasses import dataclass
import json
@dataclass
class Citation:
"""引用来源"""
index: int
title: str
url: str
snippet: str
class AnswerGenerator:
"""AI回答生成器"""
def __init__(self, config: dict):
self.config = config
from openai import AsyncOpenAI
self.client = AsyncOpenAI(api_key=config.get("api_key"))
self.model = config.get("answer_model", "gpt-4o")
async def generate(self, query: str,
context_docs: List) -> Tuple[str, List[Citation]]:
"""
基于检索文档生成回答
参数:
query: 用户查询
context_docs: 检索到的文档列表
返回:
(answer, citations): 回答文本和引用列表
"""
if not context_docs:
return "抱歉，未找到相关信息来回答您的问题。", []
# 构建引用上下文
citations = []
context_parts = []
for i, doc in enumerate(context_docs[:8], 1): # 最多使用8个文档
citation = Citation(
index=i,
title=doc.title or f"来源 {i}",
url=doc.url or "",
snippet=doc.content[:300]
)
citations.append(citation)
context_parts.append(
f"[{i}] {citation.title}\n{doc.content[:1000]}"
)
context_text = "\n\n".join(context_parts)
# 生成回答
system_prompt = """你是一个专业的AI搜索助手。请基于提供的参考资料回答用户的问题。
要求:
1. 回答必须基于提供的参考资料，不要编造信息
2. 使用引用标注，格式为 [1] [2] 等，对应参考资料的编号
3. 回答应该全面、准确、有条理
4. 如果参考资料不足以完整回答问题，明确指出信息不足的部分
5. 保持客观中立的语调
6. 使用Markdown格式化回答"""
user_prompt = f"""参考资料:
{context_text}
用户问题: {query}
请基于以上参考资料回答问题，并在相关内容后标注引用来源。"""
response = await self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
],
temperature=0.3,
max_tokens=2000,
stream=False
)
answer = response.choices[0].message.content
return answer, citations
async def generate_stream(self, query: str,
context_docs: List):
"""
流式生成回答（适用于实时展示）
返回: 异步生成器，逐token输出
"""
citations = []
context_parts = []
for i, doc in enumerate(context_docs[:8], 1):
citation = Citation(
index=i,
title=doc.title or f"来源 {i}",
url=doc.url or "",
snippet=doc.content[:300]
)
citations.append(citation)
context_parts.append(f"[{i}] {citation.title}\n{doc.content[:1000]}")
context_text = "\n\n".join(context_parts)
system_prompt = "你是一个专业的AI搜索助手。基于参考资料回答问题，使用[1][2]等标注引用。"
user_prompt = f"参考资料:\n{context_text}\n\n问题: {query}"
stream = await self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
],
temperature=0.3,
max_tokens=2000,
stream=True
)
async for chunk in stream:
if chunk.choices[0].delta.content:
yield chunk.choices[0].delta.content
# 最后yield引用信息
yield "\n\n---CITATIONS---\n"
yield json.dumps([
{"index": c.index, "title": c.title, "url": c.url}
for c in citations
], ensure_ascii=False)
class FollowUpGenerator:
"""后续问题建议生成器"""
def __init__(self, client):
self.client = client
async def suggest_followups(self, query: str, answer: str,
citations: List[Citation]) -> List[str]:
"""基于当前问答生成后续问题建议"""
prompt = f"""基于以下问答对，生成3个用户可能感兴趣的后续问题。
要求: 问题应该自然、具体、有探索性。
原始问题: {query}
回答摘要: {answer[:500]}
返回JSON格式: {{"questions": ["问题1", "问题2", "问题3"]}}"""
response = await self.client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
temperature=0.7,
response_format={"type": "json_object"}
)
result = json.loads(response.choices[0].message.content)
return result.get("questions", [])[:3]
搜索Agent设计
Agent架构
搜索Agent能够自主规划搜索策略，处理复杂查询，并进行多步推理。
"""
搜索Agent设计
"""
from typing import List, Dict, Any, Optional, Callable
from dataclasses import dataclass, field
from enum import Enum
import json
class AgentAction(Enum):
SEARCH = "search"
SUMMARIZE = "summarize"
COMPARE = "compare"
ASK_CLARIFICATION = "ask_clarification"
FINAL_ANSWER = "final_answer"
@dataclass
class AgentStep:
"""Agent执行步骤"""
action: AgentAction
thought: str
parameters: Dict[str, Any] = field(default_factory=dict)
result: Any = None
@dataclass
class AgentPlan:
"""Agent执行计划"""
goal: str
steps: List[AgentStep] = field(default_factory=list)
current_step: int = 0
class SearchAgent:
"""搜索Agent - 能够自主规划和执行多步搜索"""
def __init__(self, search_engine, llm_client, config: dict = None):
self.search_engine = search_engine
self.llm_client = llm_client
self.config = config or {}
self.max_steps = self.config.get("max_steps", 5)
self.tools = self._build_tools()
def _build_tools(self) -> Dict[str, Callable]:
"""构建Agent可用工具"""
return {
"search": self._tool_search,
"search_with_filter": self._tool_search_filtered,
"summarize": self._tool_summarize,
"extract_info": self._tool_extract_info,
}
async def run(self, query: str, session_id: str = None) -> Dict[str, Any]:
"""
运行搜索Agent
Agent会自主规划搜索策略，执行多步操作，最终给出综合回答
"""
steps = []
context = []
final_answer = None
for step_num in range(self.max_steps):
# 1. 规划下一步动作
plan = await self._plan_next_step(query, context, steps)
if plan["action"] == "final_answer":
final_answer = plan.get("answer", "")
break
# 2. 执行动作
step = AgentStep(
action=AgentAction(plan["action"]),
thought=plan.get("thought", ""),
parameters=plan.get("parameters", {})
)
# 调用对应工具
tool_name = plan["action"]
if tool_name in self.tools:
step.result = await self.tools[tool_name](**plan.get("parameters", {}))
steps.append(step)
context.append({
"step": step_num + 1,
"action": plan["action"],
"thought": plan.get("thought", ""),
"result_summary": self._summarize_result(step.result)
})
# 3. 如果没有生成最终回答，基于所有步骤的结果生成
if not final_answer:
final_answer = await self._generate_final_answer(query, steps)
return {
"query": query,
"answer": final_answer,
"steps": [
{
"action": s.action.value,
"thought": s.thought,
"result": self._summarize_result(s.result)
}
for s in steps
],
"total_steps": len(steps)
}
async def _plan_next_step(self, query: str, context: List,
steps: List[AgentStep]) -> Dict:
"""规划下一步动作"""
context_str = json.dumps(context, ensure_ascii=False, indent=2)
prompt = f"""你是一个搜索Agent，负责规划搜索策略来回答用户的问题。
用户问题: {query}
已执行的步骤:
{context_str if context else "无"}
可用工具:
1. search - 搜索相关信息。参数: {{"query": "搜索词"}}
2. search_with_filter - 带过滤条件的搜索。参数: {{"query": "搜索词", "filters": {{"source": "xxx"}}}}
3. summarize - 总结已有信息。参数: {{"text": "要总结的文本"}}
4. extract_info - 从文本中提取特定信息。参数: {{"text": "文本", "info_type": "要提取的信息类型"}}
5. final_answer - 生成最终回答。参数: {{"answer": "最终回答"}}
请决定下一步动作。返回JSON格式:
{{"action": "工具名", "thought": "思考过程", "parameters": {{参数}}}}"""
response = await self.llm_client.chat.completions.create(
model=self.config.get("agent_model", "gpt-4o"),
messages=[{"role": "user", "content": prompt}],
temperature=0,
response_format={"type": "json_object"}
)
return json.loads(response.choices[0].message.content)
async def _tool_search(self, query: str) -> List[Dict]:
"""搜索工具"""
from dataclasses import asdict
result = await self.search_engine.search(
SearchQuery(text=query, mode=SearchMode.HYBRID, top_k=5)
)
return [
{
"title": doc.title,
"content": doc.content[:500],
"url": doc.url,
"score": doc.score
}
for doc in result.documents
]
async def _tool_search_filtered(self, query: str,
filters: dict = None) -> List[Dict]:
"""带过滤的搜索工具"""
result = await self.search_engine.search(
SearchQuery(text=query, mode=SearchMode.HYBRID,
top_k=5, filters=filters or {})
)
return [
{
"title": doc.title,
"content": doc.content[:500],
"url": doc.url,
"score": doc.score
}
for doc in result.documents
]
async def _tool_summarize(self, text: str) -> str:
"""总结工具"""
response = await self.llm_client.chat.completions.create(
model="gpt-4o-mini",
messages=[{
"role": "user",
"content": f"请总结以下内容的关键信息:\n\n{text[:3000]}"
}],
temperature=0,
max_tokens=500
)
return response.choices[0].message.content
async def _tool_extract_info(self, text: str, info_type: str) -> str:
"""信息提取工具"""
response = await self.llm_client.chat.completions.create(
model="gpt-4o-mini",
messages=[{
"role": "user",
"content": f"从以下文本中提取{info_type}:\n\n{text[:3000]}"
}],
temperature=0,
max_tokens=500
)
return response.choices[0].message.content
async def _generate_final_answer(self, query: str,
steps: List[AgentStep]) -> str:
"""基于所有步骤的结果生成最终回答"""
results_text = "\n\n".join([
f"步骤{i+1} ({s.action.value}):\n{self._summarize_result(s.result)}"
for i, s in enumerate(steps)
if s.result
])
response = await self.llm_client.chat.completions.create(
model=self.config.get("answer_model", "gpt-4o"),
messages=[
{
"role": "system",
"content": "你是一个专业的AI搜索助手，基于搜索结果给出准确、全面的回答。"
},
{
"role": "user",
"content": f"用户问题: {query}\n\n搜索结果:\n{results_text}\n\n请综合以上信息回答用户的问题。"
}
],
temperature=0.3
)
return response.choices[0].message.content
def _summarize_result(self, result: Any) -> str:
"""将结果转为简短摘要"""
if result is None:
return "无结果"
if isinstance(result, str):
return result[:300]
if isinstance(result, list):
return json.dumps(result[:3], ensure_ascii=False)[:500]
return str(result)[:300]
搜索质量评估
评估指标体系
搜索质量评估是持续优化搜索引擎的关键。主要评估指标包括：
"""
搜索质量评估模块
"""
from typing import List, Dict, Set, Tuple
import numpy as np
from dataclasses import dataclass
@dataclass
class EvaluationResult:
"""评估结果"""
precision_at_k: float
recall_at_k: float
mrr: float # Mean Reciprocal Rank
ndcg: float # Normalized Discounted Cumulative Gain
map_score: float # Mean Average Precision
hit_rate: float
class SearchEvaluator:
"""搜索质量评估器"""
def __init__(self):
pass
def evaluate(self, predictions: List[List[str]],
ground_truth: List[Set[str]],
k: int = 10) -> EvaluationResult:
"""
综合评估搜索质量
参数:
predictions: 每个查询的预测结果列表 (按相关性排序的文档ID)
ground_truth: 每个查询的相关文档集合
k: 评估的top-k
"""
precisions = []
recalls = []
mrrs = []
ndcgs = []
aps = []
hits = []
for pred, gt in zip(predictions, ground_truth):
pred_k = pred[:k]
precisions.append(self._precision_at_k(pred_k, gt))
recalls.append(self._recall_at_k(pred_k, gt))
mrrs.append(self._mrr(pred, gt))
ndcgs.append(self._ndcg(pred_k, gt, k))
aps.append(self._average_precision(pred, gt))
hits.append(1.0 if any(p in gt for p in pred_k) else 0.0)
return EvaluationResult(
precision_at_k=np.mean(precisions),
recall_at_k=np.mean(recalls),
mrr=np.mean(mrrs),
ndcg=np.mean(ndcgs),
map_score=np.mean(aps),
hit_rate=np.mean(hits)
)
def _precision_at_k(self, predicted: List[str],
relevant: Set[str]) -> float:
"""P@K: top-K结果中相关文档的比例"""
if not predicted:
return 0.0
hits = sum(1 for p in predicted if p in relevant)
return hits / len(predicted)
def _recall_at_k(self, predicted: List[str],
relevant: Set[str]) -> float:
"""R@K: 相关文档中被检索到的比例"""
if not relevant:
return 0.0
hits = sum(1 for p in predicted if p in relevant)
return hits / len(relevant)
def _mrr(self, predicted: List[str], relevant: Set[str]) -> float:
"""MRR: 第一个相关结果的排名倒数"""
for i, p in enumerate(predicted):
if p in relevant:
return 1.0 / (i + 1)
return 0.0
def _ndcg(self, predicted: List[str], relevant: Set[str],
k: int) -> float:
"""NDCG@K: 归一化折损累积增益"""
dcg = 0.0
for i, p in enumerate(predicted[:k]):
rel = 1.0 if p in relevant else 0.0
dcg += rel / np.log2(i + 2) # i+2 因为log2(1)=0
# 理想排序的DCG
ideal_rels = sorted(
[1.0 if p in relevant else 0.0 for p in predicted[:k]],
reverse=True
)
idcg = sum(
rel / np.log2(i + 2)
for i, rel in enumerate(ideal_rels)
)
return dcg / idcg if idcg > 0 else 0.0
def _average_precision(self, predicted: List[str],
relevant: Set[str]) -> float:
"""AP: 平均精度"""
if not relevant:
return 0.0
hits = 0
sum_precision = 0.0
for i, p in enumerate(predicted):
if p in relevant:
hits += 1
sum_precision += hits / (i + 1)
return sum_precision / len(relevant)
class ABTestEvaluator:
"""A/B测试评估器"""
def __init__(self):
self.results_a = []
self.results_b = []
def record_interaction(self, group: str, query: str,
clicked_rank: int = None,
dwell_time: float = None,
satisfied: bool = False):
"""记录用户交互"""
record = {
"query": query,
"clicked_rank": clicked_rank,
"dwell_time": dwell_time,
"satisfied": satisfied
}
if group == "a":
self.results_a.append(record)
else:
self.results_b.append(record)
def analyze(self) -> Dict:
"""分析A/B测试结果"""
def compute_metrics(results):
if not results:
return {}
clicks = [r["clicked_rank"] for r in results if r["clicked_rank"]]
return {
"total_queries": len(results),
"click_through_rate": len(clicks) / len(results),
"average_click_position": np.mean(clicks) if clicks else 0,
"satisfaction_rate": sum(
1 for r in results if r["satisfied"]
) / len(results),
"average_dwell_time": np.mean(
[r["dwell_time"] for r in results if r["dwell_time"]]
) if any(r["dwell_time"] for r in results) else 0
}
metrics_a = compute_metrics(self.results_a)
metrics_b = compute_metrics(self.results_b)
return {
"group_a": metrics_a,
"group_b": metrics_b,
"comparison": {
"ctr_lift": (
(metrics_b.get("click_through_rate", 0) -
metrics_a.get("click_through_rate", 0))
/ max(metrics_a.get("click_through_rate", 1e-10), 1e-10)
),
"satisfaction_lift": (
(metrics_b.get("satisfaction_rate", 0) -
metrics_a.get("satisfaction_rate", 0))
/ max(metrics_a.get("satisfaction_rate", 1e-10), 1e-10)
)
}
}
class RelevanceLabeler:
"""相关性标注工具（用于构建评估数据集）"""
def __init__(self):
self.annotations = []
def add_annotation(self, query: str, doc_id: str,
relevance: int, annotator: str = ""):
"""
添加相关性标注
relevance: 0-3的评分
0: 完全不相关
1: 略微相关
2: 部分相关
3: 高度相关
"""
self.annotations.append({
"query": query,
"doc_id": doc_id,
"relevance": relevance,
"annotator": annotator
})
def get_ground_truth(self, query: str,
threshold: int = 1) -> Set[str]:
"""获取某个查询的相关文档集合"""
return {
a["doc_id"]
for a in self.annotations
if a["query"] == query and a["relevance"] >= threshold
}
def export_dataset(self) -> List[Dict]:
"""导出评估数据集"""
return self.annotations
最佳实践与生产部署
性能优化建议
- 缓存策略：对热门查询结果进行缓存，减少重复计算
- 异步并发：充分利用异步IO，提高并发处理能力
- 批量处理：Embedding和Reranking操作尽量批量执行
- 索引分片：大规模数据集采用分布式索引
- 降级策略：当某个检索通道不可用时，自动降级到其他通道
"""
生产部署配置示例
"""
import os
PRODUCTION_CONFIG = {
# Embedding配置
"embedding": {
"model": "text-embedding-3-small",
"api_key": os.getenv("OPENAI_API_KEY"),
"batch_size": 100,
"cache_ttl": 3600, # 缓存1小时
},
# 向量数据库配置
"vector_db": {
"backend": "milvus",
"host": os.getenv("MILVUS_HOST", "localhost"),
"port": int(os.getenv("MILVUS_PORT", "19530")),
"collection_name": "search_documents",
"index_type": "HNSW",
"metric_type": "COSINE",
"index_params": {
"M": 16,
"efConstruction": 256
},
"search_params": {
"ef": 128
}
},
# BM25配置
"bm25": {
"k1": 1.2,
"b": 0.75,
"backend": "elasticsearch",
"es_url": os.getenv("ES_URL", "http://localhost:9200"),
"index_name": "search_docs"
},
# Reranking配置
"reranker": {
"model": "BAAI/bge-reranker-v2-m3",
"top_k": 10,
"batch_size": 32
},
# LLM配置
"llm": {
"model": "gpt-4o",
"temperature": 0.3,
"max_tokens": 2000
},
# 搜索Agent配置
"agent": {
"max_steps": 5,
"agent_model": "gpt-4o",
"answer_model": "gpt-4o"
},
# 性能配置
"performance": {
"search_timeout_ms": 3000,
"answer_timeout_ms": 15000,
"cache_enabled": True,
"cache_size": 10000,
"max_concurrent_searches": 100
}
}
部署架构
┌──────────────┐
│ Load Balancer│
└──────┬───────┘
│
┌────────────┼────────────┐
│ │ │
┌─────┴─────┐ ┌───┴────┐ ┌────┴─────┐
│ Search API │ │Search │ │ Search │
│ Server 1 │ │Server 2│ │ Server 3 │
└─────┬─────┘ └───┬────┘ └────┬─────┘
│ │ │
┌─────┴────────────┴────────────┴─────┐
│ Service Layer │
│ ┌─────┐ ┌──────┐ ┌───────┐ │
│ │Query│ │Retri-│ │Rerank │ │
│ │Proc │ │eval │ │Service│ │
│ └─────┘ └──────┘ └───────┘ │
│ ┌─────┐ ┌──────┐ ┌───────┐ │
│ │LLM │ │Agent │ │Answer │ │
│ │Pool │ │Engine│ │Gen │ │
│ └─────┘ └──────┘ └───────┘ │
└──────────────┬─────────────────────┘
│
┌──────────────┼─────────────────────┐
│ Data Layer │
│ ┌──────┐ ┌──────┐ ┌──────────┐ │
│ │Milvus│ │ES │ │Redis │ │
│ │Cluster│ │Cluster│ │Cache │ │
│ └──────┘ └──────┘ └──────────┘ │
└────────────────────────────────────┘
监控与告警
"""
搜索引擎监控模块
"""
import time
from typing import Dict, List
from dataclasses import dataclass, field
from collections import defaultdict
@dataclass
class SearchMetrics:
"""搜索指标"""
total_queries: int = 0
total_latency_ms: float = 0
cache_hits: int = 0
cache_misses: int = 0
errors: int = 0
latency_buckets: Dict[str, int] = field(
default_factory=lambda: {
"<100ms": 0, "100-500ms": 0,
"500ms-1s": 0, "1-3s": 0, ">3s": 0
}
)
class SearchMonitor:
"""搜索引擎监控器"""
def __init__(self):
self.metrics = SearchMetrics()
self.query_logs: List[Dict] = []
self.slow_queries: List[Dict] = []
def record_search(self, query: str, latency_ms: float,
results_count: int, from_cache: bool = False,
error: str = None):
"""记录一次搜索"""
self.metrics.total_queries += 1
self.metrics.total_latency_ms += latency_ms
if from_cache:
self.metrics.cache_hits += 1
else:
self.metrics.cache_misses += 1
if error:
self.metrics.errors += 1
# 延迟分布
if latency_ms < 100:
self.metrics.latency_buckets["<100ms"] += 1
elif latency_ms < 500:
self.metrics.latency_buckets["100-500ms"] += 1
elif latency_ms < 1000:
self.metrics.latency_buckets["500ms-1s"] += 1
elif latency_ms < 3000:
self.metrics.latency_buckets["1-3s"] += 1
else:
self.metrics.latency_buckets[">3s"] += 1
# 记录慢查询
if latency_ms > 1000:
self.slow_queries.append({
"query": query,
"latency_ms": latency_ms,
"results_count": results_count,
"error": error
})
def get_summary(self) -> Dict:
"""获取监控摘要"""
total = self.metrics.total_queries
if total == 0:
return {"message": "暂无查询数据"}
return {
"total_queries": total,
"average_latency_ms": self.metrics.total_latency_ms / total,
"cache_hit_rate": self.metrics.cache_hits / total,
"error_rate": self.metrics.errors / total,
"latency_distribution": self.metrics.latency_buckets,
"slow_queries_count": len(self.slow_queries),
"recent_slow_queries": self.slow_queries[-5:]
}
总结
本教程全面讲解了AI驱动的智能搜索引擎的开发，涵盖了从架构设计到生产部署的完整链路。以下是核心要点回顾：
关键技术总结
| 技术模块 | 核心要点 |
|---|---|
| 语义搜索 | Embedding模型选型、向量检索(ANN)、索引构建 |
| 查询理解 | 意图识别、实体抽取、查询改写与扩展 |
| 混合检索 | BM25+向量检索、RRF融合排序 |
| Reranking | Cross-Encoder精排、多信号融合 |
| AI回答 | 引用生成、流式输出、后续问题建议 |
| 搜索Agent | 自主规划、多步推理、工具调用 |
| 质量评估 | P@K、NDCG、MRR、A/B测试 |
进阶方向
- 多模态搜索：支持图片、视频等非文本内容的搜索
- 个性化搜索：基于用户画像和历史行为进行个性化排序
- 知识图谱增强：利用知识图谱提升实体理解和关系推理
- 实时学习：基于用户反馈持续优化搜索质量
- 跨语言搜索：支持多语言查询和文档的统一检索
推荐学习资源
- 向量数据库实践：Milvus、Qdrant、Weaviate官方文档
- Embedding模型：Sentence-Transformers、MTEB排行榜
- 搜索框架：LlamaIndex、Haystack、LangChain
- 学术论文：DPR、ColBERT、SPLADE、HyDE
本教程由AI搜索引擎技术实践编写，代码示例仅供学习参考，实际生产环境需根据具体需求进行调整和优化。
