如何高效编写可维护的skill：从架构设计到代码实现

7次阅读

没有评论

共计 2049 个字符，预计需要花费 6 分钟才能阅读完成。

在开发语音交互 skill 时，许多开发者会遇到以下典型问题：

代码结构混乱：业务逻辑、API 调用、数据处理全部混在同一个文件中
扩展困难：新增功能时需要修改大量已有代码，牵一发而动全身
错误处理缺失：异常情况直接导致服务崩溃，缺乏优雅降级机制
测试覆盖率低：难以编写单元测试，回归测试成本高
性能瓶颈：同步阻塞式处理导致响应延迟

这些问题会随着业务复杂度的提升而不断放大，最终形成难以维护的 ” 祖传代码 ”。

采用经典的三层架构，确保各层职责单一：

表现层：处理语音平台协议适配
业务逻辑层：核心业务实现
数据访问层：封装外部服务调用

通过抽象接口实现松耦合：

定义清晰的功能接口契约
通过依赖注入 (DI) 解耦具体实现
使用工厂模式管理实例创建

定义业务异常体系
实现全局异常拦截器
设计优雅降级方案

project/
│── skills/
│   ├── __init__.py
│   ├── weather_skill.py    # 业务逻辑实现
│   └── interfaces.py       # 抽象接口定义
│── adapters/
│   ├── alexa_adapter.py    # Alexa 平台适配
│   └── dialogflow_adapter.py
│── services/
│   ├── weather_service.py  # 外部 API 封装
│   └── cache_service.py
│── main.py                # 依赖组装入口

# interfaces.py
from abc import ABC, abstractmethod

class ISkillHandler(ABC):
    @abstractmethod
    def can_handle(self, intent: str) -> bool:
        pass

    @abstractmethod
    def handle(self, request: dict) -> dict:
        pass

# weather_skill.py
from typing import Optional
from .interfaces import ISkillHandler
from ..services.weather_service import WeatherService

class WeatherSkill(ISkillHandler):
    def __init__(self, weather_service: WeatherService):
        self._weather_service = weather_service

    def can_handle(self, intent: str) -> bool:
        return intent == "weather_query"

    def handle(self, request: dict) -> dict:
        try:
            city = request["city"]
            weather = self._weather_service.get_weather(city)
            return {"text": f"{city}天气是{weather}",
                "should_end_session": True
            }
        except ServiceUnavailableError:
            # 优雅降级处理
            return {"text": "天气服务暂不可用"}

使用 async/await 避免 IO 阻塞：

async def handle(self, request: dict) -> dict:
    city = request["city"]
    weather = await self._weather_service.get_weather_async(city)
    return {"text": f"{city}天气是{weather}"}