手把手：用讯飞平台API和Python打造智能对话机器人

# 使用讯飞平台API制作Python对话机器人 ## 准备工作 1. 注册讯飞开放平台账号 2. 在控制台创建应用，获取`APPID`、`APIKey`和`APISecret` 3. 开通"语音听写"和"语音合成"服务 ## 安装必要库 ```bash pip install websocket-client requests pydub ``` ## 步骤1: 语音识别(ASR) ```python import websocket import json import time import _thread as thread from pydub import AudioSegment from pydub.playback import play class ASRClient: def __init__(self, appid, api_key, api_secret): self.appid = appid self.api_key = api_key self.api_secret = api_secret self.base_url = "wss://iat-api.xfyun.cn/v2/iat" self.text = "" def on_message(self, ws, message): result = json.loads(message) if result["code"] != 0: print(f"识别错误: {result}") return data = result["data"]["result"]["ws"] for word in data: self.text += word["cw"][0]["w"] if result["data"]["status"] == 2: print(f"识别结果: {self.text}") self.text = "" def on_error(self, ws, error): print(f"WebSocket错误: {error}") def on_close(self, ws): print("WebSocket连接关闭") def on_open(self, ws): def run(*args): # 读取音频文件并发送 audio = AudioSegment.from_wav("input.wav") raw_data = audio.raw_data frame_size = 1280 # 每帧大小 for i in range(0, len(raw_data), frame_size): ws.send(raw_data[i:i+frame_size]) ws.send('{"end": true}') thread.start_new_thread(run, ()) def recognize(self, audio_file): websocket.enableTrace(False) ws = websocket.WebSocketApp( self.base_url, on_message=self.on_message, on_error=self.on_error, on_close=self.on_close ) ws.on_open = self.on_open ws.run_forever() ``` ## 步骤2: 自然语言处理(NLP) ```python import requests import json import hashlib import base64 import time class NLUClient: def __init__(self, appid, api_key): self.appid = appid self.api_key = api_key self.base_url = "https://api.xfyun.cn/v1/aiui/v1/text_semantic" def get_response(self, text): cur_time = str(int(time.time())) param = { "scene": "main", "auth_id": "your_auth_id", "data_type": "text", "text": text } param_base64 = base64.b64encode(json.dumps(param).encode('utf-8')).decode('utf-8') m2 = hashlib.md5() m2.update((self.api_key + cur_time + param_base64).encode('utf-8')) check_sum = m2.hexdigest() headers = { 'X-Appid': self.appid, 'X-CurTime': cur_time, 'X-Param': param_base64, 'X-CheckSum': check_sum, 'Content-Type': 'application/x-www-form-urlencoded; charset=utf-8' } response = requests.post(self.base_url, headers=headers) return response.json() ``` ## 步骤3: 语音合成(TTS) ```python import requests import json import hashlib import base64 import time from pydub import AudioSegment from pydub.playback import play class TTSClient: def __init__(self, appid, api_key): self.appid = appid self.api_key = api_key self.base_url = "https://tts-api.xfyun.cn/v2/tts" def synthesize(self, text): cur_time = str(int(time.time())) param = { "aue": "raw", "auf": "audio/L16;rate=16000", "voice_name": "xiaoyan", "speed": "50", "volume": "50", "pitch": "50", "engine_type": "aisound" } param_base64 = base64.b64encode(json.dumps(param).encode('utf-8')).decode('utf-8') m2 = hashlib.md5() m2.update((self.api_key + cur_time + param_base64).encode('utf-8')) check_sum = m2.hexdigest() headers = { 'X-Appid': self.appid, 'X-CurTime': cur_time, 'X-Param': param_base64, 'X-CheckSum': check_sum, 'Content-Type': 'application/x-www-form-urlencoded; charset=utf-8' } data = {'text': text} response = requests.post(self.base_url, headers=headers, data=data) if response.headers['Content-Type'] == 'audio/mpeg': with open('output.wav', 'wb') as f: f.write(response.content) audio = AudioSegment.from_wav("output.wav") play(audio) else: print(f"合成失败: {response.text}") ``` ## 步骤4: 整合对话机器人 ```python class ChatBot: def __init__(self, appid, api_key, api_secret): self.asr = ASRClient(appid, api_key, api_secret) self.nlu = NLUClient(appid, api_key) self.tts = TTSClient(appid, api_key) def run(self): print("对话机器人已启动，请说话...") while True: # 1. 语音识别 self.asr.recognize("input.wav") # 2. 自然语言理解 nlu_result = self.nlu.get_response(self.asr.text) # 3. 生成回复 reply = self.generate_reply(nlu_result) # 4. 语音合成 self.tts.synthesize(reply) def generate_reply(self, nlu_result): # 这里可以根据nlu_result的内容生成不同的回复 if "answer" in nlu_result: return nlu_result["answer"]["text"] else: return "我没有理解你的意思，请再说一遍" ``` ## 步骤5: 使用机器人 ```python if __name__ == "__main__": # 替换为你的实际API信息 APPID = "your_appid" API_KEY = "your_api_key" API_SECRET = "your_api_secret" bot = ChatBot(APPID, API_KEY, API_SECRET) bot.run() ``` ## 额外注意事项 1. 讯飞API有调用频率限制，生产环境中需要考虑限流和错误处理 2. 音频文件格式需要符合API要求，可能需要格式转换 3. 实际应用中需要添加异常处理机制，确保网络波动时程序的稳定性 ######[AI写代码神器 | 1640点数解答 | 2025-06-10 11:29:09]