背景

目前由于ChatGPT横空出世，互联网如雨后春笋冒出了非常多的类ChatGPT的大型语言模型。但是对于这些语言模型，我们应该如何将它应用到我们实际的生产中需要一个更加成熟的解决方案。

介绍

本文旨在通过介绍ChatGLM的使用来讲述如何将一个开源的语言模型应用于智能问答，知识库问答的场景中，通过一系列实操例子来理解整个应用思路。

前期准备

• 一个开源语言模型，这里推荐ChatGLM-6B，开源的、支持中英双语的对话语言模型，并且要求的显存内存非常低，可以在个人PC中轻松部署。
• python3.8+
• milvus，向量索引库
• pytorch以及运行ChatGLM-6B所需要的CUDA和NVIDIA驱动
• …

基于文档的知识库问答

实现步骤

1. 清洗知识库文档，将文档向量化并存入向量数据库
2. 用户提问
3. 将用户提问向量化并查询向量数据库得到匹配的N条知识
4. 将匹配的知识构建prompt，并通过langchain处理用户的问题
5. 调用llm搭配prompt回答用户的问题

向量索引

我们首先需要定义一个向量索引库，在这里我选用的是milvus作为向量索引库来实现我们的文档向量索引和相似度匹配的工作

为了更方便的部署，这里我采用了docker-compose来启动milvus服务。

大家可以在milvus的官方文档中看到最新版本的部署方式Install Milvus Standalone with Docker Compose
嫌麻烦也可以直接复制使用下面的yaml文件

version: '3.5'services:etcd:container_name: milvus-etcdimage: quay.io/coreos/etcd:v3.5.0environment:- ETCD_AUTO_COMPACTION_MODE=revision- ETCD_AUTO_COMPACTION_RETENTION=1000- ETCD_QUOTA_BACKEND_BYTES=4294967296- ETCD_SNAPSHOT_COUNT=50000volumes:- ${DOCKER_VOLUME_DIRECTORY:-.}/volumes/etcd:/etcdcommand: etcd -advertise-client-urls=http://127.0.0.1:2379 -listen-client-urls http://0.0.0.0:2379 --data-dir /etcdminio:container_name: milvus-minioimage: minio/minio:RELEASE.2023-03-20T20-16-18Zenvironment:MINIO_ACCESS_KEY: minioadminMINIO_SECRET_KEY: minioadminvolumes:- ${DOCKER_VOLUME_DIRECTORY:-.}/volumes/minio:/minio_datacommand: minio server /minio_datahealthcheck:test: ["CMD", "curl", "-f", "http://localhost:9000/minio/health/live"]interval: 30stimeout: 20sretries: 3standalone:container_name: milvus-standaloneimage: milvusdb/milvus:v2.2.5command: ["milvus", "run", "standalone"]environment:ETCD_ENDPOINTS: etcd:2379MINIO_ADDRESS: minio:9000volumes:- ${DOCKER_VOLUME_DIRECTORY:-.}/volumes/milvus:/var/lib/milvusports:- "19530:19530"- "9091:9091"depends_on:- "etcd"- "minio"networks:default:name: milvus

当我们创建好docker-compose.yml文件之后就可以使用命令行docker-compose up -d来启动milvus服务。

接下来就是文档预处理

文档预处理

当我们收集到足够的文档之后，我们需要对文档进行一些清洗，方便我们之后的向量匹配更加精准。
这里，我们需要完成以下步骤：

1. 连接milvus向量库
2. 创建对应的connection
3. 遍历读取文档
4. 文档预处理
5. 文档内容转向量
6. 存入向量库

为此，我们编写代码如下

import os
import re
import jieba
import torch
import pandas as pd
from pymilvus import utility
from pymilvus import connections, CollectionSchema, FieldSchema, Collection, DataType
from transformers import AutoTokenizer, AutoModelconnections.connect(alias="default",host='localhost',port='19530'
)# 定义集合名称和维度
collection_name = "document"
dimension = 768
docs_folder = "./knowledge/"tokenizer = AutoTokenizer.from_pretrained("bert-base-chinese")
model = AutoModel.from_pretrained("bert-base-chinese")# 获取文本的向量
def get_vector(text):input_ids = tokenizer(text, padding=True, truncation=True, return_tensors="pt")["input_ids"]with torch.no_grad():output = model(input_ids)[0][:, 0, :].numpy()return output.tolist()[0]def create_collection():# 定义集合字段fields = [FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True, description="primary id"),FieldSchema(name="title", dtype=DataType.VARCHAR, max_length=50),FieldSchema(name="content", dtype=DataType.VARCHAR, max_length=10000),FieldSchema(name="vector", dtype=DataType.FLOAT_VECTOR, dim=dimension),]# 定义集合模式schema = CollectionSchema(fields=fields, description="collection schema")# 创建集合if utility.has_collection(collection_name):# 如果你想继续添加新的文档可以直接 return。但你想要重新创建collection，就可以执行下面的代码# returnutility.drop_collection(collection_name)collection = Collection(name=collection_name, schema=schema, using='default', shards_num=2)# 创建索引default_index = {"index_type": "IVF_FLAT", "params": {"nlist": 2048}, "metric_type": "IP"}collection.create_index(field_name="vector", index_params=default_index)print(f"Collection {collection_name} created successfully")else:collection = Collection(name=collection_name, schema=schema, using='default', shards_num=2)# 创建索引default_index = {"index_type": "IVF_FLAT", "params": {"nlist": 2048}, "metric_type": "IP"}collection.create_index(field_name="vector", index_params=default_index)print(f"Collection {collection_name} created successfully")def init_knowledge():collection = Collection(collection_name)# 遍历指定目录下的所有文件，并导入到 Milvus 集合中docs = []for root, dirs, files in os.walk(docs_folder):for file in files:# 只处理以 .txt 结尾的文本文件if file.endswith(".txt"):file_path = os.path.join(root, file)with open(file_path, "r", encoding="utf-8") as f:content = f.read()# 对文本进行清洗处理content = re.sub(r"\s+", " ", content)title = os.path.splitext(file)[0]# 分词words = jieba.lcut(content)# 将分词后的文本重新拼接成字符串content = " ".join(words)# 获取文本向量vector = get_vector(title + content)docs.append({"title": title, "content": content, "vector": vector})# 将文本内容和向量通过 DataFrame 一起导入集合中df = pd.DataFrame(docs)collection.insert(df)print("Documents inserted successfully")if __name__ == "__main__":create_collection()init_knowledge()

可以看到，我们创建了一个名为document的collection。它包含四个字段id，title，content，vector其中vector储存的是content转化的向量。（当然，我们只是简单的实现了一个最原始的向量索引，如果你想要之后的匹配更加精准更加高效，你可以考虑将大文档按照段落切割并分别转化为向量，并且相互关联上。）

于此同时，我们采用了jieba作为分词库，对文本进行清洗，还使用了正则去除了文档中不必要的一些特殊符号。这些操作可以让我们向量匹配更加精准。

当这些步骤全部执行完毕之后，我们就可以进行用户提问匹配向量库的操作了。

用户提问匹配知识库

首先，我们需要将用户提供的查询向量转换为blob对象，以便与数据库中的向量进行比较。我们在上个步骤实现了get_vector方法来将文本转为向量，现在可以继续调用该方法来实现。

其次我们需要将问题转化的向量用来查找向量库，并得出最为匹配的几个结果。编写代码如下：

import torch
from document_preprocess import get_vector
from pymilvus import Collectioncollection = Collection("document")  # Get an existing collection.
collection.load()
DEVICE = "cuda" if torch.cuda.is_available() else "mps" if torch.backends.mps.is_available() else "cpu"# 定义查询函数
def search_similar_text(input_text):# 将输入文本转换为向量input_vector = get_vector(input_text)# 查询前三个最匹配的向量IDsimilarity = collection.search(data=[input_vector],anns_field="vector",param={"metric_type": "IP", "params": {"nprobe": 10}, "offset": 0},limit=3,expr=None,consistency_level="Strong")ids = similarity[0].ids# 通过ID查询出对应的知识库文档res = collection.query(expr=f"id in {ids}",offset=0,limit=3,output_fields=["id", "content", "title"],consistency_level="Strong")print(res)return resif __name__ == "__main__":question = input('Please enter your question: ')search_similar_text(question)

上面我们通过向量索引库计算查询出了与问题最为接近的文档并打印了出来，接下来就到了最终的获取模型回答的环节了。

通过提示模板获取准确回答

在这一步，我们需要加载ChatGLM的预训练模型，并获取回答。

from transformers import AutoModel, AutoTokenizer
from knowledge_query import search_similar_texttokenizer = AutoTokenizer.from_pretrained("THUDM/chatglm-6b-int4", trust_remote_code=True)
model = AutoModel.from_pretrained("THUDM/chatglm-6b-int4", trust_remote_code=True).half().cuda()
model = model.eval()def predict(input, max_length=2048, top_p=0.7, temperature=0.95, history=[]):res = search_similar_text(input)prompt_template = f"""基于以下已知信息，简洁和专业的来回答用户的问题。
如果无法从中得到答案，请说 "当前会话仅支持解决一个类型的问题，请清空历史信息重试"，不允许在答案中添加编造成分，答案请使用中文。已知内容:
{res}问题:
{input}
"""query = prompt_templatefor response, history in model.stream_chat(tokenizer, query, history, max_length=max_length, top_p=top_p,temperature=temperature):chatbot[-1] = (parse_text(input), parse_text(response))yield chatbot, history

上面使用了提示模板的方式，将我们查询出来的文档作为提示内容交给模型进行推理回答。到此，我们就简单实现了一个基于知识库的问答应用。

如果你想在web上像chatgpt一样提问，也可以丰富一下上面的代码

from transformers import AutoModel, AutoTokenizer
import gradio as gr
import mdtex2htmlfrom knowledge_query import search_similar_texttokenizer = AutoTokenizer.from_pretrained("THUDM/chatglm-6b-int4", trust_remote_code=True)
model = AutoModel.from_pretrained("THUDM/chatglm-6b-int4", trust_remote_code=True).half().cuda()
model = model.eval()
is_knowledge = True"""Override Chatbot.postprocess"""def postprocess(self, y):if y is None:return []for i, (message, response) in enumerate(y):y[i] = (None if message is None else mdtex2html.convert((message)),None if response is None else mdtex2html.convert(response),)return ygr.Chatbot.postprocess = postprocessdef parse_text(text):"""copy from https://github.com/GaiZhenbiao/ChuanhuChatGPT/"""lines = text.split("\n")lines = [line for line in lines if line != ""]count = 0for i, line in enumerate(lines):if "```" in line:count += 1items = line.split('`')if count % 2 == 1:lines[i] = f'<pre><code class="language-{items[-1]}">'else:lines[i] = f'<br></code></pre>'else:if i > 0:if count % 2 == 1:line = line.replace("`", "\`")line = line.replace("<", "&lt;")line = line.replace(">", "&gt;")line = line.replace(" ", "&nbsp;")line = line.replace("*", "&ast;")line = line.replace("_", "&lowbar;")line = line.replace("-", "&#45;")line = line.replace(".", "&#46;")line = line.replace("!", "&#33;")line = line.replace("(", "&#40;")line = line.replace(")", "&#41;")line = line.replace("$", "&#36;")lines[i] = "<br>"+linetext = "".join(lines)return textdef predict(input, chatbot, max_length, top_p, temperature, history):global is_knowledgechatbot.append((parse_text(input), ""))query = inputif is_knowledge:res = search_similar_text(input)prompt_template = f"""基于以下已知信息，简洁和专业的来回答用户的问题。
如果无法从中得到答案，请说 "当前会话仅支持解决一个类型的问题，请清空历史信息重试"，不允许在答案中添加编造成分，答案请使用中文。已知内容:
{res}问题:
{input}
"""query = prompt_templateis_knowledge = Falsefor response, history in model.stream_chat(tokenizer, query, history, max_length=max_length, top_p=top_p,temperature=temperature):chatbot[-1] = (parse_text(input), parse_text(response))yield chatbot, historydef reset_user_input():return gr.update(value='')def reset_state():global is_knowledgeis_knowledge = Falsereturn [], []with gr.Blocks() as demo:gr.HTML("""<h1 align="center">ChatGLM</h1>""")chatbot = gr.Chatbot()with gr.Row():with gr.Column(scale=4):with gr.Column(scale=12):user_input = gr.Textbox(show_label=False, placeholder="Input...", lines=10).style(container=False)with gr.Column(min_width=32, scale=1):submitBtn = gr.Button("Submit", variant="primary")with gr.Column(scale=1):emptyBtn = gr.Button("Clear History")max_length = gr.Slider(0, 4096, value=2048, step=1.0, label="Maximum length", interactive=True)top_p = gr.Slider(0, 1, value=0.7, step=0.01, label="Top P", interactive=True)temperature = gr.Slider(0, 1, value=0.95, step=0.01, label="Temperature", interactive=True)history = gr.State([])submitBtn.click(predict, [user_input, chatbot, max_length, top_p, temperature, history], [chatbot, history],show_progress=True)submitBtn.click(reset_user_input, [], [user_input])emptyBtn.click(reset_state, outputs=[chatbot, history], show_progress=True)demo.queue().launch(share=False, inbrowser=True)

把ChatGLM中的web_demo代码简单改写，我们就得到了一个一模一样的前端应用，不同的是它现在可以基于我们的知识库来回答问题。

小结

上述内容仅仅介绍了最简单的通过向量索引库加AI模型加提示工程来实现知识库问答的方案，其中向量索引和文档的处理非常原始与粗糙，想要实现更加精准的匹配还需要根据实际文档内容和场景来进行修改。

相关代码已上传github knowledge_with_chatglm感兴趣的同学可以 clone 下来跑一跑

使用langchain改进代码

langchain最为目前非常火的开源库，用于知识库问答也能极大的增加开发效率并且降低工作量。例如上述的文档预处理和用户提问匹配知识库两个步骤，我们用了很多代码编写来实现这个功能。但是当我们使用langchain之后就变得简单起来，下面给出代码示例：

from langchain.vectorstores import Milvus
from langchain.document_loaders import DirectoryLoader, TextLoader
from langchain.text_splitter import CharacterTextSplitter
from langchain.embeddings import HuggingFaceEmbeddings# 加载文件夹中的所有txt类型的文件
loader = DirectoryLoader('./knowledge/', glob='**/*.txt', show_progress=True, loader_cls=TextLoader,loader_kwargs={"encoding": "utf-8"})
documents = loader.load()# 初始化加载器
text_splitter = CharacterTextSplitter(chunk_size=100, chunk_overlap=0)
# 切割加载的 document
split_docs = text_splitter.split_documents(documents)embeddings = HuggingFaceEmbeddings(model_name="shibing624/text2vec-base-chinese")
vector_db = Milvus.from_documents(split_docs, embeddings, connection_args={"host": "127.0.0.1", "port": "19530"},collection_name="langchain_knowledge", drop_old=True)

通过上述代码可以看到，langchain为我们封装好了非常多的工具。例如DirectoryLoader和TextLoader可以直接让我们加载文档，配合CharacterTextSplitter可以将加载的文档分割成设定好的一片一片的集合。与此同时使用langchain提供的向量数据库工具，可以轻松将文档向量化并持久化储存。这样仅仅六行代码我们就完成了之前几十行代码才能完成的工作，且不必考虑如何创建字段，维护数据库等。

本文链接：https://my.lmcjl.com/post/18432.html