dummy

In-memory retrieval implementations.

Classes:

• StringRet –

  Retrieval step for sources that are already plain text chunks.

StringRet

StringRet(
    source: Any = None,
    splitter: TextSplitterBase | None = None,
    api_key: str = '',
    api_params: dict[str, Any] | None = None,
    cache: Cache | None = None,
    logs: dict[str, Any] | None = None,
)

Bases: RetrievalBase

Retrieval step for sources that are already plain text chunks.

Methods:

• apply –

  Apply attached configuration to the step.

• get –

  Backward-compatible alias for retrieve.

• process –

  Resolve the content for downstream steps.

• retrieve –

  Return the provided string chunks unchanged.

Source code in src/rago/retrieval/base.py

def __init__(
    self,
    source: Any = None,
    splitter: TextSplitterBase | None = None,
    api_key: str = '',
    api_params: dict[str, Any] | None = None,
    cache: Cache | None = None,
    logs: dict[str, Any] | None = None,
) -> None:
    super().__init__()
    self.source = source
    self.splitter = splitter or LangChainTextSplitter(
        'RecursiveCharacterTextSplitter'
    )
    self.api_key = api_key
    self.api_params = api_params or {}
    self.cache = cache
    self.logs = logs if logs is not None else {}

    self._validate()
    self._setup()

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)

get

get(query: str = '', source: Any = None) -> list[str]

Backward-compatible alias for retrieve.

Source code in src/rago/retrieval/base.py

def get(self, query: str = '', source: Any = None) -> list[str]:
    """Backward-compatible alias for `retrieve`."""
    return self.retrieve(query=query, source=source)

process

process(inp: Input) -> Output

Resolve the content for downstream steps.

Source code in src/rago/retrieval/base.py

def process(self, inp: Input) -> Output:
    """Resolve the content for downstream steps."""
    source = self.source if self.source is not None else inp.get('source')
    if source is None:
        source = inp.get('content')

    result = ensure_list(self.retrieve(query=inp.query, source=source))
    output = Output.from_input(inp)
    output.content = result
    output.data = result
    return output

retrieve

retrieve(query: str = '', source: Any = None) -> list[str]

Return the provided string chunks unchanged.

Source code in src/rago/retrieval/dummy.py

def retrieve(self, query: str = '', source: Any = None) -> list[str]:
    """Return the provided string chunks unchanged."""
    del query
    value = self.source if source is None else source
    return cast(list[str], ensure_list(value))