向量資料庫

發表於 更新於 分類於

dbs物件

在akasha中，chromadb建立完之後，會被儲存成dbs物件，會儲存chromadb中的文件內容、metadata、向量資料、unique id，並被使用於後續的vector similarity search。

該物件可以添加多個chromadb資料，也可與其他dbs物件互相結合，也可根據filter抽取出需要的向量資料。

建立向量資料

processMultiDB

processMultiDB可對多個文件集(list of directory)建立chromadb，並回傳dbs物件與建立不成功的檔案list，若文件內容、使用嵌入模型、chunk size相等的chromadb已存在，則不會重新創建而直接讀取。

文件內容改變也會建立新的chromadb，設定參數ignore_check=True則不進行文件內容更改與否的確認，可更快的進行chromadb的讀取。

1
2
3
4
5
6
7
8
9
10
11
import akasha
doc_paths = ["docs/pns_query", "docs/mic"]
emb_name = "openai:text-embedding-ada-002"
emb_obj = akasha.handle_embeddings(emb_name)
db, ignore_files = akasha.processMultiDB(doc_path_list=doc_paths,
    verbose=False,
    embeddings=emb_obj,
    chunk_size=1000,
    ignore_check=False)


createDB_file & createDB_directory

使用多個文字檔案創建chromadb可使用 createDB_file，使用資料夾創建chromadb則使用 createDB_directory
創建完的dbs物件(db_files, db_directory)可以直接輸入進 get_response中，減少重複讀取。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
import akasha
db_files = akasha.createDB_file(file_path = ["f1.txt","f2.docs"], embeddings="openai:text-embedding-ada-002",chunk_size=500, ignore_check=True)
db_directory = akasha.createDB_directory(doc_path= "./docs/mic/", 
embeddings="openai:text-embedding-ada-002", ignore_check=True)
qa = akasha.Doc_QA(
    verbose=True, 
    search_type="svm", 
    model="openai:gpt-3.5-turbo", 
    )

qa.get_response(
        doc_path=db_directory,
        prompt="五軸是甚麼?",
        system_prompt="請用中文回答",
)




根據關鍵字建立向量資料

在建立向量資料時，若您不想根據整個文件片段進行embedding，可以使用 akasha.db.create_keyword_chromadb，可自訂義函式來輸出想進行embedding的關鍵字，如以下範例
我們定義了一個函式 generate_llm_keyword 請語言模型根據文件段落產生3~5個關鍵字並回傳。

當我們使用 akasha.db.create_keyword_chromadb，建立向量資料(chromadb)，每個文件片段便會根據自訂義函式 generate_llm_keyword的輸出來進行embedding

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
import akasha

md_obj = akasha.handle_model("openai:gpt-4o", temperature=1)
emb_obj = akasha.handle_embeddings("openai:text-embedding-ada-002")
def generate_llm_keyword(text, keyword_num):

    prompt = f"Generate 3~{keyword_num} keywords for the following text: \n\n" + text + "\n\n use \, to separate each keyword"

    res = akasha.call_model(md_obj, prompt)
    return [res]


db, _ = akasha.db.create_keyword_chromadb(
    "docs/mic", emb_obj, 1000,
    keyword_function=generate_llm_keyword)

注意

若不使用自訂義函式，預設的函式 akasha.generate_keyword需安裝keybert套件

1
pip install keybert


使用dbs物件

init

您可以直接宣告akasha.dbs()建立空的dbs物件，也可以利用已建立的chromadb建立dbs物件。

dbs物件包含ids(每個文字段落的unique id), embeds(每個文字段落的向量), metadatas(每個文字段落的後設資料), docs(每個文字段落的內容) 。 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
import akasha
from langchain_chroma import Chroma


db1 = akasha.dbs()

### use chromadb to initialize dbs object ###
storage_directory = "chromadb/12345"
emb_obj = akasha.handle_embeddings()
docsearch = Chroma(persist_directory=storage_directory,
                           embedding_function=emb_obj)
db2 = akasha.dbs(docsearch)

print(len(db2.get_ids()))  # list[str]
print(len(db2.get_embeds())) # list[list[float]]
print(len(db2.get_metadatas())) #list[dict]
print(len(db2.get_docs())) #list[dict]

merge

dbs物件之間可以使用.merge相互結合

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
import akasha
from langchain_chroma import Chroma

emb_obj = akasha.handle_embeddings()

docsearch1 = Chroma(persist_directory="chromadb/123",
                           embedding_function=emb_obj)
db1 = akasha.dbs(docsearch1)

docsearch2 = Chroma(persist_directory="chromadb/456",
                           embedding_function=emb_obj)
db2 = akasha.dbs(docsearch2)

db2.merge(db1)

print(len(db2.get_ids()))  # list[str]
print(len(db2.get_embeds())) # list[list[float]]
print(len(db2.get_metadatas())) #list[dict]
print(len(db2.get_docs())) #list[dict]

add_chromadb

dbs物件可以添加新的chromadb資料

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
import akasha
from langchain_chroma import Chroma

emb_obj = akasha.handle_embeddings()

docsearch1 = Chroma(persist_directory="chromadb/123",
                           embedding_function=emb_obj)

docsearch2 = Chroma(persist_directory="chromadb/456",
                           embedding_function=emb_obj)

db = akasha.dbs(docsearch1)

db.add_chromadb(docsearch2)

print(len(db.get_ids()))  # list[str]
print(len(db.get_embeds())) # list[list[float]]
print(len(db.get_metadatas())) #list[dict]
print(len(db.get_docs())) #list[dict]

get_Documents

使用get_Docuemnts可以得到當前dbs物件中儲存的Documents list (包含page_contents文件內容和metadata後設資料)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
import akasha
from langchain_chroma import Chroma

emb_obj = akasha.handle_embeddings()

docsearch1 = Chroma(persist_directory="chromadb/123",
                           embedding_function=emb_obj)

db = akasha.dbs(docsearch1)

docs = db.get_Documents()



print([doc.page_contents for doc in docs])  # list[str]
print([docs.metadata for doc in docs]) # list[dict]



提取dbs物件

extract_db_by_file

extract_db_by_file可以將檔名符合file_name_list中的所有資料提取出來生成新的dbs物件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
import akasha
from langchain_chroma import Chroma

emb_obj = akasha.handle_embeddings()

docsearch1 = Chroma(persist_directory="chromadb/123",
                           embedding_function=emb_obj)

db = akasha.dbs(docsearch1)
file_name_list = ['f1.txt', 'f2.docx']

extracted_db = akasha.extract_db_by_file(db=db, file_name_list=file_name_list)

print(len(extracted_db.get_ids()))  # list[str]
print(len(extracted_db.get_embeds())) # list[list[float]]
print(len(extracted_db.get_metadatas())) #list[dict]
print(len(extracted_db.get_docs())) #list[dict]

extract_db_by_keyword

extract_db_by_keyword可以將文字段落中存在任何keyword_list中keyword的所有資料提取出來生成新的dbs物件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
import akasha
from langchain_chroma import Chroma

emb_obj = akasha.handle_embeddings()

docsearch1 = Chroma(persist_directory="chromadb/123",
                           embedding_function=emb_obj)

db = akasha.dbs(docsearch1)
keyword_list = ["資訊產業策進會", "AI人工智慧"]

extracted_db = akasha.extract_db_by_keyword(db=db, keyword_list=keyword_list)

print(len(extracted_db.get_ids()))  # list[str]
print(len(extracted_db.get_embeds())) # list[list[float]]
print(len(extracted_db.get_metadatas())) #list[dict]
print(len(extracted_db.get_docs())) #list[dict]

extract_db_by_ids

extract_db_by_ids可以將存在id_list中的的所有資料提取出來生成新的dbs物件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
import akasha
from langchain_chroma import Chroma

emb_obj = akasha.handle_embeddings()

docsearch1 = Chroma(persist_directory="chromadb/123",
                           embedding_function=emb_obj)

db = akasha.dbs(docsearch1)
id_list = ['2024-10-21-17_45_21_963065_0', '2024-10-21-17_45_23_601845_0']

extracted_db = akasha.extract_db_by_ids(db=db, id_list=id_list)

print(len(extracted_db.get_ids()))  # list[str]
print(len(extracted_db.get_embeds())) # list[list[float]]
print(len(extracted_db.get_metadatas())) #list[dict]
print(len(extracted_db.get_docs())) #list[dict]

pop_db_by_ids

pop_db_by_ids會將所選id_list中的的所有資料從dbs物件中移除

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
import akasha
from langchain_chroma import Chroma

emb_obj = akasha.handle_embeddings()

docsearch1 = Chroma(persist_directory="chromadb/123",
                           embedding_function=emb_obj)

db = akasha.dbs(docsearch1)
id_list = ['2024-10-21-17_45_21_963065_0', '2024-10-21-17_45_23_601845_0']

akasha.pop_db_by_ids(db=db, id_list=id_list)

print(len(db.get_ids()))  # list[str]
print(len(db.get_embeds())) # list[list[float]]
print(len(db.get_metadatas())) #list[dict]
print(len(db.get_docs())) #list[dict]



其他輔助函式

get_docs_from_doc

若不需要向量資料，只需要Documents(page_content, metadata)，可使用get_docs_from_doc從文件資料夾中讀取文件內容並切割成chunk_size文件段落，此函式會回傳docs(list of Documents)和不成功的檔案名稱list。 

1
2
3
4
5
6
7
8
9
10
11
12
13
import akasha


docs, ignore_files = akasha.get_docs_from_doc(doc_path="docs/mic", chunk_size=1000, ignore_check=True)

db = akasha.dbs(docsearch1)
id_list = ['2024-10-21-17_45_21_963065_0', '2024-10-21-17_45_23_601845_0']

extracted_db = akasha.extract_db_by_ids(db=db, id_list=id_list)

print([doc.page_contents for doc in docs])  # list[str]
print([docs.metadata for doc in docs]) # list[dict]

get_db_metadata

根據文件資料夾、嵌入模型、chunk size輸出所有文件段落的後設資料list，用於自查詢

update_db_metadta

將更新完的metadata list存回chromadb，用於自查詢

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

import akasha



def add_metadata(metadata_list: list):

    import json
    for metadata in metadata_list:
        metadata['testing']= '後設資料新增測試'


### set parameter ###
dir_path = "docs/pns_query"
embed_name = "openai:text-embedding-ada-002"
chunk_size = 99999  # make sure 1 file 1 chunk
emb_obj = akasha.handle_embeddings("openai:text-embedding-ada-002")
####################

# create chromadb from docs
db, _ = akasha.db.processMultiDB(dir_path, False, emb_obj, chunk_size, True)


### add metadata to chromadb ###
metadata_list = akasha.db.get_db_metadata(dir_path, embed_name, chunk_size) # get original metada from chromadb, list of dictionary
metadata_list = add_metadata(metadata_list)  # update/add metadata, you can build your own function to update metadata
akasha.db.update_db_metadata(metadata_list, dir_path, embed_name, chunk_size)  # update and save new metadatas to chromadb
print(akasha.db.get_db_metadata(dir_path, embed_name, chunk_size)[0])