在做语音助手的项目时最容易出现的落差是演示视频里它在“聊天”环节表现很好真正接到需求里就干不了活。我们这个自研智能音箱项目内部戏称为“天猫精灵”上篇已经跑通了“唤醒词—语音转文字—回复播报”的最小闭环但那只是一条能说话的链路。下集要解决的是另一件事让同一个入口能够稳定地把“开灯”“定个 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 -v2.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_HOST127.0.0.1 MQTT_PORT1883 MQTT_USERNAME MQTT_PASSWORD MQTT_CLIENT_IDvoice-server-01 # 业务阈值 INTENT_MATCH_THRESHOLD0.35 SKILL_TIMEOUT_SECONDS5 SESSION_TTL_SECONDS180# 文件路径app/config.py import os from dataclasses import dataclass from dotenv import load_dotenv load_dotenv() dataclass(frozenTrue) 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()这里用了frozenTrue的 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_factorydict) memory: Dict[str, Any] field(default_factorydict) dataclass class SkillResult: 技能执行结果。 reply: str data: Dict[str, Any] field(default_factorydict) 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_idsettings.mqtt_client_id, ) except AttributeError: self._client mqtt.Client(client_idsettings.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_delay1, max_delay30) def start(self) - None: self._client.connect_async(settings.mqtt_host, settings.mqtt_port, keepalive60) 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 fhome/device/{device_id}/cmd payload json.dumps(command, ensure_asciiFalse) try: info self._client.publish(topic, payload, qos1, timeout3) info.wait_for_publish(timeout3) 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, propertiesNone): 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_slotdevice_id, keep_sessionTrue, ) 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( replyf好的已{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_idvalue 是 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_slotminutes, keep_sessionTrue, ) 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( replyf好的{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_factorydict) updated_at: float field(default_factorytime.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_idfs-{user_id}, user_iduser_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_idsess.session_id, user_iduser_id, texttext, memorysess.memory, ) registry get_registry() if not registry: logger.error(技能注册表为空检查 skills/__init__.py 是否导入了技能模块) return SkillResult(replyFALLBACK_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(replyFALLBACK_REPLY) scored.sort(keylambda item: (item[0], item[1]), reverseTrue) 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(replyFALLBACK_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), timeoutsettings.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 resultasyncio.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( levellogging.INFO, format%(asctime)s %(levelname)s %(name)s %(message)s, ) asynccontextmanager async def lifespan(app: FastAPI): publisher.start() yield publisher.stop() app FastAPI(titleVoice Assistant, lifespanlifespan) class TextRequest(BaseModel): user_id: str Field(..., min_length1, max_length64) text: str Field(..., min_length1, max_length200) class TextResponse(BaseModel): reply: str data: dict {} app.post(/v1/voice/text, response_modelTextResponse) async def handle_text(req: TextRequest) - TextResponse: result await dispatch(req.user_id, req.text.strip()) return TextResponse(replyresult.reply, dataresult.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:...}}结果说明第一条验证了规则打分和别名归一化日志里能看到skilldevice score0.95第二条验证了need_slot触发的上下文写入紧接着再发一条“客厅灯”就能补全执行这就是多轮的完整链路第三条验证中文数字转换注意回复里用的是阿拉伯数字因为播报时阿拉伯数字更自然。5. 接口层与语音链路的工程化处理5.1 文本接口与语音接口的边界我们刻意把文本接口和语音接口分成两层原因是它们的稳定性特征完全不同。文本接口是纯计算有明确输入输出可以独立压测语音接口涉及识别和合成两个外部依赖随时可能返回空串、乱码或者超时。把两者混在一起会让排查变得很痛苦出了问题都不知道是识别错了还是意图路由错了。推荐的调用方式是语音接口只做“音频转文本 → 调用文本接口 → 文本转语音”的串联中间不掺杂任何业务判断。这样每次出问题都能先用文本接口单独复现判断问题出在哪一段。这也是我们后来定位到“识别把‘灯’识别成‘等’”这个问题的关键因为文本接口输入“打开客厅等”确实匹配不上。5.2 语音识别与合成的降级策略外部语音服务必须有降级。最常见的三种异常返回空文本、返回超长文本、请求超时。处理方式分别是空文本直接回“没听清请再说一次”不要进入路由超长文本截断到合理长度再处理避免下游技能被超长输入拖慢请求超时要设置独立的短超时比如三秒超时后给固定的安抚回复而不是让用户干等。另外要有一个开关能一键关掉语音链路只走文本。灰度期间这个开关非常有用当发现某个版本语音服务不稳定时可以立刻切到文本模式保住核心可用性而不是紧急回滚整个服务。5.3 并发与超时控制语音助手的请求特点是“突发密集、单次极短”。用户可能一句话被切分成多次识别请求也可能在短时间内连说几条指令。这种情况下同步阻塞的代码会被放大成明显的延迟。所以整条链路都坚持异步技能是async defMQTT 发布走后台线程不阻塞会话存取是内存操作很快。需要提醒的是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_sessionTrue这两个字段是配套使用的。6.3 定时提醒偶发不触发前面已经提到任务弱引用的问题这里再补充两种可能。一是服务被重启了内存任务全部丢失这是设计上的限制需要在产品层面说明“重启后需要重新设置”或者把任务持久化到数据库并在启动时重放。二是asyncio.sleep的时间单位写错把分钟当成秒或者反过来乘以 1000这类笔误在本地短时间测试时看不出来因为都在等。排查建议是给任务加了日志之后先看“registered”和“fired”两条日志的时间差是否等于预期。如果只有 registered 没有 fired且服务没重启基本就是任务被取消或者被回收了。6.4 常见问题速查表问题现象常见原因解决思路全部走兜底回复技能未导入、阈值过高、文本带空格打印注册表、下调阈值、入口做归一化多轮追问断掉keep_session未设、user_id变化、会话过期检查字段配套、统一用户标识、调大 TTL设备控制无响应broker 不可达、topic 拼错、QoS 配置不当看 publish 日志、核对主题、QoS 设为 1定时不触发任务被回收、服务重启、单位写错强引用任务、持久化任务、核对时间单位启动报TypeErrorpaho-mqtt 2.x 版本差异显式传CallbackAPIVersion请求整体变慢异步函数里写了同步阻塞调用改await asyncio.sleep、避免同步 HTTP中文时长抽不到只写了数字正则增加中文数字转换函数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 消息体里让设备侧的执行日志也能和服务端日志对齐排障时能一眼看到整条链路。这两件事改动都很小但对线上问题的定位效率提升很明显。最后想说语音类项目最容易被低估的部分其实是“边界”。用户不会按你设计的语法说话会有口音、会有停顿、会说到一半改主意。所以规则词表要持续从真实日志里补充兜底回复要足够友好异常处理要比你预想的多写一层。把这几件事做扎实系统才算真的能用。