前言

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

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

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

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

一、LLM接入LangChain

1. LangChain中LLM组成

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

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

具体内容如下：

①提示词模版

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

unsetunset核心思想unsetunset

• 1. 模版创建
• 2. 用户输入
• 3. 提示词打包

unsetunset原生实现unsetunset

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

运行查看结果：

unsetunsetLangChain实现unsetunset

# 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中，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对象

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

②模型

在过去的文章中，作者说明了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对象

运行后可以看到输出结果：

下面是作者整理的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)

unsetunset核心思想：unsetunset

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

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

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

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

unsetunset原生实现unsetunset

• 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运行结果如下：

• 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运行结果如下：

不难看出，效果很好！?

unsetunsetLangChain实现unsetunset

# 导入所需的库和模块
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对象

运行后，可以看到结果:

????我们成功通过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的集成到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



# 导入所需的库和模块
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"}))

二、构建检索问答链



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

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

1.加载向量数据库

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



为了后续使用，我们使用上一篇文章中作者封装的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")

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



可以看到，已经成功检索出与作者输入的"电磁辐射"最相似的前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)



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

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"])



对比无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))



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参数,使其输出历史信息； 运行代码进行测试：



再次输入测试：

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



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

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

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

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

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

1. 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.部署应用程序

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







OK，收工！

文章参考

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

项目地址

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