目前AI模型领域百家争鸣,尽管 LangChain 官方提供了一些接口,但是可能并不支持所有模型。如果你需要自定义模型并将其接入 LangChain 框架,这是一种可能的选择。刚好文心一言能力全面开放,本教程借助文心一言大模型讲解在 LangChain 如何自定义 LLM,点击查看《文心一言注册及调用教程》。 《LangChain 入门到实战教程》更多内容
自定义 LLM 需要实现以下必要的函数:
_call
:它需要接受一个字符串、可选的停用词,并返回一个字符串。它还可以实现第二个可选的函数:
_identifying_params
:用于帮助打印 LLM 信息。该函数应该返回一个字典。使用LLM模块来封装我们的模型接口,可以带来许多好处,其中之一就是有利于与LangChain的其他模块进行协同工作。
下面我们通过 LangChain自定义LLM 实现文心一言 ERNIE-Bot-turbo 大模型接入:
import jsonimport timefrom typing import Any, List, Mapping, Optional, Dict, Union, Tupleimport loggingimport requestsfrom langchain.callbacks.manager import CallbackManagerForLLMRunfrom langchain.llms.base import LLMfrom langchain.utils import get_from_dict_or_envfrom pydantic import Field, root_validatorlogger = logging.getLogger(__name__)def get_access_token(api_key: str, secret_key: str): """ 使用 API Key,Secret Key 获取access_token """ url = f"https://aip.baidubce.com/oauth/2.0/token?grant_type=client_credentials&client_id={api_key}&client_secret={secret_key}" payload = json.dumps("") headers = { 'Content-Type': 'application/json', 'Accept': 'application/json' } resp = requests.request("POST", url, headers=headers, data=payload) return resp.json().get("access_token")class ErnieLLm(LLM): url = "https://aip.baidubce.com/rpc/2.0/ai_custom/v1/wenxinworkshop/chat/eb-instant" model_name: str = Field(default="ERNIE-Bot-turbo", alias="model") request_timeout: Optional[Union[float, Tuple[float, float]]] = None temperature: float = 0.95 """temperature 说明: (1)较高的数值会使输出更加随机,而较低的数值会使其更加集中和确定 (2)默认0.95,范围 (0, 1.0],不能为0 (3)建议该参数和top_p只设置1个 (4)建议top_p和temperature不要同时更改 """ top_p: float = 0.8 """top_p 说明: (1)影响输出文本的多样性,取值越大,生成文本的多样性越强 (2)默认0.8,取值范围 [0, 1.0] (3)建议该参数和temperature只设置1个 (4)建议top_p和temperature不要同时更改 """ penalty_score: float = 1.0 """通过对已生成的token增加惩罚,减少重复生成的现象。说明: (1)值越大表示惩罚越大 (2)默认1.0,取值范围:[1.0, 2.0] """ ernie_api_key: Optional[str] = None """文心一言大模型 apiKey""" ernie_secret_key: Optional[str] = None """文心一言大模型 secretKey""" user_id: Optional[str] = None """表示最终用户的唯一标识符,可以监视和检测滥用行为,防止接口恶意调用""" streaming: bool = False """是否以流式接口的形式返回数据,默认false""" cache: bool = False """是否开启缓存,默认为false""" model_kwargs: Dict[str, Any] = Field(default_factory=dict) """Holds any model parameters valid for `create` call not explicitly specified.""" @root_validator() def validate_environment(cls, values: Dict) -> Dict: """Validate that api key and python package exists in environment.""" values["ernie_api_key"] = get_from_dict_or_env( values, "ernie_api_key", "ERNIE_API_KEY" ) values["ernie_secret_key"] = get_from_dict_or_env( values, "ernie_secret_key", "ERNIE_SECRET_KEY" ) return values @property def _default_params(self) -> Dict[str, Any]: """获取调用Ennie API的默认参数。""" normal_params = { "temperature": self.temperature, "top_p": self.top_p, "penalty_score": self.penalty_score, "request_timeout": self.request_timeout, } return {**normal_params, **self.model_kwargs} def _construct_query(self, prompt: str) -> Dict: """构造请求体""" query = { "messages": [ { "role": "user", "content": prompt } ], "stream": self.streaming, "temperature": self.temperature, "top_p": self.top_p, "penalty_score": self.penalty_score, "user_id": self.user_id, } return query @property def _identifying_params(self) -> Mapping[str, Any]: """Get the identifying parameters.""" return {**{"model_name": self.model_name}, **self._default_params} @property def _llm_type(self) -> str: return "ernie" def _call( self, prompt: str, stop: Optional[List[str]] = None, run_manager: Optional[CallbackManagerForLLMRun] = None, **kwargs: Any, ) -> str: """_call 实现对模型的调用""" # construct query query = self._construct_query(prompt=prompt) # post _headers = {"Content-Type": "application/json"} with requests.session() as session: resp = session.post( self.url + "?access_token=" + get_access_token(self.ernie_api_key, self.ernie_secret_key), json=query, headers=_headers, timeout=60) if resp.status_code == 200: resp_json = resp.json() predictions = resp_json["result"] return predictions return "请求失败"
在 .env 文件中配置变量:
加载配置文件:
最简单的调用:
运行结果:
也可以通过构造直接传入 ernie_api_key
和ernie_secret_key
,如:
我们可以通过调整 temperature
、top_p
、penalty_score
等参数来优化模型回答的结果。
说明: (1)较高的数值会使输出更加随机,而较低的数值会使其更加集中和确定 (2)默认0.95,范围 (0, 1.0],不能为0 (3)建议该参数和top_p只设置1个 (4)建议top_p和temperature不要同时更改
说明: (1)影响输出文本的多样性,取值越大,生成文本的多样性越强 (2)默认0.8,取值范围 [0, 1.0] (3)建议该参数和temperature只设置1个 (4)建议top_p和temperature不要同时更改
通过对已生成的token增加惩罚,减少重复生成的现象。说明: (1)值越大表示惩罚越大 (2)默认1.0,取值范围:[1.0, 2.0]
使用方法:
运行结果:
从之前的文章中我们讲过如何使用缓存以及多种缓存方式,在这里我们使用本地内存缓存,配置缓存后,如果同一个问题被第二次提问,模型可以快速给出答案。
from langchain.cache import InMemoryCache# 启动llm的缓存langchain.llm_cache = InMemoryCache()
由于在自定义LLM时,缓存参数 cache
默认为 False,这里我们需要设置为 True。同时我们通过两次执行来看看效果:
llm = ErnieLLm(temperature=0.95, top_p=1.0, cache=True)s = time.perf_counter()# 第一次调用print(llm("给我讲一个笑话"))elapsed = time.perf_counter() - sprint("\033[1m" + f"第一次调用耗时 {elapsed:0.2f} 秒." + "\033[0m")s = time.perf_counter()# 第一次调用print(llm("给我讲一个笑话"))elapsed = time.perf_counter() - sprint("\033[1m" + f"第二次调用耗时 {elapsed:0.2f} 秒." + "\033[0m")
运行结果:
当然可以,这是一个关于两只鸟的笑话:有两只小鸟,一只小鸟问:“哥哥,人们都说你长得好帅,你觉得自己帅吗?”哥哥小鸟羞涩地回答:“不,我不觉得自己帅,我只是很可爱。”而第二只小鸟打断了他:“哥呀,人家说的是你旁边的蝴蝶卷毛哈。”第一次调用耗时 2.37 秒.当然可以,这是一个关于两只鸟的笑话:有两只小鸟,一只小鸟问:“哥哥,人们都说你长得好帅,你觉得自己帅吗?”哥哥小鸟羞涩地回答:“不,我不觉得自己帅,我只是很可爱。”而第二只小鸟打断了他:“哥呀,人家说的是你旁边的蝴蝶卷毛哈。”第二次调用耗时 0.00 秒.
可以看到第二次请求所用时间近乎为0(可能是纳秒级别)。
本文主要介绍了在LangChain平台上自定义LLM的步骤和参数,并以文心一言的ERNIE-Bot-turbo模型为例进行了详细说明。文章首先介绍了自定义LLM需要实现的必要函数,包括_call
函数和_identifying_params
函数。然后,通过导入dotenv模块和配置环境变量,示例代码演示了如何加载配置文件并调用自定义LLM。接下来,文章介绍了LLM的一些关键参数,如temperature
、top_p
和penalty_score
,并展示了如何根据需要调整这些参数来优化模型的回答结果。最后,文章提到了使用缓存的方法,通过启动LLM的缓存来加速模型的响应速度,并通过两次调用的结果展示了缓存的效果。