Chainlit集成Langchain并使用通义千问AI知识库高级检索（多重查询）网页对话应用教程

前言

之前写过几篇利用Chainlit集成Langchain和国内通义千问大模型集成的知识库检索增加的网页对话应用的技术文章。文章中关于Langchain的知识库检索只是入门级别的教学，本篇文章针对Langchain的知识库高级检索技术和之前对话应用的代码进行完善。
本次主要改进的点

使用自定义提示词，之前使用的是系统默认提示词，全英文的提示词，回答的时候，经常会出现英文，这次使用中文的提示词加国内的大模型，出现英文的回答大大降低，而且还用了中文设置了系统角色的提示词。
使用多查询检索技术，即将用户问题列变成三个相似的问题，分别对知识库进行检索，并将检索结果合并处理给AI。大大提升了知识检索的准确度。

以前的文章：
《Chainlit集成Langchain并使用通义千问和智谱AI实现AI知识库检索网页对话应用》
《使用Chainlit接入通义千问快速实现一个本地文档知识问答机器人》
《使用Chainlit接入通义千问快速实现一个本地文档知识问答机器人增强版》

快速上手

创建一个文件，例如“chainlit_chat”

mkdir chainlit_chat

进入 chainlit_chat文件夹下，执行命令创建python 虚拟环境空间(需要提前安装好python sdk。 Chainlit 需要python>=3.8。,具体操作，由于文章长度问题就不在叙述，自行百度)，命令如下：

python -m venv .venv

这一步是避免python第三方库冲突，省事版可以跳过
.venv是创建的虚拟空间文件夹可以自定义

接下来激活你创建虚拟空间，命令如下：

#linux or mac
source .venv/bin/activate
#windows
.venv\Scripts\activate

在项目根目录下创建`requirements.txt`，内容如下：

langchain
chainlit
openai
chromadb
tiktoken
pymupdf
langchain_community
dashscope~=1.20.3

执行以下命令安装依赖：

pip install -r .\requirements.txt

安装后，项目根目录下会多出.chainlit 和.files文件夹和chainlit.md文件

只使用通义千问的DashScope模型服务灵积的接口

在项目根目录下创建`.env`环境变量，配置如下：

DASHSCOPE_API_KEY="sk-api_key"

DASHSCOPE_API_KEY 是阿里dashscope的服务的APIkey，代码中使用DashScope的sdk实现，所以不需要配置base_url。默认就是阿里的base_url。
阿里模型接口地址 https://dashscope.console.aliyun.com/model

在项目根目录下创建app.py文件，代码如下：

from pathlib import Path
from typing import List

import chainlit as cl
from langchain.callbacks.base import BaseCallbackHandler
from langchain.prompts import ChatPromptTemplate
from langchain.retrievers import MultiQueryRetriever
from langchain.schema import Document
from langchain.schema import StrOutputParser
from langchain.schema.runnable import RunnablePassthrough, RunnableConfig
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import (
    PyMuPDFLoader, CSVLoader, TextLoader, Docx2txtLoader
)
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_community.llms import Tongyi
from langchain_community.vectorstores import Chroma
from langchain_core.messages import SystemMessage
from langchain_core.prompts import HumanMessagePromptTemplate

chunk_size = 1024
chunk_overlap = 100

FILE_STORAGE_PATH = "data_file"

embeddings_model = DashScopeEmbeddings()


@cl.cache
def process_files(file_storage_path: str):
    file_directory = Path(file_storage_path)
    docs = []  # type: List[Document]
    text_splitter = RecursiveCharacterTextSplitter(chunk_size=chunk_size, chunk_overlap=chunk_overlap)

    for file_path in file_directory.glob("*.pdf"):
        loader = PyMuPDFLoader(str(file_path))
        docs += text_splitter.split_documents(loader.load())

    for file_path in file_directory.glob("*.csv"):
        loader = CSVLoader(str(file_path), encoding="UTF-8")
        docs += text_splitter.split_documents(loader.load())

    for file_path in file_directory.glob("*.txt"):
        loader = TextLoader(str(file_path), encoding="UTF-8")
        docs += text_splitter.split_documents(loader.load())

    for file_path in file_directory.glob("*.doc"):
        loader = Docx2txtLoader(str(file_path))
        docs += text_splitter.split_documents(loader.load())

    vector_store = Chroma.from_documents(docs, embeddings_model)

    return vector_store


llm = Tongyi(model='qwen-plus')
doc_search = process_files(FILE_STORAGE_PATH)
retriever = doc_search.as_retriever(search_kwargs={"k": 9})
multi_retriever = MultiQueryRetriever.from_llm(
    retriever=retriever, llm=llm
)



@cl.on_chat_start
async def on_chat_start():
    system_prompt = """
    # 角色
    石家庄医专学校智能客服
    ## 能力
    - 利用用户提供的省份、文理科、分数、位次等信息，比较分数和位次，给出关于本校的报考建议。
    - 解答用户提问关于学校相关问题
    """
    human_template = """Answer the question based only on the following context:
    {context}
    回答要求：
    - 如果你不清楚答案，你需要澄清。
    - 避免提及你是从 <Data></Data> 获取的知识。
    - 保持答案与 <Data></Data> 中描述的一致。
    - 使用 Markdown 语法优化回答格式。
    - 使用与问题相同的语言回答。

    问题: {question}
    """
    prompt = ChatPromptTemplate.from_messages([
        SystemMessage(content=system_prompt),
        HumanMessagePromptTemplate.from_template(human_template)
    ])

    runnable = (
            {"context": multi_retriever, "question": RunnablePassthrough()}
            | prompt
            | llm
            | StrOutputParser()
    )
    cl.user_session.set("runnable", runnable)


@cl.on_message
async def on_message(message: cl.Message):
    runnable = cl.user_session.get("runnable")
    msg = cl.Message(content="")

    class PostMessageHandler(BaseCallbackHandler):
        """
        用于处理检索程序和LLM进程的回调处理程序。
        用于将检索到的文档的源作为Chainlit元素发布。
        """

        def __init__(self, msg: cl.Message):
            BaseCallbackHandler.__init__(self)
            self.msg = msg
            self.sources = []  # To store unique pairs

        def on_retriever_end(self, documents, *, run_id, parent_run_id, **kwargs):
            for d in documents:
                source_page_pair = (d.page_content, d.metadata['row'])
                if source_page_pair not in self.sources:
                    self.sources.append(source_page_pair)

        async def on_llm_end(self, response, *, run_id, parent_run_id, **kwargs):
            if len(self.sources):
                source_names = []
                for page_content, row in self.sources:
                    source_name = f"source_{row}"
                    source_names.append(source_name)
                    self.msg.elements.append(
                        cl.Text(content=page_content, name=source_name, display="side")
                    )
                await self.msg.stream_token(f"\n\n **数据来源**: {', '.join(source_names)}")

    async for chunk in runnable.astream(
            message.content,
            config=RunnableConfig(callbacks=[
                cl.LangchainCallbackHandler(),
                PostMessageHandler(msg)
            ]),
    ):
        await msg.stream_token(chunk)

    await msg.send()