Meeting minutes pipeline.py

#!/usr/bin/env python3
"""
회의록 자동 생성 파이프라인 / Meeting minutes pipeline
=====================================================
녹음 파일 1개 → 화자별 트랜스크립트 → LLM 요약(로컬 & Claude 동시 비교) → Markdown

Pipeline:
  ① STT + 화자분리 + 정렬  : WhisperX (Whisper large-v3 + pyannote 3.1)
  ② 화자 실명 매핑          : SPEAKER_00 → 실명 dict
  ③ Markdown 트랜스크립트    : 화자별 발화 묶기
  ④ LLM 요약 (2종 비교)      : 로컬 vLLM/SGLang + Claude API

설치 / install:
    pip install whisperx anthropic openai
    # pyannote 모델 사용 동의 (HuggingFace에서 1회):
    #   huggingface.co/pyannote/speaker-diarization-3.1
    #   huggingface.co/pyannote/segmentation-3.0

실행 / run:
    export HF_TOKEN=hf_xxxxx
    export ANTHROPIC_API_KEY=sk-ant-xxxxx      # Claude 요약 쓸 때만
    export LOCAL_LLM_BASE_URL=http://localhost:8000/v1   # vLLM/SGLang 엔드포인트
    python meeting_minutes_pipeline.py meeting.m4a
"""

import os
import gc
import sys
import argparse

import torch


# =====================================================================
# 설정 / CONFIG  (여기만 상황에 맞게 수정) / edit here
# =====================================================================
class Config:
    # --- STT / diarization ---
    device = "cuda"                 # RTX PRO 2000 Blackwell Laptop (8GB VRAM)
    compute_type = "int8"           # 8GB GPU: int8 필수 / int8 required for 8GB
    batch_size = 4                  # 8GB GPU: 4로 축소 / reduced for 8GB
    whisper_model = "large-v3"      # 한국어 권장 / recommended for Korean
    language = "ko"

    # 화자 수 / speaker count — 4명 이상은 정확히 아는 게 정확도에 크게 유리
    # 정확히 알면 num_speakers 지정, 모르면 None + min/max 사용
    num_speakers = None             # 예: 5 (정확히 알면) / exact if known
    min_speakers = 4
    max_speakers = 8

    hf_token = os.environ.get("HF_TOKEN")

    # --- 화자 실명 매핑 / speaker name mapping ---
    # 1차 결과의 앞부분을 듣고 SPEAKER_XX ↔ 실명을 채워 재실행
    # First run produces SPEAKER_00~; listen to the head, then fill & rerun
    speaker_names = {
        "SPEAKER_00": "SPEAKER_00",
        "SPEAKER_01": "SPEAKER_01",
        "SPEAKER_02": "SPEAKER_02",
        "SPEAKER_03": "SPEAKER_03",
        # "SPEAKER_00": "김철수 (PM)",
        # "SPEAKER_01": "이영희 (개발)",
        # "SPEAKER_02": "박민수 (파트너사)",
    }

    # --- LLM 요약 / summarization ---
    run_local_llm = True            # 로컬 vLLM/SGLang 요약 실행
    run_claude = True               # Claude API 요약 실행

    local_llm_base_url = os.environ.get("LOCAL_LLM_BASE_URL",
                                        "http://localhost:8000/v1")
    local_llm_model = "Qwen/Qwen2.5-72B-Instruct"   # vLLM에 로드된 모델명
    claude_model = "claude-opus-4-8"


# =====================================================================
# ① STT + 화자분리 + 정렬 / transcribe + diarize + align
# =====================================================================
def transcribe_and_diarize(audio_file: str, cfg: Config):
    import whisperx

    # --- STT ---
    print("[1/3] STT (Whisper) ...")
    model = whisperx.load_model(
        cfg.whisper_model, cfg.device,
        compute_type=cfg.compute_type, language=cfg.language,
    )
    audio = whisperx.load_audio(audio_file)
    result = model.transcribe(audio, batch_size=cfg.batch_size)
    del model
    gc.collect()
    torch.cuda.empty_cache()

    # --- 단어 단위 정렬 / word-level alignment ---
    # 한국어는 kresnik/wav2vec2-large-xlsr-korean 자동 사용
    print("[2/3] 정렬 / alignment ...")
    try:
        model_a, metadata = whisperx.load_align_model(
            language_code=cfg.language, device=cfg.device)
        result = whisperx.align(
            result["segments"], model_a, metadata, audio, cfg.device)
        del model_a
        gc.collect()
        torch.cuda.empty_cache()
    except Exception as e:
        # 정렬이 튀면 segment 단위로 진행 / fall back to segment-level
        print(f"  ! 정렬 스킵 (segment 단위 진행): {e}")

    # --- 화자 분리 / diarization ---
    print("[3/3] 화자 분리 / diarization ...")
    diarize_model = whisperx.DiarizationPipeline(
        use_auth_token=cfg.hf_token, device=cfg.device)

    if cfg.num_speakers:            # 화자 수를 알면 지정 = 정확도 상승
        diarize_segments = diarize_model(audio, num_speakers=cfg.num_speakers)
    else:
        diarize_segments = diarize_model(
            audio, min_speakers=cfg.min_speakers, max_speakers=cfg.max_speakers)

    result = whisperx.assign_word_speakers(diarize_segments, result)
    return result


# =====================================================================
# ②③ Markdown 트랜스크립트 / build markdown transcript
# =====================================================================
def format_timestamp(seconds: float) -> str:
    """초 → MM:SS / seconds to MM:SS"""
    m, s = divmod(int(seconds), 60)
    return f"{m:02d}:{s:02d}"


def build_transcript(result, cfg: Config) -> str:
    names = cfg.speaker_names
    lines, current, buffer, start_ts = [], None, [], None

    for seg in result["segments"]:
        spk_id = seg.get("speaker", "UNKNOWN")
        spk = names.get(spk_id, spk_id)
        text = seg["text"].strip()
        if not text:
            continue
        # 같은 화자 연속 발화 묶기 / group consecutive turns
        if spk != current:
            if buffer:
                lines.append(f"**[{start_ts}] {current}**: {' '.join(buffer)}\n")
            current, buffer = spk, [text]
            start_ts = format_timestamp(seg["start"])
        else:
            buffer.append(text)

    if buffer:
        lines.append(f"**[{start_ts}] {current}**: {' '.join(buffer)}\n")

    return "\n".join(lines)


# =====================================================================
# ④ LLM 요약 / summarization
# =====================================================================
SUMMARY_PROMPT = """다음은 회의 녹취록입니다. 아래 형식의 Markdown 회의록으로 정리해줘.

## 회의 요약
(핵심을 3~5줄로)

## 주요 논의 사항
- (항목별 bullet)

## 결정 사항
- (확정된 것만)

## 액션 아이템
| 담당자 | 할 일 | 기한 |
|---|---|---|

## 미해결 / 후속 논의
- (있으면)

---
녹취록:
{transcript}
"""


def summarize_local(transcript: str, cfg: Config) -> str:
    """로컬 vLLM/SGLang (OpenAI 호환 API) / local OpenAI-compatible endpoint"""
    from openai import OpenAI
    client = OpenAI(base_url=cfg.local_llm_base_url, api_key="not-needed")
    resp = client.chat.completions.create(
        model=cfg.local_llm_model,
        messages=[{"role": "user",
                   "content": SUMMARY_PROMPT.format(transcript=transcript)}],
        temperature=0.3,
        max_tokens=4000,
    )
    return resp.choices[0].message.content


def summarize_claude(transcript: str, cfg: Config) -> str:
    """Claude API / Anthropic SDK"""
    import anthropic
    client = anthropic.Anthropic()   # ANTHROPIC_API_KEY 환경변수 사용
    resp = client.messages.create(
        model=cfg.claude_model,
        max_tokens=4000,
        messages=[{"role": "user",
                   "content": SUMMARY_PROMPT.format(transcript=transcript)}],
    )
    return resp.content[0].text


# =====================================================================
# main
# =====================================================================
def main():
    parser = argparse.ArgumentParser(
        description="회의록 자동 생성 파이프라인",
        formatter_class=argparse.ArgumentDefaultsHelpFormatter)
    parser.add_argument("audio", help="녹음 파일 경로 / audio file path")
    parser.add_argument("--out", default="meeting", help="출력 파일 prefix")

    # --- STT / diarization 튜닝 (OOM 시 조절) ---
    parser.add_argument("--device", default=Config.device,
                        choices=["cuda", "cpu"], help="연산 장치 / device")
    parser.add_argument("--compute-type", default=Config.compute_type,
                        choices=["float16", "float32", "int8", "int8_float16"],
                        help="양자화 정밀도. OOM 시 int8 / lower for OOM")
    parser.add_argument("--batch-size", type=int, default=Config.batch_size,
                        help="배치 크기. OOM 시 축소 / reduce for OOM")
    parser.add_argument("--whisper-model", default=Config.whisper_model,
                        help="Whisper 모델. OOM 시 medium / smaller for OOM")
    parser.add_argument("--language", default=Config.language, help="언어 코드")

    # --- 화자 수 / speaker count ---
    parser.add_argument("--num-speakers", type=int, default=Config.num_speakers,
                        help="화자 수(정확히 알 때). 지정 시 min/max 무시")
    parser.add_argument("--min-speakers", type=int, default=Config.min_speakers,
                        help="최소 화자 수")
    parser.add_argument("--max-speakers", type=int, default=Config.max_speakers,
                        help="최대 화자 수")

    # --- LLM 요약 on/off ---
    parser.add_argument("--no-local-llm", action="store_true",
                        help="로컬 vLLM/SGLang 요약 비활성화")
    parser.add_argument("--no-claude", action="store_true",
                        help="Claude API 요약 비활성화")
    args = parser.parse_args()

    # 명령행 값으로 Config 오버라이드 / CLI overrides Config
    cfg = Config()
    cfg.device = args.device
    cfg.compute_type = args.compute_type
    cfg.batch_size = args.batch_size
    cfg.whisper_model = args.whisper_model
    cfg.language = args.language
    cfg.num_speakers = args.num_speakers
    cfg.min_speakers = args.min_speakers
    cfg.max_speakers = args.max_speakers
    cfg.run_local_llm = Config.run_local_llm and not args.no_local_llm
    cfg.run_claude = Config.run_claude and not args.no_claude

    if not cfg.hf_token:
        print("! HF_TOKEN 환경변수가 필요합니다 (pyannote 모델용).")
        sys.exit(1)

    # ① STT + diarization
    result = transcribe_and_diarize(args.audio, cfg)

    # ②③ 트랜스크립트 / transcript
    transcript = build_transcript(result, cfg)
    transcript_path = f"{args.out}_transcript.md"
    with open(transcript_path, "w", encoding="utf-8") as f:
        f.write("# 회의 녹취록 (화자별)\n\n" + transcript)
    print(f"\n✓ 트랜스크립트 저장: {transcript_path}")

    # 화자 라벨이 아직 SPEAKER_XX면 여기서 멈추고 매핑 후 재요약 권장
    if all(v.startswith("SPEAKER_") for v in cfg.speaker_names.values()):
        print("\n※ 앞부분을 듣고 Config.speaker_names에 실명을 채운 뒤 "
              "재실행하면 회의록에 실명이 반영됩니다.")

    # ④ LLM 요약 (2종 비교) / dual summarization
    if cfg.run_local_llm:
        try:
            print("\n[요약] 로컬 vLLM/SGLang ...")
            local_md = summarize_local(transcript, cfg)
            path = f"{args.out}_summary_local.md"
            with open(path, "w", encoding="utf-8") as f:
                f.write(local_md)
            print(f"✓ 로컬 요약 저장: {path}")
        except Exception as e:
            print(f"! 로컬 요약 실패: {e}")

    if cfg.run_claude:
        try:
            print("\n[요약] Claude API ...")
            claude_md = summarize_claude(transcript, cfg)
            path = f"{args.out}_summary_claude.md"
            with open(path, "w", encoding="utf-8") as f:
                f.write(claude_md)
            print(f"✓ Claude 요약 저장: {path}")
        except Exception as e:
            print(f"! Claude 요약 실패: {e}")

    print("\n완료. 두 요약본을 나란히 비교해 보세요.")


if __name__ == "__main__":
    main()