Welcome to tyron's blog

常将有日思无日，莫把无时当有时

DSP与音频混音mixer

2026-07-09

David

experience

knowledge, mixer, DSP
- content
一. 音频数据相加与dsp实现

音频数据相加（常称为混音，即 Mixing）是指将两个或多个音频信号在同一时间点上的采样值进行数学叠加。此操作在数字音频处理、游戏引擎和语音识别中非常常见。

1. 核心原理

数字音频本质上是一串连续的波形采样点数据（如 PCM 数据）。将两个音频相加，即对齐它们的时间轴，并将对应位置的采样值（Amplitude）相加： $S_{mix}(t) = S_1(t) + S_2(t)$

2. 常见实现方法与算法

直接相加往往会导致数据溢出（削波失真），因此需要根据场景选择合适的算法：
- 直接求和法：将同一声道的数值简单相加。优点是声音信息保留最完整，但混合的音轨越多，越容易超过音频格式的表示范围（例如 16-bit PCM 的 +32767），导致严重的爆音。
- 归一化/平均加权法：相加后除以音轨数量（或乘以一个权重系数）。例如双轨混合： $S_{mix} = \frac{S_1 + S_2}{2}$ 优点是绝对不会溢出，音质平滑；缺点是整体音量会变小。
- 系数衰减法（非线性混音）：为了防止极端情况下的溢出并保持音量，可以采用如 $S_{mix} = S_1 + S_2 - \frac{S_1 \times S_2}{MAX_VALUE}$ 的对数压缩算法。
- 重叠相加法（Overlap-Add Method）：在频域处理音频（如短时傅里叶变换 STFT）或平滑拼接不连续点时，通过对重叠区域的信号进行加权相加，能有效消除音频处理后的杂音和断层。
3. 注意事项

在进行音频数据相加前，务必保证参与计算的音频具有相同的以下参数：
- 采样率（Sample Rate）：例如 44.1 kHz 或 48 kHz。
- 位深度（Bit Depth）：例如 16 -bit 或 24 -bit。
- 声道数（Channels）：例如单声道（Mono）或立体声（Stereo）。
4. DSP（数字信号处理）中的经典应用

在 DSP 中，音频相加（加法器 $\bigoplus$）是构建复杂音频效果的最核心基础算子：
- 混音器（Mixer）： 将 $N$ 路不同的音频信号乘以各自的增益权重后相加。
\[y[n] = g_1 \cdot x_1[n] + g_2 \cdot x_2[n] + \dots + g_N \cdot x_N[n]\]
- 延迟与回声效果器（Delay & Echo）： 将当前信号与过去某个时间点的信号（延时信号）相加，创造出空间感。
\[y[n] = x[n] + g \cdot x[n-D]\]
- 数字滤波器（FIR / IIR Filter）： 无论是低通、高通还是均衡器（EQ），其本质都是将不同延时阶段的采样点乘以系数后进行一连串的乘加组合（MAC 运算）。
\[y[n] = \sum_{k=0}^{M} b_k \cdot x[n-k] = b_0x[n] + b_1x[n-1] + b_2x[n-2] + \dots\]
- 主动降噪（ANC）： 收集环境噪音 $x[n]$，通过 DSP 计算出反相信号 $-x[n]$，然后将两者相加实现相位抵消：
\[y[n] = x[n] + (-x[n]) = 0\]
5. 硬件级优化考量
1. MAC 单元： DSP 芯片有专门的乘累加（Multiply-Accumulate）硬件单元，可以在一个时钟周期内完成 A * B + C。
2. 定点数饱和指令： 在定点数 DSP 芯片中，通常有硬件支持的饱和指令（如 ARM 的 __QADD），会自动让 $32767 + 1 = 32767$，防止数值翻转。
3. SIMD 加速： 现代 CPU 和 DSP 支持 SIMD（如 ARM 的 NEON，x86 的 SSE/AVX），可以在一个指令周期内同时让 4 个或 8 个采样点进行相加，极大提升多路混音性能。
6. DSP 芯片级/底层算法的实现考量

当我们在嵌入式 DSP 芯片（如 TI 的 C6000 系列、ARM Cortex-M 系列的 CMSIS-DSP）或底层 C 语言中做音频相加时，必须考虑硬件特性：

① MAC 单元（乘累加运算）

DSP 芯片之所以处理音频快，是因为它有专门的 MAC (Multiply-Accumulate) 硬件单元。它可以在一个时钟周期内完成 A * B + C 的操作。因此，在 DSP 算法中，音频相加往往是和乘法（音量控制/系数）绑定在一起执行的。

② 固定小数点（Fixed-point） vs 浮点数（Floating-point）
- 定点数 DSP（如 16-bit 整数）： 算力要求低、芯片便宜，但极易溢出。
- DSP 优化手法： 使用饱和加法（Saturated Add）指令。普通 CPU 中 $32767 + 1 = -32768$（符号位翻转导致巨恶劣爆音）；而 DSP 芯片通常有硬件支持的饱和指令（如 ARM 的 __QADD），会自动让 $32767 + 1 = 32767$，将失真降到最低。
- 浮点数 DSP（如 32-bit Float）： 现代调音台、音频插件（VST）首选。动态范围极大（超过 $1500 \text{ dB}$），相加时几乎不需要担心溢出问题，只需在最终输出（如 DAC 解码前）做一次硬剪切或限幅（Limiter）即可。
③ SIMD（单指令多数据）加速

现代 CPU 和 DSP 支持 SIMD（如 ARM 的 NEON 指令集，x86 的 SSE/AVX）。传统的加法是一个一个点相加；使用 SIMD 可以在一个指令周期内，同时让 4 个或 8 个采样点进行相加，这让实时处理多路高采样率（如 $96\text{kHz} / 192\text{kHz}$）的音频混音变得极其轻松。

二. 处理实时流（buffer）数据时，需要注意什么？

在处理实时流（Buffer）数据时，音频相加（混音）的最大挑战在于超低延迟要求以及防止突发性爆音（Clip）。

1. 实时流处理的核心步骤

实时流通常以固定大小的缓冲区（Buffer / Block）为单位进行传输（如 256、512 或 1024 个采样点）。
1. 对齐时钟：确保所有输入流的 Buffer 块大小和采样率完全一致。
2. 数据类型转换：将原始字节流（如 Int16 编码的 PCM）转换为浮点数（Float32）进行计算。
3. 混合与增益控制：对各路数据进行加权求和。
4. 限幅（Clipping/Limiting）：防止叠加后的数据超出安全范围。
5. 还原编码：将 Float32 重新转回原始的 PCM 字节流输出。
2. 推荐的实时流混音算法

在实时系统中，不建议使用简单的“平均法”（ (A+B)/2 ），因为这会导致每增加一路音频，整体音量就明显塌陷。推荐使用以下两种方式：

方案 A：Float32 累加 + 软限幅（推荐）

将所有输入 Buffer 转换成 [-1.0, 1.0] 的 float 数组，直接相加。在输出前通过 软限幅（Soft-Limiter / AGC） 公式，既能保留音量，又能优雅地处理溢出，避免硬截断带来的刺耳咔哒声（Click）。常用的实时软限幅公式（正切畸变）： $f(x) = \tanh(x)$

或者更轻量、计算更快的公式： $f(x) = \begin{cases} x & \vert{}x\vert{} \le 0.8 \cr 0.8 + 0.2 \times \tanh(\frac{\vert{}x\vert{}-0.8}{0.2}) & \vert{}x\vert{} > 0.8 \end{cases}$

方案 B：定点数（Int16）直接饱和相加

如果运行在嵌入式或对性能要求极高的底层系统（如 C/C++），可以使用硬件支持的 饱和加法（Saturated Add） 。当相加结果超过 +32767 时自动等于 32767，超过 -32768 时自动等于 -32768。
- 注：此方法在高爆音时仍会有轻微失真，但计算速度极快。
3. Python 实时流混音示例（基于 NumPy）

如果您的实时流是 PyAudio、WebRTC 或网络 Socket 传输的 bytes 缓冲，可以用以下高效的向量化代码进行混音：
```
import numpy as np
def mix_live_buffers(buffer_a: bytes, buffer_b: bytes) -> bytes:
    """
    对两个 16-bit PCM 实时 Buffer 进行混合
    """
    # 1. 将原始字节流高效转换为 Float32 数组 (范围 -1.0 到 1.0)
    data_a = np.frombuffer(buffer_a, dtype=np.int16).astype(np.float32) / 32768.0
    data_b = np.frombuffer(buffer_b, dtype=np.int16).astype(np.float32) / 32768.0
    
    # 2. 线性相加 (可在此处乘以权重，如 data_a * 0.8 + data_b * 0.5)
    mixed_float = data_a + data_b
    
    # 3. 实时软限幅 (Soft-Limiting)：防止硬截断爆音
    # 使用 np.tanh 可以平滑地将超出 [-1, 1] 的部分压缩回安全范围
    mixed_clipped = np.tanh(mixed_float)
    
    # 4. 还原为 16-bit PCM 字节流
    mixed_pcm = (mixed_clipped * 32767.0).astype(np.int16)
    return mixed_pcm.tobytes()
```
4. 实时流混音的避坑指南
- 非阻塞对齐：实时流经常面临“A 流到了，B 流还没到”的情况。必须设立一个环形缓冲区（Ring Buffer / FIFO Queue）。如果某一路断流，该路应当自动补零（静音），不能阻塞整个音频线程。
- 延迟与抖动：Buffer 越小延迟越低，但越容易卡顿。网络流混音前通常需要加入 Jitter Buffer（抖动缓冲区）来平滑网络波动。
- 淡入淡出（Fade）：当某一路实时语音忽然加入或离开时，不要直接切断，应该在 Buffer 的前几毫秒做线性音量渐变，否则听众会听到明显的“啪”声。
三. 当用的编程语言是 c，实时流是网络传输数据，如何处理？

在 C 语言中处理网络传输的音频实时流（Buffer）混音，最核心的原则是高性能、零拷贝（或少拷贝）以及异步容错。由于网络流存在延迟和丢包，你不能直接把刚收到的网络包进行相加。以下是标准的 C 语言工业级实时网络混音方案：

1. 核心架构设计

网络实时混音必须使用 Jitter Buffer（抖动缓冲区）架构。不能让音频播放线程去等待网络接收线程。
```
graph LR
    %% 第一行逻辑
    Input(网络数据包 UDP/RTP) --> ThreadA[线程 A: 网络接收]
    ThreadA --> Write[写入对应的环形队列 Ring Buffer A/B]

    %% 第二行逻辑
    Timer([播放设备/定时器回调]) --> ThreadB[线程 B: 混音线程]
    Read{定时读取各队列并混音} <--> ThreadB
    ThreadB --> Output(输出音频)

    %% 虚线表示数据存储/交换关系
    Write -.-> Read
```
2. 核心 C 语言代码实现

这里提供一个基于 Int16 PCM 格式、使用 饱和加法（Saturated Add） 的高效混音实现。饱和加法直接利用 CPU 边界判断，是 C 语言实时音频处理中最快速、最省算力的防爆音方法。
```
#include <stdio.h>
#include <stdint.h>
#include <string.h>

// 饱和加法：防止 16位有符号整数溢出爆音
inline int16_t saturate_add(int32_t sample1, int32_t sample2) {
    int32_t mixed = sample1 + sample2;
    if (mixed > 32767)  return 32767;
    if (mixed < -32768) return -32768;
    return (int16_t)mixed;
}

/**
 * 实时网络流混音函数
 * @param buffer_a    网络流A的缓冲区
 * @param buffer_b    网络流B的缓冲区
 * @param out_buffer  混音后的输出缓冲区
 * @param samples     本次 Buffer 的采样点数量（注意：不是字节数！字节数 = samples * 2）
 * @param weight_a    流A的音量权重 (0.0 到 1.0)
 * @param weight_b    流B的音量权重 (0.0 到 1.0)
 */
void mix_network_buffers(const int16_t* buffer_a, const int16_t* buffer_b, 
                         int16_t* out_buffer, size_t samples, 
                         float weight_a, float weight_b) 
{
    // 如果某一路网络流断流（Buffer为空），直接处理另一路
    if (buffer_a == NULL && buffer_b == NULL) {
        memset(out_buffer, 0, samples * sizeof(int16_t));
        return;
    }
    
    for (size_t i = 0; i < samples; i++) {
        // 1. 获取采样点，若断流则补静音(0)
        int32_t s1 = buffer_a ? (int32_t)(buffer_a[i] * weight_a) : 0;
        int32_t s2 = buffer_b ? (int32_t)(buffer_b[i] * weight_b) : 0;
        
        // 2. 混合并进行饱和截断
        out_buffer[i] = saturate_add(s1, s2);
    }
}
```
3. 网络流混音的 3 大致命坑与 C 语言解决方案

① 丢包与断流（Packet Loss & Discontinuity）
- 现象：网络抖动导致流 A 的数据包迟到，混音线程此时必须输出声音。
- 对策：如果网络环形缓冲区（Ring Buffer）为空，混音线程绝不能阻塞等待。必须直接视流 A 为全 0（静音）继续运行。
- 高阶优化：使用 PLC（丢包补偿）算法，在断流时复制前一个包的尾部并做淡出，避免声音突变。
② 时钟异步与漂移（Clock Drift）
- 现象：发送端 A、发送端 B 和你的接收端服务器，三者的声卡晶振频率有微小差异。运行几小时后，由于速率不同，某一路的 Buffer 会越来越大（积压延迟）或越来越小（卡顿）。
- 对策：建立水线（Watermark）机制。
- 如果环形队列中积压的数据超过上限（如 > 60ms），则跳过（Drop）一小段采样。
  - 如果积压数据过低（如 < 10ms），则复制/插值（Interpolate）一小段采样。
③ 突发加入的“啪”声（Click Noise）
- 现象：当某路网络音频突然连上并开始说话时，信号从 0 突变到某个值，会产生刺耳的咔哒声。
- 对策：在流刚加入的第一个 Buffer（大约 5-10ms）应用淡入（Gain 从 0.0 渐变到 1.0）；在断开或静音时应用淡出（Gain 从 1.0 渐变到 0.0）。
4. 推荐使用的开源 C 库

如果是在生产环境开发，不建议全部从零手写，可以集成以下成熟的轻量级 C 库：
1. WebRTC Audio Processing (Media Process)：包含了世界上最强大的 Jitter Buffer、NetEQ（动态时钟对齐）以及 AGC（自动增益控制）。
2. Opus Codec (libopus)：如果你的网络流是压缩的，Opus 自带了丢包补偿（PLC）和重采样功能。
四. 需要支持多路（大于2路）网络音频同时混合

在 C 语言中支持多路（大于2路）网络音频混音，最优雅且高效的架构是 “基线累加 + 单次饱和截断” 。这意味着我们要把所有声道的采样点先加到一个高精度的变量（如 int32_t 或 float）中，最后统一做一次饱和处理，而不是两两两两相加，这样可以最大程度减少性能开销并降低失真。

以下是专为多路网络流设计的 C 语言解决方案。

1. 数据结构设计：管理多路流

为了灵活应对 3 路、4 路甚至数十路音频，需要用结构体来管理每一路网络音频的状态（包含其缓冲区指针、当前音量权重、是否有效等）。
```
#include <stdio.h>
#include <stdint.h>
#include <string.h>
#include <stdlib.h>
// 单路音频流的状态typedef struct {
    const int16_t* buffer;  // 指向当前需要混音的 Buffer（从该流的 RingBuffer 中取出）
    float volume;           // 该路音频的音量权重 (0.0 ~ 1.0)
    int is_active;          // 该路当前是否有数据（0 = 断流/静音，1 = 有数据）
} AudioStreamInput;
```
2. 多路混音核心 C 代码

该实现采用 int32_t 累加器。如果通道数极多（例如超过 100 路），为了彻底防止累加阶段就溢出，可以将累加器改为 float，这里以常规的多路网络会议/语音场景（int32_t）为例：
```
// 高效饱和截断函数
inline int16_t saturate_add_multi(int32_t sum) {
    if (sum > 32767)  return 32767;
    if (sum < -32768) return -32768;
    return (int16_t)sum;
}

/**
 * 多路实时网络流混音
 * @param streams      输入流数组
 * @param stream_count 输入流的总路数
 * @param out_buffer   混音后的输出缓冲区
 * @param samples      本次 Buffer 的采样点数量
 */
void mix_multi_network_buffers(const AudioStreamInput* streams, size_t stream_count, 
                               int16_t* out_buffer, size_t samples) 
{
    // 1. 遍历每一个采样点
    for (size_t i = 0; i < samples; i++) {
        int32_t accumulator = 0;
        int active_count = 0;

        // 2. 累加所有有效流在当前位置的采样点
        for (size_t j = 0; j < stream_count; j++) {
            if (streams[j].is_active && streams[j].buffer != NULL) {
                // 应用音量权重并累加
                accumulator += (int32_t)(streams[j].buffer[i] * streams[j].volume);
                active_count++;
            }
        }

        // 3. 动态增益控制 (可选优化)
        // 如果同时说话的人太多，声音会整体过载失真。
        // 可以根据当前激发的流数量，进行轻微的衰减（例如下面这行代码，根据需要取消注释）：
        // if (active_count > 2) { accumulator = (int32_t)(accumulator * 0.8f); }

        // 4. 统一进行一次饱和截断并输出
        out_buffer[i] = saturate_add_multi(accumulator);
    }
}
```
3. 工程化应用示例（伪代码）

在实际的网络音频服务器或客户端中，你的混音线程通常会这样调用它：
```
#define MAX_STREAMS 8
#define BUFFER_SAMPLES 480 // 假设 48kHz, 10ms 的数据

void audio_mixer_thread_loop() {
    AudioStreamInput inputs[MAX_STREAMS];
    int16_t output_buffer[BUFFER_SAMPLES];

    while (running) {
        // 1. 等待定时器（例如每 10ms 触发一次）
        wait_for_10ms_timer();

        // 2. 从各自的 Jitter Buffer 中提取数据
        for (int i = 0; i < MAX_STREAMS; i++) {
            // 从第 i 路用户的环形缓冲区读取 10ms 数据
            // 如果提取失败（丢包/断流），ring_buffer_pop 应返回 NULL
            inputs[i].buffer = ring_buffer_pop(user_ring_buffers[i], BUFFER_SAMPLES);
            inputs[i].volume = user_volumes[i]; // 用户设定的音量
            inputs[i].is_active = (inputs[i].buffer != NULL) ? 1 : 0;
        }

        // 3. 执行多路混音
        mix_multi_network_buffers(inputs, MAX_STREAMS, output_buffer, BUFFER_SAMPLES);

        // 4. 将混音后的数据推送到声卡播放，或者编码发给其他网络用户
        send_to_speaker_or_network(output_buffer, BUFFER_SAMPLES);
    }
}
```
4. 多路网络混音的商业级设计要点
- 发言者检测（VAD - Voice Activity Detection）：在大于 2 路的场景下（如 50 人的会议室），如果把所有人的网络流都强行加进来，即使不说话，他们各自背景里的 “底噪”叠加起来也会变成巨大的噪音 。必须引入 VAD 算法，或者设置一个能量门限（Threshold）。当某路流的平均能量低于设定值时，直接将 is_active 设为 0（视为静音），不参与混音。
- 现代 CPU 性能加速（SIMD / AVX）：如果流的数量非常大，C 语言的 for 循环可以通过编译优化加速。在 GCC / Clang 中编译时，请务必开启 -O3 优化开关，编译器会自动将上述代码转化为 SIMD（单指令多数据） 指令（如 Intel 的 AVX 或 ARM 的 NEON），实现 4 个或 8 个采样点同时相加，效率提升数倍。
- 内存对齐：分配 Buffer 时，尽量使用支持字节对齐的函数（如 posix_memalign 或 _aligned_malloc），这能让 CPU 在执行多路数据指针读取时达到最高速度。
Read All
vscode 插件 Git Graph 使用(转)

2026-06-30

David

experience

knowledge, base, gitgraph, vscode
- content
vscode 插件 Git Graph 使用(原)

gitgraph

enable gitgraph

fetch:

create branch:

checkout：

commit or checkout:

push:

pop stash:

stash:

commit:

push to remote:

push branch:
Read All
在 Ubuntu 22.04上搭建TFTP服务器和NFS服务器

2026-06-18

David

experience

knowledge, base, tftp server, nfs server
- content
在 Ubuntu 22.04 上搭建 TFTP 服务器和 NFS 服务器是嵌入式开发（如 Linux 内核调试、根文件系统挂载）的经典组合。以下是详细的搭建步骤：

一、搭建 TFTP 服务器

TFTP（简单文件传输协议）常用于在开发板启动时下载 Linux 内核镜像（uImage/zImage）和设备树文件（.dtb）。

1. 安装 TFTP 相关软件

打开终端，执行以下命令安装 TFTP 服务端和客户端：

Bash
```
sudo apt update
sudo apt install tftpd-hpa tftp-hpa
```
2. 创建 TFTP 工作目录

创建一个用于存放传输文件的目录，并赋予最高权限，确保开发板可以正常读写：

Bash
```
sudo mkdir -p /home/tftpboot
sudo chmod 777 /home/tftpboot
```
3. 修改配置文件

编辑 TFTP 的配置文件 /etc/default/tftpd-hpa：

Bash
```
sudo nano /etc/default/tftpd-hpa
```
将内容修改为以下配置（主要修改 TFTP_DIRECTORY 和 TFTP_OPTIONS）：

Plaintext
```
TFTP_USERNAME="tftp"
TFTP_DIRECTORY="/home/tftpboot"
TFTP_ADDRESS=":69"
TFTP_OPTIONS="--secure --create"
```
提示：--create 参数允许客户端在服务器上创建新文件（即支持上传）。

4. 重启并检查服务

保存退出后，重启 TFTP 服务并将其设置为开机自启：

Bash
```
sudo systemctl restart tftpd-hpa
sudo systemctl enable tftpd-hpa
```
检查服务状态，确保显示 active (running)：

Bash
```
sudo systemctl status tftpd-hpa
```
5. 本地测试 TFTP

你可以通过本地回环测试来验证是否搭建成功：

Bash
```
echo "TFTP Test File" > /home/tftpboot/test.txt
cd /tmp
tftp 127.0.0.1
# 进入 tftp 命令行后输入：
get test.txt
quit

# 查看 /tmp 目录下是否成功获取文件
cat test.txt
```
二、搭建 NFS 服务器

NFS（网络文件系统）常用于将 Ubuntu 上的某个目录（如根文件系统 rootfs）挂载到开发板上，方便直接修改和调试代码而无需反复烧录。

1. 安装 NFS 服务端

Bash
```
sudo apt install nfs-kernel-server
```
2. 创建 NFS 共享目录

创建一个用于挂载的文件目录：

Bash
```
sudo mkdir -p /home/nfsroot
sudo chmod 777 /home/nfsroot
```
3. 配置共享目录权限

编辑 /etc/exports 文件来指定允许访问的客户端和权限：

Bash
```
sudo nano /etc/exports
```
在文件末尾添加以下内容（根据你的网络环境调整）：

Plaintext
```
/home/nfsroot *(rw,sync,no_root_squash,no_subtree_check)
```
参数说明：
- *：代表允许所有的 IP 访问（在局域网开发环境中很方便。如果为了安全，可以写具体网段如 119.168.1.0/24）。
- rw：读写权限。
- sync：资料同步写入内存与硬盘。
- no_root_squash：如果客户端使用的是 root 用户，则对该共享目录也拥有 root 权限（嵌入式开发必备）。
- no_subtree_check：不检查子目录权限，提高性能。
4. 解决 Ubuntu 22.04 的 NFS 版本兼容问题（重要 ⚠️）

避坑指南：Ubuntu 22.04 默认关闭了 NFS v2 和 v3，只开启了 NFS v4。但是很多老旧的开发板 U-Boot 或内核只支持 NFS v2 或 v3。如果你在挂载时遇到 Protocol not supported 或一直连不上，请按以下步骤开启 v3 支持。
1. 编辑 /etc/default/nfs-kernel-server：
  
  Bash
  sudo nano /etc/default/nfs-kernel-server
  修改或添加以下内容，确保不禁用 v3：
  
  Plaintext
  RPCNFSDOPTS=""
2. 编辑 /etc/nfs.conf：
  
  Bash
  sudo nano /etc/nfs.conf
  找到 [nfsd] 标签，取消注释并修改 vers3 如下：
  
  Plaintext
  [nfsd] vers3=y
5. 重启 NFS 服务

配置完成后，重启相关服务并导出共享目录：

Bash
```
sudo systemctl restart rpcbind
sudo systemctl restart nfs-kernel-server
sudo systemctl enable nfs-kernel-server
```
使用以下命令查看当前生效的 NFS 共享目录：

Bash
```
showmount -e
```
如果输出显示 /home/nfsroot *，说明配置成功。

6. 本地测试 NFS

你可以尝试在本地挂载自己，验证服务是否正常：

Bash
```
sudo mount -t nfs -o nfsvers=3 127.0.0.1:/home/nfsroot /mnt
# 检查是否成功挂载，随后卸载
ls /mnt
sudo umount /mnt
```
三、常见注意事项
1. 防火墙问题：如果开发板依然无法连接，建议直接关闭 Ubuntu 的防火墙（仅限本地安全开发环境）：
  
  Bash
  sudo ufw disable
2. IP 地址在同一网段：请确保你的 Ubuntu 主机、Windows 宿主机（如果是虚拟机的话）以及开发板的 IP 地址处于同一个网段，且能够互相 ping 通。
Read All
Github上这款免费开源的Bash脚本教程

2026-01-23

David

experience

knowledge base
参考：
1. introduction-to-bash-scripting
2. 挖到宝了！Github上这款免费开源的Bash脚本教程，让我从编程小白逆袭效率大神
content
Read All
更改Linux/Unix 环境变量

2026-01-23

David

experience

knowledge base
- 方案 1：使用 source 命令（推荐！）
- 方案 2：在 shell 配置文件中永久设置
情况一：在终端中直接设置（生效）
```
export PATH=$PATH:/new/path
```
这是在当前 shell（父进程）中直接修改 PATH 后续所有命令都继承这个新 PATH

情况二：在脚本中设置（不生效）
```
# setpath.sh
export PATH=$PATH:/new/path
echo $PATH  # 脚本内能看到新 PATH
```
```
./setpath.sh    # 或 bash setpath.sh
echo $PATH      # 终端中 PATH 未改变！
```
为什么？

当你运行 ./setpath.sh 时，系统会启动一个子 shell 来执行脚本脚本中的 export 只修改了子 shell 的环境变量脚本执行完毕后，子 shell 退出，修改不会回传给父 shell（你的终端）

📌 Unix/Linux 的设计原则：子进程不能修改父进程的环境变量（安全机制）

✅ 正确解决方案

方案 1：使用 source 命令（推荐！）
```
source setpath.sh# 
或简写
. setpath.sh
```
原理：
- source 不会启动子 shell，而是在当前 shell 中逐行执行脚本
- 所有变量修改都作用于当前终端
方案 2：在 shell 配置文件中永久设置

如果希望每次打开终端都生效，将 PATH 添加到配置文件：

对当前用户：
```
echo 'export PATH="$PATH:/opt/myapp/bin"' >> ~/.bashrc
# 或 ~/.zshrc（如果你用 zsh）
```
然后重载配置：
```
source ~/.bashrc
```
对所有用户（需 root 权限）：
```
sudo echo 'export PATH="$PATH:/opt/myapp/bin"' >> /etc/environment
```
Read All

构建自己的本地知识库

2026-01-07

David

experience

knowledge base

一、整体架构（完全本地化）
二、关键组件与工具推荐（开源 + 本地运行）
三、数据安全与隐私保障
四、简易搭建流程（示例）
五、进阶建议
总结
🧰 第一步：准备工作
📁 第二步：准备你的知识文档
🧩 第三步：编写文档加载与分块脚本
🧠 第四步：构建本地向量数据库（使用 BGE 中文嵌入模型）
🔍 第五步：本地语义检索（不依赖大模型，直接返回相关段落）
🤖 第六步（可选）：使用本地大模型生成答案（RAG）
🔒 隐私与安全说明
📌 附录：完全离线使用（提前下载模型）
- A. 下载嵌入模型
- B. 下载 Ollama 模型后复制到目标机器
✅ 总结
需要图形界面以及自动监控文档更新
在 ingest.py 中启用 Unstructured 的 PDF/DOCX loader

在“仅用于自己查询、不上传任何信息到云端”的前提下，构建本地知识库的核心目标是：完全离线、数据安全、高效检索。下面是一个系统化的构建方案，涵盖工具选择、数据处理、检索方式等关键环节：

一、整体架构（完全本地化）

[本地文档/数据源]  
       ↓  
[文档解析 & 向量化] → [本地向量数据库]  
       ↓  
[本地大模型（LLM）] ←→ [本地检索系统]  
       ↓  
[用户界面 / 查询入口]

所有组件均运行在你的个人电脑或局域网服务器上，无需联网。

二、关键组件与工具推荐（开源 + 本地运行）

1. 文档解析与预处理

支持格式：PDF、Word、TXT、Markdown、网页快照、笔记（如 Obsidian/Logseq 导出）等。
工具：
- Unstructured（Python 库）：智能解析多种文档。
- PyPDF2 / pdfplumber：处理 PDF。
- BeautifulSoup：解析 HTML。
- 对于 Obsidian 等笔记系统，可直接读取 Markdown 文件。

2. 文本分块（Chunking）

将长文档切分为适合检索的片段（如 300–500 字）。
可使用重叠分块（overlap）保留上下文。
工具：LangChain 的 RecursiveCharacterTextSplitter 或自定义逻辑。

3. 文本向量化（Embedding）

将文本转换为向量（用于语义搜索）。
推荐本地嵌入模型（无需联网）：
- BAAI/bge-small-zh-v1.5（中文优化，轻量高效）
- m3e-small（中文社区常用）
- all-MiniLM-L6-v2（英文/多语言，极小）
使用 sentence-transformers 库本地加载模型。

4. 本地向量数据库

存储和检索向量，支持相似性搜索。
推荐：
- ChromaDB：轻量、嵌入式、Python 友好，支持持久化。
- FAISS（Facebook）：高效，适合静态知识库。
- Qdrant（可本地运行）：功能更强，支持过滤、多向量等。
- SQLite + Vector 扩展（如 sqlite-vss）：极简方案。

✅ 所有数据库均可配置为仅保存在本地磁盘，不联网。

5. 本地大语言模型（用于生成答案）

可选：如果你希望模型基于检索到的内容“总结”或“回答”，而非仅返回原文。
推荐本地 LLM（7B 以下可在消费级 GPU 或 CPU 运行）：
- 中文：Qwen1.5-4B-Chat、Yi-1.5-6B-Chat、DeepSeek-Coder（若含代码）
- 多语言/英文：Mistral-7B、Phi-3-mini（微软，性能强且小）
运行框架：
- Ollama（最简单，一键运行模型）
- LM Studio（图形界面，支持聊天和 API）
- llama.cpp（CPU 友好，量化后可在普通笔记本运行）

6. 检索增强生成（RAG）本地实现

流程：
1. 用户提问 → 用本地嵌入模型转为向量。
2. 在本地向量库中搜索最相关的 3–5 个文本块。
3. 将问题 + 检索结果拼接为提示（prompt），送入本地 LLM 生成答案。
可用 LangChain 或 LlamaIndex 构建本地 RAG 流程（设置 offline=True，禁用网络）。

7. 用户界面（可选）

命令行（最简单）
Web 界面（如用 Gradio 或 Streamlit 快速搭建本地 UI）
Obsidian 插件（如 Smart Connections + Local LLM 集成）

三、数据安全与隐私保障

所有数据存储在本地磁盘（如 ~/my_knowledge_base/）。
向量数据库文件（如 Chroma 的 persist directory）也保存在本地。
使用 ollama serve 或 lmstudio 时，确保禁用远程访问（默认仅 localhost）。
可对敏感数据加密（如使用 VeraCrypt 创建加密卷存放知识库）。

四、简易搭建流程（示例）

安装 Python 和必要库：

pip install langchain chromadb sentence-transformers unstructured

下载中文嵌入模型（首次运行自动下载到本地）：

from sentence_transformers import SentenceTransformer
model = SentenceTransformer('BAAI/bge-small-zh-v1.5')

构建向量库（以 Chroma 为例）：

from langchain_chroma import Chroma
from langchain_community.embeddings import HuggingFaceBgeEmbeddings

embeddings = HuggingFaceBgeEmbeddings(model_name="BAAI/bge-small-zh-v1.5")
vectorstore = Chroma.from_documents(
    documents=your_docs,
    embedding=embeddings,
    persist_directory="./chroma_db"
)

本地查询（无需 LLM 也可直接返回相关段落）：

results = vectorstore.similarity_search("你的问题", k=3)
print(results)

若需 LLM 生成答案，用 Ollama 启动模型：
```
ollama pull qwen:4b
ollama run qwen:4b
```
再通过 API 或 LangChain 调用。

五、进阶建议

定期备份：知识库文件（向量库 + 原始文档）定期加密备份。
增量更新：支持新增/删除文档后更新向量库。
多模态扩展：未来可加入图片 OCR（Tesseract + PaddleOCR 本地运行）。

总结

✅ 完全本地
✅ 无数据上传
✅ 中文友好
✅ 开源免费
✅ 可在普通电脑运行（尤其使用 4B 以下模型 + Chroma/FAISS）

只要你有一台个人电脑（最好是 16GB+ 内存，如有独立 GPU 更佳），就可以搭建一个私密、安全、高效的个人知识库系统。

以下是一个完全本地化、无需联网、不上传任何数据到云端的中文个人知识库构建教程，使用开源工具和模型。我们将使用：

文档格式：本地 Markdown / TXT / PDF（以 Markdown 为例）
嵌入模型：BAAI/bge-small-zh-v1.5（中文优化，本地运行）
向量数据库：ChromaDB（轻量、支持持久化）
大模型（可选）：Qwen 4B（通过 Ollama 本地运行）
框架：Python + LangChain（仅用于流程编排，不联网）
运行环境：Windows / macOS / Linux 均可（推荐 Python 3.10+）

🧰 第一步：准备工作

1.1 安装 Python（建议 3.10 或 3.11）

从 python.org 下载并安装。

1.2 创建项目目录

mkdir ~/local_knowledge_base
cd ~/local_knowledge_base

1.3 创建虚拟环境（推荐）

python -m venv venv
source venv/bin/activate      # macOS/Linux
# 或
venv\Scripts\activate         # Windows

1.4 安装依赖包

pip install langchain \
            langchain-community \
            langchain-chroma \
            chromadb \
            sentence-transformers \
            unstructured \
            unstructured[local-inference] \
            pdf2image \
            PyPDF2 \
            ollama \
            python-dotenv

⚠️ 如果你不需要 PDF 解析，可跳过 pdf2image 和 PyPDF2。

⚠️ 首次运行 sentence-transformers 会自动下载模型到本地缓存（约 130MB），但不需要联网之后也能用。如果你完全不能联网，请提前在有网环境下载好模型（见附录）。

📁 第二步：准备你的知识文档

在项目目录下创建一个 docs/ 文件夹，放入你的文档，例如：

local_knowledge_base/
├── docs/
│   ├── note1.md
│   ├── note2.md
│   └── my_book_summary.txt

支持格式：.txt, .md, .pdf（后续可扩展）

🧩 第三步：编写文档加载与分块脚本

创建文件：ingest.py

# ingest.py
import os
from langchain_community.document_loaders import (
    DirectoryLoader,
    TextLoader,
    UnstructuredMarkdownLoader,
    PyPDFLoader
)
from langchain_text_splitters import RecursiveCharacterTextSplitter

# 配置路径
DOCS_DIR = "./docs"
PERSIST_DIR = "./chroma_db"

def load_documents():
    """加载所有支持的文档"""
    documents = []

    # 加载 .txt 文件
    txt_loader = DirectoryLoader(DOCS_DIR, glob="*.txt", loader_cls=TextLoader, loader_kwargs={"encoding": "utf-8"})
    documents.extend(txt_loader.load())

    # 加载 .md 文件
    md_loader = DirectoryLoader(DOCS_DIR, glob="*.md", loader_cls=UnstructuredMarkdownLoader)
    documents.extend(md_loader.load())

    # 加载 .pdf 文件（可选）
    # pdf_loader = DirectoryLoader(DOCS_DIR, glob="*.pdf", loader_cls=PyPDFLoader)
    # documents.extend(pdf_loader.load())

    print(f"✅ 共加载 {len(documents)} 个文档片段")
    return documents

def split_documents(documents):
    """将文档切分为小块"""
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=500,
        chunk_overlap=50,
        length_function=len,
        is_separator_regex=False,
    )
    chunks = text_splitter.split_documents(documents)
    print(f"✅ 切分为 {len(chunks)} 个文本块")
    return chunks

if __name__ == "__main__":
    docs = load_documents()
    chunks = split_documents(docs)

    # 保存到临时文件（用于下一步嵌入）
    os.makedirs(PERSIST_DIR, exist_ok=True)
    with open(os.path.join(PERSIST_DIR, "chunks.txt"), "w", encoding="utf-8") as f:
        for i, chunk in enumerate(chunks):
            f.write(f"--- Chunk {i} ---\n")
            f.write(chunk.page_content + "\n\n")
    print("✅ 文本块已保存（供调试）")

运行：

python ingest.py

✅ 此时会在 chroma_db/chunks.txt 中看到切分后的文本（可选，仅调试用）

🧠 第四步：构建本地向量数据库（使用 BGE 中文嵌入模型）

创建文件：build_vector_db.py

# build_vector_db.py
from langchain_chroma import Chroma
from langchain_community.embeddings import HuggingFaceBgeEmbeddings
from ingest import load_documents, split_documents
import os

# 嵌入模型配置（完全本地）
embedding_model_name = "BAAI/bge-small-zh-v1.5"
model_kwargs = {'device': 'cpu'}          # 若有 GPU 可改为 'cuda'
encode_kwargs = {'normalize_embeddings': True}  # BGE 推荐设置

embeddings = HuggingFaceBgeEmbeddings(
    model_name=embedding_model_name,
    model_kwargs=model_kwargs,
    encode_kwargs=encode_kwargs,
    query_instruction="为这个句子生成表示以用于检索相关文章："  # BGE 中文模型推荐 query 指令
)

PERSIST_DIR = "./chroma_db"

def build_vector_db():
    print("🔍 正在加载文档...")
    docs = load_documents()
    chunks = split_documents(docs)

    print("🧠 正在生成向量并构建数据库（首次运行会下载模型到本地缓存）...")
    vectorstore = Chroma.from_documents(
        documents=chunks,
        embedding=embeddings,
        persist_directory=PERSIST_DIR,
        collection_name="my_knowledge"
    )
    print(f"✅ 向量数据库已保存到 {PERSIST_DIR}")

if __name__ == "__main__":
    build_vector_db()

运行：

python build_vector_db.py

⏳ 首次运行会自动下载 BAAI/bge-small-zh-v1.5 模型（约 130MB）到 ~/.cache/huggingface/，之后完全离线可用。

✅ 数据库文件将保存在 ./chroma_db/，仅在本地磁盘。

🔍 第五步：本地语义检索（不依赖大模型，直接返回相关段落）

创建文件：query.py

# query.py
from langchain_chroma import Chroma
from langchain_community.embeddings import HuggingFaceBgeEmbeddings

embedding_model_name = "BAAI/bge-small-zh-v1.5"
model_kwargs = {'device': 'cpu'}
encode_kwargs = {'normalize_embeddings': True}

embeddings = HuggingFaceBgeEmbeddings(
    model_name=embedding_model_name,
    model_kwargs=model_kwargs,
    encode_kwargs=encode_kwargs,
    query_instruction="为这个句子生成表示以用于检索相关文章："
)

vectorstore = Chroma(
    persist_directory="./chroma_db",
    embedding_function=embeddings,
    collection_name="my_knowledge"
)

def search(query: str, k: int = 3):
    docs = vectorstore.similarity_search(query, k=k)
    print(f"\n🔍 问题：{query}\n")
    for i, doc in enumerate(docs):
        print(f"【结果 {i+1}】（来源: {doc.metadata.get('source', '未知')}）")
        print(doc.page_content[:500] + "...\n")
    return docs

if __name__ == "__main__":
    while True:
        question = input("\n请输入你的问题（输入 'quit' 退出）：")
        if question.lower() == 'quit':
            break
        search(question)

运行：

python query.py

✅ 此时你可以直接提问，系统会返回最相关的原文片段，完全本地、无网络请求。

🤖 第六步（可选）：使用本地大模型生成答案（RAG）

6.1 安装 Ollama（https://ollama.com/）

下载并安装 Ollama（支持 Windows/macOS/Linux）
安装后，默认只监听 localhost，不联网

6.2 拉取本地中文模型（首次需要联网，之后离线可用）

ollama pull qwen:4b

模型大小约 2.5GB（4-bit 量化版），可在 16GB 内存笔记本 CPU 运行（稍慢）。

6.3 创建 RAG 查询脚本：`rag_query.py`

# rag_query.py
from langchain_chroma import Chroma
from langchain_community.embeddings import HuggingFaceBgeEmbeddings
from langchain_ollama import OllamaLLM
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

# === 配置 ===
embedding_model_name = "BAAI/bge-small-zh-v1.5"
llm_model = "qwen:4b"  # Ollama 中的模型名

# === 嵌入与向量库 ===
embeddings = HuggingFaceBgeEmbeddings(
    model_name=embedding_model_name,
    model_kwargs={'device': 'cpu'},
    encode_kwargs={'normalize_embeddings': True},
    query_instruction="为这个句子生成表示以用于检索相关文章："
)

vectorstore = Chroma(
    persist_directory="./chroma_db",
    embedding_function=embeddings,
    collection_name="my_knowledge"
)
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

# === LLM 与 Prompt ===
llm = OllamaLLM(model=llm_model, temperature=0.3)

template = """
你是一个知识助手，请根据以下提供的上下文信息回答问题。
只使用上下文中的内容，不要编造。如果上下文没有相关信息，请回答“根据现有资料无法回答”。

上下文：
{context}

问题：{question}
答案：
"""

prompt = ChatPromptTemplate.from_template(template)

# === RAG 链 ===
rag_chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

# === 查询 ===
if __name__ == "__main__":
    while True:
        question = input("\n🧠 请输入你的问题（'quit' 退出）：")
        if question.lower() == 'quit':
            break
        print("\n🤔 正在思考...\n")
        answer = rag_chain.invoke(question)
        print(f"✅ 答案：{answer}\n")

⚠️ 确保 Ollama 正在运行（终端输入 ollama serve，或后台已启动）

运行：

python rag_query.py

✅ 此时系统会：

用本地嵌入模型检索相关段落

将段落 + 问题交给本地 Qwen 模型

生成自然语言答案 全程无任何数据上传到网络

🔒 隐私与安全说明

所有数据（原始文档、向量库、模型缓存）均保存在你的电脑。
ChromaDB 默认保存在 ./chroma_db/
Hugging Face 模型缓存路径：~/.cache/huggingface/hub/
Ollama 模型路径：
- macOS: ~/.ollama/models/
- Windows: C:\Users\<user>\.ollama\models\
- Linux: ~/.ollama/models/
可将整个 local_knowledge_base 文件夹放入加密磁盘（如 VeraCrypt）进一步保护。

📌 附录：完全离线使用（提前下载模型）

如果你完全不能联网，请在有网络的机器上提前下载：

A. 下载嵌入模型

from sentence_transformers import SentenceTransformer
model = SentenceTransformer('BAAI/bge-small-zh-v1.5')
model.save('./bge-small-zh')  # 保存到本地

然后修改代码中 model_name='./bge-small-zh'

B. 下载 Ollama 模型后复制到目标机器

Ollama 模型可直接复制 ~/.ollama/models/ 到另一台机器（需相同架构）。

✅ 总结

你现在已经拥有一个：

✅ 100% 本地运行
✅ 支持中文
✅ 无需上传任何数据
✅ 可扩展（PDF/Word/笔记）
✅ 支持语义检索 + 智能问答

的私人知识库系统！

需要图形界面以及自动监控文档更新

在已有完全本地、离线、隐私安全的知识库基础上，我们可以进一步：

添加图形界面（GUI）：使用 Streamlit 快速搭建 Web 界面（仅本地访问）
实现自动监控文档更新：使用 watchdog 监听 docs/ 目录变化，自动增量更新向量库

✅ 依然 不联网、不上传任何数据，所有组件运行在本地。

✅ 最终目标架构

[docs/ 目录] ←─(watchdog 监控)─┐
                              ↓
[用户通过浏览器访问 http://localhost:8501]  
                              ↓  
[Streamlit GUI] → [检索/问答逻辑] → [ChromaDB + 本地 LLM]
                              ↑  
（文档变更时自动触发 rebuild 向量库）

第一步：安装额外依赖

pip install streamlit watchdog

第二步：实现“自动监控文档更新 + 增量更新向量库”

注意：ChromaDB 本身不支持高效增量删除，但我们可以：

简单方案：监控到变化 → 重新构建整个向量库（适合 <1000 文档）

进阶方案：记录文档 hash，只更新变更/新增的 chunk（本文采用简单方案，清晰可靠）

创建 `auto_updater.py`

# auto_updater.py
import os
import time
import hashlib
from pathlib import Path
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
from build_vector_db import build_vector_db  # 复用之前写的构建函数

DOCS_DIR = "./docs"
STATE_FILE = "./chroma_db/update_state.txt"

def compute_docs_hash():
    """计算 docs/ 目录下所有文件的 hash，用于判断是否变化"""
    hash_md5 = hashlib.md5()
    paths = sorted(Path(DOCS_DIR).rglob("*"))
    for path in paths:
        if path.is_file():
            stat = path.stat()
            hash_md5.update(str(stat.st_mtime).encode())
            hash_md5.update(str(stat.st_size).encode())
    return hash_md5.hexdigest()

def load_last_hash():
    if os.path.exists(STATE_FILE):
        with open(STATE_FILE, "r") as f:
            return f.read().strip()
    return None

def save_current_hash():
    current_hash = compute_docs_hash()
    os.makedirs(os.path.dirname(STATE_FILE), exist_ok=True)
    with open(STATE_FILE, "w") as f:
        f.write(current_hash)

class DocUpdateHandler(FileSystemEventHandler):
    def __init__(self, callback):
        self.callback = callback
        self.last_trigger = 0

    def on_any_event(self, event):
        # 防抖：5 秒内只触发一次
        if time.time() - self.last_trigger > 5:
            if event.src_path.endswith(('.txt', '.md', '.pdf')):
                print(f"\n📁 检测到文档变更: {event.src_path}")
                self.last_trigger = time.time()
                self.callback()

def start_watcher():
    """启动文件监控"""
    observer = Observer()
    event_handler = DocUpdateHandler(on_docs_changed)
    observer.schedule(event_handler, DOCS_DIR, recursive=True)
    observer.start()
    print(f"👀 正在监控 {os.path.abspath(DOCS_DIR)} 目录...")

    # 初始检查是否需要首次构建
    if not os.path.exists("./chroma_db/chroma.sqlite3"):  # ChromaDB 默认文件
        print("🆕 首次运行：正在构建向量库...")
        build_vector_db()
        save_current_hash()
    else:
        current_hash = compute_docs_hash()
        last_hash = load_last_hash()
        if current_hash != last_hash:
            print("🔄 检测到历史变更：正在重建向量库...")
            build_vector_db()
            save_current_hash()

    return observer

def on_docs_changed():
    """文档变更时的回调"""
    build_vector_db()
    save_current_hash()
    print("✅ 向量库已更新！")

第三步：改造 `build_vector_db.py`（支持被调用）

修改 build_vector_db.py，使其可被其他模块调用而不重复初始化模型：

# build_vector_db.py（修改版）
from langchain_chroma import Chroma
from langchain_community.embeddings import HuggingFaceBgeEmbeddings
from ingest import load_documents, split_documents
import os
import shutil

PERSIST_DIR = "./chroma_db"
embedding_model_name = "BAAI/bge-small-zh-v1.5"
model_kwargs = {'device': 'cpu'}
encode_kwargs = {'normalize_embeddings': True}
query_instruction = "为这个句子生成表示以用于检索相关文章："

# 全局嵌入模型（避免重复加载）
_EMBEDDING_MODEL = None

def get_embedding_model():
    global _EMBEDDING_MODEL
    if _EMBEDDING_MODEL is None:
        _EMBEDDING_MODEL = HuggingFaceBgeEmbeddings(
            model_name=embedding_model_name,
            model_kwargs=model_kwargs,
            encode_kwargs=encode_kwargs,
            query_instruction=query_instruction
        )
    return _EMBEDDING_MODEL

def build_vector_db():
    """构建向量数据库（覆盖写入）"""
    # 删除旧数据库（Chroma 不支持高效更新）
    if os.path.exists(PERSIST_DIR):
        shutil.rmtree(PERSIST_DIR)
        os.makedirs(PERSIST_DIR)

    docs = load_documents()
    chunks = split_documents(docs)

    embeddings = get_embedding_model()
    print("🧠 正在生成向量...")
    Chroma.from_documents(
        documents=chunks,
        embedding=embeddings,
        persist_directory=PERSIST_DIR,
        collection_name="my_knowledge"
    )
    print("✅ 向量库重建完成")

第四步：创建图形界面（Streamlit）

创建 `app.py`

# app.py
import streamlit as st
from langchain_chroma import Chroma
from build_vector_db import get_embedding_model
from langchain_ollama import OllamaLLM
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser
import os

# === 配置 ===
PERSIST_DIR = "./chroma_db"
USE_RAG = True  # 设为 False 则只返回检索结果

# === 初始化 ===
@st.cache_resource
def get_retriever():
    embeddings = get_embedding_model()
    vectorstore = Chroma(
        persist_directory=PERSIST_DIR,
        embedding_function=embeddings,
        collection_name="my_knowledge"
    )
    return vectorstore.as_retriever(search_kwargs={"k": 3})

@st.cache_resource
def get_rag_chain():
    retriever = get_retriever()
    llm = OllamaLLM(model="qwen:4b", temperature=0.3)
    template = """
你是一个知识助手，请根据以下提供的上下文信息回答问题。
只使用上下文中的内容，不要编造。如果上下文没有相关信息，请回答“根据现有资料无法回答”。

上下文：
{context}

问题：{question}
答案：
"""
    prompt = ChatPromptTemplate.from_template(template)
    return (
        {"context": retriever, "question": RunnablePassthrough()}
        | prompt
        | llm
        | StrOutputParser()
    )

# === Streamlit UI ===
st.set_page_config(page_title="我的本地知识库", layout="wide")
st.title("🧠 我的本地知识库（完全离线）")

# 输入框
query = st.text_input("请输入你的问题：", placeholder="例如：项目的关键时间节点是什么？")

if query:
    with st.spinner("正在检索和思考..."):
        if USE_RAG:
            try:
                answer = get_rag_chain().invoke(query)
                st.subheader("✅ 答案")
                st.write(answer)
            except Exception as e:
                st.error(f"模型调用失败（请确认 Ollama 正在运行）：{str(e)}")
                USE_RAG = False  # 回退到仅检索
        
        # 显示检索到的原文（始终显示）
        retriever = get_retriever()
        docs = retriever.invoke(query)
        st.subheader("📚 相关原文片段")
        for i, doc in enumerate(docs):
            source = doc.metadata.get("source", "未知")
            st.markdown(f"**来源 {i+1}:** `{os.path.basename(source)}`")
            st.text_area("", doc.page_content, height=150, key=f"doc_{i}")

# 侧边栏
with st.sidebar:
    st.header("⚙️ 系统状态")
    db_exists = os.path.exists(os.path.join(PERSIST_DIR, "chroma.sqlite3"))
    st.success("🟢 向量库已加载") if db_exists else st.warning("🔴 向量库未构建")
    
    ollama_status = "🟢 Ollama 可用（若启用 RAG）" if USE_RAG else "⚪ 仅检索模式"
    st.write(ollama_status)
    
    st.info("📁 文档目录: `./docs`")
    st.info("🔒 所有数据仅保存在本地")

第五步：启动完整系统

创建启动脚本 `start.sh`（Linux/macOS）或 `start.bat`（Windows）

`start.sh`

#!/bin/bash
source venv/bin/activate

# 启动监控
python -c "
from auto_updater import start_watcher
import time
observer = start_watcher()
try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    observer.stop()
    observer.join()
" &
WATCHER_PID=$!

# 启动 Streamlit
streamlit run app.py --server.port=8501 --server.address=127.0.0.1

# 停止监控
kill $WATCHER_PID

Windows (`start.bat`)

@echo off
call venv\Scripts\activate

start /B python -c "from auto_updater import start_watcher; import time; o=start_watcher(); time.sleep(100000)" ^

streamlit run app.py --server.port=8501 --server.address=127.0.0.1

💡 更简单的做法：开两个终端分别运行：

# 终端 1：启动监控
python -c "from auto_updater import start_watcher; start_watcher(); import time; time.sleep(100000)"

# 终端 2：启动 Web 界面
streamlit run app.py

第六步：使用说明

启动系统：
```
chmod +x start.sh  # Linux/macOS
./start.sh
```
浏览器会自动打开 http://localhost:8501
添加/修改文档：
- 直接编辑 ./docs/ 下的 .md / .txt 文件
- 保存后，5 秒内自动重建向量库
- 刷新网页或重新提问即可看到新内容
隐私保障：
- Streamlit 默认只监听 127.0.0.1（仅本机访问）
- 无任何外网连接

🔧 可选增强

功能	实现方式
支持 PDF/Word	在 `ingest.py` 中启用 `Unstructured` 的 PDF/DOCX loader
显示文档修改时间	在 `auto_updater.py` 中记录变更日志
多用户（局域网）	修改 Streamlit 启动参数 `--server.address=0.0.0.0`（⚠️ 仅限可信内网）
加密存储	将整个项目目录放入 VeraCrypt 加密卷

✅ 总结

你现在拥有一个：

✅ 图形化界面（通过浏览器访问）
✅ 自动监控文档更新并重建知识库
✅ 支持检索 + 智能问答（RAG）
✅ 100% 本地运行，零数据上传
✅ 中文优化，隐私安全

的完整个人知识库系统！

💡 提示：首次运行会较慢（下载模型 + 构建向量库），之后所有操作均在本地快速完成。

在 ingest.py 中启用 Unstructured 的 PDF/DOCX loader

要在 ingest.py 中启用 Unstructured 的 PDF 和 DOCX（Word）文档加载功能，你需要：

安装额外的依赖（支持 PDF/DOCX 解析）
在 ingest.py 中添加对应的加载器
确保 Unstructured 使用本地模型（避免联网）

由于你要求完全本地、不上传任何信息到云端，我们将配置 Unstructured 使用 本地 OCR（如 Tesseract） 和 本地解析器，禁用其默认可能调用的远程 API（如 unstructured-api）。

✅ 第一步：安装必要的本地依赖

1.1 安装 Python 依赖

# 已有基础
pip install unstructured[local-inference]

# 新增：PDF 和 DOCX 支持
pip install unstructured[pdf]      # 包含 pdf2image, PyPDF2
pip install unstructured[docx]     # 包含 python-docx

# 可选但推荐（提高 PDF 表格/图文解析质量）
pip install pymupdf          # 更快的 PDF 解析（替代 pdf2image）
pip install poppler-utils     # 用于 pdf2image（需系统安装）

⚠️ 注意：unstructured[pdf] 默认会尝试使用 pdf2image，它依赖系统级的 Poppler。

1.2 安装系统级依赖（Windows）

（1）安装 Tesseract OCR（用于 PDF 中扫描图像的文字识别）

下载地址：https://github.com/UB-Mannheim/tesseract/wiki
安装时勾选 “Additional language data (e.g. Chinese)” → 选择 chi_sim（简体中文）和 chi_tra（繁体）
安装完成后，将 Tesseract 路径（如 C:\Program Files\Tesseract-OCR）加入系统 PATH 环境变量

（2）安装 Poppler（用于 `pdf2image`）

下载预编译版：https://github.com/oschwartz10612/poppler-windows/releases/
解压到 C:\poppler
将 C:\poppler\Library\bin 加入系统 PATH

✅ 验证安装：
tesseract --version
pdftoppm -h

✅ 第二步：修改 `ingest.py` 启用 PDF/DOCX 加载器

更新你的 ingest.py 文件如下（关键：禁用远程 API，强制本地解析）：

# ingest.py （更新版：支持 PDF / DOCX / TXT / MD，完全本地）

import os
from langchain_community.document_loaders import (
    DirectoryLoader,
    TextLoader,
    UnstructuredMarkdownLoader,
)
from langchain_community.document_loaders.unstructured import UnstructuredFileLoader
from typing import List
import warnings
warnings.filterwarnings("ignore", category=UserWarning, module="unstructured")

DOCS_DIR = "./docs"
PERSIST_DIR = "./chroma_db"

def load_documents() -> List:
    """加载所有支持的文档（TXT, MD, PDF, DOCX），完全本地解析"""
    documents = []

    # === .txt 文件 ===
    txt_loader = DirectoryLoader(
        DOCS_DIR,
        glob="*.txt",
        loader_cls=TextLoader,
        loader_kwargs={"encoding": "utf-8"}
    )
    documents.extend(txt_loader.load())

    # === .md 文件 ===
    md_loader = DirectoryLoader(
        DOCS_DIR,
        glob="*.md",
        loader_cls=UnstructuredMarkdownLoader,
        loader_kwargs={"encoding": "utf-8"}
    )
    documents.extend(md_loader.load())

    # === .pdf 文件 ===
    pdf_files = [str(f) for f in Path(DOCS_DIR).glob("*.pdf")]
    for pdf_file in pdf_files:
        print(f"📄 正在解析 PDF: {os.path.basename(pdf_file)}")
        loader = UnstructuredFileLoader(
            pdf_file,
            mode="single",  # 或 "elements"（更细粒度）
            strategy="fast",  # "hi_res" 更准但慢，需 OCR
            # 关键：强制本地，不调用 API
            api_url=None,   # 禁用远程 API
            # 启用本地 OCR（如果 PDF 是扫描件）
            ocr_languages=["chi_sim", "eng"],  # 中英混合
        )
        documents.extend(loader.load())

    # === .docx 文件 ===
    docx_files = [str(f) for f in Path(DOCS_DIR).glob("*.docx")]
    for docx_file in docx_files:
        print(f"📝 正在解析 DOCX: {os.path.basename(docx_file)}")
        loader = UnstructuredFileLoader(
            docx_file,
            mode="single",
            strategy="fast",
            api_url=None,  # 禁用远程 API
        )
        documents.extend(loader.load())

    print(f"✅ 共加载 {len(documents)} 个文档片段")
    return documents

# --- 保留原有的 split_documents 函数 ---
from langchain_text_splitters import RecursiveCharacterTextSplitter
from pathlib import Path

def split_documents(documents):
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=500,
        chunk_overlap=50,
        length_function=len,
        is_separator_regex=False,
    )
    chunks = text_splitter.split_documents(documents)
    print(f"✅ 切分为 {len(chunks)} 个文本块")
    return chunks

if __name__ == "__main__":
    docs = load_documents()
    chunks = split_documents(docs)

    os.makedirs(PERSIST_DIR, exist_ok=True)
    with open(os.path.join(PERSIST_DIR, "chunks.txt"), "w", encoding="utf-8") as f:
        for i, chunk in enumerate(chunks):
            f.write(f"--- Chunk {i} ---\n")
            f.write(chunk.page_content + "\n\n")
    print("✅ 文本块已保存（供调试）")

✅ 第三步：验证配置（确保无网络请求）

UnstructuredFileLoader 默认在 api_url=None 时会使用 本地解析（partition 函数），不会连接任何远程服务。

你可以通过以下方式确认：

禁用网络（拔网线 / 关 WiFi）
运行 python ingest.py
如果能正常解析 PDF/DOCX → 说明完全本地

🔒 安全提示：Unstructured 的 UnstructuredAPIFileLoader 才会联网，我们用的是 UnstructuredFileLoader（本地版）。

✅ 第四步（可选）：提高 PDF 解析质量

如果你的 PDF 是扫描件（图片型），确保 Tesseract 已正确安装并支持中文：

# 在 UnstructuredFileLoader 中显式指定 OCR 选项
loader = UnstructuredFileLoader(
    pdf_file,
    strategy="hi_res",          # 启用高级布局分析
    ocr_languages="chi_sim+eng", # 中英混合
    include_page_breaks=False,
    api_url=None,
)

💡 strategy="hi_res" 会调用 layoutparser 和 Tesseract，需要额外安装：
pip install unstructured[local-inference]
# 已包含 layoutparser, detectron2（但 detectron2 在 Windows 安装复杂）
如果安装失败，可降级使用 strategy="ocr_only" 或 "fast"。

✅ 文件目录示例

local_knowledge_base/
└── docs/
    ├── notes.md
    ├── summary.txt
    ├── report.pdf        ← 支持！
    └── meeting.docx      ← 支持！

放入这些文件后，运行：

python ingest.py
python build_vector_db.py

即可将 PDF/DOCX 内容纳入知识库。

⚠️ 常见问题解决

问题	解决方案
`TesseractNotFoundError`	安装 Tesseract 并加入 `PATH`
`poppler not installed`	安装 Poppler 并加入 `PATH`
中文 OCR 识别乱码	安装 `chi_sim` 语言包（Tesseract 安装时勾选）
DOCX 表格丢失	`unstructured` 对复杂表格支持有限，可考虑转为 PDF 或 Markdown 预处理

✅ 总结

你现在已经成功在 ingest.py 中启用了：

✅ PDF 解析（文本型 + 扫描型 OCR）
✅ DOCX 解析（文字、标题、列表）
✅ 完全本地运行
✅ 不上传任何数据到云端

所有文档（包括 PDF/DOCX）都会被切块、向量化，并纳入你的本地知识库，可通过图形界面查询。

Read All

1/37

Welcome to tyron's blog

DSP与音频混音mixer

一. 音频数据相加与dsp实现

1. 核心原理

2. 常见实现方法与算法

3. 注意事项

4. DSP（数字信号处理）中的经典应用

5. 硬件级优化考量

6. DSP 芯片级/底层算法的实现考量

① MAC 单元（乘累加运算）

② 固定小数点（Fixed-point） vs 浮点数（Floating-point）

③ SIMD（单指令多数据）加速

二. 处理实时流（buffer）数据时，需要注意什么？

1. 实时流处理的核心步骤

2. 推荐的实时流混音算法

方案 A：Float32 累加 + 软限幅（推荐）

方案 B：定点数（Int16）直接饱和相加

3. Python 实时流混音示例（基于 NumPy）

4. 实时流混音的避坑指南

三. 当用的编程语言是 c，实时流是网络传输数据，如何处理？

1. 核心架构设计

2. 核心 C 语言代码实现

3. 网络流混音的 3 大致命坑与 C 语言解决方案

① 丢包与断流（Packet Loss & Discontinuity）

② 时钟异步与漂移（Clock Drift）

③ 突发加入的“啪”声（Click Noise）

4. 推荐使用的开源 C 库

四. 需要支持多路（大于2路）网络音频同时混合

1. 数据结构设计：管理多路流

2. 多路混音核心 C 代码

3. 工程化应用示例（伪代码）

4. 多路网络混音的商业级设计要点

vscode 插件 Git Graph 使用(转)

在 Ubuntu 22.04上搭建TFTP服务器和NFS服务器

一、 搭建 TFTP 服务器

1. 安装 TFTP 相关软件

2. 创建 TFTP 工作目录

3. 修改配置文件

4. 重启并检查服务

5. 本地测试 TFTP

二、 搭建 NFS 服务器

1. 安装 NFS 服务端

2. 创建 NFS 共享目录

3. 配置共享目录权限

4. 解决 Ubuntu 22.04 的 NFS 版本兼容问题（重要 ⚠️）

5. 重启 NFS 服务

6. 本地测试 NFS

三、 常见注意事项

Github上这款免费开源的Bash脚本教程

更改Linux/Unix 环境变量

方案 1：使用 source 命令（推荐！）

方案 2：在 shell 配置文件中永久设置

构建自己的本地知识库

一、整体架构（完全本地化）

二、关键组件与工具推荐（开源 + 本地运行）

1. 文档解析与预处理

2. 文本分块（Chunking）

3. 文本向量化（Embedding）

4. 本地向量数据库

5. 本地大语言模型（用于生成答案）

6. 检索增强生成（RAG）本地实现

7. 用户界面（可选）

三、数据安全与隐私保障

四、简易搭建流程（示例）

五、进阶建议

总结

🧰 第一步：准备工作

1.1 安装 Python（建议 3.10 或 3.11）

1.2 创建项目目录

1.3 创建虚拟环境（推荐）

1.4 安装依赖包

📁 第二步：准备你的知识文档

🧩 第三步：编写文档加载与分块脚本

🧠 第四步：构建本地向量数据库（使用 BGE 中文嵌入模型）

🔍 第五步：本地语义检索（不依赖大模型，直接返回相关段落）

🤖 第六步（可选）：使用本地大模型生成答案（RAG）

6.1 安装 Ollama（https://ollama.com/）

6.2 拉取本地中文模型（首次需要联网，之后离线可用）

6.3 创建 RAG 查询脚本：rag_query.py

🔒 隐私与安全说明

一、搭建 TFTP 服务器

二、搭建 NFS 服务器

三、常见注意事项

6.3 创建 RAG 查询脚本：`rag_query.py`

创建 `auto_updater.py`

第三步：改造 `build_vector_db.py`（支持被调用）

创建 `app.py`

创建启动脚本 `start.sh`（Linux/macOS）或 `start.bat`（Windows）

`start.sh`

Windows (`start.bat`)

（2）安装 Poppler（用于 `pdf2image`）

✅ 第二步：修改 `ingest.py` 启用 PDF/DOCX 加载器