基于阿里云ASR的AI电销机器人架构与实现

对话状态机实现

class DialogSM:
    STATES = ['GREETING', 'PRODUCT_INTRO', 'OBJECTION_HANDLING', 'CLOSING']

    def __init__(self):
        self.state = 'GREETING'
        self.context = {}

    def transit(self, intent):
        """基于意图的状态转移"""
        prev = self.state
        if self.state == 'GREETING':
            if intent == 'POSITIVE':
                self.state = 'PRODUCT_INTRO'
            else:
                self.state = 'OBJECTION_HANDLING'
        # ...其他状态转移逻辑
        logging.info(f"State changed: {prev} -> {self.state}")
        return self.get_response()

    def get_response(self):
        """获取当前状态对应话术"""
        responses = {
            'GREETING': "您好，请问是{name}先生吗？",
            'PRODUCT_INTRO': "我们最新推出的产品有三个核心优势...",
            # ...其他状态话术
        }
        return responses[self.state].format(**self.context)

避坑指南

ASR配额不足降级方案

1. 动态采样率切换：
   • 当剩余配额<20%时，自动切换至8kHz采样率
   • 识别模式从实时流改为分片识别（每2秒发送一次）

本地兜底模型：

def fallback_asr(audio):
    if USE_LOCAL_MODEL:
        return local_model.transcribe(audio)
    else:
        raise ASRQuotaExceeded()

中断重试幂等设计

1. 语音包序列号：每个数据包附加递增seq_id
2. 服务端去重：ASR服务记录最近5秒处理的seq_id
3. 客户端补偿：超时未响应时重发相同seq_id的包

敏感词过滤实现

import ahocorasick

class KeywordFilter:
    def __init__(self):
        self.automaton = ahocorasick.Automaton()
        for word in load_sensitive_words():
            self.automaton.add_word(word.lower(), word)
        self.automaton.make_automaton()

    def check(self, text):
        hits = []
        for end_idx, word in self.automaton.iter(text.lower()):
            hits.append((end_idx - len(word) + 1, end_idx, word))
        return hits

性能优化对比

测试环境：4核8G云主机，100并发请求

优化建议：

1. 常规流量使用异步IO模式
2. 高峰期切换至批处理模式（每10条请求打包发送）
3. 设置单实例最大并发不超过80%

安全实施方案

密钥轮换策略

1. 双密钥热切换：
   • 系统同时保存新旧两套密钥
   • 通过API网关的Header参数指定使用版本
   • 旧密钥保留7天后自动失效

自动更新流程：

def rotate_key():
    new_key = generate_key()
    update_key_in_vault(new_key)
    # 先验证新密钥
    test_client = AcsClient(new_key, new_secret)
    if test_client.check_valid():
        switch_traffic_to_new_key()
        deactivate_old_key_after(7days)

语音数据加密

1. 传输层：强制使用TLS1.3
2. 访问控制：基于STS的临时访问令牌

存储加密：

def encrypt_audio(audio):
    iv = os.urandom(16)
    cipher = AES.new(STORAGE_KEY, AES.MODE_CFB, iv)
    return iv + cipher.encrypt(audio)

开放问题讨论

如何设计ASR结果的置信度兜底机制？以下是几个思考方向：

1. 多模型投票：同时调用2-3个ASR服务，取置信度最高的结果
2. 上下文校验：用历史对话内容验证当前识别结果的合理性
3. 人工确认：当置信度<80%时触发二次确认流程