Config

load_pipeline_from_yaml

load_pipeline_from_yaml(path: str | Path) -> 'Pipeline'

Build a :class:Pipeline from a YAML config file.

See the module docstring for the schema. Raises ValueError with a clear message on unknown component types or missing required keys.

Source code in src/verifiable_rag/config.py

def load_pipeline_from_yaml(path: str | Path) -> "Pipeline":
    """Build a :class:`Pipeline` from a YAML config file.

    See the module docstring for the schema. Raises ``ValueError`` with
    a clear message on unknown component types or missing required keys.
    """
    try:
        import yaml
    except ImportError as exc:
        raise ImportError(
            "PyYAML is required for YAML config loading. "
            "Install with: pip install 'verifiable-rag[yaml]'"
        ) from exc

    from verifiable_rag.chunkers import ContextualChunker, LLMContextualizer
    from verifiable_rag.indexers import HybridIndex
    from verifiable_rag.parsers._cache import CachingParser
    from verifiable_rag.parsers.composite import CompositeParser
    from verifiable_rag.pipeline import Pipeline

    path = Path(path)
    with path.open() as fh:
        cfg = yaml.safe_load(fh) or {}

    # Required components
    parser_obj = _build_parser(cfg.get("parser", {}))
    chunker_obj = _build_chunker(cfg.get("chunker", {}))
    embedder_obj = _instantiate(
        "embedder",
        cfg["embedder"]["type"] if "embedder" in cfg else "bge",
        cfg.get("embedder", {}).get("config", {}),
    )
    indexer_obj = _build_indexer(cfg.get("indexer", {}))
    generator_obj = _instantiate(
        "generator",
        cfg["generator"]["type"] if "generator" in cfg else "prompted",
        cfg.get("generator", {}).get("config", {}),
    )

    # Optional components
    reranker_obj = None
    if "reranker" in cfg and cfg["reranker"].get("type") not in (None, "none"):
        reranker_obj = _instantiate(
            "reranker",
            cfg["reranker"]["type"],
            cfg["reranker"].get("config", {}),
        )

    verifier_obj = None
    if "verifier" in cfg and cfg["verifier"].get("type") not in (None, "none"):
        verifier_obj = _instantiate(
            "verifier",
            cfg["verifier"]["type"],
            cfg["verifier"].get("config", {}),
        )

    pipeline_kw = cfg.get("pipeline", {}) or {}
    return Pipeline(
        parser=parser_obj,
        chunker=chunker_obj,
        embedder=embedder_obj,
        indexer=indexer_obj,
        reranker=reranker_obj,
        generator=generator_obj,
        verifier=verifier_obj,
        strictness=pipeline_kw.get("strictness", "balanced"),
        top_k_retrieve=pipeline_kw.get("top_k_retrieve", 20),
        top_k_rerank=pipeline_kw.get("top_k_rerank", 8),
    )

Registry

register

register(component: str, name: str) -> Callable[[Callable[..., Any]], Callable[..., Any]]

Decorator: register a factory under REGISTRY[component][name].

Source code in src/verifiable_rag/config.py

def register(component: str, name: str) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
    """Decorator: register a factory under ``REGISTRY[component][name]``."""
    if component not in _REGISTRY:
        raise ValueError(
            f"Unknown component {component!r}; choose from {sorted(_REGISTRY)}"
        )

    def _wrap(fn: Callable[..., Any]) -> Callable[..., Any]:
        _REGISTRY[component][name] = fn
        return fn

    return _wrap

registered_types

registered_types() -> dict[str, list[str]]

Public view of the registry — {component: [type1, type2, ...]}.

Useful for discoverability::

for component, types in registered_types().items():
    print(f"{component}: {types}")

Source code in src/verifiable_rag/config.py

def registered_types() -> dict[str, list[str]]:
    """Public view of the registry — ``{component: [type1, type2, ...]}``.

    Useful for discoverability::

        for component, types in registered_types().items():
            print(f"{component}: {types}")
    """
    return {k: sorted(v) for k, v in _REGISTRY.items()}