150浏览
查看: 150|回复: 1

[M10教程] 行空板扩展板:星火认知大模型封装与funsion_Call使用

[复制链接]
本帖最后由 hello创客 于 2024-9-29 22:21 编辑

【前言】
   非常荣幸能参与本次活动,原本单独制作一个作品完成本次帖子的内容,但由于比赛内容安排比较多,没有时间完成作品内容,但是指导学生利用行空板与扩展板完成两个相关作品,作品处于比赛阶段,因此只能公开部分照片,之后征求同意后也会考虑开源出来。拓展板非常好用,之后也会让更多孩子使用

行空板扩展板:星火认知大模型封装与funsion_Call使用图1               行空板扩展板:星火认知大模型封装与funsion_Call使用图2

【项目背景】
   由于行空板内置的Python版本比较低,常见的大模型的Python库无法兼容低版本的Python,于是有了想要封装讯飞星火认知大模型的想法,参考mind+官网的讯飞语音库的内容,很快就确定确定封装思路。
行空板扩展板:星火认知大模型封装与funsion_Call使用图3
行空板扩展板:星火认知大模型封装与funsion_Call使用图4行空板扩展板:星火认知大模型封装与funsion_Call使用图5

【封装思路】
通过讯飞星火认知大模型的调试中心,可以很方便实现了解Python的调用流程,我们需要做的是将websocket包含到里面,然后增加部分提示以及其他函数。
行空板扩展板:星火认知大模型封装与funsion_Call使用图6         行空板扩展板:星火认知大模型封装与funsion_Call使用图7

  1. import _thread as thread
  2. import base64
  3. import datetime
  4. import hashlib
  5. import hmac
  6. import json
  7. import time
  8. from urllib.parse import urlparse
  9. import ssl
  10. from datetime import datetime
  11. from time import mktime
  12. from urllib.parse import urlencode
  13. from wsgiref.handlers import format_date_time
  14. import websocket  # 使用websocket_client
  15. #配置大模型版本
  16. domain_name = "4.0Ultra"    # v4.0版本
  17. #云端环境的服务地址
  18. Spark_url = "wss://spark-api.xf-yun.com/v4.0/chat"  # v4.0环境的地址
  19. class Ws_Param(object):
  20.      # 是否空闲
  21.     STATE_IDLE = "Idle"
  22.     STATE_RUNNING = "Running"
  23.     STATE_COMPLETED = "Completed"
  24.     STATE_ERROR = "Error"
  25.     # 初始化
  26.     def __init__(self, APPID, APIKey, APISecret):
  27.         global Spark_url
  28.         self.APPID = APPID
  29.         self.APIKey = APIKey
  30.         self.APISecret = APISecret
  31.         self.host = urlparse(Spark_url).netloc
  32.         self.path = urlparse(Spark_url).path
  33.         self.Spark_url = Spark_url
  34.         # 是否运行结束、结果如何标识符
  35.         self.state = self.STATE_IDLE
  36.         # 完整内容
  37.         self.answer = ''
  38.         # 对话内容
  39.         self.data = None
  40.         
  41.     # 生成url
  42.     def create_url(self):
  43.         # 生成RFC1123格式的时间戳
  44.         now = datetime.now()
  45.         date = format_date_time(mktime(now.timetuple()))
  46.         # 拼接字符串
  47.         signature_origin = "host: " + self.host + "\n"
  48.         signature_origin += "date: " + date + "\n"
  49.         signature_origin += "GET " + self.path + " HTTP/1.1"
  50.         # 进行hmac-sha256进行加密
  51.         signature_sha = hmac.new(self.APISecret.encode('utf-8'), signature_origin.encode('utf-8'),
  52.                                  digestmod=hashlib.sha256).digest()
  53.         signature_sha_base64 = base64.b64encode(signature_sha).decode(encoding='utf-8')
  54.         authorization_origin = f'api_key="{self.APIKey}", algorithm="hmac-sha256", headers="host date request-line", signature="{signature_sha_base64}"'
  55.         authorization = base64.b64encode(authorization_origin.encode('utf-8')).decode(encoding='utf-8')
  56.         # 将请求的鉴权参数组合为字典
  57.         v = {
  58.             "authorization": authorization,
  59.             "date": date,
  60.             "host": self.host
  61.         }
  62.         # 拼接鉴权参数,生成url
  63.         url = self.Spark_url + '?' + urlencode(v)
  64.         # 此处打印出建立连接时候的url,参考本demo的时候可取消上方打印的注释,比对相同参数时生成的url与自己代码生成的url是否一致
  65.         return url
  66.     # 收到websocket错误的处理
  67.     def on_error(self,ws, error):
  68.         print("### error:", error)
  69.    
  70.     # 收到websocket关闭的处理
  71.     def on_close(self,ws,one,two):
  72.         print(" ")
  73.     # 收到websocket连接建立的处理
  74.     def on_open(self,ws):
  75.         self.state = self.STATE_RUNNING
  76.         thread.start_new_thread(self.run, (ws,))
  77.     def run(self,ws, *args):
  78.         global domain_name
  79.         data = json.dumps(self.gen_params(appid=ws.appid, domain= domain_name,question=ws.question))
  80.         ws.send(data)
  81.     # 询问函数
  82.     def ask(self,question):
  83.         if self.state == self.STATE_RUNNING:
  84.             print("spark process is already running.")
  85.             return
  86.         self.answer = ''
  87.         websocket.enableTrace(False)
  88.         ws = websocket.WebSocketApp(self.create_url(), on_message=self.on_message, on_error=self.on_error, on_close=self.on_close, on_open=self.on_open)
  89.         ws.appid = self.APPID
  90.         ws.question = question
  91.         ws.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE})
  92.     # 收到websocket消息的处理
  93.     def on_message(self,ws, message):
  94.         data = json.loads(message)
  95.         code = data['header']['code']
  96.         if code != 0:
  97.             self.state = self.STATE_ERROR
  98.             print(f'请求错误: {code}, {data}')
  99.             ws.close()
  100.         else:
  101.             choices = data["payload"]["choices"]
  102.             status = choices["status"]
  103.             content = choices["text"][0]["content"]
  104.             # print(content,end ="")
  105.             self.answer += content
  106.             if status == 2:            
  107.                 if choices["text"][0].get("function_call") is not None:
  108.                     # print("调用函数",end='')
  109.                     # print(choices["text"][0]["function_call"])
  110.                     self.answer = choices["text"][0]["function_call"]
  111.                 else:
  112.                     self.data['payload']['message']['text'].append({"role":"assistant","content":self.answer})
  113.                 ws.close()
  114.                 self.state = self.STATE_COMPLETED
  115.     def get_full_answer(self):
  116.         while  True:
  117.             if self.state == self.STATE_COMPLETED:
  118.                 return self.answer
  119.     def gen_params(self,appid, domain,question):
  120.         """
  121.         通过appid和用户的提问来生成请参数
  122.         """
  123.         if self.data is None:
  124.             self.data = {
  125.             "payload": {
  126.             "functions": {
  127.                 "text": [
  128.                     {
  129.                         "name": "天气查询",
  130.                         "description": "天气插件可以提供天气相关信息。你可以提供指定的地点信息、指定的时间点或者时间段信息,来精准检索到天气信息。",
  131.                         "parameters": {
  132.                             "type": "object",
  133.                             "properties": {
  134.                                 "date": {
  135.                                     "description": "日期。",
  136.                                     "type": "string"
  137.                                 },
  138.                                 "location": {
  139.                                     "description": "地点,比如北京。",
  140.                                     "type": "string"
  141.                                 }
  142.                             },
  143.                             "required": [
  144.                                 "location"
  145.                             ]
  146.                         }
  147.                     },
  148.                     {
  149.                         "name": "ws2812",
  150.                         "description": "点亮ws2812灯光或控制灯光,通过调整红色、绿色、蓝色调整ws2812灯的颜色,如果是紫色或橙色灯,需要同时修改红绿蓝三个颜色的值,使用亮度去调整灯光的亮度",
  151.                         "parameters": {
  152.                             "type": "object",
  153.                             "properties": {
  154.                                 "red": {
  155.                                     "description": "红光数值",
  156.                                     "type": "number"
  157.                                 },
  158.                                 "green": {
  159.                                     "description": "绿光数值",
  160.                                     "type": "number"
  161.                                 },
  162.                                 "blue":{
  163.                                     "description":"蓝光数值",
  164.                                     "type":"number"
  165.                                 },
  166.                                 "light":{
  167.                                     "description":"亮度,数字范围0~255,255最亮,0熄灭",
  168.                                     "type":"number"
  169.                                 }
  170.                             },
  171.                             "required": [
  172.                                 "red","green","blue","light"
  173.                             ]
  174.                         }
  175.                     }
  176.                 ]
  177.             },
  178.             "message": {
  179.                 "text": [
  180.                     {
  181.                         "role": "system",
  182.                         "content": "你是一名非常厉害的创客老师,带领许多孩子打过许多比赛,积极帮助老师与同学解答各种问题。,但是要求回答简短,中文回答,控制80字以内,同时回答完后加一个语气词,如了、啦、嗯"
  183.                     },
  184.                     {
  185.                         "role": "user",
  186.                         "content": question
  187.                     }
  188.                 ]
  189.             }
  190.             },
  191.             "parameter": {
  192.             "chat": {
  193.                 "max_tokens": 4096,
  194.                 "domain": domain,
  195.                 "top_k": 3,
  196.                 "temperature": 0.5
  197.             }
  198.             },
  199.             "header": {
  200.                 "app_id": appid
  201.             }
  202.             }
  203.         else:
  204.             self.data['payload']['message']['text'].append({"role":"user","content":question})
  205.         return self.data
  206.     def get_status(self):
  207.         return self.state
复制代码



【代码测试】
将库文件放在同一个文件下,使用import spark的方式即可调用大模型使用到的方法,以下为实例代码内容。

  1. from spark import Ws_Param
  2. from pinpong.board import Board
  3. from pinpong.board import NeoPixel
  4. from pinpong.board import Board,Pin
  5. from pinpong.extension.unihiker import *
  6. import json
  7. import time
  8. APPID = ''
  9. APISecret = ''
  10. APIKey = ''
  11. options = {}
  12. business_args = {"aue":"raw","vcn":"xiaoyan","tte":"utf8","speed":50,"volume":50,"pitch":50,"bgs":0}
  13. options["business_args"] = business_args
  14. Board().begin()
  15. np1 = NeoPixel(Pin((Pin.P13)),3)
  16. xf = Ws_Param(APPID,APIKey,APISecret)
  17. def RGB(red,green,blue,light):
  18.     np1.brightness(light)
  19.     np1[0] = (red,green,blue)
  20. while True:
  21.     try:
  22.         asw = input("输入你的问题:")
  23.     except:
  24.         print("输入有误")
  25.         continue
  26.     xf.ask(asw)
  27.     result = xf.get_full_answer()
  28.         print(result)
  29.       
  30.         
复制代码


【函数调用说明】
Function call 作为大模型能力扩展的核心,支持大模型在交互过程中识别出需要调度的外部接口:
注:当前Spark Max/4.0 Ultra 支持了该功能;需要请求参数payload.functions中申明大模型需要辨别的外部接口
行空板扩展板:星火认知大模型封装与funsion_Call使用图8行空板扩展板:星火认知大模型封装与funsion_Call使用图9
通过参考说明,以及示例程序,我们可以尝试完成其中一个案例,案例演示对一个灯颜色的调整和亮度的改变。

1、自定义函数内容

  1. {
  2.     "name": "ws2812",
  3.     "description": "点亮ws2812灯光或控制灯光,通过调整红色、绿色、蓝色调整ws2812灯的颜色,如果是紫色或橙色灯,需要同时修改红绿蓝三个颜色的值,使用亮度去调整灯光的亮度",
  4.     "parameters": {
  5.         "type": "object",
  6.         "properties": {
  7.             "red": {
  8.                 "description": "红光数值",
  9.                 "type": "number"
  10.             },
  11.             "green": {
  12.                 "description": "绿光数值",
  13.                 "type": "number"
  14.             },
  15.             "blue":{
  16.                 "description":"蓝光数值",
  17.                 "type":"number"
  18.             },
  19.             "light":{
  20.                 "description":"亮度,数字范围0~255,255最亮,0熄灭",
  21.                 "type":"number"
  22.             }
  23.         },
  24.         "required": [
  25.             "red","green","blue","light"
  26.         ]
  27.     }
  28. }
复制代码


2、在主程序中进行测试调用

  1. from spark import Ws_Param
  2. from pinpong.board import Board
  3. from pinpong.board import NeoPixel
  4. from pinpong.board import Board,Pin
  5. from pinpong.extension.unihiker import *
  6. import json
  7. import time
  8. APPID = ''
  9. APISecret = ''
  10. APIKey = ''
  11. options = {}
  12. business_args = {"aue":"raw","vcn":"xiaoyan","tte":"utf8","speed":50,"volume":50,"pitch":50,"bgs":0}
  13. options["business_args"] = business_args
  14. Board().begin()
  15. np1 = NeoPixel(Pin((Pin.P13)),3)
  16. xf = Ws_Param(APPID,APIKey,APISecret)
  17. def RGB(red,green,blue,light):
  18.     np1.brightness(light)
  19.     np1[0] = (red,green,blue)
  20. while True:
  21.     try:
  22.         asw = input("输入你的问题:")
  23.     except:
  24.         print("输入有误")
  25.         continue
  26.     xf.ask(asw)
  27.     result = xf.get_full_answer()
  28.     if type(result) is dict:
  29.         if result.get("name") == 'ws2812':
  30.             print(result)
  31.             color = json.loads(result["arguments"])
  32.             RGB(color["red"],color["green"],color["blue"],color["light"])
  33.     else:
  34.         print(result)
  35.       
  36.         
复制代码


【程序进阶】
利用讯飞语音以及讯飞星火认知大模型完成对程序的控制,程序执行效果没有使用流式,整体反应会比较慢,但程序效果基本实现完成的语音交互。

  1. from spark import Ws_Param
  2. from pinpong.board import Board
  3. from pinpong.board import NeoPixel
  4. from pinpong.board import Board,Pin
  5. from pinpong.extension.unihiker import *
  6. import json
  7. import time
  8. from unihiker import Audio
  9. from df_xfyun_speech import XfIat
  10. from df_xfyun_speech import XfTts
  11. APPID = ''
  12. APISecret = ''
  13. APIKey = ''
  14. options = {}
  15. business_args = {"aue":"raw","vcn":"xiaoyan","tte":"utf8","speed":50,"volume":50,"pitch":50,"bgs":0}
  16. options["business_args"] = business_args
  17. Board().begin()
  18. np1 = NeoPixel(Pin((Pin.P13)),3)
  19. u_audio = Audio()
  20. xf = Ws_Param(APPID,APIKey,APISecret)
  21. iat = XfIat(APPID, APIKey, APISecret)
  22. tts = XfTts(APPID, APIKey, APISecret, options)
  23. def RGB(red,green,blue,light):
  24.     np1.brightness(light)
  25.     np1[0] = (red,green,blue)
  26. while True:
  27.     if (button_a.is_pressed()==True):
  28.         print("开始录音")
  29.         u_audio.record("record.wav",5)
  30.         asw = iat.recognition("record.wav")
  31.         # try:
  32.         #     asw = input("输入你的问题:")
  33.         # except:
  34.         #     print("输入有误")
  35.         #     continue
  36.         print(asw)
  37.         xf.ask(asw)
  38.         result = xf.get_full_answer()
  39.         
  40.         if type(result) is dict:
  41.             if result.get("name") == 'ws2812':
  42.                 tts.synthesis("执行自定义函数了", "speech.wav")
  43.                 print(result)
  44.                 color = json.loads(result["arguments"])
  45.                 RGB(color["red"],color["green"],color["blue"],color["light"])
  46.                 u_audio.play("speech.wav")
  47.                 time.sleep(1)
  48.         else:
  49.             tts.synthesis(result, "speech.wav")
  50.             print(result)
  51.             u_audio.play("speech.wav")
  52.             time.sleep(1)
复制代码


行空板扩展板:星火认知大模型封装与funsion_Call使用图10
电路图




【项目总结】
第一次发帖,请多多包涵。在使用过程中,对话还需要更加优化,比如当下使用websocket流式,还有远程部署到,但想到给学生使用足矣,后面也会继续优化。


zoey不种土豆  超级版主

发表于 3 天前

期待优化后的分享~
回复

使用道具 举报

您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

为本项目制作心愿单
购买心愿单
心愿单 编辑
[[wsData.name]]

硬件清单

  • [[d.name]]
btnicon
我也要做!
点击进入购买页面
上海智位机器人股份有限公司 沪ICP备09038501号-4

© 2013-2024 Comsenz Inc. Powered by Discuz! X3.4 Licensed

mail