rag-patterns

RAG 模式

标准 RAG 流水线

Documents → Chunk → Embed → Store (vector DB)
Query → Embed → Retrieve → Augment prompt → Generate answer

分块策略

from langchain_text_splitters import RecursiveCharacterTextSplitter

# Recommended defaults
splitter = RecursiveCharacterTextSplitter(
    chunk_size=800,      # chars (not tokens)
    chunk_overlap=200,
    separators=["\n\n", "\n", ". ", " ", ""],
)
chunks = splitter.split_documents(docs)

策略	适用场景	分块大小
固定大小并重叠	通用文本	500-1000 字符
递归字符分块	结构化文档	500-1000 字符
语义分块（按含义）	长篇内容	可变
文档感知分块（Markdown 标题）	技术文档	按章节划分

元数据增强

for chunk in chunks:
    chunk.metadata.update({
        "source": doc.metadata["source"],
        "section": extract_section_title(chunk),
        "doc_id": doc.metadata["id"],
        "chunk_index": i,
    })

检索策略

混合搜索（关键词 + 语义）

from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever

bm25 = BM25Retriever.from_documents(docs, k=5)
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

hybrid = EnsembleRetriever(
    retrievers=[bm25, vector_retriever],
    weights=[0.3, 0.7],
)

重排序

from cohere import Client

cohere = Client(api_key=COHERE_API_KEY)

def rerank(query: str, documents: list[str], top_n: int = 5):
    response = cohere.rerank(
        model="rerank-english-v3.0",
        query=query,
        documents=documents,
        top_n=top_n,
    )
    return [documents[r.index] for r in response.results]

多查询检索

# Generate multiple query variations for better recall
prompt = """Generate 3 different versions of this question
to retrieve relevant documents: {question}"""

queries = llm.invoke(prompt).split("\n")
all_docs = set()
for q in queries:
    all_docs.update(retriever.invoke(q))

提示词构建

SYSTEM_PROMPT = """Answer based only on the provided context.
If the context doesn't contain the answer, say "I don't have enough information."
Cite sources using [Source: filename] format.

Context:
{context}"""

def format_context(docs, max_tokens=3000):
    context_parts = []
    for doc in docs:
        source = doc.metadata.get("source", "unknown")
        context_parts.append(f"[Source: {source}]\n{doc.page_content}")
    return "\n\n---\n\n".join(context_parts)

评估

指标	衡量内容	工具
上下文相关性	检索到的文档是否相关？	RAGAS、人工评估
忠实度	答案是否与上下文一致？	RAGAS
答案相关性	答案是否回应了问题？	RAGAS
检索召回率	是否检索到了正确的文档？	自定义评估集

from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_precision

result = evaluate(dataset, metrics=[faithfulness, answer_relevancy, context_precision])

反模式

反模式	修正方法
分块过大（>1500 字符）	使用 500-1000 字符的分块，并设置 200 字符重叠
分块没有元数据	存储来源、章节和页码
未评估检索效果	构建评估集，衡量召回率和精确率
将所有分块塞入提示词	限制为 top-K（3-5），并使用重排序
忽略混合搜索	结合 BM25 + 向量搜索，以提升召回率
未追踪引用或来源	在整个流水线中传递元数据

生产环境检查清单

• 使用评估集调优分块策略
• 启用混合搜索（BM25 + 向量）
• 对检索结果进行重排序
• 在答案中标注来源
• 为超出范围的问题设置防护措施
• 监控：检索延迟、答案质量评分
• 为新文档建立增量索引