超越朴素 RAG

朴素 RAG 的工作方式

def naive_rag(query, k=5):
    q = embed(query)
    chunks = vector_store.search(q, k)        # top-k by cosine
    context = '\n\n'.join(c.text for c in chunks)
    return llm('Context:\n' + context + '\n\nQ: ' + query)

检索召回率与精确率

# The core dilemma
# small k -> may miss the gold chunk (recall problem)
# large k -> distractors crowd context (precision + cost problem)
# Advanced RAG decouples 'retrieve many' from 'use few'

嵌入不匹配问题

# Query:  'how do I revoke a token?'
# Doc:    'Token invalidation is performed via the /sessions endpoint.'
# Lexically and semantically distant -> bi-encoder may miss it
sim = cos(embed('how do I revoke a token?'),
          embed('Token invalidation via /sessions'))  # may be low

中间内容丢失

# Retrieval rank != attention rank
# Place the highest-relevance chunk at the START or END,
# never stranded in the middle of a large concatenation.

对干扰内容的敏感性

# Empirically: appending a single highly-similar but WRONG chunk
# can flip a previously-correct answer. RAG quality depends on
# keeping distractors OUT, not just getting the gold chunk IN.

分块病变

def structure_aware_chunks(doc, max_tokens=400, overlap=50):
    sections = split_by_headings(doc)        # respect document structure
    chunks = []
    for sec in sections:
        for c in sliding_window(sec.text, max_tokens, overlap):
            chunks.append(Chunk(c, meta={'section': sec.title}))
    return chunks

仅语义检索的缺口

def rrf(dense_ranks, sparse_ranks, k0=60):
    scores = {}
    for ranks in (dense_ranks, sparse_ranks):
        for rank, doc_id in enumerate(ranks):
            scores[doc_id] = scores.get(doc_id, 0) + 1.0 / (k0 + rank)
    return sorted(scores, key=scores.get, reverse=True)

过时且无法验证的上下文

def filtered_retrieve(q, after_date):
    cands = vector_store.search(embed(q), k=50)
    fresh = [c for c in cands if c.meta['date'] >= after_date]
    return fresh  # then re-rank; generator must cite c.id

没有反馈，就没有适应

def adaptive_rag(q):
    chunks = retrieve(q)
    if relevance_score(q, chunks) < 0.4:
        q2 = rewrite_query(q)            # reformulate and retry
        chunks = retrieve(q2)
    if relevance_score(q, chunks) < 0.4:
        return 'I could not find this in the sources.'
    return generate(q, chunks)

高级 RAG 技术栈

def advanced_rag(q):
    cands = hybrid_retrieve(rewrite_query(q), k=50)  # high recall
    top = rerank(q, cands)[:8]                       # precision
    ctx = compress(q, top)                            # focus + fit
    return generate_with_citations(q, ctx)            # verifiable

先测量，再优化

def diagnose(eval_set):
    return {
        'recall@5':  recall_at_k(eval_set, k=5),     # retrieval health
        'recall@50': recall_at_k(eval_set, k=50),    # ceiling with rerank
        'answer_acc': answer_accuracy(eval_set),      # generation health
    }

快速检查

回顾