扩展点

RAGSpine 的可插拔性不是插件注册表——就是普通的结构化类型。每个 外部依赖都是一个 Python Protocol；核心依赖抽象，绝不依赖 厂商 SDK。新增一个 provider、向量存储、重排器或 OCR 引擎只需新增一个 文件，且每个重量级 SDK 都是延迟导入的，因此核心可以干净地导入，并借助 确定性的 MockProvider 完全离线运行。

缝的清单

当前的检索与工作流缝

Chunker.chunk(text, meta, *, max_chars, overlap_chars) -> list[Chunk] 掌管分块策略。 make_chunker 支持确定性默认、版面、父子、句子窗口、法律、QA、 书籍与语义等实现，外加 ragspine.chunkers entry-point 组。 实现必须保留 doc_id 与 source_locator；层级/窗口数据是附加信息， 不能替代来源链路。

LibraryRouter 选出一个或多个 RoutableLibrary ID。MultiIndexRetriever 独立调用 选中的各个 NarrativeRetriever 并对结果做 RRF 融合；路由器失败时 回退到全部库。FilterExtractor 可以提出元数据谓词，但过滤器始终是 经过校验的收窄操作。

TemplateMatcher 接收的目录引用只包含匹配元数据，不含工作流提示词或 凭据。内置的词法与 ONNX 匹配器返回排好序的候选；脚手架策略 施加分数/差距阈值并校验选中的模板。自定义匹配器不会 获得文件写入或执行权限。

LLMProvider

归 corespine 所有，并从 src/ragspine/agent/llm_provider.py 再导出——agent 工具使用循环驱动的唯一方法。它 使用 OpenAI chat-completions 形态（单个 chat 方法）。AnthropicProvider 延迟导入 anthropic SDK 并把该形态映射到 Anthropic API；MockProvider 既不需要密钥也不需要网络。

EmbeddingBackend

src/ragspine/retrieval/lexical/retrieval.py — 向量通道的 依赖注入点。默认是 none（纯 BM25）；注入它即可加上向量通道。

ListwiseJudge

src/ragspine/retrieval/rerank/listwise_rerank.py — 可选的 LLM 列表式重排器 缝。真实实现是 Claude（经 build_listwise_prompt / parse_listwise_response）；测试使用确定性的 fake。缺席时回退到 RRF 顺序。

OcrBackend

src/ragspine/extraction/extractors/pdf_scanned_extractor.py — 扫描版 PDF 的 OCR/VLM 缝。真实后端（PaddleOCR）运行在 Ubuntu + GPU 上；逻辑测试使用 fake，使 渲染 → 映射 → 阈值 → 审核的流程在无 GPU 机器上也完全可测。

NarrativeRetriever

src/ragspine/agent/agent.py — 注入 answer_question 的叙事检索 实现。鸭子类型；省略时叙事路径会诚实降级。

TaskQueue

src/ragspine/service/tasks/task_queue.py — 异步作业队列。生产用 RQQueue（RQ + Redis）， 测试用 FakeQueue（同步内联）。rq / redis 仅在 RQQueue 内部延迟导入。

SourceConnector

src/ragspine/ingestion/source/connector.py — 知识库来源缝。连接器 从你的知识库所在之处产出原始文档；默认的 FilesystemConnector 不需要 extra，HTTP / Notion 连接器在 [connectors] extra 之后延迟导入 httpx。

内置实现按名字注册在模块级表中；第三方连接器通过 ragspine.source_connectors entry-point 组发现（名字冲突时内置实现获胜）。 可用 make_source_connector(spec=None, **kwargs) 工厂或 RAGSPINE_SOURCE_CONNECTOR 环境变量选择（none / None → None）：

RelationExtractor

src/ragspine/graph/extractor.py — 由 build_relation_graph(..., relation_extractor=None) 消费的可选叙事关系插槽。取默认值 None 时，基础图 逐字节一致（没有额外的边）。参见 ADR 0015。

通过 make_relation_extractor(spec=None, *, provider=None, profile=None, **kwargs) 或 RAGSPINE_RELATION_EXTRACTOR 选择（none → None；deterministic / rule / cooccurrence → 确定性默认实现；llm / on → LLM 抽取器，无 provider 可用时 诚实降级为 None）。

实现一个 Protocol 并注入它

由于这些缝是结构化的，你只需实现相应方法并把实例传进去——无需 注册，无需基类。下面是一个基于 OpenAI 的 LLMProvider，严格对应真实的 chat 签名与 ChatCompletion 形态（这些 dataclass 来自 corespine，并从 ragspine.agent.llm_provider 再导出）：

from typing import Any

from openai import OpenAI  # lazy: only your file imports the SDK
from corespine import ChatCompletion, Choice, ResponseMessage, Usage


class OpenAIProvider:
    """A custom LLMProvider — no subclassing, just the chat method."""

    def __init__(self, model: str = "gpt-4o") -> None:
        self._client = OpenAI()
        self._model = model

    def chat(
        self,
        messages: list[dict[str, Any]],
        *,
        tools: list[dict[str, Any]] | None = None,
    ) -> ChatCompletion:
        # The messages are already in the OpenAI chat-completions shape, so they pass
        # through; the SDK response is the same shape, so the mapping back is direct.
        resp = self._client.chat.completions.create(
            model=self._model,
            messages=messages,
            tools=tools or [],  # adapt / omit as needed
        )
        choice = resp.choices[0]
        message = ResponseMessage(role="assistant", content=choice.message.content)
        u = resp.usage
        usage = (
            Usage(
                prompt_tokens=u.prompt_tokens,
                completion_tokens=u.completion_tokens,
                total_tokens=u.total_tokens,
            )
            if u is not None
            else None
        )
        return ChatCompletion(
            choices=(Choice(index=0, message=message, finish_reason=choice.finish_reason or "stop"),),
            usage=usage,
            model=resp.model,
            id=resp.id,
        )

在 MockProvider 原本所在的位置注入即可：

from ragspine.agent.agent import answer_question
from ragspine.storage.fact_store import SqliteFactStore
from my_openai_provider import OpenAIProvider

store = SqliteFactStore("data/fact_metric.db")
store.init_schema()
result = answer_question("...", store, OpenAIProvider())