Meeting Transcript

Whisper Diarizer

오디오 파일을 분석하여 음성을 텍스트로 변환(STT)하고 화자 분리(diarization)하여 두 결과를 합쳐서 누가 언제 무슨 말을 했는지를 표로 저장하는 완전 자동화된 음성 분석 파이프라인

---

whisper-large-v3-turbo 모델 준비

(Google Colab을 이용할 경우 과정 생략)

1. openai/whisper-large-v3-turbo 사이트에 접속

2. 필요한 package 설치

pip install --upgrade pip
pip install --upgrade transformers datasets[audio] accelerate

3. FFMPEG 설치

• ffmpeg-git-full.7z 를 다운로드하여 압축 풀기

4. FFMPEG 실행

• whisper-large-v3-turbo 모델을 사용할 때마다 해당 코드 추가 (Google Colab에서는 생략 가능)

import os

ffmpeg_path = YOUR-FFMPEG-PATH
os.environ["PATH"] += os.pathsep + ffmpeg_path

5. PyTorch 설치

• PC 환경에 맞는 설치 명령어를 확인하여 설치

---

speaker-diarization-3.1 모델 준비

1. Hugging Face에서 토큰 발급

• 로그인 > Settings > Access Tokens > Create new token

2. pyannote/speaker-diarization-3.1에서 [Agree and access repository] 누르기

• Company/university: personal
• Website: I have no website

3. pyannote/segmentation-3.0에서 [Agree and access repository] 누르기

• Company/university: personal
• Website: I have no website

4. Token permission 허용

• Settings > Access Tokens > : > Edit permissions > Repositories permissions에서 ‘pyannote/segmentation-3.0’, ‘speaker-diarization-3.1’ 등록 > ‘Read access to contents of selected repos’ 체크

5. 필요한 패키지 설치

• Google Colab에서는 numpy 설치하지 않아도 됨

pip install pyannote.audio
pip install numpy
pip install python-dotenv

6. 토큰 사용

• token.env 파일 생성 및 토큰 적용

HUGGING_FACE_TOKE=YOUT-TOKEN

import os
from dotenv import load_dotenv

env_path =  "token.env"
load_dotenv(dotenv_path=env_path)

HUGGING_FACE_TOKEN = os.getenv("HUGGING_FACE_TOKEN")

---

Whisper Diarizer 코드 작성

• 전체 코드

import torch
import pandas as pd
from transformers import AutoModelForSpeechSeq2Seq, AutoProcessor, pipeline
from pyannote.audio import Pipeline

def whisper_stt(
    audio_file_path: str,
    output_file_path: str = "./output.csv"
):

    device = "cuda:0" if torch.cuda.is_available() else "cpu"
    torch_dtype = torch.float16 if torch.cuda.is_available() else torch.float32
    model_id = "openai/whisper-large-v3-turbo"

    model = AutoModelForSpeechSeq2Seq.from_pretrained(
        model_id, torch_dtype=torch_dtype,
        low_cpu_mem_usage=True,
        use_safetensors=True
    )
    model.to(device)

    processor = AutoProcessor.from_pretrained(model_id)

    pipe = pipeline(
        "automatic-speech-recognition",
        model=model,
        tokenizer=processor.tokenizer,
        feature_extractor=processor.feature_extractor,
        torch_dtype=torch_dtype,
        device=device,
        return_timestamps=True,  # 청크별로 타임스탬프를 반환
        chunk_length_s=10,  # 입력 오디오를 10초씩 나누기
        stride_length_s=2,  # 2초씩 겹치도록 청크 나누기
    )

    result = pipe(audio_file_path)
    df = whisper_to_dataframe(result, output_file_path)

    return result, df

def whisper_to_dataframe(result, output_file_path):
    start_end_text = []

    for chunk in result["chunks"]:
        start = chunk["timestamp"][0]
        end = chunk["timestamp"][1]
        text = chunk["text"].strip()
        start_end_text.append([start, end, text])
        df = pd.DataFrame(start_end_text, columns=["start", "end", "text"])
        df.to_csv(output_file_path, index=False, sep="|")

    return df

def speaker_diarization(
        audio_file_path: str,
        output_rttm_file_path: str,
        output_csv_file_path: str
    ):
    pipeline = Pipeline.from_pretrained(
        "pyannote/speaker-diarization-3.1",
        use_auth_token=HUGGING_FACE_TOKEN
    )

    if torch.cuda.is_available():
        pipeline.to(torch.device("cuda"))

    diarization_pipeline = pipeline(audio_file_path)

    # 화자 분리 원본(rttm) 파일 저장
    with open(output_rttm_file_path, "w", encoding='utf-8') as rttm:
        diarization_pipeline.write_rttm(rttm)

    # 화자 분리 원본(rttm) 파일 불러오기
    df_rttm = pd.read_csv(
        output_rttm_file_path,
        sep=' ',
        header=None,
        names=['type', 'file', 'chnl', 'start', 'duration', 'C1', 'C2', 'speaker_id', 'C3', 'C4']
    )

    df_rttm["end"] = df_rttm["start"] + df_rttm["duration"]

    # 연속된 발화를 하나의 구간으로 묶기
    df_rttm["number"] = None
    df_rttm.at[0, "number"] = 0

    for i in range(1, len(df_rttm)):
        if df_rttm.at[i, "speaker_id"] != df_rttm.at[i-1, "speaker_id"]:
            df_rttm.at[i, "number"] = df_rttm.at[i-1, "number"] + 1
        else:
            df_rttm.at[i, "number"] = df_rttm.at[i-1, "number"]

    df_rttm_grouped = df_rttm.groupby("number").agg(
        start=pd.NamedAgg(column='start', aggfunc='min'),
        end=pd.NamedAgg(column='end', aggfunc='max'),
        speaker_id=pd.NamedAgg(column='speaker_id', aggfunc='first')
    )

    df_rttm_grouped["duration"] = df_rttm_grouped["end"] - df_rttm_grouped["start"]

    # 화자 분리 csv 파일 저장
    df_rttm_grouped.to_csv(
        output_csv_file_path,
        index=False,
        encoding='utf-8'
    )

    return df_rttm_grouped

def stt_to_rttm(
        audio_file_path: str,
        stt_output_file_path: str,
        rttm_file_path: str,
        rttm_csv_file_path: str,
        final_output_csv_file_path: str
    ):

    result, df_stt = whisper_stt(
        audio_file_path,
        stt_output_file_path
    )

    df_rttm = speaker_diarization(
        audio_file_path,
        rttm_file_path,
        rttm_csv_file_path
    )

    # df_stt와 df_rttm을 시간 기준으로 매칭해서,각 화자에게 어떤 텍스트가 해당되는지를 자동으로 배정
    df_rttm["text"] = ""

    for i_stt, row_stt in df_stt.iterrows():
        overlap_dict = {}
        for i_rttm, row_rttm in df_rttm.iterrows():
            overlap = max(0, min(row_stt["end"], row_rttm["end"]) - max(row_stt["start"], row_rttm["start"])) # row_rttm에 대해 row_stt과 시간이 얼마나 겹치는지 계산
            overlap_dict[i_rttm] = overlap

        # 가장 겹친 정도가 큰 화자 구간을 선택
        max_overlap = max(overlap_dict.values())
        max_overlap_idx = max(overlap_dict, key=overlap_dict.get)

        # 화자 구간에 텍스트 추가
        if max_overlap > 0:
            df_rttm.at[max_overlap_idx, "text"] += row_stt["text"] + "\n"

    # 최종 결과 파일 저장
    df_rttm.to_csv(
        final_output_csv_file_path,
        index=False,
        sep='|',
        encoding='utf-8'
    )

    return df_rttm

if __name__ == "__main__":
    audio_file_path = "./data/conversation.mp3" # 원본 오디오 파일
    stt_output_file_path = "./data/conversation.csv"    # STT 결과 파일
    rttm_file_path = "./data/diarization.rttm" # 화자 분리 원본(rttm) 파일
    rttm_csv_file_path = "./data/diarization.csv" # 화자 분리 CSV 파일
    final_csv_file_path = "./data/final.csv" # 최종 결과 파일

    df_rttm = stt_to_rttm(
        audio_file_path,
        stt_output_file_path,
        rttm_file_path,
        rttm_csv_file_path,
        final_csv_file_path
    )

• 함수별 설명 및 결과

  whisper_stt

  • 코드

  def whisper_stt(
      audio_file_path: str,
      output_file_path: str = "./output.csv"
  ):
  
      device = "cuda:0" if torch.cuda.is_available() else "cpu"
      torch_dtype = torch.float16 if torch.cuda.is_available() else torch.float32
      model_id = "openai/whisper-large-v3-turbo"
  
      model = AutoModelForSpeechSeq2Seq.from_pretrained(
          model_id, torch_dtype=torch_dtype,
          low_cpu_mem_usage=True,
          use_safetensors=True
      )
      model.to(device)
  
      processor = AutoProcessor.from_pretrained(model_id)
  
      pipe = pipeline(
          "automatic-speech-recognition",
          model=model,
          tokenizer=processor.tokenizer,
          feature_extractor=processor.feature_extractor,
          torch_dtype=torch_dtype,
          device=device,
          return_timestamps=True,  # 청크별로 타임스탬프를 반환
          chunk_length_s=10,  # 입력 오디오를 10초씩 나누기
          stride_length_s=2,  # 2초씩 겹치도록 청크 나누기
      )
  
      result = pipe(audio_file_path)
      df = whisper_to_dataframe(result, output_file_path)
  
      return result, df

  
  

  • 설명

    • Whisper 모델을 불러오고 설정
    • 입력 오디오 파일을 10초 단위로 나눠서 텍스트로 변환
    • 결과를 CSV로 저장하고, DataFrame으로 반환

  
  

  • result

  {
     "text":"지금부터 저랑 그 역할극을 합시다 ...",
     "chunks":[
        {
           "timestamp":(1.0, 11.0),
           "text":" 지금부터 저랑 그 역할극을 합시다 역할극을 스탠딩 ..."
        },
        {
           "timestamp":(12.0, 17.0),
           "text":"그래서 좀 재밌고 자연스럽고 유머러스하게 저랑 대화 ..."
        },
        ...
     ]
  }

  whisper_to_dataframe

  • 코드

  def whisper_to_dataframe(result, output_file_path):
      start_end_text = []
  
      for chunk in result["chunks"]:
          start = chunk["timestamp"][0]
          end = chunk["timestamp"][1]
          text = chunk["text"].strip()
          start_end_text.append([start, end, text])
          df = pd.DataFrame(start_end_text, columns=["start", "end", "text"])
          df.to_csv(output_file_path, index=False, sep="|")
  
      return df

  
  

  • 설명

    • Whisper STT 결과 (result["chunks"])에서 시작/끝 시간 + 텍스트를 추출
    • 해당 정보를 표 (DataFrame) 형태로 정리 및 반환
    • CSV 파일로 저장 (| 구분자 사용)

  
  

  • df
    
  speaker_diarization

  • 코드

  def speaker_diarization(
          audio_file_path: str,
          output_rttm_file_path: str,
          output_csv_file_path: str
      ):
      pipeline = Pipeline.from_pretrained(
          "pyannote/speaker-diarization-3.1",
          use_auth_token=HUGGING_FACE_TOKEN
      )
  
      if torch.cuda.is_available():
          pipeline.to(torch.device("cuda"))
  
      diarization_pipeline = pipeline(audio_file_path)
  
      # 화자 분리 원본(rttm) 파일 저장
      with open(output_rttm_file_path, "w", encoding='utf-8') as rttm:
          diarization_pipeline.write_rttm(rttm)
  
      # 화자 분리 원본(rttm) 파일 불러오기
      df_rttm = pd.read_csv(
          output_rttm_file_path,
          sep=' ',
          header=None,
          names=['type', 'file', 'chnl', 'start', 'duration', 'C1', 'C2', 'speaker_id', 'C3', 'C4']
      )
  
      df_rttm["end"] = df_rttm["start"] + df_rttm["duration"]
  
      # 연속된 발화를 하나의 구간으로 묶기
      df_rttm["number"] = None
      df_rttm.at[0, "number"] = 0
  
      for i in range(1, len(df_rttm)):
          if df_rttm.at[i, "speaker_id"] != df_rttm.at[i-1, "speaker_id"]:
              df_rttm.at[i, "number"] = df_rttm.at[i-1, "number"] + 1
          else:
              df_rttm.at[i, "number"] = df_rttm.at[i-1, "number"]
  
      df_rttm_grouped = df_rttm.groupby("number").agg(
          start=pd.NamedAgg(column='start', aggfunc='min'),
          end=pd.NamedAgg(column='end', aggfunc='max'),
          speaker_id=pd.NamedAgg(column='speaker_id', aggfunc='first')
      )
  
      df_rttm_grouped["duration"] = df_rttm_grouped["end"] - df_rttm_grouped["start"]
  
      # 화자 분리 csv 파일 저장
      df_rttm_grouped.to_csv(
          output_csv_file_path,
          index=False,
          encoding='utf-8'
      )
  
      return df_rttm_grouped

  
  

  • 설명

    • 오디오 파일에서 누가 언제 말했는지 구분
    • 결과를 RTTM + CSV 파일로 저장
    • DataFrame 형태로 결과 반환

  
  

  • df_rttm
    

  
  

  • df_rttm_grouped
    
  stt_to_rttm

  • 코드

  def stt_to_rttm(
          audio_file_path: str,
          stt_output_file_path: str,
          rttm_file_path: str,
          rttm_csv_file_path: str,
          final_output_csv_file_path: str
      ):
  
      result, df_stt = whisper_stt(
          audio_file_path,
          stt_output_file_path
      )
  
      df_rttm = speaker_diarization(
          audio_file_path,
          rttm_file_path,
          rttm_csv_file_path
      )
  
      # df_stt와 df_rttm을 시간 기준으로 매칭해서,각 화자에게 어떤 텍스트가 해당되는지를 자동으로 배정
      df_rttm["text"] = ""
  
      for i_stt, row_stt in df_stt.iterrows():
          overlap_dict = {}
          for i_rttm, row_rttm in df_rttm.iterrows():
              overlap = max(0, min(row_stt["end"], row_rttm["end"]) - max(row_stt["start"], row_rttm["start"])) # row_rttm에 대해 row_stt과 시간이 얼마나 겹치는지 계산
              overlap_dict[i_rttm] = overlap
  
          # 가장 겹친 정도가 큰 화자 구간을 선택
          max_overlap = max(overlap_dict.values())
          max_overlap_idx = max(overlap_dict, key=overlap_dict.get)
  
          # 화자 구간에 텍스트 추가
          if max_overlap > 0:
              df_rttm.at[max_overlap_idx, "text"] += row_stt["text"] + "\n"
  
      # 최종 결과 파일 저장
      df_rttm.to_csv(
          final_output_csv_file_path,
          index=False,
          sep='|',
          encoding='utf-8'
      )
  
      return df_rttm

  
  

  • 설명

    • STT + Diarization을 결합하여 화자별 대화 텍스트를 만듦
    • CSV 파일 저장

  
  

  • df_rttm
    

Meeting Transcript

회의 음성의 STT 및 화자 분리 결과를 기반으로 GPT를 활용해 회의 내용을 요약하고 발화를 교정한 뒤, 다양한 형식(JSON, Markdown, Word)으로 자동 저장하는 회의록 생성 자동화 프로그램

---

패키지 설치

pip install python-docx

---

Meeting Transcript 코드 작성

import pandas as pd
from openai import OpenAI
from dotenv import load_dotenv
import os
import json
from docx import Document

def load_openai_client(env_path):

    load_dotenv(dotenv_path=env_path)
    api_key = os.getenv('OPENAI_API_KEY')

    return OpenAI(api_key=api_key)

def load_meeting_note(csv_path):

    df = pd.read_csv(csv_path, sep='|')

    name_dict = {
        "SPEAKER_00": "AI",
        "SPEAKER_01": "이성용",
    }
    df["name"] = df["speaker_id"].apply(lambda x: name_dict[x])
    df.dropna(inplace=True)

    return df

# OpenAI API를 사용해 요약 결과 생성
def generate_summary(client, df):

    meeting_note_txt = df[['start', 'end', 'name', 'text']].to_json(orient='records', force_ascii=False)

    system_prompt = f'''
    너는 회의 내용을 요약하는 봇이다. 아래 회의록을 읽고, 주요 내용을 요약하라.
    결과는 마크다운 형식으로 작성한다.

    아래 형식에 맞추어 작성하라.

    # 회의 제목

    ## 주요 내용

    ## 참석자별 입장

    ## 결정 사항

    =============== 이하 회의록 ===============

    { meeting_note_txt }
    '''

    response = client.chat.completions.create(
        model='gpt-4o',
        temperature=0.1,
        messages=[
            {"role": "system", "content": system_prompt},
        ]
    )

    summary = response.choices[0].message.content
    summary = summary.strip()

    return summary, meeting_note_txt

# GPT로 녹취록 교정하기
def correct_transcripts(client, df, summary, meeting_note_txt):
    meeting_note_dict = df.to_dict(orient='records')

    for row in meeting_note_dict:
        if not isinstance(row['text'], str): # 텍스트가 아닌 다른 형식인 경우 통과
            continue

        correction_system_prompt = f'''
            너는 주어진 회의 녹취록을 수정 및 보완하는 봇이다.
            주어진 회의 녹취록은 STT로 작성된 결과이므로, 이 중에 오류가 있는 부분을 찾아내고 수정하라.

            표현, 말투는 최대한 원본과 일치하도록 유지하되, 잘못 받아적은 내용을 수정하라.
            원본의 내용을 최대한 살리되, 잘못된 내용만 수정하라.
            어떤 경우에는 A라는 사람이 말한 내용을 B가 말한 것으로 잘못 기록된 경우도 있을 수 있다. 이런 경우에는 문맥을 고려하여 수정하라.

            원본 내용 중 빠지거나, 추가되거나, 잘못 기록된 부분을 찾아내어 수정하라. 수정된 메시지 이외에는 아무것도 작성하지 말라.
            --------------------------------
            회의 요약문 : {summary}
            --------------------------------
            회의 녹취록 전문: {meeting_note_txt}
            --------------------------------
            확인할 메시지 원본: {row['text']}
        '''

        response = client.chat.completions.create(
            model='gpt-4o',
            temperature=0.1,
            messages=[
                {"role": "system", "content": correction_system_prompt},
            ]
        )

        correction = response.choices[0].message.content
        correction = correction.strip()

        row['corrected_text'] = correction

    return meeting_note_dict

# JSON 저장
def save_json(meeting_note_dict, path):
    with open(path, 'w', encoding='utf-8') as f:
        json.dump(meeting_note_dict, f, ensure_ascii=False, indent=4)

# Markdown 저장
def save_markdown(meeting_note_dict, path):
    md = ""

    for row in meeting_note_dict:
        md += f"- **{row['name']}** : {row['corrected_text']}\n"

    with open(path, 'w', encoding='utf-8') as f:
        f.write(md)

# Word(docx) 저장
def save_docx(meeting_note_dict, path):
    document = Document()
    document.add_heading('회의록', level=1)

    table = document.add_table(rows=1, cols=2)
    table.style = 'Table Grid'

    hdr_cells = table.rows[0].cells
    hdr_cells[0].text = 'Speaker'
    hdr_cells[1].text = 'Content'

    for row in meeting_note_dict:
        row_cells = table.add_row().cells
        row_cells[0].text = row['name']
        row_cells[1].text = row['corrected_text']

    document.save(path)

# 결과 저장
def save_outputs(meeting_note_dict, json_path, md_path, docx_path):
    save_json(meeting_note_dict, json_path)
    save_markdown(meeting_note_dict, md_path)
    save_docx(meeting_note_dict, docx_path)

if __name__ == "__main__":
    client = load_openai_client("api_key.env")
    df = load_meeting_note("./data/final.csv")

    summary, meeting_note_txt = generate_summary(client, df)
    meeting_note_dict = correct_transcripts(client, df, summary, meeting_note_txt)

    save_outputs(
        meeting_note_dict,
        json_path='./data/meeting_note.json',
        md_path='./data/meeting_note.md',
        docx_path='./data/meeting_note.docx'
    )

---

결과

meeting_note.docx

meeting_note.json

meeting_note.md

출처: 이성용, 「Do it! LLM을 활용한 AI 에이전트 개발 입문 - GPT API+딥시크+라마+랭체인+랭그래프+RAG」, 이지스퍼블리싱