前沿重器[46] RAG开源项目Qanything源码阅读2-离线文件处理

前沿重器

栏目主要给大家分享各种大厂、顶会的论文和分享，从中抽取关键精华的部分和大家分享，和大家一起把握前沿技术。具体介绍：仓颉专项：飞机大炮我都会，利器心法我还有。（算起来，专项启动已经是20年的事了！）

2023年文章合集发布了！在这里：又添十万字-CS的陋室2023年文章合集来袭

往期回顾

• 前沿重器[41] | 综述-面向大模型的检索增强生成（RAG）
• 前沿重器[42] | self-RAG-大模型决策的典型案例探究
• 前沿重器[43] | 谷歌中科院新文：CRAG-可矫正的检索增强生成
• 前沿重器[44] | Adaptive-RAG:根据难度自适应检索方案
• 前沿重器[45] RAG开源项目Qanything源码阅读1-概述+服务

书接上文，最近选了一个开源的RAG项目进行进一步学习：https://github.com/netease-youdao/QAnything，后续一连几篇，会分几篇，从我的角度，给大家介绍这个项目，预计的目录如下：

• 概述+服务：项目设计、模块划分以及部署细节。
• 文件解析处理：上传文件和文件处理的方法。（本期）
• 在线推理流程：给定query进入后到给出回复结果的全流程处理。

本期是离线的文件处理，即对多种不同的文件进行详细的阐述，之前我也有写过类似的文章（心法利器[110] | 知识文档处理和使用流程），不过没落到代码层面，这次借着源码阅读的机会，正好介绍一下：

• 文件上传。
• 文件读取和切片。
• 索引构造。

提前说明，这里忽略了大量的业务代码，聚焦在文件处理和相关算法本身，如新建用户、知识库、文件删除，会有选择的忽略，有需要的可以参考我在文中的思路，在代码里找到对应的位置。

文件上传

文件上传是指将文件从前端传到后端的流程，这个流程的工作在docs\API.md有提到。首先是接口字段：

参数名	参数值	是否必填	参数类型	描述说明
files	文件二进制	是	File	需要上传的文件，可多选，目前仅支持[md,txt,pdf,jpg,png,jpeg,docx,xlsx,pptx,eml,csv]
user_id	zzp	是	String	用户 id
kb_id	KBb1dd58e8485443ce81166d24f6febda7	是	String	知识库 id
mode	soft	否	String	上传模式，soft：知识库内存在同名文件时当前文件不再上传，strong：文件名重复的文件强制上传，默认值为 soft

至于文件的上传，作者给出了两种模式，分别是同步和异步。

客户端

客户端只需要请求服务即可，这里穿插一下同步异步请求，以及文件上传的细节，这个直接参考源码就好了，首先是同步的请求源码：

          
            
import os  
import requests  
  
url = "http://{your\_host}:8777/api/local\_doc\_qa/upload\_files"  
folder_path = "./docx\_data"  # 文件所在文件夹，注意是文件夹！！  
data = {  
    "user\_id": "zzp",  
    "kb\_id": "KB6dae785cdd5d47a997e890521acbe1c9",  
 "mode": "soft"  
}  
  
files = []  
for root, dirs, file_names in os.walk(folder_path):  
    for file_name in file_names:  
        if file_name.endswith(".md"):  # 这里只上传后缀是md的文件，请按需修改，支持类型：  
            file_path = os.path.join(root, file_name)  
            files.append(("files", open(file_path, "rb")))  
  
response = requests.post(url, files=files, data=data)  
print(response.text)  

        

• 发请求用的是通用的requests包。
• 因为是本地测试，所以使用的就是比较直接的本地文件，直接open就行，文件字段存的是open变量，注意打开方式是rb。

至于异步，则会会复杂一些。

          
            
import argparse  
import os  
import sys  
import json  
import aiohttp  
import asyncio  
import time  
import random  
import string  
  
files = []  
for root, dirs, file_names in os.walk("./docx\_data"):  # 文件夹  
    for file_name in file_names:  
        if file_name.endswith(".docx"):  # 只上传docx文件  
            file_path = os.path.join(root, file_name)  
            files.append(file_path)  
print(len(files))  
response_times = []  
  
async def send\_request(round\_, files):  
    print(len(files))  
    url = 'http://{your\_host}:8777/api/local\_doc\_qa/upload\_files'  
    data = aiohttp.FormData()  
    data.add_field('user\_id', 'zzp')  
    data.add_field('kb\_id', 'KBf1dafefdb08742f89530acb7e9ed66dd')  
    data.add_field('mode', 'soft')  
  
    total_size = 0  
    for file_path in files:  
        file_size = os.path.getsize(file_path)  
        total_size += file_size  
        data.add_field('files', open(file_path, 'rb'))  
    print('size:', total_size / (1024 * 1024))  
    try:  
        start_time = time.time()  
        async with aiohttp.ClientSession() as session:  
            async with session.post(url, data=data) as response:  
                end_time = time.time()  
                response_times.append(end_time - start_time)  
                print(f"round\_:{round\_}, 响应状态码: {response.status}, 响应时间: {end\_time - start\_time}秒")  
    except Exception as e:  
        print(f"请求发送失败: {e}")  
  
async def main():  
    start_time = time.time()  
    num = int(sys.argv[1])  // 一次上传数量，http协议限制一次请求data不能大于100M，请自行控制数量  
    round_ = 0  
    r_files = files[:num]  
    tasks = []  
    task = asyncio.create_task(send_request(round_, r_files))  
    tasks.append(task)  
    await asyncio.gather(*tasks)  
  
    print(f"请求完成")  
    end_time = time.time()  
    total_requests = len(response_times)  
    total_time = end_time - start_time  
    qps = total_requests / total_time  
    print(f"total\_time:{total\_time}")  
  
if __name__ == '\_\_main\_\_':  
    asyncio.run(main())  

        

请求用的是aiohttp，而且使用的是python的协程，即asyncio一套的python技术，具体细节可以参考这篇博客：https://blog.csdn.net/m0\_68949064/article/details/132805165。协程在高密度的http请求下，能有效提升CPU的使用率，提升综合性能，毕竟在请求等待过程，可以做很多别的事，就避免CPU空跑了。

服务端

服务端则比较复杂了，文件上传后要经过大量的校验，并且需要返回最终的处理结果。

文件上传的接口是/api/local_doc_qa/upload_files，我们可以在handlers.py里面找到，排除掉一些校验代码，handlers里面的核心代码是这段（upload_files函数下）：

          
            
for file, file_name in zip(files, file_names):  
    if file_name in exist_file_names:  
        continue  
    file_id, msg = local_doc_qa.milvus_summary.add_file(user_id, kb_id, file_name, timestamp)  
    debug_logger.info(f"{file\_name}, {file\_id}, {msg}")  
    local_file = LocalFile(user_id, kb_id, file, file_id, file_name, local_doc_qa.embeddings)  
    local_files.append(local_file)  
    local_doc_qa.milvus_summary.update_file_size(file_id, len(local_file.file_content))  
    data.append(  
        {"file\_id": file_id, "file\_name": file_name, "status": "gray", "bytes": len(local_file.file_content),  
            "timestamp": timestamp})  
asyncio.create_task(local_doc_qa.insert_files_to_milvus(user_id, kb_id, local_files))  

        

这里面的几个关键的函数：

• local_doc_qa.milvus_summary.add_file：向指定知识库下面增加文件，这是一个mysql操作，要在mysql数据库内记录在案。
• local_doc_qa.insert_files_to_milvus：将文档加入到milvus中，当然这里也包含了文件切片、推理向量、存入数据库等一系列操作。

回到服务，这里最终还是会收集各种处理的信息，最终以json形式形式返回，这里包括状态码、返回信息以及必要的数据信息（例如文件id、上传后的文件名、更新时间等）

          
            
return sanic_json({"code": 200, "msg": msg, "data": data})  

        

文件处理核心流程

继续往里面看，这个函数的代码不是很长，我直接放了：

          
            
async def insert\_files\_to\_milvus(self, user\_id, kb\_id, local\_files: List[LocalFile]):  
    debug_logger.info(f'insert\_files\_to\_milvus: {kb\_id}')  
    milvus_kv = self.match_milvus_kb(user_id, [kb_id])  
    assert milvus_kv is not None  
    success_list = []  
    failed_list = []  
  
    for local_file in local_files:  
        start = time.time()  
        try:  
            local_file.split_file_to_docs(self.get_ocr_result)  
            content_length = sum([len(doc.page_content) for doc in local_file.docs])  
        except Exception as e:  
            error_info = f'split error: {traceback.format\_exc()}'  
            debug_logger.error(error_info)  
            self.milvus_summary.update_file_status(local_file.file_id, status='red')  
            failed_list.append(local_file)  
            continue  
        end = time.time()  
        self.milvus_summary.update_content_length(local_file.file_id, content_length)  
        debug_logger.info(f'split time: {end - start} {len(local\_file.docs)}')  
        start = time.time()  
        try:  
            local_file.create_embedding()  
        except Exception as e:  
            error_info = f'embedding error: {traceback.format\_exc()}'  
            debug_logger.error(error_info)  
            self.milvus_summary.update_file_status(local_file.file_id, status='red')  
            failed_list.append(local_file)  
            continue  
        end = time.time()  
        debug_logger.info(f'embedding time: {end - start} {len(local\_file.embs)}')  
  
        self.milvus_summary.update_chunk_size(local_file.file_id, len(local_file.docs))  
        ret = await milvus_kv.insert_files(local_file.file_id, local_file.file_name, local_file.file_path,  
                                            local_file.docs, local_file.embs)  
        insert_time = time.time()  
        debug_logger.info(f'insert time: {insert\_time - end}')  
        if ret:  
            self.milvus_summary.update_file_status(local_file.file_id, status='green')  
            success_list.append(local_file)  
        else:  
            self.milvus_summary.update_file_status(local_file.file_id, status='yellow')  
            failed_list.append(local_file)  
    debug_logger.info(  
        f"insert\_to\_milvus: success num: {len(success\_list)}, failed num: {len(failed\_list)}")  

        

除开各种校验和数据的同步更新，主要经历的是这几个流程：

• local_file.split_file_to_docs：文件的切片，这里还涉及不同类型的文件处理，例如md、图片等。
• local_file.create_embedding：看名字就知道了，向量化。
• milvus_kv.insert_files：存入milvus。

这就是文件上传后核心要经历的4个流程，即文件读取、文件切片、向量化和入库，接下来我会逐个展开讲。

文件读取和切片

文件读取和切片在代码里有不少是混合的，所以我也合在一起说了。在代码里，我们能看到，他们目前支持的是这几种格式：md,txt,pdf,jpg,png,jpeg,docx,xlsx,pptx,eml,csv，另外还有一个基于url的网页，大概就是这几块的内容，代码里对这几个类型都提供了处理代码，我来逐步解析。

load_and_split

在开始之前，必须了解一下文件读取的这基类BaseLoader，这里对加载、切分都有详细的预定义。这里向大家关注的点只有一个，就是load_and_split，我只把有关的部分放出来，这是一个支持在自定义好加载组件和切片组建后，一条龙使用的函数，注意这个BaseLoader是在langchain_core里的，不是在Qanything项目里的。

          
            
class BaseLoader(ABC):  
    def load\_and\_split(  
        self, text\_splitter: Optional[TextSplitter] = None  
    ) -> List[Document]:  
        """Load Documents and split into chunks. Chunks are returned as Documents.  
  
        Do not override this method. It should be considered to be deprecated!  
  
        Args:  
            text\_splitter: TextSplitter instance to use for splitting documents.  
              Defaults to RecursiveCharacterTextSplitter.  
  
        Returns:  
            List of Documents.  
        """  
  
        if text_splitter is None:  
            try:  
                from langchain_text_splitters import RecursiveCharacterTextSplitter  
            except ImportError as e:  
                raise ImportError(  
                    "Unable to import from langchain\_text\_splitters. Please specify "  
                    "text\_splitter or install langchain\_text\_splitters with "  
                    "`pip install -U langchain-text-splitters`."  
                ) from e  
  
            _text_splitter: TextSplitter = RecursiveCharacterTextSplitter()  
        else:  
            _text_splitter = text_splitter  
        docs = self.load()  
        return _text_splitter.split_documents(docs)  

        

有这个基类后，只需要继承这个积累就能写自己的加载器了，至于文档切分器，则可以在load_and_split使用的时候传进去，例如这样：

          
            
loader = MyRecursiveUrlLoader(url=self.url)  
textsplitter = ChineseTextSplitter(pdf=False, sentence_size=sentence_size)  
docs = loader.load_and_split(text_splitter=textsplitter)  

        

MyRecursiveUrlLoader是URL加载器（具体后面会讲），初始化以后，再定义一个中文的切分器ChineseTextSplitter（具体后面也会讲），然后直接用loader.load_and_split(text_splitter=textsplitter)即可把加载、切片都给搞定了。

下面就来分开把加载和切片两者的操作讲一遍。

文件读取

在这个基类下，根据不同需要，会有各种不一样的加载器，用于应对多种不同的格式，自定义的加载器直接从BaseLoader继承即可。

• MyRecursiveUrlLoader，URL加载器，即网络链接下的内容加载，内部直接用了langchain的WebBaseLoader，网页解析则使用的是BeautifulSoup，算是爬虫技术里的老朋友了，BeautifulSoup主要用于解析代码里暗藏的url，方便进一步查询。
• UnstructuredFileLoader，直接从langchain里面加载的，from langchain.document_loaders import UnstructuredFileLoader。这个也就只用在了markdown里面（.md）。
• TextLoader，也是直接从langchain里面加载的from langchain.document_loaders import UnstructuredFileLoader, TextLoader 。这个也就只用在了txt里面（.txt）。
• UnstructuredPaddlePDFLoader，这个是专门用在pdf文件里的，作者自己写的类，继承自前面提到的UnstructuredFileLoader，但不局限在此，主要重写的是_get_elements函数，内部写了一个函数pdf_ocr_txt，首先用fitz读取pdf每页的图片，然后用ocr_engine来解析（请求orc接口，本项目里用的是一个triton部署的paddleocr服务），最后用unstructured下的一个函数partition_text来完成切片（pip install unstructured），当然后续还会有针对中文的综合切片，后面会说。
• UnstructuredPaddleImageLoader，用来解析图片的工具，对应jpg、png、jpeg后缀文件。同样继承自UnstructuredFileLoader，和PDF不同的是加载部分，图片加载使用的是cv2，加载后和PDF的处理一样，都是走一遍ocr_engine和partition_text。
• UnstructuredWordDocumentLoader用于处理docx文件，来自langchain。
• xlsx使用的是pandas，值得注意的是engine使用的是openpyxl，另外文件读取后，作者会把内容转为csv，然后用CSVLoader来处理。
• CSVLoader顾名思义处理的是csv文件，这里用的是csv.DictReader来读取的。
• UnstructuredPowerPointLoader用于读取PPT，从langchain里面加载的，from langchain.document_loaders import UnstructuredPowerPointLoader。
• UnstructuredEmailLoader用于读取邮件格式的文件.eml，也是从langchain中加载的，from langchain.document_loaders import UnstructuredEmailLoader。

至此，所有支持的文件加载都在这里了，这些文件加载都挺有借鉴意义的，后续在做自己的RAG系统的过程中，也可以考虑直接使用。

文件切片

文件切片作者也是写成了通用的工具，方便调用，而且这个相比各种文件格式，这里的泛用性会更高，毕竟都解析成文本了，这个比较通用ChineseTextSplitter，继承自langchain的from langchain.text_splitter import CharacterTextSplitter，重写后，更符合中文的使用习惯。直接来看源码吧。

          
            
class ChineseTextSplitter(CharacterTextSplitter):  
    def \_\_init\_\_(self, pdf: bool = False, sentence\_size: int = SENTENCE\_SIZE, **kwargs):  
        super().__init__(**kwargs)  
        self.pdf = pdf  
        self.sentence_size = sentence_size  
  
    def split\_text1(self, text: str) -> List[str]:  
        if self.pdf:  
            text = re.sub(r"\n{3,}", "\n", text)  
            text = re.sub('\s', ' ', text)  
            text = text.replace("\n\n", "")  
        sent_sep_pattern = re.compile('([﹒﹔﹖﹗．。！？]["’”」』]{0,2}|(?=["‘“「『]{1,2}|$))')  # del ：；  
        sent_list = []  
        for ele in sent_sep_pattern.split(text):  
            if sent_sep_pattern.match(ele) and sent_list:   
                sent_list[-1] += ele  
            elif ele:  
                sent_list.append(ele)  
        return sent_list  
  
    def split\_text(self, text: str) -> List[str]:   ##此处需要进一步优化逻辑  
        if self.pdf:  
            text = re.sub(r"\n{3,}", r"\n", text)  
            text = re.sub('\s', " ", text)  
            text = re.sub("\n\n", "", text)  
  
        text = re.sub(r'([;；.!?。！？\?])([^”’])', r"\1\n\2", text)  # 单字符断句符  
        text = re.sub(r'(\.{6})([^"’”」』])', r"\1\n\2", text)  # 英文省略号  
        text = re.sub(r'(\…{2})([^"’”」』])', r"\1\n\2", text)  # 中文省略号  
        text = re.sub(r'([;；!?。！？\?]["’”」』]{0,2})([^;；!?，。！？\?])', r'\1\n\2', text)  
        # 如果双引号前有终止符，那么双引号才是句子的终点，把分句符\n放到双引号后，注意前面的几句都小心保留了双引号  
        text = text.rstrip()  # 段尾如果有多余的\n就去掉它  
        # 很多规则中会考虑分号;，但是这里我把它忽略不计，破折号、英文双引号等同样忽略，需要的再做些简单调整即可。  
        ls = [i for i in text.split("\n") if i]  
        for ele in ls:  
            if len(ele) > self.sentence_size:  
                ele1 = re.sub(r'([,，.]["’”」』]{0,2})([^,，.])', r'\1\n\2', ele)  
                ele1_ls = ele1.split("\n")  
                for ele_ele1 in ele1_ls:  
                    if len(ele_ele1) > self.sentence_size:  
                        ele_ele2 = re.sub(r'([\n]{1,}| {2,}["’”」』]{0,2})([^\s])', r'\1\n\2', ele_ele1)  
                        ele2_ls = ele_ele2.split("\n")  
                        for ele_ele2 in ele2_ls:  
                            if len(ele_ele2) > self.sentence_size:  
                                ele_ele3 = re.sub('( ["’”」』]{0,2})([^ ])', r'\1\n\2', ele_ele2)  
                                ele2_id = ele2_ls.index(ele_ele2)  
                                ele2_ls = ele2_ls[:ele2_id] + [i for i in ele_ele3.split("\n") if i] + ele2_ls[  
                                                                                                       ele2_id + 1:]  
                        ele_id = ele1_ls.index(ele_ele1)  
                        ele1_ls = ele1_ls[:ele_id] + [i for i in ele2_ls if i] + ele1_ls[ele_id + 1:]  
  
                id = ls.index(ele)  
                ls = ls[:id] + [i for i in ele1_ls if i] + ls[id + 1:]  
        return ls  

        

实际使用的应该是split_text，不带1那个，这里涉及了很多逻辑和替换，主要都是为了做句子片段的划分，这里的正则大家也可以多多了解和尝试。

在此基础上，都会再过第二次切分，这次切分旨在对长度太长（800tokens+）的进行进一步切分，此处使用的是langchain的RecursiveCharacterTextSplitter。

          
            
from langchain.text_splitter import RecursiveCharacterTextSplitter  
text_splitter = RecursiveCharacterTextSplitter(  
    separators=["\n", ".", "。", "!", "！", "?", "？", "；", ";", "……", "…", "、", "，", ",", " "],  
    chunk_size=400,  
    length_function=num_tokens,  
)  

        

后面，为了确保信息的存储的可查性（检索这段话后，能找到对应的文章），还把文件id和文件名都给记录到doc内（说白了就是正排）。

          
            
# 这里给每个docs片段的metadata里注入file\_id  
for doc in docs:  
    doc.metadata["file\_id"] = self.file_id  
    doc.metadata["file\_name"] = self.url if self.url else os.path.split(self.file_path)[-1]  

        

索引构造

在对文本进行好切片后，就可以开始跑模型准备向数据库灌数据了。此处我把他叫做索引构造，主要包括数据转化和灌库两个操作。

核心的代码同样是在local_doc_qa.insert_files_to_milvus这个函数下，这里面create_embedding就是构造向量的过程，在前面的章节（前沿重器[45] RAG开源项目Qanything源码阅读1-概述+服务）有提及，向量化的模型是单独用triton部署的，所以此处是直接请求模型服务获取的。

          
            
CUDA_VISIBLE_DEVICES=$gpu\_id1 nohup /opt/tritonserver/bin/tritonserver --model-store=/model_repos/QAEnsemble_embed_rerank --http-port=9000 --grpc-port=9001 --metrics-port=9002 --log-verbose=1 > /workspace/qanything_local/logs/debug_logs/embed_rerank_tritonserver.log 2>&1 &  

        

而请求方面，先放一个调用的关键入口。

          
            
def create\_embedding(self):  
    self.embs = self.emb_infer._get_len_safe_embeddings([doc.page_content for doc in self.docs])  

        

这里实际的调用挺深的，首先对于local，有YouDaoLocalEmbeddings，这里是包装向量模型的，里面更多是考虑并发的concurrent代码，向量是内部的embedding_client（一个EmbeddingClient实例）负责的（当然EmbeddingClient下还有concurrent的代码），这个应该才是算法比较关心的部分吧，我直接把EmbeddingClient的核心代码放出来。

          
            
import os  
import math  
import numpy as np  
import time  
  
from typing import Optional  
  
import onnxruntime as ort  
from tritonclient import utils as client_utils  
from tritonclient.grpc import InferenceServerClient, InferInput, InferRequestedOutput  
from transformers import AutoTokenizer  
  
WEIGHT2NPDTYPE = {  
    "fp32": np.float32,  
    "fp16": np.float16,  
}  
  
class EmbeddingClient:  
    DEFAULT_MAX_RESP_WAIT_S = 120  
    embed_version = "local\_v0.0.1\_20230525\_6d4019f1559aef84abc2ab8257e1ad4c"  
  
    def \_\_init\_\_(  
        self,  
        server\_url: str,  
        model\_name: str,  
        model\_version: str,  
        tokenizer\_path: str,  
        resp\_wait\_s: Optional[float] = None,  
    ):  
        self._server_url = server_url  
        self._model_name = model_name  
        self._model_version = model_version  
        self._response_wait_t = self.DEFAULT_MAX_RESP_WAIT_S if resp_wait_s is None else resp_wait_s  
        self._tokenizer = AutoTokenizer.from_pretrained(tokenizer_path)  
  
    def get\_embedding(self, sentences, max\_length=512):  
        # Setting up client  
      
        inputs_data = self._tokenizer(sentences, padding=True, truncation=True, max_length=max_length, return_tensors='np')  
        inputs_data = {k: v for k, v in inputs_data.items()}  
      
        client = InferenceServerClient(url=self._server_url)  
        model_config = client.get_model_config(self._model_name, self._model_version)  
        model_metadata = client.get_model_metadata(self._model_name, self._model_version)  
      
        inputs_info = {tm.name: tm for tm in model_metadata.inputs}  
        outputs_info = {tm.name: tm for tm in model_metadata.outputs}  
        output_names = list(outputs_info)  
        outputs_req = [InferRequestedOutput(name_) for name_ in outputs_info]  
        infer_inputs = []  
        for name_ in inputs_info:  
            data = inputs_data[name_]  
            infer_input = InferInput(name_, data.shape, inputs_info[name_].datatype)  
      
            target_np_dtype = client_utils.triton_to_np_dtype(inputs_info[name_].datatype)  
            data = data.astype(target_np_dtype)  
      
            infer_input.set_data_from_numpy(data)  
            infer_inputs.append(infer_input)  
      
        results = client.infer(  
            model_name=self._model_name,  
            model_version=self._model_version,  
            inputs=infer_inputs,  
            outputs=outputs_req,  
            client_timeout=120,  
        )  
        y_pred = {name_: results.as_numpy(name_) for name_ in output_names}  
        embeddings = y_pred["output"][:,0]  
        norm_arr = np.linalg.norm(embeddings, axis=1, keepdims=True)  
        embeddings_normalized = embeddings / norm_arr  
        return embeddings_normalized.tolist()  
      
    def getModelVersion(self):  
        return self.embed_version  

        

• 首先可以看到，tokenizer依旧是本服务做的。
• 服务的请求主要是client负责，triton是一个grpc接口（GRPC我很早之前写过，可以参考系统学习），输入和输出的数据结构参考InferInput和InferRequestedOutput。
• 细节，对模型的输出结果，结果作者还做了额外的处理，主要是做了一个归一化，用np.linalg.norm求了二范数（默认），然后想了都除以了这个二范数。
• 有留意到，对模型的版本，作者有可以保留，方便进行模型迭代的版本可控性。

GRPC文章：

• ML&DEV[9] | gRPC初体验
• ML&DEV[10] | gRPC的应用
• 心法利器[6] | python grpc实践

完成后，就可以开始灌库了，milvus_kv.insert_files。milvus自己是有开源的库的，即pymilvus，作者自己写了一个完整的类MilvusClient，至于pymilvus具体教程大家可以看：https://zhuanlan.zhihu.com/p/676124465。这里我不展开具体的使用方法了，不过还是可以从灌库的源码里挑出一些重要的细节。

          
            
async def insert\_files(self, file\_id, file\_name, file\_path, docs, embs, batch\_size=1000):  
    debug_logger.info(f'now inser\_file {file\_name}')  
    now = datetime.now()  
    timestamp = now.strftime("%Y%m%d%H%M")  
    loop = asyncio.get_running_loop()  
    contents = [doc.page_content for doc in docs]  
    num_docs = len(docs)  
    for batch_start in range(0, num_docs, batch_size):  
        batch_end = min(batch_start + batch_size, num_docs)  
        data = [[] for _ in range(len(self.sess.schema))]  
  
        for idx in range(batch_start, batch_end):  
            cont = contents[idx]  
            emb = embs[idx]  
            chunk_id = f'{file\_id}\_{idx}'  
            data[0].append(chunk_id)  
            data[1].append(file_id)  
            data[2].append(file_name)  
            data[3].append(file_path)  
            data[4].append(timestamp)  
            data[5].append(cont)  
            data[6].append(emb)  
  
        # 执行插入操作  
        try:  
            debug_logger.info('Inserting into Milvus...')  
            mr = await loop.run_in_executor(  
                self.executor, partial(self.partitions[0].insert, data=data))  
            debug_logger.info(f'{file\_name} {mr}')  
        except Exception as e:  
            debug_logger.error(f'Milvus insert file\_id:{file\_id}, file\_name:{file\_name} failed: {e}')  
            return False  
  
    # 混合检索  
    if self.hybrid_search:  
        debug_logger.info(f'now inser\_file for es: {file\_name}')  
        for batch_start in range(0, num_docs, batch_size):  
            batch_end = min(batch_start + batch_size, num_docs)  
            data_es = []  
            for idx in range(batch_start, batch_end):  
                data_es_item = {  
                    'file\_id': file_id,  
                    'content': contents[idx],  
                    'metadata': {  
                        'file\_name': file_name,  
                        'file\_path': file_path,  
                        'chunk\_id': f'{file\_id}\_{idx}',  
                        'timestamp': timestamp,  
                    }  
                }  
                data_es.append(data_es_item)  
  
            try:  
                debug_logger.info('Inserting into es ...')  
                mr = await self.client.insert(data=data_es, refresh=batch_end==num_docs)  
                debug_logger.info(f'{file\_name} {mr}')  
            except Exception as e:  
                debug_logger.error(f'ES insert file\_id: {file\_id}\nfile\_name: {file\_name}\nfailed: {e}')  
                return False  
  
    return True  

        

• milvus使用的是pymilvus工具来读写，其中self.partitions[0].insert就是用存储数据的，此处可以注意到data内有很多不同的字段。
• 执行代码使用的是loop.run_in_executor，有留意到，在MilvusClient内有一个self.executor，这个的定义在这个类的__init__内，self.executor = ThreadPoolExecutor(max_workers=10)，这里新建了一个线程池，新技能get。
• 下方是ES的数据灌入。个人感觉，这个ES数据处理写在这个位置并不是很合适，应该单独出来处理，毕竟混合代码不太好看到。

小结

本文离线的文件处理，我看了挺久，而且写的时间也很长。我自己看完的收获还挺大的，原本是对文档处理比较生疏，但这次看完对这块的理解比较深了，而且通过通篇阅读，也能了解到作者的设计思路，希望大家也能在阅读本文的过程中有所收获吧。

下一篇，在线推理，敬请期待。