rag-pipeline-builder

RAG 流水线构建器

设计端到端 RAG 流水线，实现准确的文档检索与生成。

流水线架构

Documents → Chunking → Embedding → Vector Store → Retrieval → Reranking → Generation

分块策略

# Semantic chunking (recommended)
from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,        # Characters per chunk
    chunk_overlap=200,      # Overlap between chunks
    separators=["\n\n", "\n", ". ", " ", ""],
    length_function=len,
)

chunks = splitter.split_text(document.text)

# Add metadata to each chunk
for i, chunk in enumerate(chunks):
    chunks[i] = {
        "text": chunk,
        "metadata": {
            "source": document.filename,
            "page": calculate_page(i),
            "chunk_id": f"{document.id}_chunk_{i}",
        }
    }

元数据架构

interface ChunkMetadata {
  // Source information
  document_id: string;
  source: string;
  url?: string;

  // Location
  page?: number;
  section?: string;
  chunk_index: number;

  // Content classification
  content_type: "text" | "code" | "table" | "list";
  language?: string;

  // Timestamps
  created_at: Date;
  updated_at: Date;

  // Retrieval optimization
  keywords: string[];
  summary?: string;
  importance_score?: number;
}

向量存储设置

# Pinecone example
import pinecone
from langchain.vectorstores import Pinecone
from langchain.embeddings import OpenAIEmbeddings

pinecone.init(api_key="...", environment="...")

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

vectorstore = Pinecone.from_documents(
    documents=chunks,
    embedding=embeddings,
    index_name="knowledge-base",
    namespace="production",
)

检索策略

# Hybrid search (dense + sparse)
def hybrid_retrieval(query: str, k: int = 5):
    # Dense retrieval (semantic)
    dense_results = vectorstore.similarity_search(query, k=k*2)

    # Sparse retrieval (keyword - BM25)
    sparse_results = bm25_search(query, k=k*2)

    # Combine and rerank
    combined = reciprocal_rank_fusion(dense_results, sparse_results)

    return combined[:k]

# Metadata filtering
results = vectorstore.similarity_search(
    query,
    k=5,
    filter={
        "content_type": "code",
        "language": "python",
    }
)

重排序

from sentence_transformers import CrossEncoder

reranker = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')

def rerank_results(query: str, results: List[Document], top_k: int = 3):
    # Score each result against query
    pairs = [(query, doc.page_content) for doc in results]
    scores = reranker.predict(pairs)

    # Sort by score
    scored_results = list(zip(results, scores))
    scored_results.sort(key=lambda x: x[1], reverse=True)

    return [doc for doc, score in scored_results[:top_k]]

查询增强

# Query expansion
def expand_query(query: str) -> str:
    expansion_prompt = f"""
    Generate 3 alternative phrasings of this query:
    "{query}"

    Return as JSON array of strings.
    """
    alternatives = llm(expansion_prompt)
    return [query] + alternatives

# Multi-query retrieval
def multi_query_retrieval(query: str, k: int = 5):
    queries = expand_query(query)
    all_results = []

    for q in queries:
        results = vectorstore.similarity_search(q, k=k)
        all_results.extend(results)

    # Deduplicate and rerank
    unique_results = deduplicate(all_results)
    return rerank_results(query, unique_results, k)

评估计划

# Define golden dataset
golden_dataset = [
    {
        "query": "How do I authenticate users?",
        "expected_docs": ["auth_guide.md", "user_management.md"],
        "relevant_chunks": ["chunk_123", "chunk_456"],
    },
]

# Metrics
def evaluate_retrieval(dataset):
    results = {
        "precision": [],
        "recall": [],
        "mrr": [],  # Mean Reciprocal Rank
        "ndcg": []  # Normalized Discounted Cumulative Gain
    }

    for item in dataset:
        retrieved = retrieval_fn(item["query"])
        retrieved_ids = [doc.metadata["chunk_id"] for doc in retrieved]

        # Calculate metrics
        relevant = set(item["relevant_chunks"])
        retrieved_set = set(retrieved_ids)

        precision = len(relevant & retrieved_set) / len(retrieved_set)
        recall = len(relevant & retrieved_set) / len(relevant)

        results["precision"].append(precision)
        results["recall"].append(recall)

    return {k: sum(v)/len(v) for k, v in results.items()}

上下文窗口管理

def fit_context_window(chunks: List[Document], max_tokens: int = 4000):
    """Select chunks that fit in context window"""
    total_tokens = 0
    selected_chunks = []

    for chunk in chunks:
        chunk_tokens = count_tokens(chunk.page_content)
        if total_tokens + chunk_tokens <= max_tokens:
            selected_chunks.append(chunk)
            total_tokens += chunk_tokens
        else:
            break

    return selected_chunks

最佳实践

1. 分块大小：一般文本为 500-1000 个字符
2. 重叠：分块之间重叠 10-20%
3. 元数据：使用丰富的元数据进行筛选
4. 混合搜索：结合语义搜索与关键词搜索
5. 重排序：使用交叉编码器进行最终排序
6. 评估：使用包含各项指标的黄金数据集
7. 上下文管理：不要超出模型限制

输出检查清单

• 已定义分块策略
• 已记录元数据架构
• 已配置向量存储
• 已实现检索算法
• 已添加重排序流水线
• 查询增强（可选）
• 已管理上下文窗口
• 已创建评估数据集
• 已实现指标
• 已建立性能基准