本帖最后由 云天 于 2024-1-9 18:58 编辑 【项目背景】 大模型定制训练平台 ，满足开发者更多场景调用需求，带来优质人机交互体验。讯飞星火FunctionCall： 【项目设计】 机器人 理解我说的话（话的内容为控制机器人做动作的指令：抬起手、放下手，但话的内容不固定），做出相应的动作。同时配合讯飞语音识别和语音合成功能，用户可以与行空板机器人进行语音对话。【演示视频】
【Function Call分析】 星火认知大模型Web API文档 | 讯飞开放平台文档中心 (xfyun.cn) ):我:明天我应该穿什么衣服？
星火:{"header":{"code":0,"message":"Success","sid":"cht000b8e77@dx18ce887dd74b8f3550","status":2},"payload":{"choices":{"status":2,"seq":0,"text":[{"content":"","role":"assistant","content_type":"text","function_call":{"arguments":"{"datetime":"明天","location":"CURRENT_CITY"}","name":"天气查询"},"index":0}]},"usage":{"text":{"question_tokens":6,"prompt_tokens":6,"completion_tokens":0,"total_tokens":6}}}} 复制代码 假设我们正在开发一个聊天类的AI应用，它需要根据用户输入的问题来生成相应的回答。我们在提问上并没有直接问明天的天气如何，而是问「我明天应该穿什么衣服？」 此时 语言完全理解了我的问题，而且知道穿衣建议是需要了解相关天气情况的。那么机器人应该返回一个关于今天天气的字符串。为了实现这个功能，我们可以定义一个名为“天气查询”的函数，它接受两个参数：一个表示日期的字符串、一个表示所在城市的字符串。然后，我们可以将这个函数的描述传递给大模型，让它生成一个包含调用该函数所需参数的JSON对象。最后，当用户输入一个问题时，我们可以让大模型分析问题并确定是否需要调用“天气查询”函数。如果需要，我们可以从JSON对象中提取所需的参数，并将它们传递给"天气查"询函数。这样，我们就可以根据用户输入的问题来生成相应的回答了。
为了让机器理解人类的意图，过去我们想方设法去「约束用户行为」或者「猜测用户意图』。但时代已经开始变了，通过 Function Call， 我们只需要在发送给 GPT 请求时加一个functions
【修改functions】
定制了一个"机器人控制"functions，描述为：“机器人控制可以控制机器人的左右手相应动作，你可以让机器人抬起、放下左右手。可以是只控制一只手，也可能同时控制两只手。” "payload": {
"message": {
"text": question
},
"functions": {
"text": [
{
"name": "机器人控制",
"description": "机器人控制可以控制机器人的左右手相应动作，你可以让机器人抬起、放下左右手。可以是只控制一只手，也可能同时控制两只手",
"parameters": {
"type": "object",
"properties": {
"hand": {
"type": "string",
"description": "控制对象，左手或右手或左右手"
},
"action": {
"type": "string",
"description": "动作，抬起或放下"
}
},
"required": [
"hand","action"
]
}
}
]
}
}
复制代码
我:让机器人抬起左手
星火:{"header":{"code":0,"message":"Success","sid":"cht000bf06b@dx18ce8ce92c5b8f3550","status":2},"payload":{"choices":{"status":2,"seq":0,"text":[{"content":"","role":"assistant","content_type":"text","function_call":{"arguments":"{"action":"抬起","hand":"左手"}","name":"机器人控制"},"index":0}]},"usage":{"text":{"question_tokens":4,"prompt_tokens":4,"completion_tokens":0,"total_tokens":4}}}}
我:放下右手
星火:{"header":{"code":0,"message":"Success","sid":"cht000bf6ba@dx18ce8cee8eab8f2550","status":2},"payload":{"choices":{"status":2,"seq":0,"text":[{"content":"","role":"assistant","content_type":"text","function_call":{"arguments":"{"action":"放下","hand":"右手"}","name":"机器人控制"},"index":0}]},"usage":{"text":{"question_tokens":2,"prompt_tokens":6,"completion_tokens":0,"total_tokens":6}}}}
我:抬起右手
星火:{"header":{"code":0,"message":"Success","sid":"cht000b09e3@dx18ce8d1b7e69a4b550","status":2},"payload":{"choices":{"status":2,"seq":0,"text":[{"content":"","role":"assistant","content_type":"text","function_call":{"arguments":"{"action":"抬起","hand":"右手"}","name":"机器人控制"},"index":0}]},"usage":{"text":{"question_tokens":2,"prompt_tokens":2,"completion_tokens":0,"total_tokens":2}}}}
我:让两只手都抬起
星火:{"header":{"code":0,"message":"Success","sid":"cht000b0bea@dx18ce8d202b79a4b550","status":2},"payload":{"choices":{"status":2,"seq":0,"text":[{"content":"","role":"assistant","content_type":"text","function_call":{"arguments":"{"action":"抬起","hand":"双手"}","name":"机器人控制"},"index":0}]},"usage":{"text":{"question_tokens":5,"prompt_tokens":7,"completion_tokens":0,"total_tokens":7}}}} 复制代码 【定义行空板控制舵机函数】
def RobotControl(hand,action):
if hand=="左手":
if action=="抬起":
servoL.write_angle(90)
else:
servoL.write_angle(178)
elif hand=="右手":
if action=="抬起":
servoR.write_angle(90)
else:
servoR.write_angle(2)
elif hand=="双手":
if action=="抬起":
servoL.write_angle(90)
servoR.write_angle(90)
else:
servoR.write_angle(2)
servoL.write_angle(178)
复制代码 我:把右手举起来
星火:{"header":{"code":0,"message":"Success","sid":"cht000b8d3e@dx18ce96fcae5b8f2550","status":2},"payload":{"choices":{"status":2,"seq":0,"text":[{"content":"","role":"assistant","content_type":"text","function_call":{"arguments":"{"action":"抬起","hand":"右手"}","name":"机器人控制"},"index":0}]},"usage":{"text":{"question_tokens":4,"prompt_tokens":4,"completion_tokens":0,"total_tokens":4}}}} 复制代码 【简易机器人制作】 【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_time
import websocket # 使用websocket_client
from pinpong.extension.unihiker import *
from pinpong.board import Servo
from pinpong.board import Board
from pinpong.board import Pin
Board().begin()
pin1 = Pin(Pin.D23)
pin2 = Pin(Pin.D22)
servoL = Servo(pin1)#左手舵机
servoR = Servo(pin2)#右手舵机
#左右手放平
servoL.write_angle(178)
servoR.write_angle(2)
hand=""
action=""
answer = ""
class Ws_Param(object):
# 初始化
def __init__(self, APPID, APIKey, APISecret, Spark_url):
self.APPID = APPID
self.APIKey = APIKey
self.APISecret = APISecret
self.host = urlparse(Spark_url).netloc
self.path = urlparse(Spark_url).path
self.Spark_url = Spark_url
# 生成url
def 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
}
# 拼接鉴权参数，生成url
url = 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):
global hand,action
#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 answer
answer += content
# print(1)
if choices["text"][0]["function_call"] is not None:
function_call=choices["text"][0]["function_call"]
if function_call["name"]=="机器人控制":
arguments=json.loads(function_call["arguments"])
action=arguments["action"]
hand=arguments["hand"]
RobotControl(hand,action)
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
},
"functions": {
"text": [
{
"name": "机器人控制",
"description": "“机器人控制”可以控制机器人的左右手相应动作，你可以使用动作指令让机器人抬起、放下左右手。动作指令只包括抬起和放下。可以是只控制一只手，也可能同时控制双手",
"parameters": {
"type": "object",
"properties": {
"hand": {
"type": "string",
"description": "控制对象，左手或右手或左右手"
},
"action": {
"type": "string",
"description": "动作，抬起或放下"
}
},
"required": [
"hand","action"
]
}
}
]
}
}
}
return data
def RobotControl(hand,action):
if hand=="左手":
if action=="抬起":
servoL.write_angle(90)
else:
servoL.write_angle(178)
elif hand=="右手":
if action=="抬起":
servoR.write_angle(90)
else:
servoR.write_angle(2)
elif hand=="双手":
if action=="抬起":
servoL.write_angle(90)
servoR.write_angle(90)
else:
servoR.write_angle(2)
servoL.write_angle(178)
def 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 = appid
ws.question = question
ws.domain = domain
ws.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE})
复制代码 【使用语音交互】 1.语音输入 复制代码 import pyaudio,wave
import numpy as np
def listen():
temp = 20
CHUNK = 1024
FORMAT = pyaudio.paInt16
CHANNELS = 1
RATE = 16000
RECORD_SECONDS = 2
WAVE_OUTPUT_FILENAME = 'record.wav'
mindb=1500 #最小声音，大于则开始录音，否则结束
delayTime=1.3 #小声1.3秒后自动终止
p = pyaudio.PyAudio()
stream = p.open(format=FORMAT,
channels=CHANNELS,
rate=RATE,
input=True,
frames_per_buffer=CHUNK)
#snowboydecoder.play_audio_file()
print("开始!计时")
frames = []
flag = False # 开始录音节点
stat = True #判断是否继续录音
stat2 = False #判断声音小了
tempnum = 0 #tempnum、tempnum2、tempnum3为时间
tempnum2 = 0
while stat:
data = stream.read(CHUNK,exception_on_overflow = False)
audio_data = np.frombuffer(data, dtype=np.short)
temp = np.max(audio_data)
if temp > mindb and flag==False:
flag =True
print("开始录音")
tempnum2=tempnum
if flag:
frames.append(data)
if(temp < mindb and stat2==False):
stat2 = True
tempnum2 = tempnum
print("声音小，且之前是是大的或刚开始，记录当前点")
if(temp > mindb):
stat2 =False
tempnum2 = tempnum
#刷新
if(tempnum > tempnum2 + delayTime*15 and stat2==True):
print("间隔%.2lfs后开始检测是否还是小声"%delayTime)
if(stat2 and temp < mindb):
stat = False
#还是小声，则stat=True
print("小声！")
else:
stat2 = False
print("大声！")
print(str(temp) + " " + str(tempnum))
tempnum = tempnum + 1
if tempnum > 3600:
tempnum=0 #超时直接退出
#stat = False
print("录音结束")
stream.stop_stream()
stream.close()
p.terminate()
wf = wave.open(WAVE_OUTPUT_FILENAME, 'wb')
wf.setnchannels(CHANNELS)
wf.setsampwidth(p.get_sample_size(FORMAT))
wf.setframerate(RATE)
wf.writeframes(b''.join(frames))
wf.close()
复制代码 import xunfeiasr
xunfeiasr.xunfeiasr_set(APPID=appid,APISecret=api_secret,APIKey=api_key)
复制代码 Input = input("\n" +"我:")
复制代码 listening.listen()
ShiBieNaRong=xunfeiasr.xunfeiasr(r"record.wav")
复制代码 "description": "“机器人控制”可以控制机器人的左右手相应动作，你可以使用动作指令让机器人抬起、放下左右手。动作指令只包括抬起和放下。可以是只控制一只手，也可能同时控制双手", 复制代码
import SparkApi
import listening
#以下密钥信息从控制台获取
appid = "*********" #填写控制台中获取的 APPID 信息
api_secret = "***********" #填写控制台中获取的 APISecret 信息
api_key ="**************" #填写控制台中获取的 APIKey 信息
import xunfeiasr
xunfeiasr.xunfeiasr_set(APPID=appid,APISecret=api_secret,APIKey=api_key)
#用于配置大模型版本，默认“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"
text =[]
# length = 0
def getText(role,content):
jsoncon = {}
jsoncon["role"] = role
jsoncon["content"] = content
text.append(jsoncon)
return text
def getlength(text):
length = 0
for content in text:
temp = content["content"]
leng = len(temp)
length += leng
return length
def checklen(text):
while (getlength(text) > 8000):
del text[0]
return text
if __name__ == '__main__':
text.clear
while(1):
listening.listen()
Input=xunfeiasr.xunfeiasr(r"record.wav")
question = checklen(getText("user",Input))
SparkApi.answer =""
print("星火:",end = "")
SparkApi.main(appid,api_key,api_secret,Spark_url,domain,question)
getText("assistant",SparkApi.answer)
# print(str(text))
复制代码 2.行空板屏幕表情 from unihiker import GUI
u_gui=GUI()
表情=u_gui.draw_emoji(emoji="Peace",x=0,y=0,duration=0.2) 复制代码 3.测试视频：
4.机器人语音回答 from df_xfyun_speech import XfTts
from unihiker import Audio
u_audio = Audio()
options = {}
business_args = {"aue":"raw","vcn":"xiaoyan","tte":"utf8","speed":50,"volume":50,"pitch":50,"bgs":0}
options["business_args"] = business_args
tts = XfTts(appid, api_key, api_secret, options) 复制代码 if SparkApi.answer!="":
tts.synthesis(answer+"……", "speech.wav")
u_audio.play("speech.wav")
复制代码
if SparkApi.hand!="":
tts.synthesis("好的"+SparkApi.hand+SparkApi.action+"了", "speech.wav")
u_audio.play("speech.wav") 复制代码 【主程序完整代码】
import SparkApi
import listening
import time
#以下密钥信息从控制台获取
appid = "*******************" #填写控制台中获取的 APPID 信息
api_secret = "***************************" #填写控制台中获取的 APISecret 信息
api_key ="*******************************" #填写控制台中获取的 APIKey 信息
import xunfeiasr
xunfeiasr.xunfeiasr_set(APPID=appid,APISecret=api_secret,APIKey=api_key)
from df_xfyun_speech import XfTts
from unihiker import Audio
u_audio = Audio()
options = {}
business_args = {"aue":"raw","vcn":"xiaoyan","tte":"utf8","speed":50,"volume":50,"pitch":50,"bgs":0}
options["business_args"] = business_args
global tts
tts = XfTts(appid, api_key, api_secret, options)
#用于配置大模型版本，默认“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"
text =[]
from unihiker import GUI
u_gui=GUI()
表情=u_gui.draw_emoji(emoji="Peace",x=0,y=0,duration=0.2)
# length = 0
def getText(role,content):
jsoncon = {}
jsoncon["role"] = role
jsoncon["content"] = content
text.append(jsoncon)
return text
def getlength(text):
length = 0
for content in text:
temp = content["content"]
leng = len(temp)
length += leng
return length
def checklen(text):
while (getlength(text) > 8000):
del text[0]
return text
if __name__ == '__main__':
text.clear
while(1):
listening.listen()
表情.config(emoji="Wink")
Input=xunfeiasr.xunfeiasr(r"record.wav")
print(Input)
question = checklen(getText("user",Input))
SparkApi.answer =""
SparkApi.hand=""
print("星火:",end = "")
SparkApi.main(appid,api_key,api_secret,Spark_url,domain,question)
if SparkApi.answer!="":
tts.synthesis(SparkApi.answer+"……", "speech.wav")
u_audio.play("speech.wav")
time.sleep(3)
if SparkApi.hand!="":
tts.synthesis("好的"+SparkApi.hand+SparkApi.action+"了", "speech.wav")
u_audio.play("speech.wav")
getText("assistant",SparkApi.answer)
表情.config(emoji="Peace")
# print(str(text))
复制代码 【注：System指令 】 System指令可对模型在多轮对话中的角色信息进行快捷约束，避免模型定制，即要求模型按照约定的方式进行后续响应。相比prompt约束，可以解决prompt拼接不连贯、不支持多轮的问题。
置顶卡
变色卡
千斤顶
活跃会员
宣传大使
创客造
编辑选择奖
志“童”道合
编辑选择奖
编辑选择奖
