config

Declarative configuration helpers for composable Rago pipelines.

Classes:

• Cache –

  Resolve a cache backend into step configuration.

• DB –

  Resolve a vector database backend into step configuration.

• Logs –

  Public alias for attaching logs declaratively.

Cache

Cache(
    backend: str = 'file',
    target_dir: Path | str = '.rago-cache',
)

Bases: ParametersBase

Resolve a cache backend into step configuration.

Methods:

• apply –

  Merge additional configuration into this object.

• process –

  Return the input unchanged for configuration-only objects.

Attributes:

• params (dict[str, Any]) –

  Expose the underlying parameter mapping.

Source code in src/rago/config.py

def __init__(
    self,
    backend: str = 'file',
    target_dir: Path | str = '.rago-cache',
) -> None:
    backend_name = backend.lower()
    if backend_name != 'file':
        raise ValueError(f'Unsupported cache backend: {backend}')

    cache = CacheFile(target_dir=target_dir)
    super().__init__(cache=cache)

params property

params: dict[str, Any]

Expose the underlying parameter mapping.

apply

apply(parameters: Any) -> None

Merge additional configuration into this object.

Source code in src/rago/base.py

def apply(self, parameters: Any) -> None:
    """Merge additional configuration into this object."""
    self.data.update(config_to_dict(parameters))

process

process(inp: Input) -> Output

Return the input unchanged for configuration-only objects.

Source code in src/rago/base.py

def process(self, inp: Input) -> Output:
    """Return the input unchanged for configuration-only objects."""
    return inp.to_output()

DB

DB(backend: str = 'faiss', **kwargs: Any)

Bases: ParametersBase

Resolve a vector database backend into step configuration.

Methods:

• apply –

  Merge additional configuration into this object.

• process –

  Return the input unchanged for configuration-only objects.

Attributes:

• params (dict[str, Any]) –

  Expose the underlying parameter mapping.

Source code in src/rago/config.py

def __init__(self, backend: str = 'faiss', **kwargs: Any) -> None:
    backend_name = backend.lower()
    db: DBBase

    if backend_name == 'faiss':
        from rago.augmented.db.faiss import FaissDB

        db = FaissDB()
    elif backend_name == 'chroma':
        from rago.augmented.db.chroma import ChromaDB

        db = ChromaDB(**kwargs)
    else:
        raise ValueError(f'Unsupported DB backend: {backend}')

    super().__init__(db=db)

params property

params: dict[str, Any]

Expose the underlying parameter mapping.

apply

apply(parameters: Any) -> None

Merge additional configuration into this object.

Source code in src/rago/base.py

def apply(self, parameters: Any) -> None:
    """Merge additional configuration into this object."""
    self.data.update(config_to_dict(parameters))

process

process(inp: Input) -> Output

Return the input unchanged for configuration-only objects.

Source code in src/rago/base.py

def process(self, inp: Input) -> Output:
    """Return the input unchanged for configuration-only objects."""
    return inp.to_output()

Logs

Logs(target: dict[str, Any] | None = None)

Bases: Logs

Public alias for attaching logs declaratively.

Methods:

• apply –

  Merge additional configuration into this object.

• process –

  Return the input unchanged for configuration-only objects.

Attributes:

• params (dict[str, Any]) –

  Expose the underlying parameter mapping.

Source code in src/rago/extensions/logs.py

def __init__(self, target: dict[str, Any] | None = None) -> None:
    super().__init__(logs=target if target is not None else {})

params property

params: dict[str, Any]

Expose the underlying parameter mapping.

apply

apply(parameters: Any) -> None

Merge additional configuration into this object.

Source code in src/rago/base.py

def apply(self, parameters: Any) -> None:
    """Merge additional configuration into this object."""
    self.data.update(config_to_dict(parameters))

process

process(inp: Input) -> Output

Return the input unchanged for configuration-only objects.

Source code in src/rago/base.py

def process(self, inp: Input) -> Output:
    """Return the input unchanged for configuration-only objects."""
    return inp.to_output()