Whisper 음성 전사(Transcription)에 화자 식별(Speaker Diarization)을 통합하는 구현: 시간 중첩을 이용한

음성 전사(Transcription)를 할 때 흔히 다음과 같은 출력을 원하게 됩니다.

Speaker 0: 오늘은 회의록을 확인하겠습니다.
Speaker 1: 네, 우선 스케줄부터 확인하죠.
Speaker 0: 다음 회차까지의 작업도 정리하고 싶습니다.

Whisper는 전사 정확도가 높으며, 로컬에서도 whisper.cpp를 사용하면 다루기 쉽습니다.

하지만 Whisper의 주요 역할은 "무엇을 말했는가"를 텍스트화하는 것입니다.

"누가 언제 말했는가"를 구분하려면, 별도의 화자 분리(Speaker Diarization) 타임라인이 필요합니다.

이 기사에서는 Whisper의 전사 결과에 별도의 화자 분리 타임라인을 겹쳐서 화자 라벨을 붙이는 방법을 설명합니다.

여기서 말하는 화자 식별은 인물의 실명을 맞히는 인증이 아닙니다.

음성 내의 발화 구간을 Speaker 0, Speaker 1과 같이 나누는, 이른바 화자 분리(Speaker Diarization)입니다.

구현 예시는 TypeScript 스타일로 작성하지만, 사고방식은 Python이나 다른 언어에서도 동일합니다.

필요한 것은 Whisper의 세그먼트(Segment) 시간과 화자 분리 측의 타임라인을 동일한 시간 단위로 다루는 것입니다.

이 구현에서 수행하는 작업은 매우 단순합니다.

• 음성 파일을 ffmpeg로 16kHz / mono / 16-bit PCM WAV로 변환한다
• 동일한 WAV를 Whisper와 로컬 화자 분리에 전달한다
• Whisper로부터 "문자열 + 시작 시간 + 종료 시간"을 가져온다
• 화자 분리로부터 "시작 시간 + 종료 시간 + 화자 번호"를 가져온다
• Whisper의 각 세그먼트에 시간이 가장 많이 중첩된 화자를 할당한다
• 인접한 동일 화자의 세그먼트를 합친다
• 최종적으로 Speaker N: transcript 형태로 정형화한다

도식화하면 다음과 같습니다.

음성 파일
|
v
...

포인트는 텍스트와 화자를 직접 연결하려고 하지 않는 것입니다.

텍스트 내용으로 추측하는 것이 아니라, 양쪽의 결과를 **시간축(Time axis)**으로 결합합니다.

이 방식이 구현이 단순하고 테스트하기 쉬우며, Whisper 측의 인식 품질을 그대로 활용할 수 있습니다.

Whisper와 조합하여 화자 식별을 구현하려면 주로 다음 요소가 필요합니다.

요소	역할
음성 정규화 (Audio Normalization)	입력 음성을 16kHz / mono / PCM WAV 등으로 맞춤
...

Whisper 실행은 whisper.cpp의 CLI를 자식 프로세스로 호출합니다.

화자 분리는 최종적으로 다음과 같은 타임라인을 반환할 수 있으면 충분합니다.

export type SpeakerTimelineSegment = {
  start: number; // 초
  end: number; // 초
  ...
}

이 SpeakerTimelineSegment[]를 Whisper의 JSON 출력에 나중에 맞춥니다.

먼저, 화자 식별 통합에 필요한 데이터 구조를 작게 정의합니다.

요청(Request) 측에서는 화자 식별 사용 여부와, 알고 있는 경우의 최대 화자 수를 전달합니다.

export type TranscribeOptions = {
  speakerLabel: boolean;
  maxSpeakers?: number;
  ...
}

출력(Output) 측에서는 각 세그먼트에 speakerLabel을 선택적(optional)으로 갖게 합니다.

export type TranscriptSegment = {
  speakerLabel?: string;
  transcript: string;
  ...
}

화자 분리를 사용할 수 없거나 실패한 경우에도 speakerLabel이 없는 전사 결과로 반환할 수 있습니다.

이 optional 설계는 매우 중요합니다.

화자 식별은 편리하지만, 전사 본체보다 깨지기 쉬운 처리입니다.

모델 파일이 없거나, 네이티브 모듈을 로드할 수 없거나, 음성 품질이 나쁘거나, 화자를 분리할 수 없는 경우가 발생할 수 있습니다.

그 경우에도 Whisper의 전사 자체가 실패하지 않도록 하기 위해 speakerLabel?: string으로 설정합니다.

처리 입구는 음성 파일과 옵션을 받아 전사 결과를 반환하는 함수로 만들어 두면 다루기 쉽습니다.

export async function transcribeWithSpeakerLabels(
inputPath: string,
options: TranscribeOptions,
...

여기서 중요한 것은 처리를 세 가지로 나누는 것입니다.

runWhisper()는 Whisper의 JSON을 반환합니다.

diarize()는 화자 타임라인(Speaker Timeline)을 반환합니다.

parseWhisperJson()은 두 결과를 시간축(Time axis)을 기준으로 맞춥니다.

이렇게 나누어 두면 Whisper만 교체하거나, 화자 분리(Speaker Diarization)만 별도의 구현으로 변경하기가 쉬워집니다.

또한 화자 분리는 선택 사항(optional)이므로, speakerLabel=false인 경우에는 Whisper만으로 동작할 수 있습니다.

Whisper와 화자 분리 결과를 시간으로 맞추려면, 두 작업 모두 동일한 음성을 바라보고 있어야 합니다.

이를 위해 업로드된 음성은 먼저 ffmpeg를 사용하여 WAV로 변환합니다.

private async convertToWav(inputPath: string, wavPath: string): Promise<void> {
const ffmpegBin = this.config.ffmpegBin ?? 'ffmpeg';
const result = await this.commandRunner(ffmpegBin, [
...

-ar 16000으로 16kHz, -ac 1로 모노(mono), pcm_s16le로 16-bit PCM 설정을 적용합니다.

이 input.wav를 Whisper와 화자 분리 양쪽 모두에 전달합니다.

만약 이를 서로 다른 음성 파일로 처리하면 시간축이 미세하게 어긋나게 되어, 후속 단계인 매칭(Matching) 과정이 불안정해집니다.

Whisper 측은 whisper.cpp의 CLI를 사용합니다.

Whisper를 CLI로 실행할 때는 JSON 출력을 활성화해야 합니다.

private async runWhisperCandidate(
candidate: WhisperCandidate,
modelPath: string,
...

중요한 것은 -oj 옵션입니다.

이를 통해 Whisper의 결과를 JSON으로 받을 수 있습니다.

이 JSON에는 구현상 transcription[].offsets 또는 segments[].start/end와 같은 시간 정보가 포함됩니다.

나중에 화자 타임라인과 맞추기 위해서는 이 시간 정보가 반드시 필요합니다.

Whisper의 전사(Transcription)와 화자 분리는 동일한 WAV 파일에 대해 실행할 수 있습니다.

따라서 구현 시에는 화자 분리를 먼저 Promise로 시작하고, Whisper가 완료될 때까지 기다린 후 대조(Matching)합니다.

Whisper 실행과 화자 분리를 병렬로 처리하는 코드

다음은 본문과 관련된 부분만 추린 구현 예시입니다.

private async runWhisper(
wavPath: string,
jobDir: string,
...

이러한 구조로 만든 데에는 두 가지 이유가 있습니다.

첫 번째는 대기 시간을 줄이기 위해서입니다.

Whisper가 끝난 뒤에 화자 분리를 시작하면, 단순히 처리 시간이 합산되어 늘어나게 됩니다.

두 번째는 책임을 분리하기 위해서입니다.

runWhisperRaw()는 Whisper의 JSON을 반환할 뿐입니다.

runWhisperDiarization()은 화자 타임라인을 반환할 뿐입니다.

parseWhisperJson()이 이 두 가지를 결합합니다.

이렇게 나누어 두면 각각을 테스트하기가 훨씬 용이합니다.

화자 분리는 선택 사항입니다.

따라서 실패할 경우 undefined를 반환하고, Whisper의 전사 결과만으로 작업을 계속 진행합니다.

화자 분리의 폴백(Fallback) 처리

본문에서는 구현의 본질을 잘 보여줄 수 있도록 어댑터(Adapter) 이름을 일반화하였습니다.

실제 코드도 동일한 방침을 따라, 사용 불가능하거나 미구현되었거나 실패했을 경우 undefined를 반환합니다.

private async runWhisperDiarization(
wavPath: string,
options: DiarizationOptions,
...

여기서 중요한 점은 화자 분리의 실패가 전사 전체의 실패로 이어지지 않게 하는 것입니다.

사용자 입장에서는 화자 라벨이 포함된 결과가 가장 이상적이기 때문입니다.

하지만 화자 라벨을 얻지 못하는 경우라도, 전사(Transcription) 본문이 반환되는 편이 더 실용적입니다.

따라서, parseWhisperJson(raw, timeline)의 timeline은 선택 사항(optional)으로 되어 있습니다.

화자 분리(Speaker Diarization) 측은 Whisper와 동일한 WAV를 읽어 SpeakerTimelineSegment[]를 반환합니다.

예를 들어 sherpa-onnx 계열의 화자 분리에서는 다음 두 종류의 ONNX 모델로 화자 타임라인(Speaker Timeline)을 만들 수 있습니다.

모델	용도
pyannote-segmentation.onnx	화자가 바뀌는 구간을 찾음
speaker-embedding-campplus-zh-en.onnx	각 구간의 화자 임베딩 (Speaker Embedding)을 생성

구현상으로는 화자 분리만 실행할 때 이 두 가지만 확보합니다.

const DIARIZATION_MODEL_FILES = [
'pyannote-segmentation.onnx',
'speaker-embedding-campplus-zh-en.onnx',
...

화자 수가 지정된 경우에는 고정 클러스터(Fixed Cluster) 수를, 지정되지 않은 경우에는 임계값 클러스터링 (Threshold Clustering)을 사용합니다.

const numClusters = maxSpeakers > 0 ? maxSpeakers : -1;
const diarizer = new sherpa.OfflineSpeakerDiarization({
segmentation: {
...

maxSpeakers를 지정하면 numClusters에 그 값을 전달합니다.

예를 들어 회의 참가자가 2명이라는 것을 알고 있다면, maxSpeakers=2로 설정하여 클러스터링을 안정적으로 만들 수 있습니다.

인원수를 모르는 경우에는 -1로 설정하여 임계값(Threshold) 기반으로 나눕니다.

Whisper의 JSON 형식은 실행 방법이나 호환 형식에 따라 조금씩 다릅니다.

이 구현에서는 먼저 offsets.from/to를 우선시하고, 없을 경우 start/end를 확인합니다.

const whisperSegmentRange = (
segment: WhisperJsonSegment,
): { start: number; end: number } | undefined => {
...

offsets는 밀리초(ms) 단위이므로 초(s)로 변환합니다.

화자 타임라인은 초 단위입니다.
단위를 맞추지 않으면 중첩(Overlap) 계산이 망가집니다.

또한, end >= start 확인 과정도 포함되어 있습니다.
음성 처리에서는 손상된 입력이나 호환되지 않는 JSON으로 인해 이상한 값이 들어올 가능성이 있습니다.
여기서 최소한의 정규화(Normalization)를 해두면 후속 로직을 단순하게 만들 수 있습니다.

가장 중요한 부분은 여기입니다.
Whisper의 1개 세그먼트에 대해, 화자 타임라인 내에서 시간이 가장 많이 중첩되는 화자를 선택합니다.

export const speakerForRange = (
range: { start: number; end: number },
timeline: SpeakerTimelineSegment[],
...

생각하는 방식은 다음과 같습니다.

Whisper segment:
1.0s ---------------- 2.0s
Speaker timeline:
...

단순히 "시작 시간이 가까운 화자"를 선택하지 않는 것이 핵심입니다.

Whisper의 세그먼트는 화자의 전환(Speaker Turn)과 완전히 일치하지 않을 수 있습니다.
하나의 Whisper 세그먼트가 화자 전환 지점을 약간 걸쳐 있는 경우가 있습니다.
그럴 경우 시작 시간만 보면 오류가 발생하기 쉽습니다.
구간의 중첩을 확인하는 것이 더 자연스럽습니다.

단, 짧은 무음이나 VAD(Voice Activity Detection) 경계에서 어떤 화자 구간과도 중첩되지 않을 수도 있습니다.
그럴 경우에는 가장 가까운 화자를 폴백(Fallback)으로 반환합니다.
타임라인이 비어 있을 때만 undefined를 반환합니다.

이 폴백 덕분에 수백 밀리초의 틈새에서 speakerLabel이 갑자기 사라지는 것을 방지할 수 있습니다.

다음으로, Whisper JSON을 앱의 응답 형식으로 변환합니다.

Whisper JSON을 Transcript[]로 변환하는 코드

다음은 parseWhisperJson입니다.

화자 식별(Speaker Diarization)과 관련된 부분을 중심으로 추출한 구현 예시입니다.

const parseWhisperJson = (
raw: string,
speakerTimeline?: SpeakerTimelineSegment[],
...

여기서 수행하는 작업은 다음 4가지입니다.

• Whisper JSON의 언어 코드(Language Code)를 읽음
• transcription / segments / results.audio_segments 중 하나에서 세그먼트(Segment)를 읽음
• 세그먼트에 화자 정보가 없으면, 화자 타임라인(Speaker Timeline)과의 시간 중첩을 통해 보완
• 마지막으로 인접한 동일 화자의 세그먼트를 결합

Whisper JSON의 형식이 여러 가지이므로, 입구 부분에서는 다소 유연하게 처리하도록 구성했습니다.

다만, 내부 처리 형식은 TranscriptSegment[]와 같은 단순한 구조로 통일합니다.

이러한 "외부 형식은 유연하게 받고, 내부 형식은 고정한다"는 방침을 따르면, 후속 단계인 표시, 저장, 정렬, 테스트가 용이해집니다.

화자 라벨(Speaker Label)을 붙인 후 그대로 반환하면 세그먼트가 너무 잘게 나누어질 수 있습니다.

예를 들어,

Speaker 0: こんにちは 世界
Speaker 0: ありがとう
Speaker 1: さようなら

보다,

Speaker 0: こんにちは世界ありがとう
Speaker 1: さようなら

쪽이 더 읽기 쉽습니다.

따라서 인접한 동일 화자의 세그먼트는 결합합니다.

export const mergeAdjacentSpeakerSegments = (segments: TranscriptSegment[]): TranscriptSegment[] =>
segments.reduce((prev, item) => {
if (prev.length === 0 || prev[prev.length - 1].speakerLabel !== item.speakerLabel) {
...

또한, 일본어의 경우 Whisper 출력에 포함된 공백이 읽기 불편할 때가 있습니다.

그래서 언어 코드가 일본어(ja)인 경우에는 공백을 제거합니다.

const normalizeTranscriptText = (text: string, languageCode: string): string => {
const normalized = text.trim();
return languageCode.startsWith('ja') ? normalized.replace(/ /g, '') : normalized;
...

이 두 가지 처리를 통해 최종적인 전사(Transcription) 결과가 훨씬 읽기 좋아집니다.

화자 식별은 필요한 실행 모듈이나 모델이 있는 경우에만 사용할 수 있습니다.

따라서 본격적으로 통합할 때는, 처리를 시작하기 전에 "화자 분리(Speaker Diarization)를 사용할 수 있는지"를 판별하는 함수를 준비해 두는 것이 편리합니다.

Whisper의 화자 식별 지원 여부는 고정값이 아니라, 로컬 화자 분리 기능을 사용할 수 있는지에 따라 동적으로 결정합니다.

private async detectWhisperDiarizationSupport(): Promise<boolean> {
if (typeof this.localDiarizationEngine.diarizeFile !== 'function') {
return false;
...

이 함수가 false를 반환하는 경우에는 화자 분리를 실행하지 않고, Whisper의 전사 결과만 반환합니다.

이렇게 설계해 두면 다음과 같은 상황에서도 전체 프로세스가 망가지는 것을 방지할 수 있습니다.

• 모델이 아직 없는 경우
• 네이티브 모듈(Native Module)을 불러올 수 없는 경우
• 대상 OS에서 지원하지 않는 경우

화자 식별을 사용할 수 없을 때는 speakerLabel이 없는 Whisper 결과로 폴백(Fallback)합니다.

최종적으로 speakerLabel이 있으면 화자 이름을 앞에 붙이고, 없으면 본문만 출력합니다.

const formatTranscript = (segments: TranscriptSegment[]): string =>
segments
.map((segment) =>
...

출력 예시는 다음과 같습니다.

Speaker 0: こんにちは世界ありがとう
Speaker 1: さようなら

여기서 주의할 점은, Speaker 0은 인물의 실명이 아니라는 것입니다.

화자 분리 (Speaker Diarization)를 통해 알 수 있는 것은 "같은 목소리의 클러스터 (Cluster)"입니다.

그것이 누구인지는 별도의 본인 확인이나 사용자 입력이 없는 한 알 수 없습니다.

즉, 이 레이어의 책임은 Speaker 0, Speaker 1과 같이 구간을 나누는 데까지입니다.

화자 식별은 겉모습만으로 확인하면 깨지기 쉽습니다.

테스트에서는 시간 중첩 (Time overlap)의 contract를 준수합니다.

최대 중첩으로 화자를 할당하는 테스트

const whisperSegments = [
{
  text: 'こんにちは 世界',
  from: 0,
  to: 1000
},
{
  text: 'ありがとう',
  from: 1000,
  to: 2000
},
...

이 테스트에서 확인하는 것은 주로 다음 3가지 점입니다.

• Whisper의 두 번째 세그먼트가 Speaker 0과 0.9초, Speaker 1과 0.1초 중첩되므로 Speaker 0이 된다.
• 인접한 Speaker 0의 세그먼트가 결합된다.
• maxSpeakers가 화자 분리에 인계된다.

나아가, speakerForRange 단독으로도 테스트하고 있습니다.

const timeline: SpeakerTimelineSegment[] = [
{ start: 0, end: 2, speaker: 0 },
{ start: 2, end: 4, speaker: 1 },
...

중첩이 발생하는 경우에는 먼저 나타난 화자를 선택합니다.

중첩이 없는 경우에는 가장 가까운 화자를 선택합니다.

타임라인이 비어 있는 경우에만 화자 없음으로 처리합니다.

이 부분은 작은 로직이지만, 실제 사용자 경험 (UX)에 상당히 큰 영향을 미칩니다.

Whisper는 받아쓰기 (Transcription)에 강점이 있습니다.

따라서 화자 식별을 넣기 위해 받아쓰기 본문을 다른 처리로 대체할 필요는 없습니다.

Whisper가 출력한 텍스트를 그대로 사용하고, 화자 라벨 (Speaker label)만 나중에 추가합니다.

이 방식이 책임이 더 명확합니다.

텍스트 내용만으로 화자를 추측하는 것은 어렵습니다.

예를 들어 "네", "그렇군요", "부탁드립니다"와 같은 짧은 발화는 내용만으로 화자를 판단할 수 없습니다.

반면, Whisper와 화자 분리 모두 시간 정보 (Time information)를 가지고 있습니다.

따라서 결합 키 (Join key)로 시간을 사용합니다.

화자 식별은 음성 품질이나 환경의 영향을 받습니다.

화자 분리가 실패하더라도 Whisper의 결과까지 실패하게 만들면 사용자 경험이 나빠집니다.

그러므로 화자 분리는 optional로 설정하여, 실패 시에는 라벨이 없는 받아쓰기로 반환합니다.

화자 식별을 사용할 수 있는지 여부는 호출 측의 설정값만으로는 판단할 수 없습니다.

필요한 실행 모듈이나 모델의 유무를 확인하여 supportsSpeakerDiarization을 반환합니다.

호출 측은 그에 따라 화자 분리를 활성화할지 여부를 결정합니다.

이 설계에서는 가장 중요한 로직이 speakerForRange()로 분리되어 있습니다.

음성 파일을 사용하지 않더라도,

speakerForRange({ start: 1.5, end: 3.5 }, timeline)

와 같이 순수 함수 (Pure function)로서 테스트할 수 있습니다.

음성 처리 전체를 매번 돌리지 않아도 화자 할당의 contract를 준수할 수 있습니다.

화자 식별과 통합하려면 Whisper의 출력에는 반드시 세그먼트 시간이 필요합니다.

예를 들어 whisper.cpp에서는 -oj를 전달하여 JSON을 읽어옵니다.

시간축을 맞추기 위해 양쪽 모두에 동일한 input.wav를 전달합니다.

입력 음성을 별도로 변환하면 샘플 레이트 (Sample rate)나 무음 처리의 차이로 인해 시간이 어긋날 수 있습니다.

Whisper의 offsets는 밀리초 (ms) 단위이고, 화자 타임라인은 초 (s) 단위가 되기 쉽습니다.

이 구현에서는 offsets.from / 1000과 같이 초 단위로 변환한 후 비교합니다.

프로덕션 환경을 위해서는 화자 분리가 실패하더라도 받아쓰기는 반환하는 설계가 다루기 쉽습니다.

speakerLabel?: string과 같이 optional로 설정해 두면 후속 처리도 자연스럽게 fallback할 수 있습니다.

Whisper의 세그먼트는 세밀하게 나뉩니다.

화자 라벨을 붙인 후, 같은 화자가 이어진다면 결합하는 것이 가독성이 좋습니다.

회의 참가자 수를 알고 있는 경우, 최대 화자 수를 지정할 수 있으면 클러스터링 (Clustering)이 안정되기 쉽습니다.

예를 들어 1~10 사이의 정수만 허용한다면 다음과 같이 검증할 수 있습니다.

export const speakerNumSchema = z
.number()
.int({ message: '화자의 최대 수는 정수로 설정해 주세요' })
...

• Whisper의 전사 (Transcription) 결과에 화자 라벨을 붙이려면, Whisper 세그먼트 (Segment)와 화자 타임라인 (Speaker Timeline)을 시간 중첩 (Time Overlap)으로 대조하면 됩니다.
• 구현의 핵심은 runWhisper(), parseWhisperJson(), speakerForRange() 세 가지이며, Whisper 본문은 그대로 사용하고 Speaker N만 사후에 추가합니다.
• 화자 분리 (Speaker Diarization)는 선택 사항 (Optional)으로 설정하여, 실패 시 라벨이 없는 Whisper 결과를 반환하도록 함으로써 실용적인 관점에서 견고한 전사 기능을 만들 수 있습니다.

Whisper와 화자 식별을 통합할 때 너무 어렵게 생각할 필요는 없습니다.

Whisper는 "무엇을 말했는가"를 출력합니다.

화자 분리는 "누가 언제 말했는가"를 출력합니다.

이 두 가지를 시간축 (Time Axis)에서 맞추면, Speaker 0: ...와 같은 실용적인 결과를 만들 수 있습니다.

이 글에서는 Whisper의 JSON 세그먼트, 화자 분리 타임라인, 그리고 speakerForRange()를 통한 최대 중첩 판정을 조합하는 방법을 설명했습니다.

구현 자체는 규모가 작지만, 음성 변환, Whisper 실행, 화자 분리, 시간축 정렬 (Alignment), 폴백 (Fallback), 테스트까지 연결하면 상당히 쓸모 있는 형태가 됩니다.

유사한 기능을 만들 경우에는 먼저 SpeakerTimelineSegment[]를 생성하고, Whisper 세그먼트에 대해 최대 중첩을 기준으로 Speaker N을 부여하는 것부터 시작하는 것이 좋다고 생각합니다.