Bases: BaseReader
awadb阅读器。
通过现有的awadb客户端检索文档。
然后可以在下游的LlamaIndex数据结构中使用这些文档。
Parameters:
| Name |
Type |
Description |
Default |
client |
client
|
|
required
|
Source code in llama_index/readers/awadb/base.py
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62 | class AwadbReader(BaseReader):
"""awadb阅读器。
通过现有的awadb客户端检索文档。
然后可以在下游的LlamaIndex数据结构中使用这些文档。
Args:
client (awadb.client): 一个awadb客户端。"""
def __init__(self, client: Any):
"""使用参数进行初始化。"""
import_err_msg = "`awadb` package not found, please run `pip install awadb`"
try:
pass
except ImportError:
raise ImportError(import_err_msg)
self.awadb_client = client
def load_data(
self,
query: np.ndarray,
k: int = 4,
separate_documents: bool = True,
) -> List[Document]:
"""从Faiss加载数据。
Args:
query（np.ndarray）：查询向量的二维numpy数组。
k（int）：要检索的最近邻居的数量。默认为4。
separate_documents（Optional[bool]）：是否返回单独的文档。默认为True。
Returns:
List[Document]：文档列表。
"""
results = self.awadb_client.Search(
query,
k,
text_in_page_content=None,
meta_filter=None,
not_include_fields=None,
)
documents = []
for item_detail in results[0]["ResultItems"]:
documents.append(Document(text=item_detail["embedding_text"]))
if not separate_documents:
# join all documents into one
text_list = [doc.get_content() for doc in documents]
text = "\n\n".join(text_list)
documents = [Document(text=text)]
return documents
|
load_data
load_data(
query: ndarray,
k: int = 4,
separate_documents: bool = True,
) -> List[Document]
从Faiss加载数据。
Returns:
Source code in llama_index/readers/awadb/base.py
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62 | def load_data(
self,
query: np.ndarray,
k: int = 4,
separate_documents: bool = True,
) -> List[Document]:
"""从Faiss加载数据。
Args:
query（np.ndarray）：查询向量的二维numpy数组。
k（int）：要检索的最近邻居的数量。默认为4。
separate_documents（Optional[bool]）：是否返回单独的文档。默认为True。
Returns:
List[Document]：文档列表。
"""
results = self.awadb_client.Search(
query,
k,
text_in_page_content=None,
meta_filter=None,
not_include_fields=None,
)
documents = []
for item_detail in results[0]["ResultItems"]:
documents.append(Document(text=item_detail["embedding_text"]))
if not separate_documents:
# join all documents into one
text_list = [doc.get_content() for doc in documents]
text = "\n\n".join(text_list)
documents = [Document(text=text)]
return documents
|