本文介紹基于云原生內存數據庫Tair與大語言模型（Large Language Model, LLM）構建企業專屬Chatbot的解決方案。
背景信息
大語言模型已具備了相當豐富的基礎知識、閱讀理解和邏輯推理能力。但想在實際使用過程中充分發揮它的潛力，仍需提供額外的信息進行輔助，原因如下：
不了解私域數據：盡管大語言模型已通過大量訓練樣本，但無法保證其已學習、掌握指定的私域數據或新知識。
不具備多輪對話能力：大語言模型能夠接收的請求大小（即Token數）是有限制的，通常是4 KB～32 KB之間，并且響應速度隨著請求變大而變慢。以ChatGLM為例，將用戶的歷史記錄作為請求的一部分持續追加，當超過Token數限制時就需要丟棄老的聊天記錄，無法實現長期、多輪對話能力。
因此，可以使用云原生內存數據庫Tair作為大語言模型的外部存儲，將私域數據和長期對話記錄存儲在Tair向量檢索中，結合Tair高效的向量檢索能力與大語言模型的AI能力，使企業專屬Chatbot更加智能。
Tair特性與優勢
Tair向量檢索是在云原生內存數據庫Tair的基礎上，以內置引擎的方式，為用戶提供實時高性能、多模混合檢索、簡單易用的向量數據庫服務。
實時高性能：所有操作均在內存中進行，支持FLAT、HNSW索引算法，支持歐式距離、向量內積、余弦距離等多種距離函數，索引創建與向量數據點查詢的時延均小于1ms，更多信息請參見TairVector性能白皮書。
多模混合檢索：完全兼容Redis生態與使用方式，同時還支持向量檢索、全文檢索及其他多種擴展數據結構，更多信息請參見Tair擴展數據結構。
簡單易用：1 GB規格（低成本）起步，支持在線擴、縮容，最高可擴容至16 TB集群。
應用場景
私域數據問答
提前將私域數據寫入Tair向量中，同時也可以在Tair中構建一份全文索引。在請求LLM前，Tair支持提供向量檢索和全文檢索兩路召回，再通過Prompt潤色，一并將問題與私域數據提交給LLM，可實現企業專屬Chatbot定制。
長Session會話
將用戶的歷史會話記錄存儲在Session中（支持設置TTL等過期機制）。在請求LLM前，通過Tair向量檢索技術將相關歷史信息檢索出來，再通過Prompt潤色后，一并提交給LLM，可實現基于長期、多輪對話下的上下文感知能力。
快速體驗
在阿里云云速搭CADT平臺中已上線大模型結合Tair構建企業級專屬Chatbot解決方案。
該方案模板中已預部署ECS和Tair實例，預安裝了前置安裝包，Demo中已實現了私域數據問答、長Session會話功能，您可以結合教程快速體驗專屬Chatbot，更多信息請參見大模型結合Tair構建企業級專屬Chatbot。
代碼說明
本項目來自于開源項目langchain-ChatGLM，其中向量數據庫使用Tair、模型為ChatGLM-6B，請自覺遵守相關協議及法律法規等。
如下圖所示，灰色箭頭表示文本（TXT）文件經過一系列處理，把向量信息以及切分后的原始文本一起寫入到Tair中。橙色箭頭表示在用戶發起一個查詢后，先從Tair中檢索出最相似的TopK個向量，然后與用戶問題進行Prompt，一并提交給LLM，讓LLM更好地回答用戶的問題。
以下代碼僅說明Tair與大語言模型結合的部分，具體Demo運行，請參見快速體驗章節。
# -*- coding: utf-8 -*-
# !/usr/bin/env python
import torch
from langchain.document_loaders import TextLoader
from langchain.embeddings.huggingface import HuggingFaceEmbeddings
from transformers import AutoTokenizer, AutoModel
from langchain.vectorstores import Tair
from utils import ChineseTextSplitter
from langchain.schema import Document
from typing import List
from tair import Tair as TairClient
TOPK = 2
EMBEDDING_DEVICE = "cuda" if torch.cuda.is_available() else "mps" if torch.backends.mps.is_available() else "cpu"
PROMPT_TEMPLATE =
"""
已知信息：
{context}
根據上述已知信息，簡潔回答用戶的問題。如果無法從中得到答案，請說 “根據已知信息無法回答該問題” 或 “沒有提供足夠的相關信息”，答案請使用中文。 問題是：{question}
"""
PROMPT_TEMPLATE_SESSION =
"""
已知用戶以前提供的問題：
{context}
根據上述已知信息，簡潔回答用戶問題。如果無法從中得到答案，請自行回答用戶問題，答案請使用中文。 問題是：{question}
"""
class ChatBot():
"""
使用開源模型，請遵循相關協議及法律法規。參數說明：
tair_url: Tair的連接地址，例如"redis://account_name:account_password@r-bp10xx****.redis.rds.aliyuncs.com:6379"。
GanymedeNil/text2vec-large-chinese：需下載"https://huggingface.co/GanymedeNil/text2vec-large-chinese"，并替換為本地目錄。
THUDM/chatglm-6b：需下載"https://huggingface.co/THUDM/chatglm-6b"，并替換為本地目錄。
"""
def __init__(self, tair_url):
self.tair_url = tair_url
self.text_embeddings_model= HuggingFaceEmbeddings(model_name='GanymedeNil/text2vec-large-chinese',
model_kwargs={'device': EMBEDDING_DEVICE})
self.tokenizer = AutoTokenizer.from_pretrained("THUDM/chatglm-6b", trust_remote_code=True)
self.llm_model = AutoModel.from_pretrained("THUDM/chatglm-6b", trust_remote_code=True).half().cuda()
self.tair_session_client = TairClient.from_url(tair_url)
"""
該代碼示例的私域數據存儲langchain的index中。
"""
def get_langchain_tair(self, index_name="langchain"):
return Tair(self.text_embeddings_model, self.tair_url, index_name)
"""
結合EXPIRE可以給某個index設置過期時間。
"""
def expire_index(self, index_name):
return self.tair_session_client.expire(index_name, 1200)
"""
判斷不存在某個index，不存在返回True, 存在返回False。
"""
def not_exists_index(self, session_id):
index = self.tair_session_client.tvs_get_index(session_id)
if index is not None:
return False
return True
"""
與大模型交互接口，該代碼示例的history為空，LLM中對history的處理，請參考"https://github.com/imClumsyPanda/langchain-ChatGLM/blob/master/webui.py"。
"""
def chat(self, question:str):
response, history = self.llm_model.chat(self.tokenizer, question, history=[])
return response
"""
寫入私域數據，該方法需輸入文件路徑。
ChineseTextSplitter請參考"https://github.com/imClumsyPanda/langchain-ChatGLM/blob/master/textsplitter/chinese_text_splitter.py"。
"""
def insert_text(self,filepath:str):
loader = TextLoader(filepath, autodetect_encoding=True)
textsplitter = ChineseTextSplitter(pdf=False, sentence_size=100)
docs= loader.load_and_split(textsplitter)
self.insert_text_to_tair(docs,filepath)
"""
Tair.from_documents具體實現，請參考"https://github.com/langchain-ai/langchain/blob/master/libs/community/langchain_community/vectorstores/tair.py"。
最終也是調用Tair的TVS.HSET接口寫入向量與標量。
"""
def insert_text_to_tair(self,docs:List[Document]=None,filename:str=""):
Tair.from_documents(docs, self.text_embeddings_model, tair_url=self.tair_url)
"""
結合私域數據，做prompt提交給大模型。
"""
def chat_by_prompt(self,query:str,topk:int=TOPK):
context = self.get_langchain_tair().similarity_search(query, k=topk)
context = "\n".join([doc.page_content for doc in context])
prompt = PROMPT_TEMPLATE.replace("{question}", query).replace("{context}", context)
response = self.chat(prompt)
return response
"""
獲取用戶歷史會話Session，根據用戶的歷史提示回答問題。
"""
def get_prompt_by_tair_session(self, query, session_id):
if self.not_exists_index(session_id):
prompt = query
return prompt
related_docs_with_score = self.get_langchain_tair(session_id).similarity_search(query, k=self.top_k)
if len(related_docs_with_score)>0:
context = "\n".join([doc.page_content for doc in related_docs_with_score])
prompt = PROMPT_TEMPLATE_SESSION.replace("{question}", query).replace("{context}", context)
else:
prompt = query
return prompt
"""
將用戶歷史回答寫入Session。關于Session如何存儲仍需不斷探索，您可以根據自己的需求進行定制化開發。
"""
def insert_tair_session(self, query, resp, session_id):
text = f"{query}"
Tair.from_texts([text], self.embeddings, None, session_id, "content", "metadata", tair_url=self.insert_tair_session, index_type="FLAT")
self.expire_index(session_id)效果展示
私域數據問答
下圖展示了上傳私域數據，回答私域數據問題的示例。
長Session會話
下圖展示了根據用戶已提供的信息，回答用戶問題的示例。