Class Introduction

Function

Provides the sqlite knowledge base to store chunk information after splitting.

Prototype

from mx_rag.storage.document_store import SQLiteDocstore
SQLiteDocstore(db_path, encrypt_fn, decrypt_fn)

Parameters

Parameter	Data Type	Required/Optional	Description
db_path	String	Required	Storage path of a relational database. The path must be a valid path. The path can contain a maximum of 1024 characters. The file name cannot exceed 200 characters and cannot contain two consecutive dots (..). The storage path cannot be in the path list: ["/etc", "/usr/bin", "/usr/lib", "/usr/lib64", "/sys/", "/dev/", "/sbin", "/tmp"].
encrypt_fn	Callable[[str], str]	Optional	Callback method. The return value is a string and its length cannot exceed 128 × 1024 × 1024. It encrypts the chunk content of ChunkModel and outputs a string. When the add operation is performed, the data stored by the database are chunks processed by encrypt_fn. NOTICE: If the file to be uploaded contains personal data such as bank account numbers, ID card numbers, passport numbers, and passwords, set this parameter to ensure personal data security.
decrypt_fn	Callable[[str], str]	Optional	Callback method. The return value is a string and its length cannot exceed 16 × 1024 × 1024. It decrypts the chunk content of ChunkModel and outputs a string. After the search operation is performed, the returned data are chunks processed by decrypt_fn.

Example

from mx_rag.storage.document_store import MxDocument, SQLiteDocstore
def encrypt_fn(value):
    # Secure encryption method
    return value
def decrypt_fn(value):
    # Secure decryption method
    return value
chunk_store = SQLiteDocstore(db_path="./sql.db", encrypt_fn=encrypt_fn, decrypt_fn=decrypt_fn)
text = ["Example", "Text"]
metadata = [{} for _ in text]
doc = [MxDocument(page_content=t, metadata=m, document_name="1.docx") for t, m in zip(text, metadata)]
document_id = 1
chunk_store.add(doc, document_id)
idx = chunk_store.get_all_chunk_id()
document = chunk_store.search(idx[0])
print(document.page_content)
chunk_store.update(idx[:2], ["text1", "text2"])
print(chunk_store.delete(document_id))
chunk_store.search_by_document_id(document_id)

Parent topic: SQLiteDocstore