11b - RAG 流水线详解

简单文本（MD/TXT/HTML）──▶ 直接读取或轻量解析
结构化 PDF ──────────────▶ PyMuPDF / Docling
复杂 PDF（表格/图表）────▶ LlamaParse / Unstructured
扫描件/图片 ─────────────▶ LlamaParse（OCR）/ Unstructured
Office 文档 ─────────────▶ Unstructured / Apache Tika
代码文件 ────────────────▶ 直接读取 + 语法感知分块

from unstructured.partition.auto import partition
 
# 自动检测格式并解析
elements = partition(filename="report.pdf")
 
# 按元素类型分类处理
for element in elements:
    print(f"类型: {type(element).__name__}")
    print(f"内容: {element.text[:100]}...")
    print(f"元数据: {element.metadata}")
    print("---")

from llama_parse import LlamaParse
 
parser = LlamaParse(
    result_type="markdown",  # 输出 Markdown 格式
    num_workers=4,           # 并行处理
    language="zh",           # 中文文档
)
 
documents = parser.load_data("complex_report.pdf")
for doc in documents:
    print(doc.text[:500])

from docling.document_converter import DocumentConverter
 
converter = DocumentConverter()
result = converter.convert("internal_doc.pdf")
 
# 导出为 Markdown
markdown_output = result.document.export_to_markdown()
print(markdown_output[:500])

请分析以下文档解析结果的质量，并提出改进建议：

## 文档信息
- 文件名：[文件名]
- 格式：[PDF/Word/HTML/...]
- 页数：[页数]

## 解析结果样本
[粘贴解析后的前 500 字]

## 请评估
1. 文本完整性：是否有内容丢失？
2. 结构保留：标题层级、表格、列表是否正确保留？
3. 元数据质量：是否提取了有用的元数据？
4. 建议：推荐使用哪种解析工具或参数调整？

┌─────────────────────────────────────────────────┐
│           分块参数调优指南                         │
├─────────────────────────────────────────────────┤
│                                                 │
│  块大小（Chunk Size）                             │
│  ├── 256-512 tokens：高精度检索，适合 FAQ/短文档   │
│  ├── 512-1024 tokens：通用推荐，平衡精度和上下文   │
│  └── 1024-2048 tokens：长上下文需求，技术文档      │
│                                                 │
│  重叠（Overlap）                                  │
│  ├── 推荐：块大小的 10-20%                        │
│  ├── 过小：上下文断裂                             │
│  └── 过大：冗余增加、存储浪费                      │
│                                                 │
│  经验法则                                         │
│  ├── 递归分块 + 400-800 tokens + 20% 重叠        │
│  │   = 大多数场景的最佳起点                        │
│  └── 根据评估指标迭代调优                          │
│                                                 │
└─────────────────────────────────────────────────┘

from langchain.text_splitter import RecursiveCharacterTextSplitter
 
splitter = RecursiveCharacterTextSplitter(
    chunk_size=800,
    chunk_overlap=160,  # 20% 重叠
    separators=["\n\n", "\n", "。", ".", " ", ""],
    length_function=len,
)
 
chunks = splitter.split_text(document_text)
print(f"生成 {len(chunks)} 个块")

from langchain.text_splitter import RecursiveCharacterTextSplitter
 
# 中文文档优化的分隔符层级
splitter = RecursiveCharacterTextSplitter(
    chunk_size=600,
    chunk_overlap=120,
    separators=[
        "\n\n\n",   # 章节分隔
        "\n\n",      # 段落分隔
        "\n",        # 行分隔
        "。",        # 中文句号
        "；",        # 中文分号
        "，",        # 中文逗号
        ".",         # 英文句号
        " ",         # 空格
        "",          # 字符级（最后手段）
    ],
)
 
chunks = splitter.split_documents(documents)

from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings
 
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
 
semantic_splitter = SemanticChunker(
    embeddings,
    breakpoint_threshold_type="percentile",
    breakpoint_threshold_amount=95,  # 相似度低于 95 分位时切分
)
 
semantic_chunks = semantic_splitter.split_text(document_text)

from langchain.text_splitter import Language, RecursiveCharacterTextSplitter
 
# Python 代码分块
python_splitter = RecursiveCharacterTextSplitter.from_language(
    language=Language.PYTHON,
    chunk_size=1000,
    chunk_overlap=100,
)
 
# TypeScript 代码分块
ts_splitter = RecursiveCharacterTextSplitter.from_language(
    language=Language.TS,
    chunk_size=1000,
    chunk_overlap=100,
)
 
code_chunks = python_splitter.split_text(python_source_code)

from llama_index.core.node_parser import (
    SentenceSplitter,
    HierarchicalNodeParser,
)
from llama_index.core.retrievers import AutoMergingRetriever
 
# 创建层级分块器：大块 2048、中块 512、小块 128
node_parser = HierarchicalNodeParser.from_defaults(
    chunk_sizes=[2048, 512, 128],
)
 
nodes = node_parser.get_nodes_from_documents(documents)
 
# 检索时用小块匹配，返回时自动合并为父块
retriever = AutoMergingRetriever(
    vector_retriever,
    storage_context,
    simple_ratio_thresh=0.5,  # 超过 50% 子块命中则返回父块
)

请为以下文档推荐最佳分块策略：

## 文档特征
- 类型：[技术文档/法律合同/代码库/FAQ/论文/...]
- 平均长度：[页数或字数]
- 语言：[中文/英文/混合]
- 结构化程度：[高（有清晰标题层级）/ 中 / 低（纯文本流）]
- 查询类型：[精确事实查找 / 概念理解 / 代码搜索 / ...]

## 请推荐
1. 分块策略（固定/递归/语义/结构/代码感知）
2. 建议的块大小和重叠比例
3. 是否需要父子分块
4. 需要注意的特殊处理

你的场景是什么？
│
├── 快速原型 / 预算有限
│   └── ✅ text-embedding-3-small（$0.02/百万 tokens）
│
├── 生产级通用场景
│   ├── 英文为主 ──▶ ✅ voyage-3（性价比最优）
│   ├── 中文为主 ──▶ ✅ BGE-M3 或 GTE-Qwen2（中文优化）
│   └── 多语言 ──▶ ✅ Cohere embed-v4（多语言领先）
│
├── 最高精度需求
│   └── ✅ voyage-3-large 或 mistral-embed
│
├── 代码/技术文档
│   └── ✅ voyage-3-large（代码领域优化）
│
├── 隐私/自托管需求
│   └── ✅ BGE-M3 / Jina v3 / GTE-Qwen2（开源自托管）
│
└── 超长文档（>8K tokens）
    └── ✅ voyage-3（32K）/ GTE-Qwen2（32K）/ NV-Embed-v2（32K）

from openai import OpenAI
 
client = OpenAI()
 
def get_embeddings(texts: list[str], model="text-embedding-3-small"):
    """批量获取文本嵌入向量"""
    response = client.embeddings.create(
        input=texts,
        model=model,
    )
    return [item.embedding for item in response.data]
 
# 单条嵌入
embedding = get_embeddings(["RAG 流水线是什么？"])[0]
print(f"向量维度: {len(embedding)}")  # 1536
 
# 批量嵌入（推荐，减少 API 调用）
texts = ["文档片段1", "文档片段2", "文档片段3"]
embeddings = get_embeddings(texts)

# text-embedding-3 系列支持 Matryoshka 维度缩减
response = client.embeddings.create(
    input=["RAG 流水线详解"],
    model="text-embedding-3-large",
    dimensions=256,  # 从 3072 缩减到 256，存储减少 12x
)
 
# 精度损失约 2-5%，但存储和检索速度大幅提升

from sentence_transformers import SentenceTransformer
 
# 加载 BGE-M3 多语言模型
model = SentenceTransformer("BAAI/bge-m3")
 
texts = [
    "RAG 系统的核心是检索增强生成",
    "向量数据库存储文档的嵌入表示",
]
 
# 本地推理，数据不出服务器
embeddings = model.encode(texts, normalize_embeddings=True)
print(f"向量维度: {embeddings.shape[1]}")  # 1024

import voyageai
 
client = voyageai.Client()
 
result = client.embed(
    texts=["RAG 流水线的检索阶段"],
    model="voyage-3-large",
    input_type="document",  # 文档嵌入用 "document"，查询用 "query"
)
 
embedding = result.embeddings[0]
print(f"向量维度: {len(embedding)}")  # 1024

请帮我选择最适合的嵌入模型：

## 项目需求
- 文档语言：[中文/英文/多语言]
- 文档类型：[技术文档/法律文件/代码/通用]
- 文档规模：[千级/万级/百万级]
- 查询类型：[语义搜索/精确匹配/代码搜索]
- 部署方式：[云 API/自托管/混合]
- 月预算：[金额]
- 延迟要求：[实时 <100ms / 准实时 <500ms / 批处理]

## 请推荐
1. 首选模型及理由
2. 备选模型
3. 预估月成本
4. 需要注意的限制

import chromadb
 
# 创建持久化客户端
client = chromadb.PersistentClient(path="./chroma_db")
 
# 创建或获取集合
collection = client.get_or_create_collection(
    name="my_documents",
    metadata={"hnsw:space": "cosine"},  # 使用余弦相似度
)
 
# 添加文档
collection.add(
    documents=["RAG 是检索增强生成", "向量数据库存储嵌入"],
    metadatas=[{"source": "doc1"}, {"source": "doc2"}],
    ids=["id1", "id2"],
)
 
# 查询
results = collection.query(
    query_texts=["什么是 RAG？"],
    n_results=3,
)
print(results["documents"])

from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct
 
# 连接 Qdrant（本地或云端）
client = QdrantClient(url="http://localhost:6333")
 
# 创建集合
client.create_collection(
    collection_name="documents",
    vectors_config=VectorParams(
        size=1536,              # 向量维度
        distance=Distance.COSINE,
    ),
)
 
# 插入向量
client.upsert(
    collection_name="documents",
    points=[
        PointStruct(
            id=1,
            vector=embedding_vector,
            payload={"text": "文档内容", "source": "report.pdf", "page": 5},
        ),
    ],
)
 
# 搜索
results = client.search(
    collection_name="documents",
    query_vector=query_embedding,
    limit=5,
)

-- 启用 pgvector 扩展
CREATE EXTENSION IF NOT EXISTS vector;
 
-- 创建文档表
CREATE TABLE documents (
    id SERIAL PRIMARY KEY,
    content TEXT NOT NULL,
    metadata JSONB,
    embedding vector(1536),  -- 1536 维向量
    created_at TIMESTAMP DEFAULT NOW()
);
 
-- 创建 HNSW 索引（推荐，比 IVFFlat 更快）
CREATE INDEX ON documents
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);
 
-- 相似度搜索
SELECT id, content, metadata,
       1 - (embedding <=> $1::vector) AS similarity
FROM documents
ORDER BY embedding <=> $1::vector
LIMIT 5;

from llama_index.core import VectorStoreIndex
 
# 构建向量索引
index = VectorStoreIndex.from_documents(documents)
 
# 创建检索器
retriever = index.as_retriever(
    similarity_top_k=10,  # 返回 Top-10 结果
)
 
# 检索
nodes = retriever.retrieve("RAG 系统如何处理中文文档？")
for node in nodes:
    print(f"相关度: {node.score:.4f} | {node.text[:100]}...")

from llama_index.retrievers.bm25 import BM25Retriever
 
# 创建 BM25 检索器
bm25_retriever = BM25Retriever.from_defaults(
    nodes=all_nodes,
    similarity_top_k=10,
)
 
# BM25 擅长精确关键词匹配
results = bm25_retriever.retrieve("pgvector HNSW 索引配置")

from llama_index.core.retrievers import QueryFusionRetriever
 
# 融合向量检索和 BM25 检索
hybrid_retriever = QueryFusionRetriever(
    retrievers=[vector_retriever, bm25_retriever],
    similarity_top_k=10,
    num_queries=1,           # 不做查询扩展
    mode="reciprocal_rerank", # 使用 RRF 融合排序
)
 
results = hybrid_retriever.retrieve("如何优化 RAG 检索质量？")

from qdrant_client import QdrantClient, models
 
client = QdrantClient(url="http://localhost:6333")
 
# Qdrant 原生支持混合搜索（稠密 + 稀疏）
results = client.query_points(
    collection_name="documents",
    prefetch=[
        # 稠密检索
        models.Prefetch(
            query=dense_embedding,
            using="dense",
            limit=20,
        ),
        # 稀疏检索
        models.Prefetch(
            query=models.SparseVector(
                indices=sparse_indices,
                values=sparse_values,
            ),
            using="sparse",
            limit=20,
        ),
    ],
    # RRF 融合
    query=models.FusionQuery(fusion=models.Fusion.RRF),
    limit=10,
)

RRF_score(d) = Σ 1 / (k + rank_i(d))

其中：
- d = 文档
- k = 常数（通常 60）
- rank_i(d) = 文档 d 在第 i 个检索器中的排名

请帮我设计 RAG 系统的检索策略：

## 系统信息
- 知识库规模：[文档数量和总大小]
- 文档类型：[技术文档/FAQ/代码/混合]
- 查询特征：[自然语言/关键词/混合]
- 延迟要求：[<100ms / <500ms / <1s]

## 典型查询示例
1. [示例查询 1]
2. [示例查询 2]
3. [示例查询 3]

## 请推荐
1. 检索方法（稠密/稀疏/混合/多路）
2. 融合算法和权重建议
3. Top-K 参数建议
4. 是否需要查询转换（重写/分解/HyDE）

import cohere
 
co = cohere.Client("your-api-key")
 
# 对检索结果重排序
rerank_results = co.rerank(
    model="rerank-v3.5",
    query="RAG 系统如何处理中文文档？",
    documents=[node.text for node in retrieved_nodes],
    top_n=5,  # 只保留 Top-5
)
 
for result in rerank_results.results:
    print(f"排名: {result.index} | 相关度: {result.relevance_score:.4f}")

from sentence_transformers import CrossEncoder
 
# 加载交叉编码器重排序模型
reranker = CrossEncoder("BAAI/bge-reranker-v2-m3", max_length=512)
 
# 计算查询-文档对的相关度分数
query = "RAG 系统如何处理中文文档？"
pairs = [(query, node.text) for node in retrieved_nodes]
scores = reranker.predict(pairs)
 
# 按分数排序
ranked_indices = scores.argsort()[::-1]
top_nodes = [retrieved_nodes[i] for i in ranked_indices[:5]]

from llama_index.core.postprocessor import SentenceTransformerRerank
 
# 创建重排序后处理器
reranker = SentenceTransformerRerank(
    model="BAAI/bge-reranker-v2-m3",
    top_n=5,
)
 
# 在查询引擎中使用
query_engine = index.as_query_engine(
    similarity_top_k=20,          # 先召回 20 个
    node_postprocessors=[reranker], # 重排序后保留 5 个
)
 
response = query_engine.query("RAG 流水线有哪些阶段？")

from llama_index.core.postprocessor import LongContextReorder
 
# 长上下文重排序：将最相关的内容放在开头和结尾
# （研究表明 LLM 对中间内容的注意力较弱——"Lost in the Middle"）
reorder = LongContextReorder()
 
query_engine = index.as_query_engine(
    similarity_top_k=10,
    node_postprocessors=[reranker, reorder],  # 先重排序，再重排列
)

from langchain.retrievers.document_compressors import LLMChainExtractor
from langchain_openai import ChatOpenAI
 
# 使用 LLM 提取每个文档中与查询最相关的部分
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
compressor = LLMChainExtractor.from_llm(llm)
 
# 压缩后的文档只包含与查询直接相关的句子
compressed_docs = compressor.compress_documents(
    documents=retrieved_docs,
    query="RAG 系统的分块策略有哪些？",
)

检索结果（Top-20）
    │
    ▼
┌──────────────┐
│  去重（Dedup） │  ← 去除重复或高度相似的文档
└──────┬───────┘
       │
       ▼
┌──────────────┐
│ 重排序（Rerank）│  ← 交叉编码器精排，保留 Top-10
└──────┬───────┘
       │
       ▼
┌──────────────────┐
│ 上下文压缩         │  ← 提取每个文档中最相关的部分
│（Context Compress）│
└──────┬───────────┘
       │
       ▼
┌──────────────────┐
│ 长上下文重排列     │  ← 最相关内容放开头和结尾
│（Lost-in-Middle） │
└──────┬───────────┘
       │
       ▼
  送入 LLM 生成

def build_rag_prompt(query: str, context_nodes: list) -> str:
    """组装 RAG Prompt"""
    context_parts = []
    for i, node in enumerate(context_nodes, 1):
        source = node.metadata.get("file_name", "未知来源")
        context_parts.append(f"[来源 {i}: {source}]\n{node.text}")
 
    context_str = "\n\n---\n\n".join(context_parts)
 
    return f"""你是一个基于知识库的问答助手。请严格根据以下检索到的上下文信息回答用户问题。
 
## 规则
1. 只使用提供的上下文信息回答问题
2. 在回答中用 [来源 N] 标注信息出处
3. 如果上下文中没有相关信息，明确说明
4. 不要编造上下文中不存在的信息
 
## 上下文信息
{context_str}
 
## 用户问题
{query}
 
## 请回答（附带来源引用）："""

from openai import OpenAI
 
client = OpenAI()
 
def generate_with_citations(query: str, context_nodes: list) -> str:
    """生成带引用的回答"""
    prompt = build_rag_prompt(query, context_nodes)
 
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "你是一个严谨的知识库问答助手，回答时必须标注来源。"},
            {"role": "user", "content": prompt},
        ],
        temperature=0.1,  # 低温度，减少幻觉
        max_tokens=1024,
    )
 
    return response.choices[0].message.content
 
# 使用
answer = generate_with_citations(
    query="RAG 系统的分块策略有哪些？",
    context_nodes=reranked_nodes,
)
print(answer)

from pydantic import BaseModel
 
class Citation(BaseModel):
    source: str
    page: int | None = None
    quote: str
 
class RAGResponse(BaseModel):
    answer: str
    citations: list[Citation]
    confidence: float  # 0-1
 
# 使用 OpenAI 结构化输出
response = client.beta.chat.completions.parse(
    model="gpt-4o-mini",
    messages=[
        {"role": "system", "content": "基于上下文回答问题，返回结构化引用。"},
        {"role": "user", "content": prompt},
    ],
    response_format=RAGResponse,
)
 
rag_response = response.choices[0].message.parsed
print(f"回答: {rag_response.answer}")
print(f"置信度: {rag_response.confidence}")
for cite in rag_response.citations:
    print(f"  来源: {cite.source} | 引用: {cite.quote}")

def stream_rag_response(query: str, context_nodes: list):
    """流式生成 RAG 回答"""
    prompt = build_rag_prompt(query, context_nodes)
 
    stream = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "你是知识库问答助手。"},
            {"role": "user", "content": prompt},
        ],
        temperature=0.1,
        stream=True,
    )
 
    for chunk in stream:
        if chunk.choices[0].delta.content:
            print(chunk.choices[0].delta.content, end="", flush=True)
 
stream_rag_response("RAG 流水线有哪些阶段？", reranked_nodes)

你是一个基于知识库的专业问答助手。

## 核心规则
1. **严格基于上下文**：只使用提供的上下文信息回答，不要使用你的训练知识
2. **标注来源**：每个关键信息点用 [来源 N] 标注出处
3. **诚实不确定性**：如果上下文信息不足以完整回答，明确说明哪些部分无法确认
4. **不编造**：绝对不要编造上下文中不存在的信息

## 上下文信息
[检索到的文档片段，每个片段标注来源编号]

## 用户问题
[用户查询]

## 回答格式
1. 直接回答问题
2. 关键信息标注 [来源 N]
3. 如有不确定之处，在末尾说明
4. 最后给出置信度评估（高/中/低）

"""
生产级 RAG 流水线完整示例
技术栈：LlamaIndex + Qdrant + OpenAI + Cohere Rerank
"""
from llama_index.core import (
    VectorStoreIndex,
    SimpleDirectoryReader,
    Settings,
    StorageContext,
)
from llama_index.core.node_parser import SentenceSplitter
from llama_index.vector_stores.qdrant import QdrantVectorStore
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.llms.openai import OpenAI
from llama_index.postprocessor.cohere_rerank import CohereRerank
from qdrant_client import QdrantClient
 
# ============================================
# 阶段 1：配置全局设置
# ============================================
Settings.llm = OpenAI(model="gpt-4o-mini", temperature=0.1)
Settings.embed_model = OpenAIEmbedding(
    model="text-embedding-3-small",
    dimensions=512,  # 维度缩减，节省存储
)
 
# ============================================
# 阶段 2：文档摄入与解析
# ============================================
documents = SimpleDirectoryReader(
    input_dir="./knowledge_base",
    recursive=True,
    required_exts=[".md", ".txt", ".pdf", ".py", ".ts"],
    filename_as_id=True,
).load_data()
 
print(f"加载了 {len(documents)} 个文档")
 
# ============================================
# 阶段 3：分块
# ============================================
splitter = SentenceSplitter(
    chunk_size=512,
    chunk_overlap=100,
    paragraph_separator="\n\n",
)
 
# ============================================
# 阶段 4：向量存储（Qdrant）
# ============================================
qdrant_client = QdrantClient(url="http://localhost:6333")
vector_store = QdrantVectorStore(
    client=qdrant_client,
    collection_name="tech_knowledge_base",
)
storage_context = StorageContext.from_defaults(vector_store=vector_store)
 
# ============================================
# 阶段 5：构建索引（嵌入 + 存储）
# ============================================
index = VectorStoreIndex.from_documents(
    documents,
    storage_context=storage_context,
    transformations=[splitter],
    show_progress=True,
)
 
# ============================================
# 阶段 6：配置检索 + 重排序 + 生成
# ============================================
cohere_rerank = CohereRerank(
    model="rerank-v3.5",
    top_n=5,
)
 
query_engine = index.as_query_engine(
    similarity_top_k=20,                # 先召回 20 个
    node_postprocessors=[cohere_rerank], # 重排序后保留 5 个
    response_mode="tree_summarize",      # 树状摘要模式
)
 
# ============================================
# 阶段 7：查询
# ============================================
response = query_engine.query(
    "我们的 API 认证机制是如何实现的？请详细说明。"
)
 
print("回答:", response.response)
print("\n来源:")
for node in response.source_nodes:
    print(f"  - {node.metadata.get('file_name', '未知')} "
          f"(相关度: {node.score:.4f})")