FreeSWITCH 簡單圖形化界面43 - 使用百度的unimrcp搞個智能話務臺，用的在線的ASR和TTS

FreeSWITCH 簡單圖形化界面43 - 使用百度的unimrcp搞個智能話務臺

0、一個fs的web配置界面預覽
1、安裝unimrcp模塊
2、安裝完成后，配置FreeSWITCH。
- 2.1 有界面的配置
- - 2.1.1 mod_unimrcp模塊配置
  - 2.1.2 mod_unimrcp客戶端配置
- 2.2 無界面的配置
3、呼叫規則
4、編寫流程
- 4.1 python代碼
- 4.2 主要是detect_speech和speak
5、效果
6、一些參考的官網網址

0、一個fs的web配置界面預覽

http://myfs.f3322.net:8020/
用戶名：admin，密碼：admin

FreeSWITCH界面安裝參考：https://blog.csdn.net/jia198810/article/details/137820796

1、安裝unimrcp模塊

參考鏈接：
1、https://docs.qq.com/pdf/DVFFub1JpSmNMRXdJ] 百度mrcp章節
2、https://blog.csdn.net/jia198810/article/details/118461176?spm=1011.2415.3001.5331

安裝完成后，配置unimrcp服務器的端口為6060。

如果這里模塊沒有安裝成功，就不用繼續了哈，請劃過。

2、安裝完成后，配置FreeSWITCH。

2.1 有界面的配置

2.1.1 mod_unimrcp模塊配置

在這里插入圖片描述

2.1.2 mod_unimrcp客戶端配置

在這里插入圖片描述

2.2 無界面的配置

如果使用的是沒有界面的FreeSWITCH，配置如下：

<configuration name="unimrcp.conf" description="UniMRCP Client"><settings><!--  UniMRCP profile to use for TTS  --><param name="default-tts-profile" value="baidu"/><!--  UniMRCP profile to use for ASR  --><param name="default-asr-profile" value="baidu"/><!--  UniMRCP logging level to appear in freeswitch.log.  Options are:EMERGENCY|ALERT|CRITICAL|ERROR|WARNING|NOTICE|INFO|DEBUG  --><param name="log-level" value="info"/><!--  Enable events for profile creation, open, and close  --><param name="enable-profile-events" value="true"/><param name="max-connection-count" value="100"/><param name="offer-new-connection" value="1"/><param name="request-timeout" value="3000"/></settings>
<profiles>
<profile name="baidu" version="2"><param name="server-ip" value="127.0.0.1"/><!--  Mrcp Server 端口號  --><param name="server-port" value="6060"/><param name="resource-location" value=""/><param name="client-ip" value="127.0.0.1"/><param name="client-port" value="6061"/><param name="sip-transport" value="udp"/><param name="rtp-ip" value="127.0.0.1"/><param name="rtp-port-min" value="30000"/><param name="rtp-port-max" value="30500"/><param name="speechsynth" value="speechsynthesizer"/><param name="speechrecog" value="speechrecognizer"/><param name="codecs" value="PCMU PCMA L16/96/8000"/><!--  Add any default MRCP params for SPEAK requests here  --><synthparams> </synthparams><!--  Add any default MRCP params for RECOGNIZE requests here  --><recogparams><!-- param name="start-input-timers" value="false"/ --></recogparams></profile>
</profiles>
</configuration>

3、呼叫規則

<extension name="測試:11120"><condition expression="^(11120)$" field="destination_number"><action application="python" data="baidu-mrcpserver"/><action application="hangup"/></condition>
</extension>

如下圖：
在這里插入圖片描述

4、編寫流程

編寫baidu-mrcpserver.py代碼，放在/usr/local/freeswitch/scripts/下面，如下代碼用的python，lua腳本換下語法即可，Freeswitch的lua API和python的API是通用的，名稱都一樣。

4.1 python代碼

#!/usr/local/python3/bin/python3
from freeswitch import *
"""
baidu-mrcpserver.py
撥號規則:
<extension name="測試:11120"><condition expression="^(11120)$" field="destination_number"><action application="python" data="baidu-mrcpserver"/><action application="hangup"/></condition>
</extension># 識別用戶輸入后，交由第三方處理，并讀出結果。
"""
from xml.dom import minidom
from openai import OpenAI
import redeepseek_api_key = "i am a key "# 解析百度的asr結果,XML格式的
def parse_xml(xml_string):text = ""try:dom = minidom.parseString(xml_string)input_element = dom.getElementsByTagName('input')[0]text = input_element.firstChild.nodeValueexcept Exception as e:consoleLog("error", f"解析xml出錯:{e}")return text# 說
def speak(session, text):if text:session.consoleLog("info", f"正在說：{text}")session.execute("speak", f"unimrcp:baidu||{text}")# 開始聽
def start_listen(session):# timeout為語音識別時間,單位為毫秒# session.consoleLog("info", f"開始語音識別")# 設置識別超時時間timeout = 10000session.execute("detect_speech", f"param speech-timeout {timeout}")session.execute("detect_speech", "unimrcp baidu builtin:grammar/baidu")# 暫停聽
def pause_listen(session):session.consoleLog("info", "暫停語音識別")session.execute("detect_speech", "pause")# 結束聽
def stop_listen(session):session.consoleLog("info", "結束語音識別")session.execute("detect_speech", "stop")# 恢復聽
def resume_listen(session):session.consoleLog("info", "恢復語音識別")# resume 不好使,用detect_speech代替# session:execute("detect_speech", "resume");start_listen(session)# 連說帶聽,不會觸發回調
def speak_and_listen(session, text, time=5000):session.execute("set","tts_engine=unimrcp")session.execute("play_and_detect_speech",f"say:{text} detect:unimrcp:baidu builtin:grammar/baidu")session.execute("sleep", f"{time}")# 第三方處理
# 這里可以編寫對于語音識別的處理,這里是調用deepseek的python接口處理語音識別結果
def process_result(session, result):if "掛機" in result or "退出" in result or "結束" in result or "再見" in result or "結束通話" in result or "掛斷" in result or "拜拜" in result:speak(session, "再見")session.hangup()return "break"if session.ready():speak(session, f"正在查詢{result},請稍等")try:sentence_delimiters = r'([，。！？,\.!\?\n])'  # [1,3,7](@ref)buffer = ""current_sentence = ""client = OpenAI(api_key=deepseek_api_key, base_url="https://api.deepseek.com")response = client.chat.completions.create(model="deepseek-chat",messages=[{"role": "system", "content": "生活助手"},{"role": "user", "content": f"{result},不要超過100字回復,簡潔回答,不要出現表情、星號、井號等特殊符號"},],stream=True  # 設置 stream 參數為 True)for chunk in response:if chunk.choices and chunk.choices[0].delta.content:content = chunk.choices[0].delta.contentbuffer += content# 實時檢測分隔符while True:match = re.search(sentence_delimiters, buffer)if not match:break# 分割出完整句子split_pos = match.end()sentence_part = buffer[:split_pos]buffer = buffer[split_pos:]# 將句子片段添加到當前句子current_sentence += sentence_part# 判斷是否遇到句子結束符if match.group() in ['。', '.', '！', '!', '？', '?', '\n']:  # [3,7](@ref)# 分段說,避免句子太長,tts生成的太慢speak(session,current_sentence.strip())current_sentence = ""# 輸出緩沖區剩余內容if buffer.strip():speak(session,current_sentence.strip())except Exception as e:consoleLog("info", f"處理結果出錯:{e}")return "break"# 只有在detect_speech成功后才會調用,play_and_detect_speech不會調用
# 最大識別次數,超過次數掛機
max_retries = 3
def on_input(session, callback_type, obj, arg):global max_retriesif max_retries <= 0:# 重試次數用完,結束監聽speak(session, "通話超時，感謝您的使用")session.hangup()return "break"# 處理用戶按鍵if callback_type == "dtmf":session.consoleLog("info", f"用戶按了DTMF,收集到的按鍵:{obj.digit}")return ""# 處理事件if callback_type == "event":event_type = obj.getHeader("Speech-Type")# 觸發了detected-speech事件if event_type == "detected-speech":speech_output = obj.getBody()if speech_output:text = parse_xml(speech_output)if text and len(text) >= 2:# 獲取結果后進行其他處理process_result(session, text)speak(session, "還有什么可以幫到您?")else:# 識別結果太短,重新開始監聽speak(session, "沒聽清,請再說一遍")else:# 識別結果太短,重新開始監聽speak(session, "沒聽清,請再說一遍")max_retries -= 1# 立即結束本次處理return "break"# 立即結束本次處理return "break"# 立即結束本次處理return "break"# 通話入口
def handler(session, args):# 應答session.answer()# 是否向event_socket發送asr事件,我們是撥號規則處理asr事件,所以不需要# session.setVariable("fire_asr_events", "true")# 注冊事件回調,python的回調函數不要加引號session.setInputCallback(on_input)# 歡迎語speak(session,text="請問有什么可以幫助您的?")# 一直聽,識別,讀結果,直到max_retries用完while session.ready():start_listen(session)session.streamFile("silence_stream://90000000")# 掛機session.hangup()

4.2 主要是detect_speech和speak

5、效果

freeswitch 百度mrcp測試

6、一些參考的官網網址

mod_unimrcp
https://developer.signalwire.com/freeswitch/FreeSWITCH-Explained/Modules/mod_unimrcp_6586728
MRCP的參數設置
No-Input-Timeout, Recognition-Timeout, Start-Input-Timers
https://www.rfc-editor.org/rfc/rfc6787
speak
https://developer.signalwire.com/freeswitch/FreeSWITCH-Explained/Modules/mod-dptools/6587123/#0-about
play_and_detect_speech
https://developer.signalwire.com/freeswitch/FreeSWITCH-Explained/Modules/mod-dptools/6586714#lua
detect_speech
https://developer.signalwire.com/freeswitch/FreeSWITCH-Explained/Modules/mod-dptools/6586756#about
詳細的語音識別流程,識別用戶分機的案例
https://developer.signalwire.com/freeswitch/FreeSWITCH-Explained/Client-and-Developer-Interfaces/Lua-API-Reference/Lua-examples/Lua-ASR-TTS-Directory-example_1049011/