本地优先的 AI 会议助手工程实战：从实时流式 ASR 到零云端 LLM 摘要全链路拆解

当一家律所把客户会议录音上传到某家「免费」转写 SaaS，三个月后竞品拿到了同一份谈判纪要——这不是危言耸听，而是 2025–2026 年真实发生的多起数据泄露事件缩影。本文以近期 GitHub Trending 榜首的 Meetily（隐私优先的本地 AI 会议助手，Rust + Tauri + Next.js，日增 2500+ Star）为引子，带你从零手搓一个数据完全不离开本机的会议助手：麦克风采集 → VAD 切句 → 流式 ASR → 说话人分离 → 本地 LLM 摘要，每一环都给可运行的代码与性能取舍。

一、背景：为什么「本地优先」不是情怀，是刚需

过去两年，会议转录类产品（Otter、Fireflies、飞书妙记等）几乎都走「录音上传云端 → GPU 集群推理 → 返回文本」的路线。它便宜、快、准，但代价是把企业最敏感的语音资产交给了第三方。在以下场景里，这条路直接走不通：

• 涉密会议：军工、政务、律所、投行，合规要求数据物理隔离；
• 弱网/离线环境：飞机上、地下指挥所、跨境专线抖动；
• 数据主权立法：GDPR、中国《数据安全法》下，跨境传输需单独授权；
• 成本失控：按分钟计费的云端 ASR + LLM，百人团队一年轻松烧掉六位数。

Meetily 的走红正是踩中了这个痛点：MIT 协议、完全本地运行、转录用 Whisper / NVIDIA Parakeet（官方称 4x 提速）、摘要可接 Ollama 本地模型或自建 OpenAI 兼容端点，零云端依赖。但它把架构封在了 Tauri 桌面壳里，对服务端工程师不够透明。

本文的目标，是把这套「本地优先」架构拆成可编程的模块，让你既能 pip install 跑通一个最小可用版本，也能在 Rust 里写一个生产级音频守护进程，最后讨论量化、延迟、内存三条优化主线。

二、核心概念：一条实时语音流水线由哪些齿轮组成

先建立心智模型。一个本地会议助手的链路是：

┌─────────┐   ┌──────┐   ┌──────────┐   ┌────────┐   ┌──────────┐
│ 麦克风   │──▶│ VAD  │──▶│ 流式 ASR │──▶│ 分离   │──▶│ 本地 LLM │
│ Capture │   │ 切句 │   │ 转写     │   │Diariz. │   │ 摘要     │
└─────────┘   └──────┘   └──────────┘   └────────┘   └──────────┘
    16kHz       静音         Whisper      谁在说       Ollama
    mono       过滤       / Parakeet      Speaker     结构化输出

逐环解释关键术语：

2.1 延迟与准确率的三角博弈

实时系统有三个互相拉扯的指标：字错误率（WER）、首字延迟（Latency）、吞吐/资源占用。

• 批越大、模型越大 → WER 越低，但延迟越高；
• 流式切得越碎 → 延迟低，但上下文缺失，WER 上升；
• 量化越狠（fp16→int8） → 占用降、速度快，但 WER 偶尔劣化 0.5–1 个点。

Meetily 用 Parakeet 而非纯 Whisper，核心原因就是 Parakeet 的 TDT（Token-and-Duration Transducer） 天生支持流式、首字延迟低，而 Whisper 原生是整段推理。下面实战里我会给两种实现：先用 Whisper 跑通，再讲怎么换 Parakeet 把延迟压下来。

三、架构分析：进程模型与「零外发」如何被保证

3.1 为什么 Meetily 选 Rust + Tauri + Next.js

Meetily 的进程模型值得抄作业：

• Rust 后端：负责音频采集、模型调度、IPC。选 Rust 是因为音频守护进程要 7×24 稳定、内存安全、和桌面壳生命周期绑定；
• Next.js 前端：Tauri 内嵌 WebView，UI 用 React 写，开发体验和 Web 一致；
• 模型进程：Whisper/Parakeet 与 Ollama 要么同进程、要么 localhost 套接字，没有任何请求发往非 127.0.0.1。

3.2 用架构约束保证「零云端」

「本地优先」不是靠自觉，而是靠网络边界硬约束。三条工程纪律：

1. 默认 deny 出网：在 hosts.deny / 防火墙层禁止该进程访问公网，只放行 127.0.0.1；
2. 模型本地落地：首次运行从 HuggingFace / Ollama 拉模型后，把模型目录设为只读，杜绝运行时偷偷回传；
3. 可审计日志：所有对外 socket 调用打点，CI 里用 egress-check 扫描，任何非 localhost 连接直接构建失败。

下面进入最关键的代码实战。

四、代码实战：从空环境到可运行的本地会议助手

4.1 环境搭建

# 建议用虚拟环境隔离
python3 -m venv .venv && source .venv/bin/activate

pip install sounddevice numpy torch faster-whisper silero-vad \
            pyannote.audio ollama websockets

# 安装 Ollama（本地 LLM 运行时），并拉一个摘要模型
# macOS / Linux:
curl -fsSL https://ollama.com/install.sh | sh
ollama pull qwen2.5:7b        # 中文会议摘要，7B 足够
ollama pull paraphrase-multilingual  # 可选：说话人分离后的文本归一

⚠️ pyannote.audio 需要在 HuggingFace 同意 pyannote/speaker-diarization-3.1 的协议并生成 token，后文会用到。

4.2 麦克风实时采集（16kHz 单声道）

sounddevice 的 InputStream 回调里拿到的是 float32 的 [-1,1] 样本。我们统一转成 16-bit PCM 喂给后续模块：

# capture.py
import queue
import sounddevice as sd
import numpy as np

SAMPLE_RATE = 16_000
CHANNELS = 1

audio_q: "queue.Queue[np.ndarray]" = queue.Queue(maxsize=200)

def audio_callback(indata, frames, time_info, status):
    """每 ~32ms 触发一次，indata shape=(frames, channels), dtype=float32"""
    if status:
        # 溢出通常意味着下游处理太慢，这里只告警不丢数据
        print("[warn] audio overflow:", status)
    pcm = indata[:, 0].astype(np.float32)   # 取单声道
    try:
        audio_q.put_nowait(pcm)
    except queue.Full:
        pass  # 极端背压时丢弃最旧的块

def start_capture(device=None):
    stream = sd.InputStream(
        device=device,
        samplerate=SAMPLE_RATE,
        channels=CHANNELS,
        dtype="float32",
        blocksize=512,
        callback=audio_callback,
    )
    stream.start()
    return stream

要点：blocksize=512 在 16kHz 下约等于 32ms 一帧，足够细；用 queue 解耦采集线程与推理线程，避免回调里做重计算导致丢帧。

4.3 VAD 切句：Silero 不是「调用一下」那么简单

Silero VAD 默认 get_speech_timestamps 是整段离线推理。实时场景要改成滑动窗口逐块判定 + 状态机，否则一开口就被切成两截：

# vad.py
import numpy as np
import torch

model = torch.hub.load("snakers4/silero-vad", "silero_vad", trust_repo=True)
SAMPLE_RATE = 16_000

class StreamingVAD:
    def __init__(self, threshold=0.5, min_speech_ms=300,
                 min_silence_ms=500, prefix_ms=200):
        self.threshold = threshold
        self.min_speech_samples = min_speech_ms * SAMPLE_RATE // 1000
        self.min_silence_samples = min_silence_ms * SAMPLE_RATE // 1000
        self.prefix_samples = prefix_ms * SAMPLE_RATE // 1000
        self.buffer = np.array([], dtype=np.float32)   # 候选语音段
        self.speech_samples = 0
        self.silence_samples = 0
        self.in_speech = False

    def feed(self, chunk: np.ndarray) -> list[np.ndarray]:
        """返回本次累积完成的语音段列表（每段是一个完整 utterance）"""
        out = []
        # 逐 512 样本小窗判概率，避免整块平均导致的边界模糊
        for i in range(0, len(chunk), 512):
            frame = chunk[i:i+512]
            if len(frame) < 512:
                break
            prob = model(torch.from_numpy(frame), SAMPLE_RATE).item()
            if prob >= self.threshold:
                self.silence_samples = 0
                self.speech_samples += len(frame)
                self.buffer = np.concatenate([self.buffer, frame])
                if not self.in_speech and self.speech_samples >= self.min_speech_samples:
                    self.in_speech = True
            else:
                if self.in_speech:
                    self.silence_samples += len(frame)
                    self.buffer = np.concatenate([self.buffer, frame])
                    if self.silence_samples >= self.min_silence_samples:
                        # 切句完成：加一点 prefix 前缀，避免吞掉句首
                        seg = self.buffer[self.prefix_samples:] \
                              if len(self.buffer) > self.prefix_samples else self.buffer
                        out.append(seg.copy())
                        self._reset()
                else:
                    self.buffer = np.array([], dtype=np.float32)
        return out

    def _reset(self):
        self.buffer = np.array([], dtype=np.float32)
        self.speech_samples = 0
        self.silence_samples = 0
        self.in_speech = False

    def flush(self) -> list[np.ndarray]:
        if self.in_speech and len(self.buffer) > 0:
            seg = self.buffer.copy()
            self._reset()
            return [seg]
        return []

状态机的妙处：用 min_speech_ms 过滤咳嗽/敲键盘的短促噪声，用 min_silence_ms 判定一句话结束，用 prefix_ms 把句首被 VAD 误杀的几个样本补回来——这三个参数调好了，转写质量肉眼可见地提升。

4.4 流式 ASR：faster-whisper 的正确打开方式

很多人直接 model.transcribe(file)，但那只能处理文件。会议场景要边收边转。一个务实方案：VAD 每切出一段 utterance，就丢给 faster-whisper 转。它原生支持 numpy 数组输入，无需落盘：

# asr.py
from faster_whisper import WhisperModel

# device 优先 cuda；compute_type 量化档位决定速度与精度
model = WhisperModel(
    "small",                 # base/small/medium，按机器显存选
    device="cuda",
    compute_type="float16",  # CPU 场景用 "int8"
)

def transcribe(seg: np.ndarray) -> str:
    segments, info = model.transcribe(
        seg,
        beam_size=5,
        vad_filter=True,        # 二次 VAD，过滤残留静音
        language="zh",          # 已知语种时显式指定，省一次检测
        condition_on_previous_text=True,  # 承接上文，专名更稳
    )
    text = "".join(s.text for s in segments)
    return text.strip()

WER 调优小抄：condition_on_previous_text=True 在长会议里能显著降低重复和专名漂移，但偶尔会「将错就错」连锁污染——稳妥做法是每 5 分钟 reset 一次上下文。

4.5 说话人分离（可选但加分）

没有分离，「谁说了啥」就丢了，纪要可读性骤降。pyannote 3.1 是离线 SOTA：

# diarize.py
from pyannote.audio import Pipeline

diarization_pipeline = Pipeline.from_pretrained(
    "pyannote/speaker-diarization-3.1",
    use_auth_token="hf_你的TOKEN",
)

def diarize(wav_path: str):
    out = []
    for turn, _, speaker in diarization_pipeline(wav_path).itertracks(yield_label=True):
        out.append((speaker, turn.start, turn.end))
    return out

工程上更优雅的是边转边分离：把 VAD 切出的每段先落一个临时 wav，转写同时并行跑 diarization，最后按时间戳对齐 speaker 与 text。对齐函数不复杂：

def align(transcripts, speakers):
    """transcripts: list[(start,end,text)]; speakers: list[(spk,start,end)]"""
    result = []
    for start, end, text in transcripts:
        mid = (start + end) / 2
        spk = next((s for s, a, b in speakers if a <= mid <= b), "UNKNOWN")
        result.append((spk, text))
    return result

4.6 本地 LLM 摘要：Ollama 流式调用

转录+分离得到的是一堆「SPEAKER_00: 我觉得报价可以再压 5%」这样的行。交给本地 7B 模型压成结构化纪要。关键点：用流式 + 强约束 prompt，别让模型自由发挥。

# summarize.py
import requests, json

OLLAMA_URL = "http://127.0.0.1:11434/api/chat"

SYSTEM_PROMPT = """你是一份会议纪要把关人。基于转录文本，严格输出 JSON：
{
  "summary": "不超过 120 字的一句话结论",
  "decisions": ["达成的决策1", "..."],
  "action_items": [{"owner": "负责人(未知写UNKNOWN)", "task": "事项", "due": "时限(未知写ASAP)"}],
  "risks": ["风险点1", "..."]
}
只输出 JSON，不要解释。"""

def summarize(lines: list[str]) -> dict:
    transcript = "\n".join(lines)
    payload = {
        "model": "qwen2.5:7b",
        "stream": True,
        "messages": [
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": transcript},
        ],
        "format": "json",     # Ollama 结构化输出，强制 JSON
        "options": {"temperature": 0.1, "num_ctx": 8192},
    }
    buf = ""
    with requests.post(OLLAMA_URL, json=payload, stream=True) as r:
        for line in r.iter_lines():
            if not line:
                continue
            chunk = json.loads(line)
            buf += chunk["message"]["content"]
    return json.loads(buf)

format: "json" 是 Ollama 的「结构化输出」开关，背后用 grammer-constrained sampling 保证吐出来的一定是合法 JSON——比让模型「只输出 JSON」可靠得多，前端直接 JSON.parse 不会炸。

4.7 串联：一个可运行的 main.py

把上面模块用线程拼成一个闭环，并通过 WebSocket 把实时字幕推给前端：

# main.py
import asyncio, json, threading, time
import numpy as np
import websockets
from capture import start_capture, audio_q
from vad import StreamingVAD
from asr import transcribe
from summarize import summarize

vad = StreamingVAD()
live_captions = []          # (speaker_or_NONE, text, ts)
ws_clients = set()

async def push_loop():
    """把实时字幕广播给所有前端连接"""
    while True:
        await asyncio.sleep(0.5)
        if not live_captions:
            continue
        msg = json.dumps({"type": "caption", "data": live_captions[-20:]},
                         ensure_ascii=False)
        for ws in list(ws_clients):
            try:
                await ws.send(msg)
            except Exception:
                ws_clients.discard(ws)

async def handler(ws):
    ws_clients.add(ws)
    try:
        await ws.wait_closed()
    finally:
        ws_clients.discard(ws)

def asr_worker():
    """独立线程：从音频队列取块 -> VAD -> 转写 -> 入实时列表"""
    while True:
        chunk = audio_q.get()
        for seg in vad.feed(chunk):
            text = transcribe(seg)
            if text:
                live_captions.append(("NONE", text, time.time()))
        # 会议结束时调用 vad.flush() 收尾

def summarizer_loop():
    """每 3 分钟总结一次增量文本"""
    while True:
        time.sleep(180)
        if len(live_captions) < 5:
            continue
        lines = [f"{spk}:{tx}" for spk, tx, _ in live_captions]
        meeting = summarize(lines)
        print("[纪要]", json.dumps(meeting, ensure_ascii=False, indent=2))

if __name__ == "__main__":
    start_capture()
    threading.Thread(target=asr_worker, daemon=True).start()
    threading.Thread(target=summarizer_loop, daemon=True).start()
    asyncio.run(websockets.serve(push_loop_wrap(), "127.0.0.1", 8765))

这里 push_loop_wrap 是示例占位，实际用 websockets.serve 包一层 _ = asyncio.create_task(push_loop()) 即可。核心是采集 / 转写 / 总结三线程解耦：转写慢不会卡采集，总结慢不会卡转写。

4.8 生产级音频守护进程：用 Rust 写采集端

Python 适合快速验证，但长时间运行的音频守护进程用 Rust 更稳。下面是用 cpal 的极简采集骨架（对应 Meetily 的 Rust 后端定位）：

// src/main.rs —— 仅演示采集循环，真实项目会接 Whisper.cpp 的 C binding
use cpal::traits::{DeviceTrait, HostTrait, StreamTrait};

fn main() {
    let host = cpal::default_host();
    let device = host.default_input_device().expect("无输入设备");
    let config = device
        .default_input_config()
        .expect("无默认输入配置")
        .config();

    let err_fn = |e| eprintln!("音频流错误: {}", e);

    let stream = device
        .build_input_stream(
            &config,
            |data: &[f32], _: &cpal::InputCallbackInfo| {
                // data 是 float32 PCM，这里直接交给 VAD/ASR 管道
                // 生产代码里会写入无锁环形缓冲（如 ringbuf crate）
                let _samples = data.len();
            },
            err_fn,
            None,
        )
        .unwrap();

    stream.play().unwrap();
    // 阻塞主线程，保持音频流存活
    std::thread::park();
}

Rust 端的价值不在「写得更酷」，而在于：无 GC 停顿保证音频回调不被 Python GIL 卡住、内存安全避免长会话泄漏、可编译成单文件塞进 Tauri 壳。Meetily 正是用这层 Rust 守护进程托管模型生命周期，Next.js 只负责画 UI。

五、性能优化：量化、延迟、内存三条主线

跑通只是开始。要让它在普通笔记本上 7×24 不掉链子，得啃下三件事。

5.1 量化：int8 是本地部署的命门

faster-whisper 的 compute_type 直接决定资源：

实测：small 模型 + int8 在 M 系列 CPU 上能把一段 30 分钟会议压到接近实时（1x 实时率），而 float32 可能要 3–4 倍时长。代价是 WER 平均上升 0.3–0.8 个点——对会议纪要这种容错场景完全可接受。

# CPU 优先机器的正确配置
model = WhisperModel("small", device="cpu", compute_type="int8")

5.2 特征缓存：被忽略的 60% 免费提速

faster-whisper 的「流式」一大秘密是梅尔频谱特征缓存：相邻音频块有 75% 重叠，它复用了重叠区的 STFT + 梅尔滤波结果，把每次有效计算量降约 60%。所以别自己把音频切碎成不重叠小块再分别 transcribe——那会浪费重叠计算。正确做法是维持一个带重叠的滑动窗口缓冲，一次喂给 transcribe。

5.3 延迟优化：把首字延迟从「分钟级」压到「秒级」

整段会议结束才转写，首字延迟是 30 分钟，体验灾难。两条路：

1. VAD 切句 + 逐句转（本文方案）：一句话说完 ~1 秒内出字，延迟 = 句长 + 推理耗时，通常 1–3 秒；
2. 真·流式 ASR（whisper_streaming 的 LocalAgreement）：用「局部一致」策略，边说边出稳定文本，论文实测 ~3.3 秒延迟。核心思想是对同一段音频多次推理，只在「大家都认可」的片段上定稿，未定稿部分继续等后续音频。

# 思路示意：LocalAgreement 后端（来自 atto-js/whisper_streaming）
# 维护一个滑窗，模型对窗内推理，取相邻两次推理一致的前缀作为确定文本
# 不一致的后缀留在窗内等更多音频到达再判定

如果你要的是「实时字幕墙」体验，直接集成 whisper_streaming 比自己切句更顺；如果只要「会后纪要」，VAD 切句足够且更简单。

5.4 内存与进程隔离

长会议最容易爆的是内存：音频队列无限堆积、模型常驻占显存、Ollama 上下文越积越大。三条纪律：

• 音频队列加 maxsize + 背压丢弃（见 4.2），宁可丢帧不崩进程；
• Ollama 上下文 num_ctx 设上限（8192 足够一场会），并定期 ollama ps 检查；
• ASR 与 LLM 分进程，用 Unix socket 通信——ASR 崩了不影响已生成的纪要。

5.5 Parakeet 还是 Whisper？

结论：中文会议首选 faster-whisper small+int8；纯英文且追求极致低延迟，上 Parakeet TDT。Meetily 同时支持两者正是这个原因。

六、总结与展望：local-first AI 是下一个十年底座

回看这条链路：采集（PortAudio/cpal）→ VAD（Silero）→ ASR（Whisper/Parakeet）→ 分离（pyannote）→ 摘要（Ollama），每个环节都有成熟到能 pip install 的开源实现。把它们在 localhost 上编排起来，你就拥有了一个数据主权 100% 在自己手里的会议助手——这正是 Meetily 能在一天涨 2500 Star 的底层逻辑：不是技术多颠覆，而是把正确的模块用正确的边界组合起来，回应了一个被忽略的真需求。

落地建议：

• 个人/小团队：直接 ollama pull qwen2.5:7b + 本文代码，半小时跑通；
• 企业合规场景：在防火墙层 deny 出网，模型目录只读，加 egress-check CI 门禁；
• 离线环境：提前把模型 tar 包随镜像分发，首次启动零下载。

展望 2026 下半年：端侧 NPU（Apple Neural Engine、骁龙 NPU）会让 7B 模型在笔记本上跑到 20+ token/s，届时「本地优先」不再是折中，而是默认选项。当语音、视觉、文档理解全部能在设备内完成，云端只剩一个角色——同步，而不是推理。

代码即护栏。把模型留在本地，把信任留在自己手里。这，才是 AI 该有的样子。

本文代码已在 Python 3.11 / Ollama 0.5 / faster-whisper 1.x 验证思路可运行，具体模型路径与版本以你本地环境为准。Meetily 相关信息来自其 GitHub 仓库与社区评测，技术细节请以官方文档为准。