RAGchain.retrieval package

Submodules

RAGchain.retrieval.base module

class RAGchain.retrieval.base.BaseRetrieval

Bases: Runnable[str, RetrievalResult], ABC

Base Retrieval class for all retrieval classes.

property InputType: type[Input]

The type of input this runnable accepts specified as a type annotation.

property OutputType: type[Output]

The type of output this runnable produces specified as a type annotation.

as_ingest()

static create_db(db_type: str, db_path: dict) → BaseDB

You must implement this when you create new DB class. selector-ModuleSelector cant import because of circular import.

abstract delete(ids: List[str | UUID])

delete passages from vector representation of passages by ids.

static duplicate_check(db_origin_list: list[dict]) → dict[tuple, list[int]]

Check duplicated db origin in one retrieval. For example, db_origin = {“db_type”: “mongo_db”,

“db_path”: {“mongo_url”: “…”, “db_name”: “…”, “collection_name”: “…”}}

result = {((“db_type”: “mongo_db”),

(‘db_path’,((‘mongo_url’: “…”), (‘db_name’: “…”), (‘collection_name’: “…”)))): [0, 2], …}

fetch_data(ids: List[str | UUID]) → List[Passage]

fetch passages from each db. This can fetch data from multiple db. :param ids: list of passage ids

fetch_data_from_db_origin(ids: List[str | UUID], db_origin: dict, target_ids: List[int]) → List[Passage]

fetch_each_db(final_db_origin: dict[tuple, list[int]], ids: List[str | UUID]) → List[Passage]

check_dict = {((“db_type”: “mongo_db”),

((‘mongo_url’: “~”), (‘db_name’: “~”), (‘collection_name’: “~”))): [0, 2], …}

abstract ingest(passages: List[Passage])

ingest passages to vector representation of passages.

invoke(input: Input, config: RunnableConfig | None = None) → Output

retrieve passages from user query. :param input: user query. str type. :param config: RunnableConfig. Default is None. You can set top_k option in config. config[‘configurable’] key is “retrieval_options”. You can put like this. Example:

runnable.invoke(“your query”, config={“configurable”: {“retrieval_options”: {“top_k”: 10}}})

is_created(db_type: str, db_path: dict)

abstract retrieve(query: str, top_k: int = 5) → List[Passage]

retrieve passages at ingested vector representation of passages.

abstract retrieve_id(query: str, top_k: int = 5) → List[str | UUID]

retrieve passage ids at ingested vector representation of passages.

abstract retrieve_id_with_scores(query: str, top_k: int = 5) → tuple[List[str | UUID], List[float]]

retrieve passage ids and similarity scores at ingested vector representation of passages.

retrieve_with_filter(query: str, top_k: int = 5, content: List[str] | None = None, filepath: List[str] | None = None, content_datetime_range: List[tuple[datetime, datetime]] | None = None, importance: List[int] | None = None, multi_num: int = 2, retrieve_range_mult: int = 8, max_trial: int = 5, **kwargs)

retrieve passages which matches filter_dict conditions. :param query: query string :param top_k: passages count to retrieve :param content: content list to filter :param filepath: filepath list to filter :param content_datetime_range: content_datetime_range list to filter :param importance: importance list to filter :param kwargs: metadata_etc to filter :param multi_num: multiplier when failed to retrieve enough passages :param retrieve_range_mult: multiplier for retrieve range :param max_trial: max trial count for retrieve

search_data(ids: List[str | UUID], content: List[str] | None = None, filepath: List[str] | None = None, content_datetime_range: List[tuple[datetime, datetime]] | None = None, importance: List[int] | None = None, **kwargs) → List[Passage]

search passages from each db with given filters. This can search data from multiple db. :param ids: list of passage ids :param content: content list to filter :param filepath: filepath list to filter :param content_datetime_range: content_datetime_range list to filter :param importance: importance list to filter :param kwargs: metadata_etc to filter. Put metadata_etc key as kwargs key and metadata_etc value as kwargs value.

search_data_from_db_origin(ids: List[str | UUID], db_origin: dict, target_ids: List[int], content: List[str] | None = None, filepath: List[str] | None = None, content_datetime_range: List[tuple[datetime, datetime]] | None = None, importance: List[int] | None = None, **kwargs)

search_each_db(final_db_origin: dict[tuple, list[int]], ids: List[str | UUID], content: List[str] | None = None, filepath: List[str] | None = None, content_datetime_range: List[tuple[datetime, datetime]] | None = None, importance: List[int] | None = None, **kwargs) → List[Passage]

class RAGchain.retrieval.base.RunnableRetrievalIngest(retrieval: BaseRetrieval)

Bases: Runnable[List[Passage], List[Passage]]

property InputType: type[Input]

The type of input this runnable accepts specified as a type annotation.

property OutputType: type[Output]

The type of output this runnable produces specified as a type annotation.

invoke(input: Input, config: RunnableConfig | None = None) → Output

Transform a single input into an output. Override to implement.

Args:

input: The input to the runnable. config: A config to use when invoking the runnable.

The config supports standard keys like ‘tags’, ‘metadata’ for tracing purposes, ‘max_concurrency’ for controlling how much work to do in parallel, and other keys. Please refer to the RunnableConfig for more details.

Returns:

The output of the runnable.

RAGchain.retrieval.bm25_retrieval module

class RAGchain.retrieval.bm25_retrieval.BM25Retrieval(save_path: str, tokenizer_name: str = 'gpt2')

Bases: BaseRetrieval

BM25Retrieval class for BM25 retrieval. Save bm25 representation as pkl file and retrieve it. Default data structure looks like this: {

“tokens” : [], # 2d list of tokens “passage_id” : [], # 2d list of passage_id. Type must be UUID or string.

}

delete(ids: List[str | UUID])

delete passages from vector representation of passages by ids.

ingest(passages: List[Passage])

ingest passages to vector representation of passages.

static load_data(save_path: str)

persist(save_path: str)

Persist data to save_path as pickle file.

retrieve(query: str, top_k: int = 5) → List[Passage]

retrieve passages at ingested vector representation of passages.

retrieve_id(query: str, top_k: int = 5) → List[str | UUID]

retrieve passage ids at ingested vector representation of passages.

retrieve_id_with_scores(query: str, top_k: int = 5) → tuple[List[str | UUID], List[float]]

retrieve passage ids and similarity scores at ingested vector representation of passages.

RAGchain.retrieval.hybrid module

class RAGchain.retrieval.hybrid.HybridRetrieval(retrievals: List[BaseRetrieval], weights: List[float] | None = None, p: int = 500, method: str = 'cc', rrf_k: int = 60)

Bases: BaseRetrieval

Hybrid Retrieval class for retrieve passages from multiple retrievals. You can combine retrieval scores with rrf algorithm or convex combination algorithm.

delete(ids: List[str | UUID])

delete passages from vector representation of passages by ids.

ingest(passages: List[Passage])

ingest passages to vector representation of passages.

static min_max_normalization(arr: ndarray)

retrieve(query: str, top_k: int = 5) → List[Passage]

retrieve passages at ingested vector representation of passages.

retrieve_id(query: str, top_k: int = 5) → List[str | UUID]

retrieve passage ids at ingested vector representation of passages.

retrieve_id_with_scores(query: str, top_k: int = 5) → tuple[List[str | UUID], List[float]]

retrieve passage ids and similarity scores at ingested vector representation of passages.

retrieve_id_with_scores_parallel(retrieval: BaseRetrieval, query: str, top_k: int) → Series

RAGchain.retrieval.hyde module

class RAGchain.retrieval.hyde.HyDERetrieval(retrieval: BaseRetrieval, llm: BaseLanguageModel, system_prompt: str | None = None)

Bases: BaseRetrieval

HyDE Retrieval, which inspired by “Precise Zero-shot Dense Retrieval without Relevance Labels” (https://arxiv.org/pdf/2212.10496.pdf) At retrieval, LLM model creates hypothetical passage. And then, retrieve passages using hypothetical passage as query.

BASIC_SYSTEM_PROMPT = 'Please write a passage to answer the question'

delete(ids: List[str | UUID])

delete passages from vector representation of passages by ids.

ingest(passages: List[Passage])

ingest passages to vector representation of passages.

retrieve(query: str, top_k: int = 5, *args, **kwargs) → List[Passage]

retrieve passages at ingested vector representation of passages.

retrieve_id(query: str, top_k: int = 5, *args, **kwargs) → List[str | UUID]

retrieve passage ids at ingested vector representation of passages.

retrieve_id_with_scores(query: str, top_k: int = 5, *args, **kwargs) → tuple[List[str | UUID], List[float]]

retrieve passage ids and similarity scores at ingested vector representation of passages.

RAGchain.retrieval.vectordb_retrieval module

class RAGchain.retrieval.vectordb_retrieval.VectorDBRetrieval(vectordb: VectorStore)

Bases: BaseRetrieval

VectorDBRetrieval is a retrieval class that uses VectorDB as a backend. First, embed the passage content using an embedding model. Then, store the embedded vector in VectorDB. When retrieving, embed the query and search the most similar vectors in VectorDB. Lastly, return the passages that have the most similar vectors.

delete(ids: List[str | UUID])

delete passages from vector representation of passages by ids.

ingest(passages: List[Passage])

ingest passages to vector representation of passages.

retrieve(query: str, top_k: int = 5) → List[Passage]

retrieve passages at ingested vector representation of passages.

retrieve_id(query: str, top_k: int = 5) → List[str | UUID]

retrieve passage ids at ingested vector representation of passages.

retrieve_id_with_scores(query: str, top_k: int = 5) → tuple[List[str | UUID], List[float]]

retrieve passage ids and similarity scores at ingested vector representation of passages.

Module contents