base

Base classes for composable Rago pipelines.

Classes:

• ParametersBase –

  Base class for declarative step configuration.

• Pipeline –

  Composable pipeline of arbitrary Rago steps.

• PipelineBase –

  Base pipeline class that can be built declaratively.

• StepBase –

  Abstract base for all pipeline steps.

Functions:

• config_to_dict –

  Normalize step configuration objects into a plain dictionary.

• ensure_list –

  Normalize scalar or iterable values into a list.

ParametersBase

ParametersBase(**kwargs: Any)

Bases: UserDict[str, Any]

Base class for declarative 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/base.py

def __init__(self, **kwargs: Any) -> None:
    super().__init__(kwargs)

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()

Pipeline

Pipeline()

Bases: PipelineBase

Composable pipeline of arbitrary Rago steps.

Methods:

• process –

  Run this pipeline when embedded as a step.

• prompt –

  Run the pipeline and return the primary result value.

• run –

  Run all configured steps for the given query and source.

Source code in src/rago/base.py

def __init__(self) -> None:
    self.stack: list[StepBase] = []
    self.global_params: list[Any] = []

process

process(inp: Input) -> Output

Run this pipeline when embedded as a step.

Source code in src/rago/base.py

def process(self, inp: Input) -> Output:
    """Run this pipeline when embedded as a step."""
    source = inp.get('source')
    data = inp.get('data', inp.get('content', source))
    extra = {
        key: value
        for key, value in inp.items()
        if key not in {'content', 'data', 'query', 'source'}
    }
    return self.run(inp.query, source=source, data=data, **extra)

prompt

prompt(
    query: str,
    source: Any = None,
    data: Any = None,
    **kwargs: Any,
) -> Any

Run the pipeline and return the primary result value.

Source code in src/rago/base.py

def prompt(
    self,
    query: str,
    source: Any = None,
    data: Any = None,
    **kwargs: Any,
) -> Any:
    """Run the pipeline and return the primary result value."""
    output = self.run(query=query, source=source, data=data, **kwargs)
    if 'result' in output:
        return output.result
    if 'content' in output:
        return output.content
    return output

run

run(
    query: str = '',
    source: Any = None,
    data: Any = None,
    **kwargs: Any,
) -> Output

Run all configured steps for the given query and source.

Source code in src/rago/base.py

def run(
    self,
    query: str = '',
    source: Any = None,
    data: Any = None,
    **kwargs: Any,
) -> Output:
    """Run all configured steps for the given query and source."""
    if source is None and data is not None:
        source = data

    content = kwargs.pop('content', None)
    if content is None:
        content = data if data is not None else source

    cur_output = Output(
        query=query,
        source=source,
        data=content,
        content=content,
        **kwargs,
    )

    for step in self.stack:
        cur_input = cur_output.as_input()
        cur_input.query = query
        cur_output = step.process(cur_input)

    return cur_output

PipelineBase

PipelineBase()

Bases: ABC

Base pipeline class that can be built declaratively.

Methods:

• process –

  Run this pipeline when embedded as a step.

• run –

  Run the pipeline.

Source code in src/rago/base.py

def __init__(self) -> None:
    self.stack: list[StepBase] = []
    self.global_params: list[Any] = []

process

process(inp: Input) -> Output

Run this pipeline when embedded as a step.

Source code in src/rago/base.py

def process(self, inp: Input) -> Output:
    """Run this pipeline when embedded as a step."""
    source = inp.get('source')
    data = inp.get('data', inp.get('content', source))
    extra = {
        key: value
        for key, value in inp.items()
        if key not in {'content', 'data', 'query', 'source'}
    }
    return self.run(inp.query, source=source, data=data, **extra)

run abstractmethod

run(
    query: str = '',
    source: Any = None,
    data: Any = None,
    **kwargs: Any,
) -> Output

Run the pipeline.

Source code in src/rago/base.py

@abstractmethod
def run(
    self,
    query: str = '',
    source: Any = None,
    data: Any = None,
    **kwargs: Any,
) -> Output:
    """Run the pipeline."""

StepBase

StepBase()

Bases: ABC

Abstract base for all pipeline steps.

Methods:

• apply –

  Apply attached configuration to the step.

• process –

  Process the current pipeline input.

Source code in src/rago/base.py

def __init__(self) -> None:
    self.cache: Any = None
    self.logs: dict[str, Any] = {}

apply

apply(parameters: Any) -> None

Apply attached configuration to the step.

Source code in src/rago/base.py

def apply(self, parameters: Any) -> None:
    """Apply attached configuration to the step."""
    if parameters is None:
        return

    if _is_cache_backend(parameters):
        self.cache = parameters
        return

    if _is_vector_db(parameters):
        setattr(self, 'db', parameters)
        return

    if _is_text_splitter(parameters):
        setattr(self, 'splitter', parameters)
        return

    for key, value in config_to_dict(parameters).items():
        if key == 'cache':
            self.cache = value
        elif key == 'logs':
            self.logs = value if value is not None else {}
        else:
            setattr(self, key, value)

process abstractmethod

process(inp: Input) -> Output

Process the current pipeline input.

Source code in src/rago/base.py

@abstractmethod
def process(self, inp: Input) -> Output:
    """Process the current pipeline input."""

config_to_dict

config_to_dict(parameters: Any) -> dict[str, Any]

Normalize step configuration objects into a plain dictionary.

Source code in src/rago/base.py

def config_to_dict(parameters: Any) -> dict[str, Any]:
    """Normalize step configuration objects into a plain dictionary."""
    if parameters is None:
        return {}
    if isinstance(parameters, ParametersBase):
        return dict(parameters.params)
    if isinstance(parameters, Mapping):
        return dict(parameters)
    return {}

ensure_list

ensure_list(value: Any) -> list[Any]

Normalize scalar or iterable values into a list.

Source code in src/rago/base.py

def ensure_list(value: Any) -> list[Any]:
    """Normalize scalar or iterable values into a list."""
    if value is None:
        return []
    if isinstance(value, list):
        return value
    if isinstance(value, tuple):
        return list(value)
    if isinstance(value, str):
        return [value]
    if isinstance(value, Iterable):
        return list(value)
    return [value]