nat.retriever.nemo_retriever.retriever#

Attributes#

logger

Exceptions#

CollectionUnavailableError

Common base class for all non-exit exceptions.

Classes#

`Collection`
`RetrieverPayload`
`NemoRetriever`	Client for retrieving document chunks from a Nemo Retriever service.
`NemoLangchainRetriever`	Abstract base class for a document retrieval system.

Functions#

`_wrap_nemo_results`(output, content_field)
`_wrap_nemo_single_results`(output, content_field)
`_flatten`(→ list[str])

Module Contents#

logger#

class Collection(/, **data: Any)#

Bases: pydantic.BaseModel

id: str#

name: str#

meta: Any#

pipeline: str#

created_at: str#

class RetrieverPayload(/, **data: Any)#

Bases: pydantic.BaseModel

query: str#

top_k: int = None#

exception CollectionUnavailableError#

Bases: nat.retriever.models.RetrieverError

Common base class for all non-exit exceptions.

Initialize self. See help(type(self)) for accurate signature.

class NemoRetriever( uri: str | pydantic.HttpUrl, timeout: int = 60, nvidia_api_key: str = None, \*\*kwargs, )#

Bases: nat.retriever.interface.Retriever

Client for retrieving document chunks from a Nemo Retriever service.

base_url = ''#

timeout = 60#

_search_func#

api_key#

_bound_params = []#

bind(\*\*kwargs) → None#

Bind default values to the search method. Cannot bind the query parameter.

Args:: kwargs: Key value pairs corresponding to the default values of search parameters.

get_unbound_params() → list[str]#: Returns a list of unbound parameters which will need to be passed to the search function.

async get_collections(client) → list[Collection]#: Get a list of all available collections as pydantic Collection objects.

async get_collection_by_name(collection_name, client) → Collection#: Retrieve a collection using its name. Will return the first collection found if the name is ambiguous.

async search(query: str, \*\*kwargs)#: Retireve max(top_k) items from the data store based on vector similarity search (implementation dependent).

async _search( query: str, collection_name: str, top_k: str, output_fields: list[str] = None, )#: Retrieve document chunks from the configured Nemo Retriever Service.

_wrap_nemo_results(output: list[dict], content_field: str)#

_wrap_nemo_single_results(output: dict, content_field: str)#

_flatten(obj: dict, output_fields: list[str]) → list[str]#

class NemoLangchainRetriever(/, **data: Any)#

Bases: langchain_core.retrievers.BaseRetriever, pydantic.BaseModel

Abstract base class for a document retrieval system.

A retrieval system is defined as something that can take string queries and return the most ‘relevant’ documents from some source.

Usage:

A retriever follows the standard Runnable interface, and should be used via the standard Runnable methods of invoke, ainvoke, batch, abatch.

Implementation:

When implementing a custom retriever, the class should implement the _get_relevant_documents method to define the logic for retrieving documents.

Optionally, an async native implementations can be provided by overriding the _aget_relevant_documents method. validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

client: NemoRetriever#

abstractmethod _get_relevant_documents(query, *, run_manager, \*\*kwargs)#

Get documents relevant to a query.

Args:: query: String to find relevant documents for. run_manager: The callback handler to use.
Returns:: List of relevant documents.

async _aget_relevant_documents(query, *, run_manager, \*\*kwargs)#

Asynchronously get documents relevant to a query.

Args:: query: String to find relevant documents for run_manager: The callback handler to use
Returns:: List of relevant documents