构建RAG应用-Datawhale笔记

将LLM接入LangChain

LangChain 为基于 LLM 开发自定义应用提供了高效的开发框架，便于开发者迅速地激发 LLM 的强大能力，搭建 LLM 应用。LangChain 也同样支持多种大模型，内置了 OpenAI、LLAMA 等大模型的调用接口。

基于LangChain调用ChatGPT

我这里使用的AzureChatOpenAI（免去在服务器上搭魔法的麻烦）

llm = AzureChatOpenAI(azure_endpoint="https://xxx.com/",
                    openai_api_key="xxx",
                    api_version="2024-02-01",
                    temperature=0.0,
                    deployment_name = 'gpt-4') #部署的模型为gpt-4
output = llm.invoke("请你自我介绍一下自己！")
print(output) 

Prompt（提示模板）

在我们开发大模型应用时，大多数情况下不会直接将用户的输入直接传递给 LLM。通常，他们会将用户输入添加到一个较大的文本中，称为提示模板，该文本提供有关当前特定任务的附加上下文。 PromptTemplates 正是帮助解决这个问题！它们捆绑了从用户输入到完全格式化的提示的所有逻辑。

举一个翻译的例子

from langchain.prompts.chat import ChatPromptTemplate
​
template = "你是一个翻译助手，可以帮助我将 {input_language} 翻译成 {output_language}."
human_template = "{text}"
​
chat_prompt = ChatPromptTemplate.from_messages([
    ("system", template),
    ("human", human_template),
])
​
text = "我带着比身体重的行李，\
游入尼罗河底，\
经过几道闪电 看到一堆光圈，\
不确定是不是这里。\
"
messages  = chat_prompt.format_messages(input_language="中文", output_language="英文", text=text)
messages
output  = llm.invoke(messages)
output

结果：

AIMessage(content='I carried luggage heavier than my body and dived into the bottom of the Nile River. After passing through several flashes of lightning, I saw a pile of halos, not sure if this is the place.')

Output parse （输出解析器）

OutputParsers 将语言模型的原始输出转换为可以在下游使用的格式。 OutputParsers 有几种主要类型，包括：

• 将 LLM 文本转换为结构化信息（例如 JSON）

• 将 ChatMessage 转换为字符串

• 将除消息之外的调用返回的额外信息（如 OpenAI 函数调用）转换为字符串

最后，我们将模型输出传递给 output_parser，它是一个 BaseOutputParser，这意味着它接受字符串或 BaseMessage 作为输入。 StrOutputParser 特别简单地将任何输入转换为字符串。

from langchain_core.output_parsers import StrOutputParser
​
output_parser = StrOutputParser() #创建一个实例
output_parser.invoke(output) #传入的output为上一次得到的output

结果：

'I carried luggage heavier than my body and dived into the bottom of the Nile River. After passing through several flashes of lightning, I saw a pile of halos, not sure if this is the place.'

从上面结果可以看到，我们通过输出解析器成功将 ChatMessage 类型的输出解析为了字符串

补充LCEL语法实现链

chain = chat_prompt | llm | output_parser
chain.invoke({"input_language":"中文", "output_language":"英文","text": text})

结果为

'I carried luggage heavier than my body and dived into the bottom of the Nile River. After passing through several flashes of lightning, I saw a pile of halos, not sure if this is the place.'

LCEL（LangChain Expression Language，Langchain的表达式语言），LCEL是一种新的语法，是LangChain工具包的重要补充，他有许多优点，使得我们处理LangChain和代理更加简单方便。

chain = prompt | model | output_parser

上面代码中我们使用 LCEL 将不同的组件拼凑成一个链，在此链中，用户输入传递到提示模板，然后提示模板输出传递到模型，然后模型输出传递到输出解析器。| 的符号类似于 Unix 管道运算符，它将不同的组件链接在一起，将一个组件的输出作为下一个组件的输入。

构建检索问答链

在上一次搭建数据库中，我们已经知道如何根据本地知识文档，搭建一个向量知识库，接下来的内容里，我们将使用搭建好的向量数据库，对 query 查询问题进行召回，并将召回结果和 query 结合起来构建 prompt，输入到大模型中进行问答。

这里补充一下召回的概念：

召回(Recall) 是信息检索系统一个重要概念，指的是从整个数据集中检索出与用户查询条件相关的一小部分数据项的过程。在上述提到的召回，其目的是从向量数据库中找到与用户输入的查询问题（query）语义上最接近或最相关的问题或信息。

加载向量数据库

# 定义 Embeddings
# 向量数据库持久化路径
persist_directory = '../C3 搭建知识库/data_base/vector_db/chroma'
embeddings = AzureOpenAIEmbeddings(
                    azure_deployment="text-embedding-3-large",
                    model="text-embedding-3-large",  #embedding model
                    azure_endpoint="https://xxx.com/",
                    openai_api_key="xxx",
                    api_version="2024-02-01"
            )
# 加载数据库
vectordb = Chroma(
    persist_directory=persist_directory,  # 允许我们将persist_directory目录保存到磁盘上
    embedding_function=embeddings
)
print(f"向量库中存储的数量：{vectordb._collection.count()}")  #20

接下来我们可以测试一下加载的向量数据库，使用一个问题query进行向量检索，

（相似性搜索前，请确保你已安装了 OpenAI 开源的快速分词工具 tiktoken 包：pip install tiktoken）

question = "什么是prompt"
docs = vectordb.similarity_search(question,k=3)
print(f"检索到的内容数：{len(docs)}")
for i, doc in enumerate(docs):
    print(f"检索到的第{i}个内容: \n {doc.page_content}", end="\n-----------------------------------------------------\n")

创建一个LLM

import os 
from langchain_openai import AzureChatOpenAI
llm = AzureChatOpenAI(azure_endpoint="https://xxx.com/",
                    openai_api_key="xxx",
                    api_version="2024-02-01",temperature=0.0,deployment_name = 'gpt-4')
output = llm.invoke("请你自我介绍一下自己！")
print(output) 

结果：

content='您好！我是一个由OpenAI开发的大型语言模型，名为ChatGPT。我的主要功能是理解和生成自然语言文本，这意味着我可以帮助回答问题、提供信息、撰写文章、创作故事，以及执行其他与语言相关的任务。我是基于大量的文本数据训练而成的，这让我能够理解和生成多种语言的文本。虽然我拥有广泛的知识基础，但我的信息是截至到2023年4月的，之后的新信息和事件我可能不会了解。我旨在提供帮助和信息，但请记住，我不是人类，我的回答可能不总是完美无缺。希望我能在您需要的时候提供帮助！' response_metadata={'token_usage': {'completion_tokens': 228, 'prompt_tokens': 20, 'total_tokens': 248}, 'model_name': 'gpt-4', 'system_fingerprint': 'fp_2f57f81c11', 'prompt_filter_results': [{'prompt_index': 0, 'content_filter_results': {'hate': {'filtered': False, 'severity': 'safe'}, 'self_harm': {'filtered': False, 'severity': 'safe'}, 'sexual': {'filtered': False, 'severity': 'safe'}, 'violence': {'filtered': False, 'severity': 'safe'}}}], 'finish_reason': 'stop', 'logprobs': None, 'content_filter_results': {'hate': {'filtered': False, 'severity': 'safe'}, 'self_harm': {'filtered': False, 'severity': 'safe'}, 'sexual': {'filtered': False, 'severity': 'safe'}, 'violence': {'filtered': False, 'severity': 'safe'}}}

构建检索问答链

from langchain.prompts import PromptTemplate
​
template = """使用以下上下文来回答最后的问题。请注意答案一定来自上下文内容，如果不是来自上下文内容或者你不知道答案，就说你不知道，不要试图编造答
案。最多使用三句话。尽量使答案简明扼要。总是在回答的最后说“谢谢你的提问！”。
{context}
问题: {question}
"""
​
QA_CHAIN_PROMPT = PromptTemplate(input_variables=["context","question"],
                                 template=template)
#再创建一个基于模板的检索链
from langchain.chains import RetrievalQA
​
qa_chain = RetrievalQA.from_chain_type(llm,
                                       retriever=vectordb.as_retriever(),#从向量数据库中检索，并将结果转换为检索器
                                       return_source_documents=True,
                                       chain_type_kwargs={"prompt":QA_CHAIN_PROMPT})

创建检索 QA 链的方法 RetrievalQA.from_chain_type() 有如下参数：

• llm：指定使用的 LLM

• 指定 chain type : RetrievalQA.from_chain_type(chain_type="map_reduce")，也可以利用load_qa_chain()方法指定chain type。

• 返回源文档：通过RetrievalQA.from_chain_type()方法中指定：return_source_documents=True参数；也可以使用RetrievalQAWithSourceChain()方法，返回源文档的引用（坐标或者叫主键、索引）

• 自定义 prompt ：通过在RetrievalQA.from_chain_type()方法中，指定chain_type_kwargs参数，而该参数：chain_type_kwargs = {"prompt": PROMPT}

基于召回效果和query结合起来的prompt效果

question_1 = "什么是南瓜书？"
question_2 = "王阳明是谁？"
result = qa_chain({"query": question_1})
print("大模型+知识库后回答 question_1 的结果：")
print(result["result"])
result = qa_chain({"query": question_2})
print("大模型+知识库后回答 question_2 的结果：")
print(result["result"])

结果：

大模型+知识库后回答 question_1 的结果：
南瓜书是一本旨在对《机器学习》（西瓜书）里比较难理解的公式加以解析，并对部分公式补充具体推导细节的书籍，主要面向数学基础较弱的读者，帮助他们更好地理解机器学习的相关知识。谢谢你的提问！
大模型+知识库后回答 question_2 的结果：
我不知道，谢谢你的提问！

添加历史对话的记忆功能

现在我们已经实现了通过上传本地知识文档，然后将他们保存到向量知识库，通过将查询问题与向量知识库的召回结果进行结合输入到 LLM 中，我们就得到了一个相比于直接让 LLM 回答要好得多的结果。在与语言模型交互时，你可能已经注意到一个关键问题 - 它们并不记得你之前的交流内容。

解决办法是使用 ConversationBufferMemory ，它保存聊天消息历史记录的列表，这些历史记录将在回答问题时与问题一起传递给聊天机器人，从而将它们添加到上下文中。

from langchain.memory import ConversationBufferMemory
​
memory = ConversationBufferMemory(
    memory_key="chat_history",  # 与 prompt 的输入变量保持一致。
    return_messages=True  # 将以消息列表的形式返回聊天记录，而不是单个字符串
)

对话检索链

对话检索链（ConversationalRetrievalChain）在检索问答链的基础上，增加了处理对话历史的能力。

它的工作流程是:

1. 将之前的对话与新问题合并生成一个完整的查询语句。

2. 在向量数据库中搜索该查询的相关文档。

3. 获取结果后,存储所有答案到对话记忆区。

4. 用户可在 UI 中查看完整的对话流程。

from langchain.chains import ConversationalRetrievalChain
​
retriever=vectordb.as_retriever()
​
qa = ConversationalRetrievalChain.from_llm(
    llm,
    retriever=retriever,
    memory=memory
)
question = "我可以学习到关于提示工程的知识吗？"
result = qa({"question": question})
print(result['answer'])

结果：

是的，您可以学习到关于提示工程的知识。本模块内容基于吴恩达老师的《Prompt Engineering for Developer》课程编写，旨在分享使用提示词开发大语言模型应用的最佳实践和技巧。课程将介绍设计高效提示的原则，包括编写清晰、具体的指令和给予模型充足思考时间等。通过学习这些内容，您可以更好地利用大语言模型的性能，构建出色的语言模型应用。

部署知识助手

使用Streamlit

Streamlit 是一种快速便捷的方法，可以直接在 Python 中通过友好的 Web 界面演示机器学习模型。在本课程中，我们将学习如何使用它为生成式人工智能应用程序构建用户界面。在构建了机器学习模型后，如果你想构建一个 demo 给其他人看，也许是为了获得反馈并推动系统的改进，或者只是因为你觉得这个系统很酷，所以想演示一下：Streamlit 可以让您通过 Python 接口程序快速实现这一目标，而无需编写任何前端、网页或 JavaScript 代码。

import streamlit as st
from langchain_openai import AzureChatOpenAI
import os
from langchain_openai import AzureOpenAIEmbeddings
from langchain_core.output_parsers import StrOutputParser
from langchain.prompts import PromptTemplate
from langchain.chains import RetrievalQA
from langchain.vectorstores.chroma import Chroma
from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationalRetrievalChain
from dotenv import load_dotenv, find_dotenv
_ = load_dotenv(find_dotenv())  
​
def generate_response(input_text):
    llm = AzureChatOpenAI(azure_endpoint="https:xxx.com/",
                      api_key="xxx",
                      azure_deployment="gpt-4",
                      api_version="2024-02-15-preview",
                      temperature=0.7)
    output = llm.invoke(input_text)
    output_parser = StrOutputParser()
    output = output_parser.invoke(output)
    #st.info(output)
    return output
​
def get_vectordb():
    # 定义 Embeddings
    embeddings = AzureOpenAIEmbeddings(
                    azure_deployment="text-embedding-3-large",
                    model="text-embedding-3-large",
                    azure_endpoint="xxx.com/",
                    openai_api_key="xxx",
                    api_version="2024-02-01"
            )
    # 向量数据库持久化路径
    persist_directory = 'data_base/vector_db/chroma'
    # 加载数据库
    vectordb = Chroma(
        persist_directory=persist_directory, 
        embedding_function=embeddings
    )
    return vectordb
​
#带有历史记录的问答链
def get_chat_qa_chain(question:str):
    vectordb = get_vectordb()
    llm = AzureChatOpenAI(azure_endpoint="https:xxx.com/",
                      api_key="xxx",
                      azure_deployment="gpt-4",
                      api_version="2024-02-15-preview",
                      temperature=0.0)
    memory = ConversationBufferMemory(
        memory_key="chat_history",  # 与 prompt 的输入变量保持一致。
        return_messages=True  # 将以消息列表的形式返回聊天记录，而不是单个字符串
    )
    retriever=vectordb.as_retriever()
    qa = ConversationalRetrievalChain.from_llm(
        llm,
        retriever=retriever,
        memory=memory
    )
    result = qa({"question": question})
    return result['answer']
​
#不带历史记录的问答链
def get_qa_chain(question:str):
    vectordb = get_vectordb()
    llm = AzureChatOpenAI(azure_endpoint="https:xxx.com/",
                      api_key="xxx",
                      azure_deployment="gpt-4",
                      api_version="2024-02-15-preview",
                      temperature=0.0)
    template = """使用以下上下文来回答最后的问题。请注意答案一定要来自上下文的内容,如果不是来自上下文的内容或者你不知道答案，就说你不知道，不要试图编造答
        案。最多使用三句话。尽量使答案简明扼要。总是在回答的最后说“谢谢你的提问！”。
        {context}
        问题: {question}
        """
    QA_CHAIN_PROMPT = PromptTemplate(input_variables=["context","question"],
                                 template=template)
    qa_chain = RetrievalQA.from_chain_type(llm,
                                       retriever=vectordb.as_retriever(),
                                       return_source_documents=True,
                                       chain_type_kwargs={"prompt":QA_CHAIN_PROMPT})
    result = qa_chain({"query": question})
    return result["result"]
​
​
# Streamlit 应用程序界面
def main():
    st.title('🦜🔗 动手学大模型应用开发')
    # 添加一个选择按钮来选择不同的模型
    #selected_method = st.sidebar.selectbox("选择模式", ["qa_chain", "chat_qa_chain", "None"])
    selected_method = st.radio(
        "你想选择哪种模式进行对话？",
        ["None", "qa_chain", "chat_qa_chain"],
        captions = ["不使用检索问答的普通模式", "不带历史记录的检索问答模式", "带历史记录的检索问答模式"])
​
    # 用于跟踪对话历史
    if 'messages' not in st.session_state:
        st.session_state.messages = []
​
    messages = st.container(height=300)
    if prompt := st.chat_input("Say something"):
        # 将用户输入添加到对话历史中
        st.session_state.messages.append({"role": "user", "text": prompt})
​
        if selected_method == "None":
            # 调用 respond 函数获取回答
            answer = generate_response(prompt)
        elif selected_method == "qa_chain":
            answer = get_qa_chain(prompt)
        elif selected_method == "chat_qa_chain":
            answer = get_chat_qa_chain(prompt)
​
        # 检查回答是否为 None
        if answer is not None:
            # 将LLM的回答添加到对话历史中
            st.session_state.messages.append({"role": "assistant", "text": answer})
​
        # 显示整个对话历史
        for message in st.session_state.messages:
            if message["role"] == "user":
                messages.chat_message("user").write(message["text"])
            elif message["role"] == "assistant":
                messages.chat_message("assistant").write(message["text"])   
​
​
if __name__ == "__main__":
    main()

​

在终端运行这个代码文件

 streamlit run streamlit_app.py

访问给出的URL地址，尝试问个问题：

至此，搭建完成。