基于阿里云ASR的AI电销机器人源码解析与部署指南

2. 阿里云SDK安全封装

推荐使用环境变量存储敏感信息：

# utils/config.py
import os
from dotenv import load_dotenv

load_dotenv()

class ASRConfig:
    APP_KEY = os.getenv('ALI_ASR_KEY')  # 通过RAM子账号获取
    APP_SECRET = os.getenv('ALI_ASR_SECRET')  # 定期轮换
    SAMPLE_RATE = 16000  # 电话音频常用采样率
    FORMAT = 'pcm'  # 原始音频格式

3. 双向音频流处理

# services/audio_stream.py
import websockets
from typing import AsyncGenerator
import json

async def stream_audio(call_id: str) -> AsyncGenerator[str, None]:
    """处理双向音频流"""
    async with websockets.connect(WS_ENDPOINT) as ws:
        # 发送启动参数
        await ws.send(json.dumps({
            "header": {
                "message_id": call_id,
                "namespace": "SpeechRecognizer",
                "name": "StartRecognition",
                "format": "pcm",
                "sample_rate": 16000
            }
        }))
        # 实时接收识别结果
        while True:
            resp = await ws.recv()
            data = json.loads(resp)
            if 'payload' in data:
                text = data['payload']['result']
                yield text  # 返回识别文本

四、生产环境部署指南

1. 阿里云RAM配置

最小权限策略示例：

{
  "Version": "1",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "nls-filetrans:*",
        "nls-cloud-meta:*"
      ],
      "Resource": "*"
    }
  ]
}

2. Nginx优化配置

# /etc/nginx/conf.d/asr.conf
upstream asr_app {
    server 127.0.0.1:8000;
    keepalive 32;  # 保持长连接
}

server {
    listen 443 ssl;
    server_name yourdomain.com;
    ssl_certificate /path/to/cert.pem;
    ssl_certificate_key /path/to/key.pem;

    location / {
        proxy_pass http://asr_app;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
    }
}

Gunicorn启动参数建议：

gunicorn -w 4 -k gevent -b 0.0.0.0:8000 app:app
# -w 根据CPU核心数设置 (2-4倍)
# -k 使用gevent协程提高并发

五、关键问题解决方案

1. Token自动刷新机制

# services/auth.py
from datetime import datetime, timedelta

class TokenManager:
    def __init__(self):
        self._token = None
        self._expire_time = None

    @property
    def token(self) -> str:
        if not self._token or datetime.now() > self._expire_time:
            self._refresh_token()
        return self._token

    def _refresh_token(self):
        # 调用阿里云Token接口
        resp = requests.post('https://nls-meta.cn-shanghai.aliyuncs.com', ...)
        self._token = resp.json()['Token']['Id']
        self._expire_time = datetime.now() + timedelta(hours=1)

2. ASR结果后处理

# utils/text_cleaner.py
import re

def clean_asr_result(text: str) -> str:
    """清理识别结果中的无效内容"""
    # 去除语气词
    text = re.sub(r'嗯 | 啊 | 呃 | 这个 | 那个', '', text)
    # 合并重复标点
    text = re.sub(r'([，。！？])\1+', r'\1', text)
    return text.strip()

六、安全合规实践

1. 录音文件加密

# services/storage.py
from cryptography.fernet import Fernet
import os

class AudioStorage:
    def __init__(self):
        self.cipher = Fernet(os.getenv('AES_KEY'))

    def save(self, call_id: str, audio_data: bytes):
        encrypted = self.cipher.encrypt(audio_data)
        with open(f'/data/{call_id}.enc', 'wb') as f:
            f.write(encrypted)

2. GDPR数据删除

# api/gdpr.py
from flask import jsonify

@app.route('/gdpr/delete', methods=['DELETE'])
def delete_user_data():
    user_id = request.json['user_id']
    # 伪代码示例
    delete_recordings(user_id)
    delete_call_logs(user_id)
    return jsonify({'status': 'completed'})

七、后续优化方向

当前方案已实现基础语音交互，下一步可考虑：

1. 意图识别升级：如何结合NLP模型识别客户购买意向？
   • 方案一：在ASR文本后接入BERT分类模型
   • 方案二：实时分析语音情感特征（音高/语速变化）
2. 智能打断机制：当客户长时间沉默时，如何自动触发话术引导？
3. 多轮对话管理：设计状态机处理复杂的业务咨询流程