专题 · RAG 检索管线¶

agentframework.retrieval 是把文档变成可检索知识的能力包。它把每个环节定义成协议 + 内置实现，你可以替换任意一段（如换真实 embedding、换第三方向量库）而不用改工作流。

1. 管线总览¶

文本/文档 ─► 切分 splitter ─► 向量化 embedder ─► 入库 store
                                              │  (VectorStore 协议)
用户问题 ─► 检索 retriever (create_retriever)
        └─ 命中 ─► [KnowledgeBaseSearch 节点] ─► 注入 ChatLLM(context=...)

主入口与分工：

对象	作用
`RecursiveCharacterSplitter`	把长文本切成块（`TextSplitter` 的内置实现）
`BaseEmbedder` / `HashEmbedder`	文本 → 向量（协议 + 内置哈希实现）
`VectorStore` / `MemoryVectorStore`	向量读写（协议 + 内置内存实现）
`create_retriever(...)`	组装检索器（embed query → store.search，可选 rerank / BM25 混合）
`ingest_document / ingest_documents`	一段式导入：切分 → embed → 入库（同 source 幂等）
`KnowledgeBaseSearch`	把 retriever 接进工作流的节点

工作流里的用法见指南 02；本页讲每个协议/实现本身，便于你替换或对接真实组件。

2. splitter · 切分¶

from agentframework.retrieval import DocumentChunk, RecursiveCharacterSplitter

splitter = RecursiveCharacterSplitter(chunk_size=800, chunk_overlap=None)
chunks: list[DocumentChunk] = splitter.split_text(long_text, metadata={"source": "doc1"})

DocumentChunk：text: str；metadata: dict。
抽象基类 TextSplitter：split_text(text, *, metadata=None) -> list[DocumentChunk]，可继承实现自己的切分策略。
RecursiveCharacterSplitter 参数：
chunk_size：目标块大小（默认 800，<=0 抛 ValueError）。
chunk_overlap：相邻块重叠（默认 min(100, chunk_size//4)，越界抛 ValueError）。
切分按优先级分隔符递归：["\n\n", "\n", "。", "！", "？", ". ", "! ", "? ", " ", ""]，无分隔符则固定大小硬切，相邻块保留重叠。

3. embedder · 向量化¶

from agentframework.retrieval import BaseEmbedder, HashEmbedder

embedder = HashEmbedder(dimension=32)  # 内置：确定性哈希，离线可跑（非语义）
vec = embedder.embed(["文本一", "文本二"])  # list[list[float]]，每行 dimension 维

协议 BaseEmbedder：dimension 属性 + embed(texts) -> list[list[float]]（embed_one 默认实现）。
HashEmbedder：确定性哈希 + L2 归一化，仅供测试/无外部依赖——只能精确匹配，不具备语义。要语义检索，实现一个真实中文 embedding 的 BaseEmbedder 替换即可（其余环节不变）。

4. store · 向量存储¶

存储是协议 VectorStore，由使用方实现（框架不绑死某个向量库）：

from typing import Protocol


class VectorStore(Protocol):
    def add(self, chunks, vectors) -> None: ...  # 批量写块 + 向量
    def search(self, vector, *, top_k=5) -> list[SearchResult]: ...  # 向量检索
    def delete_by_source(self, source: str) -> int: ...  # 按来源删除
    def count(self) -> int: ...

SearchResult：chunk: DocumentChunk；score: float（0~1，越大越相关）。
MemoryVectorStore：内置内存实现，余弦相似度，演示/测试用。
对接第三方（如 ChromaDB）：实现这 4 个方法即可。完整示例见 rag_kb 示例（实现 VectorStore 对接 chromadb.PersistentClient，块落盘、幂等重建）。

5. retriever · 检索¶

用工厂组装：

from agentframework.retrieval import create_retriever

retriever = create_retriever(store, embedder, top_k=5, reranker=None, bm25_weight=0.0)
hits = retriever.search("租房提取要什么材料")  # list[SearchResult]
cites = retriever.search_with_citations(
    "租房提取要什么材料"
)  # list[dict] [{source, chunk_index, text, score}]

top_k：返回条数（默认 5）。
reranker：可选回调 Callable[[str, list[SearchResult]], list[SearchResult]]，rerank 后再截断。
bm25_weight：0（默认）= 纯向量；>0 = 向量分×(1-w) + 简化 BM25 关键词分×w 的混合检索。
search_with_citations：归一化为带引用的 dict 列表（score 保留 4 位），是 KnowledgeBaseSearch 输出 results 的来源。

6. ingest · 一段式导入¶

from agentframework.retrieval import ingest_document, ingest_documents

ingest_document(
    long_text,
    store=store,
    embedder=embedder,
    source="zhijin_guide",  # source 非空串
    splitter=RecursiveCharacterSplitter(chunk_size=50, chunk_overlap=10),
    metadata=None,
)

ingest_documents(
    [{"text": "...", "source": "a"}, {"text": "...", "source": "b", "metadata": {...}}],
    store=store,
    embedder=embedder,
    splitter=splitter,
)

ingest_document 安全语义：先切分 / 向量化成功，再删旧、再写——同一 source 重复导入是幂等的，中途失败不会丢掉旧数据（避免失败导入把已入库内容冲掉）。自动补 chunk_index 元数据，校验向量维度一致。

7. 端到端最小链路¶

把上面串起来，即可得到指南 02 里接入 KnowledgeBaseSearch 的 retriever：

from agentframework.retrieval import (
    HashEmbedder,
    MemoryVectorStore,
    RecursiveCharacterSplitter,
    create_retriever,
    ingest_document,
)

store = MemoryVectorStore()
embedder = HashEmbedder(dimension=32)
ingest_document(
    "租房提取公积金需准备：身份证、租房合同、银行卡…",
    store=store,
    embedder=embedder,
    source="guide",
    splitter=RecursiveCharacterSplitter(chunk_size=50, chunk_overlap=10),
)
retriever = create_retriever(store, embedder, top_k=2)

8. 生产替换建议¶

环节	演示实现	生产替换为
embedder	`HashEmbedder`	真实中文 embedding（实现 `BaseEmbedder`）
store	`MemoryVectorStore`	ChromaDB / Milvus / ES（实现 `VectorStore`，参考 `rag_kb.py`）
检索	纯向量 / 简化 BM25	配 `reranker` + `bm25_weight` 混合