Qwen3-ASR-1.7B实战指南：对接LangChain构建语音输入的RAG问答系统

1. 项目背景与价值

想象一下这样的场景：你正在开车，突然想到一个专业问题需要查询，但双手不能离开方向盘。或者你在实验室做实验，手上沾满试剂，却需要立即查阅技术文档。传统的打字搜索方式在这些场景下完全不可行。

这就是语音输入RAG系统的价值所在。通过将Qwen3-ASR-1.7B语音识别模型与LangChain框架结合，我们可以构建一个能够"听懂"问题并"智能回答"的系统。你只需要说出问题，系统就能自动识别语音内容，从知识库中检索相关信息，并生成准确的回答。

这种技术组合特别适合：

• 移动场景下的信息查询（驾驶、步行、运动时）
• 双手被占用的专业工作环境（实验室、工厂、手术室）
• 为视觉障碍人群提供无障碍的信息访问方式
• 多语言环境下的即时翻译和问答需求

2. 环境准备与依赖安装

在开始构建系统之前，我们需要准备相应的环境。以下是完整的依赖清单和安装步骤。

2.1 核心依赖包

首先创建并激活Python虚拟环境：

python -m venv asr-rag-env
source asr-rag-env/bin/activate  # Linux/Mac
# 或者
asr-rag-env\Scripts\activate    # Windows

安装必要的Python包：

pip install torch torchaudio transformers
pip install langchain langchain-community chromadb
pip install pydub librosa soundfile
pip install fastapi uvicorn python-multipart

2.2 音频处理工具安装

对于音频文件处理，还需要安装ffmpeg：

# Ubuntu/Debian
sudo apt-get install ffmpeg

# CentOS/RHEL
sudo yum install ffmpeg

# MacOS
brew install ffmpeg

# Windows
# 从 https://ffmpeg.org/download.html 下载并添加到系统PATH

3. 系统架构设计

我们的语音RAG系统采用模块化设计，主要包括四个核心组件：

语音输入 → 语音识别 → 文本处理 → 知识检索 → 智能回答

3.1 整体架构图

class VoiceRAGSystem:
    def __init__(self):
        self.asr_model = None      # 语音识别模块
        self.vector_db = None      # 向量数据库
        self.retriever = None      # 检索器
        self.llm_chain = None      # 语言模型链
        
    def process_audio(self, audio_file):
        # 1. 语音转文本
        text = self.transcribe_audio(audio_file)
        
        # 2. 检索相关知识
        relevant_docs = self.retrieve_documents(text)
        
        # 3. 生成回答
        answer = self.generate_answer(text, relevant_docs)
        
        return answer

4. Qwen3-ASR-1.7B语音识别集成

4.1 模型加载与初始化

首先实现语音识别模块，用于将音频转换为文本：

import torch
from transformers import AutoModelForSpeechSeq2Seq, AutoProcessor
import torchaudio

class QwenASRTranscriber:
    def __init__(self, model_name="Qwen/Qwen3-ASR-1.7B"):
        self.device = "cuda" if torch.cuda.is_available() else "cpu"
        self.torch_dtype = torch.float16 if self.device == "cuda" else torch.float32
        
        # 加载模型和处理器
        self.processor = AutoProcessor.from_pretrained(model_name)
        self.model = AutoModelForSpeechSeq2Seq.from_pretrained(
            model_name,
            torch_dtype=self.torch_dtype,
            low_cpu_mem_usage=True,
            use_safetensors=True
        ).to(self.device)
        
    def transcribe_audio(self, audio_path):
        # 加载和预处理音频
        waveform, sample_rate = torchaudio.load(audio_path)
        
        # 重采样到16kHz（模型要求的采样率）
        if sample_rate != 16000:
            resampler = torchaudio.transforms.Resample(sample_rate, 16000)
            waveform = resampler(waveform)
        
        # 处理音频并生成转录
        inputs = self.processor(
            waveform.squeeze().numpy(),
            sampling_rate=16000,
            return_tensors="pt",
            padding=True
        )
        
        with torch.no_grad():
            generated_ids = self.model.generate(
                inputs.input_values.to(self.device),
                attention_mask=inputs.attention_mask.to(self.device)
            )
        
        transcription = self.processor.batch_decode(
            generated_ids, skip_special_tokens=True
        )[0]
        
        return transcription

4.2 音频预处理工具

为了处理各种格式的音频文件，我们创建一个音频预处理工具类：

import os
from pydub import AudioSegment

class AudioPreprocessor:
    @staticmethod
    def convert_to_wav(input_path, output_path=None):
        """将各种音频格式转换为WAV格式"""
        if output_path is None:
            output_path = os.path.splitext(input_path)[0] + ".wav"
        
        # 支持多种音频格式
        audio = AudioSegment.from_file(input_path)
        audio = audio.set_frame_rate(16000).set_channels(1)
        audio.export(output_path, format="wav")
        return output_path
    
    @staticmethod
    def normalize_audio(input_path, output_path):
        """标准化音频音量"""
        audio = AudioSegment.from_file(input_path)
        normalized_audio = audio.normalize()
        normalized_audio.export(output_path, format="wav")
        return output_path

5. LangChain RAG系统构建

5.1 知识库准备与向量化

首先我们需要准备知识库并将其向量化存储：

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain.schema import Document

class KnowledgeBaseManager:
    def __init__(self, persist_directory="./chroma_db"):
        self.embeddings = HuggingFaceEmbeddings(
            model_name="sentence-transformers/all-MiniLM-L6-v2"
        )
        self.text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=1000,
            chunk_overlap=200
        )
        self.persist_directory = persist_directory
        self.vectorstore = None
    
    def create_knowledge_base(self, documents):
        """从文档创建知识库"""
        # 分割文档
        docs = []
        for doc_content in documents:
            splits = self.text_splitter.split_text(doc_content)
            docs.extend([Document(page_content=text) for text in splits])
        
        # 创建向量存储
        self.vectorstore = Chroma.from_documents(
            documents=docs,
            embedding=self.embeddings,
            persist_directory=self.persist_directory
        )
        self.vectorstore.persist()
        return self.vectorstore
    
    def load_knowledge_base(self):
        """加载已有的知识库"""
        self.vectorstore = Chroma(
            persist_directory=self.persist_directory,
            embedding_function=self.embeddings
        )
        return self.vectorstore

5.2 检索增强生成链

构建完整的RAG问答链：

from langchain.chains import RetrievalQA
from langchain_community.llms import Ollama  # 或者使用OpenAI、Azure等

class RAGQASystem:
    def __init__(self, vectorstore, model_name="llama2"):
        self.vectorstore = vectorstore
        self.llm = Ollama(model=model_name)
        self.qa_chain = self._create_qa_chain()
    
    def _create_qa_chain(self):
        """创建QA链"""
        retriever = self.vectorstore.as_retriever(
            search_type="similarity",
            search_kwargs={"k": 4}
        )
        
        qa_chain = RetrievalQA.from_chain_type(
            llm=self.llm,
            chain_type="stuff",
            retriever=retriever,
            return_source_documents=True
        )
        return qa_chain
    
    def ask_question(self, question):
        """提问并获取回答"""
        result = self.qa_chain({"query": question})
        return {
            "answer": result["result"],
            "source_documents": result["source_documents"]
        }

6. 完整系统集成

现在我们将所有组件集成到一个完整的系统中：

import logging
from typing import Dict, Any

logger = logging.getLogger(__name__)

class VoiceRAGSystem:
    def __init__(self, knowledge_base_path="./knowledge_base"):
        # 初始化各组件
        self.transcriber = QwenASRTranscriber()
        self.audio_preprocessor = AudioPreprocessor()
        self.kb_manager = KnowledgeBaseManager()
        
        # 加载知识库
        try:
            self.vectorstore = self.kb_manager.load_knowledge_base()
        except:
            logger.warning("知识库不存在，请先创建知识库")
            self.vectorstore = None
        
        self.rag_system = None
        if self.vectorstore:
            self.rag_system = RAGQASystem(self.vectorstore)
    
    def process_voice_query(self, audio_file_path: str) -> Dict[str, Any]:
        """处理语音查询的完整流程"""
        try:
            # 1. 预处理音频
            processed_audio = self.audio_preprocessor.convert_to_wav(audio_file_path)
            
            # 2. 语音转文本
            logger.info("开始语音识别...")
            question_text = self.transcriber.transcribe_audio(processed_audio)
            logger.info(f"识别结果: {question_text}")
            
            # 3. RAG问答
            if self.rag_system:
                result = self.rag_system.ask_question(question_text)
                return {
                    "success": True,
                    "question": question_text,
                    "answer": result["answer"],
                    "sources": [doc.page_content for doc in result["source_documents"]]
                }
            else:
                return {
                    "success": False,
                    "error": "RAG系统未初始化",
                    "question": question_text
                }
                
        except Exception as e:
            logger.error(f"处理语音查询时出错: {str(e)}")
            return {
                "success": False,
                "error": str(e)
            }
    
    def add_to_knowledge_base(self, documents):
        """向知识库添加新文档"""
        self.vectorstore = self.kb_manager.create_knowledge_base(documents)
        self.rag_system = RAGQASystem(self.vectorstore)
        return True

7. Web服务接口

为了提供易用的服务，我们创建一个FastAPI Web接口：

from fastapi import FastAPI, File, UploadFile, HTTPException
from fastapi.responses import JSONResponse
import tempfile
import os

app = FastAPI(title="语音RAG问答系统")
voice_rag_system = None

@app.on_event("startup")
async def startup_event():
    """启动时初始化系统"""
    global voice_rag_system
    try:
        voice_rag_system = VoiceRAGSystem()
        print("语音RAG系统初始化完成")
    except Exception as e:
        print(f"系统初始化失败: {e}")

@app.post("/api/voice-query")
async def process_voice_query(file: UploadFile = File(...)):
    """处理语音查询接口"""
    if not voice_rag_system:
        raise HTTPException(status_code=503, detail="系统未初始化")
    
    # 保存上传的音频文件
    with tempfile.NamedTemporaryFile(delete=False, suffix=".wav") as temp_file:
        content = await file.read()
        temp_file.write(content)
        temp_path = temp_file.name
    
    try:
        # 处理语音查询
        result = voice_rag_system.process_voice_query(temp_path)
        return JSONResponse(content=result)
    finally:
        # 清理临时文件
        os.unlink(temp_path)

@app.post("/api/add-knowledge")
async def add_knowledge(document: str):
    """向知识库添加新知识"""
    if not voice_rag_system:
        raise HTTPException(status_code=503, detail="系统未初始化")
    
    try:
        success = voice_rag_system.add_to_knowledge_base([document])
        return {"success": success, "message": "知识添加成功"}
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"添加知识失败: {str(e)}")

@app.get("/health")
async def health_check():
    """健康检查接口"""
    return {"status": "healthy", "system_initialized": voice_rag_system is not None}

8. 实战演示与测试

8.1 系统启动与测试

启动Web服务：

uvicorn main:app --host 0.0.0.0 --port 8000 --reload

使用curl测试语音查询：

# 上传音频文件进行查询
curl -X POST "http://localhost:8000/api/voice-query" \
  -H "accept: application/json" \
  -H "Content-Type: multipart/form-data" \
  -F "file=@question_audio.wav"

8.2 示例应用场景

场景一：技术文档查询

# 准备技术文档知识库
documents = [
    "Qwen3-ASR-1.7B支持52种语言识别，包括30种通用语言和22种中文方言...",
    "LangChain是一个用于开发LLM应用的框架，提供链、代理、记忆等组件...",
    "RAG（检索增强生成）技术通过检索相关知识来增强LLM的回答准确性..."
]

# 语音提问："LangChain的主要组件有哪些？"
# 系统会自动识别并返回准确答案

场景二：多语言支持

# 用英语提问："What languages does Qwen3-ASR support?"
# 用粤语提问："Qwen3-ASR识唔识听广东话？"
# 系统都能正确识别并回答

9. 性能优化与部署建议

9.1 性能优化技巧

# 使用批处理提高识别效率
def batch_transcribe(audio_paths):
    """批量语音识别"""
    results = []
    for path in audio_paths:
        try:
            text = self.transcriber.transcribe_audio(path)
            results.append({"file": path, "text": text, "success": True})
        except Exception as e:
            results.append({"file": path, "error": str(e), "success": False})
    return results

# 实现缓存机制减少重复处理
from functools import lru_cache

class CachedTranscriber(QwenASRTranscriber):
    @lru_cache(maxsize=100)
    def transcribe_audio_cached(self, audio_path):
        """带缓存的语音识别"""
        return self.transcribe_audio(audio_path)

9.2 部署建议

硬件要求：

• GPU：至少8GB显存（用于ASR模型和嵌入模型）
• RAM：16GB以上
• 存储：根据知识库大小调整，建议50GB以上SSD

生产环境部署：

# 使用gunicorn多进程部署
gunicorn -w 4 -k uvicorn.workers.UvicornWorker main:app

# 使用docker容器化部署
docker build -t voice-rag-system .
docker run -p 8000:8000 --gpus all voice-rag-system

10. 总结与展望

通过本教程，我们成功构建了一个基于Qwen3-ASR-1.7B和LangChain的语音输入RAG问答系统。这个系统不仅能够准确识别多语言语音输入，还能从知识库中检索相关信息并生成智能回答。

关键收获：

1. 语音识别集成：学会了如何集成和调用Qwen3-ASR-1.7B进行高质量的语音转文本
2. RAG系统构建：掌握了使用LangChain构建检索增强生成系统的完整流程
3. 系统集成：了解了如何将不同组件集成为一个完整的应用系统
4. 实践部署：获得了从开发到部署的完整项目经验

未来扩展方向：

• 支持实时语音流识别，实现真正的对话式交互
• 集成文本转语音(TTS)功能，提供语音回答输出
• 增加多模态支持，处理包含图像的复杂查询
• 实现用户个性化知识库和查询历史记忆

这个项目展示了现代AI技术的强大组合能力，通过将专业的语音识别模型与灵活的LangChain框架结合，我们可以构建出真正实用的人工智能应用。

---

获取更多AI镜像

想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。