Qwen3-ASR-1.7B实战案例：法律咨询热线录音→案由分类+关键词提取流水线-智慧文博士

Qwen3-ASR-1.7B实战案例：法律咨询热线录音→案由分类+关键词提取流水线

1. 引言：当法律咨询遇上AI语音识别

想象一下，一个市级法律援助中心，每天要接听成百上千个咨询电话。接线员一边接听，一边手忙脚乱地记录，还要在挂断后花大量时间整理录音、归纳案由、提取关键信息。这不仅效率低下，还容易因为记录不全或理解偏差，导致后续服务跟不上。

现在，情况可以完全不同了。我们借助Qwen3-ASR-1.7B这个高精度的语音识别模型，可以搭建一条自动化流水线：电话录音进来，自动转成文字，然后智能分析出咨询属于哪个案由（比如劳动争议、婚姻家庭、合同纠纷），并提取出核心关键词（比如“拖欠工资”、“离婚财产分割”、“违约金”）。整个过程，从录音到结构化报告，可能只需要几分钟。

这篇文章，我就带你一步步搭建这个“法律咨询热线智能处理流水线”。我们不仅会用Qwen3-ASR-1.7B把录音变成文字，还会结合一些简单的文本处理技术，完成案由分类和关键词提取。即使你之前没怎么接触过语音识别，跟着做下来，也能掌握这套非常实用的解决方案。

2. 为什么选择Qwen3-ASR-1.7B？

在开始动手之前，我们先简单了解一下这次的核心工具。Qwen3-ASR-1.7B是阿里云通义千问团队推出的开源语音识别模型，你可以把它理解为一个“耳朵”特别灵的AI。

它有几个特点，特别适合我们这个法律咨询的场景：

听得准：1.7B的参数量属于“高精度”版本，对于法律咨询中可能出现的专业术语、地方口音，它的识别准确率比小模型更有保障。准确是法律工作的生命线。
听得懂“方言”：它支持22种中文方言。这意味着，即使咨询者说的是带口音的普通话，甚至是粤语、四川话，它也能较好地识别，大大提升了系统的普适性。
自动判断语言：我们不需要事先告诉它录音是中文还是英文，它能自己判断。这对于处理一些涉外或双语咨询很有帮助。
开箱即用：它已经封装成了Web镜像，我们部署好后，通过一个简单的网页就能上传音频、获取识别结果，非常方便。

简单来说，我们需要一个既准又稳的“耳朵”，Qwen3-ASR-1.7B正好符合要求。

3. 实战准备：部署与初识你的“AI耳朵”

我们的第一步，是把Qwen3-ASR-1.7B这个“AI耳朵”部署并运行起来。

3.1 一键部署与访问

如果你在CSDN星图镜像广场找到了Qwen3-ASR-1.7B的镜像，部署过程通常是一键完成的。部署成功后，你会得到一个访问地址，格式类似：

https://gpu-你的实例ID-7860.web.gpu.csdn.net/

用浏览器打开这个地址，你会看到一个简洁的Web界面。主要功能就两个：上传音频文件和查看识别结果。

3.2 第一次“听力测试”

为了确保一切正常，我们先做个简单测试。

准备测试音频：你可以用手机录一段自己说的话，比如：“您好，我想咨询一下，公司拖欠我三个月工资，我该怎么办？” 保存为MP3或WAV格式。
上传与识别：在Web界面点击上传，选择你的测试音频。语言选项保持“auto”（自动检测）。然后点击“开始识别”。
查看结果：稍等片刻，页面会显示识别出的语言（如“zh”代表中文）和转写文本。

如果能看到准确的中文文本，恭喜你，你的“AI耳朵”已经就位，听力正常！

4. 构建核心流水线：从录音到结构化信息

现在，“耳朵”有了，我们要给它配上“大脑”，构建完整的处理流水线。整个流程可以分为三个核心步骤，我们用下面的流程图来直观展示：

graph TD A[原始咨询录音.mp3] --> B(Qwen3-ASR-1.7B语音识别); B --> C{识别成功?}; C -- 是 --> D[得到纯文本转录稿]; C -- 否 --> E[记录错误并跳过]; D --> F(文本预处理与清洗); F --> G(案由分类模型); F --> H(关键词提取模块); G --> I[输出案由类别<br>如: 劳动争议]; H --> J[输出关键词列表<br>如: 拖欠工资, 仲裁]; I --> K[生成结构化报告]; J --> K; K --> L{是否批量处理?}; L -- 是 --> M[汇总成批次分析报表]; L -- 否 --> N[输出单条咨询分析结果];

接下来，我们详细拆解每一个步骤的具体实现。

4.1 第一步：语音转文本（ASR）

这是流水线的起点。我们需要编写一个Python函数，能够自动调用我们部署好的Qwen3-ASR-1.7B服务，把录音文件变成文字。

由于Web界面是交互式的，我们需要通过编程方式（模拟HTTP请求）来调用它。假设我们的服务提供了API接口（通常镜像会提供），调用方式如下：

import requests import json import time def transcribe_audio(audio_file_path, asr_service_url): """ 调用Qwen3-ASR服务进行语音识别 :param audio_file_path: 音频文件路径 :param asr_service_url: ASR服务的API地址，例如 'http://localhost:7860/api/transcribe' :return: 识别后的文本字符串 """ try: # 1. 打开音频文件 with open(audio_file_path, 'rb') as f: files = {'file': f} # 2. 构建请求数据，语言设为自动检测 data = {'language': 'auto'} # 3. 发送POST请求到ASR服务 response = requests.post(asr_service_url, files=files, data=data) response.raise_for_status() # 检查请求是否成功 # 4. 解析返回的JSON结果 result = response.json() # 假设返回格式为 {'text': '识别出的文本', 'language': 'zh'} transcribed_text = result.get('text', '') return transcribed_text except FileNotFoundError: print(f"错误：音频文件未找到 - {audio_file_path}") return "" except requests.exceptions.RequestException as e: print(f"错误：调用ASR服务失败 - {e}") return "" except json.JSONDecodeError: print("错误：无法解析ASR服务的响应") return "" # 使用示例 if __name__ == "__main__": # 替换为你的音频文件路径和服务地址 audio_path = "法律咨询录音_示例.mp3" service_url = "https://gpu-xxxx-7860.web.gpu.csdn.net/api/transcribe" text = transcribe_audio(audio_path, service_url) if text: print("识别结果：") print(text) else: print("语音识别失败。")

代码说明：

这个函数的核心是使用requests库向ASR服务发送一个包含音频文件的POST请求。
我们设置了language='auto'，让模型自动检测语言，这对法律咨询场景很实用。
函数包含了基本的错误处理，比如文件找不到、网络请求失败等，确保流水线的健壮性。

4.2 第二步：文本预处理与清洗

从ASR模型得到的文本可能包含一些我们需要清理的内容，比如多余的空白符、标点符号，或者一些识别错误产生的无意义词。清洗后的文本能让后续分析更准确。

import re import jieba from collections import Counter def clean_and_preprocess(text): """ 清洗和预处理识别出的文本 :param text: 原始识别文本 :return: 清洗后的文本， 以及分词列表 """ if not text: return "", [] # 1. 基础清洗：去除首尾空白，将多个空格/换行符合并为一个空格 cleaned_text = re.sub(r'\s+', ' ', text.strip()) # 2. 针对法律咨询场景的简单纠错（示例：常见口语化表达修正） # 例如，有人可能说“欠我工钱”，ASR可能识别为“欠我工钱”，我们可映射为更正式的“拖欠工资” correction_map = { r'欠我工钱': '拖欠工资', r'离昏': '离婚', r'打官司': '诉讼', # 可以根据实际识别错误库扩展 } for wrong, correct in correction_map.items(): cleaned_text = re.sub(wrong, correct, cleaned_text) # 3. 使用jieba进行中文分词 # 对于法律文本，可以考虑加载自定义词典以提高分词准确性 # jieba.load_userdict("legal_terms_dict.txt") words = jieba.lcut(cleaned_text) # 4. 去除停用词（如“的”、“了”、“呢”等对分类无意义的词） # 这里使用一个简单的内置列表，实际应用中应使用更全面的停用词表 stop_words = set(['的', '了', '呢', '啊', '吗', '是', '在', '我', '有', '和', '就']) filtered_words = [word for word in words if word not in stop_words and len(word) > 1] return cleaned_text, filtered_words # 使用示例 if __name__ == "__main__": # 模拟一段ASR识别结果，可能包含一些噪音 raw_text = "嗯 您好 啊 我想咨询一下 就是公司 拖欠了我 三个月的 工资 我该怎么办呢" cleaned_text, keywords = clean_and_preprocess(raw_text) print("清洗后文本:", cleaned_text) print("分词及过滤后关键词:", keywords)

代码说明：

clean_and_preprocess函数做了四件事：基础清洗、简单纠错、中文分词、去除停用词。
纠错映射(correction_map) 是一个很实用的技巧。你可以根据ASR模型在特定领域（如法律）的常见识别错误，不断补充这个映射表，显著提升后续处理的准确性。
分词使用了jieba库，它是处理中文文本的利器。如果有很多法律专业术语，建议加载自定义词典。
最终，我们得到了干净的文本和一份有意义的词语列表，为下一步分析做好准备。

4.3 第三步：案由分类与关键词提取

这是流水线的“大脑”部分。我们将使用规则匹配与简单统计相结合的方法来实现。对于生产环境，你可以考虑训练专门的分类模型，但规则方法对于初期快速搭建和验证非常有效。

def analyze_legal_consultation(cleaned_text, word_list): """ 分析法律咨询文本，进行案由分类和关键词提取 :param cleaned_text: 清洗后的完整文本 :param word_list: 预处理后的词语列表 :return: 案由类别， 关键词列表 """ # 1. 定义案由分类规则（关键词匹配） case_category_rules = { '劳动争议': ['工资', '拖欠工资', '加班费', '劳动合同', '解雇', '开除', '工伤', '仲裁'], '婚姻家庭': ['离婚', '财产分割', '抚养权', '子女抚养', '夫妻', '婚前财产', '家暴'], '合同纠纷': ['合同', '违约', '违约金', '协议', '履行', '解除合同', '买卖合同', '租赁合同'], '侵权纠纷': ['受伤', '赔偿', '人身损害', '交通事故', '医疗事故', '名誉权'], '债务纠纷': ['借款', '欠款', '债务', '债权人', '债务人', '利息'], '其他': [] # 默认类别 } # 2. 案由分类：计算每个类别的匹配得分 category_scores = {} for category, keywords in case_category_rules.items(): score = 0 for keyword in keywords: # 在清洗后的完整文本中查找关键词（避免分词不准确的影响） if keyword in cleaned_text: score += 2 # 完整关键词匹配权重更高 # 同时在分词列表中查找 if keyword in word_list: score += 1 if score > 0: category_scores[category] = score # 3. 确定案由类别 if category_scores: # 选择得分最高的类别 predicted_category = max(category_scores, key=category_scores.get) else: predicted_category = '其他' # 4. 关键词提取：从分词列表中提取高频且重要的词 # 计算词频 word_freq = Counter(word_list) # 过滤掉过于常见或无意义的词（可根据需要扩展） common_but_meaningless = ['咨询', '一下', '怎么办', '如何', '问题', '法律'] meaningful_words = [(word, freq) for word, freq in word_freq.items() if word not in common_but_meaningless] # 按词频排序，取前5个作为关键词 meaningful_words.sort(key=lambda x: x[1], reverse=True) top_keywords = [word for word, freq in meaningful_words[:5]] return predicted_category, top_keywords # 使用示例：串联前三步 if __name__ == "__main__": # 假设这是从ASR服务获取的原始文本 asr_raw_text = "律师您好，公司单方面解除劳动合同，还没有支付经济补偿金，我能不能申请劳动仲裁？" # 第二步：文本预处理 cleaned_text, word_list = clean_and_preprocess(asr_raw_text) print(f"预处理后文本: {cleaned_text}") print(f"分词列表: {word_list}") # 第三步：案由分类与关键词提取 category, keywords = analyze_legal_consultation(cleaned_text, word_list) print("\n=== 分析结果 ===") print(f"案由分类: {category}") print(f"提取关键词: {keywords}")

代码说明：

analyze_legal_consultation函数是流水线的核心分析模块。
案由分类：我们采用基于规则的关键词匹配。为每个案由定义一组关键词，计算咨询文本与各组关键词的匹配度，得分最高的即为预测案由。这种方法直观、可控，非常适合业务规则明确的场景。
关键词提取：我们使用词频统计（Counter）来找出文本中最重要的词语。通过过滤掉“咨询”、“怎么办”等常见但无分析价值的词，我们得到了能反映咨询核心内容的关键词列表。
这个示例展示了从一段模拟咨询文本到最终分析结果的完整过程。你可以看到，系统成功将其分类为“劳动争议”，并提取出了“劳动合同”、“解除”、“经济补偿金”、“仲裁”等关键信息。

5. 组装完整流水线与批量处理

现在，我们把前三个步骤串联起来，形成一个完整的、可以处理单个录音文件的函数。然后，我们再扩展它，使其能够处理一个文件夹里的所有录音，实现批量处理。

import os import pandas as pd from datetime import datetime def process_single_consultation(audio_path, asr_service_url): """处理单条法律咨询录音的完整流水线""" print(f"正在处理: {os.path.basename(audio_path)}") # 步骤1: 语音转文本 raw_text = transcribe_audio(audio_path, asr_service_url) if not raw_text: print(f" -> 语音识别失败，跳过。") return None # 步骤2: 文本预处理 cleaned_text, word_list = clean_and_preprocess(raw_text) # 步骤3: 案由分类与关键词提取 category, keywords = analyze_legal_consultation(cleaned_text, word_list) # 组装结果 result = { '音频文件': os.path.basename(audio_path), '识别文本': cleaned_text, '案由分类': category, '关键词': ', '.join(keywords), '处理时间': datetime.now().strftime('%Y-%m-%d %H:%M:%S') } print(f" -> 完成。案由: {category}, 关键词: {keywords}") return result def batch_process_consultations(audio_folder, asr_service_url, output_csv='consultation_results.csv'): """批量处理一个文件夹内的所有音频文件""" # 支持常见的音频格式 supported_formats = ('.mp3', '.wav', '.flac', '.m4a', '.ogg') all_results = [] # 遍历文件夹 for filename in os.listdir(audio_folder): if filename.lower().endswith(supported_formats): audio_path = os.path.join(audio_folder, filename) result = process_single_consultation(audio_path, asr_service_url) if result: all_results.append(result) # 将结果保存到CSV文件 if all_results: df = pd.DataFrame(all_results) df.to_csv(output_csv, index=False, encoding='utf-8-sig') # 使用utf-8-sig支持Excel中文 print(f"\n批量处理完成！共处理 {len(all_results)} 个文件。") print(f"结果已保存至: {output_csv}") # 打印简要统计 print("\n=== 案由分类统计 ===") print(df['案由分类'].value_counts()) return df else: print("未找到支持的音频文件或处理全部失败。") return None # 使用示例 if __name__ == "__main__": # 配置你的ASR服务地址和音频文件夹路径 ASR_SERVICE_URL = "https://gpu-xxxx-7860.web.gpu.csdn.net/api/transcribe" # 请替换为实际地址 AUDIO_FOLDER = "./law_consultation_audios" # 存放录音的文件夹 # 运行批量处理 results_df = batch_process_consultations(AUDIO_FOLDER, ASR_SERVICE_URL)

代码说明：

process_single_consultation函数将前三步完美串联，输入一条录音，输出一个包含所有分析结果的字典。
batch_process_consultations函数实现了批量自动化。它会扫描指定文件夹下的所有音频文件，逐一处理，并将最终结果汇总到一个CSV表格中。这份表格可以直接用Excel打开，方便法务人员查看和归档。
我们还添加了简单的统计功能，可以快速查看各类案由的分布情况，这对于法律援助中心进行工作分析和资源调配非常有价值。