基于智谱AI和LangChain实现RAG应用(一)

1. 使用 LangChain 调用智谱 GLM

1.1 自定义chatglm

#!/usr/bin/env python
# -*- encoding: utf-8 -*-

from typing import Any, List, Mapping, Optional, Dict
from langchain_core.callbacks.manager import CallbackManagerForLLMRun
from langchain_core.language_models.llms import LLM
from zhipuai import ZhipuAI

import os

# 继承自 langchain.llms.base.LLM
class ZhipuAILLM(LLM):
    # 默认选用 glm-4
    model: str = "glm-4"
    # 温度系数
    temperature: float = 0.1
    # API_Key
    api_key: str = None
    
    def _call(self, prompt : str, stop: Optional[List[str]] = None,
                run_manager: Optional[CallbackManagerForLLMRun] = None,
                **kwargs: Any):
        client = ZhipuAI(
            api_key = self.api_key
        )

        def gen_glm_params(prompt):
            '''
            构造 GLM 模型请求参数 messages

            请求参数：
                prompt: 对应的用户提示词
            '''
            messages = [{"role": "user", "content": prompt}]
            return messages
        
        messages = gen_glm_params(prompt)
        response = client.chat.completions.create(
            model = self.model,
            messages = messages,
            temperature = self.temperature
        )

        if len(response.choices) > 0:
            return response.choices[0].message.content
        return "generate answer error"


    # 首先定义一个返回默认参数的方法
    @property
    def _default_params(self) -> Dict[str, Any]:
        """获取调用API的默认参数。"""
        normal_params = {
            "temperature": self.temperature,
            }
        # print(type(self.model_kwargs))
        return {**normal_params}

    @property
    def _llm_type(self) -> str:
        return "Zhipu"

    @property
    def _identifying_params(self) -> Mapping[str, Any]:
        """Get the identifying parameters."""
        return {**{"model": self.model}, **self._default_params}
    

定义了一个名为 ZhipuAILLM 的类，它继承自 LLM（这是 LangChain 框架中的一个基类，用于表示语言模型）。该类封装了与智谱AI（ZhipuAI）进行交互的逻辑，主要用于调用智谱AI的 GLM 模型进行文本生成任务。以下是代码的详细解释：

1. 头部注释

#!/usr/bin/env python
# -*- encoding: utf-8 -*-

• #!/usr/bin/env python：这是 Unix/Linux 脚本的 Shebang，表示这个脚本应该用 python 解释器来运行。
• # -*- encoding: utf-8 -*-：声明文件使用 UTF-8 编码，确保代码可以处理多语言文本（如中文）。

2. 导入模块

from typing import Any, List, Mapping, Optional, Dict
from langchain_core.callbacks.manager import CallbackManagerForLLMRun
from langchain_core.language_models.llms import LLM
from zhipuai import ZhipuAI

import os

• typing 模块：提供类型提示（Type Hints）的支持，提升代码的可读性和安全性。
• langchain_core.callbacks.manager.CallbackManagerForLLMRun：用于管理和追踪语言模型运行时的回调函数。
• langchain_core.language_models.llms.LLM：LangChain 框架中的 LLM 基类，ZhipuAILLM 继承自这个类。
• zhipuai.ZhipuAI：智谱AI的客户端，用于与智谱AI的服务进行交互。

3. 定义 ZhipuAILLM 类

class ZhipuAILLM(LLM):
    model: str = "glm-4"
    temperature: float = 0.1
    api_key: str = None

• model: 默认模型设为 "glm-4"，这代表使用 GLM-4 模型进行推理。
• temperature: 温度系数，控制生成文本的随机性。较低的值（如 0.1）会使生成的文本更确定性。
• api_key: 用于身份验证的 API 密钥，必须在实例化时提供。

4. 定义 _call 方法

def _call(self, prompt: str, stop: Optional[List[str]] = None,
          run_manager: Optional[CallbackManagerForLLMRun] = None,
          **kwargs: Any):
    client = ZhipuAI(api_key=self.api_key)

    def gen_glm_params(prompt):
        messages = [{"role": "user", "content": prompt}]
        return messages

    messages = gen_glm_params(prompt)
    response = client.chat.completions.create(
        model=self.model,
        messages=messages,
        temperature=self.temperature
    )

    if len(response.choices) > 0:
        return response.choices[0].message.content
    return "generate answer error"

• _call 方法：这个方法定义了当你调用语言模型时实际发生的操作。

  • prompt: 用户提供的输入文本。
  • stop: 可选参数，用于指定生成文本的停止条件。
  • run_manager: 可选参数，用于管理模型运行时的回调。
  • **kwargs: 允许传递其他可选参数。

• client = ZhipuAI(api_key=self.api_key): 创建一个智谱AI的客户端实例。

• gen_glm_params(prompt)：内部函数，用于生成请求参数 messages，其中包含用户的输入提示。

• messages：构造的请求参数列表。

• response = client.chat.completions.create(...)：向智谱AI发送请求，使用指定模型和参数生成文本。

• if len(response.choices) > 0：检查生成的响应中是否有内容，如果有，返回生成的文本内容；否则，返回错误信息。

5. 定义 _default_params 属性

@property
def _default_params(self) -> Dict[str, Any]:
    normal_params = {
        "temperature": self.temperature,
    }
    return {**normal_params}

• _default_params：返回默认的模型参数，例如温度（temperature）。这用于在没有明确指定参数时，提供一些默认值。

6. 定义 _llm_type 属性

@property
def _llm_type(self) -> str:
    return "Zhipu"

• _llm_type：返回模型的类型标识符，在这里是 "Zhipu"，表示这是一个智谱AI的模型。

7. 定义 _identifying_params 属性

@property
def _identifying_params(self) -> Mapping[str, Any]:
    return {**{"model": self.model}, **self._default_params}

_identifying_params：返回一个字典，包含标识模型实例的所有参数，如模型名称和默认参数。这个属性可以用来唯一标识该语言模型实例。

这个类封装了对智谱AI的 GLM 模型的调用，可以用于集成在 LangChain 框架中，执行文本生成任务。主要功能包括：

• 设置模型类型和参数（如温度系数）。
• 生成请求参数并与智谱AI的服务进行交互。
• 返回生成的文本或错误信息。

1.2 接入 langchain

当前涉及到构建相关知识库，参考链接：

大模型入门到精通——使用Embedding API及搭建本地知识库(一)

大模型入门到精通——使用Embedding API及搭建本地知识库(二)

定义ZhipuAIEmbeddings如下：

from __future__ import annotations

import logging
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 ZhipuAIEmbeddings(BaseModel, Embeddings):
    """`Zhipuai Embeddings` embedding models."""

    client: Any
    """`zhipuai.ZhipuAI"""

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

        Args:

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

            values (Dict): 包含配置信息的字典。如果环境中有zhipuai库，则将返回实例化的ZhipuAI类；否则将报错 'ModuleNotFoundError: No module named 'zhipuai''.
        """
        from zhipuai import ZhipuAI
        _ = load_dotenv(find_dotenv())
        api_key = os.environ["API_key"]
        values["client"] = ZhipuAI(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.create(
            model="embedding-2",
            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]
    
    
    async def aembed_documents(self, texts: List[str]) -> List[List[float]]:
        """Asynchronous Embed search docs."""
        raise NotImplementedError("Please use `embed_documents`. Official does not support asynchronous requests")

    async def aembed_query(self, text: str) -> List[float]:
        """Asynchronous Embed query text."""
        raise NotImplementedError("Please use `aembed_query`. Official does not support asynchronous requests")
    
def vectorize_text(embeddings_model,chunks):
   
    return embeddings_model.embed_documents(chunks)

from dotenv import find_dotenv, load_dotenv
import os

# 读取本地/项目的环境变量。

# find_dotenv()寻找并定位.env文件的路径
# load_dotenv()读取该.env文件，并将其中的环境变量加载到当前的运行环境中
# 如果你设置的是全局的环境变量，这行代码则没有任何作用。
_ = load_dotenv(find_dotenv())

# 获取环境变量 API_KEY
api_key = os.environ["API_key"] #填写控制台中获取的 APIKey 信息

1. 实例化 ZhipuAILLM 类，并传入参数创建一个模型对象 zhipuai_model。

zhipuai_model = ZhipuAILLM(model="glm-4", temperature=0.1, api_key=api_key)

参数解释:

1. model="glm-4":

   • 指定了使用的模型版本为 "glm-4"。这个参数告诉 ZhipuAILLM 类使用 GLM-4 模型进行推理和生成文本。
   • 你还可以指定其他模型版本（如 "glm-4-0520"），如果模型版本支持不同的特性或更新，这样的配置可以灵活切换不同的模型。

2. temperature=0.1:

   • 设置生成文本的温度系数（temperature）。温度系数控制生成文本的随机性：
     • 较低的值（如 0.1）意味着生成的文本更具有确定性，输出更加集中于模型认为最可能的结果。
     • 较高的值会增加生成内容的多样性和创造性，但也可能导致结果不太一致。

3. api_key=api_key:

   • api_key 智谱AI API 密钥

2. 调用如下

zhipuai_model("你好，请你自我介绍一下！")

zhipuai_model("请介绍一下百度网站")

2. 构建检索问答链

2.1 加载向量数据库

• 配置系统路径以导入模块。
• 从环境变量中读取智谱AI API 密钥。
• 定义用于生成文本嵌入的对象。
• 设置向量数据库的保存路径。
• 加载并实例化向量数据库，用于存储和检索嵌入向量。

## 加载向量数据库
import sys
sys.path.append("./data_base/vector_db/chroma") # 将父目录放入系统路径中
# 使用智谱 Embedding API，注意，需要将上一章实现的封装代码下载到本地
from langchain.vectorstores.chroma import Chroma
from dotenv import load_dotenv, find_dotenv
import os

_ = load_dotenv(find_dotenv())    # read local .env file
zhipuai_api_key = os.environ['API_key']
# 定义 Embeddings
embedding = ZhipuAIEmbeddings()

# 向量数据库持久化路径
persist_directory = './data_base/vector_db/chroma'

# 加载数据库
vectordb = Chroma(
    persist_directory=persist_directory,  # 允许我们将persist_directory目录保存到磁盘上
    embedding_function=embedding
)

print(f"向量库中存储的数量：{vectordb._collection.count()}")

question = "什么是prompt engineering?"
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")

2.2 创建LLM

import os 
OPENAI_API_KEY = os.environ["API_key"]

zhipuai_model = ZhipuAILLM(model = "glm-4", temperature = 0.1, api_key = api_key)  
zhipuai_model.invoke("请你介绍一下自己")

2.3 构建检索问答链

1. 提示模版定义

通过插入特定的上下文和问题来生成提示，适用于自然语言处理模型。它确保模型生成的回答简洁明确，并在回答结束时添加

from langchain.prompts import PromptTemplate

template = """使用以下上下文来回答最后的问题。如果你不知道答案，就说你不知道，不要试图编造答
案。最多使用三句话。尽量使答案简明扼要。总是在回答的最后说“谢谢你的提问！”。
{context}
问题: {question}
"""

QA_CHAIN_PROMPT = PromptTemplate(input_variables=["context","question"],
                                 template=template)

2. 构建检索问答链

1. 问题输入: 用户提出一个问题。
2. 文档检索: 使用向量数据库检索与问题相关的文档或上下文。
3. 答案生成: 智谱AI模型根据检索到的上下文生成简明的答案，遵循提供的提示模板。
4. 返回结果: 返回生成的答案和用于生成答案的源文档（如果启用了 return_source_documents）。

RetrievalQA.from_chain_type 构建检索问答链

from langchain.chains import RetrievalQA

qa_chain = RetrievalQA.from_chain_type(zhipuai_model,
                                       retriever=vectordb.as_retriever(),
                                       return_source_documents=True,
                                       chain_type_kwargs={"prompt":QA_CHAIN_PROMPT})

3. 问答链测试

• 定义问题

question_1 = "什么是南瓜书？"
question_2 = "王阳明是谁？"

1. 大模型+知识库

• 打印回答1

result = qa_chain({"query": question_1})
print("大模型+知识库后回答 question_1 的结果：")
print(result["result"])

• 打印回答2

result = qa_chain({"query": question_2})
print("大模型+知识库后回答 question_2 的结果：")
print(result["result"])

2. 大模型回复

• 打印回答1

prompt_template = """请回答下列问题:
                            {}""".format(question_1)

### 基于大模型的问答
zhipuai_model.predict(prompt_template)

• 打印回答2

prompt_template = """请回答下列问题:
                            {}""".format(question_2)

### 基于大模型的问答
zhipuai_model.predict(prompt_template)

2.4 添加记忆功能

question = "为什么这门课需要教这方面的知识？"
result = qa({"question": question})
print(result['answer'])

实现一个对话式的检索问答链，其中结合了智谱AI模型、向量数据库和会话记忆来处理连续的问题。

1. 导入必要的模块

from langchain.chains import ConversationalRetrievalChain
from langchain.memory import ConversationBufferMemory

• ConversationalRetrievalChain: 用于创建一个支持多轮对话的检索问答链。与标准的检索问答链不同，它能够保持上下文，并在多轮对话中使用之前的对话历史。
• ConversationBufferMemory: 提供了会话记忆功能，能够保存对话的历史记录，从而在对话过程中引用之前的内容。

2. 创建会话记忆 memory

memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True
)

• memory_key="chat_history": 定义用于存储对话历史的键名，与后续提示模板中的变量名保持一致，以便在生成答案时能够引用这些历史记录。
• return_messages=True: 指定返回的对话历史记录将以消息列表的形式返回，而不是作为一个单一的字符串。这对于在多轮对话中保留消息的顺序和内容非常有用。

3. 创建检索器 retriever

retriever = vectordb.as_retriever()

• vectordb.as_retriever(): 使用之前加载的向量数据库 vectordb 生成一个检索器对象。这个检索器负责从数据库中检索与当前问题相关的内容。

4. 创建对话检索问答链 qa

qa = ConversationalRetrievalChain.from_llm(
    zhipuai_model,
    retriever=retriever,
    memory=memory
)

• ConversationalRetrievalChain.from_llm: 这是一个工厂方法，用于创建一个对话式检索问答链实例。
  • zhipuai_model: 使用的语言模型对象（之前实例化的 ZhipuAILLM）。
  • retriever=retriever: 使用的检索器对象，用于从向量数据库中检索相关信息。
  • memory=memory: 会话记忆对象，保存并管理对话的历史记录，使得模型能够在多轮对话中使用上下文信息生成更连贯的回答。

5. 处理第一个问题

question = "我可以学习到关于提示工程的知识吗？"
result = qa({"question": question})
print(result['answer'])

• question: 第一个问题，询问关于“提示工程”的知识。
• qa({"question": question}): 将问题传递给问答链进行处理。问答链会通过以下步骤生成答案：
  1. 检索与问题相关的文档或内容。
  2. 使用模型生成一个基于检索内容的回答。
  3. 利用会话记忆引用之前的对话（如果有）。
• print(result['answer']): 输出模型生成的答案。

6. 处理第二个问题

question = "为什么这门课需要教这方面的知识？"
result = qa({"question": question})
print(result['answer'])

• question: 第二个问题，询问为什么课程需要教授“提示工程”相关的知识。
• 由于对话记忆功能的存在，模型能够引用之前的对话上下文来生成更准确和连贯的回答。

参考

https://datawhalechina.github.io/llm-universe/#/C4/2.构建检索问答链?id=_2-创建一个-llm