第 28 章:智能家居实验
第 28 章:智能家居实验
本文整理自 Datawhale 开源项目 datawhalechina/jev-cookbook(CC BY-NC-SA 4.0),源文件:main/05_智能家居实验。
用同一套房间状态练习四件事:理解用户意图、检查动作依赖、由代码控制副作用、观察完整调用链。这里的 3D 房间是模拟器,不是实际家居设备控制器。
系统怎样工作
用户语句 → state 与候选问题 → Jev 返回类型化判断 → 应用代码检查权限和依赖 → 3D 模拟器更新状态 → 追踪面板记录结果。
Jev 负责语义判断,例如用户想开哪盏灯、指令是否包含多个动作;代码负责设备白名单、动作映射、顺序、失败处理和状态写入。模型回答“很可能是开灯”不等于程序已经授权开灯。
sequenceDiagram
actor User as 用户
participant App as 智能家居应用
participant Jev as 类型化判断接口
participant Policy as 确定性策略
participant Sim as 3D 模拟器
User->>App: 自然语言指令
App->>Jev: State 与候选问题
Jev-->>App: 选项概率与判断结果
App->>Policy: 检查目标、权限和动作依赖
alt 状态有效且动作允许
Policy->>Sim: 按依赖顺序执行
Sim-->>App: 更新模拟状态
else 不确定或不允许
Policy-->>App: 拒绝、澄清或升级
App-->>User: 说明处理结果
end复合指令的执行清单
用户:「关厨房灯,然后锁书房门」
动作 A:关闭 kitchen.light ─┐
├─ A 成功后再执行 B
动作 B:锁定 study.door ─┘
执行前:检查用户有权控制这两台设备,并读取当前设备版本
执行中:按依赖顺序运行;失败时记录已完成动作,不盲目重放整条指令
重试时:沿用同一 action_id,设备端 / 应用端防止重复副作用
执行后:读取新状态并写入追踪记录这里的“然后”表示顺序约束;若两项操作互不依赖,应用才可并行执行。以上是阅读 Notebook 时可对照的控制清单,不表示示例模拟器实现了真实设备授权或 exactly-once 语义。
Notebook 与演示覆盖什么
| 环节 | 场景 | 建议观察 |
|---|---|---|
| 推测性扇出 | “把咖啡烧上” | 一次请求打包多个潜在判断,应用只读取实际分支所需答案;记录问题数、输入量和往返 |
| 复合指令 | “关厨房灯,然后锁书房门” | 判断是否有依赖;有先后关系时由代码串行执行,无依赖时才考虑并行 |
| 同设备连续动作 | “先开客厅灯再关掉” | 检查状态读写是否保留指令顺序,避免并行竞争产生随机终态 |
| 状态查询 | “客厅灯开着吗” | 将语义解析与本地模拟状态读取分开 |
| 生成式模型对照 | 同一任务经不同处理链路 | 控制模型、提示词、网络、预热和输出长度;比较端到端延迟与费用,不把单次结果当普遍倍数 |
| 语音输入 | 麦克风流式识别 | 区分 ASR 首字延迟、识别完成时间、Jev 判断和动作执行时间 |
Notebook 展示一次投机调用;浏览器应用支持实时追踪。具体概率、耗时和费用会随模型版本、网络及配置变化,查看时请同时记录运行日期和所用端点。文字交互可以只运行 Notebook;语音与 3D 体验需要启动本地演示,并按其说明配置额外服务。
理论补充:依赖图、竞态与幂等
把多条动作看成有向图:若 B 必须等 A 完成,写入边 A → B;互不依赖的动作才可并行。它定义的是动作之间的部分顺序,不必把所有操作一律串行。模型可以提供“可能相关”的信号,但应用要用明确策略把它转换成执行顺序。阈值和动作关系属于本地策略,应由代表性指令集验证。
并发工作流要特别留意:
- 竞态:两个动作同时读取旧状态并写入冲突结果。
- 幂等:重试同一动作时避免重复开锁、重复扣款或重复创建任务。RFC 9110 §9.2.2用“多次相同请求的预期效果等同于一次请求”定义 HTTP 方法幂等性;业务动作也应设计可安全重试的语义,但日志等附带副作用仍可能重复。
- 过期状态:判断和执行之间设备状态可能已变化;执行前重新读取关键状态,或通过版本号条件更新,避免基于旧 state 执行。
- 失败恢复:记录已完成步骤,并定义补偿、回滚或人工介入方式。
现实系统还应在动作前检查用户权限、设备允许的动作和安全联锁。语义判断不能绕过这些确定性检查。
运行与下一步
配套服务与 3D 页面在 smart_home_demo。运行方法、环境变量和演示边界以该目录和 Notebook 内说明为准。先读第三章架构模式,再运行智能家居 Notebook;想看更完整的端到端应用,可接着读第七章。
对应官方示例 Smart Home Demo 的可运行复刻:3D 仿真屋 + 语音/文字指令 + 真实 Jev 投机提示管线。
配套内容在文件夹 smart_home_demo/:
serve_smart_home.py—— 本地服务(静态页 + Jev/ASR/chat 代理,密钥只走环境变量)smart_home_playground.html—— 单文件 3D 应用(Three.js + 追踪面板 + 成本统计)
上游仓库:Bald0Wang/jev-playground → smart-home(含实验报告 README 与演练场笔记本)。
1. 官方 Demo 的三个模式
- 投机提示(Speculative Prompting):一次
/v1/systemone调用捆绑全部问题——意图、是否复合、范围、房间、设备类别,外加五类设备的动作预判(灯/风扇/音响/家电/门锁,哪怕与本指令无关也问)。代码识别主目标后在代码端剪枝无关分支,零二次往返。 - 复合拆分:noul 判定复合(如"关厨房灯然后锁书房门")→ 交给小模型拆原子指令 → 每条并行再走一次 Jev。
- 路由守卫:常识问题("1989 世界大赛冠军")由 Jev 快速识别为信息请求,转交通用大模型回答——快而确定的走 Jev,开放式的走 LLM。
我们的复刻在此之上加了串/并行工作流编排(noul 判"谁必须先于谁")与传统 LLM(step-5-preview)同题对照、成本/token 实时统计。
2. 启动配套应用
服务默认端口 8810(本格用 8843 独立实例,避免和你正开着的页面打架)。需要 TYPESAFE_API_KEY;语音与 LLM 对照另需 STEPFUN_API_KEY(阶跃)。
import subprocess, sys, time, json, urllib.request, os
from pathlib import Path
PORT = 8843
BASE = f"http://127.0.0.1:{PORT}"
def up():
try:
return urllib.request.urlopen(f"{BASE}/api/health", timeout=2).status == 200
except Exception:
return False
if not up():
env = dict(os.environ)
env.setdefault("TYPESAFE_API_KEY", os.environ.get("TYPESAFE_API_KEY", ""))
subprocess.Popen([sys.executable, "serve_smart_home.py", "--port", str(PORT), "--no-open"],
cwd=str(next((d / "smart_home_demo" for d in [Path.cwd(), Path.cwd() / "05_智能家居实验"]
if (d / "smart_home_demo").exists()), Path.cwd() / "05_智能家居实验/smart_home_demo")), env=env,
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
for _ in range(20):
if up(): break
time.sleep(0.5)
print(json.dumps(json.loads(urllib.request.urlopen(f"{BASE}/api/health", timeout=2).read()),
ensure_ascii=False)){"jev": true, "asr": false, "chat": false, "asr_model": "stepaudio-2.5-asr", "chat_model": "step-1o-turbo-vision", "answer_model": "step-3.5-flash", "compare_model": "step-5-preview"}3. 在 Notebook 里复刻一次投机调用
不看页面也能体验核心机制:把官方式样的问题包(意图 + 复合判断 + 设备类别 + 三类动作预判)一次发给 Jev,观察返回的完整概率分布——包括与本指令无关的"投机"答案。
import json, urllib.request
state = {"utterance": "把咖啡烧上",
"home": {"rooms": ["living_room", "kitchen", "office", "bedroom", "entrance"],
"current_state": {"coffee": False, "light_kitchen": True}}}
turn = {"turn_on": "打开/启动", "turn_off": "关闭/停止", "leave_unchanged": "保持现状"}
payload = {"state": state, "model": "jev-latest", "questions": {
"intent": {"type": "choice", "instructions": "这句指令的意图是什么?", "criteria": {
"smart_home_command": "控制智能家居设备",
"information_request": "与智能家居无关的通用知识问题",
"smart_home_query": "询问设备当前状态"}},
"is_compound": {"type": "noul", "instructions": "这条指令包含两个或以上互相独立的操作吗?"},
"category": {"type": "choice", "instructions": "目标设备属于哪一类?", "criteria": {
"lighting": "灯", "appliance": "家电(如咖啡机)", "lock": "门锁",
"fan": "风扇", "speaker": "音响", "not_device_specific": "不涉及设备"}},
"appliance_action": {"type": "choice", "instructions": "投机预判:家电应该怎样?", "criteria": turn},
"light_action": {"type": "choice", "instructions": "投机预判:灯应该怎样?", "criteria": turn},
"lock_action": {"type": "choice", "instructions": "投机预判:门锁应该怎样?",
"criteria": {"lock": "上锁", "unlock": "解锁", "leave_unchanged": "保持"}},
}}
req = urllib.request.Request(f"{BASE}/api/jev", data=json.dumps(payload).encode(),
headers={"Content-Type": "application/json"})
wire = json.loads(urllib.request.urlopen(req, timeout=30).read())
print("上游耗时", wire["upstream_ms"], "ms")
for name, a in wire["body"]["answers"].items():
dist = a.get("probabilities") or {"noul": a.get("noul")}
top = max(dist, key=dist.get)
mark = " ← 采纳" if name in ("intent", "category", "appliance_action") else "(投机,代码端剪枝)"
print(f" {name:17s} -> {a.get('choice', a.get('noul'))!s:16s} "
f"top={top}({dist[top]:.2f}){mark}")上游耗时 777 ms
intent -> smart_home_command top=smart_home_command(1.00) ← 采纳
is_compound -> 0.09 top=noul(0.09)(投机,代码端剪枝)
category -> appliance top=appliance(1.00) ← 采纳
appliance_action -> turn_on top=turn_on(1.00) ← 采纳
light_action -> leave_unchanged top=leave_unchanged(0.97)(投机,代码端剪枝)
lock_action -> leave_unchanged top=leave_unchanged(0.97)(投机,代码端剪枝)观察与理解: 与咖啡机无关的 light_action / lock_action 也返回了完整分布(大概率是"保持现状"或不确定)——这就是投机提示:问都问了,真用到门锁的指令来时不用再跑一趟;本次用不到的,代码端直接丢弃。
4. 打开 3D 应用体验
内嵌窗口可直接操作(拖拽旋转 3D、点设备、切管线、双路对比);麦克风建议在独立标签页体验(部分浏览器限制 iframe 内录音)。
from IPython.display import HTML
HTML(f"""
独立标签页:{BASE}
· 语音需 STEPFUN_API_KEY · 无 TYPESAFE_API_KEY 时页面自动用本地模拟引擎
""")