在做语音助手的项目时，最容易出现的落差是：演示视频里它在“聊天”环节表现很好，真正接到需求里就干不了活。我们这个自研智能音箱项目（内部戏称为“天猫精灵”）上篇已经跑通了“唤醒词—语音转文字—回复播报”的最小闭环，但那只是一条能说话的链路。下集要解决的是另一件事：让同一个入口能够稳定地把“开灯”“定个 25 分钟闹钟”这类指令落地，并且新增功能时不用改主流程。下面这份笔记覆盖意图路由、技能插件化、槽位抽取、多轮会话、异步定时、MQTT 下发、接口层与线上排错，适合已经写过一点 Python 异步代码、想把语音类 Demo 做成可维护服务的同学。

1. 项目背景与“下集”要解决的问题

1.1 上集回顾：最小可用闭环长什么样

上篇的代码量很小，一条链路上的四个环节串起来就算成功：设备端录音、调用语音识别拿到文本、用if "开灯" in text这种硬判断生成回复、再调用语音合成播出去。这段代码在一个人调试时完全够用，因为它只有一条路径，输入输出都是确定的。

问题在于它的扩展方式极其脆弱。每加一个功能，就要在同一个函数里再塞一个if，而且判断顺序决定结果，谁写在前面谁赢。当指令涨到十几条，函数会变成一个又长又难读的分支树，改一个关键词可能顺带打断另一个功能，测试也无从下手。这个阶段的项目不是“错”，它只是还没到需要架构的时候；而一旦要交付给真实使用者，就必须重构。

1.2 下集目标：从“能对话”到“能干活”

这一篇我们给自己定的目标很具体，一共四条。第一条是把意图识别从硬编码分支中拆出来，形成独立的路由层，任何新指令只能通过“注册”进入系统，不允许修改路由代码。第二条是定义技能（Skill）的标准协议，让每个功能自己负责匹配、抽参和执行，互不感知。第三条是补上多轮会话，让“开灯”这种缺参数的指令能够追问一轮再执行，而不是直接失败。第四条是补齐工程细节：超时控制、异常兜底、日志埋点、配置外置。

这四条做完之后，系统的形态会从“一个大函数”变成“一个调度器 + 若干插件”。这个变化带来的直接收益是：新同学接手只需要按模板写一个新的技能文件，不需要读懂整个调度器；线上出问题时日志能告诉你“命中了哪个技能、抽到了哪些槽位、执行耗时多少”，而不是只看到一句笼统的报错。

1.3 三个必须分清的术语：意图、槽位、技能

先把术语统一，后面代码才不会歧义。意图（Intent）指的是用户“想干什么”，比如device.on、timer.create，它是动作的名字，不包含对象。槽位（Slot）指的是动作需要的参数，比如开灯这个意图里，“灯”是设备槽位，“卧室”是位置槽位，它们决定了动作作用在谁身上。

技能（Skill）是工程上的概念，它把“识别意图 + 抽槽位 + 执行动作”打包成一个可注册的类。一个技能可以对应多个意图（比如设备技能同时处理开和关），也可以只处理一个。把意图和技能分开看，是因为意图是面向用户的语义，技能是面向代码的实现单元，两者不必一一对应。想清楚这层关系，后面的路由设计就顺了。

1.4 为什么不用大模型直接端到端

有同学会问，既然现在大模型这么强，为什么不把用户的话直接丢给它，让它输出结构化 JSON？这当然是一条路，本文的架构也完全可以把意图识别换成模型调用，接口不用改。但在当前阶段我们仍然选择规则优先，原因有三个，都是工程侧的考虑。

一是延迟。语音交互对响应时间敏感，用户说完话到听到回复，超过两秒就会明显觉得卡。本地规则匹配是微秒级，而一次模型调用即使很快也在几百毫秒量级，加上网络抖动不确定性更大。二是可测性。规则匹配的输入输出是确定的，可以写单元测试；模型输出存在波动，测试只能做模糊断言。三是成本与稳定性，规则没有调用费用，也不会因为外部服务波动而整体不可用。

合理做法是分层：规则先兜住高频、格式固定的指令（开灯、定时、音量），当规则全部低于阈值时，再把这句话交给模型兜底，让模型只负责“闲聊”和“模糊意图”，不负责关键动作。这样既保住了响应速度和确定性，也不会让用户觉得系统很笨。

2. 环境准备与工程结构

2.1 运行环境与依赖说明

本文示例使用 Python 3.10 及以上版本，原因是代码里用到了X | None这种联合类型写法以及asyncio较新的 API。如果你的环境是 3.8/3.9，把类型注解改成Optional[...]即可，逻辑不用动。Web 层用 FastAPI，它原生支持异步，和我们的技能体系（异步函数）天然对齐；消息下发用 paho-mqtt，用于把控制指令推给设备侧。

版本方面需要提醒一句：paho-mqtt 从 1.x 到 2.x 有一个破坏性变更，创建客户端时要求显式传入回调 API 版本，直接沿用旧写法会抛TypeError。本文后面给出的封装做了兼容处理，两种版本都能跑。其余依赖（FastAPI、uvicorn、pydantic）请以你本地实际安装的版本为准，本文重点演示配置与代码思路，不绑定具体小版本号。

# 建议在虚拟环境中操作 python -m venv .venv source .venv/bin/activate # Windows 用 .venv\Scripts\activate pip install fastapi uvicorn paho-mqtt python-dotenv

如果你要验证 MQTT 链路，本地可以起一个轻量 broker 用于测试，测试环境里怎么部署都行，但不要用生产 broker 做实验：

# 仅用于本地联调，示例命令按你实际安装的 broker 调整 mosquitto -p 1883 -v

2.2 目录结构设计

我们采用“路由层 + 技能层 + 基础设施层”三段式，目录不深，但边界清晰。路由层只做分发，技能层只做业务，基础设施层封装 MQTT、会话存储、配置读取这些通用能力。这样划分之后，每个目录的职责一句话就能说清，后续加功能时不需要反复权衡“这段代码该放哪”。

voice-assistant/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 入口 │ ├── router.py # 意图路由与调度 │ ├── config.py # 配置加载 │ ├── session.py # 会话上下文存储 │ ├── infra/ │ │ ├── __init__.py │ │ └── mqtt_client.py # MQTT 发布封装 │ └── skills/ │ ├── __init__.py │ ├── base.py # 技能基类与注册表 │ ├── device.py # 家电控制技能 │ └── timer.py # 定时/番茄钟技能 ├── .env └── requirements.txt

有一点要特别注意：skills/__init__.py不是空文件，它负责导入所有技能模块，触发注册装饰器执行。如果只写了技能类却忘了在__init__.py里导入，程序启动后注册表是空的，所有指令都会走到兜底回复。这个坑非常隐蔽，因为代码看起来毫无问题，只是“没生效”，后面排错章节会再提一次。

2.3 配置外置与环境变量

配置一律不写死在代码里，尤其是 broker 地址、端口、认证信息和各种超时阈值。这些值在开发、测试、生产三套环境里必然不同，写死意味着发版前要改代码，风险很高。我们用.env文件承载，通过python-dotenv加载，代码里只保留默认值兜底。

# 文件路径：.env（不要提交到代码仓库） MQTT_HOST=127.0.0.1 MQTT_PORT=1883 MQTT_USERNAME= MQTT_PASSWORD= MQTT_CLIENT_ID=voice-server-01 # 业务阈值 INTENT_MATCH_THRESHOLD=0.35 SKILL_TIMEOUT_SECONDS=5 SESSION_TTL_SECONDS=180

# 文件路径：app/config.py import os from dataclasses import dataclass from dotenv import load_dotenv load_dotenv() @dataclass(frozen=True) class Settings: mqtt_host: str = os.getenv("MQTT_HOST", "127.0.0.1") mqtt_port: int = int(os.getenv("MQTT_PORT", "1883")) mqtt_username: str | None = os.getenv("MQTT_USERNAME") or None mqtt_password: str | None = os.getenv("MQTT_PASSWORD") or None mqtt_client_id: str = os.getenv("MQTT_CLIENT_ID", "voice-server-01") intent_threshold: float = float(os.getenv("INTENT_MATCH_THRESHOLD", "0.35")) skill_timeout: float = float(os.getenv("SKILL_TIMEOUT_SECONDS", "5")) session_ttl: int = int(os.getenv("SESSION_TTL_SECONDS", "180")) settings = Settings()

这里用了frozen=True的 dataclass，好处是配置在进程生命周期内不可变，避免某个函数顺手改了全局阈值导致行为漂移。阈值单独抽出来也有实际意义：意图匹配的分数线需要根据真实语料调，调参时只改环境变量重启即可，不用重新构建镜像。

3. 核心机制拆解：意图路由与技能插件化

3.1 路由层到底在解决什么问题

路由层的职责只有一句：给定一段用户文本和一个会话上下文，决定交给哪个技能处理，以及带上哪些参数。它不该关心设备怎么开、闹钟怎么定，那些是技能的事。这条边界一旦守住，路由代码的复杂度就是恒定的，不会随着功能增加而膨胀。

我们采用“打分 + 取最高”的策略，而不是“按顺序匹配第一个命中”。原因在于关键词存在天然的重叠：打开和开都会被设备技能命中，五分钟既可能出现在定时里也可能出现在倒计时里。顺序匹配要求开发者精确掌握所有技能的优先级关系，这种隐式依赖迟早会出错。打分机制则把冲突显式化：每个技能自己声明匹配度，调度器只做比较，谁高用谁，分数一样还可以按优先级字段兜底。

3.2 技能基类与注册装饰器的设计

先看基类。它定义了三样东西：技能名（唯一标识）、匹配方法（返回 0 到 1 的分数）、处理方法（返回结果对象）。结果对象里除了回复文本，还额外带了need_slot和keep_session两个字段，前者用于触发多轮追问，后者告诉会话层“这次需要记住上下文”。这两个字段是多轮对话的基础，缺了它们就只能一次性问答。

# 文件路径：app/skills/base.py from __future__ import annotations from dataclasses import dataclass, field from typing import Any, Dict, List, Optional, Type @dataclass class SkillContext: """一次请求的上下文，随请求创建，随请求销毁。""" session_id: str user_id: str text: str slots: Dict[str, Any] = field(default_factory=dict) memory: Dict[str, Any] = field(default_factory=dict) @dataclass class SkillResult: """技能执行结果。""" reply: str data: Dict[str, Any] = field(default_factory=dict) need_slot: Optional[str] = None # 非空表示需要用户补充该槽位 keep_session: bool = False # 是否保留会话上下文继续多轮 class Skill: name: str = "" priority: int = 0 async def match(self, ctx: SkillContext) -> float: """返回 0~1 的匹配度，0 表示完全不匹配。""" return 0.0 async def handle(self, ctx: SkillContext) -> SkillResult: """执行技能，子类必须实现。""" raise NotImplementedError _REGISTRY: Dict[str, Skill] = {} def register(skill_cls: Type[Skill]) -> Type[Skill]: """类装饰器：实例化并登记到全局注册表。""" instance = skill_cls() if not instance.name: raise ValueError(f"{skill_cls.__name__} 缺少 name 属性") if instance.name in _REGISTRY: raise ValueError(f"技能名重复：{instance.name}") _REGISTRY[instance.name] = instance return skill_cls def get_registry() -> Dict[str, Skill]: return _REGISTRY

注册表用模块级字典实现，进程启动时一次性填充，运行期只读。这带来一个额外好处：启动后可以打印注册表内容做一次自检，能提前发现“技能没被导入”“技能名重复”这类问题。至于技能实例是单例还是每次新建，这里选单例，因为技能本身应当无状态，所有请求相关数据都放在SkillContext里，不放在self上。

3.3 匹配分数怎么算才靠谱

匹配算法不需要多复杂，但必须可解释。我们的做法是三层加权：关键词命中给基础分，设备别名或数字等实体命中给加分，文本长度做归一化惩罚（避免一句话很长但只命中一个字的误判）。最终分数截断在 0 到 1 之间，小于阈值的直接视为不匹配。

这样做的好处是调试直观。线上发现某句话被错误路由，看日志里的分数和命中的关键词就知道原因，而不是面对一个黑盒分数猜。对于中文短指令，规则匹配的准确率其实相当高，因为用户表达方式有限，跑一段时间把没命中的语句捞出来补充词表，覆盖率就能稳步上去。

3.4 中文数字与设备别名的抽取

槽位抽取里最容易踩坑的是中文数字。用户说“定个二十五分钟的闹钟”，文本里不会出现阿拉伯数字，直接正则\d+会抽空。所以需要一层中文数字转换，覆盖 1 到 99 就够了，更大的数字在语音指令里极少出现。

# 文件路径：app/skills/utils.py from typing import Optional _DIGITS = { "零": 0, "一": 1, "二": 2, "两": 2, "三": 3, "四": 4, "五": 5, "六": 6, "七": 7, "八": 8, "九": 9, } def cn_to_int(text: str) -> Optional[int]: """把 1~99 的中文数字或阿拉伯数字转成 int，失败返回 None。""" if not text: return None text = text.strip() if text.isdigit(): return int(text) if text == "十": return 10 if "十" in text: left, _, right = text.partition("十") tens = _DIGITS.get(left, 1) if left else 1 ones = _DIGITS.get(right, 0) if right else 0 if tens is None or ones is None: return None return tens * 10 + ones if len(text) == 1: return _DIGITS.get(text) return None

设备别名则用一个显式映射表，把用户的各种叫法收敛到统一的设备 ID。这一步很关键：如果下游 MQTT 主题按用户口语构造，主题会变得五花八门，设备订阅端根本没法写。统一 ID 之后，“客厅的灯”“客厅灯”“大灯”都落到light.living_room，下游只认这一个值。

# 文件路径：app/skills/device_alias.py DEVICE_ALIASES = { "light.living_room": ["客厅的灯", "客厅灯", "大灯", "客厅大灯"], "light.bedroom": ["卧室的灯", "卧室灯", "房间的灯"], "ac.bedroom": ["卧室空调", "房间空调", "卧室的空调"], } def find_device(text: str) -> str | None: """从文本中找设备 ID，命中越长的别名优先级越高。""" best_id, best_len = None, 0 for device_id, aliases in DEVICE_ALIASES.items(): for alias in aliases: if alias in text and len(alias) > best_len: best_id, best_len = device_id, len(alias) return best_id

注意这里用“最长别名优先”而不是遍历顺序优先。因为“客厅灯”是“客厅的灯”的子串，如果按字典顺序先撞上短的，可能匹配到别的设备。这类细节在别名表变大之后会变得非常重要，早点写对能省很多排查时间。

4. 完整实战：两个可直接落地的技能

4.1 实现家电控制技能（含 MQTT 下发）

先写基础设施层的 MQTT 封装。要点有三个：客户端只创建一次并复用；断线后自动重连（loop_start会处理，但要有重连日志）；发布失败不能抛到业务层，要降级成“设备暂时没响应”的友好回复。生产环境还要考虑 QoS 和保留消息，本地联调先用默认值即可。

# 文件路径：app/infra/mqtt_client.py import json import logging from typing import Any, Dict import paho.mqtt.client as mqtt from app.config import settings logger = logging.getLogger(__name__) class DevicePublisher: def __init__(self) -> None: # paho-mqtt 2.x 要求显式指定回调 API 版本，1.x 没有该参数 try: self._client = mqtt.Client( mqtt.CallbackAPIVersion.VERSION2, client_id=settings.mqtt_client_id, ) except AttributeError: self._client = mqtt.Client(client_id=settings.mqtt_client_id) if settings.mqtt_username: self._client.username_pw_set( settings.mqtt_username, settings.mqtt_password ) self._client.on_connect = self._on_connect self._client.on_disconnect = self._on_disconnect self._client.reconnect_delay_set(min_delay=1, max_delay=30) def start(self) -> None: self._client.connect_async(settings.mqtt_host, settings.mqtt_port, keepalive=60) self._client.loop_start() def stop(self) -> None: self._client.loop_stop() self._client.disconnect() def publish_command(self, device_id: str, command: Dict[str, Any]) -> bool: topic = f"home/device/{device_id}/cmd" payload = json.dumps(command, ensure_ascii=False) try: info = self._client.publish(topic, payload, qos=1, timeout=3) info.wait_for_publish(timeout=3) logger.info("mqtt publish ok topic=%s payload=%s", topic, payload) return True except Exception as exc: # 网络异常、超时都在这里兜住 logger.error("mqtt publish failed topic=%s err=%s", topic, exc) return False @staticmethod def _on_connect(client, userdata, flags, reason_code, properties=None): logger.info("mqtt connected rc=%s", reason_code) @staticmethod def _on_disconnect(client, userdata, *args): logger.warning("mqtt disconnected, will auto reconnect") publisher = DevicePublisher()

connect_async配合loop_start是这套封装的精髓：连接过程放在后台线程，主线程不会被 broker 不可达卡住。这一点很重要，因为语音请求走的是异步主循环，一旦在这里同步阻塞，整个服务都会变慢。发布失败时返回False而非抛异常，让业务层可以决定怎么回复用户。

技能侧的实现就很薄了，只负责识别意图、抽取参数、调用发布。注意它不关心 MQTT 细节，也不关心回复措辞之外的任何事：

# 文件路径：app/skills/device.py import logging from app.infra.mqtt_client import publisher from app.skills.base import Skill, SkillContext, SkillResult, register from app.skills.device_alias import find_device logger = logging.getLogger(__name__) ON_WORDS = ("打开", "开一下", "开启", "开") OFF_WORDS = ("关闭", "关掉", "关一下", "关") @register class DeviceSkill(Skill): name = "device" priority = 10 async def match(self, ctx: SkillContext) -> float: text = ctx.text has_action = any(w in text for w in ON_WORDS + OFF_WORDS) device_id = find_device(text) if not has_action: return 0.0 if device_id is None: # 有动作但没说是哪个设备，给个中间分，靠追问补槽位 return 0.4 return 0.95 if device_id else 0.4 async def handle(self, ctx: SkillContext) -> SkillResult: text = ctx.text device_id = ctx.slots.get("device_id") or find_device(text) if device_id is None: ctx.memory["pending_intent"] = "device.toggle" return SkillResult( reply="你要控制哪个设备？可以说客厅灯或者卧室空调。", need_slot="device_id", keep_session=True, ) action = "off" if any(w in text for w in OFF_WORDS) else "on" ok = publisher.publish_command(device_id, {"action": action}) if not ok: return SkillResult( reply="指令发出去了，但设备暂时没有响应，你可以稍后再试一次。", data={"device_id": device_id, "action": action, "ok": False}, ) action_cn = "打开" if action == "on" else "关闭" return SkillResult( reply=f"好的，已{action_cn}{device_id}。", data={"device_id": device_id, "action": action, "ok": True}, )

4.2 实现定时技能（异步任务与任务表）

定时看起来简单，实际是最容易出问题的技能。核心有三点：任务必须持有一个强引用，否则可能被垃圾回收导致“闹钟不响”；任务完成或异常后必须从任务表里移除，否则内存只增不减；服务重启后内存任务会全部丢失，这一点必须在使用说明里写清楚，不能假装它没有。

# 文件路径：app/skills/timer.py import asyncio import logging import re import time from typing import Dict from app.skills.base import Skill, SkillContext, SkillResult, register from app.skills.utils import cn_to_int logger = logging.getLogger(__name__) # 任务表：key 是 user_id，value 是 asyncio.Task _TASKS: Dict[str, asyncio.Task] = {} NUM_PATTERN = re.compile(r"([零一二两三四五六七八九十\d]+)\s*分钟") async def _fire_reminder(user_id: str, minutes: int, text: str) -> None: """到点执行的回调。真实项目里这里调用 TTS 主动播报。""" try: await asyncio.sleep(minutes * 60) logger.info("reminder fired user=%s minutes=%s text=%s", user_id, minutes, text) # 示例：这里接入播报通道，例如投递到 TTS 队列 # await tts_queue.put({"user_id": user_id, "text": f"{minutes}分钟到了"}) except asyncio.CancelledError: logger.info("reminder cancelled user=%s", user_id) raise finally: _TASKS.pop(user_id, None) @register class TimerSkill(Skill): name = "timer" priority = 20 async def match(self, ctx: SkillContext) -> float: if "闹钟" in ctx.text or "提醒" in ctx.text or "倒计时" in ctx.text: return 0.9 if NUM_PATTERN.search(ctx.text) and "分钟" in ctx.text: return 0.7 return 0.0 async def handle(self, ctx: SkillContext) -> SkillResult: match = NUM_PATTERN.search(ctx.text) if not match: return SkillResult( reply="请告诉我要多少分钟，比如定个二十五分钟的闹钟。", need_slot="minutes", keep_session=True, ) minutes = cn_to_int(match.group(1)) if minutes is None or minutes <= 0 or minutes > 24 * 60: return SkillResult(reply="这个时长我处理不了，请说一分钟到二十四小时之间的时间。") user_id = ctx.user_id old = _TASKS.pop(user_id, None) if old and not old.done(): old.cancel() task = asyncio.create_task(_fire_reminder(user_id, minutes, ctx.text)) _TASKS[user_id] = task logger.info("reminder registered user=%s minutes=%s", user_id, minutes) return SkillResult( reply=f"好的，{minutes}分钟后提醒你。", data={"minutes": minutes, "created_at": int(time.time())}, )

这里有个容易被忽略的细节：asyncio.create_task返回的任务如果没有被任何变量引用，事件循环只持弱引用，任务可能在中途被回收，表现为“闹钟有时候响有时候不响”，非常难复现。所以我们用模块级字典_TASKS强引用它。同时新任务注册前先取消同名旧任务，避免用户连说两次“定十分钟”结果收到两次提醒。

4.3 会话上下文与多轮追问

多轮的关键在于“把上一轮没说完的意图记住”。我们的做法是：当技能返回need_slot非空时，调度器把pending_intent、pending_skill一起写进会话存储；下一次同一用户发消息，先看会话里有没有待补槽位，有的话直接把文本交给原技能处理，跳过意图匹配。

# 文件路径：app/session.py import time from dataclasses import dataclass, field from typing import Any, Dict from app.config import settings @dataclass class Session: session_id: str user_id: str memory: Dict[str, Any] = field(default_factory=dict) updated_at: float = field(default_factory=time.time) class SessionStore: def __init__(self) -> None: self._data: Dict[str, Session] = {} def get(self, user_id: str) -> Session: sess = self._data.get(user_id) if sess and time.time() - sess.updated_at > settings.session_ttl: self._data.pop(user_id, None) sess = None if sess is None: sess = Session(session_id=f"s-{user_id}", user_id=user_id) self._data[user_id] = sess sess.updated_at = time.time() return sess def save(self, sess: Session) -> None: sess.updated_at = time.time() self._data[sess.user_id] = sess def clear(self, user_id: str) -> None: self._data.pop(user_id, None) store = SessionStore()

这是进程内实现，单机演示完全够用，但必须说清楚它的局限：多实例部署时会话不共享，用户第二次请求落到另一台机器上就“失忆”了。生产环境应换成 Redis，把Session序列化后存进去，接口保持不变即可平滑迁移。这段提示不是废话，很多人在扩容之后才发现多轮功能时好时坏。

4.4 调度器：把一切串起来

调度器只做四件事：加载会话、判断待补槽位、遍历技能打分、执行最高分技能。所有兜底和超时也在这里统一处理，技能内部不需要写 try/except，分层更干净。

# 文件路径：app/router.py import asyncio import logging from app.config import settings from app.session import store from app.skills.base import SkillContext, SkillResult, get_registry logger = logging.getLogger(__name__) FALLBACK_REPLY = "这个我还不太会，你可以试试让我开灯，或者定个闹钟。" async def dispatch(user_id: str, text: str) -> SkillResult: sess = store.get(user_id) ctx = SkillContext( session_id=sess.session_id, user_id=user_id, text=text, memory=sess.memory, ) registry = get_registry() if not registry: logger.error("技能注册表为空，检查 skills/__init__.py 是否导入了技能模块") return SkillResult(reply=FALLBACK_REPLY) # 1) 多轮优先：上一轮有未完成的技能，直接续上 pending = ctx.memory.get("pending_skill") if pending and pending in registry: return await _run_skill(pending, registry[pending], ctx, sess) # 2) 打分选优 scored = [] for name, skill in registry.items(): try: score = await skill.match(ctx) except Exception as exc: logger.warning("skill match error name=%s err=%s", name, exc) score = 0.0 if score > 0: scored.append((score, skill.priority, name, skill)) if not scored: return SkillResult(reply=FALLBACK_REPLY) scored.sort(key=lambda item: (item[0], item[1]), reverse=True) score, _, name, skill = scored[0] logger.info("dispatch user=%s text=%s skill=%s score=%.2f", user_id, text, name, score) if score < settings.intent_threshold: return SkillResult(reply=FALLBACK_REPLY) return await _run_skill(name, skill, ctx, sess) async def _run_skill(name: str, skill, ctx: SkillContext, sess) -> SkillResult: try: result = await asyncio.wait_for(skill.handle(ctx), timeout=settings.skill_timeout) except asyncio.TimeoutError: logger.error("skill timeout name=%s text=%s", name, ctx.text) return SkillResult(reply="处理超时了，请再说一次。") except Exception as exc: logger.exception("skill error name=%s err=%s", name, exc) return SkillResult(reply="刚刚出了点问题，请稍后再试。") if result.need_slot: ctx.memory["pending_skill"] = name ctx.memory["pending_slot"] = result.need_slot else: ctx.memory.pop("pending_skill", None) ctx.memory.pop("pending_slot", None) if result.keep_session: store.save(sess) else: ctx.memory.clear() store.clear(ctx.user_id) return result

asyncio.wait_for这层超时保护很重要。技能一旦出现阻塞（比如某次 MQTT 发布卡在网络上），没有超时就会拖住整个请求，而且这类问题在本地永远测不出来，只在网络抖动时偶发。统一加超时之后，最坏情况也只是用户听到一句“处理超时”，而不是整个服务雪崩。

4.5 启动与验证

最后把 Web 入口补上，请求体只留最小字段：用户标识和文本。真实接语音时，文本由识别环节提供，这一层不需要知道音频怎么来的。

# 文件路径：app/main.py import logging from contextlib import asynccontextmanager from fastapi import FastAPI from pydantic import BaseModel, Field from app.infra.mqtt_client import publisher from app.router import dispatch from app.skills import * # noqa: F401,F403 触发技能注册 logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s %(message)s", ) @asynccontextmanager async def lifespan(app: FastAPI): publisher.start() yield publisher.stop() app = FastAPI(title="Voice Assistant", lifespan=lifespan) class TextRequest(BaseModel): user_id: str = Field(..., min_length=1, max_length=64) text: str = Field(..., min_length=1, max_length=200) class TextResponse(BaseModel): reply: str data: dict = {} @app.post("/v1/voice/text", response_model=TextResponse) async def handle_text(req: TextRequest) -> TextResponse: result = await dispatch(req.user_id, req.text.strip()) return TextResponse(reply=result.reply, data=result.data)

skills/__init__.py里必须显式导入各个技能模块，这是注册生效的前提：

# 文件路径：app/skills/__init__.py from app.skills import device # noqa: F401 from app.skills import timer # noqa: F401

启动并验证：

uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload

# 正常控制设备 curl -X POST http://127.0.0.1:8000/v1/voice/text \ -H "Content-Type: application/json" \ -d '{"user_id":"u001","text":"打开客厅灯"}' # 预期：{"reply":"好的，已打开light.living_room。","data":{"device_id":"light.living_room","action":"on","ok":true}} # 缺设备参数，触发多轮追问 curl -X POST http://127.0.0.1:8000/v1/voice/text \ -H "Content-Type: application/json" \ -d '{"user_id":"u002","text":"把灯打开"}' # 预期：{"reply":"你要控制哪个设备？可以说客厅灯或者卧室空调。","data":{}} # 中文数字时长 curl -X POST http://127.0.0.1:8000/v1/voice/text \ -H "Content-Type: application/json" \ -d '{"user_id":"u003","text":"定个二十五分钟的闹钟"}' # 预期：{"reply":"好的，25分钟后提醒你。","data":{"minutes":25,"created_at":...}}

结果说明：第一条验证了规则打分和别名归一化，日志里能看到skill=device score=0.95；第二条验证了need_slot触发的上下文写入，紧接着再发一条“客厅灯”就能补全执行，这就是多轮的完整链路；第三条验证中文数字转换，注意回复里用的是阿拉伯数字，因为播报时阿拉伯数字更自然。

5. 接口层与语音链路的工程化处理

5.1 文本接口与语音接口的边界

我们刻意把文本接口和语音接口分成两层，原因是它们的稳定性特征完全不同。文本接口是纯计算，有明确输入输出，可以独立压测；语音接口涉及识别和合成两个外部依赖，随时可能返回空串、乱码或者超时。把两者混在一起，会让排查变得很痛苦，出了问题都不知道是识别错了还是意图路由错了。

推荐的调用方式是：语音接口只做“音频转文本 → 调用文本接口 → 文本转语音”的串联，中间不掺杂任何业务判断。这样每次出问题，都能先用文本接口单独复现，判断问题出在哪一段。这也是我们后来定位到“识别把‘灯’识别成‘等’”这个问题的关键，因为文本接口输入“打开客厅等”确实匹配不上。

5.2 语音识别与合成的降级策略

外部语音服务必须有降级。最常见的三种异常：返回空文本、返回超长文本、请求超时。处理方式分别是：空文本直接回“没听清，请再说一次”，不要进入路由；超长文本截断到合理长度再处理，避免下游技能被超长输入拖慢；请求超时要设置独立的短超时（比如三秒），超时后给固定的安抚回复，而不是让用户干等。

另外要有一个开关，能一键关掉语音链路只走文本。灰度期间这个开关非常有用，当发现某个版本语音服务不稳定时，可以立刻切到文本模式保住核心可用性，而不是紧急回滚整个服务。

5.3 并发与超时控制

语音助手的请求特点是“突发密集、单次极短”。用户可能一句话被切分成多次识别请求，也可能在短时间内连说几条指令。这种情况下，同步阻塞的代码会被放大成明显的延迟。所以整条链路都坚持异步：技能是async def，MQTT 发布走后台线程不阻塞，会话存取是内存操作很快。

需要提醒的是，asyncio里千万不要写time.sleep()，它会阻塞整个事件循环，导致所有并发请求一起变慢。这个错误非常常见，尤其是从同步代码迁移过来时顺手写了time.sleep(1)做节流。正确的写法是await asyncio.sleep(1)。同理，requests这类同步 HTTP 库在异步函数里也要避免，需要时用线程池包装或者换异步客户端。

6. 常见问题与排查思路

6.1 所有指令都返回兜底回复

这是上线后第一类高频问题，现象是所有输入都回“这个我还不太会”。最常见原因就是技能注册表为空：写了device.py和timer.py，但skills/__init__.py里没有导入，装饰器根本没执行。排查方法是启动时打印一下注册表内容，或者在main.py里加一行日志输出get_registry().keys()。

第二个原因通常是阈值设得太高。如果你把INTENT_MATCH_THRESHOLD调到了 0.9，那么所有中间分（比如缺设备时的 0.4）都会被过滤掉。第三个原因是文本里带了多余的字符，比如语音识别结果带标点，“打开客厅灯。”这句里打开能命中，一般不会出问题，但如果识别成“打 开 客 厅 灯”这种带空格的，关键词就匹配不上了。建议入口统一做一次去空格和标点归一化。

6.2 多轮追问不生效

现象是用户说“把灯打开”，系统问“你要控制哪个设备”，用户回“客厅灯”，系统又回到兜底回复。这时要检查三件事：会话是否被写入了pending_skill、下次请求的user_id是否一致、会话是否已过期。如果用户标识每次请求都变（比如用设备号加时间戳拼的），会话自然对不上，这个坑在联调时特别常见。

还有一个隐蔽原因是keep_session忘记设为True。看第 4.4 节的调度器代码，keep_session为False时会走store.clear()，把刚写进去的待补槽位信息全清了。所以凡是返回need_slot的技能，必须同时设keep_session=True，这两个字段是配套使用的。

6.3 定时提醒偶发不触发

前面已经提到任务弱引用的问题，这里再补充两种可能。一是服务被重启了，内存任务全部丢失，这是设计上的限制，需要在产品层面说明“重启后需要重新设置”，或者把任务持久化到数据库并在启动时重放。二是asyncio.sleep的时间单位写错，把分钟当成秒，或者反过来乘以 1000，这类笔误在本地短时间测试时看不出来，因为都在等。

排查建议是给任务加了日志之后，先看“registered”和“fired”两条日志的时间差是否等于预期。如果只有 registered 没有 fired，且服务没重启，基本就是任务被取消或者被回收了。

6.4 常见问题速查表

7. 最佳实践与工程建议

7.1 日志要能回答三个问题

日志的设计标准很朴素：出问题时能不能从日志里直接看出“命中了哪个技能、抽到了什么槽位、耗时多少”。所以每条请求至少要打三行：入参（用户、归一化后的文本）、路由结果（技能名、分数）、执行结果（成功与否、耗时）。不要打全量音频或完整请求体，涉及用户隐私的内容要脱敏。

另外建议给每次请求生成一个trace_id，在调度器和技能之间传递，这样一条请求的所有日志能串起来。如果项目已经接入链路追踪系统，把trace_id塞进上下文即可，改动量很小。等并发量上来之后，没有trace_id的日志基本等于没有日志。

7.2 配置与密钥的管理边界

MQTT_USERNAME、MQTT_PASSWORD这类凭证绝对不能进代码仓库，.env要加进.gitignore。生产环境推荐用容器编排平台提供的密钥管理能力注入环境变量，而不是把.env文件复制到服务器上。同时注意不要把这些值打进日志，我们上面_on_connect的回调里只打了reason_code，就是为了避免误打认证信息。

阈值类配置则可以大胆外置，鼓励调参。意图匹配的分数线本来就该根据真实语料不断优化，把它做成环境变量，让调参不需要改代码、不需要发版，是性价比很高的工程习惯。

7.3 技能开发的约定俗成

为了让技能保持可维护，我们定了几条约定，团队内一直沿用。第一，技能必须有唯一name，命名用“领域.动作”的小写形式，比如device、timer，将来扩展可以叫weather、music。第二，技能不写try/except，异常统一在调度器兜住，避免每个技能重复写一遍错误处理。第三，技能不直接操作数据库和网络客户端，一律通过infra层封装调用，方便替换和 mock。

第四，也是最重要的一条：技能不持有请求状态。所有请求相关数据放在SkillContext，所有跨请求状态放在会话存储。因为技能实例是单例，一旦把用户数据写到self上，就会出现 A 用户的参数串到 B 用户的请求里，这种 bug 极其危险且难以复现。

7.4 安全边界与权限控制

设备控制类功能必须做权限校验，否则任何人都能通过接口控制别人的设备。最小实现是：请求里的user_id经过认证层校验后才能进入路由，设备 ID 与用户的绑定关系存在数据库里，发布指令前先查一次“这个用户是否有权控制这个设备”。本文示例为了聚焦主线省略了这一步，但在实际项目中它是不可省略的。

另外要牢记：涉及设备开关、定时任务这类会改变物理世界状态的操做，一定要在测试环境充分验证，并对高风险操作（比如批量关闭、总闸控制）加二次确认。生产环境的设备控制指令建议加操作审计日志，记录谁在什么时候对什么设备做了什么，便于事后追溯。

7.5 性能优化的优先顺序

优化要有顺序，不要一上来就折腾。第一优先级是消除阻塞，把异步函数里的同步调用找出来，收益最大。第二优先级是给外部依赖加超时和熔断，避免一个慢依赖拖垮整个服务。第三才是考虑缓存，比如把设备别名表、用户权限关系缓存到本地内存并定期刷新，减少数据库查询。

至于并发量本身，语音助手的 QPS 通常不高，单机异步足够支撑相当规模。真正需要关注的是尾延迟，也就是最慢那 1% 的请求，因为语音交互对慢的感知极其明显。建议在日志里记录耗时分布，重点关注 P99，而不是平均耗时。

8. 后续可以继续深入的方向

到这里，这套自研语音助手已经从“能对话”走到了“能干活”。回顾一下这条链路上真正起作用的几个设计：技能用装饰器注册，新功能只增不改；调度器用打分代替顺序匹配，把冲突显式化；need_slot加keep_session支撑多轮；统一超时和异常兜底守住可用性。这几条加起来，才让一个 Demo 具备了持续迭代的基础。

接下来有几个可以继续推进的方向，按投入产出排序。第一是把会话存储换成 Redis，为多实例部署做准备，接口不变，改动集中在一个类里。第二是给定时任务加持久化，服务重启后从数据库重放未完成的任务，彻底解决“重启丢闹钟”。第三是在规则匹配全部低于阈值时接入模型兜底，只让模型处理闲聊和模糊意图，关键动作仍走确定性路径。

接口层还可以补两件事：一是给设备控制加幂等键，避免用户重复说同一句话导致指令重复下发；二是把trace_id贯穿到 MQTT 消息体里，让设备侧的执行日志也能和服务端日志对齐，排障时能一眼看到整条链路。这两件事改动都很小，但对线上问题的定位效率提升很明显。

最后想说，语音类项目最容易被低估的部分其实是“边界”。用户不会按你设计的语法说话，会有口音、会有停顿、会说到一半改主意。所以规则词表要持续从真实日志里补充，兜底回复要足够友好，异常处理要比你预想的多写一层。把这几件事做扎实，系统才算真的能用。