core

Rago core pipeline.

Classes:

• Rago –

  Composable Rago pipeline.

Rago

Rago(
    retrieval: StepBase | None = None,
    augmented: StepBase | None = None,
    generation: StepBase | None = None,
)

Bases: Pipeline

Composable Rago pipeline.

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.

Attributes:

• logs (dict[str, dict[str, Any]]) –

  Expose step logs using stable public names when possible.

Source code in src/rago/core.py

def __init__(
    self,
    retrieval: StepBase | None = None,
    augmented: StepBase | None = None,
    generation: StepBase | None = None,
) -> None:
    super().__init__()
    for step in (retrieval, augmented, generation):
        if step is not None:
            self | step

logs property

logs: dict[str, dict[str, Any]]

Expose step logs using stable public names when possible.

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