本地Ollama语音聊天机器人（流式语音版）-QZQ

# -*- coding: utf-8 -*-

import tkinter as tk
from tkinter import scrolledtext, messagebox
import threading
import json
import queue
import os
import re
import time

from ollama import chat

# 语音相关库
try:
import sounddevice as sd
import vosk
import pyttsx3
VOICE_AVAILABLE = True
except ImportError:
VOICE_AVAILABLE = False

class OllamaChatApp:
def __init__(self, root):
self.root = root
self.root.title("🤖 本地Ollama语音聊天机器人（流式语音版）")
self.root.geometry("920×800")
self.root.minsize(520, 420)

# ===== 对话历史 =====
self.messages = []
self.is_streaming = False

# ===== Vosk 语音识别 =====
self.vosk_model = None
self.audio_queue = queue.Queue()
self.is_listening = False
self.stream = None

# ===== TTS 语音合成 =====
self.tts_engine = None
self.tts_enabled = tk.BooleanVar(value=False)
self.tts_queue = queue.Queue() # TTS 消息队列
self.tts_thread = None # 唯一 TTS 消费线程
self.tts_lock = threading.Lock() # 初始化锁

# 流式播报模式：True=边说边播，False=整段播完再读
self.tts_stream_mode = tk.BooleanVar(value=True)

# 流式 TTS 的句子缓冲
self.tts_sentence_buffer = ""
self._prev_stream_len = 0

# ===== 获取已安装模型 =====
try:
import ollama
models = ollama.list()["models"]
self.model_names = [m["model"] for m in models]
except Exception:
self.model_names = ["llama3.1", "qwen2.5:7b", "deepseek-r1:7b"]

# ==================== 顶部：模型选择 ====================
top_frame = tk.Frame(root)
top_frame.pack(pady=8, fill=tk.X, padx=10)

tk.Label(top_frame, text="选择模型:").pack(side=tk.LEFT, padx=5)
self.model_var = tk.StringVar(value=self.model_names[0] if self.model_names else "")
self.model_menu = tk.OptionMenu(top_frame, self.model_var, *self.model_names)
self.model_menu.pack(side=tk.LEFT, padx=5)

self.clear_btn = tk.Button(top_frame, text="🗑️ 清空对话", command=self.clear_chat)
self.clear_btn.pack(side=tk.RIGHT, padx=5)

# ==================== 语音控制栏 ====================
voice_frame = tk.Frame(root)
voice_frame.pack(pady=5, fill=tk.X, padx=10)

self.voice_btn = tk.Button(
voice_frame,
text="🎤 点击说话",
command=self.toggle_listen,
bg="#2196F3",
fg="white",
font=("微软雅黑", 11, "bold"),
padx=15,
pady=5,
state="normal" if VOICE_AVAILABLE else "disabled"
)
self.voice_btn.pack(side=tk.LEFT, padx=5)

self.status_label = tk.Label(
voice_frame,
text="语音状态：就绪",
fg="#666666",
font=("微软雅黑", 10)
)
self.status_label.pack(side=tk.LEFT, padx=15)

# TTS 开关
self.tts_check = tk.Checkbutton(
voice_frame,
text="🔊 语音播报",
variable=self.tts_enabled,
font=("微软雅黑", 10),
command=self.on_tts_toggle
)
self.tts_check.pack(side=tk.RIGHT, padx=5)

# 流式播报开关
self.stream_check = tk.Checkbutton(
voice_frame,
text="⚡ 流式播报",
variable=self.tts_stream_mode,
font=("微软雅黑", 10)
)
self.stream_check.pack(side=tk.RIGHT, padx=5)

if not VOICE_AVAILABLE:
self.status_label.config(
text="语音库未安装，请 pip install vosk sounddevice pyttsx3",
fg="#FF0000"
)

# ==================== 中间：对话显示区 ====================
self.chat_display = scrolledtext.ScrolledText(
root, wrap=tk.WORD, font=("微软雅黑", 11),
state="disabled", height=25
)
self.chat_display.pack(padx=10, pady=5, fill=tk.BOTH, expand=True)

self.chat_display.tag_config("user", foreground="#0066CC", font=("微软雅黑", 11, "bold"))
self.chat_display.tag_config("assistant", foreground="#008000", font=("微软雅黑", 11))
self.chat_display.tag_config("streaming", foreground="#008000", font=("微软雅黑", 11), underline=True)
self.chat_display.tag_config("error", foreground="#FF0000")
self.chat_display.tag_config("voice", foreground="#9C27B0", font=("微软雅黑", 11, "italic"))

# ==================== 底部：输入框和发送 ====================
bottom_frame = tk.Frame(root)
bottom_frame.pack(pady=10, padx=10, fill=tk.X)

self.input_entry = tk.Text(bottom_frame, height=3, font=("微软雅黑", 11), wrap=tk.WORD)
self.input_entry.pack(side=tk.LEFT, fill=tk.BOTH, expand=True)
self.input_entry.bind("<Return>", lambda e: self.send_message())

self.send_btn = tk.Button(
bottom_frame, text="发送 (Enter)", command=self.send_message,
bg="#4CAF50", fg="white", font=("微软雅黑", 10), padx=10
)
self.send_btn.pack(side=tk.RIGHT, padx=(10, 0))

# 欢迎消息
self.append_message("assistant", "👋 你好！点击「🎤 点击说话」进行语音输入，或直接打字。")

# 预加载 Vosk 模型
if VOICE_AVAILABLE:
threading.Thread(target=self.load_vosk_model, daemon=True).start()

# ==================== Vosk 语音识别 ====================

def load_vosk_model(self):
model_path = "vosk-model-small-cn-0.22"
try:
if not os.path.exists(model_path):
self.root.after(0, lambda: self.status_label.config(
text=f"未找到模型目录: {model_path}", fg="#FF0000"
))
return
vosk.SetLogLevel(-1)
self.vosk_model = vosk.Model(model_path)
self.root.after(0, lambda: self.status_label.config(
text="语音状态：模型已加载，可以说话", fg="#008000"
))
except Exception as e:
self.root.after(0, lambda: self.status_label.config(
text=f"模型加载失败: {e}", fg="#FF0000"
))

def toggle_listen(self):
if self.is_listening:
self.stop_listen()
else:
self.start_listen()

def start_listen(self):
if self.vosk_model is None:
messagebox.showwarning("提示", "语音模型还没加载完，请稍候…")
return

self.is_listening = True
self.voice_btn.config(text="⏹️ 停止录音", bg="#F44336")
self.status_label.config(text="🎙️ 正在聆听…请说话", fg="#2196F3")

while not self.audio_queue.empty():
try:
self.audio_queue.get_nowait()
except queue.Empty:
break

threading.Thread(target=self.audio_capture_thread, daemon=True).start()
threading.Thread(target=self.recognize_thread, daemon=True).start()

def stop_listen(self):
self.is_listening = False
if self.stream:
try:
self.stream.stop()
self.stream.close()
except Exception:
pass
self.stream = None
self.voice_btn.config(text="🎤 点击说话", bg="#2196F3")
self.status_label.config(text="语音状态：就绪", fg="#666666")

def audio_capture_thread(self):
try:
self.stream = sd.RawInputStream(
samplerate=16000,
blocksize=8000,
dtype='int16',
channels=1,
callback=self._audio_callback
)
self.stream.start()
while self.is_listening:
time.sleep(0.1)
except Exception as e:
self.root.after(0, lambda: self.status_label.config(
text=f"麦克风错误: {e}", fg="#FF0000"
))
self.is_listening = False

def _audio_callback(self, indata, frames, time_info, status):
if self.is_listening:
self.audio_queue.put(bytes(indata))

def recognize_thread(self):
rec = vosk.KaldiRecognizer(self.vosk_model, 16000)
partial_text = ""

while self.is_listening:
try:
data = self.audio_queue.get(timeout=0.3)
except queue.Empty:
continue

if rec.AcceptWaveform(data):
result = json.loads(rec.Result())
text = result.get("text", "").replace(" ", "")
if text:
partial_text = text
self.root.after(0, self.fill_input, text)
self.root.after(0, self.auto_send_voice, text)
break
else:
partial = json.loads(rec.PartialResult()).get("partial", "").replace(" ", "")
if partial:
self.root.after(0, self.update_partial, partial)

if self.is_listening:
final = json.loads(rec.FinalResult()).get("text", "").replace(" ", "")
if final and final != partial_text:
self.root.after(0, self.fill_input, final)
self.root.after(0, self.auto_send_voice, final)

self.root.after(0, self.stop_listen)

def update_partial(self, text):
self.status_label.config(text=f"🎙️ 识别中: {text}", fg="#2196F3")

def fill_input(self, text):
self.input_entry.delete("1.0", tk.END)
self.input_entry.insert("1.0", text)

def auto_send_voice(self, text):
if text.strip():
self.append_message("voice", f"🎤 语音识别：{text}")
self.send_message()

# ==================== TTS 语音合成 ====================

def on_tts_toggle(self):
if self.tts_enabled.get():
if self.tts_engine is None:
self._create_tts_engine()
elif self.tts_thread is None or not self.tts_thread.is_alive():
self.tts_thread = threading.Thread(
target=self._tts_worker, daemon=True
)
self.tts_thread.start()
print("[TTS] 引擎线程已重启")

def _create_tts_engine(self):
"""只启动 TTS 消费线程，引擎在 worker 里每次新建"""
with self.tts_lock:
if self.tts_thread is not None and self.tts_thread.is_alive():
return
try:
# 探测 pyttsx3 是否可用
test_engine = pyttsx3.init()
test_engine.stop()
del test_engine

self.tts_engine = True # 只是标记
self.tts_thread = threading.Thread(
target=self._tts_worker, daemon=True
)
self.tts_thread.start()
print("[TTS] 引擎线程已启动（每次播报重建引擎）")
except Exception as e:
messagebox.showerror("TTS错误", f"语音合成初始化失败: {e}")
self.tts_enabled.set(False)
self.tts_engine = None

def _tts_worker(self):
"""TTS 消费线程：每次播报都新建引擎，避免 SAPI5 状态污染"""
while True:
try:
text = self.tts_queue.get()
except Exception:
continue

if text is None:
break
if not text.strip():
self.tts_queue.task_done()
continue

try:
engine = pyttsx3.init()
engine.setProperty('rate', 180)
engine.setProperty('volume', 0.9)

voices = engine.getProperty('voices')
for voice in voices:
vid = voice.id.lower()
vname = voice.name.lower()
if ('chinese' in vname or 'zh' in vid or 'huihui' in vid
or 'kangkang' in vid or 'yaoyao' in vid):
engine.setProperty('voice', voice.id)
break

engine.say(text)
engine.runAndWait()
engine.stop()
del engine

except Exception as e:
print(f"[TTS] 播放错误: {e}")
finally:
try:
self.tts_queue.task_done()
except Exception:
pass

def speak(self, text, drop_old=False):
"""把文本丢进 TTS 队列"""
if not self.tts_enabled.get() or self.tts_engine is None:
return
if not text or not text.strip():
return

clean = re.sub(r'[*#`_~>\\[\\]()【】「」]', '', text).strip()
if not clean:
return

# 仅在整段播报模式下允许丢旧消息
if drop_old:
try:
if self.tts_queue.qsize() > 3:
while self.tts_queue.qsize() > 1:
self.tts_queue.get_nowait()
self.tts_queue.task_done()
except Exception:
pass

self.tts_queue.put(clean)

# ==================== 流式 TTS 句子切分 ====================

def _flush_tts_sentences(self, force=False):
"""
把 tts_sentence_buffer 里已完成的句子切出来，丢进 TTS 队列。
force=True 时把剩余全部切出来（流式结束时用）。
"""
if not self.tts_enabled.get():
return

buffer = self.tts_sentence_buffer
if not buffer:
return

# ✅ 强断句符
end_marks = "。！？!?；;\\n"
# ✅ 弱断句符（逗号、顿号、冒号）
soft_marks = "，,、：:"

sentences = []
start = 0
for i, ch in enumerate(buffer):
if ch in end_marks:
seg = buffer[start:i + 1].strip()
if seg:
sentences.append(seg)
start = i + 1
elif ch in soft_marks and (i – start) >= 8: # ✅ 从 15 降到 8
seg = buffer[start:i + 1].strip()
if seg:
sentences.append(seg)
start = i + 1

remainder = buffer[start:]

# ✅ 关键：流式模式下，若剩余缓冲已够长（比如 >= 20 字）且迟迟没标点，
# 也先播一段，避免"一直不响"
if not force and len(remainder) >= 20:
# 从中间找个空格或直接硬切
cut = remainder.rfind(" ", 0, 20)
if cut <= 0:
cut = 20
sentences.append(remainder[:cut].strip())
remainder = remainder[cut:]

if force:
if remainder.strip():
sentences.append(remainder.strip())
self.tts_sentence_buffer = ""
else:
self.tts_sentence_buffer = remainder

for s in sentences:
if s:
print(f"[TTS-流式] 播报片段: {s[:30]}…")
self.speak(s, drop_old=False)

# ==================== 聊天模块 ====================

def append_message(self, role, content):
self.chat_display.config(state="normal")
if role == "user":
self.chat_display.insert(tk.END, "🧑 你：\\n", "user")
elif role == "assistant":
self.chat_display.insert(tk.END, "🤖 助手：\\n", "assistant")
elif role == "streaming":
self.chat_display.insert(tk.END, "🤖 助手：\\n", "streaming")
elif role == "voice":
self.chat_display.insert(tk.END, content + "\\n", "voice")
self.chat_display.see(tk.END)
self.chat_display.config(state="disabled")
return
else:
self.chat_display.insert(tk.END, "⚠️ 错误：\\n", "error")

self.chat_display.insert(tk.END, content + "\\n")
self.chat_display.see(tk.END)
self.chat_display.config(state="disabled")

def clear_chat(self):
if self.is_streaming:
return
self.messages = []
self.chat_display.config(state="normal")
self.chat_display.delete(1.0, tk.END)
self.chat_display.config(state="disabled")
self.append_message("assistant", "🗑️ 对话已清空，重新开始吧！")

def send_message(self):
if self.is_streaming:
return
user_input = self.input_entry.get("1.0", tk.END).strip()
if not user_input:
return

self.input_entry.delete("1.0", tk.END)
if not user_input.startswith("🎤"):
self.append_message("user", user_input)

self.messages.append({"role": "user", "content": user_input})
self.send_btn.config(state="disabled", text="⏳ 生成中…")
self.is_streaming = True
# ✅ 双重保险：发送时就重置流式状态
self.tts_sentence_buffer = ""
self._prev_stream_len = 0
threading.Thread(target=self.get_ollama_response, daemon=True).start()

def get_ollama_response(self):
try:
self.root.after(0, self.prepare_streaming_area)
full_reply = ""
stream = chat(
model=self.model_var.get(),
messages=self.messages,
stream=True
)
for chunk in stream:
content = chunk["message"]["content"]
full_reply += content
self.root.after(0, self.update_streaming, full_reply)

self.root.after(0, self.on_stream_finish, full_reply)
except Exception as e:
self.root.after(0, self.on_response_error, str(e))

def prepare_streaming_area(self):
self.chat_display.config(state="normal")
self.chat_display.insert(tk.END, "🤖 助手：\\n", "assistant")
self.streaming_start_pos = self.chat_display.index("end-2c")
self.chat_display.config(state="disabled")

# 重置流式 TTS 状态
self.tts_sentence_buffer = ""
self._prev_stream_len = 0

def update_streaming(self, text):
self.chat_display.config(state="normal")
end_pos = self.chat_display.index("end-2c")
if getattr(self, 'streaming_start_pos', None):
self.chat_display.delete(self.streaming_start_pos, end_pos)
self.chat_display.insert(self.streaming_start_pos, text)
self.chat_display.see(tk.END)
self.chat_display.config(state="disabled")

# 流式播报模式：实时把新增文字累积，凑够句子就播
if (self.tts_enabled.get() and self.tts_engine is not None
and self.tts_stream_mode.get()):
if len(text) > self._prev_stream_len:
new_part = text[self._prev_stream_len:]
self.tts_sentence_buffer += new_part
self._prev_stream_len = len(text)
self._flush_tts_sentences(force=False)

def on_stream_finish(self, reply):
self.chat_display.config(state="normal")
end_pos = self.chat_display.index("end-2c")
if getattr(self, 'streaming_start_pos', None):
self.chat_display.delete(self.streaming_start_pos, end_pos)
self.chat_display.insert(self.streaming_start_pos, reply + "\\n")
self.chat_display.config(state="disabled")

self.messages.append({"role": "assistant", "content": reply})

self.send_btn.config(state="normal", text="发送 (Enter)")
self.is_streaming = False

# 语音播报
if self.tts_enabled.get() and self.tts_engine is not None:
if self.tts_stream_mode.get():
# 流式模式：把缓冲里剩余的尾巴播完
self._flush_tts_sentences(force=True)
else:
# 整段模式：整段播报（可丢弃旧消息）
self.speak(reply, drop_old=True)

def on_response_error(self, error_msg):
if getattr(self, 'streaming_start_pos', None):
self.chat_display.config(state="normal")
end_pos = self.chat_display.index("end-2c")
self.chat_display.delete(self.streaming_start_pos, end_pos)
self.chat_display.config(state="disabled")

self.append_message("error", f"请求失败：{error_msg}\\n请检查Ollama服务是否运行。")
self.send_btn.config(state="normal", text="发送 (Enter)")
self.is_streaming = False

# 清理流式 TTS 缓冲
self.tts_sentence_buffer = ""
self._prev_stream_len = 0

# ==================== 退出清理 ====================

def on_closing(self):
self.is_listening = False
if self.stream:
try:
self.stream.stop()
self.stream.close()
except Exception:
pass

try:
self.tts_queue.put(None)
except Exception:
pass

self.root.destroy()

# ========== 启动程序 ==========
if __name__ == "__main__":
root = tk.Tk()
app = OllamaChatApp(root)
root.protocol("WM_DELETE_WINDOW", app.on_closing)
root.mainloop()