Esp32Robot入门12-智能家居控制01：接入Home Assistant（场景联动：通过本地大模型控制办公室空调与灯光）

📌 文章简介：大模型智能机器人不仅是一个会聊天的伙伴，更是未来智能家居的“超级控制中枢”。今天，我们将打破云端闭环的壁垒，将基于 ESP32-S3 的智能语音助手接入本地物联网王者——Home Assistant（HA）。我们将深度科普 HA 在局域网自动化控制中的核心地位及网络发现协议，手把手教你在个人创作者后台生成高安全性的“长期访问令牌”（Long-Lived Access Token），并解密小智服务端（xiaozhi-esp32-server）的 config.json 与环境变量配置。此外，我们将通过精美的 Mermaid 序列图还原“从语音到物理控制”的端到端数据流，并提供一份开箱即用的 Python HTTP REST API 调试代码，助你轻松实现通过本地大模型控制办公室空调与灯光的梦幻场景！

一、 前言：大模型硬件时代，智能家居的“本地化脑干接入”

在传统的智能家居方案中，我们经常使用小爱同学、天猫精灵或小度音箱。这些产品虽然极大地方便了我们的生活，但作为极客和开发者，我们不得不面对以下几个直击痛点的硬伤：

  1. 🚨 严重依赖外网：一旦家里的宽带断网，或者厂家的云端服务器遭遇故障，智能音箱就会瞬间变成“哑巴”和“砖头”，连最简单的关灯操作都无法完成。
  2. 🔒 隐私泄露隐患：每一次语音采集、每一条控制指令都要上传到云端进行 ASR 识别与语义解析，对于注重隐私的家庭或企业办公室场景，这无疑埋下了一颗安全地雷。
  3. 🧱 生态闭环严重：跨品牌的设备极难联动，小米的音箱很难原生控制涂鸦的灯，或者直接接入非官方支持的自制硬件。

随着边缘计算的爆发，本地大模型（如 Qwen3.6-35B） 的推理能力与开源物联网核心 Home Assistant (HA) 的结合，为我们提供了一条完美的破局之路！

搭载 ESP32-S3 的智能语音机器人作为硬件载体，负责局域网内的语音采集与音频播放；小智网关服务端在本地处理音频并调度本地大模型，完成意图解析；最终，通过局域网的 REST API 直接向 Home Assistant 发送物理控制指令。

这，就是 100% 局域网运行、安全、极速的“本地化脑干接入”方案！ 本文将作为智能家居控制系列的第一弹，带你打通这套硬核架构的“任督二脉”！

二、 Home Assistant 本地中枢与生态准备

在正式编写代码前，我们需要先理解并准备好我们的本地物联网中枢——Home Assistant。

1. Home Assistant 在本地自动化中的至高地位

Home Assistant (HA) 是目前全球最强大的开源智能家居自动化平台。它不生产任何硬件，但它几乎“包容一切”。

  • 极强的设备兼容性：它能把 Zigbee、Bluetooth、Wi-Fi、Modbus 甚至是空调红外转发器等上万种不同协议、不同品牌的硬件，统一抽象为 HA 内部的“实体（Entity）”。
  • 本地网络发现协议（mDNS/SSDP）：HA 在局域网内通过组播 DNS（mDNS）和简单服务发现协议（SSDP）来自动侦测新设备。例如，当你的 ESP32 机器人或小智网关上线时，它们可以通过 mDNS 向局域网宣告自己的存在，实现“免配置自发现”。

2. 手把手创建“长期访问令牌”（Long-Lived Access Token）

由于我们的 ESP32 服务端网关需要作为一个独立的第三方系统接入 HA 并调用其 API，我们必须进行安全身份认证。HA 提供了极其安全的 长期访问令牌（Long-Lived Access Token） 机制。它基于 JWT（JSON Web Token）架构，有效期长达 10 年，且可以随时在后台手动撤销，非常适合本地局域网脚本和网关开发。

下面是详细的生成步骤：

💡 操作指南：

  1. 打开浏览器，登录你的 Home Assistant Web 控制台（默认地址通常是 http://192.168.X.X:8123）。
  2. 点击左下角你的 个人头像（用户资料设置页）。
  3. 在右侧个人资料页面中，向下滑动到最底部。
  4. 找到 “长期访问令牌 (Long-Lived Access Tokens)” 区域，点击 “创建令牌 (Create Token)”。
  5. 在弹出的对话框中，输入一个识别名称（例如：xiaozhi_esp32_robot），然后点击确定。
  6. 此时屏幕上会弹出一串极其冗长的 Base64 密文。【⚠️ 警告】请立即完整复制这串 Token 并保存到安全的本地记事本中！一旦关闭此弹窗，你将再也无法查看它的内容，只能删除重新创建！
🔑 认证方式对比表
认证方式 安全性等级 维护成本 局域网离线可用性 适用开发场景
用户名 & 密码 🚨 低（容易被暴力破解，不适宜硬编码） 高（修改密码需同步更新代码） 是 仅限于浏览器控制台登录
长期访问令牌 (JWT) 🛡️ 极高（单设备独立 Token，可随时撤销） 低（一次生成，永久有效） 是 第三方私有网关/ESP32服务端对接（首选）
OAuth 2.0 授权流 🛡️ 极高（支持细粒度权限范围 scope 限制） 高（需要搭建 OAuth 认证回调服务） 否（本地化配置极其繁琐） 跨公网多租户平台集成（如 Alexa/Google Home 接入）

3. 办公室物理设备实体准备

为了配合本教程的实战场景，我们在 Home Assistant 中虚拟或接入了两个典型的办公室设备，请记住它们的 实体 ID（Entity ID），后续的代码和配置都会围绕它们展开：

  • 办公室大灯：light.office_light （支持开关 on/off 和亮度 brightness 调节）
  • 办公室空调：climate.office_air_conditioner （支持模式 hvac_mode 和目标温度 temperature 调节）

三、 网关服务端的对接配置

小智网关服务端（xiaozhi-esp32-server）扮演着把“自然语言意图”翻译为“物联控制指令”的角色。我们需要在服务端中配置 HA 的局域网 API 访问端点和长期访问令牌。

1. 配置文件 config.json 详细解密

在小智服务端的根目录下，打开或新建 config.json 配置文件。我们加入 home_assistant 的专属配置节点。请仔细阅读以下配置文件中的每一行中文注释：

{
  "server": {
    "port": 8000,
    "log_level": "info"
  },
  "home_assistant": {
    "enabled": true,
    "host": "http://192.168.31.250:8123",
    "token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiIzNTRmYmJhNmUyMDk0OTFlODk0NDFkYTJiMjg0OTI2YiIsImlhdCI6MTcxNjI3NTc1MiwiZXhwIjoyMDMxODMxNzUyfQ.YourActualTokenHere_ReplaceThisWithYourOwnTokenValue",
    "request_timeout_ms": 5000,
    "entity_aliases": {
      "大灯": "light.office_light",
      "顶灯": "light.office_light",
      "办公室灯": "light.office_light",
      "空调": "climate.office_air_conditioner",
      "冷气": "climate.office_air_conditioner"
    }
  }
}

2. 生产环境环境变量配置（.env）

如果你是使用 Docker 容器化部署小智服务端，推荐通过 .env 环境变量文件来进行解耦配置，防止 Token 泄露到代码仓库中：

# =========================================================================
# Home Assistant 局域网集成配置
# =========================================================================
# 是否启用物联网联动插件 (true / false)
HA_ENABLED=true

# Home Assistant 本地局域网 API 端点（切记不要写公网域名，以保证毫秒级响应）
HA_BASE_URL=http://192.168.31.250:8123

# 填入刚刚在 HA 个人后台生成的 Long-Lived Access Token
HA_ACCESS_TOKEN=eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiIzNTRmYmJh...

# 请求超时限制（毫秒），建议设为 5000ms 避免网络死锁
HA_TIMEOUT=5000

3. 大模型 Tool Use (Function Calling) 意图解析设计原理

小智网关通过在 System Prompt 中向本地大模型（如 Qwen3.6-35B）声明“可调用的工具（Tools）”，来实现语义控制。其底层向大模型发送的 JSON 描述类似于：

{
  "name": "control_home_assistant",
  "description": "控制本地 Home Assistant 智能家居设备（如电灯、空调、窗帘等）。支持开关、亮度调节、温度模式设置。",
  "parameters": {
    "type": "object",
    "properties": {
      "entity_id": {
        "type": "string",
        "description": "要控制的设备实体ID，例如 light.office_light 或 climate.office_air_conditioner"
      },
      "action": {
        "type": "string",
        "enum": ["turn_on", "turn_off", "set_brightness", "set_temperature", "set_hvac_mode"],
        "description": "要执行的动作"
      },
      "value": {
        "type": "number",
        "description": "可选参数。当 action 为 set_brightness 时代表亮度百分比 (1-100)；当 action 为 set_temperature 时代表温度数值 (16-30)"
      }
    },
    "required": ["entity_id", "action"]
  }
}

当用户说：“帮我把办公室空调温度调到24度。” 大模型通过推理，并不会直接去调用 API，而是会输出如下的结构化 Tool Call：

{
  "tool": "control_home_assistant",
  "arguments": {
    "entity_id": "climate.office_air_conditioner",
    "action": "set_temperature",
    "value": 24
  }
}

网关捕获到这个 Tool Call 后，在本地代码中通过 HTTP 协议向 Home Assistant 发起真实的 API 请求，这就完成了整套智能控制链路。

四、 联动控制数据流全景解析

1. 端到端数据流控制序列图

我们可以通过以下精美的 Mermaid 序列图，直观地观察到从“用户发出语音”到“空调/灯光做出响应”的完整端到端数据流向：

🔌 物理空调/灯光 🏡 Home Assistant 🧠 本地大模型 (Qwen) 🗣️ Faster-Whisper (ASR) 🖥️ 小智网关 (Node.js) 🤖 ESP32 语音终端 🔌 物理空调/灯光 🏡 Home Assistant 🧠 本地大模型 (Qwen) 🗣️ Faster-Whisper (ASR) 🖥️ 小智网关 (Node.js) 🤖 ESP32 语音终端 硬件使用 I2S 采集音频 并进行 OPUS 压缩编码 LLM解析语义并触发 Function Calling (Tool Call) par [并行执行物理控制] 👤 用户 "把办公室的灯打开，空调温度调到24度" 1 WebSocket 发送流式音频数据 2 音频流丢入本地 ASR 引擎 3 返回识别文本: "把办公室的灯打开，空调温度调到24度" 4 携带智能家居 System Tools 的推理请求 5 返回 Tool Call JSON: {"tool": "control_ha", ...} 6 POST /api/services/light/turn_on (entity_id: light.office_light) 7 通过 Zigbee 协议下发开灯指令 8 返回灯已点亮状态 9 POST /api/services/climate/set_temperature (entity_id: ..., temp: 24) 10 通过 红外/Wi-Fi 调节空调至 24℃ 11 返回空调运行状态 12 返回 HTTP 200 OK 执行成功 13 将设备执行结果反馈给大模型作为上下文 14 生成自然语言回复: "好的，办公室大灯已为您开启，空调已成功调节至24度。" 15 TTS 语音合成并推送音频流 (WebRTC/WebSocket) 16 扬声器播放: "好的，办公室大灯已为您开启..." 17 👤 用户

五、 纯 Python 编写的局域网 REST API 联动测试脚本

为了在不启动整个大模型和语音网关的情况下，快速验证我们的网络通路、实体 ID 和长期访问令牌是否正确，我们需要编写一个独立的、高健壮性的 Python 测试脚本。

该脚本使用 requests 库，采用面向对象的设计思路，完美封装了 Home Assistant 的 REST API 操作，包含查询实体状态、开关灯光、调节亮度和设置空调温度。

1. 完整源码 ha_test_client.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""
文件名: ha_test_client.py
描述: 智能家居控制测试脚本，用于验证小智网关与本地 Home Assistant (HA) REST API 的网络连通性与控制逻辑。
主要功能:
  1. 验证 HA 长期访问令牌 (Long-Lived Access Token) 是否有效。
  2. 获取特定实体 (如灯光 light.office_light) 的当前状态。
  3. 向 HA 发送控制服务请求 (开灯、调光、设置空调温度)。
"""

import sys
import json
import requests

# 控制台彩色输出定义 (ANSI Escape Codes)
COLOR_GREEN = "\033[92m"
COLOR_YELLOW = "\033[93m"
COLOR_RED = "\033[91m"
COLOR_BLUE = "\033[94m"
COLOR_RESET = "\033[0m"

# =========================================================================
# ⚙️ 配置区域 - 请根据你本地的实际环境进行修改！
# =========================================================================
HA_BASE_URL = "http://192.168.31.250:8123"  # 你的 Home Assistant 局域网访问地址
HA_TOKEN = "YOUR_LONG_LIVED_ACCESS_TOKEN_HERE"  # 替换为你自己生成的长期访问令牌

class HomeAssistantClient:
    """
    Home Assistant REST API 本地测试客户端类
    """
    def __init__(self, base_url, token):
        self.base_url = base_url.rstrip('/')
        self.headers = {
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json"
        }

    def ping_api(self):
        """
        验证 API 连通性与 Token 有效性
        请求端点: GET /api/
        """
        url = f"{self.base_url}/api/"
        print(f"{COLOR_BLUE}[INFO]{COLOR_RESET} 正在验证与 Home Assistant 的连接: {url}...")
        try:
            response = requests.get(url, headers=self.headers, timeout=5)
            if response.status_code == 200:
                data = response.json()
                print(f"{COLOR_GREEN}[SUCCESS]{COLOR_RESET} 成功连接到 Home Assistant！HA 状态: {data.get('message')}")
                return True
            else:
                print(f"{COLOR_RED}[ERROR]{COLOR_RESET} 认证失败，请检查 Token。HTTP 状态码: {response.status_code}")
                return False
        except requests.exceptions.RequestException as e:
            print(f"{COLOR_RED}[ERROR]{COLOR_RESET} 局域网连接超时或失败，请检查 IP 地址和端口！详情: {e}")
            return False

    def get_entity_state(self, entity_id):
        """
        获取指定实体的当前状态与属性
        请求端点: GET /api/states/<entity_id>
        """
        url = f"{self.base_url}/api/states/{entity_id}"
        try:
            response = requests.get(url, headers=self.headers, timeout=5)
            if response.status_code == 200:
                state_data = response.json()
                state_value = state_data.get("state")
                attributes = state_data.get("attributes", {})
                friendly_name = attributes.get("friendly_name", entity_id)
                print(f"{COLOR_GREEN}[SUCCESS]{COLOR_RESET} 设备 [{friendly_name}] 当前状态: {COLOR_YELLOW}{state_value}{COLOR_RESET}")
                return state_data
            elif response.status_code == 404:
                print(f"{COLOR_RED}[ERROR]{COLOR_RESET} 找不到该实体 ID: {entity_id}，请检查 HA 中是否存在！")
                return None
            else:
                print(f"{COLOR_RED}[ERROR]{COLOR_RESET} 获取状态失败，HTTP 状态码: {response.status_code}")
                return None
        except Exception as e:
            print(f"{COLOR_RED}[ERROR]{COLOR_RESET} 请求异常: {e}")
            return None

    def call_service(self, domain, service, service_data):
        """
        调用 Home Assistant 服务执行具体控制动作
        请求端点: POST /api/services/<domain>/<service>
        """
        url = f"{self.base_url}/api/services/{domain}/{service}"
        print(f"{COLOR_BLUE}[INFO]{COLOR_RESET} 正在发送控制命令: [{domain}.{service}] -> 数据: {json.dumps(service_data, ensure_ascii=False)}")
        try:
            response = requests.post(url, headers=self.headers, json=service_data, timeout=5)
            if response.status_code == 200:
                print(f"{COLOR_GREEN}[SUCCESS]{COLOR_RESET} 命令下发成功！物理设备已响应。")
                return response.json()
            else:
                print(f"{COLOR_RED}[ERROR]{COLOR_RESET} 命令执行失败，HTTP 状态码: {response.status_code}，详情: {response.text}")
                return None
        except Exception as e:
            print(f"{COLOR_RED}[ERROR]{COLOR_RESET} 请求服务异常: {e}")
            return None


# =========================================================================
# 🚀 脚本主入口
# =========================================================================
if __name__ == "__main__":
    # 1. 检查配置项是否被替换
    if HA_TOKEN == "YOUR_LONG_LIVED_ACCESS_TOKEN_HERE":
        print(f"{COLOR_RED}[ERROR]{COLOR_RESET} 请先在代码中配置您真实的 HA_TOKEN 长期访问令牌！")
        sys.exit(1)

    # 2. 实例化客户端
    client = HomeAssistantClient(base_url=HA_BASE_URL, token=HA_TOKEN)

    # 3. 运行连通性测试
    if not client.ping_api():
        print(f"{COLOR_RED}[FATAL]{COLOR_RESET} 基础网络或认证未通过，程序退出。")
        sys.exit(1)

    print("\n" + "="*50 + "\n")

    # 4. 测试场景 1：办公室大灯 (light.office_light) 状态查询与控制
    target_light = "light.office_light"
    print(f"--- 💡 正在执行灯光控制测试 [{target_light}] ---")
    
    # 步骤 A: 查询初始状态
    client.get_entity_state(target_light)

    # 步骤 B: 发送开灯指令，并设置亮度为 80% (亮度范围通常为 0-255)
    # 计算 80% 对应的亮度值: 255 * 0.8 ≈ 204
    light_on_data = {
        "entity_id": target_light,
        "brightness": 204
    }
    client.call_service("light", "turn_on", light_on_data)

    # 步骤 C: 再次查询状态，验证是否成功开启
    client.get_entity_state(target_light)
    
    print("\n" + "="*50 + "\n")

    # 5. 测试场景 2：办公室空调 (climate.office_air_conditioner) 状态查询与控制
    target_climate = "climate.office_air_conditioner"
    print(f"--- ❄️ 正在执行空调控制测试 [{target_climate}] ---")

    # 步骤 A: 查询空调当前运行属性（如当前温度、设定温度、当前模式）
    climate_state = client.get_entity_state(target_climate)
    if climate_state:
        attrs = climate_state.get("attributes", {})
        current_temp = attrs.get("current_temperature", "未知")
        target_temp = attrs.get("temperature", "未知")
        print(f"空调当前室内温度: {COLOR_YELLOW}{current_temp}℃{COLOR_RESET}，目标设定温度: {COLOR_YELLOW}{target_temp}℃{COLOR_RESET}")

    # 步骤 B: 调节空调设定温度为 24℃
    climate_temp_data = {
        "entity_id": target_climate,
        "temperature": 24
    }
    client.call_service("climate", "set_temperature", climate_temp_data)

    # 步骤 C: 切换空调模式为制冷 (cool) 并开启
    climate_mode_data = {
        "entity_id": target_climate,
        "hvac_mode": "cool"
    }
    client.call_service("climate", "set_hvac_mode", climate_mode_data)

    print("\n" + "="*50 + "\n")
    print(f"{COLOR_GREEN}[FINISH]{COLOR_RESET} 本次局域网 REST API 控制连通性测试全部完成！")

2. 运行与排障指南

在本地终端中，推荐使用 uv 这一高效包管理工具来运行该脚本：

  1. 安装依赖：
    该测试脚本仅依赖 requests 库，如果尚未安装，可直接通过 uv 极速安装： 
    uv pip install requests -i https://pypi.tuna.tsinghua.edu.cn/simple
  2. 运行测试： 
    uv run python ha_test_client.py
  3. 常见局域网排障提示：
    • 请求超时 (Timeout)：请确保测试运行的电脑与 Home Assistant 宿主机处于同一个局域网网段内，并且确认没有开启可能导致广播隔离的路由器安全设置（如 AP 隔离）。
    • 401 Unauthorized：99% 是因为你的长期访问令牌复制不完整（注意不要遗漏开头或结尾的字符），或者 Token 已经在 HA 后台被意外撤销。
    • 404 Not Found：请仔细比对 Home Assistant 中的“开发工具 -> 状态”页面，确认你的实体 ID 是否拼写正确，例如是 light.office_light 还是 light.office_light_1。

六、 ✅ 本文总结

通过本篇教程，我们成功实现了大模型机器人的“物联网脑干接入”的第一步：

  1. 💡 理清了本地物联网的优势：利用本地 Home Assistant 摆脱对云端服务器的依赖，确保了毫秒级的局域网响应速度与绝对的隐私安全。
  2. 🔑 掌握了高安全级的 Token 生成：学会了如何在 HA 个人后台生成JWT架构的“长期访问令牌”，保证了开发接入的规范与安全。
  3. ⚙️ 打通了网关配置：深度解密了小智网关服务端的 config.json 与 .env 环境变量配置方式，明确了参数的具体含义。
  4. 📊 还原了控制全景数据流：利用 Mermaid 绘制了从语音输入到物理开关的“端到端序列图”，并编写了纯 Python 版本的 API 控制类，实现了设备状态的精准查询与指令下发。

📢 下一篇预告

基础链路已经打通，接下来我们要进入整套系统最激动人心的部分了！

下一篇中，我们将深入《Esp32Robot入门13-智能家居控制02：编写HA Agent Tool（智能体实战：基于LangChain赋予大模型设备控制工具链）》。我们将手把手教你使用 LangChain 框架为本地大模型（Qwen3.6）构建智能家居的 Tool Agent，让大模型学会像人类一样思考，并自动根据用户的模糊指令（例如：“太热了，帮我降降温”或者“光线有点刺眼”）来自主进行逻辑判断、参数拆解，并自动匹配并调用工具，控制 Home Assistant 设备！

精彩不容错过，我们下期见！👋

# 智能家居
Logo
https://edu.csdn.net/learn/39067/627173?utm_source=2019755004

汇聚全球AI编程工具，助力开发者即刻编程。

更多推荐

cover

四天蒸发近2000亿，宇树科技股价近乎“腰斩”；苹果举办库克告别派对：9月1日正式卸任CEO；Anthropic两款新Claude模型曝光 | 极客头条

avatar AI编程社区
cover

Claude Code反超GitHub Copilot登顶第一、90%程序员已用上Agent，最新AI编码调查报告来了

avatar AI编程社区
cover

如何用好 Claude Code 写后端代码

avatar AI编程社区