RAG 检索增强 实战路线：从最小可用到生产级

def minimal_rag(docs, question, k=4):
    chunks = []
    for doc in docs:
        # 固定长度分块，最小可用阶段够用，后文会换掉
        for i in range(0, len(doc.text), 1000):
            chunks.append(Chunk(text=doc.text[i:i + 1000], source=doc.name))

    index = VectorIndex()
    for c in chunks:
        index.add(c.id, embed(c.text))

    hits = index.search(embed(question), top_k=k)   # 只按向量相似度排序
    context = "\n\n".join(f"[{i + 1}] {h.text}" for i, h in enumerate(hits))
    return llm(f"只依据下方片段回答，缺少依据时直接说明。\n{context}\n问题：{question}")

def chunk_document(doc):
    """按标题层级切分，表格整体成块，超长块递归细分并保留重叠。"""
    out = []
    for section in doc.sections():          # H2 / H3 构成的层级单元
        path = "/".join(section.heading_path)
        for unit in section.units():        # 段落、列表项、整张表格
            text = unit.text
            if len(tokenize(text)) > 1200:
                pieces = split_recursive(text, size=1200, overlap=200)
            else:
                pieces = [text]
            for p in pieces:
                out.append(Chunk(
                    text=p,
                    doc_title=doc.title,
                    section_path=path,
                    updated_at=doc.updated_at,
                    source=doc.source_system,
                ))
    return out

RRF_K = 60

def rrf_merge(rank_lists, weights=None):
    """rank_lists: [[chunk_id, ...], ...]，每个列表已按分数从高到低排好。"""
    scores = {}
    for li, lst in enumerate(rank_lists):
        w = 1.0 if weights is None else weights[li]
        for rank, cid in enumerate(lst, start=1):
            scores[cid] = scores.get(cid, 0.0) + w / (RRF_K + rank)
    return sorted(scores.items(), key=lambda x: x[1], reverse=True)

@dataclass
class EvalRecord:
    question: str
    expected_chunk_ids: list[str]
    retrieved_chunk_ids: list[str]     # 重排后进上下文的那批
    answer: str
    cited_ids: list[str]
    unsupported: bool                  # 引用的 ID 不在 retrieved 集合内

def run_eval(golden_set):
    rows = []
    for case in golden_set:
        hits, answer, cited = pipeline(case.question)
        recalled = len(set(case.expected_chunk_ids) & set(hits))
        rows.append(EvalRecord(
            question=case.question,
            expected_chunk_ids=case.expected_chunk_ids,
            retrieved_chunk_ids=[h.id for h in hits],
            answer=answer,
            cited_ids=cited,
            unsupported=any(c not in [h.id for h in hits] for c in cited),
        ))
    return {
        "recall@k": mean(len(set(r.expected_chunk_ids) & set(r.retrieved_chunk_ids)) / len(r.expected_chunk_ids) for r in rows),
        "mrr": mean(first_hit_rank(r) for r in rows),
        "unsupported_rate": mean(r.unsupported for r in rows),
    }