我用AI大模型管理智能家电，全网唯一完整教程+源代码_qwen_agent_key

作者：Monodyee | 2024-04-23 07:25:48

踩

qwen_agent_key

本期视频：

我用AI大模型管理智能家电，全网唯一完整教程+源代码

我用钉钉接入AI大模型，还有home assistant，自己设计了一个AI智能管家。那么他跟小爱智能音箱比有什么优点呢？我认为有3点，首先是极高的智能性，我的这套代码可以替换成任意一种AI大模型，比如ChatGPT4，甚至即将上线的ChatGPT5，这就比智障小爱同学强了一百个天猫精灵。然后是高度的可定制化，我使用Python代码编写程序，不再局限于米家APP的有限配置，具有更高的自由度跟可玩性，理论上它能调动家里任意设备，实现任意的自动化流程。目前我这个项目还只是个100多行代码的小Demo，我只调通了基本流程，很多不足还是需要改进。很适合爱学习爱折腾的观众朋友深入研究，这就是第三个优点，折腾过程中可以学到很多Python编程，大模型应用等相关知识。爬爬虾之前视频讲过家庭服务器，HomeAssistant，还有钉钉接入AI大模型等等的基础内容，所以这期有些步骤我会讲的比较快，建议大家补一下之前视频对一些工具有更多了解。

我先介绍一下技术选型，我选的聊天工具是钉钉，钉钉方案最大优点是不需要公网IP，正好适合部署在家庭服务器上，大模型是通义千问，目前是限时免费的。智能家居当然是Home Asssistant, 可玩性最高的智能家居管理平台。下面马上开始干货教程，希望各位观众一键三连鼓励下新人UP。

1. 注册钉钉

https://oa.dingtalk.com/register_new.htm?showmenu=false#/

2. 创建应用

进入钉钉开发者后台，登录后，点击创建应用，填写应用相关信息。

点击添加应用能力，选择 "机器人" 能力，点击添加。

配置机器人信息后点击发布。发布后，点击“点击调试”，会自动创建测试群聊，可在客户端查看:

点击版本管理与发布，创建新版本发布

点击凭证与基础信息，获取Client ID和Client Secret两个参数

3. 获取通义千问Access Key

注册并实名认证阿里云

官方地址：阿里云登录 - 欢迎登录阿里云，安全稳定的云计算服务平台

获取AccessKey AccessKeySecret

进入阿里云百炼阿里云登录 - 欢迎登录阿里云，安全稳定的云计算服务平台

应用管理->模型->调用获取 AgentKey和AppId

4. 部署HomeAssistant

home Assistant是一个开源项目，他是智能家居的终极解决方案，可以将各种厂商的智能设备接入同一平台。

关于home Assistant的部署使用可以看一下这两期教程

旧电脑不要扔！用来部署智能家居终极方案HomeAssistant，一键安装部署，易学低成本。 - 哔哩哔哩

手机静音找不到怎么办？原创全网最优雅的找手机方法米家智能网关接入home assistant - 哔哩哔哩

Windows平台最简单的部署方法就是使用这个开源项目，HA的一键启动包

GitHub - AlexxIT/HassWP: Portable version of Home Assistant for Windows (no need to install)

部署完成后，我们获取一个home Assistant 的Token，

首先进入Home Assistant的设置页面 http://127.0.0.1:8123/profile

点击创建令牌，获取一个很长的Token保存好。

5. 编写大模型提示词

我们使用自然语言跟大模型交流，大模型选择一个合适工具函数，把函数名称还有参数返回给我，再使用Python具体执行这个函数，这个就是大模型Function Call的一个实现思路。

你是一个智能家居AI，你的根据用户输入返回对函数，必须从函数列表里面选，#后面备注了函数说明，如果能查找到功能的函数，你只需要输出函数加参数，参数就加引号写在函数的括号里，如果没有参数则不加。如果查寻不到任何函数，就输出"对不起，办不到"

函数列表：

find_temperature() #查询房间温度，参数是房间名称

open_air_purifier() #打开空气净化器无参数

close_air_purifier() #关闭空气净化器无参数

6. 程序代码


# 安装python依赖 pip3 install dingtalk_stream broadscope-bailian
 
import json
import logging
import uuid
 
import requests
from broadscope_bailian import ChatQaMessage
from dingtalk_stream import AckMessage
import dingtalk_stream
import broadscope_bailian
# 2.5小节获取的两个参数
DINGTALK_CLIENT_ID = "xxxxxxxxxxxxxxxxxxxx"
DINGTALK_CLIENT_SECRET = "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
# 3.1 3.2小节获取的四个参数
QWEN_ACCESS_KEY_ID = "xxxxxxxxxxxxxxxxxxxx"
QWEN_ACCESS_KEY_SECRET = "xxxxxxxxxxxxxxxxxxxx"
QWEN_AGENT_KEY = "xxxxxxxxxxxxxxxxxxxx"
QWEN_APP_ID = "xxxxxxxxxxxxxxxxxxxx"
# 第4节获取的token
HOME_ASSISTANT_TOKEN = "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
# AI大模型的提示词
PROMPT = '''
你是一个智能家居AI，你的根据用户输入返回对函数，必须从函数列表里面选，#后面备注了函数说明，如果能查找到功能的函数，你只需要输出函数加参数，参数就加引号写在函数的括号里，如果没有参数则不加。如果查寻不到任何函数，就输出"对不起，办不到"
函数列表：
find_temperature()  #查询房间温度，参数是房间名称
open_air_purifier()  #打开空气净化器 无参数
close_air_purifier()  #关闭空气净化器 无参数
'''
 
# 控制设备函数的具体实现，使用Home Assistant的Restful API
# 具体见文档https://developers.home-assistant.io/docs/api/rest/
# 小米参数 https://home.miot-spec.com/
def open_air_purifier():
    url = "http://127.0.0.1:8123/api/services/xiaomi_miot/set_miot_property"
    payload = json.dumps({
        "entity_id": "fan.zhimi_v6_c97a_air_purifier",
        "siid": 2,
        "piid": 1,
        "value": True
    })
    headers = {
        'Content-Type': 'application/json',
        'Authorization': f'Bearer {HOME_ASSISTANT_TOKEN}'
    }
    response = requests.request("POST", url, headers=headers, data=payload)
    return "命令已发送"
 
 
def close_air_purifier():
    url = "http://127.0.0.1:8123/api/services/xiaomi_miot/set_miot_property"
    payload = json.dumps({
        "entity_id": "fan.zhimi_v6_c97a_air_purifier",
        "siid": 2,
        "piid": 1,
        "value": False
    })
    headers = {
        'Content-Type': 'application/json',
        'Authorization': f'Bearer {HOME_ASSISTANT_TOKEN}'
    }
    response = requests.request("POST", url, headers=headers, data=payload)
    return "命令已发送"
 
 
def find_temperature(room_name):
 
    entity_id = ""
    if room_name == "客厅":
        entity_id = "sensor.a4c138ba6acb_temperature"
    if room_name == "卧室":
        entity_id = "sensor.a4c138906273_temperature"
 
    url = f"http://127.0.0.1:8123/api/states/{entity_id}"
 
    payload = json.dumps({
        "entity_id": "sensor.a4c138ba6acb_temperature",
        "method": ""
    })
    headers = {
        'Content-Type': 'application/json',
        'Authorization': f'Bearer {HOME_ASSISTANT_TOKEN}'
    }
 
    response = requests.request("GET", url, headers=headers, data=payload)
    return json.loads(response.text).get("state")
 
 
class Qwen:
    def __init__(self):
        self.session_id = str(uuid.uuid4()).replace("-", "")
        self.client = broadscope_bailian.AccessTokenClient(access_key_id=QWEN_ACCESS_KEY_ID,
                                                           access_key_secret=QWEN_ACCESS_KEY_SECRET,
                                                           agent_key=QWEN_AGENT_KEY)
        self.token = self.client.get_token()
        self.chat_history = [ChatQaMessage(PROMPT, "好的，请输入")]
        self.completions = broadscope_bailian.Completions(token=self.token)
 
    def get_response(self, prompt):
        resp = self.completions.call(app_id=QWEN_APP_ID, history=self.chat_history, prompt=prompt)
        response = resp.get("Data").get("Text")
        return response
 
 
def setup_logger():
    logger = logging.getLogger()
    handler = logging.StreamHandler()
    handler.setFormatter(
        logging.Formatter('%(asctime)s %(name)-8s %(levelname)-8s %(message)s [%(filename)s:%(lineno)d]'))
    logger.addHandler(handler)
    logger.setLevel(logging.INFO)
    return logger
 
 
class EchoTextHandler(dingtalk_stream.ChatbotHandler):
    def __init__(self, logger: logging.Logger = None):
        super(dingtalk_stream.ChatbotHandler, self).__init__()
        if logger:
            self.logger = logger
        # 初始化AI模型
        self.qwen = Qwen()
 
    async def process(self, callback: dingtalk_stream.CallbackMessage):
        incoming_message = dingtalk_stream.ChatbotMessage.from_dict(callback.data)
        text = incoming_message.text.content.strip()
        ai_response = self.qwen.get_response(text)
        # 执行模型返回的函数
        print(ai_response)
        try:
            result = eval(ai_response)
        except Exception as e:
            print("模型返回无法执行: ", e)
            result = ai_response
        self.reply_text(result, incoming_message)
        return AckMessage.STATUS_OK, 'OK'
 
 
def main():
    logger = setup_logger()
 
    credential = dingtalk_stream.Credential(DINGTALK_CLIENT_ID, DINGTALK_CLIENT_SECRET)
    client = dingtalk_stream.DingTalkStreamClient(credential)
    client.register_callback_handler(dingtalk_stream.chatbot.ChatbotMessage.TOPIC, EchoTextHandler(logger))
    client.start_forever()
 
 
if __name__ == '__main__':
    main()

7. PS.小米硬件功能编写指南

小米siid, piid查询
小米/米家产品库 - Xiaomi Miot Spec

Home Assistant Restful API
REST API | Home Assistant Developer Docs

Restful API 调试


curl --location 'http://localhost:8123/api/services/xiaomi_miot/set_miot_property' \
--header 'Content-Type: application/json' \
--header 'Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiIwNDRjYTE4ZTEyOGI0NmY5ODNiNDU0ZDc3NDgyNjQyYiIsImlhdCI6MTcwODA3NTI5NiwiZXhwIjoyMDIzNDM1Mjk2fQ.rC_sN-FmIriOMwyXU8CtUM9Vg-doVHZszwqfSqDi8rQ' \
--data '{
    "entity_id":"fan.zhimi_v6_c97a_air_purifier",
    "siid": 2,
    "piid": 1,
    "value": true
}'


curl --location --request GET 'http://localhost:8123/api/states/sensor.a4c138ba6acb_temperature' \
--header 'Content-Type: application/json' \
--header 'Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiIwNDRjYTE4ZTEyOGI0NmY5ODNiNDU0ZDc3NDgyNjQyYiIsImlhdCI6MTcwODA3NTI5NiwiZXhwIjoyMDIzNDM1Mjk2fQ.rC_sN-FmIriOMwyXU8CtUM9Vg-doVHZszwqfSqDi8rQ' \
--data '{
    "entity_id":"sensor.a4c138ba6acb_temperature",
    "method":""
}'

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/Monodyee/article/detail/472781