动手学大模型应用开发，第6天：前后端搭建

第一章 项目代码简析

根据前面讲解的内容，我们逐步学习了如何调用差别的 LLM API，如那里置惩罚知识库文档搭建向量数据库，再如何设计 Prompt 搭建基于数据库的检索问答链。如今，我们可以回顾前面学过的所有内容，联合项目设计思路，将上述学习内容实现成代码，并按项目层次封装，来支持后续部署的调用。在这一章，我们会联合项目代码与前面学习的内容，讲解我们是如何封装项目代码以接口的形式为部署层提供核心功能的。
一、LLM 调用

基于前文内容，我们可以将百度文心、讯飞星火、智谱 GLM 等国内大模型接口封装成自定义的 LLM，然后接入到 LangChain 架构中。由于我们的项目需要将上述三种大模型接口都进行封装来同一到我们的项目框架中，为了项目的简洁，我们起首定义了一个自定义 LLM 的基类：

1. class Self_LLM(LLM):
2.     # 自定义 LLM
3.     # 继承自 langchain.llms.base.LLM
4.     # 原生接口地址
5.     url : str =  None
6.     # 默认选用 GPT-3.5 模型，即目前一般所说的百度文心大模型
7.     model_name: str = "gpt-3.5-turbo"
8.     # 访问时延上限
9.     request_timeout: float = None
10.     # 温度系数
11.     temperature: float = 0.1
12.     # API_Key
13.     api_key: str = None
14.     # 必备的可选参数
15.     model_kwargs: Dict[str, Any] = Field(default_factory=dict)

复制代码

Self_LLM 类定义在 /llm/self_llm.py 文件中，抽出调用三种大模型共同需要的参数（例如模型调用网址、利用模型名字、温度系数、API_Key 等）。在 Self_LLM 类基础上，我们分别继承了三个大模型 API 的自定义 LLM：Wenxin_LLM（/llm/wenxin_llm.py）、Spark_LLM（/llm/spark_llm.py）、ZhipuAILLM（/llm/zhipuai_llm.py），三个子类分别定义了本 API 所独有的参数，并基于 API 调用方式重写了 _call 方法。假如对封装方法有所疑问，可以具体阅读第二章，我们有讲解每一种 API 的调用方法。封装的自定义 LLM，实则就是将各自的调用方法按照给定的结构重写到 _call 方法中。
通过如上封装，在上一层检索问答链搭建时可以直接调用各个自定义 LLM 类，从而无需关注差别 API 调用的细节。
同时，为调用方便，我们也封装了同一的 get_completion 函数（/llm/call_llm.py）：

1. def get_completion(prompt :str, model :str, temperature=0.1,api_key=None,
2.                     secret_key=None, access_token=None, appid=None,
3.                     api_secret=None, max_tokens=2048) -> str

复制代码

该函数将四种模型 API 的原生接口封装在一起，旨在通过这一个函数来调用所有的模型。在这个函数内部，我们剖析了 model 参数的值，分别映射到差别的 API 调用函数中：

1.     if model in ["gpt-3.5-turbo", "gpt-3.5-turbo-16k-0613", "gpt-3.5-turbo-0613", "gpt-4", "gpt-4-32k"]:
2.         return get_completion_gpt(prompt, model, temperature, api_key, max_tokens)
3.     elif model in ["ERNIE-Bot", "ERNIE-Bot-4", "ERNIE-Bot-turbo"]:
4.         return get_completion_wenxin(prompt, model, temperature, api_key, secret_key)
5.     elif model in ["Spark-1.5", "Spark-2.0"]:
6.         return get_completion_spark(prompt, model, temperature, api_key, appid, api_secret, max_tokens)
7.     elif model in ["chatglm_pro", "chatglm_std", "chatglm_lite"]:
8.         return get_completion_glm(prompt, model, temperature, api_key, max_tokens)
9.     else:
10.         return "不正确的模型"

复制代码

对于其中映射的每一个子函数（包罗 get_completion_gpt、get_completion_wenxin 等），我们都以类似于第二章讲解的方式进行了封装。在后续调用中，可以直接利用 get_completion 函数，通过传入差别的模型参数和 API_KEY 认证，可以隐蔽掉 API 调用细节。
二、数据库构建

我们将在第四章中讲解过的构建项目数据库方法封装成 create_db 函数，函数中封装了各种文件范例的源文件处置惩罚方法和最终向量数据库的构建。假如当地没有构建过向量数据库，可以直接调用该方法：

1. def create_db(files=DEFAULT_DB_PATH, persist_directory=DEFAULT_PERSIST_PATH, embeddings="openai"):
2.     """
3.     该函数用于加载源数据文件，切分文档，生成文档的嵌入向量，创建向量数据库。
5.     参数:
6.     file: 存放文件的路径。
7.     embeddings: 用于生产 Embedding 的模型
9.     返回:
10.     vectordb: 创建的数据库。
11.     """

复制代码

在该函数内部，我们构造了一个文件加载映射函数，该函数会针对源文件范例分配差别的文件加载器，从而实现对不通过文件的处置惩罚：

1. def file_loader(file, loaders):
2.     # 对于多种文档的 FileLoader 映射
3.     if isinstance(file, tempfile._TemporaryFileWrapper):
4.         file = file.name
5.     if not os.path.isfile(file):
6.         [file_loader(os.path.join(file, f), loaders) for f in  os.listdir(file)]
7.         return
8.     file_type = file.split('.')[-1]
9.     if file_type == 'pdf':
10.         loaders.append(PyMuPDFLoader(file))
11.     elif file_type == 'md':
12.         loaders.append(UnstructuredMarkdownLoader(file))
13.     elif file_type == 'txt':
14.         loaders.append(UnstructuredFileLoader(file))
15.     return

复制代码

同时，针对已构造向量数据库需要调用的环境，我们也封装了 get_vectordb 函数，在检索问答链中只需直接调用该函数获取已构建的数据库即可：

1. def get_vectordb(file_path:str=None, persist_path:str=None, embedding = "openai",embedding_key:str=None):
2.     """
3.     返回向量数据库对象
4.     输入参数：
5.     question：
6.     llm:
7.     vectordb:向量数据库(必要参数),一个对象
8.     template：提示模版（可选参数）可以自己设计一个提示模版，也有默认使用的
9.     embedding：可以使用zhipuai等embeddin，不输入该参数则默认使用 openai embedding，注意此时api_key不要输错
10.     """

复制代码

三、检索问答链

基于 LLM 层与 Database 层的封装，我们可以在应用层搭建自定义的检索问答链，封装实现项目的核心功能。
起首我们封装了一个 LLM 映射函数，该函数会根据传入 model 参数的差别映射到差别的 LLM 对象，从而实现差别 API 来源 LLM 的切换：

1. def model_to_llm(model:str=None, temperature:float=0.0, appid:str=None, api_key:str=None,Spark_api_secret:str=None,Wenxin_secret_key:str=None):
2.         """
3.         星火：model,temperature,appid,api_key,api_secret
4.         百度文心：model,temperature,api_key,api_secret
5.         智谱：model,temperature,api_key
6.         OpenAI：model,temperature,api_key
7.         """
8.         if model in ["gpt-3.5-turbo", "gpt-3.5-turbo-16k-0613", "gpt-3.5-turbo-0613", "gpt-4", "gpt-4-32k"]:
9.             if api_key == None:
10.                 api_key = parse_llm_api_key("openai")
11.             llm = ChatOpenAI(model_name = model, temperature = temperature , openai_api_key = api_key)
12.         elif model in ["ERNIE-Bot", "ERNIE-Bot-4", "ERNIE-Bot-turbo"]:
13.             if api_key == None or Wenxin_secret_key == None:
14.                 api_key, Wenxin_secret_key = parse_llm_api_key("wenxin")
15.             llm = Wenxin_LLM(model=model, temperature = temperature, api_key=api_key, secret_key=Wenxin_secret_key)
16.         elif model in ["Spark-1.5", "Spark-2.0"]:
17.             if api_key == None or appid == None and Spark_api_secret == None:
18.                 api_key, appid, Spark_api_secret = parse_llm_api_key("spark")
19.             llm = Spark_LLM(model=model, temperature = temperature, appid=appid, api_secret=Spark_api_secret, api_key=api_key)
20.         elif model in ["chatglm_pro", "chatglm_std", "chatglm_lite"]:
21.             if api_key == None:
22.                 api_key = parse_llm_api_key("zhipuai")
23.             llm = ZhipuAILLM(model=model, zhipuai_api_key=api_key, temperature = temperature)
24.         else:
25.             raise ValueError(f"model{model} not support!!!")
26.         return llm

复制代码

在该映射器的基础上，我们构建了我们的自定义检索问答链。我们分别构建了两种检索问答链，QA_chain_self（/qa_chain/QA_chain_self.py）和 Chat_QA_chain_self（/qa_chain/Chat_QA_chain_self.py），分别对应普通的检索问答链和加入历史会话的检索问答链。两种自定义检索问答链内部实现细节类似，只是调用了差别的 LangChain 链。
在自定义检索问答链内部，我们起首在构造函数中定义了恒久参数的赋值，而且调用了 LLM 映射器和数据库构建（或加载）函数，从而实现了调用问答链的所有预备工作。然后我们定义了 answer 函数，该函数会对用户的题目调用检索问答链并输出回答，同时，在每一次调用 answer 的时间我们都可以动态改变温度系数和 top_k 参数：

1.     def answer(self, question:str=None, temperature = None, top_k = 4):
2.         """"
3.         核心方法，调用问答链
4.         arguments:
5.         - question：用户提问
6.         """
8.         if len(question) == 0:
9.             return ""
10.         
11.         if temperature == None:
12.             temperature = self.temperature
13.             
14.         if top_k == None:
15.             top_k = self.top_k
17.         result = self.qa_chain({"query": question, "temperature": temperature, "top_k": top_k})
18.         return result["result"]   

复制代码

在完成上述封装之后，我们无需关注底层细节，只需实例化一个自定义检索问答链对象，并调用其 answer 方法即可实现本项目的核心功能。同时，实例化时无需关注差别 API 的调用差异，直接传入需要调用的模型参数即可。
后续的服务层部署，我们会直接在上述封装代码的基础上进行调用，即直接调用自定义检索问答链，而不再重新实现全部过程。
第二章、Gradio 的介绍与前端界面的搭建
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。