2025. 1. 2. 22:36ㆍProject Log/학부 졸업프로젝트
Whisper STT 구현 내용
Whisper STT 모듈을 구현했다. 지난번에 플러터 패키지 STT도 사용해 보았는데, 조금 더 정확한 모델을 사용하고 싶어서 OpenAI Whisper를 사용하게 되었다.
'클라이언트 Flutter 앱에서 음성 녹음 ➝ 웹소켓을 통해 실시간으로 서버에 전달 ➝ django 앱 Whisper 모델에서 음성을 텍스트로 변환 ➝ 웹소켓을 통해 서버에서 클라이언트로 변환된 텍스트를 전달'하는 방식이다.
만들면서 정말 에러가 많이 나서 마음속으로 아래 곰처럼 '와아아악!!!!!!!!'을 몇 번씩 외친 것 같다. 에러에 초연해야 하는데, 아직 갈 길이 멀다. 그래도 더 많은 고난을 겪지 않고 오늘 목표를 달성해서 행복하다.
Django 기본 디렉토리 구조
먼저 django 앱을 생성하는 커맨드와 기본 앱 디렉토리 구조를 간단히 정리했다. 지난번에 다 같이 만들었던 도커 파일은 수정이 좀 필요해서, 일단은 로컬에서 진행했다.
먼저 django 기본 프로젝트를 생성한다.
django-admin startproject voice_django
지금은 각종 파일을 더 추가한 상태라 디렉토리 트리가 더 복잡하긴 하다. 붉은색 박스로 표시한 파일들은 위의 커맨드로 기본 생성되는 것이다. 각 파일의 기능을 간단히 정리하겠다.
- manage.py: Django 프로젝트와 다양한 방법으로 상호작용 하는 커맨드라인의 유틸리티
- voice_django: 프로젝트의 파이썬 패키지가 포함되는 디렉토리
- voice_django/__init__.py: 파이썬으로 하여금 이 디렉토리를 패키지처럼 다루라고 알려주는 용도의 단순한 빈 파일
- voice_django/asgi.py: 현재 프로젝트를 서비스하기 위한 ASGI-호환 웹 서버의 진입점
- voice_django/settings.py: 프로젝트의 환경 및 구성을 저장
- voice_django/urls.py: URL 선언을 저장
- voice_django/wsgi.py: 현재 프로젝트를 서비스하기 위한 WSGI 호환 웹 서버의 진입점
사실 위의 설명도 중요하지만, 직접 구현하면서 각 파일의 용도가 더 와 닿았던 것 같다. 오늘 각 파일에 대해 기억에 남는 점을 기록하려고 한다.
1) __init__.py
먼저 __init__.py 는 아래 커맨드를 통해 생성할 수 있다. 이걸 생성 안 해서 에러를 마주쳤기에 매우 중요하다.
touch __init__.py
상황은 다음과 같다. 프로젝트 폴더에 my_whisper라는 디렉토리를 직접 만들고, 그 아래에 consumers.py와 routing.py 파일을 생성한 상황이었다. asgi.py에 분명히 'from my_whisper.routing import websocket_urlpatterns'를 적었는데 이 경로를 인식하지 못하고 에러가 나왔다. 그 이유는 바로 my_whisper 디렉토리 안에 __init__.py 를 안 만들어서 그렇다. 위의 커맨드로 파일을 생성하니 에러가 해결되었다.
2) manage.py
서버를 작동시키는 기본 코드는 manage.py를 사용하는 것이다.
python manage.py runserver
3) WSGI와 Gunicorn 서버 vs ASGI와 Daphne 서버
그런데 이 외에 프로덕션 환경에서 서버를 동작시키는 다른 방법도 존재한다. WSGI는 Gunicorn 서버로 실행할 수 있고, ASGI는 Daphne 서버로 실행할 수 있다. WSGI는 동기식 통신을, ASGI는 비동기식 통신을 지원한다.
Django는 전통적으로 WSGI를 따르는 웹 애플리케이션이라고 한다. ASGI는 HTTP 요청뿐만 아니라 웹소켓 같은 양방향 비동기 프로토콜을 지원한다. Django는 기본적으로 ASGI를 지원하지 않기 때문에, Channels 라이브러리를 사용해서 Django 애플리케이션에 ASGI 기능을 추가하는 형태라고 한다.
오늘 나도 ASGI를 주로 다루었기 때문에, Daphne 서버를 실행한 채로 웹소켓을 테스트했다.
daphne voice_django.asgi:application
구글링하다 보니 supervisord를 사용하여 두 종류의 서버를 병렬로 실행하고 Nginx로 라우팅하는 방식도 있었는데, 이는 다음에 알아보기로 했다.
서버 Django 앱 구현 방식
다운로드한 라이브러리
[Command]
pip install channels
pip install librosa
pip install --upgrade --no-deps --force-reinstall git+https://github.com/openai/whisper.git
[Version]
channels==4.2.0
librosa==0.10.2.post1
openai-whisper==20240930
voice_django/asgi.py
- AuthMiddlewareStack은 웹소켓 요청에 대해 사용자 인증을 처리한다.
- my_whisper/routing.py의 websocket_urlpatterns를 임포트하여, URLRouter로 라우팅한다.
import os
from django.core.asgi import get_asgi_application
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.auth import AuthMiddlewareStack
from my_whisper.routing import websocket_urlpatterns
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'voice_django.settings')
application = ProtocolTypeRouter({
"http": get_asgi_application(),
"websocket": AuthMiddlewareStack(
URLRouter(websocket_urlpatterns)
),
})
voice_django/settings.py
- 10.0.2.2는 안드로이드 에뮬레이터가 호스트 컴퓨터와 통신할 때 사용하는 IP이다.
- INSTALLED_APPS는 Django 프로젝트에 활성화된 애플리케이션 리스트이다.
- ASGI_APPLICATION은 "{django 프로젝트명}.asgi.application"으로 지정하면 된다.
ALLOWED_HOSTS = ['10.0.2.2']INSTALLED_APPS = [
(생략),
"channels",
"whisper",
]ASGI_APPLICATION = "voice_django.asgi.application"
voice_django/urls.py
기본 파일에서 변경한 부분이 없다.
voice_django/wsgi.py
기본 파일에서 변경한 부분이 없다.
my_whisper/__init__.py
my_whisper 디렉토리를 직접 생성하고, 'touch __init__.py' 커맨드로 파일을 생성한다. my_whisper 디렉토리명은 주의할 점이 있다. 그냥 whisper 디렉토리로 생성했다가, openai-whisper 라이브러리 경로랑 충돌해서 한참 동안 에러가 났었다. 경로 충돌이 일어나지 않도록 고유한 이름으로 생성해야 한다.
my_whisper/consumers.py
WhisperConsumer 클래스에 있는 각 함수의 역할을 간단히 적어보았다.
- connect 함수는 웹소켓 연결을 수락하고, Whisper 모델을 로드한다. (현재는 "base" 모델을 사용한다.)
- receive 함수는 클라이언트로부터 base64로 인코딩된 오디오 데이터를 받는다. base64 디코딩과 numpy array로의 변환을 거친 후 transcribe 함수를 통해 텍스트로 변환한다. 그리고 변환된 텍스트를 클라이언트로 전송한다. ensure_ascii 옵션을 False로 지정한 이유가 있다. 영어로 말하면 영어 텍스트로 잘 변환되는데, 한국어로 말하면 변환 텍스트가 'uc770'과 같은 코드로 나왔다. ensure_ascii 옵션을 지정하면 한국어 텍스트로 잘 변환된다.
- audio_bytes_to_numpy 함수는 오디오 바이트를 numpy array로 변환한다. librosa 라이브러리 추가 설치가 필요하다.
import json
import whisper
import numpy as np
import base64
from channels.generic.websocket import AsyncWebsocketConsumer
from io import BytesIO
import librosa
class WhisperConsumer(AsyncWebsocketConsumer):
async def connect(self):
self.room_group_name = 'whisper_room'
await self.accept()
self.model = whisper.load_model("base")
async def disconnect(self, close_code):
pass
async def receive(self, text_data):
data = json.loads(text_data)
audio_base64 = data['audio'] # base64로 인코딩된 오디오 데이터
audio_bytes = base64.b64decode(audio_base64)
audio = self.audio_bytes_to_numpy(audio_bytes)
result = self.model.transcribe(audio, language='ko')
text = result['text']
await self.send(text_data=json.dumps({
'message': text
}, ensure_ascii=False))
def audio_bytes_to_numpy(self, audio_bytes):
audio_data = BytesIO(audio_bytes)
audio, sr = librosa.load(audio_data, sr=16000) # sr: sample rate
return np.array(audio)
my_whisper/routing.py
urls.py와 구조가 비슷하지만, 별도로 생성한 이유가 있다. voice_django/urls.py는 HTTP 프로토콜 관련 라우팅을 하고, my_whisper/routing.py는 WebSocket 관련 라우팅을 한다.
- re_path는 정규 표현식 기반으로 URL 패턴을 정의한다.
- 현재 디렉토리 경로에서 consumers.py를 임포트 한다.
- 'ws/whisper/$'는 WebSocket URL 패턴을 지정한 것이다. Flutter 클라이언트 측 코드의 URL을 보면 'ws://10.0.2.2:8000/ws/whisper/'로 맵핑된 것을 알 수 있다.
- as_asgi()는 cosumers.py에 있는 Consumer 클래스를 ASGI 서버와 통신할 수 있는 애플리케이션 객체로 변환한다.
# whisper/routing.py
from django.urls import re_path
from . import consumers
websocket_urlpatterns = [
re_path(r'ws/whisper/$', consumers.WhisperConsumer.as_asgi()),
]
클라이언트 Flutter 앱 구현 방식
/android/app/src/main/AndroidManifest.xml (Android 권한)
<uses-permission android:name="android.permission.RECORD_AUDIO"/>
<uses-permission android:name="android.permission.INTERNET"/>
/ios/Runner/Info.plist (iOS 권한)
<key>NSMicrophoneUsageDescription</key>
<string>We need access to your microphone for speech-to-text functionality.</string>
<key>NSAppTransportSecurity</key>
<dict>
<key>NSAllowsArbitraryLoads</key>
<true/>
</dict>
pubspec.yaml
- 아래 커맨드로 web_socket_channel, flutter_sound, permission_handler, http 패키지를 추가한다.
flutter pub add {패키지명}dependencies:
flutter:
sdk: flutter
cupertino_icons: ^1.0.8
web_socket_channel: ^3.0.1
flutter_sound: ^9.17.8
permission_handler: ^11.3.1
http: ^1.2.2
/lib/main.dart
- 아래 커맨드로 webflutter pub add {패키지명}
- WebSocketChannel.connect에 적절한 URL을 지정해야 한다. 현재는 안드로이드 에뮬레이터가 호스트 컴퓨터와 통신할 때 사용하는 IP와 routing.py에서 지정했던 URL 패턴을 조합하여 적는다.
- Permission.microphone.request() 함수로 마이크 사용 권한 승인을 받도록 한다.
- startRecorder()와 stopRecorder()로 녹음 시작 및 종료를 하고, 오디오를 base64 인코딩한다.
- channel.sink.add로 웹소켓을 통해 서버로 오디오 파일을 전달한다.
- sendPostRequest는 오늘 구현하는 웹소켓 통신과는 관련이 없는 부분이다. 에뮬레이터와 호스트 컴퓨터와의 통신에서 사용되는 IP를 확인하는 용도로 간단하게 구현했던 부분이다.
- print("Received data from server: $serverData"); 로깅을 통해 터미널에서 서버로부터 변환된 텍스트가 잘 들어오는지 확인했다.
- transcribedText가 존재하면 UI에 띄우고, 없는 경우 "Waiting for speech…" 를 띄우도록 했다.
import 'dart:convert';
import 'package:flutter/material.dart';
import 'package:flutter_sound/flutter_sound.dart';
import 'package:web_socket_channel/web_socket_channel.dart';
import 'package:http/http.dart' as http;
import 'dart:io';
import 'package:permission_handler/permission_handler.dart';
void main() {
runApp(MyApp());
}
class MyApp extends StatelessWidget {
@override
Widget build(BuildContext context) {
return MaterialApp(
title: 'Flutter Whisper',
home: SpeechToTextPage(),
);
}
}
class SpeechToTextPage extends StatefulWidget {
@override
_SpeechToTextPageState createState() => _SpeechToTextPageState();
}
class _SpeechToTextPageState extends State<SpeechToTextPage> {
late WebSocketChannel channel;
final FlutterSoundRecorder _recorder = FlutterSoundRecorder();
bool isRecording = false;
String transcribedText = '';
@override
void initState() {
super.initState();
channel = WebSocketChannel.connect(
Uri.parse('ws://10.0.2.2:8000/ws/whisper/'),
);
_initRecorder();
requestPermission();
}
void _initRecorder() async {
print("Initializing recorder...");
await _recorder.openRecorder();
print("Recorder initialized.");
}
void requestPermission() async {
print("Requesting microphone permission...");
final status = await Permission.microphone.request();
if (status.isGranted) {
print("Microphone permission granted");
} else {
print("Microphone permission denied");
}
}
void _startRecording() async {
print("Starting recording...");
await _recorder.startRecorder(toFile: 'audio.wav');
setState(() {
isRecording = true;
});
print("Recording started.");
}
void _stopRecording() async {
print("Stopping recording...");
final path = await _recorder.stopRecorder();
setState(() {
isRecording = false;
});
print("Recording stopped.");
if (path != null) {
print("Recording saved at: $path");
final audioFile = File(path);
final audioBytes = await audioFile.readAsBytes();
final audioBase64 = base64Encode(audioBytes);
print("Audio encoded to base64.");
channel.sink.add(jsonEncode({
'audio': audioBase64,
}));
print("Audio data sent to server.");
}
}
void _sendPostRequest() async {
final url = Uri.parse('http://10.0.2.2:8000');
print("Sending POST request...");
try {
final response = await http.post(
url,
headers: {'Content-Type': 'application/json'},
body: jsonEncode({'message': 'This is a test POST request'}),
);
if (response.statusCode == 200) {
print('POST request successful: ${response.body}');
} else {
print('POST request failed with status: ${response.statusCode}');
}
} catch (e) {
print('Error sending POST request: $e');
}
}
@override
void dispose() {
print("Disposing resources...");
_recorder.closeRecorder();
channel.sink.close();
super.dispose();
print("Resources disposed.");
}
@override
Widget build(BuildContext context) {
return Scaffold(
appBar: AppBar(
title: Text("Whisper Speech-to-Text"),
),
body: Center(
child: Column(
mainAxisAlignment: MainAxisAlignment.center,
children: [
if (isRecording)
IconButton(
icon: Icon(Icons.stop),
onPressed: _stopRecording,
)
else
IconButton(
icon: Icon(Icons.mic),
onPressed: _startRecording,
),
SizedBox(height: 20),
ElevatedButton(
onPressed: _sendPostRequest,
child: Text("Send POST Request"),
),
SizedBox(height: 20),
StreamBuilder(
stream: channel.stream,
builder: (context, snapshot) {
if (snapshot.connectionState == ConnectionState.active) {
var serverData = snapshot.data;
if (serverData is String) {
print("Received data from server: $serverData");
transcribedText = serverData; // Directly update the text
}
return Container(
padding: EdgeInsets.all(12),
margin: EdgeInsets.symmetric(horizontal: 20),
color: Colors.blue.shade50,
child: Text(
transcribedText.isNotEmpty
? transcribedText
: "Waiting for speech...",
style: TextStyle(fontSize: 18),
),
);
} else {
return Container(
padding: EdgeInsets.all(12),
margin: EdgeInsets.symmetric(horizontal: 20),
color: Colors.grey.shade200,
child: Text(
"Waiting for speech...",
style: TextStyle(fontSize: 18),
),
);
}
},
),
],
),
),
);
}
}
구현 결과
에뮬레이터와 서버 간의 연결 확인용으로 만들었던 버튼을 누르면, http POST 요청이 잘 가는 것을 볼 수 있다. (파란색)
웹 소켓 WSCONNECT 표시도 볼 수 있다. (빨간색) 마이크 버튼을 누르고 말하면 웹 소켓을 통해 음성이 서버로 전달되며, 서버에서 텍스트로 변환하여 클라이언트에 전달한다. 양방향 통신을 통해 클라이언트에 실시간으로 텍스트가 표시되었다.
To do
- 현재는 "base" 모델을 사용했기 때문에 로드 속도가 빠르지만, 정확도가 부족하다. "large" 모델이나 상위 버전 모델은 정확도가 높은 대신에 로드 시간이 꽤 걸린다. 모델 종류와 로드 시간을 어떻게 처리할지 생각해야 한다.
- transcribe 할 때 language='ko' 옵션을 주었으나, 다른 나라 언어가 섞여서 나오는 문제가 있다. 옵션을 더 뜯어봐야 한다.
- 에뮬레이터를 사용하면 ALLOWED_HOSTS에 '10.0.2.2'를 추가하고 URL에도 이걸 적으면 된다. 그러나 실제 배포 환경에서의 웹소켓 통신 URL는 다를 수 있어서 찾아봐야 한다.
- Android 에뮬레이터에서만 테스트해 봐서, 실제 Android 기기와 iOS 기기에서도 테스트해 봐야 한다.
* 참고 사이트
- 공식 Django Docs(첫 번째 장고 앱 작성하기, part 1): https://docs.djangoproject.com/ko/5.1/intro/tutorial01/
- [Django] Django rest framework로 웹소켓 채팅 서버 구현하기(1): https://velog.io/@mimijae/Django-Django-rest-framework로-웹소켓-채팅-서버-구현하기1
- Django Channels 배포: https://gunjoon.tistory.com/54