RAG：开发者智能文档搜索

检索增强生成：企业级智能文档搜索新范式

RAG 将向量搜索与生成式 AI 相结合，从私有文档库中精准提取答案。系统可从 PDF、Excel 文件、代码等多种格式中检索相关内容，并基于原始资料生成解释——显著降低幻觉问题。

流程始于预处理阶段：文档被解析、按内容类型智能分块，并转换为嵌入向量存储在向量数据库中。

多格式文件文本提取

系统支持多种格式的无损解析，每种格式均配备专用提取器。

SUPPORTED = {
    '.pdf':  lambda f: ''.join(p.extract_text() or '' for p in PyPDF2.PdfReader(f).pages),
    '.docx': lambda f: '\n'.join(p.text for p in Document(f).paragraphs),
    '.pptx': lambda f: '\n'.join(
        shape.text
        for slide in Presentation(f).slides
        for shape in slide.shapes
        if hasattr(shape, "text") and shape.text.strip()
    ),
    '.xlsx': extract_xlsx,
    '.csv':  extract_csv,
    # .txt, .json, .py, .js, etc.
}

输出为干净、可直接处理的文本。

按内容类型智能分块

文本不按固定大小切割，而是语义化分割。首先识别内容类型：

def detect_content_type(text):
    if re.search(r'<table|</td>|<tr>', text, re.IGNORECASE):
        return 'table'
    tab_lines = sum(1 for line in text.splitlines() if '\t' in line)
    if tab_lines > max(3, len(text.splitlines()) * 0.4):
        return 'table'
    if re.search(r'function\s+\w+\s*\(|\bdef\s+\w+\s*\(|\bclass\s+\w+\b', text):
        return 'code'
    return 'prose'

• 表格：每个分块保留重复的表头
• 代码：按函数/类边界切分，不破坏逻辑结构
• 正文：使用句子嵌入实现语义边界划分

针对表格的处理如下：

def split_table_by_rows(text, chunk_size):
    lines = [l for l in text.splitlines() if l.strip()]
    header = lines[0]
    chunks = []
    cur_lines = [header]
    for line in lines[1:]:
        if sum(len(l) for l in cur_lines) + len(line) >= chunk_size:
            chunks.append('\n'.join(cur_lines))
            cur_lines = [header, line]
        else:
            cur_lines.append(line)
    if cur_lines:
        chunks.append('\n'.join(cur_lines))
    return chunks

语义分块通过相邻句子间的余弦相似度计算，采用自适应阈值（第25百分位）。

嵌入向量生成

使用 text-embedding-3-small 等模型（1536维）对分块进行向量化，向量与元数据一同存入数据库。

def get_embeddings(texts, batch_size=100):
    embeddings = []
    for i in range(0, len(texts), batch_size):
        resp = requests.post(
            "https://openrouter.ai/api/v1/embeddings",
            headers=HEADERS,
            json={
                "model": "openai/text-embedding-3-small",
                "input": texts[i:i + batch_size],
                "encoding_format": "float"
            },
        ).json()
        embeddings.extend(d["embedding"] for d in resp.get("data", []))
    return np.array(embeddings)

多阶段搜索机制

查询被转化为向量后，经大语言模型改写为三种变体，再通过混合搜索进行处理。

混合搜索：向量 + BM25

• 向量搜索：在嵌入空间中计算余弦相似度
• BM25：基于 TF-IDF，k1=1.5，b=0.75，擅长精确匹配

结果通过 RRF（Reciprocal Rank Fusion）融合：

# RRF 融合
for rank, idx in enumerate(sorted_by_vector):
    rrf_scores[idx] += 1.0 / (k_rrf + rank)
for rank, idx in enumerate(sorted_by_bm25):
    rrf_scores[idx] += bm25_normalized[idx] * (1.0 / (k_rrf + rank))

重排序与过滤规则

前 50 个候选结果由大语言模型评估，最终选出 5–10 个最优项。强制包含规则如下：

• 引号中的内容（如「...」或 "..."）
• 数字或编号（正则表达式：\b(№|N|number)\s*\d+\b）

quote_matches = re.findall(r'«([^»]+)»|"([^"+])"', query)
for match in quote_matches:
    quote = (match[0] or match[1]).strip()
    if len(quote) > 5:
        for i, chunk in enumerate(chunks_data):
            if quote in chunk['content']:
                forced_indices.append(i)
                break

性能优化策略

分层搜索机制

先通过文档摘要识别顶层相关文档；再在这些文档内部进行分块搜索。

doc_summaries = [chunks_data_by_doc[doc_id][0]['content'][:500] for doc_id in all_doc_ids]
doc_embs = get_embeddings(doc_summaries)
sims = np.dot(doc_embs_normalized, query_embedding_normalized)
top_doc_ids = [all_doc_ids[i] for i in np.argsort(sims)[::-1][:5]]
chunks_data = [ch for ch in chunks_data if ch['doc_id'] in top_doc_ids]

迭代式搜索

对于复杂查询，系统通过多轮迭代，利用大语言模型反馈持续优化结果。

核心优势总结

• RAG 通过严格基于相关分块生成答案，大幅减少幻觉现象
• 混合搜索（向量 + BM25 + RRF）相比纯向量搜索准确率提升 20–30%
• 语义分块有效保留表格与代码的上下文信息，降低噪声干扰
• 分层策略显著加速大规模文档库（>1万篇）的检索效率
• 强制过滤规则确保包含数字或引号内容的事实被完整召回

— Editorial Team