目录
- 现有的一些功能
- 记录一些过程中遇到的问题
- 安装llama_cpp
- 1、安装ollama和部署deepseek R1
- 2、使用本地部署的deepseek R1模型
- 3、语音识别
- 4、代码实现
- 运行演示
现有的一些功能
1、正常与人沟通，但受限于电脑性能，还存在一定延迟；
2、可以根据交流内容修改提示词，例如用户名称；
3、拥有文本和json两种输出形式，为未来拓展至智能机器人提供可能；
4、根据简单指令实现关闭程序操作。
更多功能和设想等待开发与实现中。。。
记录一些过程中遇到的问题
安装llama_cpp
正常是直接install llama-cpp-python，但是在编译过程中会出现各种各样的报错，这里推荐根据环境直接安装预编译版本的llama-cpp。 我是Windows环境，python版本是3.11，所以直接下载llama_cpp_python-0.3.2-cp311-cp311-win_amd64.whl并安装即可。
1、安装ollama和部署deepseek R1
这一步比较简单，直接按照官网教程一步步安装即可，安装完成之后，就可以打开终端，下载自己想要部署在本地的大模型了，例如想在本地部署deepseek R1 1.5B模型，那么在终端内输入ollama run deepseek-r1:1.5b即可，注意，除了目前开源的671B模型，其余ollama提供的R1模型均为蒸馏模型，具体可在官网找到详细介绍。
2、使用本地部署的deepseek R1模型
使用也非常简单，最直接的是在终端内直接运行和使用，注意不要输错模型的名称即可。但是这种方法毕竟不够优美，缺少一个与用户自然交互的前台。有两款比较好用的前台推荐给大家，一个是AnythingLLM，很多文章都基于这款软件去使用本地部署的大模型，灵活且具有一定可玩性；但其实面向我们普通使用者而言我更推荐的是page assist，作为一款浏览器插件，简直不要太方便，可以在浏览器中直接调用本地部署的大模型来帮助答疑解惑，用的还是很频繁，上手也很简单，基本上有手就行。
当然，以上只是作为普通使用者推荐的一些使用本地大模型的方法，作为开发者，我们当然希望能够在程序中调用大模型，帮助我们完成更加个性化的操作，这里我们还是使用ollama所提供的API。由于我们已经安装了llama_cpp，所以理论上我们可以直接使用llama_cpp所提供的接口即可，但是很遗憾，目前llama_cpp预编译版本没有最新版的，直接调用deepseek模型会出现无法加载模型的问题，而一些其他比较早期的模型，例如Llama系列或者Qwen系列是不会出现这个问题。上网一番搜索发现，需要将llama_cpp升级为最新，所以就尴尬了。好在我们还有ollama提供的API可以使用。
3、语音识别
解决完使用本地大模型的问题后，接下来我们要实现语音识别。我们的目标是能够访问本地的模型，与人进行语音对话交谈。那就需要将人的语音，转换为文本，再与大模型交互。原本我是想使用本地部署openai的开源语音模型whisper，但是笔记本性能太过有限，一番折腾后最终还是选择用线上的语音识别模型。目前所使用的是科大讯飞的实时语音转写API，在网络条件还不错的情况下识别速度很快，而且准确性较高，最重要的是新用户一年免费使用50完次，还是很香的。
此外，讯飞的星火大模型，据说在文本生成、语言理解等方面超越GPT 4-Turbo，而且也有免费token，之后也可以试试。
4、代码实现
有了调用本地大模型的API和在线语音的API后，我们就非常轻松的实现一个语音交互助手。直接上代码：
import websocket
import hashlib
import base64
import hmac
import json
import pyaudio
import sys
import re
from ollama import Client
import pyttsx3
from typing import Dict, List
import _thread as thread
from urllib.parse import urlencode
from wsgiref.handlers import format_date_time
from time import mktime
from datetime import datetime
import sslSTATUS_FIRST_FRAME = 0
STATUS_CONTINUE_FRAME = 1
STATUS_LAST_FRAME = 2class VoiceAssistant:def __init__(self, endpoint, mode, model_name):self.ollama_endpoint = endpointself.mode = modeself.history: List[Dict] = []self.model_name = model_nameself.username = Noneself.is_first_interaction = Trueself.presets = {"text": {"system": "你是一个有帮助的助手，请用非常简单直接的方式回答"},"json": {"system": "请始终用JSON格式回答，包含'response'和'sentiment'字段","response_template": {"response": "", "sentiment": ""}}}self.tts_engine = pyttsx3.init()self.setup_voice_engine()self.ws_param = Nonedef setup_voice_engine(self):"""配置语音引擎参数"""voices = self.tts_engine.getProperty('voices')self.tts_engine.setProperty('voice', voices[0].id)self.tts_engine.setProperty('rate', 150)def update_ws_param(self, ws_param, username="AI助手"):self.ws_param = ws_paramself.presets = {"text": {"system": "你的名字叫" + username + "，用户的名字叫主人，每次回答加上用户的名字，并且用非常简短直接的方式回答"},"json": {"system": "请始终用JSON格式回答，包含'response'和'sentiment'字段","response_template": {"response": "", "sentiment": ""}}}def update_username(self, username="AI助手"):self.presets = {"text": {"system": "你的名字叫" + username + "，每次回答请带上自己的名字，并且用非常简短直接的方式回答"},"json": {"system": "请始终用JSON格式回答，包含'response'和'sentiment'字段","response_template": {"response": "", "sentiment": ""}}}class Ws_Param:def __init__(self, APPID, APIKey, APISecret, vad_eos=10000):self.APPID = APPIDself.APIKey = APIKeyself.APISecret = APISecretself.CommonArgs = {"app_id": self.APPID}self.BusinessArgs = {"domain": "iat", "language": "zh_cn", "accent": "mandarin", "vad_eos": int(vad_eos)}def create_url(self):now = datetime.now()date = format_date_time(mktime(now.timetuple()))signature_origin = "host: ws-api.xfyun.cn\ndate: {}\nGET /v2/iat HTTP/1.1".format(date)signature_sha = hmac.new(self.APISecret.encode('utf-8'), signature_origin.encode('utf-8'),digestmod=hashlib.sha256).digest()signature_sha = base64.b64encode(signature_sha).decode('utf-8')authorization_origin = 'api_key="{}", algorithm="hmac-sha256", headers="host date request-line", signature="{}"'.format(self.APIKey, signature_sha)authorization = base64.b64encode(authorization_origin.encode('utf-8')).decode('utf-8')v = {"authorization": authorization, "date": date, "host": "ws-api.xfyun.cn"}return 'wss://ws-api.xfyun.cn/v2/iat?' + urlencode(v)def generate_response(prompt: str, assistant: VoiceAssistant) -> str:"""处理命令并生成响应"""# 命令处理if re.search(r"切换(到|为)JSON模式", prompt):assistant.mode = "json"return "已切换到JSON模式"elif re.search(r"切换(到|为)文本模式", prompt):assistant.mode = "text"return "已切换到文本模式"# 正常响应生成current_preset = assistant.presets[assistant.mode]messages = [{"role": "system", "content": current_preset["system"]},*assistant.history[-5:],{"role": "user", "content": prompt}]try:client = Client(host=assistant.ollama_endpoint)response = client.chat(model=assistant.model_name, messages=messages)return response["message"]["content"]except Exception as e:return f"请求API出错: {str(e)}"def parse_response(response: str, mode: str) -> str:"""解析不同模式的响应"""s1 = "<think>"s2 = "</think>"new_response = deleteByStartAndEnd(response, s1, s2)if mode == "json":try:data = json.loads(new_response)return data.get("response", "无效的JSON格式")except json.JSONDecodeError:return "响应解析失败"return new_response
def deleteByStartAndEnd(s, start, end):# 找出两个字符串在原始字符串中的位置，开始位置是：开始始字符串的最左边第一个位置，结束位置是：结束字符串的最右边的第一个位置x1 = s.index(start)x2 = s.index(end) + len(end) # s.index()函数算出来的是字符串的最左边的第一个位置# 找出两个字符串之间的内容x3 = s[x1:x2]# 将内容替换为控制符串result = s.replace(x3, "")return result
def speak(text: str, assistant: VoiceAssistant):"""文本转语音输出"""print(f"{assistant.username}回答: {text}")assistant.tts_engine.say(text)assistant.tts_engine.runAndWait()start_listening(assistant) # 语音输出完成后重新开始监听def on_message(ws, message, assistant):try:result = json.loads(message)if result['code'] == 0:text = ''.join([w['w'] for item in result['data']['result']['ws'] for w in item['cw']])print(f"用户输入: {text}")if re.search(r"退出|再见|关闭|结束|停止|拜拜", text):ws.close()sys.exit(0)elif assistant.is_first_interaction:assistant.username = text.strip() or "AI助手"response = f"感谢您为我命名，现在我叫{assistant.username}，请问有什么可以帮您？"assistant.is_first_interaction = Falseassistant.update_username(assistant.username)else:raw_response = generate_response(text, assistant)response = parse_response(raw_response, assistant.mode)speak(response, assistant)except Exception as e:print("处理错误:", e)def on_error(ws, error):print(f"### 错误: {error}")if "SSL" in str(error) or "EOF" in str(error):print("检测到SSL错误，尝试重新连接...")start_listening(ws.assistant) # 需要确保assistant对象可以通过ws访问def on_close(ws, close_status_code, close_msg):print(f"### 连接关闭 ### 状态码: {close_status_code}, 消息: {close_msg}")def on_open(ws):print("### 连接已打开 ###")def run(*args):audio_generator = record_audio()send_audio(ws, audio_generator)thread.start_new_thread(run, ())def send_audio(ws, audio_generator):status = STATUS_FIRST_FRAMEprint("开始发送音频...")try:for chunk in audio_generator:if not ws.sock or not ws.sock.connected: # 新增连接状态检查print("连接已断开，停止发送音频")break# 原有发送逻辑保持不变...data = {"common": ws.ws_param.CommonArgs,"business": ws.ws_param.BusinessArgs,"data": {"status": STATUS_FIRST_FRAME if status == 0 else STATUS_CONTINUE_FRAME,"format": "audio/L16;rate=16000","audio": base64.b64encode(chunk).decode('utf-8')}}ws.send(json.dumps(data))status = STATUS_CONTINUE_FRAMEif ws.sock and ws.sock.connected:ws.send(json.dumps({"data": {"status": STATUS_LAST_FRAME}}))except Exception as e:print("发送错误:", e)ws.close() # 确保关闭失效连接def record_audio(rate=16000, chunk_size=1024):p = pyaudio.PyAudio()stream = p.open(format=pyaudio.paInt16, channels=1,rate=rate, input=True, frames_per_buffer=chunk_size)try:while True:yield stream.read(chunk_size)except KeyboardInterrupt:print("停止录音")finally:stream.stop_stream()stream.close()p.terminate()def start_listening(assistant):"""启动新的语音识别会话"""print("开始新的语音识别会话...")websocket.enableTrace(False)ws_url = assistant.ws_param.create_url() ws = websocket.WebSocketApp(ws_url,on_message=lambda ws, msg: on_message(ws, msg, assistant),on_error=on_error,on_close=on_close)ws.ws_param = assistant.ws_paramws.on_open = on_open# 修改运行参数ws.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE})
if __name__ == "__main__":with open("config.json") as f:config = json.load(f)# 初始化语音助手voice_assistant = VoiceAssistant(endpoint=config["endpoint"],mode=config["mode"],model_name=config["model_name"])# 配置语音识别参数ws_param = Ws_Param(APPID=config["APPID"],APIKey=config["APIKey"],APISecret=config["APISecret"],vad_eos=config["vad_eos"])voice_assistant.update_ws_param(ws_param)# 初始问候voice_assistant.tts_engine.say("您好，请为我命名")voice_assistant.tts_engine.runAndWait()# 开始首次监听start_listening(voice_assistant)# 保持主线程运行while True:pass
另外一个配置文件(config.json)如下格式：
{"APPID": "Websocket服务接口认证信息","APIKey": "Websocket服务接口认证信息","APISecret": "Websocket服务接口认证信息","vad_eos": "10000","endpoint": "http://localhost:11434","mode": "text","model_name": "deepseek-r1:1.5b"
}其中前三项为讯飞语音接口的验证信息，注册后根据自己的接口信息进行修改即可，"vad_eos"为录音时长，最多不超过60秒，即600000毫秒，以毫秒为单位；”endpoint“为访问ollama的本地大模型的端口，默认为11434； "mode"为输出形式，即text或json；"model_name"为部署在本地的大模型名字，确保下载的模型和本程序使用的模型保持一直即可。
运行演示
打开ollama，运行程序效果如下：
