Python音声認識アプリ開発実践：多言語対応ASRで業務効率化を実現する実装パターン

python -m venv asr_env
source asr_env/bin/activate  # Windows: asr_env\Scripts\activate

pip install openai-whisper
pip install torch torchvision torchaudio
pip install soundfile librosa

import whisper

# モデルのロード（初回はダウンロードが発生）
model = whisper.load_model("base")

# 音声ファイルの文字起こし
result = model.transcribe("audio.mp3", language="ja")
print(result["text"])

# 言語自動検出の場合
result_auto = model.transcribe("audio.mp3")
print(f"検出言語: {result_auto['language']}")
print(result_auto["text"])

import speech_recognition as sr

recognizer = sr.Recognizer()

with sr.AudioFile("audio.wav") as source:
    audio = recognizer.record(source)
    
try:
    # Google Speech API を使用
    text = recognizer.recognize_google(audio, language="ja-JP")
    print(text)
except sr.UnknownValueError:
    print("音声を認識できませんでした")
except sr.RequestError as e:
    print(f"APIエラー: {e}")

import librosa
import numpy as np
import soundfile as sf

def preprocess_audio(input_path, output_path, target_sr=16000):
    # 音声ファイルの読み込み
    audio, sr = librosa.load(input_path, sr=None)
    
    # サンプリングレートの変換
    if sr != target_sr:
        audio = librosa.resample(audio, orig_sr=sr, target_sr=target_sr)
    
    # 無音区間のトリミング
    audio_trimmed, _ = librosa.effects.trim(audio, top_db=20)
    
    # ノイズゲートの適用
    audio_filtered = librosa.effects.preemphasis(audio_trimmed)
    
    # 正規化
    audio_normalized = librosa.util.normalize(audio_filtered)
    
    # 保存
    sf.write(output_path, audio_normalized, target_sr)
    
    return output_path

import webrtcvad
import wave
import struct

def detect_speech_segments(audio_path, aggressiveness=2):
    vad = webrtcvad.Vad(aggressiveness)  # 0-3: 高いほど厳格
    
    with wave.open(audio_path, 'rb') as wf:
        sample_rate = wf.getframerate()
        frames = wf.readframes(wf.getnframes())
    
    # 30ms ごとにフレームを分割
    frame_duration = 30  # ms
    frame_size = int(sample_rate * frame_duration / 1000) * 2
    
    speech_segments = []
    for i in range(0, len(frames), frame_size):
        frame = frames[i:i+frame_size]
        if len(frame) < frame_size:
            break
            
        is_speech = vad.is_speech(frame, sample_rate)
        if is_speech:
            speech_segments.append((i, i+frame_size))
    
    return speech_segments

import pyaudio
import numpy as np
import whisper
from queue import Queue
import threading

class RealtimeASR:
    def __init__(self, model_size="base"):
        self.model = whisper.load_model(model_size)
        self.audio_queue = Queue()
        self.is_running = False
        
        # PyAudio の設定
        self.CHUNK = 1024
        self.FORMAT = pyaudio.paInt16
        self.CHANNELS = 1
        self.RATE = 16000
        
    def audio_callback(self, in_data, frame_count, time_info, status):
        self.audio_queue.put(in_data)
        return (in_data, pyaudio.paContinue)
    
    def start_stream(self):
        self.is_running = True
        p = pyaudio.PyAudio()
        
        stream = p.open(
            format=self.FORMAT,
            channels=self.CHANNELS,
            rate=self.RATE,
            input=True,
            frames_per_buffer=self.CHUNK,
            stream_callback=self.audio_callback
        )
        
        stream.start_stream()
        
        # 認識スレッドを起動
        recognition_thread = threading.Thread(target=self.process_audio)
        recognition_thread.start()
        
        return stream, p
    
    def process_audio(self):
        buffer = []
        while self.is_running:
            if not self.audio_queue.empty():
                data = self.audio_queue.get()
                buffer.append(np.frombuffer(data, dtype=np.int16))
                
                # 3秒分のバッファが溜まったら認識
                if len(buffer) >= self.RATE * 3 / self.CHUNK:
                    audio_data = np.concatenate(buffer)
                    audio_float = audio_data.astype(np.float32) / 32768.0
                    
                    result = self.model.transcribe(audio_float, language="ja")
                    print(f"認識結果: {result['text']}")
                    
                    buffer = []

from fastapi import FastAPI, WebSocket
from fastapi.responses import HTMLResponse
import asyncio

app = FastAPI()

@app.websocket("/ws/transcribe")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    
    asr = RealtimeASR()
    stream, p = asr.start_stream()
    
    try:
        while True:
            # 認識結果を WebSocket で送信
            if not asr.audio_queue.empty():
                result = await process_recognition(asr)
                await websocket.send_json({"text": result})
            
            await asyncio.sleep(0.1)
            
    except Exception as e:
        print(f"エラー: {e}")
    finally:
        asr.is_running = False
        stream.stop_stream()
        stream.close()
        p.terminate()

class MeetingTranscriptionPipeline:
    def __init__(self):
        self.whisper_model = whisper.load_model("large")
        self.summarizer = pipeline("summarization", model="facebook/bart-large-cnn")
        
    def process_meeting(self, audio_path):
        # Step 1: 前処理
        processed_audio = self.preprocess(audio_path)
        
        # Step 2: 文字起こし
        transcription = self.transcribe(processed_audio)
        
        # Step 3: 話者分離
        speaker_segments = self.diarize(transcription)
        
        # Step 4: 要約生成
        summary = self.generate_summary(transcription["text"])
        
        return {
            "transcription": transcription,
            "speakers": speaker_segments,
            "summary": summary
        }
    
    def transcribe(self, audio_path):
        result = self.whisper_model.transcribe(
            audio_path,
            language="ja",
            task="transcribe",
            verbose=False
        )
        return result
    
    def generate_summary(self, text, max_length=150):
        # 長文を分割して要約
        chunks = self.split_text(text, max_chunk_size=1024)
        summaries = []
        
        for chunk in chunks:
            summary = self.summarizer(chunk, max_length=max_length, min_length=30)
            summaries.append(summary[0]["summary_text"])
        
        return " ".join(summaries)

from pyannote.audio import Pipeline

def diarize_speakers(audio_path, transcription):
    # 話者分離パイプラインの初期化
    pipeline = Pipeline.from_pretrained("pyannote/speaker-diarization")
    
    # 話者分離の実行
    diarization = pipeline(audio_path)
    
    # タイムスタンプと話者を対応付け
    speaker_segments = []
    for segment, _, speaker in diarization.itertracks(yield_label=True):
        speaker_segments.append({
            "start": segment.start,
            "end": segment.end,
            "speaker": speaker,
            "text": get_text_for_segment(transcription, segment.start, segment.end)
        })
    
    return speaker_segments