聊天机器人之接入智能回复（四）

智能回复实现思路

目前市场上有许多大语言模型，他能帮助我们更好的解决问题或者陪伴我们聊天。最先兴起的就是OpenAI的ChatGPT，但是我们国内使用不太方便。所以这里我使用咱们国内的平台做一个简单展示。

首先这里咱们使用的是讯飞星火认知大模型，这个提供了免费额度的API使用，推荐可以进行实名认证可以获取更多的额度。

具体实现就是利用前面俩篇文章提到的接收和发送消息的功能来组合实现，当咱们接收到某个消息的时候咱们可以把接收到的消息传递给语言模型，让他给我们回复，得到回复之后咱们再使用之前的发送接口，将信息发送回去，这样就实现了一个智能回复。

星火API 使用

我从官网下载了一个python使用案例，稍做修改，直接上代码。
SparkApi.py的代码

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})

具体使用代码 test.py，咱们这里使用v3的版本。

import SparkApi
#以下密钥信息从控制台获取
appid = "填写你自己的"     #填写控制台中获取的 APPID 信息
api_secret = "填写你自己的"   #填写控制台中获取的 APISecret 信息
api_key ="填写你自己的"    #填写控制台中获取的 APIKey 信息#用于配置大模型版本，默认“general/generalv2”
# domain = "general"   # v1.5版本
domain = "generalv3"   # v3.0版本
# domain = "generalv2"    # v2.0版本
#云端环境的服务地址
Spark_url = "wss://spark-api.xf-yun.com/v3.1/chat"  # 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环境的地址text =[]# length = 0def getText(role,content):jsoncon = {}jsoncon["role"] = rolejsoncon["content"] = contenttext.append(jsoncon)return textdef getlength(text):length = 0for content in text:temp = content["content"]leng = len(temp)length += lengreturn lengthdef checklen(text):while (getlength(text) > 8000):del text[0]return textdef get_answer(question):question = getText("user", question)SparkApi.main(appid, api_key, api_secret, Spark_url, domain, question)getText("assistant", SparkApi.answer)print(SparkApi.answer)return SparkApi.answerif __name__ == '__main__':text.clearwhile (1):Input = input("\n" + "我:")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))

运行代码效果

在这里插入图片描述

具体实现自动回信

这里咱们就需要对之前写的接收消息的代码进行修改。

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Author  : QS
# @QQ      : 376494614
# @File    : test_flask_reciver.pyimport requests
from flask import Flask, request, jsonify
from flask_cors import CORS
import json
from test import get_answerapp = Flask(__name__)port = 6999
url = "http://127.0.0.1"
test_url = f"{url}:{port}/"
post_url = f"{url}:{port}/send"headers = {"Content-Type": "application/json"
}def reply_msg(reciver, question):answer = get_answer(question)data = {"type": 1, "data": {"wxid": reciver, "msg": answer}}requests.post(post_url, headers=headers, data=json.dumps(data))@app.route('/reciver', methods=["POST"])
def diy_recive_msg():if request.method == 'POST':payload = request.get_data(as_text=True)# 按照顺序打印输出# payload = json.dumps(payload, sort_keys=True, indent=2, ensure_ascii=False)# print('[全数据]\r', payload)datas = json.loads(payload)print(datas)msg = datas["msg"]if "\u2005" in msg and "@" in msg:# 代表为atmsg_split = msg.split("\u2005")# 这里只有at当前用户一个人的时候生效if len(msg_split) < 2:returnif msg_split[0] == "@你的昵称":  # 这里引号内替换为你自己的昵称sender = datas["chatid"]reply_msg(sender, msg_split[1])if datas["type"] == 1:# 为私信sender = datas["wxid"]reply_msg(sender, msg)return datasif __name__ == '__main__':# 下面的代码是消息接收的部分# app.config.from_object(Config())app.config['JSON_AS_ASCII'] = FalseCORS(app, supports_credentials=True)# 开启定时任务# scheduler = APScheduler()# scheduler.init_app(app)# scheduler.start()app.run(host='127.0.0.1', port=6666, debug=True, use_reloader=False)