大模型从入门到精通——基于智谱AI和LangChain实现RAG应用(一)

基于智谱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.LLMclassZhipuAILLM(LLM):# 默认选用 glm-4
    model:str="glm-4"# 温度系数
    temperature:float=0.1# API_Key
    api_key:str=Nonedef_call(self, prompt :str, stop: Optional[List[str]]=None,
                run_manager: Optional[CallbackManagerForLLMRun]=None,**kwargs: Any):
        client = ZhipuAI(
            api_key = self.api_key
        )defgen_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
        )iflen(response.choices)>0:return response.choices[0].message.content
        return"generate answer error"# 首先定义一个返回默认参数的方法@propertydef_default_params(self)-> Dict[str, Any]:"""获取调用API的默认参数。"""
        normal_params ={"temperature": self.temperature,}# print(type(self.model_kwargs))return{**normal_params}@propertydef_llm_type(self)->str:return"Zhipu"@propertydef_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

类

classZhipuAILLM(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)defgen_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
    )iflen(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

属性

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

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

6. 定义

_llm_type

属性

@propertydef_llm_type(self)->str:return"Zhipu"

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

7. 定义

_identifying_params

属性

@propertydef_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__)classZhipuAIEmbeddings(BaseModel, Embeddings):"""`Zhipuai Embeddings` embedding models."""

    client: Any
    """`zhipuai.ZhipuAI"""@root_validator()defvalidate_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
    
    defembed_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
    
    defembed_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]asyncdefaembed_documents(self, texts: List[str])-> List[List[float]]:"""Asynchronous Embed search docs."""raise NotImplementedError("Please use `embed_documents`. Official does not support asynchronous requests")asyncdefaembed_query(self, text:str)-> List[float]:"""Asynchronous Embed query text."""raise NotImplementedError("Please use `aembed_query`. Official does not support asynchronous requests")defvectorize_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 inenumerate(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