一、引言
你是否曾经在堆积如山的本地文件中苦苦寻找一个文档，却因为记不清文件名而束手无策？你是否幻想过对着电脑说一句“搜索文件修改时间为2026年1月至5月的PDF文件”，文件就自动出现在眼前？
本文介绍了一套基于海光DCU K100_AI国产算力平台打造的语音驱动的智能文件搜索系统。系统后端部署了Qwen3-ASR-1.7B语音识别模型和Qwen3.5-9B大语言模型，前端则是一个运行在Windows上的Python GUI程序，集成了Everything搜索引擎的SDK。
用户只需对着麦克风说出自然语言指令（如“搜索文件修改时间为2026年1月至5月的PDF文件”），系统便会自动完成语音识别→语义理解→Everything搜索→结果展示的完整链路。整个过程无需键盘输入，真正实现了“动口不动手” 。
海光DCU K100_AI单卡拥有64GB显存，FP16算力达196 TFLOPS，配合海光DTK 26.04工具包和vLLM推理框架，能够稳定流畅地同时承载ASR和大模型的推理服务。本文将完整介绍系统架构、部署方案、核心代码以及踩坑经验，希望能为国产算力平台上的AI应用开发提供一份实用的参考。
二、系统方案设计
2.1 总体架构
系统采用客户端-服务端分离架构：
-
服务端（海光DCU服务器，Ubuntu 22.04，DTK 26.04，K100_AI）：部署Qwen3-ASR-1.7B和Qwen3.5-9B两个模型，分别提供语音识别和语义理解能力。
-
客户端（Windows PC）：运行Python GUI程序，负责麦克风音频采集、调用服务端API、执行Everything本地搜索并展示结果。
2.2 核心工作流程
第一步：语音采集与VAD静音检测
客户端通过PyAudio库从麦克风采集16kHz单声道PCM音频流。使用WebRTC VAD（Voice Activity Detection）库实时检测语音活动，将连续语音帧缓存为音频段，当检测到连续静音帧数超过阈值（默认30帧）时，判定一个完整的语音指令结束。
第二步：ASR语音识别
语音段被封装为WAV格式，通过HTTP POST请求发送至Qwen3-ASR-1.7B服务端的/v1/audio/transcriptions接口。模型返回转录文本后，客户端会过滤掉预设的噪声词表（如“嗯”、“啊”、“uh”等），避免误触发搜索。
第三步：大模型语义理解与工具调用
将ASR输出的文本作为用户查询，送入Qwen3.5-9B大模型。模型被预先注入了search_files工具定义，包含query（Everything搜索语法）、max_results、sort_by、sort_order等参数。大模型自动判断用户意图，生成符合Everything高级语法的搜索语句——例如“找一下2026年创建的关于深度学习的PDF文件”会被转换为
ext:pdf datecreated:2026/1/1-2026/5/31 深度学习
第四步：Everything高速本地搜索
客户端通过ctypes直接调用Everything64.dll的API，执行大模型生成的搜索语句。Everything利用NTFS的USN日志机制实现毫秒级文件检索，结果包含文件名、路径、大小、修改时间、创建时间等完整元数据。程序是在windows操作系统上运行，须安装Everything 64位版本软件，并下载SDK包：Everything-SDK.zip，下载链接为：https://www.voidtools.com/Everything-SDK.zip，解压后提取dll目录里面的Everything64.dll。
第五步：结果展示与交互
搜索结果以表格形式呈现在GUI中，支持按列排序、双击打开文件、右键复制路径、打开文件所在位置以及“打开方式”对话框等丰富操作。
2.3 技术选型亮点
-
Everything SDK：直接调用DLL而非HTTP服务，响应速度更快、无额外端口占用。
-
vLLM推理框架：支持PagedAttention和Continuous Batching，显著提升显存利用率和推理吞吐。
-
WebRTC VAD：轻量级实时语音活动检测，无需额外模型，适合客户端本地运行。
三、系统工作流程图
流程图详解
1. 用户语音输入 → 麦克风采集
用户对着麦克风说出自然语言指令（例如：“找一下去年项目验收的PDF”）。客户端通过 PyAudio 库以 16kHz 采样率、16-bit 位深、单声道格式实时读取音频流。
2. VAD 静音检测（语音活动检测）
每 30ms 一帧的音频数据送入 WebRTC VAD 模块（敏感度设为 mode=1）。系统会：
-
持续累积被判定为“语音”的帧；
-
一旦检测到连续
30帧（约 0.9 秒）的静音，即认为用户说完了一整句话，触发语音段结束事件。
该机制有效避免环境噪声误触发，同时保证较快的响应速度。
3. 封装为 WAV 并发送 ASR 服务
将累积的语音帧合成为一个完整的音频段，封装成标准 WAV 格式（16000Hz / 单声道 / 16-bit PCM），通过 HTTP POST 请求发送给服务端的 Qwen3-ASR-1.7B（监听 8084 端口）。
4. ASR 转录与文本后处理
服务端返回的原始文本可能带有 <|zh|><asr_text> 等语言标记前缀，客户端会使用正则表达式 re.sub(r'^[^<]*<asr_text>\s*', '', raw_text) 将其去除。随后，文本会与预定义的噪声词表（包括“嗯”、“啊”、“uh”、“um”等）进行匹配，若命中则直接丢弃，避免无意义的搜索请求。
5. 大模型语义理解（Qwen3.5-9B）
若文本有效，客户端将其作为用户消息，连同预定义的 search_files 工具定义（包含 query、max_results、sort_by、sort_order 参数）一起发送至 Qwen3.5-9B 服务（监听 8085 端口）。该模型启用了 --enable-auto-tool-choice，能够自动判断需要调用工具，并生成符合 Everything 高级语法的查询字符串。
例如：用户说“找一下2026年创建的关于深度学习的PDF”，模型会生成
ext:pdf datecreated:2026/1/1-2026/12/31 深度学习。
6. 解析工具调用参数
客户端接收到模型的 tool_call 响应后，解析 JSON 参数，提取出最终的 query、max_results（默认 1000）、排序字段和排序方向。
7. Everything 本地高速搜索
客户端通过 ctypes 直接调用 Everything64.dll 的 API：
-
Everything_SetSearchW(query)设置查询条件； -
Everything_SetMax(max_results)限制结果数； -
Everything_QueryW(True)执行查询； -
循环
Everything_GetNumResults()获取每条结果的文件名、路径、大小、修改时间和创建时间。
整个过程利用 Everything 的 NTFS USN 日志机制，实现毫秒级检索。
8. 结果展示与用户交互
搜索结果以表格形式呈现在 Tkinter Treeview 中，包含五列：文件名、路径、大小、修改时间、创建时间。用户可进行以下操作：
-
双击任意行：使用系统默认程序打开文件；
-
右键菜单：复制完整路径、打开文件所在位置（
explorer /select）、弹出“打开方式”对话框（通过 PowerShell 调用Shell.Application）。
| 颜色块 | 表示阶段 | 含义 |
|---|---|---|
| 金色 | 用户输入 | 语音指令起点 |
| 蓝色 | 客户端采集与 VAD | 本地音频处理和静音检测 |
| 绿色 | ASR 识别 | 语音到文本的转换（服务端） |
| 橙色 | 大模型理解 | 语义解析和工具调用生成（服务端） |
| 紫色 | Everything 搜索 | 本地高速文件检索 |
| 红色 | 结果操作 | 用户与搜索结果的交互 |
四、海光DCU环境部署
4.1 硬件环境
| 项目 | 配置 |
|---|---|
| CPU | 海光 x86架构处理器 |
| GPU | 海光DCU K100_AI × 8（单卡64GB显存） |
| 算力 | 单卡FP16 196 TFLOPS |
| 显存带宽 | 896 GB/s |
| 操作系统 | Ubuntu 22.04 |
| DCU工具包 | DTK 26.04 |
4.2 Qwen3-ASR-1.7B 部署
使用的Docker镜像：
harbor.sourcefind.cn:5443/dcu/admin/base/custom:vllm0.15.1-ubuntu22.04-dtk26.04-0130-py3.10-20260220
⚠️ 关键注意：Qwen3-ASR-1.7B部署时，transformers版本必须小于5，推荐使用 4.57.6。否则会导致模型加载失败或推理异常。
启动脚本：
HIP_VISIBLE_DEVICES=6 vllm serve '/home/models/Qwen3-ASR-1.7B' \
--trust-remote-code \
--gpu-memory-utilization 0.3 \
--limit-mm-per-prompt '{"audio": 1}' \
--port 8084 \
--served-model-name Qwen3-ASR-1.7B
参数说明：
-
HIP_VISIBLE_DEVICES=6：指定使用第6号DCU显卡 -
--gpu-memory-utilization 0.3：显存利用率限制为30%，为其他服务预留空间 -
--limit-mm-per-prompt '{"audio": 1}'：每个请求最多处理1个音频文件 -
--port 8084：服务监听端口
Qwen3-ASR-1.7B是Qwen团队开源的自动语音识别模型，支持30种语言和22种中文方言的识别，在Librispeech等公开基准测试中达到SOTA水平。
4.3 Qwen3.5-9B 部署
使用的Docker镜像：
42.228.13.241:5000/jenkins/model_test_env/vllm:0.21.0-ubuntu22.04-dtk26.04-py3.10-20260624-2240
启动脚本 ：
#!/bin/bash
export VLLM_SPEC_DECODE_EAGER=1
export VLLM_MLA_DISABLE=0
export VLLM_USE_FLASH_MLA=1
export VLLM_RPC_TIMEOUT=1800000
export HIP_VISIBLE_DEVICES=6
export ALLREDUCE_STREAM_WITH_COMPUTE=1
# 海光CPU绑定核，通过 hy-smi --showtopo 参考numa节点
export VLLM_NUMA_BIND=1
export VLLM_RANK0_NUMA=0
export VLLM_RANK1_NUMA=0
export VLLM_RANK2_NUMA=0
export VLLM_RANK3_NUMA=0
export VLLM_RANK4_NUMA=0
export VLLM_RANK5_NUMA=0
export VLLM_RANK6_NUMA=0
export VLLM_RANK7_NUMA=0
export NCCL_MAX_NCHANNELS=16
export NCCL_MIN_NCHANNELS=16
vllm serve "/home/models/Qwen3.5-9B" \
--gpu-memory-utilization 0.4 \
--port 8085 \
--max-model-len 32768 \
--max-num-seqs 32 \
--served-model-name Qwen3.5-9B \
--tensor-parallel-size 1 \
--enable-auto-tool-choice \
--tool-call-parser qwen3_xml \
--trust-remote-code \
--enable-prefix-caching \
--enable-chunked-prefill \
--api-key PassWord@123456
关键参数说明：
-
--enable-auto-tool-choice：启用自动工具选择功能，让模型能够自主决定是否调用工具 -
--tool-call-parser qwen3_xml：指定Qwen3系列的XML格式工具调用解析器 -
--enable-prefix-caching：启用前缀缓存，提升重复查询的推理效率 -
--enable-chunked-prefill：启用分块预填充，优化长文本处理 -
--api-key PassWord@123456：API访问密钥，客户端需保持一致
NUMA绑定说明：海光CPU为多NUMA节点架构，通过VLLM_NUMA_BIND=1和VLLM_RANK*_NUMA环境变量将vLLM进程绑定到特定NUMA节点，可显著降低内存访问延迟，提升推理性能
五、部署踩坑与解决
5.1 transformers版本不兼容
问题：部署Qwen3-ASR-1.7B时，使用最新版transformers（≥5.0）导致模型加载失败，报错涉及Qwen3ASRModel类定义不兼容。
解决：将transformers版本降级至 4.57.6：
pip install transformers==4.57.6
5.2 音频格式不匹配
问题：客户端发送的WAV音频采样率、声道数、位深与服务端预期不一致，导致转录返回空结果或乱码。
解决：客户端须严格遵循服务端要求：
采样率：16000 Hz
声道数：单声道（Mono）
采样位深：16-bit PCM
音频格式：WAV封装
5.3 ASR返回文本包含多余前缀
问题：Qwen3-ASR的转录结果有时会包含 <|zh|><asr_text> 之类的语言标记前缀，导致后续大模型解析异常。
解决：在客户端增加正则过滤：
import re
clean_text = re.sub(r'^[^<]*<asr_text>\s*', '', raw_text).strip()
5.4 vLLM在海光DCU上的性能调优
问题：vLLM在海光DCU上运行时的默认参数可能导致显存利用率不足或推理延迟偏高。
解决：
-
合理设置
--gpu-memory-utilization（ASR用0.3，大模型用0.4），为两个服务在同一张卡上共存预留空间 -
启用
--enable-prefix-caching和--enable-chunked-prefill提升推理效率 -
通过
VLLM_NUMA_BIND=1绑定NUMA节点，减少跨节点内存访问延迟
5.5 Everything64.dll加载失败
问题：Python通过ctypes加载Everything64.dll时报错“找不到指定的模块”。
解决：
-
确保Everything软件已在Windows上安装并运行
-
将Everything64.dll放置在脚本所在目录或系统PATH中
-
使用绝对路径加载：
ctypes.WinDLL(r"D:\audio_search\Everything64.dll")
六、程序代码及运行截图
6.1 程序完整代码
import asyncio
import json
import os
import re
import threading
import traceback
import subprocess
import tkinter as tk
from tkinter import ttk, scrolledtext, messagebox
from openai import OpenAI
import ctypes
from ctypes import wintypes
import datetime
# ==================== 语音识别所需导入 ====================
import pyaudio
import webrtcvad
import requests
import io
import wave
import collections
# ==================== 配置加载 ====================
DEFAULT_CONFIG = {
"everything_dll": r"D:\audio_search\Everything64.dll",
"llm_base_url": "http://192.168.222.65:8085/v1",
"llm_model": "Qwen3.5-9B",
"llm_api_key": "PassWord@123456",
"max_results_limit": 1000,
"enable_thinking": False,
# ASR 语音识别配置（vLLM /v1/audio/transcriptions 接口）
"asr_api_url": "http://192.168.222.65:8084/v1/audio/transcriptions",
"asr_model": "Qwen3-ASR-1.7B",
"asr_api_key": "PassWord@123456",
"asr_sample_rate": 16000,
"asr_frame_duration": 30, # VAD 帧长（毫秒），仅支持 10/20/30
"asr_vad_mode": 1, # VAD 敏感度 0~3
"asr_silence_thresh": 30 # 连续静音帧数后认为语音结束
}
CONFIG_FILE = "config.json"
def load_config():
if os.path.exists(CONFIG_FILE):
try:
with open(CONFIG_FILE, 'r', encoding='utf-8') as f:
config = json.load(f)
for key, value in DEFAULT_CONFIG.items():
config.setdefault(key, value)
return config
except Exception as e:
print(f"读取配置文件失败: {e}，将使用默认配置")
return DEFAULT_CONFIG.copy()
else:
try:
with open(CONFIG_FILE, 'w', encoding='utf-8') as f:
json.dump(DEFAULT_CONFIG, f, indent=4, ensure_ascii=False)
print(f"已创建默认配置文件: {CONFIG_FILE}")
except Exception as e:
print(f"无法创建配置文件: {e}")
return DEFAULT_CONFIG.copy()
config = load_config()
EVERYTHING_DLL = config["everything_dll"]
LLM_BASE_URL = config["llm_base_url"]
LLM_MODEL = config["llm_model"]
LLM_API_KEY = config["llm_api_key"]
MAX_RESULTS_LIMIT = config["max_results_limit"]
# ASR 参数
ASR_API_URL = config["asr_api_url"]
ASR_MODEL = config["asr_model"]
ASR_API_KEY = config["asr_api_key"]
ASR_SAMPLE_RATE = config["asr_sample_rate"]
ASR_FRAME_DURATION = config["asr_frame_duration"]
ASR_VAD_MODE = config["asr_vad_mode"]
ASR_SILENCE_THRESH = config["asr_silence_thresh"]
# 噪声词表（可调整）
NOISE_WORDS = {"¿Qué?", "No.", "ok.", "Yeah.", "hmm", "uh", "um", "ah", "oh", "mhm", "嗯。","嗯","啊。", "啊", "哦。","哦","哎。"}
# ==================== 路径处理工具函数 ====================
def build_full_path(dir_path, file_name):
"""安全拼接路径，修复 Windows 盘符缺少反斜杠的问题（例如 'D:' -> 'D:\\'）"""
if not dir_path and not file_name:
return None
if not file_name:
return dir_path
if not dir_path:
return file_name
if os.name == 'nt' and re.match(r'^[A-Za-z]:$', dir_path):
dir_path += '\\'
if os.path.basename(dir_path) == file_name:
return dir_path
return os.path.join(dir_path, file_name)
# ==================== Everything SDK 直接调用封装 ====================
everything_dll = ctypes.WinDLL(EVERYTHING_DLL)
everything_dll.Everything_SetSearchW.argtypes = [wintypes.LPCWSTR]
everything_dll.Everything_SetSearchW.restype = None
everything_dll.Everything_SetMax.argtypes = [wintypes.DWORD]
everything_dll.Everything_SetMax.restype = None
everything_dll.Everything_SetSort.argtypes = [wintypes.DWORD]
everything_dll.Everything_SetSort.restype = None
everything_dll.Everything_SetRequestFlags.argtypes = [wintypes.DWORD]
everything_dll.Everything_SetRequestFlags.restype = None
everything_dll.Everything_QueryW.argtypes = [wintypes.BOOL]
everything_dll.Everything_QueryW.restype = wintypes.BOOL
everything_dll.Everything_GetNumResults.argtypes = []
everything_dll.Everything_GetNumResults.restype = wintypes.DWORD
everything_dll.Everything_GetResultFileNameW.argtypes = [wintypes.DWORD]
everything_dll.Everything_GetResultFileNameW.restype = wintypes.LPCWSTR
everything_dll.Everything_GetResultPathW.argtypes = [wintypes.DWORD]
everything_dll.Everything_GetResultPathW.restype = wintypes.LPCWSTR
everything_dll.Everything_GetResultSize.argtypes = [wintypes.DWORD, ctypes.POINTER(ctypes.c_longlong)]
everything_dll.Everything_GetResultSize.restype = wintypes.BOOL
everything_dll.Everything_GetResultDateModified.argtypes = [wintypes.DWORD, ctypes.POINTER(ctypes.c_longlong)]
everything_dll.Everything_GetResultDateModified.restype = wintypes.BOOL
everything_dll.Everything_GetResultDateCreated.argtypes = [wintypes.DWORD, ctypes.POINTER(ctypes.c_longlong)]
everything_dll.Everything_GetResultDateCreated.restype = wintypes.BOOL
EVERYTHING_SORT_NAME_ASC = 1
EVERYTHING_SORT_NAME_DESC = 2
EVERYTHING_SORT_PATH_ASC = 3
EVERYTHING_SORT_PATH_DESC = 4
EVERYTHING_SORT_SIZE_ASC = 5
EVERYTHING_SORT_SIZE_DESC = 6
EVERYTHING_SORT_DATE_MODIFIED_ASC = 13
EVERYTHING_SORT_DATE_MODIFIED_DESC = 14
EVERYTHING_SORT_DATE_CREATED_ASC = 11
EVERYTHING_SORT_DATE_CREATED_DESC = 12
EVERYTHING_REQUEST_FILE_NAME = 0x1
EVERYTHING_REQUEST_PATH = 0x2
EVERYTHING_REQUEST_SIZE = 0x10
EVERYTHING_REQUEST_DATE_MODIFIED = 0x40
EVERYTHING_REQUEST_DATE_CREATED = 0x80
def search_files_with_everything(query, max_results=1000, sort_by=None, sort_order="ascending"):
sort_map = {
("name", "ascending"): EVERYTHING_SORT_NAME_ASC,
("name", "descending"): EVERYTHING_SORT_NAME_DESC,
("path", "ascending"): EVERYTHING_SORT_PATH_ASC,
("path", "descending"): EVERYTHING_SORT_PATH_DESC,
("size", "ascending"): EVERYTHING_SORT_SIZE_ASC,
("size", "descending"): EVERYTHING_SORT_SIZE_DESC,
("date_modified", "ascending"): EVERYTHING_SORT_DATE_MODIFIED_ASC,
("date_modified", "descending"): EVERYTHING_SORT_DATE_MODIFIED_DESC,
("date_created", "ascending"): EVERYTHING_SORT_DATE_CREATED_ASC,
("date_created", "descending"): EVERYTHING_SORT_DATE_CREATED_DESC,
}
sort_key = (sort_by, sort_order) if sort_by else ("name", "ascending")
sort_type = sort_map.get(sort_key, EVERYTHING_SORT_NAME_ASC)
everything_dll.Everything_SetSort(sort_type)
everything_dll.Everything_SetRequestFlags(
EVERYTHING_REQUEST_FILE_NAME |
EVERYTHING_REQUEST_PATH |
EVERYTHING_REQUEST_SIZE |
EVERYTHING_REQUEST_DATE_MODIFIED |
EVERYTHING_REQUEST_DATE_CREATED
)
everything_dll.Everything_SetSearchW(query)
everything_dll.Everything_SetMax(max_results)
if not everything_dll.Everything_QueryW(True):
return []
num = everything_dll.Everything_GetNumResults()
results = []
for i in range(num):
name = everything_dll.Everything_GetResultFileNameW(i)
path = everything_dll.Everything_GetResultPathW(i)
size_val = ctypes.c_longlong()
everything_dll.Everything_GetResultSize(i, ctypes.byref(size_val))
modified_val = ctypes.c_longlong()
everything_dll.Everything_GetResultDateModified(i, ctypes.byref(modified_val))
created_val = ctypes.c_longlong()
everything_dll.Everything_GetResultDateCreated(i, ctypes.byref(created_val))
date_modified = ""
date_created = ""
if modified_val.value:
try:
date_modified = datetime.datetime.fromtimestamp(modified_val.value / 10000000 - 11644473600).strftime("%Y-%m-%d %H:%M:%S")
except:
pass
if created_val.value:
try:
date_created = datetime.datetime.fromtimestamp(created_val.value / 10000000 - 11644473600).strftime("%Y-%m-%d %H:%M:%S")
except:
pass
if not date_created:
try:
full_path = build_full_path(path, name)
if full_path and os.path.exists(full_path):
ctime = os.path.getctime(full_path)
date_created = datetime.datetime.fromtimestamp(ctime).strftime("%Y-%m-%d %H:%M:%S")
except Exception:
pass
results.append({
"Name": name or "",
"Path": path or "",
"Size": size_val.value,
"DateModified": date_modified,
"DateCreated": date_created
})
return results
# ==================== 大模型调用与工具定义 ====================
client = OpenAI(base_url=LLM_BASE_URL, api_key=LLM_API_KEY)
TOOLS = [
{
"type": "function",
"function": {
"name": "search_files",
"description": (
"使用 Everything 搜索引擎查找本地文件。"
"query 支持 Everything 高级查询语法，例如 ext:pdf、content:关键词、datecreated:、size:>1mb 等。"
"sort_by 可选：name, path, size, date_modified, date_created"
"sort_order 可选：ascending, descending"
"示例：搜索创建时间为2024年文件名包含“逆向”的docx，生成的query：ext:docx datecreated:2024/1/1-2024/12/31 逆向"
),
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "Everything 查询字符串"},
"max_results": {"type": "integer", "description": "最大结果数，默认1000，最大1000"},
"sort_by": {"type": "string", "enum": ["name", "path", "size", "date_modified", "date_created"]},
"sort_order": {"type": "string", "enum": ["ascending", "descending"]}
},
"required": ["query"]
}
}
}
]
def process_user_query(user_input: str, enable_thinking: bool = True):
debug_lines = []
messages = [
{"role": "system", "content": (
"你是一个文件搜索助手。根据用户的自然语言生成 Everything 搜索参数。"
"query 中可直接使用 Everything 高级语法。max_results 不填则默认1000。"
)},
{"role": "user", "content": user_input}
]
try:
extra_body = {
"chat_template_kwargs": {
"enable_thinking": enable_thinking
}
}
response = client.chat.completions.create(
model=LLM_MODEL,
messages=messages,
tools=TOOLS,
tool_choice="auto",
extra_body=extra_body
)
except Exception as e:
debug_lines.append(f"大模型调用失败: {e}")
return None, "\n".join(debug_lines), f"调用大模型失败: {e}"
choice = response.choices[0]
debug_lines.append(f"模型原始响应: {choice.message}")
if not choice.message.tool_calls:
direct_reply = choice.message.content or "（无文本回复）"
debug_lines.append(f"大模型直接回复: {direct_reply}")
return None, "\n".join(debug_lines), "大模型未生成搜索指令，请更明确地描述。"
tool_call = choice.message.tool_calls[0]
if tool_call.function.name != "search_files":
debug_lines.append(f"未知工具: {tool_call.function.name}")
return None, "\n".join(debug_lines), f"大模型调用了未知工具: {tool_call.function.name}"
try:
arguments = json.loads(tool_call.function.arguments)
except json.JSONDecodeError as e:
debug_lines.append(f"参数 JSON 解析失败: {e}")
return None, "\n".join(debug_lines), f"参数解析失败: {e}"
query = arguments.get("query")
max_results = min(arguments.get("max_results", 1000), MAX_RESULTS_LIMIT)
sort_by = arguments.get("sort_by")
sort_order = arguments.get("sort_order", "ascending")
debug_lines.append(f"生成的搜索语句: {query}")
debug_lines.append(f"最大结果数: {max_results}")
debug_lines.append(f"排序: {sort_by} ({sort_order})")
debug_lines.append("正在执行本地搜索...")
try:
results = search_files_with_everything(query, max_results, sort_by, sort_order)
error = None
except Exception:
results = []
error = traceback.format_exc()
if error:
debug_lines.append(f"搜索异常: {error}")
return None, "\n".join(debug_lines), f"搜索执行错误: {error}"
debug_lines.append(f"返回结果数: {len(results)}")
if results:
first_keys = list(results[0].keys())
debug_lines.append(f"第一条记录的键: {first_keys}")
debug_lines.append("前3条结果预览:")
for item in results[:3]:
debug_lines.append(str(item))
return results, "\n".join(debug_lines), None
# ==================== GUI 界面（使用 HTTP ASR 接口） ====================
class SearchGUI:
def __init__(self, root):
self.root = root
root.title("本地大模型文件搜索 + 语音识别")
root.geometry("1100x850")
# ---------- 文件搜索原有状态 ----------
self.sort_column = None
self.sort_reverse = False
self.size_map = {}
self.file_paths = {}
self.right_click_iid = None
self.thinking_var = tk.BooleanVar(value=config.get("enable_thinking", False))
# ---------- 语音识别状态 ----------
self.asr_active = False # 识别是否运行中
self.asr_thread = None
self.asr_stop_event = threading.Event()
self.asr_session = requests.Session()
if ASR_API_KEY:
self.asr_session.headers['Authorization'] = f'Bearer {ASR_API_KEY}'
# 语音搜索结果防抖定时器 ID
self._asr_search_after_id = None
# ---------- 文件搜索输入区域 ----------
frame_input = ttk.Frame(root, padding="5")
frame_input.pack(fill=tk.X, padx=10, pady=5)
ttk.Label(frame_input, text="输入查询（自然语言）：").pack(side=tk.LEFT)
self.query_var = tk.StringVar()
self.entry_query = ttk.Entry(frame_input, textvariable=self.query_var, width=60)
self.entry_query.pack(side=tk.LEFT, padx=5, fill=tk.X, expand=True)
self.entry_query.bind("<Return>", lambda e: self.start_search())
self.btn_search = ttk.Button(frame_input, text="搜索", command=self.start_search)
self.btn_search.pack(side=tk.LEFT, padx=5)
self.chk_thinking = ttk.Checkbutton(
frame_input,
text="启用思考模式",
variable=self.thinking_var
)
self.chk_thinking.pack(side=tk.LEFT, padx=5)
self.btn_clear = ttk.Button(frame_input, text="清空结果", command=self.clear_results)
self.btn_clear.pack(side=tk.LEFT, padx=5)
# ---------- 语音识别区域 ----------
asr_frame = ttk.LabelFrame(root, text="实时语音识别 (vLLM ASR)", padding="5")
asr_frame.pack(fill=tk.X, padx=10, pady=5)
asr_btn_frame = ttk.Frame(asr_frame)
asr_btn_frame.pack(fill=tk.X, pady=2)
self.asr_stream_btn = ttk.Button(asr_btn_frame, text="开始识别", command=self.toggle_asr_stream)
self.asr_stream_btn.pack(side=tk.LEFT, padx=5)
self.asr_status_label = ttk.Label(asr_btn_frame, text="空闲", foreground="red")
self.asr_status_label.pack(side=tk.LEFT, padx=10)
self.asr_text = scrolledtext.ScrolledText(asr_frame, height=5, wrap=tk.WORD, state=tk.DISABLED)
self.asr_text.pack(fill=tk.BOTH, expand=True)
# ---------- 状态标签（全局） ----------
self.status_var = tk.StringVar(value="就绪")
ttk.Label(root, textvariable=self.status_var, foreground="gray").pack(anchor=tk.W, padx=10)
# ---------- 文件搜索结果 + 调试信息 ----------
main_paned = ttk.PanedWindow(root, orient=tk.VERTICAL)
main_paned.pack(fill=tk.BOTH, expand=True, padx=10, pady=5)
frame_result = ttk.Frame(main_paned)
main_paned.add(frame_result, weight=3)
columns = ("name", "path", "size", "date_modified", "date_created")
self.tree = ttk.Treeview(frame_result, columns=columns, show="headings", selectmode="extended")
self.tree.heading("name", text="文件名")
self.tree.heading("path", text="路径")
self.tree.heading("size", text="大小")
self.tree.heading("date_modified", text="修改时间")
self.tree.heading("date_created", text="创建时间")
for col in columns:
self.tree.heading(col, command=lambda c=col: self.sort_by_column(c))
self.tree.column("name", width=200)
self.tree.column("path", width=350)
self.tree.column("size", width=80)
self.tree.column("date_modified", width=150)
self.tree.column("date_created", width=150)
scroll_y = ttk.Scrollbar(frame_result, orient=tk.VERTICAL, command=self.tree.yview)
scroll_x = ttk.Scrollbar(frame_result, orient=tk.HORIZONTAL, command=self.tree.xview)
self.tree.configure(yscrollcommand=scroll_y.set, xscrollcommand=scroll_x.set)
self.tree.grid(row=0, column=0, sticky="nsew")
scroll_y.grid(row=0, column=1, sticky="ns")
scroll_x.grid(row=1, column=0, sticky="ew")
frame_result.grid_rowconfigure(0, weight=1)
frame_result.grid_columnconfigure(0, weight=1)
self.tree.bind("<Double-1>", self.on_double_click)
self.tree.bind("<Button-3>", self.on_right_click)
self.context_menu = tk.Menu(self.tree, tearoff=0)
self.context_menu.add_command(label="复制完整路径和文件名", command=self.copy_full_path)
self.context_menu.add_command(label="打开文件所在位置", command=self.open_file_location)
self.context_menu.add_command(label="打开方式...", command=self.open_with_dialog)
frame_debug = ttk.Frame(main_paned)
main_paned.add(frame_debug, weight=1)
ttk.Label(frame_debug, text="调试信息：").pack(anchor=tk.W)
self.debug_text = scrolledtext.ScrolledText(frame_debug, height=8, state='normal')
self.debug_text.pack(fill=tk.BOTH, expand=True)
self.lbl_count = ttk.Label(root, text="", foreground="blue")
self.lbl_count.pack(anchor=tk.W, padx=10, pady=2)
self.update_column_headings()
# 绑定窗口关闭事件
root.protocol("WM_DELETE_WINDOW", self.on_close)
# ==================== 语音识别核心方法（HTTP 转录接口） ====================
def toggle_asr_stream(self):
"""开始/停止实时语音识别"""
if not self.asr_active:
self._start_asr()
else:
self._stop_asr()
def _start_asr(self):
"""启动后台识别线程"""
if self.asr_active:
return
self.asr_active = True
self.asr_stop_event.clear()
self.asr_stream_btn.config(text="停止识别")
self.asr_status_label.config(text="识别中", foreground="green")
self.asr_thread = threading.Thread(target=self._asr_worker, daemon=True)
self.asr_thread.start()
def _stop_asr(self):
"""停止识别线程"""
if not self.asr_active:
return
# 取消可能正在等待的搜索
if self._asr_search_after_id is not None:
self.root.after_cancel(self._asr_search_after_id)
self._asr_search_after_id = None
self.asr_active = False
self.asr_stop_event.set()
self.asr_stream_btn.config(text="开始识别")
self.asr_status_label.config(text="空闲", foreground="red")
def _asr_worker(self):
"""识别线程主函数：循环读取麦克风数据，VAD分割并发送"""
p = pyaudio.PyAudio()
chunk = int(ASR_SAMPLE_RATE * ASR_FRAME_DURATION / 1000) # 每帧样本数
vad = webrtcvad.Vad(ASR_VAD_MODE)
stream = None
try:
stream = p.open(format=pyaudio.paInt16,
channels=1,
rate=ASR_SAMPLE_RATE,
input=True,
frames_per_buffer=chunk)
# VAD 状态
voiced_frames = []
silence_counter = 0
speech_active = False
while not self.asr_stop_event.is_set():
try:
frame = stream.read(chunk, exception_on_overflow=False)
except Exception as e:
self._asr_log(f"音频读取错误: {e}")
break
is_speech = vad.is_speech(frame, ASR_SAMPLE_RATE)
if is_speech:
voiced_frames.append(frame)
silence_counter = 0
if not speech_active:
speech_active = True
else:
if speech_active:
voiced_frames.append(frame) # 保留尾部静音
silence_counter += 1
if silence_counter >= ASR_SILENCE_THRESH:
# 语音段结束，发送
self._send_audio_segment(voiced_frames)
voiced_frames = []
silence_counter = 0
speech_active = False
# 非激活状态的静音直接丢弃
except Exception as e:
self._asr_log(f"识别线程异常: {e}")
finally:
if stream:
stream.stop_stream()
stream.close()
p.terminate()
def _send_audio_segment(self, frame_list):
if not frame_list:
return
# 生成 WAV 字节流
wav_io = io.BytesIO()
with wave.open(wav_io, 'wb') as wf:
wf.setnchannels(1)
wf.setsampwidth(2) # 16-bit
wf.setframerate(ASR_SAMPLE_RATE)
wf.writeframes(b''.join(frame_list))
wav_io.seek(0)
try:
response = self.asr_session.post(
ASR_API_URL,
files={'file': ('audio.wav', wav_io, 'audio/wav')},
data={'model': ASR_MODEL},
timeout=10
)
response.raise_for_status()
result = response.json()
raw_text = result.get('text', '').strip()
# ---- 新增：去掉 "xxx<asr_text>" 前缀 ----
import re
clean_text = re.sub(r'^[^<]*<asr_text>\s*', '', raw_text).strip()
if not clean_text:
self._asr_log("[识别] <空结果或仅含标记>")
return
if self._is_noise(clean_text):
self._asr_log(f"🔇 已过滤噪声: {clean_text}")
return
self._asr_log(f"🎤 {clean_text}")
self._on_asr_result(clean_text)
except requests.exceptions.RequestException as e:
self._asr_log(f"[错误] 转录请求失败: {e}")
def _is_noise(self, text):
t = text.strip().lower().rstrip('.')
return t in NOISE_WORDS
def _asr_log(self, msg):
"""线程安全地在语音识别文本框中追加消息"""
def append():
self.asr_text.config(state=tk.NORMAL)
self.asr_text.insert(tk.END, msg + "\n")
self.asr_text.see(tk.END)
self.asr_text.config(state=tk.DISABLED)
self.root.after(0, append)
# ---------- 语音结果触发搜索（带防抖） ----------
def _on_asr_result(self, text):
if self._asr_search_after_id is not None:
self.root.after_cancel(self._asr_search_after_id)
self._asr_search_after_id = self.root.after(500, self._execute_search_from_asr, text)
def _execute_search_from_asr(self, text):
self._asr_search_after_id = None
self.query_var.set(text)
self.start_search()
# ==================== 原有文件搜索方法 ====================
def start_search(self):
query = self.query_var.get().strip()
if not query:
messagebox.showwarning("输入为空", "请输入查询条件")
return
thinking = self.thinking_var.get()
self.btn_search.config(state=tk.DISABLED)
self.status_var.set("正在与大模型交互...")
self.debug_text.delete("1.0", tk.END)
self.debug_text.insert(tk.END, "处理中...\n")
self.root.update()
def task():
results, debug_info, error = process_user_query(query, enable_thinking=thinking)
self.root.after(0, self.on_search_complete, results, debug_info, error)
threading.Thread(target=task, daemon=True).start()
def on_search_complete(self, results, debug_info, error):
self.btn_search.config(state=tk.NORMAL)
self.debug_text.delete("1.0", tk.END)
if debug_info:
self.debug_text.insert(tk.END, debug_info)
for row in self.tree.get_children():
self.tree.delete(row)
self.size_map.clear()
self.file_paths.clear()
self.sort_column = None
self.sort_reverse = False
self.update_column_headings()
if error:
self.status_var.set(f"错误: {error}")
messagebox.showerror("搜索失败", error)
return
if not results:
self.status_var.set("未找到任何文件")
self.lbl_count.config(text="0 个结果")
return
for item in results:
name = item.get("Name", "")
path = item.get("Path", "")
size_raw = item.get("Size", "")
date_mod = item.get("DateModified", "")
date_cre = item.get("DateCreated", "")
try:
size_display = self.format_size(int(size_raw))
except (ValueError, TypeError):
size_display = str(size_raw) if size_raw else ""
iid = self.tree.insert("", tk.END, values=(name, path, size_display, date_mod, date_cre))
self.size_map[iid] = size_raw
full_path = build_full_path(path, name)
if full_path:
self.file_paths[iid] = full_path
self.status_var.set(f"搜索完成，显示 {len(results)} 条结果")
self.lbl_count.config(text=f"共 {len(results)} 条结果")
def on_double_click(self, event):
selection = self.tree.selection()
if not selection:
return
iid = selection[0]
file_path = self.file_paths.get(iid)
if not file_path:
messagebox.showwarning("路径缺失", "无法获取该文件的完整路径，可能解析异常。")
return
if not os.path.exists(file_path):
messagebox.showerror("文件不存在", f"文件未找到:\n{file_path}")
return
try:
self.status_var.set(f"正在打开: {os.path.basename(file_path)}")
self.root.update()
if os.name == 'nt':
os.startfile(file_path)
elif os.name == 'posix':
subprocess.run(['xdg-open', file_path], check=False)
else:
messagebox.showwarning("平台不支持", "当前操作系统不支持直接打开文件。")
except Exception as e:
messagebox.showerror("打开失败", f"无法打开文件:\n{file_path}\n\n错误: {e}")
finally:
self.status_var.set("就绪")
def on_right_click(self, event):
iid = self.tree.identify_row(event.y)
if not iid:
return
self.tree.selection_set(iid)
self.right_click_iid = iid
self.context_menu.tk_popup(event.x_root, event.y_root)
def copy_full_path(self):
iid = getattr(self, 'right_click_iid', None)
if not iid:
return
path = self.file_paths.get(iid, "")
if path:
self.root.clipboard_clear()
self.root.clipboard_append(path)
self.status_var.set("路径已复制到剪贴板")
else:
messagebox.showwarning("路径缺失", "无法获取文件完整路径。")
def open_file_location(self):
iid = getattr(self, 'right_click_iid', None)
if not iid:
return
path = self.file_paths.get(iid, "")
if not path:
messagebox.showwarning("路径缺失", "无法获取文件完整路径。")
return
if not os.path.exists(path):
messagebox.showerror("文件不存在", f"文件未找到:\n{path}")
return
try:
if os.name == 'nt':
subprocess.run(['explorer', '/select,', os.path.normpath(path)])
self.status_var.set(f"已打开文件位置: {path}")
else:
folder = os.path.dirname(path)
if os.name == 'posix':
subprocess.run(['xdg-open', folder], check=False)
else:
messagebox.showwarning("平台不支持", "当前操作系统不支持此操作。")
except Exception as e:
messagebox.showerror("打开位置失败", f"操作失败:\n{path}\n\n错误: {e}")
def open_with_dialog(self):
iid = getattr(self, 'right_click_iid', None)
if not iid:
return
path = self.file_paths.get(iid, "")
if not path:
messagebox.showwarning("路径缺失", "无法获取文件完整路径。")
return
if not os.path.exists(path):
messagebox.showerror("文件不存在", f"文件未找到:\n{path}")
return
if os.name != 'nt':
messagebox.showwarning("平台不支持", "“打开方式”对话框仅在 Windows 上可用。")
return
file_name = os.path.basename(path)
dir_path = os.path.dirname(path)
try:
ps_script = (
f"$shell=New-Object -ComObject Shell.Application;"
f"$folder=$shell.Namespace('{dir_path}');"
f"$file=$folder.ParseName('{file_name}');"
f"$file.InvokeVerb('openas')"
)
proc = subprocess.run(
['powershell', '-NoProfile', '-ExecutionPolicy', 'Bypass', '-Command', ps_script],
capture_output=True, text=True, timeout=10
)
if proc.returncode == 0:
self.status_var.set(f"已打开方式对话框: {file_name}")
return
else:
self.debug_text.insert(tk.END, f"\n[打开方式] PowerShell 失败: {proc.stderr}\n")
self.debug_text.see(tk.END)
except Exception as e:
self.debug_text.insert(tk.END, f"\n[打开方式] PowerShell 异常: {e}\n")
self.debug_text.see(tk.END)
try:
os.startfile(path, 'openas')
self.status_var.set(f"已打开方式对话框: {file_name}")
return
except Exception as e:
self.debug_text.insert(tk.END, f"\n[打开方式] startfile openas 失败: {e}\n")
self.debug_text.see(tk.END)
try:
os.startfile(path)
self.status_var.set(f"已用默认程序打开: {file_name}")
return
except Exception as e:
self.debug_text.insert(tk.END, f"\n[打开方式] startfile 直接打开也失败: {e}\n")
self.debug_text.see(tk.END)
try:
subprocess.run(['explorer', '/select,', os.path.normpath(path)])
messagebox.showinfo(
"提示",
"无法自动弹出“打开方式”对话框，已打开文件所在位置。\n请右键该文件手动选择“打开方式”。"
)
except Exception:
messagebox.showerror("错误", "所有方法均失败，请检查系统设置。")
def _check_file_association(self, ext):
try:
result = subprocess.run(
['cmd', '/c', 'assoc', ext],
capture_output=True,
text=True,
timeout=5
)
if result.returncode == 0 and result.stdout.strip():
return '=' in result.stdout
except Exception:
pass
try:
import winreg
with winreg.OpenKey(winreg.HKEY_CLASSES_ROOT, ext) as key:
default_value = winreg.QueryValue(key, None)
if default_value:
return True
except Exception:
pass
return False
def sort_by_column(self, col):
if self.sort_column == col:
self.sort_reverse = not self.sort_reverse
else:
self.sort_column = col
self.sort_reverse = False
items = self.tree.get_children('')
if col == 'size':
data = []
for iid in items:
raw = self.size_map.get(iid, '')
try:
key = int(raw)
except (ValueError, TypeError):
key = 0
data.append((key, iid))
else:
data = [(self.tree.set(iid, col), iid) for iid in items]
data.sort(key=lambda x: x[0], reverse=self.sort_reverse)
for index, (_, iid) in enumerate(data):
self.tree.move(iid, '', index)
self.update_column_headings()
def update_column_headings(self):
base_texts = {
"name": "文件名",
"path": "路径",
"size": "大小",
"date_modified": "修改时间",
"date_created": "创建时间"
}
for col, base in base_texts.items():
text = base
if col == self.sort_column:
text += " ▼" if self.sort_reverse else " ▲"
self.tree.heading(col, text=text)
def clear_results(self):
# 清空文件搜索结果树
for row in self.tree.get_children():
self.tree.delete(row)
self.status_var.set("就绪")
self.lbl_count.config(text="")
self.size_map.clear()
self.file_paths.clear()
self.sort_column = None
self.sort_reverse = False
self.right_click_iid = None
self.update_column_headings()
# 清空语音识别文本框
self.asr_text.config(state=tk.NORMAL)
self.asr_text.delete("1.0", tk.END)
self.asr_text.config(state=tk.DISABLED)
@staticmethod
def format_size(size_bytes):
try:
size_bytes = int(size_bytes)
except (ValueError, TypeError):
return str(size_bytes)
for unit in ['B', 'KB', 'MB', 'GB', 'TB']:
if size_bytes < 1024.0:
return f"{size_bytes:.1f} {unit}"
size_bytes /= 1024.0
return f"{size_bytes:.1f} PB"
def on_close(self):
# 停止语音识别线程
if self.asr_active:
self._stop_asr()
self.root.destroy()
if __name__ == "__main__":
root = tk.Tk()
app = SearchGUI(root)
root.mainloop()
代码结构概览：
| 模块 | 功能 |
|---|---|
load_config() | 加载/创建配置文件 |
search_files_with_everything() | 通过ctypes调用Everything64.dll执行搜索 |
process_user_query() | 调用大模型解析用户意图并生成搜索参数 |
SearchGUI | Tkinter主界面，包含语音识别、搜索、结果展示 |
_asr_worker() | 后台线程：麦克风采集+VAD静音检测 |
_send_audio_segment() | 将音频段发送至ASR服务端转录 |
6.2 程序运行截图
七、总结与展望
总结
本文基于海光DCU K100_AI国产算力平台，成功搭建了一套“语音识别→语义理解→文件搜索”全链路的智能文件检索系统。系统具备以下特点：
-
全栈国产化：从硬件（海光DCU K100_AI）到软件栈（DTK 26.04、vLLM），再到AI模型（Qwen3-ASR、Qwen3.5-9B），实现了完整的国产化技术闭环。
-
交互自然：用户只需说出需求，无需记忆复杂的Everything搜索语法，大模型自动完成语义到语法的转换。
-
响应迅速：Everything引擎实现毫秒级本地文件检索，vLLM保障模型推理的高吞吐。
-
功能完备：不仅支持搜索，还提供了打开文件、复制路径、打开位置、打开方式等丰富操作。
展望
未来可以在以下方向继续优化和扩展：
-
多模态搜索：结合Qwen3系列的多模态能力，支持“根据图片内容搜索图片文件”或“根据音频内容搜索音频文件”。
-
语义向量检索：不仅依赖文件名匹配，还可通过向量数据库对文件内容进行语义检索，实现“搜内容”而非“搜文件名”。
-
实时流式ASR：目前采用VAD分段后发送的准实时方案，未来可接入WebSocket流式转录接口，实现真正的实时语音交互。
-
跨平台支持：目前客户端仅支持Windows（依赖Everything），未来可适配macOS（mdfind）和Linux（locate/plocate）。
-
多卡分布式推理：当前ASR和大模型共用一张DCU卡（
HIP_VISIBLE_DEVICES=6），未来可将两个模型部署到不同DCU卡上，进一步提升并发能力。
转载自 CSDN-专业IT技术社区
原文链接：https://blog.csdn.net/pla88888888/article/details/163276247
