基于OpenVINO和LangChain構建RAG問答系統

作者：楊亦誠 英特爾 AI 軟件工程師

背景

隨著生成式 AI 的興起，和大語言模型對話聊天的應用變得非常熱門，但這類應用往往只能簡單地和你“聊聊家?！?，并不能針對某些特定的行業，給出非常專業和精準的答案。這也是由于大語言模型（以下簡稱 LLM）在時效性和專業性上的局限所導致，現在市面上大部分開源的 LLM 幾乎都只是使用某一個時間點前的公開數據進行訓練，因此它無法學習到這個時間點之后的知識，并且也無法保證在專業領域上知識的準確性。那有沒有辦法讓你的模型學習到新的知識呢？

當然有，這里一般有 2 種方案：

Fine-tuning 微調

微調通過對特定領域數據庫進行廣泛的訓練來調整整個模型。這樣可以內化專業技能和知識。然后，微調也需要大量的數據、大量的計算資源和定期的重新訓練以保持時效性。

RAG 檢索增強生成

RAG的全稱是 Retrieval-Augmented Generation，它的原理是通過檢索外部知識來給出上下文響應，在無需對模型進行重新訓練的情況，保持模型對于特定領域的專業性，同時通過更新數據查詢庫，可以實現快速地知識更新。但 RAG 在構建以及檢索知識庫時，會占用更多額外的內存資源，其回答響應延時也取決于知識庫的大小。

從以上比較可以看出，在沒有足夠 GPU 計算資源對模型進行重新訓練的情況下，RAG 方式對普通用戶來說更為友好。因此本文也將探討如何利用 OpenVINO 以及 LangChain 工具來構建屬于你的 RAG 問答系統。

RAG 流程

雖然 RAG 可以幫助 LLM “學習”到新的知識，并給出更可靠的答案，但它的實現流程并不復雜，主要可以分為以下兩個部分：

01構建知識庫檢索

圖：構建知識庫流程

Load 載入：

讀取并解析用戶提供的非結構化信息，這里的非結構化信息可以是例如 PDF 或者 Markdown 這樣的文檔形式。

Split 分割：

將文檔中段落按標點符號或是特殊格式進行拆分，輸出若干詞組或句子，如果拆分后的單句太長，將不便于后期 LLM 理解以及抽取答案，如果太短又無法保證語義的連貫性，因此我們需要限制拆分長度（chunk size），此外，為了保證 chunk 之間文本語義的連貫性，相鄰 chunk 會有一定的重疊，在 LangChain 中我可以通過定義 Chunk overlap 來控制這個重疊區域的大小。

圖：Chunk size 和 Chunk overlap 示例

Embedding 向量化：

使用深度學習模型將拆分后的句子向量化，把一段文本根據語義在一個多維空間的坐標系里面表示出來，以便知識庫存儲以及檢索，語義將近的兩句話，他們所對應的向量相似度會相對較大，反之則較小，以此方式我們可以在檢索時，判斷知識庫里句子是否可能為問題的答案。

Store 存儲：

構建知識庫，將文本以向量的形式存儲，用于后期檢索。

02檢索和答案生成

圖：答案生成流程

Retrieve 檢索：

當用戶問題輸入后，首先會利用 embedding 模型將其向量化，然后在知識庫中檢索與之相似度較高的若干段落，并對這些段落的相關性進行排序。

Generate 生成：

將這個可能包含答案，且相關性最高的 Top K 個檢索結果，包裝為 Prompt 輸入，喂入 LLM 中，據此來生成問題所對應的的答案。

關鍵步驟

在利用 OpenVINO構建 RAG 系統過程中有以下一些關鍵步驟：

01封裝 Embedding 模型類

由于在 LangChain 的 chain pipeline 會調用 embedding 模型類中的embed_documents和 embed_query 來分別對知識庫文檔和問題進行向量化，而他們最終都會調用 encode 函數來實現每個 chunk 具體的向量化實現，因此在自定義的 embedding 模型類中也需要實現這樣幾個關鍵方法，并通過 OpenVINO進行推理任務的加速。

圖：embedding 模型推理示意

由于在 RAG 系統中的各個 chunk 之間的向量化任務往往沒有依賴關系，因此我們可以通過 OpenVINO 的 AsyncInferQueue 接口，將這部分任務并行化，以提升整個 embedding 任務的吞吐量。

for i, sentence in enumerate(sentences_sorted):
      inputs = {}
      features = self.tokenizer(
        sentence, padding=True, truncation=True, return_tensors='np')
      for key in features:
        inputs[key] = features[key]
      infer_queue.start_async(inputs, i)
    infer_queue.wait_all()
    all_embeddings = np.asarray(all_embeddings)

左滑查看更多

此外，從 HuggingFace Transfomers 庫中（https://hf-mirror.com/sentence-transformers/all-mpnet-base-v2#usage-huggingface-transformers）導出的 embedding 模型是不包含 mean_pooling 和歸一化操作的，因此我們需要在獲取模型推理結果后，再實現這部分后處理任務。并將其作為 callback function 與 AsyncInferQueue 進行綁定。

def postprocess(request, userdata):
      embeddings = request.get_output_tensor(0).data
      embeddings = np.mean(embeddings, axis=1)
      if self.do_norm:
        embeddings = normalize(embeddings, 'l2')
      all_embeddings.extend(embeddings)


    infer_queue.set_callback(postprocess)

左滑查看更多

02封裝 LLM 模型類

由于 LangChain 已經可以支持 HuggingFace 的 pipeline 作為其 LLM 對象，因此這里我們只要將 OpenVINO 的 LLM 推理任務封裝成一個 HF 的 text generation pipeline 即可（詳細方法可以參考我的上一篇文章）。此外為了流式輸出答案（逐字打?。?，需要通過 TextIteratorStreamer 對象定義一個流式生成器。

streamer = TextIteratorStreamer(
  tok, timeout=30.0, skip_prompt=True, skip_special_tokens=True
)
generate_kwargs = dict(
  model=ov_model,
  tokenizer=tok,
  max_new_tokens=256,
  streamer=streamer,
  # temperature=1,
  # do_sample=True,
  # top_p=0.8,
  # top_k=20,
  # repetition_penalty=1.1,
)
if stop_tokens is not None:
  generate_kwargs["stopping_criteria"] = StoppingCriteriaList(stop_tokens)
  
pipe = pipeline("text-generation", **generate_kwargs)
llm = HuggingFacePipeline(pipeline=pipe)

左滑查看更多

03設計 RAG prompt template

當完成檢索后，RAG 會將相似度最高的檢索結果包裝為 Prompt，讓 LLM 進行篩選與重構，因此我們需要為每個 LLM 設計一個 RAG prompt template，用于在 Prompt 中區分這些檢索結果，而這部分的提示信息我們又可以稱之為 context 上下文，以供 LLM 在生成答案時進行參考。以 ChatGLM3 為例，它的 RAG prompt template 可以是這樣的：

 "prompt_template": f"""<|system|>
    {DEFAULT_RAG_PROMPT_CHINESE }"""
    + """
    <|user|>
    問題: {question} 
    已知內容: {context} 
    回答: 
    <|assistant|>""",

左滑查看更多

其中：

● {DEFAULT_RAG_PROMPT_CHINESE}為我們事先根據任務要求定義的系統提示詞。

●{question}為用戶問題。

●{context} 為 Retriever 檢索到的，可能包含問題答案的段落。

例如，假設我們的問題是“飛槳的四大優勢是什么？”，對應從飛槳文檔中獲取的 Prompt 輸入就是：

“<|system|>
基于以下已知信息，請簡潔并專業地回答用戶的問題。如果無法從中得到答案，請說 "根據已知信息無法回答該問題" 或 "沒有提供足夠的相關信息"。不允許在答案中添加編造成分。另外，答案請使用中文。
<|user|>
問題: 飛槳的四大領先技術是什么？
已知內容: ## 安裝


PaddlePaddle最新版本: v2.5
跟進PaddlePaddle最新特性請參考我們的版本說明


四大領先技術
開發便捷的產業級深度學習框架


飛槳深度學習框架采用基于編程邏輯的組網范式，對于普通開發者而言更容易上手，符合他們的開發習慣。同時支持聲明式和命令式編程，兼具開發的靈活性和高性能。網絡結構自動設計，模型效果超越人類專家。


支持超大規模深度學習模型的訓練


飛槳突破了超大規模深度學習模型訓練技術，實現了支持千億特征、萬億參數、數百節點的開源大規模訓練平臺，攻克了超大規模深度學習模型的在線學習難題，實現了萬億規模參數模型的實時更新。
查看詳情


支持多端多平臺的高性能推理部署工具
…
<|assistant|>“

左滑查看更多

04創建 RetrievalQA 檢索

在文本分割這個任務中，LangChain 支持了多種分割方式，例如按字符數的 CharacterTextSplitter，針對 Markdown 文檔的 MarkdownTextSplitter，以及利用遞歸方法的 RecursiveCharacterTextSplitter，當然你也可以通過繼成 TextSplitter 父類來實現自定義的 split_text 方法，例如在中文文檔中，我們可以采用按每句話中的標點符號進行分割。

class ChineseTextSplitter(CharacterTextSplitter):
  def __init__(self, pdf: bool = False, **kwargs):
    super().__init__(**kwargs)
    self.pdf = pdf


  def split_text(self, text: str) -> List[str]:
    if self.pdf:
      text = re.sub(r"
{3,}", "
", text)
      text = text.replace("

", "")
    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

左滑查看更多

接下來我們需要載入預先設定的好的 prompt template，創建 rag_chain。

圖：Chroma 引擎檢索流程

這里我們使用 Chroma 作為檢索引擎，在 LangChain 中，Chroma 默認使用 cosine distance 作為向量相似度的評估方法，同時可以通過調整 db.as_retriever（search_type= "similarity_score_threshold"），或是 db.as_retriever（search_type= "mmr"）來更改默認搜索策略，前者為帶閾值的相似度搜索，后者為 max_marginal_relevance 算法。當然 Chroma 也可以被替換為 FAISS 檢索引擎，使用方式也是相似的。

此外通過定義 as_retriever函數中的 {"k": vector_search_top_k}，我們還可以改變檢索結果的返回數量，有助于幫助 LLM 獲取更多有效信息，但也為增加 Prompt 的長度，提高推理延時，因此不建議將該數值設定太高。創建 rag_chain 的完整代碼如下：

 documents = load_single_document(doc.name)


  text_splitter = TEXT_SPLITERS[spliter_name](
    chunk_size=chunk_size, chunk_overlap=chunk_overlap
  )


  texts = text_splitter.split_documents(documents)


  db = Chroma.from_documents(texts, embedding)
  retriever = db.as_retriever(search_kwargs={"k": vector_search_top_k})


  global rag_chain
  prompt = PromptTemplate.from_template(
    llm_model_configuration["prompt_template"])
  chain_type_kwargs = {"prompt": prompt}
  rag_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=retriever,
    chain_type_kwargs=chain_type_kwargs,
  )

左滑查看更多

05答案生成

創建以后的 rag_chain 對象可以通過 rag_chain.run(question) 來響應用戶的問題。將它和線程函數綁定后，就可以從 LLM 對象的 streamer 中獲取流式的文本輸出。

def infer(question):
    rag_chain.run(question)
    stream_complete.set()


  t1 = Thread(target=infer, args=(history[-1][0],))
  t1.start()
  partial_text = ""
  for new_text in streamer:
    partial_text = text_processor(partial_text, new_text)
    history[-1][1] = partial_text
    yield history

左滑查看更多

最終效果

最終效果如下圖所示，當用戶上傳了自己的文檔文件后，點擊 Build Retriever 便可以創建知識檢索庫，同時也可以根據自己文檔的特性，通過調整檢索庫的配置參數來實現更高效的搜索。當完成檢索庫創建后就可以在對話框中與 LLM 進行問答交互了。

圖：基于 RAG 的問答系統效果

總結

在醫療、工業等領域，行業知識庫的構建已經成為了一個普遍需求，通過 LLM 與 OpenVINO 的加持，我們可以讓用戶對于知識庫的查詢變得更加精準與高效，帶來更加友好的交互體驗。

審核編輯：湯梓紅