昇腾社区首页
中文
注册

upload_files

功能描述

上传文档,将文档保存在知识库中,仅知识库管理员有权限操作。如果文档重复,可选择强制覆盖。文档数据将明文存储,请注意安全风险。当上传的文档数量超过知识库的max_file_count最大限制时会失败,如果某个文档添加失败会抛出异常。

函数原型

from mx_rag.knowledge import upload_files
def upload_files(knowledge, files, loader_mng, embed_func, force)

内部工作流程

参数说明

参数名

数据类型

可选/必选

说明

knowledge

KnowledgeDB

必选

知识库对象,数据类型参见KnowledgeDB类

files

List[str]

必选

文档路径列表,路径长度取值范围为[1, 1024],文件个数默认不超过1000个。文档路径不能为软链接且不允许存在".."。

loader_mng

LoaderMng

必选

提供文档解析函数的管理类对象,数据类型参见LoaderMng类

embed_func

Callable[[List[str]], List[List[float]]]或者dict

必选

嵌入函数,把文件信息转换为向量。

直接传入回调方法默认按稠密处理,即{'dense': Callable, 'sparse': None}

传入字典格式为{'dense': x, 'sparse': y},x和y分别为稠密和稀疏向量的回调函数,xy不能同时为None。支持同时入库稠密和稀疏向量。

force

bool

可选

是否强制覆盖旧数据,如果选择否,文档重复会抛出异常。默认为“False”

返回值说明

数据类型

说明

List[str]

添加到知识库中失败的文件列表。

调用示例

from mx_rag.embedding.local import TextEmbedding
from mx_rag.knowledge import KnowledgeStore, KnowledgeDB, upload_files, delete_files, FilesLoadInfo
from mx_rag.document import LoaderMng
from mx_rag.storage.document_store import SQLiteDocstore
from mx_rag.storage.vectorstore import MindFAISS
from langchain.text_splitter import RecursiveCharacterTextSplitter
from mx_rag.knowledge import upload_dir
from mx_rag.document.loader import DocxLoader, PdfLoader, ExcelLoader

loader_mng = LoaderMng()
loader_mng.register_loader(DocxLoader, [".docx"])
loader_mng.register_loader(PdfLoader, [".pdf"])
loader_mng.register_loader(ExcelLoader, [".xlsx"])
# loader_mng.register_loader(ImageLoader, [".png"])
loader_mng.register_splitter(RecursiveCharacterTextSplitter,
                             [".docx", ".pdf", ".xlsx"])
# 设置向量检索使用的NPU卡
dev = 0
# 加载embedding模型
emb = TextEmbedding("/path/to/model", dev_id=dev)
# 初始化向量数据库
vector_store = MindFAISS(x_dim=1024, devs=[dev], 
                         load_local_index="/path/to/index", auto_save=True)
# 初始化文档chunk 关系数据库
chunk_store = SQLiteDocstore(db_path="./sql.db")
# 初始化知识管理关系数据库
knowledge_store = KnowledgeStore(db_path="./sql.db")
#添加知识库及管理员
knowledge_store.add_knowledge(knowledge_name="test", user_id='Default', role='admin')
# 初始化知识管理
knowledge_db = KnowledgeDB(knowledge_store=knowledge_store, chunk_store=chunk_store, vector_store=vector_store,
                           knowledge_name="test", user_id='Default', white_paths=["/home/"])
# 上传领域知识文档
# 调用upload_files
upload_files(knowledge=knowledge_db, files=["/path/data/test.docx"], loader_mng=loader_mng,
             embed_func=emb.embed_documents, force=True)
# 上传领域知识文档目录
# 调用upload_dir
params = FilesLoadInfo(knowledge=knowledge_db, dir_path="/path/data/files", loader_mng=loader_mng,
                       embed_func=emb.embed_documents, force=True, load_image=False)
upload_dir(params=params)
# 调用delete_files
delete_files(knowledge_db, ["test.docx"])