1. 项目概述:当AI助手能听懂你的声音

最近在折腾一个挺有意思的东西:一个完全在本地运行的、能用语音控制的AI助手。想象一下,你对着电脑说一句“帮我总结一下今天的工作邮件”,它就能调用本地的语言模型,分析你的邮件内容并给出摘要,整个过程数据不出你的电脑,既保护隐私,又响应迅速。这就是我基于 Ollama Whisper 搭建的本地语音AI代理(Voice-Controlled Local AI Agent)的核心构想。

这个项目的驱动力很直接:一方面,云端AI服务虽然强大,但存在延迟、依赖网络、数据隐私顾虑以及持续的订阅成本;另一方面,随着像 Llama 3、Mistral 这类优秀的开源大语言模型(LLM)以及 Whisper 这样的顶尖语音识别模型的出现,在消费级硬件上运行一个功能完备的本地AI栈已成为可能。Ollama 的出现更是简化了本地大模型的部署和管理,让它变得像安装一个应用一样简单。于是,一个很自然的想法就是:把 Whisper 的耳朵、Ollama 的大脑,再加上一点“胶水代码”组合起来,创造一个能听会说、完全受控于本地的智能体。

它适合谁呢?如果你是对隐私有高要求的开发者、喜欢折腾本地化AI应用的技术爱好者、或者希望为自己的智能家居、个人知识库打造一个离线智能中枢的用户,这个项目会给你提供一个清晰的实现蓝图和可复现的路径。整个过程涉及了本地模型部署、语音识别集成、简单的应用逻辑编排,虽然不涉及复杂的AI代理框架,但足以构建一个功能完整、可扩展的原型。

2. 核心架构与工具选型解析

2.1 为什么是 Ollama + Whisper?

这个组合的选择,背后是几个非常务实的考量。

Ollama 的核心价值在于其极简的模型管理。在它出现之前,在本地运行一个大模型,你需要操心模型格式转换(GGUF、GPTQ等)、加载库(llama.cpp, transformers)、内存管理、上下文长度设置等一系列繁琐问题。Ollama 通过一个统一的命令行工具和 RESTful API,把这些都封装了起来。你只需要一句 ollama run llama3 ,它就会自动下载、加载并运行 Meta 的 Llama 3 模型。它支持庞大的模型库,从轻量级的 Phi-3 到庞大的 Llama 3 70B,并且持续更新。对于我们的语音代理来说,这意味着我们可以用一套固定的 API 调用方式(HTTP POST请求)与任何它支持的模型对话,极大地降低了集成复杂度。

Whisper 则是 OpenAI 开源的自动语音识别(ASR)模型,它在准确性和多语言支持上表现卓越。关键是,它有不同规模的版本(tiny, base, small, medium, large),我们可以根据硬件性能选择。例如,在 CPU 上, whisper-tiny whisper-base 就能实现近乎实时的识别,精度对于日常指令足够。Whisper 也提供了完善的 Python 库和命令行工具,方便我们捕获麦克风输入并进行转录。

本地化 是另一个关键决策。所有数据处理——从你的声音被麦克风捕获,到转换成文字,再到发送给语言模型生成回复——全部在你的设备上完成。这消除了网络延迟,保证了在无网环境下的可用性,最重要的是,你的所有对话、指令和可能涉及的敏感信息都不会离开你的设备。这种可控性,是云端服务无法比拟的。

注意:虽然模型在本地,但请确保你下载的模型文件来源可信。Ollama 官方仓库和 Hugging Face 是相对可靠的来源。

2.2 系统工作流设计

整个代理的工作流是一个清晰的管道(Pipeline),理解这个流程是后续开发的基础:

  1. 语音捕获与预处理 :通过 Python 的 sounddevice pyaudio 库,从系统默认麦克风持续或按需录制音频流。通常需要设置采样率(如 16000 Hz)、声道数和每次读取的音频块大小。录制到的原始 PCM 数据需要保存为 WAV 等格式,以供 Whisper 处理。
  2. 语音转文本(STT) :将录制好的音频文件路径传递给 Whisper 模型。这里我们调用 whisper.load_model(“base”) 加载模型,然后使用 model.transcribe(audio_path) 得到识别出的文本。这一步的输出就是纯字符串,例如“今天天气怎么样”。
  3. 文本理解与指令路由(可选但推荐) :得到的文本可能直接是问题,也可能包含控制指令。例如,“退出”或“停止监听”应该触发程序关闭,“切换到 llama3:8b 模型”应该改变 Ollama 的对话模型。这里可以引入一个简单的规则引擎或意图识别模块(甚至可以用一个小型的本地 LLM 来做),来解析用户意图,决定是将文本直接转发给 LLM,还是执行某个控制命令。
  4. 与大语言模型(LLM)交互 :将需要处理的文本(如问题、指令)通过 HTTP POST 请求发送给本地 Ollama 服务的 API 端点(默认是 http://localhost:11434/api/generate )。请求体中需要包含模型名称、提示词(Prompt)以及一些生成参数(如温度 temperature 、最大令牌数 max_tokens )。
  5. 响应处理与执行 :Ollama 会流式(stream)或非流式地返回 JSON 格式的响应。我们解析出其中的文本回复。这个回复可能本身就是最终答案,也可能是一个可执行的指令描述(例如“已为您打开文档”)。对于后者,我们的代理需要能够调用系统函数(如打开文件、执行命令)或与其他本地服务(如日历、邮件客户端)交互。这部分定义了代理的“行动能力”。
  6. 文本转语音(TTS,可选) :为了形成完整的语音交互闭环,可以将 LLM 返回的文本通过本地 TTS 引擎(如 pyttsx3 , edge-tts 或更高质量的 Coqui TTS )合成语音并播放出来,让代理真正“说”出答案。

这个工作流中, 步骤3和步骤5是代理“智能”和“能动性”的关键 。一个简单的问答机器人只需要1、2、4步。而要构建一个能真正“做事”的代理,就必须在3和5上下功夫,实现意图识别和功能调用。

3. 环境搭建与核心组件部署

3.1 基础Python环境与依赖库

建议使用 Python 3.9 或以上版本。创建一个独立的虚拟环境是一个好习惯,可以避免包依赖冲突。

# 创建并激活虚拟环境 (以 conda 为例)
conda create -n voice-agent python=3.10
conda activate voice-agent

# 或者使用 venv
python -m venv venv
source venv/bin/activate  # Linux/Mac
# venv\Scripts\activate  # Windows

接下来安装核心的 Python 库。我们将使用 whisper 官方库进行语音识别,使用 requests aiohttp 与 Ollama API 通信,使用 sounddevice scipy 处理音频。

pip install openai-whisper  # 语音识别核心
pip install sounddevice scipy  # 音频捕获和保存
pip install requests  # 调用 Ollama API
pip install pyttsx3  # 可选的本地文本转语音(跨平台,但声音较机械)
# 或者安装更高质量的 Coqui TTS (但更大更复杂)
# pip install TTS

实操心得:在 Windows 上安装 sounddevice 可能需要 PortAudio 库。最简单的方法是安装 Anaconda,它通常会包含这些底层依赖。如果遇到问题,可以尝试先安装 pip install pipwin ,然后 pipwin install pyaudio 作为 sounddevice 的替代或补充。

3.2 Ollama 的安装与模型拉取

Ollama 的安装极其简单。访问其官网,根据你的操作系统(Windows, macOS, Linux)下载对应的安装包或执行安装脚本。

  • Linux/macOS : 通常是一行 curl 命令。
  • Windows : 直接下载 exe 安装程序。

安装完成后,打开终端(Windows 上可能是 Ollama 自己的命令行或 PowerShell),启动 Ollama 服务。它通常会作为后台服务运行。

接下来,拉取你需要的语言模型。对于语音代理,响应速度是关键,因此建议从较小的模型开始测试。

# 拉取模型(以 Llama 3 8B 为例,这是一个在速度和能力上平衡得很好的模型)
ollama pull llama3:8b
# 也可以尝试更小更快的模型
ollama pull phi3:mini
ollama pull mistral:7b

拉取完成后,你可以通过 ollama run llama3:8b 在命令行交互测试模型是否正常工作。服务启动后,默认会在 http://localhost:11434 提供 API 服务。

3.3 Whisper 模型的选择与初始化

Whisper 模型有不同的尺寸,需要在精度和速度之间权衡:

  • tiny : 最快,内存占用最小(约 80 MB),适合实时性要求极高的场景,但准确度稍低。
  • base : 速度很快,内存占用适中(约 150 MB),是实时应用的常用选择。
  • small : 精度有显著提升,速度尚可(约 500 MB)。
  • medium/large : 精度最高,但速度慢,内存占用大(>1GB),不适合实时交互。

对于本地语音代理, base small 模型通常是首选 。在代码中初始化非常简单:

import whisper

model = whisper.load_model(“base”) # 首次运行会自动下载模型文件
# 后续运行会加载本地缓存,速度很快

模型下载后默认会保存在 ~/.cache/whisper 目录。确保你的磁盘有足够空间( large 模型约 3GB)。

4. 核心功能模块实现详解

4.1 语音捕获与音频预处理模块

可靠地捕获音频是第一步。我们使用 sounddevice 进行非阻塞式录音,并设置一个能量阈值(VAD,语音活动检测)来过滤背景噪音,实现“按下说话”或“语音唤醒”的效果。

import sounddevice as sd
import numpy as np
from scipy.io.wavfile import write
import queue
import threading

class AudioRecorder:
    def __init__(self, samplerate=16000, channels=1, threshold=0.01, silence_duration=1.0):
        self.samplerate = samplerate
        self.channels = channels
        self.threshold = threshold  # 音量阈值,低于此值视为静音
        self.silence_duration = silence_duration  # 持续静音多久停止录音(秒)
        self.audio_queue = queue.Queue()
        self.is_recording = False

    def _audio_callback(self, indata, frames, time, status):
        """这是 sounddevice 的流回调函数,每次有音频块就会调用。"""
        if status:
            print(f"音频流错误: {status}")
        # 计算当前音频块的能量(均方根)
        volume_norm = np.linalg.norm(indata) / np.sqrt(len(indata))
        # 将音频数据和能量值放入队列,供主线程处理
        self.audio_queue.put((indata.copy(), volume_norm))

    def record_until_silence(self):
        """开始录音,直到检测到持续静音。返回完整的音频数据。"""
        print("开始聆听...(说话即可)")
        self.is_recording = True
        all_audio = []
        silent_frames = 0
        silence_limit = int(self.silence_duration * self.samplerate / 1024) # 假设每块1024帧

        # 创建输入流
        with sd.InputStream(callback=self._audio_callback,
                            channels=self.channels,
                            samplerate=self.samplerate,
                            blocksize=1024):
            while self.is_recording:
                try:
                    audio_chunk, volume = self.audio_queue.get(timeout=1)
                except queue.Empty:
                    continue

                all_audio.append(audio_chunk)

                if volume < self.threshold:
                    silent_frames += 1
                else:
                    silent_frames = 0

                # 如果静音帧数超过限制,停止录音
                if silent_frames > silence_limit:
                    print("检测到静音,停止录音。")
                    self.is_recording = False
                    break

        if all_audio:
            # 将所有音频块拼接成一个 numpy 数组
            recorded_audio = np.concatenate(all_audio, axis=0)
            return recorded_audio
        else:
            return None

    def save_wav(self, audio_data, filename="output.wav"):
        """将 numpy 数组保存为 WAV 文件,Whisper 需要此格式。"""
        if audio_data is not None:
            # 确保数据是 float32 格式,并缩放到 int16 范围
            audio_int16 = (audio_data * 32767).astype(np.int16)
            write(filename, self.samplerate, audio_int16)
            print(f"音频已保存至: {filename}")
            return filename
        return None

这个类实现了带静音检测的录音。 threshold 参数需要根据你的麦克风和环境噪音进行调整,可以通过录制一段静音环境来观察 volume_norm 的典型值。

4.2 集成 Whisper 实现语音识别

有了音频文件,调用 Whisper 进行转录就很简单了。但为了提升体验,我们可以添加一些预处理和后处理。

import whisper
import numpy as np

class SpeechToTextEngine:
    def __init__(self, model_size="base"):
        print(f"正在加载 Whisper {model_size} 模型...")
        self.model = whisper.load_model(model_size)
        print("模型加载完毕。")

    def transcribe_audio(self, audio_path, language=None, initial_prompt=None):
        """
        转录音频文件。
        :param audio_path: WAV文件路径
        :param language: 指定语言(如 'zh', 'en'),None则自动检测
        :param initial_prompt: 可选的初始提示,帮助模型纠正特定词汇
        :return: 识别出的文本
        """
        # 可选:在这里可以添加音频预处理,如降噪、归一化(Whisper内部已做了一些)
        # 使用 Whisper 进行转录
        result = self.model.transcribe(
            audio_path,
            language=language,
            initial_prompt=initial_prompt, # 例如,如果领域专有名词多,可以提示
            fp16=False # 如果CPU运行,确保为False
        )
        text = result["text"].strip()
        # 简单的后处理:去除多余空格,处理标点
        # 例如,Whisper 中文输出有时会有空格,可以去掉
        if language and 'zh' in language:
            text = text.replace(" ", "")
        print(f"识别结果: {text}")
        return text

    def transcribe_audio_data(self, audio_numpy_array, samplerate=16000):
        """直接转录 numpy 音频数组,避免中间文件(需要 Whisper 版本支持)。"""
        # 较新版本的 Whisper API 支持直接传入 numpy 数组
        # 这里我们为了兼容性,先保存临时文件
        import tempfile
        with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmpfile:
            tmp_path = tmpfile.name
            # 需要将 audio_numpy_array 保存为 tmp_path
            # 这里省略保存代码,可使用 scipy.io.wavfile.write
            from scipy.io.wavfile import write
            write(tmp_path, samplerate, (audio_numpy_array * 32767).astype(np.int16))
            text = self.transcribe_audio(tmp_path)
            # 删除临时文件
            import os
            os.unlink(tmp_path)
            return text

initial_prompt 参数是一个很有用的技巧。如果你知道对话的上下文(比如总是在讨论编程),可以设置一个提示如“以下是关于Python编程的对话”,这能显著提升专有名词的识别准确率。

4.3 与 Ollama LLM 的通信模块

Ollama 提供了简洁的 HTTP API。我们将封装一个类来管理对话历史、发送请求并处理流式响应。

import requests
import json

class OllamaClient:
    def __init__(self, base_url="http://localhost:11434", model="llama3:8b"):
        self.base_url = base_url
        self.model = model
        self.conversation_history = [] # 保存对话上下文

    def generate_response(self, prompt, stream=False, temperature=0.7, max_tokens=500):
        """
        向 Ollama 发送生成请求。
        :param prompt: 用户输入的提示文本
        :param stream: 是否使用流式响应(实时看到生成过程)
        :param temperature: 创造性,越高越随机
        :param max_tokens: 生成的最大令牌数
        :return: 模型生成的回复文本
        """
        url = f"{self.base_url}/api/generate"
        # 构建包含历史上下文的完整提示(简单实现)
        full_prompt = self._build_prompt_with_history(prompt)

        payload = {
            "model": self.model,
            "prompt": full_prompt,
            "stream": stream,
            "options": {
                "temperature": temperature,
                "num_predict": max_tokens,
            }
        }

        try:
            if stream:
                return self._handle_stream_response(url, payload)
            else:
                response = requests.post(url, json=payload, timeout=60)
                response.raise_for_status()
                result = response.json()
                final_response = result.get("response", "").strip()
                # 更新对话历史
                self._update_history(prompt, final_response)
                return final_response
        except requests.exceptions.ConnectionError:
            print(f"错误:无法连接到 Ollama 服务,请确保 Ollama 正在运行于 {self.base_url}")
            return None
        except requests.exceptions.Timeout:
            print("错误:请求超时,模型可能正在加载或提示过长。")
            return None
        except Exception as e:
            print(f"调用 Ollama API 时发生错误: {e}")
            return None

    def _build_prompt_with_history(self, new_prompt, history_length=5):
        """构建包含最近N轮对话历史的提示词。"""
        if not self.conversation_history:
            return new_prompt

        # 简单拼接历史对话
        history_text = "\n".join([f"User: {h['user']}\nAssistant: {h['assistant']}" for h in self.conversation_history[-history_length:]])
        combined_prompt = f"{history_text}\nUser: {new_prompt}\nAssistant:"
        return combined_prompt

    def _update_history(self, user_input, assistant_output):
        """更新对话历史记录。"""
        self.conversation_history.append({
            "user": user_input,
            "assistant": assistant_output
        })
        # 可选:限制历史长度,防止上下文过长
        if len(self.conversation_history) > 10:
            self.conversation_history.pop(0)

    def _handle_stream_response(self, url, payload):
        """处理流式响应,实时打印并收集完整回复。"""
        full_response = ""
        try:
            with requests.post(url, json=payload, stream=True, timeout=60) as resp:
                resp.raise_for_status()
                for line in resp.iter_lines():
                    if line:
                        decoded_line = line.decode('utf-8')
                        data = json.loads(decoded_line)
                        chunk = data.get("response", "")
                        print(chunk, end="", flush=True) # 实时打印
                        full_response += chunk
                        if data.get("done", False):
                            break
                print() # 换行
                # 更新历史(使用完整的回复)
                user_prompt = payload["prompt"].split("\nUser: ")[-1].replace("\nAssistant:", "").strip()
                self._update_history(user_prompt, full_response.strip())
                return full_response.strip()
        except Exception as e:
            print(f"\n流式响应处理错误: {e}")
            return full_response

    def change_model(self, new_model):
        """动态切换模型。"""
        # 首先检查新模型是否可用(可选,可调用 /api/tags 端点)
        self.model = new_model
        print(f"已切换模型至: {new_model}")
        # 清空历史,因为不同模型的上下文格式可能不同
        self.conversation_history = []

这个客户端类处理了基本的对话管理。 stream=True 在调试时非常有用,你可以看到模型是如何“思考”并逐词生成的。对于最终产品,你可能希望关闭流式以获得更快的整体响应。

4.4 简单指令路由与代理逻辑

现在,我们需要一个“大脑”来协调录音、识别、LLM对话和可能的行动。这是代理的核心逻辑。我们实现一个简单的基于关键词的指令路由。

class VoiceAgent:
    def __init__(self, stt_model="base", llm_model="llama3:8b"):
        self.recorder = AudioRecorder(threshold=0.015) # 调整阈值
        self.stt_engine = SpeechToTextEngine(model_size=stt_model)
        self.llm_client = OllamaClient(model=llm_model)
        self.is_running = True
        # 定义系统指令和对应的处理函数
        self.system_commands = {
            "退出": self._cmd_exit,
            "停止": self._cmd_exit,
            "清空历史": self._cmd_clear_history,
            "切换模型": self._cmd_change_model,
            "帮助": self._cmd_help,
        }

    def run(self):
        """主运行循环。"""
        print("本地语音AI代理已启动。")
        print("说出指令或问题,检测到静音后自动处理。")
        print("说'退出'或'停止'来结束程序。\n")

        while self.is_running:
            # 1. 录音
            audio_data = self.recorder.record_until_silence()
            if audio_data is None:
                continue

            # 2. 保存临时音频文件并转录
            import tempfile
            with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmp:
                tmp_path = tmp.name
            self.recorder.save_wav(audio_data, tmp_path)
            user_text = self.stt_engine.transcribe_audio(tmp_path, language="zh") # 假设中文
            # 删除临时文件
            import os
            os.unlink(tmp_path)

            if not user_text:
                print("未识别到有效语音,请重试。")
                continue

            # 3. 检查是否为系统指令
            cmd_detected = False
            for cmd_keyword in self.system_commands.keys():
                if cmd_keyword in user_text:
                    self.system_commands[cmd_keyword](user_text)
                    cmd_detected = True
                    break

            if cmd_detected:
                continue # 如果是系统指令,不发送给LLM,继续下一轮循环

            # 4. 发送给 LLM 处理
            print(f"\n[用户] {user_text}")
            print("[AI] ", end="", flush=True)
            response = self.llm_client.generate_response(user_text, stream=True) # 使用流式输出

            # 5. (可选)文本转语音播报
            # self.speak(response)

    def _cmd_exit(self, command_text):
        print("收到退出指令。")
        self.is_running = False

    def _cmd_clear_history(self, command_text):
        self.llm_client.conversation_history = []
        print("对话历史已清空。")

    def _cmd_change_model(self, command_text):
        # 简单解析,例如“切换模型到 mistral”
        try:
            # 这里可以做得更智能,用LLM或正则表达式提取模型名
            # 简化版:假设命令是“切换模型 mistral:7b”
            parts = command_text.split()
            if len(parts) >= 3:
                new_model = parts[2]
                self.llm_client.change_model(new_model)
            else:
                print("请指定要切换的模型名称,例如‘切换模型 mistral:7b’")
        except Exception as e:
            print(f"切换模型失败: {e}")

    def _cmd_help(self, command_text):
        help_msg = """
        可用系统指令:
        - 退出 / 停止:结束程序。
        - 清空历史:清空当前对话上下文。
        - 切换模型 [模型名]:切换到指定模型(需已通过ollama pull下载)。
        - 帮助:显示此帮助信息。
        其他任何话语都将被视为问题或指令,发送给AI模型处理。
        """
        print(help_msg)

    # 可选的 TTS 功能
    def speak(self, text):
        try:
            import pyttsx3
            engine = pyttsx3.init()
            engine.say(text)
            engine.runAndWait()
        except ImportError:
            print("未安装 pyttsx3,跳过语音播报。")
        except Exception as e:
            print(f"语音播报失败: {e}")

if __name__ == "__main__":
    agent = VoiceAgent(stt_model="base", llm_model="llama3:8b")
    agent.run()

这个 VoiceAgent 类将各个模块串联起来,形成了一个可运行的基础代理。它通过关键词匹配来处理简单的系统指令,其他所有语音输入则交给 LLM 处理。这是一个 反应式代理(Reactive Agent) ,它根据当前输入做出反应,没有复杂的长期规划能力,但对于许多自动化任务和问答场景已经足够。

5. 进阶功能与优化方向

基础版本跑通后,我们可以从多个维度增强这个代理的能力和体验。

5.1 实现连续对话与上下文管理

上面的简单历史管理( _build_prompt_with_history )在对话轮次增多后,会迅速耗尽模型的上下文窗口(Context Window)。更优的方案是使用 滑动窗口 摘要压缩 技术。

  • 滑动窗口 :只保留最近 N 条对话记录(如最近10轮)。这是最简单的,但会丢失早期的重要信息。
  • 摘要压缩 :当对话历史达到一定长度时,调用 LLM 本身对之前的对话历史生成一个简短的摘要,然后用“摘要 + 近期对话”作为新的上下文。这需要额外的 LLM 调用,但能更有效地利用上下文长度。
def summarize_conversation(self, history):
    """使用LLM对长历史进行摘要。"""
    summary_prompt = f"""请将以下对话历史浓缩成一个简洁的摘要,保留核心事实和决策。
    对话历史:
    {history}
    摘要:"""
    # 调用一个更小、更快的模型(如 phi3:mini)来生成摘要
    summary = self._call_fast_model(summary_prompt)
    return summary

5.2 集成函数调用(Function Calling)能力

要让代理从“聊天”升级为“执行”,必须赋予它调用外部工具的能力。这需要:

  1. 定义工具 :用清晰的 JSON Schema 描述每个函数(工具)的名称、描述、参数。
  2. LLM 理解与规划 :将用户指令、可用工具列表和对话历史一起给 LLM,让 LLM 判断是否需要调用工具,以及调用哪个、参数是什么。
  3. 执行与反馈 :代理执行被调用的函数,将结果返回给 LLM,由 LLM 组织最终的自然语言回复给用户。

Ollama 的部分模型(如 Llama 3.1)支持类似 OpenAI 的 function calling 格式。你可以构造特定的 Prompt 来引导模型输出结构化 JSON。

# 一个简化的函数调用处理逻辑示例
tools = [
    {
        "name": "get_weather",
        "description": "获取指定城市的当前天气",
        "parameters": {
            "type": "object",
            "properties": {
                "location": {"type": "string", "description": "城市名"}
            },
            "required": ["location"]
        }
    },
    # ... 更多工具
]

def process_with_tools(self, user_input):
    # 构建包含工具描述的 Prompt
    prompt = f"""
    你是一个助手,可以调用工具。以下是可用工具:
    {json.dumps(tools, ensure_ascii=False)}
    根据用户请求,决定是否需要调用工具。
    如果需要,请严格按以下JSON格式回复:
    {{"tool": "工具名", "parameters": {{...}}}}
    如果不需要,请正常对话。
    用户请求:{user_input}
    助手:
    """
    llm_response = self.llm_client.generate_response(prompt, stream=False, temperature=0.1)
    # 尝试解析 JSON
    try:
        action = json.loads(llm_response)
        if "tool" in action:
            # 执行对应的函数
            result = self.execute_tool(action["tool"], action["parameters"])
            # 将结果再次喂给 LLM,生成用户友好的回复
            follow_up_prompt = f"用户问:{user_input}\n你调用了工具{action['tool']},得到结果:{result}。请根据此结果生成对用户的回复。"
            final_reply = self.llm_client.generate_response(follow_up_prompt)
            return final_reply
    except json.JSONDecodeError:
        # LLM 返回的不是 JSON,直接作为普通回复
        return llm_response

5.3 性能优化与实时性提升

实时语音交互对延迟非常敏感。优化点包括:

  • Whisper 模型量化 :使用 whisper.cpp faster-whisper 等优化版本,它们通过 C++ 实现和模型量化,能大幅提升转录速度,尤其适合 CPU 环境。
  • 音频流式处理 :不要等整段话说完再送 Whisper。可以将音频缓存切成小段(如 1-2 秒),使用 Whisper 的带时间戳的转录,并实时拼接文本。这能实现“边说边转”的效果,减少用户等待感。
  • LLM 响应加速 :使用量化程度更高的模型(如 q4_0 量化),或切换到更小的模型(如 Phi-3-mini )。在 Ollama 拉取模型时,可以指定量化版本,如 ollama pull llama3:8b-q4_0
  • 硬件加速 :如果有 NVIDIA GPU,确保 Ollama 和 Whisper 都启用了 CUDA 支持。对于 Whisper,安装 pip install openai-whisper 时会自动尝试安装 CUDA 版本的 PyTorch(如果环境合适)。对于 Ollama,在支持 GPU 的系统上,它通常会优先使用 GPU。

5.4 添加唤醒词与持续监听

目前的实现是“按静音检测停止”的对话模式。更自然的交互是添加一个 唤醒词 (如“嗨,助手”),平时代理处于低功耗监听状态,只有检测到唤醒词后才开始录制并处理后续指令。这需要集成一个轻量级的离线唤醒词检测引擎,如 Porcupine (付费商业库有更准的版本)或 Vosk 中的关键词识别功能。实现逻辑变为:

while True:
    audio_chunk = listen_for_1_second()
    if wake_word_detector.process(audio_chunk):
        print("唤醒词检测到!")
        # 开始录制主要指令
        main_audio = record_until_silence()
        # ... 后续处理流程

6. 常见问题与故障排查实录

在实际搭建和运行过程中,你几乎一定会遇到下面这些问题。这里记录了我的踩坑经验和解决方案。

6.1 音频相关问题

问题:录音没有声音或全是噪音。

  • 排查1:检查默认输入设备 。运行 python -c "import sounddevice as sd; print(sd.query_devices())" 查看所有音频设备。确认你使用的设备索引是否正确。在 AudioRecorder 初始化时,可以指定 device= 参数。
  • 排查2:调整音量阈值 。环境噪音大的办公室和安静的家里,阈值 ( threshold ) 需要不同。写一个小的测试脚本,在静音和说话时打印 volume_norm 的值,据此设置一个合理的阈值(比如静音时平均值的 2-3 倍)。
  • 排查3:麦克风权限 。在 macOS 和 Linux 上通常没问题。在 Windows 上,确保 Python 解释器或终端有麦克风访问权限(系统设置 -> 隐私 -> 麦克风)。

问题:Whisper 识别中文不准,尤其是专有名词。

  • 解决1:指定语言 。在 transcribe_audio 中明确设置 language='zh' ,避免自动检测错误。
  • 解决2:使用 initial_prompt 。如果你在特定领域(如编程),可以在提示里加入领域关键词,例如 initial_prompt="以下是关于计算机编程和软件开发的对话。"
  • 解决3:升级模型 。从 base 升级到 small medium ,精度提升显著,但代价是速度变慢和内存占用增加。

6.2 Ollama 与 LLM 相关问题

问题:Ollama 服务启动失败或连接被拒绝。

  • 解决 :首先在终端直接运行 ollama serve 查看输出。常见原因是端口 11434 被占用。可以修改 Ollama 的配置(环境变量 OLLAMA_HOST )换一个端口,例如 OLLAMA_HOST=127.0.0.1:11435 ,然后代码中的 base_url 也要相应修改。
  • 检查模型是否已下载 :运行 ollama list 确认你调用的模型(如 llama3:8b )存在。如果不存在,用 ollama pull 拉取。

问题:LLM 响应慢或卡住。

  • 排查1:查看系统资源 。运行 ollama ps 查看模型运行状态和资源占用。可能是内存不足导致交换(swapping),这会极慢。考虑换用更小的模型。
  • 排查2:调整生成参数 。降低 max_tokens (比如从 500 降到 200),设置 temperature=0 来获得更确定、更快的回答。
  • 排查3:使用流式响应 。虽然整体时间可能一样,但流式 ( stream=True ) 能让用户先看到部分输出,感知上更快。

问题:对话历史混乱,LLM 忘记之前的内容。

  • 解决 :这是上下文管理的问题。首先确认你的 conversation_history 列表确实在更新。其次,检查发送给 Ollama 的 prompt 是否确实包含了历史信息。最可能的原因是上下文长度超限,模型“忘记”了开头的内容。需要实现前面提到的 历史摘要 或更严格的 滑动窗口 机制。

6.3 集成与逻辑问题

问题:系统指令误触发。 比如用户说“不要退出”,结果触发了“退出”指令。

  • 解决 :简单的关键词匹配 ( if cmd_keyword in user_text: ) 非常粗糙。可以改为 精确匹配 前缀匹配 。例如,只当用户输入 完全等于 “退出”或 “退出” 开头 时才触发。更好的办法是训练一个简单的本地文本分类模型(或用一个小型 LLM)来做意图识别,但这会引入复杂度。

问题:代理无法执行本地操作(如打开文件、搜索网页)。

  • 解决 :这就是需要实现 函数调用(Function Calling) 的地方。从简单的开始,比如定义一个 open_file(file_path) 函数,用 Python 的 os.startfile (Windows) 或 subprocess.run(['open', file_path]) (macOS) 实现。然后在指令路由环节,如果用户说“打开我的简历”,先用 LLM 提取出文件路径(可能需要结合你的文件系统索引),再调用该函数。

6.4 性能与资源优化

问题:CPU 占用率 100%,风扇狂转。

  • 解决 :这是本地运行大模型的常态。优化方向:
    1. 模型量化 :使用 q4_0 , q5_1 等量化版本的模型,能大幅减少内存占用和计算量。
    2. 使用更小的模型 Phi-3-mini (3.8B) 在许多任务上表现接近 7B 模型,但速度快得多。
    3. 硬件升级 :如果条件允许,增加内存,使用带 GPU 的机器(即使是一张消费级的 RTX 4060),体验会有质的飞跃。确保 Ollama 能识别到 GPU(运行 ollama run llama3:8b 时看输出日志)。

问题:第一次运行 Whisper 或加载新模型特别慢。

  • 解决 :这是正常的。Whisper 第一次加载某个尺寸的模型时需要从 Hugging Face 下载,模型文件较大( base 约 150MB)。Ollama 第一次运行某个模型也需要从仓库下载。确保网络通畅,且磁盘空间足够。后续运行会直接加载本地缓存,速度很快。

搭建这样一个本地语音AI代理的过程,就像在组装一个乐高机器人。Whisper 是它的耳朵,Ollama 是它的大脑,而你的代码则是它的神经系统和运动指令。从最简单的语音问答开始,逐步为它添加“手臂”(函数调用)和“记忆”(上下文管理),看着它从一个简单的复读机成长为一个能真正帮你处理事务的助手,这种成就感是使用现成云服务无法比拟的。最关键的是,整个系统的控制权完全在你手中,你可以定制它的能力,调整它的性格,而无需担心你的数据去了哪里。这或许就是本地AI最吸引人的地方。

Logo

免费领 100 小时云算力,进群参与显卡、AI PC 幸运抽奖

更多推荐