db

LocalDB to perform in-memory storage and data persistence(pickle or any filesystem) for data models like documents and dialogturn.

Classes

LocalDB(name, items, transformed_items, ...)

LocalDB with in-memory CRUD operations, data transformation/processing pipelines, and persistence.

class LocalDB(name: str | None = None, items: ~typing.List[~core.db.T] = <factory>, transformed_items: ~typing.Dict[str, ~typing.List[~core.db.U]] = <factory>, transformer_setups: ~typing.Dict[str, ~adalflow.core.component.Component] = <factory>, mapper_setups: ~typing.Dict[str, ~typing.Callable[[~core.db.T], ~typing.Any]] = <factory>, index_path: str | None = 'index.faiss')

Bases: Generic[T], Component

LocalDB with in-memory CRUD operations, data transformation/processing pipelines, and persistence.

LocalDB is highly flexible. 1. It can store any type of data items in the items attribute. 2. You can register and apply multiple transformers, and save the transformed data in the transformed_items attribute.

This is highly useful to manage experiments with different data transformations.

3. You can save the state of the LocalDB to a pickle file and load it back later. All states are restored.

   str(local_db.__dict__) == str(local_db_loaded.__dict__) should be True.

Note

The transformer should be of type Component. We made the effort in the library to make every component picklable.

CRUD operations: 1. Create a new db: db = LocalDB(name="my_db") 2. load: Load the db with data. db.load([{"text": "hello world"}, {"text": "hello world2"}]) 3. extend: Extend the db with data. db.extend([{"text": "hello world3"}]).

In default, the transformer is applied and the transformed data is extended.

4. add: Add a single item to the db. db.add({"text": "hello world4"}). In default, the transformer is applied and the transformed data is added. Unless the transformed data keeps the same length as the original data, the insert operation does not mean insert after the last item.

5. delete: Remove items by index. db.delete([0]).

6. reset: Remove all items. db.reset(), including transformed_items and transformer_setups,and mapper_setups.

Data transformation: 1. Register a transformer first and apply it later

db.register_transformer(transformer, key="test", map_fn=map_fn)
# load data
db.load([{"text": "hello world"}, {"text": "hello world2"}], apply_transformer=True)

# or load data first and apply transformer by key
db.load([{"text": "hello world"}, {"text": "hello world2"}], apply_transformer=False)
db.apply_transformer("test")

2. Add a version of transformed data to the db along with the transformer.

db.transform(transformer, key="test", map_fn=map_fn)

Data persistence: 1. Save the state of the db to a pickle file.

db.save_state("storage/local_item_db.pkl")

2. Load the state of the db from a pickle file.

db2 = LocalDB.load_state("storage/local_item_db.pkl")

3. Check if the loaded and original db are the same.

str(db.__dict__) == str(db2.__dict__) # expect True

Parameters:

• items (List[T], optional) – The original data items. Defaults to []. Can be any type such as Document, DialogTurn, dict, text, etc. The only requirement is that they should be picklable/serializable.

• transformed_items (Dict[str, List [U]], optional) – Transformed data items by key. Defaults to {}. Transformer must be of type Component.

• transformer_setups (Dict[str, Component], optional) – Transformer setup by key. Defaults to {}. It is used to save the transformer setup for later use.

• mapper_setups (Dict[str, Callable[[T], Any]], optional) – Map function setup by key. Defaults to {}.

name: str | None = None

items: List[T]

transformed_items: Dict[str, List[U]]

transformer_setups: Dict[str, Component]

mapper_setups: Dict[str, Callable[[T], Any]]

index_path: str | None = 'index.faiss'

property length

get_transformer_keys() → List[str]

get_transformed_data(key: str, filter_fn: ~typing.Callable[[~typing.Any], bool] = <function LocalDB.<lambda>>) → List[U]

Get the transformed items by key after applying a filter on metadata.

Parameters:

• key (str) – The key to identify which transformed items to retrieve.

• filter_fn (Callable[[Any], bool], optional) – The filter function to apply on the metadata. Defaults to lambda x: True.

Returns:

The filtered and transformed items.

Return type:

List[U]

register_transformer(transformer: Component, key: str | None = None, map_fn: Callable[[T], Any] | None = None) → str

Register a transformer to be used later for transforming the data.

transform(key: str) → str

transform(transformer: Component, key: str | None = None, map_fn: Callable[[T], Any] | None = None) → str

The main method to apply the transformer to the data in two ways: 1. Apply the transformer by key to the data using transform(key="test"). 2. Register and apply the transformer to the data using transform(transformer, key="test").

Parameters:

• transformer (Optional[Component], optional) – The transformer to use. Defaults to None.

• key (Optional[str], optional) – The key to use for the transformer. Defaults to None.

• map_fn (Optional[Callable[[T], Any]], optional) – The map function to use. Defaults to None.

Returns:

The key used for the transformation, from which the transformed data can be accessed.

Return type:

str

load(items: List[Any])

Load the db with new items.

Parameters:

items (List[Any]) – The items to load.

Examples:

db = LocalDB()
db.load([{"text": "hello world"}, {"text": "hello world2"}])

extend(items: List[Any], apply_transformer: bool = True)

Extend the db with new items.

delete(index: int | None = None, remove_transformed: bool = True)

Remove items by index or pop the last item. Optionally remove the transformed data as well.

Assume the transformed item has the same index as the original item. Might not always be the case.

Parameters:

• index (Optional[int], optional) – The index to remove. Defaults to None.

• remove_transformed (bool, optional) – Whether to remove the transformed data as well. Defaults to True.

add(item: Any, index: int | None = None, apply_transformer: bool = True)

Add a single item by index or append to the end. Optionally apply the transformer.

Note

The item will be transformed using the registered transformer. Only if the transformed data keeps the same length as the original data, the insert operation will work correctly.

Parameters:

• item (Any) – The item to add.

• index (int, optional) – The index to add the item at. Defaults to None.

• None (When)

• end. (the item is appended to the)

• apply_transformer (bool, optional) – Whether to apply the transformer to the item. Defaults to True.

fetch_items(condition: Callable[[T], bool]) → List[T]

Fetch items with a condition.

fetch_transformed_items(key: str, condition: Callable[[U], bool]) → List[U]

Fetch transformed items with a condition.

reset()

Reset all attributes to empty.

save_state(filepath: str = None)

Save the current state (attributes) of the DB using pickle.

Note

The transformer setups will be lost when pickling. As it might not be picklable.

classmethod load_state(filepath: str = None) → LocalDB

Load the state of the DB from a pickle file.