Modular-RAG

换嵌入或换重排要改整条链，无法并行、也无法按模块回滚。根因是阶段之间没有稳定 I/O 契约。本方法按阶段 schema 拆模块。不直接涨分，是迭代速度与归因的前提。

本文属于 RAG 工程框架中的「6 模块化架构与评估闭环」环节，聚焦「Modular RAG」方法。

定位

维度	内容
角色	阶段 I/O 契约
输入 → 输出	稳定 schema 的可替换实现
默认组合	Haystack 组件图；LlamaIndex / LangChain 模块替换
何时不用	一次性 POC、不会换件

核心机制

钉死 chunk / candidates / evidence pack / answer 的字段。替换实现不得改端口。

实现路径与心智：按阶段钉死 I/O schema，实现可替换、可回滚。底层心智：RAG 是带类型端口的组件图，不是一个 retrieve_and_generate()。没有它就无法冻结评测。

优缺点

优点：并行开发、按模块回滚。
缺点：先付接口成本。

契约与走通样例

输入（召回端口）

1	{"query": "GAPDH siRNA 处理多久", "top_k": 5}

输出（同一端口，换 ColBERT 实现）

1	{"candidates": [{"id": "P-GAPDH-01-C1", "score": 1.55, "channel": "colbert"}]}

字段不变，实现可 A/B。

社区实现

Haystack 组件图最贴近端口思维。风险：LangChain 大类把多阶段缠死。

工程落地

最小可运行示例

复制为 .py 后直接运行（仅标准库）。三个模块只通过 Hit 契约通信，禁止读对方私有句柄。

"""检索 / 精排 / 生成可替换；字段契约不变。"""
from __future__ import annotations

from dataclasses import dataclass
from typing import Protocol


@dataclass(frozen=True)
class Hit:
    """跨模块唯一契约。chunk_id / score / channel 必填。"""

    chunk_id: str
    text: str
    score: float
    channel: str


class Retriever(Protocol):
    """召回模块。生产对应 Hybrid 检索器。"""

    def search(self, query: str) -> list[Hit]: ...


class Reranker(Protocol):
    """精排模块。生产对应 Cross-encoder。"""

    def pack(self, query: str, hits: list[Hit]) -> list[Hit]: ...


class Generator(Protocol):
    """生成模块。生产对应绑定证据的 LLM。"""

    def generate(self, query: str, packed: list[Hit]) -> str: ...


class KeywordRetriever:
    def search(self, query: str) -> list[Hit]:
        return [Hit("P-GAPDH-01-C1", "20 nM siRNA for 48 h", 0.4, "bm25")]


class IdentityReranker:
    def pack(self, query: str, hits: list[Hit]) -> list[Hit]:
        del query
        return sorted(hits, key=lambda h: -h.score)


class TemplateGenerator:
    def generate(self, query: str, packed: list[Hit]) -> str:
        del query
        return packed[0].text if packed else "refuse"


class RagService:
    """用接口组装，便于单测替换。"""

    def __init__(self, retriever: Retriever, reranker: Reranker, gen: Generator) -> None:
        self.retriever, self.reranker, self.gen = retriever, reranker, gen

    def answer(self, query: str) -> str:
        hits = self.retriever.search(query)
        packed = self.reranker.pack(query, hits)
        return self.gen.generate(query, packed)


if __name__ == "__main__":
    print(RagService(KeywordRetriever(), IdentityReranker(), TemplateGenerator()).answer("siRNA 时程？"))

参数

无模型超参。契约字段才是「参数」：chunk_id、score、channel 必填。

失效—信号—螺丝

换件改字段：评测不可比。螺丝：schema 测试。
隐式全局状态：螺丝：禁止模块读对方私有索引句柄。
为模块而模块：螺丝：POC 不必拆。

规模（100 篇生物学 PDF）

不直接增加数据面资源。多套实现时磁盘与 GPU 按模块倍增。拆分联调为人周级。