LangChain 系列 8：Gradio 聊天界面与本地语音输入

上一章完成了一个可以保存多轮历史的聊天机器人，但它只能在终端中运行。普通用户更习惯在网页里输入消息，也希望直接使用麦克风说话。

本文为聊天机器人增加 Gradio 页面，并使用 MLX Whisper 在 Apple Silicon 上完成本地语音识别。语音识别不会直接代替大模型，它负责把声音转换成文字，再把文字交给上一章的 LangChain 聊天链。

1. 语音聊天的数据流

图中同时画出了文字和语音两条输入路径。它们最终都会进入同一个 ask_llm() 函数，因此模型调用、会话隔离和历史存储不需要各写一套。

一次语音问答会经过五个阶段：

1. Gradio 接收麦克风录音或上传的音频文件。
2. Gradio 把临时音频文件路径交给 MLX Whisper。
3. MLX Whisper 把音频转换成文字，并把识别结果显示在页面中。
4. LangChain 根据 session_id 读取 SQLite 历史，再把识别文本和历史消息一起交给 Qwen3。
5. Gradio 在聊天窗口中显示识别文本和模型回答，LangChain 同时把本轮消息写回 SQLite。

文字输入会跳过 Whisper，直接从第 4 步开始执行。

因此，本文实际上使用了两个模型：

• Whisper 负责语音转文字，不负责回答问题。
• Qwen3 负责理解文字并生成回答，不直接读取音频。

2. 安装运行环境

代码位于：

source/_posts/llm_learning/langchain/p08_gradio_voice_chatbot/

激活 Python 3.12 虚拟环境并安装依赖：

cd ~/Documents/git/blog/source/_posts/llm_learning
source .venv/bin/activate
python -m pip install --retries 10 --timeout 120 -r requirements.txt

新增依赖为：

gradio==5.49.1
mlx-whisper==0.4.3

mlx-whisper 还会安装 Torch、Numba 和 SciPy 等依赖，所以下载量明显大于普通 LangChain 组件。

音频解码需要 FFmpeg。macOS 可以使用 Homebrew 安装：

brew install ffmpeg

检查：

ffmpeg -version

3. 先完成本地语音识别

3.1 准备测试音频

为了让测试结果可重复，可以使用 macOS 自带的 say 生成一段中文语音，再转换成 16 kHz 单声道 WAV：

mkdir -p data

say -v Tingting \
  -o /tmp/llm_learning_sample.aiff \
  '你好，我叫小明，我今天想去杭州西湖散步。'

ffmpeg -y \
  -i /tmp/llm_learning_sample.aiff \
  -ar 16000 \
  -ac 1 \
  data/sample_voice.wav

3.2 下载语音模型并识别

02_transcribe_audio.py 的完整代码如下：

# 这个文件演示如何使用 MLX Whisper 把一段本地语音转换成文字。
# 默认读取 llm_learning/data/sample_voice.wav，也可以在命令行传入其他音频路径。

import argparse
import os
from pathlib import Path

import mlx_whisper
from modelscope import snapshot_download


project_root = Path(__file__).resolve().parents[2]
default_audio = project_root / "data" / "sample_voice.wav"
voice_model_dir = project_root / "voice_model"
voice_model_target = Path.home() / "Documents/git/llm/whisper-small-mlx/model"
voice_model_revision = "31153328991c71b2a551fd9279cee142332f0b2e"

# ModelScope 在国内可以直接访问，避免大文件下载经过系统 VPN。
os.environ.setdefault("NO_PROXY", "*")
os.environ.setdefault("no_proxy", "*")

parser = argparse.ArgumentParser(description="使用 MLX Whisper 转写语音")
parser.add_argument("audio", nargs="?", default=str(default_audio))
args = parser.parse_args()

audio_path = Path(args.audio).expanduser().resolve()
if not audio_path.exists():
    raise FileNotFoundError(f"没有找到音频文件：{audio_path}")

# 本地没有模型时，先通过 ModelScope 下载；中断后再次运行可以续传。
if not (voice_model_dir / "weights.npz").exists():
    voice_model_target.mkdir(parents=True, exist_ok=True)
    snapshot_download(
        model_id="mlx-community/whisper-small-mlx",
        revision=voice_model_revision,
        local_dir=str(voice_model_target),
    )
    if voice_model_dir.is_symlink():
        voice_model_dir.unlink()
    if not voice_model_dir.exists():
        voice_model_dir.symlink_to(voice_model_target, target_is_directory=True)

result = mlx_whisper.transcribe(
    str(audio_path),
    path_or_hf_repo=str(voice_model_dir),
    language="zh",
)

print("音频文件：", audio_path)
print("语音模型：", voice_model_dir)
print("识别结果：", result["text"].strip())

这里使用 mlx-community/whisper-small-mlx。small 模型比 tiny 更大、占用资源更多；在本文的固定中文音频中，它能够正确识别姓名、城市、景点和动作，因此继续用它完成后续页面测试。

运行：

python langchain/p08_gradio_voice_chatbot/02_transcribe_audio.py

真实输出：

音频文件： .../llm_learning/data/sample_voice.wav
语音模型： .../llm_learning/voice_model
识别结果： 你好,我叫小明,我今天想去杭州西湖散步。

模型正确识别了姓名、城市、景点和动作。

4. 使用 Gradio 创建文本聊天页面

Gradio 可以直接用 Python 描述一个本地网页。最主要的组件有：

• gr.Blocks：页面容器。
• gr.Chatbot：显示用户消息和模型回答。
• gr.Textbox：文字输入框。
• gr.Button：发送和清空按钮。
• gr.State：保存当前浏览器会话的临时状态。

页面中的聊天函数仍然调用上一章的 LangChain 链：

def chat(message: str, history: list[dict], session_id: str):
    """接收页面输入，调用 LangChain，并返回更新后的页面消息。"""
    if not isinstance(session_id, str) or not session_id:
        session_id = str(uuid4())

    if not message.strip():
        return "", history, session_id

    config = {"configurable": {"session_id": session_id}}
    response = chatbot_chain.invoke({"input": message}, config=config)
    history = history + [
        {"role": "user", "content": message},
        {"role": "assistant", "content": response.content},
    ]
    return "", history, session_id

history 是 Gradio 页面当前要显示的消息，SQLite 历史则由 RunnableWithMessageHistory 管理。两者用途不同：

• Gradio history 决定页面现在显示什么。
• LangChain 消息历史决定下一次模型调用能看到什么。

启动页面：

python langchain/p08_gradio_voice_chatbot/01_gradio_text_chatbot.py

浏览器访问：

http://127.0.0.1:7860

5. 把语音输入加入聊天页面

5.1 语音处理函数

def send_voice(audio_path: str | None, history: list[dict], session_id: str):
    """先把录音转成文字，再把识别结果交给聊天链。"""
    if not isinstance(session_id, str) or not session_id:
        session_id = str(uuid4())
    if not audio_path:
        return history, None, "请先录音或上传音频。", session_id

    result = mlx_whisper.transcribe(
        audio_path,
        path_or_hf_repo=str(voice_model_dir),
        language="zh",
    )
    transcript = result["text"].strip()
    updated_history, session_id = ask_llm(transcript, history, session_id)
    return updated_history, None, f"识别结果：{transcript}", session_id

函数的返回值依次更新：

1. 聊天窗口。
2. 音频输入组件。
3. 语音识别结果文本框。
4. 当前浏览器的会话 ID 状态。

5.2 Gradio 页面结构

with gr.Blocks(title="本地语音聊天机器人") as demo:
    gr.Markdown("## 本地语音聊天机器人")
    session_id = gr.State(lambda: str(uuid4()))
    chatbot = gr.Chatbot(type="messages", height=460)

    with gr.Tab("文字"):
        message = gr.Textbox(label="消息", placeholder="请输入一条消息")
        send_button = gr.Button("发送", variant="primary")

    with gr.Tab("语音"):
        audio = gr.Audio(
            sources=["microphone", "upload"],
            type="filepath",
            label="语音",
        )
        voice_button = gr.Button("识别并发送", variant="primary")
        transcript = gr.Textbox(label="语音识别结果", interactive=False)

    clear_button = gr.Button("清空对话")

    message.submit(
        send_text,
        [message, chatbot, session_id],
        [message, chatbot, session_id],
    )
    send_button.click(
        send_text,
        [message, chatbot, session_id],
        [message, chatbot, session_id],
    )
    voice_button.click(
        send_voice,
        [audio, chatbot, session_id],
        [chatbot, audio, transcript, session_id],
    )

type=”filepath” 表示 Gradio 将录音保存成临时文件，并把文件路径传给 Python 函数。Whisper 可以直接读取这个路径。

启动前先确认 Qwen3 服务正在运行：

"$VIRTUAL_ENV/bin/python" -m mlx_lm server \
  --model model \
  --host 127.0.0.1 \
  --port 18080 \
  --chat-template-args '{"enable_thinking": false}'

另开终端确认服务已经就绪：

curl --noproxy '*' http://127.0.0.1:18080/v1/models

再启动语音页面：

python langchain/p08_gradio_voice_chatbot/03_gradio_voice_chatbot.py

访问：

http://127.0.0.1:7861

切换到“语音”标签页后，可以看到录音或上传入口、“识别并发送”按钮和只读的识别结果文本框：

自动化测试浏览器没有麦克风设备，因此截图中的设备列表显示 No microphone found；上传本地音频仍可正常验证完整流程。实际使用麦克风时，还需要在浏览器中授予录音权限。

6. 真实语音问答结果

将前面生成的 WAV 音频交给页面语音处理函数，真实结果为：

语音识别状态：识别结果：你好,我叫小明,我今天想去杭州西湖散步。
用户消息：你好,我叫小明,我今天想去杭州西湖散步。
模型回答：你好，小明！西湖是个很美的地方，适合散步。你可以沿着湖边走走，欣赏湖光山色，还可以看看三潭印月、雷峰塔这些景点。记得带上相机，拍下美丽的风景哦！祝你玩得开心！

这次调用可以拆成两个真实结果：

• Whisper 正确完成语音转文字。
• Qwen3 根据识别结果生成了与杭州西湖相关的回答。

7. 浏览器会话为什么需要独立 ID

如果把会话 ID 固定写成：

session_id = "user123"

所有访问页面的人都会读写同一份聊天历史。本文使用：

session_id = gr.State(lambda: str(uuid4()))

每个页面会话会得到独立 UUID，并作为 LangChain 的 session_id。同一个页面中的文字输入和语音输入会共用这个 ID，因此两种输入方式看到的是同一段对话历史；另一个标签页则会得到新的 ID，不会读取前一个页面的会话。

这里需要区分两种状态：

• gr.State 只在当前 Gradio 页面会话中保存 UUID。
• SQLChatMessageHistory 根据 UUID 读取和写入 SQLite 中的模型历史。

gr.State 不是登录或身份认证机制。刷新页面或重新打开页面可能得到新的 UUID，当前示例也没有提供“输入旧会话 ID 并恢复页面消息”的功能。如果需要稳定绑定真实用户，应在登录系统中取得用户或会话标识，而不是把随机 UUID 当成账户 ID。

代码中还增加了类型判断：

if not isinstance(session_id, str) or not session_id:
    session_id = str(uuid4())

这是因为直接通过 Gradio Client 调用接口时，页面初始化事件可能尚未执行。增加兜底后，网页调用和 API 调用都不会把函数对象错误地写入 SQLite 查询参数。

8. 本章完整代码

下面的代码与 llm_learning 项目中的脚本保持一致。

01_gradio_text_chatbot.py

# 这个文件把有记忆的文本聊天机器人放进 Gradio 页面。
# 运行前需要先在项目根目录启动 Qwen3 的 mlx_lm.server。

from pathlib import Path
from uuid import uuid4

import gradio as gr
import httpx
from langchain_community.chat_message_histories import SQLChatMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_openai import ChatOpenAI


project_root = Path(__file__).resolve().parents[2]
database_path = project_root / "data" / "chat_history.db"
database_path.parent.mkdir(parents=True, exist_ok=True)
connection = f"sqlite:///{database_path}"

llm = ChatOpenAI(
    base_url="http://127.0.0.1:18080/v1",
    api_key="not-needed",
    model="default_model",
    temperature=0,
    max_tokens=256,
    http_client=httpx.Client(trust_env=False),
)

prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "你是一个友好的中文聊天助手，请简洁回答。"),
        MessagesPlaceholder(variable_name="history"),
        ("human", "{input}"),
    ]
)


def get_session_history(session_id: str) -> SQLChatMessageHistory:
    """每个浏览器会话使用独立的 SQLite 历史。"""
    return SQLChatMessageHistory(session_id=session_id, connection=connection)


chatbot_chain = RunnableWithMessageHistory(
    prompt | llm,
    get_session_history,
    input_messages_key="input",
    history_messages_key="history",
)


def chat(message: str, history: list[dict], session_id: str):
    """接收页面输入，调用 LangChain，并返回更新后的页面消息。"""
    if not isinstance(session_id, str) or not session_id:
        session_id = str(uuid4())

    if not message.strip():
        return "", history, session_id

    config = {"configurable": {"session_id": session_id}}
    response = chatbot_chain.invoke({"input": message}, config=config)
    history = history + [
        {"role": "user", "content": message},
        {"role": "assistant", "content": response.content},
    ]
    return "", history, session_id


def clear_chat(session_id: str):
    """同时清空页面和当前会话的数据库记录。"""
    if not isinstance(session_id, str) or not session_id:
        session_id = str(uuid4())
    get_session_history(session_id).clear()
    return [], "", session_id


with gr.Blocks(title="本地 Qwen3 聊天机器人") as demo:
    gr.Markdown("## 本地 Qwen3 聊天机器人")
    session_id = gr.State(lambda: str(uuid4()))
    chatbot = gr.Chatbot(type="messages", height=480)
    message = gr.Textbox(label="消息", placeholder="请输入一条消息")
    with gr.Row():
        send_button = gr.Button("发送", variant="primary")
        clear_button = gr.Button("清空")

    message.submit(
        chat,
        [message, chatbot, session_id],
        [message, chatbot, session_id],
    )
    send_button.click(
        chat,
        [message, chatbot, session_id],
        [message, chatbot, session_id],
    )
    clear_button.click(clear_chat, session_id, [chatbot, message, session_id])


if __name__ == "__main__":
    demo.launch(server_name="127.0.0.1", server_port=7860)

02_transcribe_audio.py

# 这个文件演示如何使用 MLX Whisper 把一段本地语音转换成文字。
# 默认读取 llm_learning/data/sample_voice.wav，也可以在命令行传入其他音频路径。

import argparse
import os
from pathlib import Path

import mlx_whisper
from modelscope import snapshot_download


project_root = Path(__file__).resolve().parents[2]
default_audio = project_root / "data" / "sample_voice.wav"
voice_model_dir = project_root / "voice_model"
voice_model_target = Path.home() / "Documents/git/llm/whisper-small-mlx/model"
voice_model_revision = "31153328991c71b2a551fd9279cee142332f0b2e"

# ModelScope 在国内可以直接访问，避免大文件下载经过系统 VPN。
os.environ.setdefault("NO_PROXY", "*")
os.environ.setdefault("no_proxy", "*")

parser = argparse.ArgumentParser(description="使用 MLX Whisper 转写语音")
parser.add_argument("audio", nargs="?", default=str(default_audio))
args = parser.parse_args()

audio_path = Path(args.audio).expanduser().resolve()
if not audio_path.exists():
    raise FileNotFoundError(f"没有找到音频文件：{audio_path}")

# 本地没有模型时，先通过 ModelScope 下载；中断后再次运行可以续传。
if not (voice_model_dir / "weights.npz").exists():
    voice_model_target.mkdir(parents=True, exist_ok=True)
    snapshot_download(
        model_id="mlx-community/whisper-small-mlx",
        revision=voice_model_revision,
        local_dir=str(voice_model_target),
    )
    if voice_model_dir.is_symlink():
        voice_model_dir.unlink()
    if not voice_model_dir.exists():
        voice_model_dir.symlink_to(voice_model_target, target_is_directory=True)

result = mlx_whisper.transcribe(
    str(audio_path),
    path_or_hf_repo=str(voice_model_dir),
    language="zh",
)

print("音频文件：", audio_path)
print("语音模型：", voice_model_dir)
print("识别结果：", result["text"].strip())

03_gradio_voice_chatbot.py

# 这个文件把文本聊天和本地语音识别放进同一个 Gradio 页面。
# 文字和语音最终都会转换成文本，再交给本地 Qwen3 回答。

import os
from pathlib import Path
from uuid import uuid4

import gradio as gr
import httpx
import mlx_whisper
from langchain_community.chat_message_histories import SQLChatMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_openai import ChatOpenAI
from modelscope import snapshot_download


project_root = Path(__file__).resolve().parents[2]
database_path = project_root / "data" / "chat_history.db"
database_path.parent.mkdir(parents=True, exist_ok=True)
connection = f"sqlite:///{database_path}"
voice_model_dir = project_root / "voice_model"
voice_model_target = Path.home() / "Documents/git/llm/whisper-small-mlx/model"
voice_model_revision = "31153328991c71b2a551fd9279cee142332f0b2e"

# 下载 ModelScope 权重时不经过系统 VPN，防止大文件连接频繁中断。
os.environ.setdefault("NO_PROXY", "*")
os.environ.setdefault("no_proxy", "*")

# 页面启动时确认本地语音模型存在，不存在时通过 ModelScope 下载。
if not (voice_model_dir / "weights.npz").exists():
    voice_model_target.mkdir(parents=True, exist_ok=True)
    snapshot_download(
        model_id="mlx-community/whisper-small-mlx",
        revision=voice_model_revision,
        local_dir=str(voice_model_target),
    )
    if voice_model_dir.is_symlink():
        voice_model_dir.unlink()
    if not voice_model_dir.exists():
        voice_model_dir.symlink_to(voice_model_target, target_is_directory=True)

llm = ChatOpenAI(
    base_url="http://127.0.0.1:18080/v1",
    api_key="not-needed",
    model="default_model",
    temperature=0,
    max_tokens=256,
    http_client=httpx.Client(trust_env=False),
)

prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "你是一个友好的中文聊天助手，请简洁回答。"),
        MessagesPlaceholder(variable_name="history"),
        ("human", "{input}"),
    ]
)


def get_session_history(session_id: str) -> SQLChatMessageHistory:
    """取得当前页面会话对应的 SQLite 历史。"""
    return SQLChatMessageHistory(session_id=session_id, connection=connection)


chatbot_chain = RunnableWithMessageHistory(
    prompt | llm,
    get_session_history,
    input_messages_key="input",
    history_messages_key="history",
)


def ask_llm(text: str, history: list[dict], session_id: str):
    """把文本交给有历史记录的 LangChain 聊天链。"""
    if not isinstance(session_id, str) or not session_id:
        session_id = str(uuid4())

    if not text.strip():
        return history, session_id

    config = {"configurable": {"session_id": session_id}}
    response = chatbot_chain.invoke({"input": text}, config=config)
    updated_history = history + [
        {"role": "user", "content": text},
        {"role": "assistant", "content": response.content},
    ]
    return updated_history, session_id


def send_text(text: str, history: list[dict], session_id: str):
    """处理文本输入。"""
    updated_history, session_id = ask_llm(text, history, session_id)
    return "", updated_history, session_id


def send_voice(audio_path: str | None, history: list[dict], session_id: str):
    """先把录音转成文字，再把识别结果交给聊天链。"""
    if not isinstance(session_id, str) or not session_id:
        session_id = str(uuid4())
    if not audio_path:
        return history, None, "请先录音或上传音频。", session_id

    result = mlx_whisper.transcribe(
        audio_path,
        path_or_hf_repo=str(voice_model_dir),
        language="zh",
    )
    transcript = result["text"].strip()
    updated_history, session_id = ask_llm(transcript, history, session_id)
    return updated_history, None, f"识别结果：{transcript}", session_id


def clear_chat(session_id: str):
    """清空当前页面和对应数据库历史。"""
    if not isinstance(session_id, str) or not session_id:
        session_id = str(uuid4())
    get_session_history(session_id).clear()
    return [], "", None, "", session_id


with gr.Blocks(title="本地语音聊天机器人") as demo:
    gr.Markdown("## 本地语音聊天机器人")
    session_id = gr.State(lambda: str(uuid4()))
    chatbot = gr.Chatbot(type="messages", height=460)

    with gr.Tab("文字"):
        message = gr.Textbox(label="消息", placeholder="请输入一条消息")
        send_button = gr.Button("发送", variant="primary")

    with gr.Tab("语音"):
        audio = gr.Audio(sources=["microphone", "upload"], type="filepath", label="语音")
        voice_button = gr.Button("识别并发送", variant="primary")
        transcript = gr.Textbox(label="语音识别结果", interactive=False)

    clear_button = gr.Button("清空对话")

    message.submit(
        send_text,
        [message, chatbot, session_id],
        [message, chatbot, session_id],
    )
    send_button.click(
        send_text,
        [message, chatbot, session_id],
        [message, chatbot, session_id],
    )
    voice_button.click(
        send_voice,
        [audio, chatbot, session_id],
        [chatbot, audio, transcript, session_id],
    )
    clear_button.click(
        clear_chat,
        session_id,
        [chatbot, message, audio, transcript, session_id],
    )


if __name__ == "__main__":
    demo.launch(server_name="127.0.0.1", server_port=7861)

9. 常见问题

9.1 pip 下载 Torch 时中断

首次安装实测遇到：

IncompleteRead: 58512192 bytes read, 52700853 more expected

这是网络下载中断，不是 Python 版本或依赖代码错误。重新执行并增加重试：

python -m pip install \
  --retries 10 \
  --timeout 120 \
  -r requirements.txt

pip 会复用已经缓存的其他安装包。

9.2 Hugging Face 下载权重时 TLS 断开

实测当前 VPN 环境下载 weights.npz 时出现：

SSL: UNEXPECTED_EOF_WHILE_READING

因此代码改用 ModelScope 下载相同的 mlx-community/whisper-small-mlx。ModelScope SDK 下载中断时会保留进度，再次运行脚本即可续传。

9.3 页面提示没有麦克风

先检查浏览器是否获得麦克风权限。若只是验证程序，也可以直接上传 WAV、MP3、M4A 或 FLAC 文件，不需要使用麦克风。

9.4 语音能识别，但模型没有回答

Whisper 和 Qwen3 是两个独立环节。先确认终端已经启动 mlx_lm.server，再访问：

http://127.0.0.1:18080/v1/models

/v1/models 返回成功只能证明服务进程能够响应并列出模型，不能证明聊天推理一定成功。如果语音识别结果已经显示，而聊天回答失败，应继续检查 Qwen3 服务终端中的 /v1/chat/completions 请求日志和模型加载错误，而不是重新下载 Whisper。

代码中的 httpx.Client(trust_env=False) 只负责防止本机请求误走 VPN 或系统代理，不能修复模型加载失败、内存不足或服务端推理异常。

9.5 页面在手机宽度下是否可用

本文实际检查了桌面视口和 390 × 844 移动视口。Gradio 会把输入组件自动调整为单列，聊天窗口、标签、输入框和按钮没有发生重叠。

10. 小结

这一章没有让 Qwen3 直接处理声音，而是建立了一条清晰的本地处理链：

音频 -> MLX Whisper -> 文字 -> LangChain -> Qwen3 -> Gradio

组件	输入	输出	作用
gr.Audio	麦克风或音频文件	本地临时路径	接收语音
MLX Whisper	音频路径	识别文本	本地语音转文字
gr.State	页面初始化	UUID 字符串	隔离浏览器会话
LangChain 聊天链	识别文本和历史消息	AIMessage	调用本地 Qwen3
gr.Chatbot	用户文本和模型回答	聊天页面	显示多轮对话