RAGchain.DB package

Submodules

RAGchain.DB.base module

class RAGchain.DB.base.BaseDB

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

Abstract class for using a database for passage contents.

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.

abstract create(*args, **kwargs)

Abstract method for creating a new database.

abstract create_or_load(*args, **kwargs)

Abstract method for creating a new database or loading existing database.

abstract property db_type: str

The type of the database. This attribute should be implemented by the subclasses.

abstract fetch(ids: List[UUID]) → List[Passage]

Abstract method for fetching passages from the database based on their passage IDs.

abstract get_db_origin() → DBOrigin

DBOrigin: Abstract method for retrieving DBOrigin of the database.

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.

abstract load(*args, **kwargs)

Abstract method for loading existed database.

abstract save(passages: List[Passage], upsert: bool = False)

Abstract method for saving passages to the database.

abstract search(id: List[UUID | str] | None = None, 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 Passage from DB using filter Dict. This function can search Passage using ‘id’, ‘content’, ‘filepath’ and ‘metadata_etc’. You can add more search condition in metadata_etc using **kwargs. This function is an implicit AND operation.

Parameters:

  • id – (Optional[List[Union[UUID, str]]]): List of Passage ID to search.

  • content – (Optional[List[str]]): List of Passage content to search.

  • filepath – (Optional[List[str]]): List of Passage filepath to search.

  • content_datetime_range – (Optional[List[tuple[datetime.datetime, datetime.datetime]]]): List of content_datetime range to search.

You can set start_time at the first element in tuple and end_time at the second element in tuple. Multiple time range search is possible, and the search condition is OR operation in time range. :param importance: (Optional[List[int]]): List of passages’ importance to search. **kwargs: Additional metadata to search.

RAGchain.DB.mongo_db module

class RAGchain.DB.mongo_db.MongoDB(mongo_url: str, db_name: str, collection_name: str, *args, **kwargs)

Bases: BaseDB

MongoDB class for using MongoDB as a database for passage contents.

create()

Creates the collection in the MongoDB database. Raises a ValueError if the collection already exists.

create_or_load()

Creates the collection if it does not exist, otherwise loads it.

property db_type: str

Returns the type of the database as a string.

fetch(ids: List[UUID]) → List[Passage]

Fetches the passages from MongoDB collection by their passage ids.

get_db_origin() → DBOrigin

Returns the DBOrigin object representing the MongoDB database.

load()

Loads the collection from the MongoDB database. Raises a ValueError if the collection does not exist.

save(passages: List[Passage], upsert: bool = False)

Saves the passages to MongoDB collection.

search(id: List[UUID | str] | None = None, 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 Passage from DB using filter Dict. This function can search Passage using ‘id’, ‘content’, ‘filepath’ and ‘metadata_etc’. You can add more search condition in metadata_etc using **kwargs. This function is an implicit AND operation.

Parameters:

  • id – (Optional[List[Union[UUID, str]]]): List of Passage ID to search.

  • content – (Optional[List[str]]): List of Passage content to search.

  • filepath – (Optional[List[str]]): List of Passage filepath to search.

  • content_datetime_range – (Optional[List[tuple[datetime.datetime, datetime.datetime]]]): List of content_datetime range to search.

You can set start_time at the first element in tuple and end_time at the second element in tuple. Multiple time range search is possible, and the search condition is OR operation in time range. :param importance: (Optional[List[int]]): List of passages’ importance to search. **kwargs: Additional metadata to search.

set_db()

RAGchain.DB.pickle_db module

class RAGchain.DB.pickle_db.PickleDB(save_path: str)

Bases: BaseDB

This DB stores passages in a pickle file format at your local disk.

create()

Creates a new pickle file for the database if it doesn’t exist.

create_or_load()

Creates a new pickle file if it doesn’t exist, otherwise loads the data from the existing file.

property db_type: str

Returns the type of the database as a string.

fetch(ids: List[UUID]) → List[Passage]

Retrieves the Passage objects from the database based on the given list of passage IDs.

get_db_origin() → DBOrigin

Returns a DBOrigin object that represents the origin of the database.

load()

Loads the data from the existing pickle file into the database.

save(passages: List[Passage], upsert: bool = False)

Saves the given list of Passage objects to the pickle database. It also saves the data to the Linker.

search(id: List[UUID | str] | None = None, 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 Passage from DB using filter Dict. This function can search Passage using ‘id’, ‘content’, ‘filepath’ and ‘metadata_etc’. You can add more search condition in metadata_etc using **kwargs. This function is an implicit AND operation.

Parameters:

  • id – (Optional[List[Union[UUID, str]]]): List of Passage ID to search.

  • content – (Optional[List[str]]): List of Passage content to search.

  • filepath – (Optional[List[str]]): List of Passage filepath to search.

  • content_datetime_range – (Optional[List[tuple[datetime.datetime, datetime.datetime]]]): List of content_datetime range to search.

You can set start_time at the first element in tuple and end_time at the second element in tuple. Multiple time range search is possible, and the search condition is OR operation in time range. :param importance: (Optional[List[int]]): List of passages’ importance to search. **kwargs: Additional metadata to search.

Module contents