基于LangChain构建RAG应用

news2024/11/15 21:27:03

banner

前言

Hello，大家好，我是GISer Liu😁，一名热爱AI技术的GIS开发者，上一篇文章中我们详细介绍了RAG的核心思想以及搭建向量数据库的完整过程；😲

本文将基于上一篇文章的结果进行开发，主要内容为：

  • 将LLM接入LangChain：选择LLM，然后在LangChain中使用；
  • 构建检索问答链：使用语法构建RAG问答链
  • 部署知识库助手：使用streamlit部署项目；

帮助读者快速构建RAG应用，并部署在阿里云服务器上；

一、LLM接入LangChain

1. LangChain中LLM组成

model I/O

与LLM API原生调用方法不同，在LangChain中，LLM调用过程高度抽象，其由模型(Model）、提示词模版(Prompt Template)、输出解析器(Output parser) 组成；如上图所示:

  • 提示词模版：将用户输入添加到一个提示词模板中，这个提示词模版提供有关当前特定任务的附加上下文，构建出适用于特定任务的提示词；
  • 模型：LangChain集成的各大平台模型，如ChatGPT、Claude、Mistral、ChatGLM；
  • 输出解析器：OutputParsers 将LLM的原始输出转换为可以在下游使用的格式，如json;

具体内容如下：

①提示词模版

作者在本系列的第一篇文章中就强调：提示词工程是LLM开发者重要的知识基础和必备技能，而此处的提示词模版就是提示词工程的一个应用载体；通过构建提示词模版，我们可以减少开发过程中的输入，优化用户体验，提高RAG应用处理速度；

核心思想
  • 1. 模版创建
  • 2. 用户输入
  • 3. 提示词打包
原生实现
# 原生构建提示词模版
template = """请你将由三个反引号分割的文本翻译成英文！\
text: ```{text}```
"""
# 用户输入
text = "Babylon是一个开源的JavaScript解析器和代码转换工具，用于分析和转换JavaScript代码。它是由Microsoft开发的，旨在提供一个高性能、可扩展和可靠的解析器，用于支持各种JavaScript工具和框架。"
# prompt打包
prompt = template.format(text=text)
print(prompt)

运行查看结果：
prompt

LangChain实现
# LangChain实现
from langchain.prompts.chat import ChatPromptTemplate

# system prompt template
template = "你当前是一个翻译助手，请将 {input_language} 翻译成 {output_language}."


human_template = "翻译内容：{text}"

chat_prompt = ChatPromptTemplate.from_messages([
    ("system", template),
    ("human", human_template),
])

text = "Babylon是一个开源的JavaScript解析器和代码转换工具，用于分析和转换JavaScript代码。"
messages  = chat_prompt.format_messages(input_language="中文", output_language="英文", text=text)
print(messages)
print("---------")
print(messages[0].content)

运行查看结果：
langchain-prompt-template
可以看到，在LangChain中，ChatPromptTemplate不仅支持用户输入HumanMessage的提示词模版，也支持系统提示system prompt的提示词模版；通过区分系统提示词和用户输入提示词，可以在进行重复性任务时，固定system prompt，只改变用户输入以降低工作量和时间成本；

Langchain中，一个 ChatPromptTemplate 是一个 ChatMessageTemplate 的列表。

  • ChatMessageTemplate 系统提示词模版，
  • ChatMessageTemplate 用户输入提示词模版

我们将打包好的提示词输入给LLM输出一下：

# 运行测试
import os
import openai
from dotenv import load_dotenv, find_dotenv
from langchain_mistralai.chat_models import ChatMistralAI

# 读取本地/项目的环境变量。
# find_dotenv()寻找并定位.env文件的路径
# load_dotenv()读取该.env文件，并将其中的环境变量加载到当前的运行环境中  
_ = load_dotenv(find_dotenv()); # 如果环境变量是全局的，这行代码可以省略

# 获取环境变量 OPENAI_API_KEY
api_key = os.environ['MISTRAL_API_KEY']

# 实例化一个ChatMistralAI类：然后设置其Mistral API_KEY；
llm = ChatMistralAI(api_key=api_key)
print(llm)

# 做一个输出
output = llm.invoke(messages)
print(output.content) # 其返回结果也是一个Message对象

translation
可以看到，内容已经成功翻译；

②模型

在过去的文章中，作者说明了LLM是RAG应用的核心，而LangChain 提供了对于多种大模型的封装，基于 LangChain 的接口可以便捷地调用 LLM 并将其集成在以 LangChain 为基础框架搭建的RAG个人应用中😲。

我们在此简述如何使用 LangChain 接口来调用 Mistral API Key。

这里作者本来打算使用ChatGPT的API，但是昨天OpenAI宣布禁止国内使用OpenAI的API key了，本人也就放弃了使用OpenAI的API Key，使用Mistral AI； 🤔🤔

# 使用MistralAI 的 LLM

import os
from dotenv import load_dotenv, find_dotenv
from langchain_mistralai.chat_models import ChatMistralAI

# 读取本地/项目的环境变量。
# find_dotenv()寻找并定位.env文件的路径
# load_dotenv()读取该.env文件，并将其中的环境变量加载到当前的运行环境中  
_ = load_dotenv(find_dotenv()); # 如果环境变量是全局的，这行代码可以省略

# 获取环境变量 OPENAI_API_KEY
api_key = os.environ['MISTRAL_API_KEY']

# 实例化一个ChatMistralAI类：然后设置其Mistral API_KEY；
llm = ChatMistralAI(api_key=api_key)
# 输出测试
output = llm.invoke("介绍一下什么是Babylon?") # 既可以输入打包好的Message对象，也可以输入字符串
print(output.content) # 其返回结果也是一个Message对象

运行后可以看到输出结果：
mistral LLM
下面是作者整理的ChatMistralAI支持的参数：

  • cache: 是否缓存响应。如果为真，将使用全局缓存。如果为假，则不使用缓存。
  • callbacks: 要添加到运行跟踪中的回调。
  • custom_get_token_ids: 用于计数令牌的可选编码器。
  • endpoint: 要使用的API 后端URL。默认为’https://api.mistral.ai/v1’。
  • max_concurrent_requests: 要同时执行的最大请求数。默认为64。
  • max_retries: 要重试请求的最大次数。默认为5。
  • max_tokens: 要使用的最大令牌数。
  • metadata: 要添加到运行跟踪中的元数据。
  • mistral_api_key: 要使用的API密钥。约束条件：类型=字符串，writeOnly=真，格式=密码。
  • model: 要使用的模型。默认为’mistral-small’。
  • random_seed: 要使用的随机种子。
  • safe_mode: 是否使用安全模式。默认为false。
  • streaming: 是否使用流式输出。默认为false。
  • tags: 要添加到运行跟踪中的标签。
  • temperature: 用于采样的温度。默认为0.7。
  • timeout: 请求超时（以秒为单位）。默认为120。
  • top_p: 使用核采样解码：考虑至少其概率和为top_p的最小令牌集。必须在[0, 1]的闭区间内。
  • verbose: 是否打印响应文本。可选。
③输出解析器(Output parse)
核心思想：

在实际的开发中，根据具体业务和应用场景的不同，我们希望LLM输出的结果和格式适配当前场景，例如输出JSON数据，TXT数据，或者输出CSV数据，方便下游的业务直接使用；

而LangChain提供的OutputParsers 将LLM的原始输出转换为可以在下游使用的格式。 OutputParsers 有几种主要类型，包括：

  • 将 LLM 文本转换为结构化信息（例如 JSON）
  • 将 ChatMessage 转换为字符串
  • 将除消息之外的调用返回的额外信息（如 OpenAI 函数调用）转换为字符串

这里，作者将以LLM输出解析为Json格式为例进行演示：

原生实现
  • Mistral 原生实现
# Mistral实现原生解析器实现
# 导入环境变量
from dotenv import load_dotenv
import os
import re
from mistralai.client import MistralClient
from mistralai.models.chat_completion import ChatMessage

# 从当前目录中的 .env 文件加载环境变量
load_dotenv()

api_key = os.getenv("MISTRAL_API_KEY")
model = "mistral-large-latest"

# 正则提取任务
def parse_task(content): # 从模型生成中字符串匹配提取生成的代码
    pattern = r'```task(.*?)```'  # 使用非贪婪匹配
    match = re.search(pattern, content, re.DOTALL)
    task = match.group(1) if match else content
    # task = json.loads(task,strict=False)  # 转换为json格式
    return task

user_requirement = "如何在Cesium中集成Babylon？"
prompt :str = f"""
        您是一名任务分析师，您的任务是理解用户需求、并分析和归纳用户意图，生成任务报告。
        你要生成的内容要包裹在```task```中，包含的字段有任务名、任务类型、任务内容、任务发布时间、任务完成状态，如下案例：
            "task_name":"xxx",
            "task_type":"xxx",
            "task_content":"xxx",
            "task_time":"xxx",
            "task_status":"xxx"
        生成的内容是一个json格式 用大括号json格式扩住,并将将生成的情报信息包裹在```task```中，要求使用中文、完整且精炼的语言进行描述。
        好的，请根据以下用户输入生成任务信息,严格中文输出：
        {user_requirement}
        """
# 创建模型
client = MistralClient(api_key=api_key)

# 模型输入
chat_response = client.chat(
    model=model,
    messages=[ChatMessage(role="user", content=prompt)]
)

# 从模型输出解析任务JSON
task = parse_task(chat_response.choices[0].message.content)
print(task)

Mistral运行结果如下：

(mistral

  • DeepSeek 原生实现
# deepseek 原生实现结构化输出
# 格式化输出内容
import re
import os
import json
from openai import OpenAI
api_key = os.getenv('DEEPSEEK_API_KEY')
base_url = os.getenv('BASE_URL')


# 正则提取任务
def parse_task(content): # 从模型生成中字符串匹配提取生成的代码
    pattern = r'```task(.*?)```'  # 使用非贪婪匹配
    match = re.search(pattern, content, re.DOTALL)
    task = match.group(1) if match else content
    # task = json.loads(task,strict=False)  # 转换为json格式
    return task

user_requirement = "如何在Cesium中集成Babylon？"
prompt :str = f"""
        您是一名任务分析师，您的任务是理解用户需求、并分析和归纳用户意图，生成任务报告。
        你要生成的内容要包裹在```task```中，包含的字段有任务名、任务类型、任务内容、任务发布时间、任务完成状态，如下案例：
            "task_name":"xxx",
            "task_type":"xxx",
            "task_content":"xxx",
            "task_time":"xxx",
            "task_status":"xxx"
        生成的内容是一个json格式 用大括号json格式扩住,并将将生成的情报信息包裹在```task```中，要求使用中文、完整且精炼的语言进行描述。
        好的，请根据以下用户输入生成任务信息,严格中文输出：
        {user_requirement}
        """
client = OpenAI(api_key=api_key, base_url=base_url)
response = client.chat.completions.create(
    model="deepseek-chat",
    messages=[
        {"role": "system", "content": "GIS开发全栈工程师"},
        {"role": "user", "content": prompt},
    ],
    stream=False
)
task = parse_task(response.choices[0].message.content)
print(task)

DeepSeek运行结果如下：

deepseek

不难看出，效果很好！😀

LangChain实现
# 导入所需的库和模块
from langchain_core.output_parsers import JsonOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_mistralai.chat_models import ChatMistralAI

# 获取环境变量 MISTRAL_API_KEY
api_key = os.environ['MISTRAL_API_KEY']

# 初始化一个 ChatMistralAI 模型实例，并设置温度为 0
llm = ChatMistralAI(temperature=0,api_key=api_key)

# 定义一个名为 Task 的 Pydantic 模型，用于表示任务数据结构
# 该模型包含五个字段：task_name、task_type、task_content、task_time 和 task_status
class Task(BaseModel):
    task_name: str = Field(description="分析任务，得到任务名称")
    task_type: str = Field(description="分析任务，得到任务的类型")
    task_content: str = Field(description="分析任务，得到任务的内容")
    task_time: str = Field(description="分析任务，得到任务的时间")
    task_status: str = Field(description="分析任务，得到任务的状态,完成或未完成")

# 定义一个任务查询字符串
task_query = "如何在Cesium中集成Babylon？"

# 初始化一个 JsonOutputParser 实例，用于解析模型生成的 JSON 输出
# 并将其转换为 Task 模型实例
parser = JsonOutputParser(pydantic_object=Task)

# 定义一个 PromptTemplate 实例，用于生成提示字符串
# 该模板包含一个查询变量，用于在提示字符串中插入用户输入的查询
# 该模板还包含一个格式化说明变量，用于在提示字符串中插入解析器生成的格式化说明
prompt = PromptTemplate(
    template="根据用户输入的问题得到任务JSON.\n{format_instructions}\n{query}\n",
    input_variables=["query"],
    partial_variables={"format_instructions": parser.get_format_instructions()},
)

# 创建一个管道，将提示、模型和解析器连接在一起
# 当调用该管道时，它将生成一个提示字符串，将其发送到模型以生成响应，
# 然后将响应解析为 Task 模型实例
task_chain = prompt | llm | parser

# 调用管道，使用用户输入的查询来生成任务数据
task_data = task_chain.invoke({"query": task_query})
print(task_data) # 这里得到的是解析过的结果，而不是message对象

运行后，可以看到结果:
langchian-output-json

🎉🎉🎉🤓我们成功通过JsonOutputParser将ChatMessgae类型的输出解析json格式；

需要注意的是，输出解析器输出的结果就是对应的输出格式，而不是ChatMessage!!!，可以直接进入下游业务；

④LCEL语法

在上面的代码中，作者使用了LCEL语法将这些组件组合为一条链Chain：

task_chain = prompt | llm | parser

该链(Chain)或者工作流(WorkFlow)将会：

    1. 获取输入变量 task_query;
    1. task_query变量会输入提示词模版PromptTemplate打包为一个prompt;
    1. 打包好的Prompt传递给大语言模型LLM;
    1. LLM输出的结果ChatMessage经过输出解析器JsonOutputParser解析；
    1. 最终得到解析后对应格式的结果,如Json；

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

  • LCEL提供了异步、批处理和流处理支持，使代码可以快速在不同服务器中移植。
  • LCEL拥有后备措施，解决LLM格式输出的问题。
  • LCEL增加了LLM的并行性，提高了效率。
  • LCEL内置了日志记录，即使代理变得复杂，有助于理解复杂链条和代理的运行情况。

官方文档：https://python.langchain.com/v0.1/docs/expression_language/why/#lcel

  • 核心思想：它将不同的组件链接在一起，将一个组件的输出作为下一个组件的输入。

LCEL真的方便吗？作者感觉其用起来更加抽象了🤔🤔

2.Langchain集成智谱AI

因为思路和上文的Mistral一致，作者在这里只提供完整代码；
智谱AI

# 智谱AI的集成到Langchain实现
# 导入所需的库和模块
from langchain_core.output_parsers import JsonOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_community.chat_models import ChatZhipuAI
import os

# 获取环境变量 GLM_API_KEY
api_key = os.environ['ZHIPUAI_API_KEY']

# 初始化一个 ChatMistralAI 模型实例，并设置温度为 0
llm = ChatZhipuAI(model="glm-4",temperature=0.5,api_key=api_key)

# 定义一个名为 Task 的 Pydantic 模型，用于表示任务数据结构
# 该模型包含五个字段：task_name、task_type、task_content、task_time 和 task_status
class Task(BaseModel):
    task_name: str = Field(description="分析任务，得到任务名称")
    task_type: str = Field(description="分析任务，得到任务的类型")
    task_content: str = Field(description="分析任务，得到任务的内容")
    task_time: str = Field(description="分析任务，得到任务的时间")
    task_status: str = Field(description="分析任务，得到任务的状态,完成或未完成")

# 定义一个任务查询字符串
task_query = "如何在Cesium中集成Babylon？"

# 初始化一个 JsonOutputParser 实例，用于解析模型生成的 JSON 输出
# 并将其转换为 Task 模型实例
parser = JsonOutputParser(pydantic_object=Task)

# 定义一个 PromptTemplate 实例，用于生成提示字符串
# 该模板包含一个查询变量，用于在提示字符串中插入用户输入的查询
# 该模板还包含一个格式化说明变量，用于在提示字符串中插入解析器生成的格式化说明
prompt = PromptTemplate(
    template="根据用户输入的问题得到任务JSON.\n{format_instructions}\n{query}\n",
    input_variables=["query"],
    partial_variables={"format_instructions": parser.get_format_instructions()},
)

# 创建一个管道，将提示、模型和解析器连接在一起
# 当调用该管道时，它将生成一个提示字符串，将其发送到模型以生成响应，
# 然后将响应解析为 Task 模型实例
task_chain = prompt | llm | parser

# 调用管道，使用用户输入的查询来生成任务数据
task_data = task_chain.invoke({"query": task_query})
print(task_data) # 这里得到的是解析过的结果，而不是message对象

3.Langchain集成Ollama

ollama

# 导入所需的库和模块
from langchain_community.chat_models import ChatOllama
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate

# 初始化一个 ChatOllama 模型实例，并设置模型名称为 "llama3"
# ChatOllama 支持许多可选参数，您可以悬停在 ChatOllama(...) 类上以查看最新支持的参数
llm = ChatOllama(model="qwen") # 这里我选择使用通义千问的模型

# 定义一个 ChatPromptTemplate 实例，用于生成提示字符串
# 该模板包含一个主题变量，用于在提示字符串中插入用户输入的主题
prompt = ChatPromptTemplate.from_template("讲个笑话，主题是： {topic}")

# 创建一个管道，将提示、模型和解析器连接在一起
# 当调用该管道时，它将生成一个提示字符串，将其发送到模型以生成响应，
# 然后将响应解析为字符串
# 这里使用了 LangChain 表达式语言（LCEL）来构建管道
chain = prompt | llm | StrOutputParser()

# 调用管道，使用用户输入的主题来生成短笑话
# 为了简洁起见，将响应打印在终端中
# 您可以使用 LangServe 部署您的应用程序以进行生产
print(chain.invoke({"topic": "Space travel"}))

二、构建检索问答链

rag

在上文中，我们已经学习了Langchain中LLM调用的组成以及LCEL语法的简答使用，现在我们将基于我们上一篇文章的结果与上文内容，构建一个检索问答链，也就是RAG Chain;
先说说我们的实现思路：

    1. 加载向量数据库，保证后续操作访问向量数据库；
    1. 获取用户输入，向量化后在向量数据库中进行向量相似度查询；
    1. 筛选出相似性数据，作为辅助信息的存在，并通过提示词模版打包为Prompt；
    1. 将打包好的提示词输入到LLM中，并将结果解析为字符串后返回；

1.加载向量数据库

我们之前使用的是Faiss向量数据库，这里我们继续使用,向量数据库文件路径如图，在根目录下的./db/GIS_db中：
path

为了后续使用，我们使用上一篇文章中作者封装的Langchain Mistral Embedding类,代码如下：

# 封装Mistral Embedding 
from __future__ import annotations

import logging
import os
from typing import Dict, List, Any

from langchain.embeddings.base import Embeddings
from langchain.pydantic_v1 import BaseModel, root_validator

logger = logging.getLogger(__name__)

class MistralAIEmbeddings(BaseModel, Embeddings):
    """`MistralAI Embeddings` embedding models."""

    client: Any
    """`mistralai.MistralClient`"""

    @root_validator()
    def validate_environment(cls, values: Dict) -> Dict:
        """
        实例化MistralClient为values["client"]

        Args:
            values (Dict): 包含配置信息的字典，必须包含 client 的字段.

        Returns:
            values (Dict): 包含配置信息的字典。如果环境中有mistralai库，则将返回实例化的MistralClient类；否则将报错 'ModuleNotFoundError: No module named 'mistralai''.
        """
        from mistralai.client import MistralClient
        api_key = os.getenv('MISTRAL_API_KEY')
        if not api_key:
            raise ValueError("MISTRAL_API_KEY is not set in the environment variables.")
        values["client"] = MistralClient(api_key=api_key)
        return values

    def embed_query(self, text: str) -> List[float]:
        """
        生成输入文本的 embedding.

        Args:
            texts (str): 要生成 embedding 的文本.

        Return:
            embeddings (List[float]): 输入文本的 embedding，一个浮点数值列表.
        """
        embeddings = self.client.embeddings(
            model="mistral-embed",
            input=[text]
        )
        return embeddings.data[0].embedding

    def embed_documents(self, texts: List[str]) -> List[List[float]]:
        """
        生成输入文本列表的 embedding.

        Args:
            texts (List[str]): 要生成 embedding 的文本列表.

        Returns:
            List[List[float]]: 输入列表中每个文档的 embedding 列表。每个 embedding 都表示为一个浮点值列表。
        """
        return [self.embed_query(text) for text in texts]

加载向量数据库，并测试一下运行状态；代码如下：

# 加载向量数据库
# 使用向量数据库进行检索
from langchain_community.vectorstores import FAISS

# 实例化Embedding模型
embeddings_model = MistralAIEmbeddings()
# 加载向量数据库
loaded_db = FAISS.load_local("./db/GIS_db", embeddings_model, allow_dangerous_deserialization=True)


# 计算相似度并检索最相似的文档
query = "电磁辐射"
docs = loaded_db.similarity_search(query, k=3) # 相似度最高的前3个chunk

# 输出检索结果
for doc in docs:
    print(doc.page_content+"\n-----------------\n")

运行测试代码，查看检索结果：

search_result
可以看到，已经成功检索出与作者输入的"电磁辐射"最相似的前3个内容；说明该向量数据库可用；

2.创建LLM

这里我们最好使用LangChain支持的模型，避免自己封装的问题，可以使用Ollama，Mistral、GLM、讯飞星火等；作者继续使用Mistral，下面是我们的代码：

# 继续使用Mistral
# 导入所需的库和模块
import os
from langchain_mistralai.chat_models import ChatMistralAI

# 获取环境变量 MISTRAL_API_KEY
api_key = os.environ['MISTRAL_API_KEY']

# 初始化一个 ChatMistralAI 模型实例，并设置温度为 0
llm = ChatMistralAI(temperature=0,api_key=api_key)

# 定义一个任务查询字符串
task_query = "如何在Cesium中集成Babylon？中文回答"

# 调用管道，使用用户输入的查询来生成任务数据
result = llm.invoke(task_query)
print(result.content)

keep-mistral

😏😏😏运行成功，说明我们的模型可用；

3.构建检索问答链

LangChain中提供了构建检索问答链RetrievalQA的API接口，这里我们直接使用，代码如下：

# 构建检索问答链
import os
from langchain_mistralai.chat_models import ChatMistralAI
from langchain.prompts import PromptTemplate
from langchain.chains import RetrievalQA
from langchain_community.vectorstores import FAISS

# 实例化Embedding模型
embeddings_model = MistralAIEmbeddings()
# 加载向量数据库
loaded_db = FAISS.load_local("./db/GIS_db", embeddings_model, allow_dangerous_deserialization=True)

# 获取环境变量 MISTRAL_API_KEY
api_key = os.environ['MISTRAL_API_KEY']

# 初始化一个 ChatMistralAI 模型实例，并设置温度为 0
llm = ChatMistralAI(temperature=0,api_key=api_key)

template = """使用以下上下文来回答最后的问题。如果你不知道答案或者不确定结果，只需要你不知道，不要试图编造答
案。最多使用三句话。尽量使答案简明扼要。总是在回答的最后说“谢谢你的提问！要求使用中文回答”。
{context}
问题: {question}
"""
QA_CHAIN_PROMPT = PromptTemplate(input_variables=["context","question"],
                                 template=template)

# 构建检索问答链
qa_chain = RetrievalQA.from_chain_type(llm,
                                       retriever=loaded_db.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。
  • 自定义 prompt ：通过在RetrievalQA.from_chain_type()方法中，指定chain_type_kwargs参数，而该参数：chain_type_kwargs = {“prompt”: PROMPT}
  • 返回源文档：通过RetrievalQA.from_chain_type()方法中指定：return_source_documents=True参数；也可以使用RetrievalQAWithSourceChain()方法，返回源文档的引用（坐标或者叫主键、索引）

测试查看检索效果：

question_1 = "什么是GIS？"
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"])

search_result_analysis

对比无RAG时大模型回答的结果：

# 对比有RAG和无RAG的回答结果
prompt_template = """请回答下列问题:
                            {}""".format(question_1)
### 基于大模型的问答
print(llm.predict(prompt_template)+'\n------------------------\n')

prompt_template1 = """请回答下列问题:
                            {}""".format(question_2)
### 基于大模型的问答
print(llm.predict(prompt_template1))

no-rag-result

GIS说的还凑合，但是王司徒就不清楚了，学历史的同学看看其说的对不对； 🤔🤔🤔

4.添加历史对话功能

当前我们已经实现了：

    1. 文档存储到向量数据库;
    1. 构建基于向量数据库的RAG检索问答链；

而在实际的RAG场景中，我们通常要与RAG应用中进行多轮对话来充分了解我们需要的内容；但是上面的程序只能支持单轮对话，如何才能支持多轮对话呢🤔？

①记忆（Memory）

这里我们介绍一下 LangChain 中的记忆储存模块，即如何将先前的历史对话也Embedding到LLM中的，使其具有连续对话的能力。我们将使用 ConversationBufferMemory ，它保存了聊天消息历史记录chat_history的列表，这些历史记录将在回答问题时与问题一起传递给LLM，从而将它们添加到上下文中，使得LLM拥有记忆功能，API如下：

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

我们将其集成到我们之前的代码中，下面是完整代码：

# 添加历史对话功能
import os
from langchain_mistralai.chat_models import ChatMistralAI
from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationalRetrievalChain
from langchain_community.vectorstores import FAISS
# 封装Mistral Embedding 
from __future__ import annotations

import logging
import os
from typing import Dict, List, Any

from langchain.embeddings.base import Embeddings
from langchain.pydantic_v1 import BaseModel, root_validator

logger = logging.getLogger(__name__)

class MistralAIEmbeddings(BaseModel, Embeddings):
    """`MistralAI Embeddings` embedding models."""

    client: Any
    """`mistralai.MistralClient`"""

    @root_validator()
    def validate_environment(cls, values: Dict) -> Dict:
        """
        实例化MistralClient为values["client"]

        Args:
            values (Dict): 包含配置信息的字典，必须包含 client 的字段.

        Returns:
            values (Dict): 包含配置信息的字典。如果环境中有mistralai库，则将返回实例化的MistralClient类；否则将报错 'ModuleNotFoundError: No module named 'mistralai''.
        """
        from mistralai.client import MistralClient
        api_key = os.getenv('MISTRAL_API_KEY')
        if not api_key:
            raise ValueError("MISTRAL_API_KEY is not set in the environment variables.")
        values["client"] = MistralClient(api_key=api_key)
        return values

    def embed_query(self, text: str) -> List[float]:
        """
        生成输入文本的 embedding.

        Args:
            texts (str): 要生成 embedding 的文本.

        Return:
            embeddings (List[float]): 输入文本的 embedding，一个浮点数值列表.
        """
        embeddings = self.client.embeddings(
            model="mistral-embed",
            input=[text]
        )
        return embeddings.data[0].embedding

    def embed_documents(self, texts: List[str]) -> List[List[float]]:
        """
        生成输入文本列表的 embedding.

        Args:
            texts (List[str]): 要生成 embedding 的文本列表.

        Returns:
            List[List[float]]: 输入列表中每个文档的 embedding 列表。每个 embedding 都表示为一个浮点值列表。
        """
        return [self.embed_query(text) for text in texts]


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

# 实例化Embedding模型
embeddings_model = MistralAIEmbeddings()
# 加载向量数据库
loaded_db = FAISS.load_local("./db/GIS_db", embeddings_model, allow_dangerous_deserialization=True)
# 获取环境变量 MISTRAL_API_KEY
api_key = os.environ['MISTRAL_API_KEY']

# 初始化一个 ChatMistralAI 模型实例，并设置温度为 0
llm = ChatMistralAI(temperature=0,api_key=api_key)
# 构建新的问答链，使用带有记忆的提示模板
retriever=loaded_db.as_retriever()

#构建对话问答链
qa = ConversationalRetrievalChain.from_llm(
    llm,
    retriever=retriever,
    memory=memory,
    verbose=True,
)
question = "如何学习GIS呢？要求中文回答"
result = qa({"question": question})
print(result['answer'])

这里作者增加了verbose=True参数,使其输出历史信息；
运行代码进行测试：
first-result
再次输入测试：

question = "什么意思？中文回答"
result = qa({"question": question})
print(result['answer'])

second_result

可以看到，第二次回答时，LLM理解了提问中"什么意思"所代指的对象——上一轮的 回答，并进行了进一步的解释；

除此之外，这里可以看到其是对用户的输入"什么意思，中文回答"做了一次优化：

  • 从新的用户输入的Prompt中识别意图，得到新的Prompt："可以请你解释一下"GIS"和"地理参考系统、空间分析、地理数据处理等知识"的含义吗？（中文回答）"；
  • 然后打包为新的Prompt通过检索链发给LLM得到结果；

至此，我们的RAG应用代码编写完成！🎉🎉🎉😀，接下来，我们使用Streamlit库构建应用进行部署；

三、部署个人知识库到阿里云

1. Streamlit简介

streamlit

Streamlit 是一个开源的 Python 库，它使得数据科学家和开发者能够快速构建和共享美观的机器学习模型和数据应用程序。使用 Streamlit，用户无需深入了解前端开发，即可创建交互式的 Web 应用。它的设计哲学是简单、快速和直观，使得用户可以通过编写 Python 脚本来定义应用的布局和行为。

2.构建应用程序

作者这里给出完整代码：

from __future__ import annotations
import streamlit as st
import os
import sys
sys.path.append("./")  # 将父目录放入系统路径中
from langchain_mistralai.chat_models import ChatMistralAI
from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationalRetrievalChain
from langchain_community.vectorstores import FAISS
from langchain_core.output_parsers import StrOutputParser
import logging
from typing import Dict, List, Any
from langchain.embeddings.base import Embeddings
from langchain.pydantic_v1 import BaseModel, root_validator, validator
from dotenv import load_dotenv, find_dotenv
_ = load_dotenv(find_dotenv())    # read local .env file

logger = logging.getLogger(__name__)

class MistralAIEmbeddings(BaseModel, Embeddings):
    """`MistralAI Embeddings` embedding models."""

    client: Any
    """`mistralai.MistralClient`"""

    @root_validator(allow_reuse=True)
    def validate_environment(cls, values: Dict) -> Dict:
        """
        实例化MistralClient为values["client"]

        Args:
            values (Dict): 包含配置信息的字典，必须包含 client 的字段.

        Returns:
            values (Dict): 包含配置信息的字典。如果环境中有mistralai库，则将返回实例化的MistralClient类；否则将报错 'ModuleNotFoundError: No module named 'mistralai''.
        """
        from mistralai.client import MistralClient
        api_key = os.getenv('MISTRAL_API_KEY')
        if not api_key:
            raise ValueError("MISTRAL_API_KEY is not set in the environment variables.")
        values["client"] = MistralClient(api_key=api_key)
        return values

    def embed_query(self, text: str) -> List[float]:
        """
        生成输入文本的 embedding.

        Args:
            texts (str): 要生成 embedding 的文本.

        Return:
            embeddings (List[float]): 输入文本的 embedding，一个浮点数值列表.
        """
        embeddings = self.client.embeddings(
            model="mistral-embed",
            input=[text]
        )
        return embeddings.data[0].embedding

    def embed_documents(self, texts: List[str]) -> List[List[float]]:
        """
        生成输入文本列表的 embedding.

        Args:
            texts (List[str]): 要生成 embedding 的文本列表.

        Returns:
            List[List[float]]: 输入列表中每个文档的 embedding 列表。每个 embedding 都表示为一个浮点值列表。
        """
        return [self.embed_query(text) for text in texts]

def generate_response(input_text, api_key):
    llm = ChatMistralAI(temperature=0,api_key=api_key)
    output = llm.invoke(input_text)
    output_parser = StrOutputParser()
    output = output_parser.invoke(output)
    #st.info(output)
    return output

def get_vectordb():
    # 定义 Embeddings
    embedding = MistralAIEmbeddings()
    # 向量数据库持久化路径
    persist_directory = './db/GIS_db'
    # 加载数据库
    vectordb = FAISS.load_local("./db/GIS_db", embedding, allow_dangerous_deserialization=True)

    return vectordb

#带有历史记录的问答链
def get_chat_qa_chain(question:str ,api_key:str):
    vectordb = get_vectordb()
    llm = ChatMistralAI(temperature=0,api_key=api_key)
    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,api_key:str):
    vectordb = get_vectordb()
    llm = ChatMistralAI(temperature=0,api_key=api_key)
    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('🦜🔗 动手学大模型应用开发(GISer Liu) Mistral版本')  # 创建应用程序的标题st.title
    api_key = st.sidebar.text_input('Mistral API Key', type='password')  # 添加一个文本输入框，供用户输入其 OpenAI API 密钥
    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})

        # 调用 respond 函数获取回答
        answer = generate_response(prompt, api_key)
        # 检查回答是否为 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()

3.部署应用程序

登录阿里云服务器，开放使用的端口，部署效果：
安全组

run
streamlit

OK，收工！😀😀🎉🎉🏆

文章参考

  • OpenAI官方文档
  • DeepSeek官方文档
  • Mistral官方文档
  • ChatGLM官方文档

项目地址

  • Github地址
  • 拓展阅读
  • 专栏文章

thank_watch

如果觉得我的文章对您有帮助，三连+关注便是对我创作的最大鼓励！或者一个star🌟也可以😂.

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/1868653.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

@RequestParam注解的使用及源码解析

@RequestParam注解的使用及源码解析

前言 RequestParam 注解是我们进行JavaEE开发，最常见的几个注解之一，这篇博文我们以案例和源码相结合，帮助大家更好的了解RequestParam 注解 使用案例 1.获取 URL 上的值 GetMapping("/simple") public String simple(RequestP…
阅读更多...
MySQL自学教程：1. MySQL简介与安装

MySQL自学教程：1. MySQL简介与安装

MySQL简介与安装 一、MySQL简介二、MySQL安装（一）Windows系统上的安装（二）Linux系统上的安装（以Ubuntu为例）（三）Mac OS系统上的安装三、安装后的基本配置四、总结一、MySQL简介 MySQL是一个流行的开源关系型数据库管理系统（RDBMS），广泛应用于各种业务场景，从小型个…
阅读更多...
2024肥晨赠书活动第三期：《前端工程化：基于Vue.js 3.0的设计与实践》

2024肥晨赠书活动第三期：《前端工程化：基于Vue.js 3.0的设计与实践》

文章目录 内容简介作者简介关于《前端工程化：基于Vue.js 3.0的设计与实践》文章目录文章简介《前端工程化：基于Vue.js 3.0的设计与实践》全书速览结束语 内容简介 本书以Vue.js的3.0版本为核心技术栈，围绕“前端工程化”和TypeScript的知识点…
阅读更多...
保姆级本地部署Qwen2

保姆级本地部署Qwen2

重点：Qwen2提供了CPU与GPU两种运行方式 运行成功效果图： 前提说明：如果需要用GPU，那么请在物理机安装ubuntu系统，不然显卡驱动很难安装，不建议新手部署。训练微调模型需要用到GPU。本文仅以ubuntu系统演示…
阅读更多...
vue3+ts ＜script setup lang=“ts“＞ element-plus的el-date-picker设置默认日期

vue3+ts ＜script setup lang=“ts“＞ element-plus的el-date-picker设置默认日期

效果图（单个日期）： utils.ts： /*** 格式化时间戳* param {number} timestamp 时间戳* param {string} format 格式* returns {string}*/ export const formatTimeStamp (timestamp: number, format: string) > {if (!timesta…
阅读更多...
Python魔法参数：深入解析*args和**kwargs的强大用途

Python魔法参数：深入解析*args和**kwargs的强大用途

目录 引言 基础概念解析 *args：处理位置参数 **kwargs：处理关键字参数 *args和**kwargs的实际应用场景 1. 函数装饰器中使用*args和**kwargs 2. 类构造函数中使用*args和**kwargs 3. API调用中使用**kwargs 与其他参数类型的结合使用 结合默认参数 位置参数与关键…
阅读更多...
利用powershell开展网络钓鱼

利用powershell开展网络钓鱼

要确保人们打开我们的恶意文件并执行它们，我们只需让微软努力工作多年来赢得人们的信任，然后将一些危险的宏插入到幻灯片中。 本博文将介绍如何通过屏幕顶部的一个友好的警告提示，在用户启用宏后立即运行您的宏。 首先，我们需要打…
阅读更多...
pytest-yaml-sanmu(五):跳过执行和预期失败

pytest-yaml-sanmu(五):跳过执行和预期失败

除了手动注册标记之外，pytest 还内置了一些标记可直接使用，每种内置标记都会用例带来不同的特殊效果，本文先介绍 3 种。 1. skip skip 标记通常用于忽略暂时无法执行，或不需要执行的用例。 pytest 在执行用例时，如果…
阅读更多...
手持小风扇哪个品牌好耐用？手持小风扇品牌排行榜揭晓分享

手持小风扇哪个品牌好耐用？手持小风扇品牌排行榜揭晓分享

炎炎夏日，手持小风扇、USB小风扇，成为人手一台的“网红”。这些小风扇造型小巧，可以装进包里，夏日出街或者挤公交地铁都可以拿出来吹一吹。那么这些小风扇性价比高不高呢？真的好用吗？耐用吗？根据…
阅读更多...
00. 这里整理了最全的爬虫框架（Java + Python）

00. 这里整理了最全的爬虫框架（Java + Python）

目录 1、前言 2、什么是网络爬虫 3、常见的爬虫框架 3.1、java框架 3.1.1、WebMagic 3.1.2、Jsoup 3.1.3、HttpClient 3.1.4、Crawler4j 3.1.5、HtmlUnit 3.1.6、Selenium 3.2、Python框架 3.2.1、Scrapy 3.2.2、BeautifulSoup Requests 3.2.3、Selenium 3.2.4…
阅读更多...
web前端——javaScript

web前端——javaScript

目录 一、javaScript概述 1.javaScript历史 2.JavaScript与html,css关系 二、基本语法 ①放在head中 ②放在 body中 ③写在外部的.js文件中 1.变量 2.数据类型 3.算术运算符 4.逻辑运算符 5.赋值运算 6.逻辑运算符 7.条件运算符 8.控制语句 三、函数 1…
阅读更多...
简单的text/html无法解析解决记录

简单的text/html无法解析解决记录

简单的text/html无法解析解决记录 1. bug发现 我们所有的服务都是微服务，服务间调用都是使用feign接口进行调用，正常调用都没有问题，但是某一天发现部分从esb服务调用过来到我们本地的服务，本地服务再使用feign接口调用其他微服…
阅读更多...
电脑定时重启怎么设置？用这个智能管理电脑定时任务的好帮手！

电脑定时重启怎么设置？用这个智能管理电脑定时任务的好帮手！

电脑定时重启怎么设置？用这个智能管理电脑定时任务的好帮手！电脑定时重启，这个设置其实很简单，但是很多人都不知道用电脑怎么设置，而且操作也很麻烦，并不好管理，这个时候我们需要一个非常智能的…
阅读更多...
模型情景制作-冰镇啤酒

模型情景制作-冰镇啤酒

夏日炎炎，当我们在真实世界中开一瓶冰镇啤酒的时候，我们也可以为模型世界中的人物添加一些冰镇啤酒。 下面介绍一种快速酒瓶制造方法，您只需要很少工具： 截取尽量直的流道（传说中的板件零件架）,将其夹在您的…
阅读更多...
adb push 报错 ...error: failed to copy...

adb push 报错 ...error: failed to copy...

一、现象： 原因：没有权限导致的 二、解决方法： adb root adb remount #重新加载文件系统三、再次尝试：adb push xxx.apk /system/app 结果：成功
阅读更多...
详细解释Spring事务的传播机制

详细解释Spring事务的传播机制

详细解释Spring事务的传播机制 Spring框架中，事务传播机制是指在一个事务方法调用另一个事务方法时，Spring如何管理这些方法之间的事务边界。Spring提供了七种事务传播行为，以满足不同的业务需求。下面将详细解释每种传播行为及其适用场景&a…
阅读更多...
不用翻墙，手把手教你用MAC本地版免费ComfyUI搭建Stable Diffusion工作流，让出图效率起飞

不用翻墙，手把手教你用MAC本地版免费ComfyUI搭建Stable Diffusion工作流，让出图效率起飞

AI绘图如火如荼发展了这么久，从mj到SD webUI,再到时下最热门的Comfy UI。因为显存的问题对Mac用户一直不是很友好，阻碍了大部分设计师上手学习的道路。但是Comflowy解决了这个痛点。这是一款Mac系统可用本地版的sd，一键安装，让苹果…
阅读更多...
阿里巴巴找黄金宝箱(IV)

阿里巴巴找黄金宝箱(IV)

系列文章目录 本人最近再练习算法，所以会发布自己的解题思路，希望大家多指教 文章目录 系列文章目录前言一、题目描述二、输入描述三、输出描述四、java代码五、测试用例 前言 一、题目描述 贫如洗的椎夫阿里巴巴在去砍柴的路上，无意中发现…
阅读更多...
基于SpringBoot学生信息管理系统设计和实现(源码+LW+调试文档+讲解等)

基于SpringBoot学生信息管理系统设计和实现(源码+LW+调试文档+讲解等)

💗博主介绍：✌全网粉丝10W,CSDN作者、博客专家、全栈领域优质创作者，博客之星、平台优质作者、专注于Java、小程序技术领域和毕业项目实战✌💗 🌟文末获取源码数据库🌟感兴趣的可以先收藏起来，还…
阅读更多...
操纵系统的特征调度算法

操纵系统的特征调度算法

操纵系统的特征 调度算法是操作系统用来决定各个进程/作业在CPU上执行顺序的方法。最常见的调度算法有：FCFS、SJF、HRRN、RR、HPF和MFQ。这集先介绍前三个 先来先服务 FCFS 根据作业到达的先后顺序调度，CPU会一直运行直到作业结束，所以这个…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023