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

1. 使用 LangChain 调用智谱 GLM

1.1 自定义chatglm

<code>#!/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",code>

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。