背景

目前由于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-etcd    image: quay.io/coreos/etcd:v3.5.0    environment:      - ETCD_AUTO_COMPACTION_MODE=revision      - ETCD_AUTO_COMPACTION_RETENTION=1000      - ETCD_QUOTA_BACKEND_BYTES=4294967296      - ETCD_SNAPSHOT_COUNT=50000    volumes:      - ${DOCKER_VOLUME_DIRECTORY:-.}/volumes/etcd:/etcd    command: etcd -advertise-client-urls=http://127.0.0.1:2379 -listen-client-urls http://0.0.0.0:2379 --data-dir /etcd  minio:    container_name: milvus-minio    image: minio/minio:RELEASE.2023-03-20T20-16-18Z    environment:      MINIO_ACCESS_KEY: minioadmin      MINIO_SECRET_KEY: minioadmin    volumes:      - ${DOCKER_VOLUME_DIRECTORY:-.}/volumes/minio:/minio_data    command: minio server /minio_data    healthcheck:      test: ["CMD", "curl", "-f", "http://localhost:9000/minio/health/live"]      interval: 30s      timeout: 20s      retries: 3  standalone:    container_name: milvus-standalone    image: milvusdb/milvus:v2.2.5    command: ["milvus", "run", "standalone"]    environment:      ETCD_ENDPOINTS: etcd:2379      MINIO_ADDRESS: minio:9000    volumes:      - ${DOCKER_VOLUME_DIRECTORY:-.}/volumes/milvus:/var/lib/milvus    ports:      - "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 osimport reimport jiebaimport torchimport pandas as pdfrom pymilvus import utilityfrom pymilvus import connections, CollectionSchema, FieldSchema, Collection, DataTypefrom transformers import AutoTokenizer, AutoModelconnections.connect(    alias="default",    host='localhost',    port='19530')# 定义集合名称和维度collection_name = "document"dimension = 768docs_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，就可以执行下面的代码        # return        utility.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 torchfrom document_preprocess import get_vectorfrom 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)# 查询前三个最匹配的向量ID    similarity = 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, AutoTokenizerfrom 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, AutoTokenizerimport gradio as grimport 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 = 0    for i, line in enumerate(lines):        if "```" in line:            count += 1            items = line.split('`')            if count % 2 == 1:                lines[i] = f'
{items[-1]}">'            else:                lines[i] = f'
'        else:            if i > 0:                if count % 2 == 1:                    line = line.replace("`", "\`")                    line = line.replace("<", "<")                    line = line.replace(">", ">")                    line = line.replace(" ", " ")                    line = line.replace("*", "*")                    line = line.replace("_", "_")                    line = line.replace("-", "-")                    line = line.replace(".", ".")                    line = line.replace("!", "!")                    line = line.replace("(", "(")                    line = line.replace(")", ")")                    line = line.replace("$", "$")                lines[i] = "
"+line    text = "".join(lines)    return textdef predict(input, chatbot, max_length, top_p, temperature, history):    global is_knowledge    chatbot.append((parse_text(input), ""))    query = input    if is_knowledge:        res = search_similar_text(input)        prompt_template = f"""基于以下已知信息，简洁和专业的来回答用户的问题。如果无法从中得到答案，请说 "当前会话仅支持解决一个类型的问题，请清空历史信息重试"，不允许在答案中添加编造成分，答案请使用中文。已知内容:{res}问题:{input}"""        query = prompt_template        is_knowledge = False    for 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_knowledge    is_knowledge = False    return [], []with gr.Blocks() as demo:    gr.HTML("""
ChatGLM
""")    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 Milvusfrom langchain.document_loaders import DirectoryLoader, TextLoaderfrom langchain.text_splitter import CharacterTextSplitterfrom 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)# 切割加载的 documentsplit_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://blog.csdn.net/a914541185/article/details/130150101