在之前的课程中,我带领小伙伴们使用开源项目实现了将星火模型的OpenAI-API接口适配转换封装,
但是这种做法的局限性也很强,只能使用开源项目适配过的大模型,并且由于多了一层适配代理,接口的性能也存在一定损耗。今天,我将给大家介绍一个更加通用的方案,基于LangChain平台提供的LLM基础模型,完成任意模型接口的封装。
LangChain与大模型交互的核心模型-LLM
LLM(语言逻辑模型)是LangChain平台与各种大模型进行交互的核心模型,它是一个抽象的概念,可以理解为一个能够处理语言输入和输出的黑盒。
LLM的输入是一个字符串,表示用户的请求或问题,LLM的输出也是一个字符串,表示模型的回答或结果。LLM可以根据不同的输入,调用不同的大模型,来完成不同的语言任务,如文本生成、文本理解、文本翻译等。
LLM的优势在于,它可以让开发者无需关心大模型的细节和复杂性,只需要关注语言的逻辑和意义,就可以利用大模型的能力来构建自己的应用。LLM也可以让开发者灵活地选择和切换不同的大模型,而无需修改代码或适配接口。LLM还可以让开发者自己封装自己的LLM,来实现自己的语言逻辑和功能。
如何自己封装一个LLM
要自己封装一个LLM,只需要实现以下两个必要的方法:
- 一个
_call
方法,它接受一个字符串作为输入,表示用户的请求或问题,还可以接受一些可选的停用词,用于过滤无关的词汇,它返回一个字符串作为输出,表示模型的回答或结果。 - 一个
_identifying_params
属性,它用于帮助打印这个类的信息,它返回一个字典,包含一些描述这个类的参数。
下面我们来实现一个非常简单的自定义LLM,它返回输入的前n个字符。
# 导入LangChain的LLM基类
from langchain.llm import LLM# 定义一个自定义的LLM类,继承自LLM基类
class CustomLLM(LLM):# 初始化方法,接受一个参数n,表示返回的字符数def __init__(self, n):# 调用父类的初始化方法super().__init__()# 将n赋值给self.nself.n = n# 实现_call方法,接受一个字符串input,和一些可选的停用词stop_wordsdef _call(self, input, stop_words=None):# 如果有停用词,就过滤掉输入中的停用词if stop_words:# 将输入分割成单词列表words = input.split()# 创建一个空列表,用于存放过滤后的单词filtered_words = []# 遍历单词列表for word in words:# 如果单词不在停用词列表中,就将其添加到过滤后的单词列表中if word not in stop_words:filtered_words.append(word)# 将过滤后的单词列表重新拼接成字符串input = " ".join(filtered_words)# 返回输入的前self.n个字符return input[:self.n]# 实现_identifying_params属性,返回一个字典,包含n的值@propertydef _identifying_params(self):return {"n": self.n}
基于讯飞星火api封装LLM实例
讯飞星火是一款基于人工智能的语音、图像、自然语言处理等领域的开放平台,提供了多种api接口,让开发者可以轻松地使用讯飞的技术能力。我们可以基于讯飞星火的api封装一个LLM,来实现一些语言任务,如文本翻译、文本摘要、文本分类等。
下面我们以星火大模型为例,来展示如何基于讯飞星火api封装一个LLM。我们需要先注册一个讯飞星火的账号,然后创建一个应用,获取应用的appid和appsecret,这两个参数是调用api的必要条件。我们还需要安装requests库,用于发送http请求。
官方提供的示例代码,SparkApi如下可直接使用
import _thread as thread
import base64
import datetime
import hashlib
import hmac
import json
from urllib.parse import urlparse
import ssl
from datetime import datetime
from time import mktime
from urllib.parse import urlencode
from wsgiref.handlers import format_date_timeimport websocket # 使用websocket_client
answer = ""class Ws_Param(object):# 初始化def __init__(self, APPID, APIKey, APISecret, Spark_url):self.APPID = APPIDself.APIKey = APIKeyself.APISecret = APISecretself.host = urlparse(Spark_url).netlocself.path = urlparse(Spark_url).pathself.Spark_url = Spark_url# 生成urldef create_url(self):# 生成RFC1123格式的时间戳now = datetime.now()date = format_date_time(mktime(now.timetuple()))# 拼接字符串signature_origin = "host: " + self.host + "\n"signature_origin += "date: " + date + "\n"signature_origin += "GET " + self.path + " HTTP/1.1"# 进行hmac-sha256进行加密signature_sha = hmac.new(self.APISecret.encode('utf-8'), signature_origin.encode('utf-8'),digestmod=hashlib.sha256).digest()signature_sha_base64 = base64.b64encode(signature_sha).decode(encoding='utf-8')authorization_origin = f'api_key="{self.APIKey}", algorithm="hmac-sha256", headers="host date request-line", signature="{signature_sha_base64}"'authorization = base64.b64encode(authorization_origin.encode('utf-8')).decode(encoding='utf-8')# 将请求的鉴权参数组合为字典v = {"authorization": authorization,"date": date,"host": self.host}# 拼接鉴权参数,生成urlurl = self.Spark_url + '?' + urlencode(v)# 此处打印出建立连接时候的url,参考本demo的时候可取消上方打印的注释,比对相同参数时生成的url与自己代码生成的url是否一致return url# 收到websocket错误的处理
def on_error(ws, error):print("### error:", error)# 收到websocket关闭的处理
def on_close(ws,one,two):print(" ")# 收到websocket连接建立的处理
def on_open(ws):thread.start_new_thread(run, (ws,))def run(ws, *args):data = json.dumps(gen_params(appid=ws.appid, domain= ws.domain,question=ws.question))ws.send(data)# 收到websocket消息的处理
def on_message(ws, message):# print(message)data = json.loads(message)code = data['header']['code']if code != 0:print(f'请求错误: {code}, {data}')ws.close()else:choices = data["payload"]["choices"]status = choices["status"]content = choices["text"][0]["content"]print(content,end ="")global answeranswer += content# print(1)if status == 2:ws.close()def gen_params(appid, domain,question):"""通过appid和用户的提问来生成请参数"""data = {"header": {"app_id": appid,"uid": "1234"},"parameter": {"chat": {"domain": domain,"temperature": 0.5,"max_tokens": 2048}},"payload": {"message": {"text": question}}}return datadef main(appid, api_key, api_secret, Spark_url,domain, question):# print("星火:")wsParam = Ws_Param(appid, api_key, api_secret, Spark_url)websocket.enableTrace(False)wsUrl = wsParam.create_url()ws = websocket.WebSocketApp(wsUrl, on_message=on_message, on_error=on_error, on_close=on_close, on_open=on_open)ws.appid = appidws.question = questionws.domain = domainws.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE})
我们可以基于这个api来进行我们自定义的LLM封装如下:
from llm.adaptor import SparkApi
from typing import Any, List, Mapping, Optionalfrom langchain.callbacks.manager import CallbackManagerForLLMRun
from langchain.llms.base import LLM#用于配置大模型版本,默认“general/generalv2”
# domain = "general" # v1.5版本
#domain = "generalv2" # v2.0版本
#domain = "generalv3" # v3.0版本
#云端环境的服务地址
# Spark_url = "ws://spark-api.xf-yun.com/v1.1/chat" # v1.5环境的地址
#Spark_url = "ws://spark-api.xf-yun.com/v2.1/chat" # v2.0环境的地址
#Spark_url = "ws://spark-api.xf-yun.com/v3.1/chat" # v3.0环境的地址
modal_dict={"general":"ws://spark-api.xf-yun.com/v1.1/chat","generalv2":"ws://spark-api.xf-yun.com/v2.1/chat","generalv3":"ws://spark-api.xf-yun.com/v3.1/chat"}def getText(role,content):text =[]jsoncon = {}jsoncon["role"] = rolejsoncon["content"] = contenttext.append(jsoncon)return textclass SparkLLM(LLM):appid: Optional[str] = Noneapi_secret: Optional[str] = Noneapi_key: Optional[str] = Nonemodel: Optional[str] = None@propertydef _llm_type(self) -> str:return "ErnieLLM"def _call(self,prompt: str,stop: Optional[List[str]] = None,run_manager: Optional[CallbackManagerForLLMRun] = None,**kwargs: Any,) -> str:Spark_url = modal_dict[self.model]domain = self.modelquestion = getText("user",prompt)SparkApi.answer = ""SparkApi.main(self.appid,self.api_key,self.api_secret,Spark_url,domain,question)return SparkApi.answer@propertydef _identifying_params(self) -> Mapping[str, Any]:"""Get the identifying parameters."""return {"appid": self.appid}
这样,我们就完成了基于讯飞星火api封装一个LLM的实例,我们可以在LangChain平台上使用这个LLM。
xh_app_id = ""xh_api_secret = ""xh_api_key = ""modal = "generalv3"
llm = SparkLLM(appid=xh_app_id,api_secret=xh_api_secret,api_key=xh_api_key,model=model)
当然,同样的封装方法也适用于其他大语言模型,例如百度提供了文心一言erniebot的sdk,我们的封装会比以上的示例更加简单。
总结
这篇文章就到这里结束了,希望你能够通过这篇文章,了解到如何使用LangChain平台开发基于大模型的应用,以及如何自己封装一个LLM。如果你对LangChain平台或LLM有任何疑问或建议,欢迎随时与我交流。😊
大模型&AI产品经理如何学习
求大家的点赞和收藏,我花2万买的大模型学习资料免费共享给你们,来看看有哪些东西。
1.学习路线图
第一阶段: 从大模型系统设计入手,讲解大模型的主要方法;
第二阶段: 在通过大模型提示词工程从Prompts角度入手更好发挥模型的作用;
第三阶段: 大模型平台应用开发借助阿里云PAI平台构建电商领域虚拟试衣系统;
第四阶段: 大模型知识库应用开发以LangChain框架为例,构建物流行业咨询智能问答系统;
第五阶段: 大模型微调开发借助以大健康、新零售、新媒体领域构建适合当前领域大模型;
第六阶段: 以SD多模态大模型为主,搭建了文生图小程序案例;
第七阶段: 以大模型平台应用与开发为主,通过星火大模型,文心大模型等成熟大模型构建大模型行业应用。
2.视频教程
网上虽然也有很多的学习资源,但基本上都残缺不全的,这是我自己整理的大模型视频教程,上面路线图的每一个知识点,我都有配套的视频讲解。
(都打包成一块的了,不能一一展开,总共300多集)
因篇幅有限,仅展示部分资料,需要点击下方图片前往获取
3.技术文档和电子书
这里主要整理了大模型相关PDF书籍、行业报告、文档,有几百本,都是目前行业最新的。
4.LLM面试题和面经合集
这里主要整理了行业目前最新的大模型面试题和各种大厂offer面经合集。
👉学会后的收获:👈
• 基于大模型全栈工程实现(前端、后端、产品经理、设计、数据分析等),通过这门课可获得不同能力;
• 能够利用大模型解决相关实际项目需求: 大数据时代,越来越多的企业和机构需要处理海量数据,利用大模型技术可以更好地处理这些数据,提高数据分析和决策的准确性。因此,掌握大模型应用开发技能,可以让程序员更好地应对实际项目需求;
• 基于大模型和企业数据AI应用开发,实现大模型理论、掌握GPU算力、硬件、LangChain开发框架和项目实战技能, 学会Fine-tuning垂直训练大模型(数据准备、数据蒸馏、大模型部署)一站式掌握;
• 能够完成时下热门大模型垂直领域模型训练能力,提高程序员的编码能力: 大模型应用开发需要掌握机器学习算法、深度学习框架等技术,这些技术的掌握可以提高程序员的编码能力和分析能力,让程序员更加熟练地编写高质量的代码。
1.AI大模型学习路线图
2.100套AI大模型商业化落地方案
3.100集大模型视频教程
4.200本大模型PDF书籍
5.LLM面试题合集
6.AI产品经理资源合集
👉获取方式:
😝有需要的小伙伴,可以保存图片到wx扫描二v码免费领取【保证100%免费】🆓