"""Colab 실전 테스트 노트북 생성 스크립트. 다른 환경(CPU-only, ffmpeg 없음)에서 검증할 수 없었던 부분을 Colab Pro(T4 GPU + 터미널)에서 실전 검증하기 위한 노트북을 생성한다: 1. 저장소 클론 (feat/full-platform) 2. 시스템 의존성 (ffmpeg) + venv + luke-scribe 설치 3. `detect` — T4 GPU 실제 감지 (능력 등급/정밀도/워커수) 4. 단위/통합 테스트 (127개 mock) 5. 샘플 오디오 생성 (한국어+영문 기술용어, edge-tts) 6. 실전 전사 (GPU, faster-whisper 모델 다운로드) + glossary/hotword 검증 7. API 서버 기동 + curl 스모크 (업로드 → poll → 결과) 8. 벤치마크 (turbo vs large-v3) 사용법: python scripts/build_colab_notebook.py → notebooks/luke-scribe-colab.ipynb """ from __future__ import annotations import json from pathlib import Path OUT = Path("notebooks/luke-scribe-colab.ipynb") def md(source: str) -> dict: return {"cell_type": "markdown", "metadata": {}, "source": source} def code(source: str) -> dict: return {"cell_type": "code", "execution_count": None, "metadata": {}, "outputs": [], "source": source} def cells() -> list[dict]: return [ md( "# luke_scribe — Colab 실전 테스트 노트북\n" "\n" "> 내부용 로컬 STT 전사 API (faster-whisper, hardware-adaptive, privacy-first)\n" "\n" "이 노트북은 CPU-only 개발 환경에서 **mock으로만 검증**했던 것을,\n" "Colab Pro(GPU + 터미널)에서 **실전 검증**하기 위한 것입니다.\n" "\n" "## 검증 대상\n" "\n" "- `luke-scribe detect` → T4 GPU 실제 감지 (능력 등급 T1~T3, 정밀도, 워커수)\n" "- faster-whisper 모델 다운로드 + **실제 한국어 전사** (CPU-only 환경에선 불가)\n" "- glossary/hotword 후처리 (KO+EN 기술용어 보존)\n" "- REST API 흐름 (업로드 → poll → 결과)\n" "- turbo vs large-v3 벤치마크\n" "\n" "## 준비\n" "\n" "1. 런타임 → 런타임 유형 변경 → **T4 GPU** 선택\n" "2. (Colab Pro) 터미널을 사용해도 동일한 명령을 실행할 수 있습니다\n" "3. 저장소가 private이면 아래 셀에 **Gitea 토큰** 입력 (Settings → Applications → Generate New Token)\n" "\n" "---\n" ), code( "# 0) 런타임 확인 — T4 GPU가 활성 상태여야 합니다\n" "!nvidia-smi\n" "import sys\n" "print('Python', sys.version.split()[0])\n" ), md( "## 1) 저장소 클론\n" "\n" "`feat/full-platform` 브랜치를 클론합니다.\n" "저장소가 private이면 아래 셀의 `GITEA_TOKEN`에 토큰을 입력하세요." ), code( "# 1) 클론 또는 업데이트 (private 저장소면 GITEA_TOKEN 입력)\n" "GITEA_TOKEN = '' # ← 필요 시 입력: https://git.lukehemmin.com/user/settings/applications\n" "\n" "if GITEA_TOKEN:\n" " REPO = f'https://{GITEA_TOKEN}@git.lukehemmin.com/lukehemmin/luke_scribe.git'\n" "else:\n" " REPO = 'https://git.lukehemmin.com/lukehemmin/luke_scribe.git'\n" "\n" "import os, subprocess\n" "\n" "if os.path.isdir('/content/luke_scribe/.git'):\n" " # 이미 클론된 레포 → 최신 브랜치로 갱신 (pull)\n" " print('기존 레포 감지 → pull로 갱신')\n" " subprocess.run(['git', 'fetch', 'origin'], cwd='/content/luke_scribe', check=True)\n" " subprocess.run(['git', 'checkout', 'feat/full-platform'], cwd='/content/luke_scribe', check=True)\n" " subprocess.run(['git', 'pull', '--ff-only', 'origin', 'feat/full-platform'], cwd='/content/luke_scribe', check=True)\n" "else:\n" " # 최초 실행 → 클론\n" " subprocess.run(['git', 'clone', '-b', 'feat/full-platform', REPO], cwd='/content', check=True)\n" "\n" "os.chdir('/content/luke_scribe')\n" "print('작업 디렉터리 →', os.getcwd())\n" "!git log --oneline -1\n" ), md("## 2) 시스템 의존성 + 설치\n\nffmpeg(오디오 정규화) 설치 후 venv에 luke-scribe를 설치합니다."), code( "# 2) 시스템 패키지 (ffmpeg — ffprobe/정규화에 필수)\n" "!apt-get update -qq && apt-get install -y -qq ffmpeg >/dev/null\n" "!ffmpeg -version 2>&1 | head -1\n" "!ffprobe -version 2>&1 | head -1\n" ), code( "# 3) 설치 — Colab은 시스템 pip가 표준 (venv는 Colab에서 ensurepip 오류로 실패할 수 있음)\n" "!pip install -q --upgrade pip\n" "!pip install -q -e '.[engine,api]'\n" "!pip install -q edge-tts # 샘플 음성 생성용\n" "print('설치 완료')\n" "!which luke-scribe && luke-scribe --help 2>&1 | head -8\n" ), md( "## 3) 하드웨어 감지 — T4 GPU 실제 확인\n" "\n" "CPU-only 환경에서는 `T0 / cpu / int8`이 나왔지만,\n" "Colab T4(16GB VRAM, sm_75)에서는 GPU가 감지되어야 합니다." ), code( "# 4) detect — GPU 감지 / 능력 등급 / 정밀도 / 워커수\n" "!luke-scribe detect\n" ), md("## 4) 단위/통합 테스트 (127개)\n\nmock 기반 테스트가 GPU 환경에서도 전부 통과하는지 확인합니다."), code( "# 5) 테스트 스위트\n" "!python -m pytest tests/ -q 2>&1 | tail -5\n" ), md( "## 5) 실전 전사 (GPU)\n\n" "### 5-1) 샘플 오디오 생성\n\n" "edge-tts(MS TTS)로 **한국어 + 영문 기술용어가 섞인** 샘플을 생성합니다.\n" "이 텍스트에는 `vLLM`, `Kubernetes`, `GPU` 같은 용어가 포함되어 glossary/hotword 검증에 적합합니다." ), code( "# 6) 샘플 오디오 생성 (한국어 + 기술용어, 무료 TTS)\n" "!mkdir -p samples\n" "!edge-tts --voice ko-KR-SunHiNeural --text '오늘은 vLLM 서버를 Kubernetes 클러스터에 배포하는 방법을 설명합니다. GPU 가속 추론으로 대기 시간을 줄일 수 있습니다.' --write-media samples/colab-ko-en.mp3\n" "!ffprobe -v error -show_entries format=duration -of json samples/colab-ko-en.mp3\n" ), code( "# 7) 실전 전사 — GPU 자동 감지 + 모델 다운로드 (large-v3-turbo)\n" "# 첫 실행 시 Hugging Face에서 모델을 다운로드합니다 (turbo ≈ 1.6GB, 1~2분)\n" "!time luke-scribe transcribe samples/colab-ko-en.mp3 --language ko --device auto\n" ), md( "### 5-2) 후처리 검증\n\n" "glossary(오인식 용어 복원) + hotword(용어 사전 주입) 동작을 확인합니다.\n" "`--hotword vLLM Kubernetes`를 주면 initial_prompt에 용어가 주입되어 보존률이 올라갑니다." ), code( "# 8) hotword 포함 전사 (용어 보존 강화)\n" "!luke-scribe transcribe samples/colab-ko-en.mp3 --language ko --device auto --hotword vLLM --hotword Kubernetes\n" ), md( "## 6) REST API 스모크\n\n" "서버를 백그라운드로 띄우고 **업로드 → poll → 결과(SRT)** 흐름을 검증합니다." ), code( "# 9) API 키 생성 (1회 출력)\n" "!luke-scribe key --create --scopes transcribe,admin --file /content/api_keys.json\n" ), code( "# 10) 서버 기동 (in-proc 큐, 백그라운드 — Colab에서는 %%bash --bg 또는 nohup 사용)\n" "!mkdir -p /content/logs\n" "!nohup luke-scribe serve --port 8000 > /content/logs/server.log 2>&1 &\n" "import time, urllib.request\n" "ok = False\n" "for _ in range(40):\n" " try:\n" " urllib.request.urlopen('http://localhost:8000/health', timeout=1)\n" " ok = True\n" " break\n" " except Exception:\n" " time.sleep(1)\n" "print('서버 기동 OK' if ok else '서버 기동 실패 — 로그:')\n" "if not ok:\n" " print(open('/content/logs/server.log').read()[-1500:])\n" ), code( "# 11) 업로드 → poll → 결과\n" "import json, time, urllib.request, urllib.error\n" "\n" "KEY = json.load(open('/content/api_keys.json'))['keys'][0]\n" "\n" "# multipart 업로드 (curl 사용 — 간단)\n" "import subprocess\n" "r = subprocess.run(\n" " ['curl', '-s', '-X', 'POST', 'http://localhost:8000/v1/jobs',\n" " '-H', f'X-API-Key: {KEY}',\n" " '-F', 'file=@samples/colab-ko-en.mp3',\n" " '-F', 'options={\"language\":\"ko\",\"formats\":[\"json\",\"srt\"]}'],\n" " capture_output=True, text=True,\n" ")\n" "job = json.loads(r.stdout)\n" "print('생성:', job)\n" "job_id = job['job_id']\n" "\n" "# 완료까지 poll (워커 없이 in-proc이라 즉시 완료되지는 않음 → 여기선 생성까지만 확인)\n" "print('job_id =', job_id)\n" ), md( "### 참고 — 서버에서 실제 전사까지 실행하려면\n\n" "in-proc 브로커는 큐만 받고 워커가 별도로 소비해야 합니다. 워커 스레드를 띄우거나\n" "간단하게는 배치 파이프라인을 직접 호출하는 CLI가 더 간단합니다. API 전체 흐름(워커 포함)은\n" "다음 셀에서 `python`으로 브로커 + 워커를 함께 돌려 확인합니다." ), code( "# 12) 브로커 + 워커 포함 전체 흐름 (in-proc)\n" "# 서버의 in-proc 브로커에 enqueue된 job을 같은 프로세스의 워커가 소비하는 구조는\n" "# 프로세스 분리 필요 → 여기서는 클라이언트에서 직접 Worker.drain() 호출로 검증\n" "from luke_scribe.config import Settings\n" "from luke_scribe.jobqueue.broker import InProcBroker\n" "from luke_scribe.jobqueue.worker import Worker\n" "from luke_scribe.jobqueue.jobs import Job\n" "from luke_scribe.results.store import ResultStore\n" "\n" "settings = Settings(_env_file=None, queue_backend='inproc',\n" " results_dir='/content/results', model_cache_dir=None)\n" "broker = InProcBroker(settings)\n" "store = ResultStore(settings.results_dir)\n" "\n" "job = Job(type='file', lane='batch', source_path='samples/colab-ko-en.mp3',\n" " options={'model': 'large-v3-turbo', 'language': 'ko', 'device': 'auto'})\n" "broker.enqueue(job)\n" "worker = Worker(settings=settings, broker=broker, store=store)\n" "worker.drain()\n" "\n" "result = store.read_result(job.id)\n" "print('status:', result.status)\n" "print('text:', result.text[:120])\n" "print('device:', result.execution.device, '| ct:', result.execution.compute_type, '| rtf:', result.timings.rtf)\n" ), md( "## 7) 벤치마크 (turbo vs large-v3)\n\n" "manifest에 모델 비교 항목이 있으면 실행합니다. 두 모델을 모두 다운로드하므로\n" "시간이 걸립니다 (turbo ~1.6GB + large-v3 ~3GB)." ), code( "# 13) 벤치마크 — 샘플 manifest 사용 (선택, 시간 소요)\n" "# 간단 manifest 생성\n" "import yaml\n" "manifest = {\n" " 'name': 'colab-quick',\n" " 'language': 'ko',\n" " 'targets': {'entity_preservation': 0.95, 'cer': 0.15},\n" " 'cases': [\n" " {'name': 'ko-en-tech', 'audio': 'samples/colab-ko-en.mp3',\n" " 'expected_entities': ['vLLM', 'Kubernetes', 'GPU']},\n" " ],\n" "}\n" "yaml.safe_dump(manifest, open('/content/manifest.yaml', 'w'))\n" "\n" "# 실행 (기본: turbo만 → 빠름)\n" "!luke-scribe bench /content/manifest.yaml --models large-v3-turbo --device auto --repeats 2 --output /content/bench-report.json 2>&1 | tail -20\n" ), md( "## 8) 문제 해결\n\n" "### CUDA 라이브러리 오류 (faster-whisper/CTranslate2)\n\n" "Colab T4에는 CUDA 런타임이 있지만, CTranslate2가 pip wheel의 CUDA 라이브러리를\n" "찾지 못하면 아래를 실행하세요:\n" "\n" "```python\n" "import os\n" "os.environ['LD_LIBRARY_PATH'] = (\n" " '/usr/local/cuda/lib64:' + os.environ.get('LD_LIBRARY_PATH', '')\n" ")\n" "```\n" "\n" "### 모델 다운로드 실패\n\n" "- Hugging Face 연결 필요. 재시도: `LUKESCRIBE_MODEL_DOWNLOAD_RETRIES=3`\n" "- 특정 모델만: `--model large-v3-turbo` (기본) / `--model large-v3`\n" "\n" "### OOM (16GB VRAM)\n\n" "- EngineOwner가 자동으로 정밀도를 강등합니다 (float16 → int8_float16 → int8 → CPU)\n" "- 수동 지정: `--compute-type int8_float16`\n" "\n" "### 저장소가 private인데 클론 안 됨\n\n" "1번 셀의 `GITEA_TOKEN`에 토큰을 입력하고 런타임 → **Restart session** 후 다시 실행하세요.\n" ), ] def main() -> None: nb = { "nbformat": 4, "nbformat_minor": 0, "metadata": { "colab": {"provenance": [], "gpuType": "T4"}, "kernelspec": { "display_name": "Python 3", "language": "python", "name": "python3", }, "language_info": {"name": "python", "version": "3.11"}, "accelerator": "GPU", }, "cells": cells(), } OUT.parent.mkdir(parents=True, exist_ok=True) OUT.write_text(json.dumps(nb, ensure_ascii=False, indent=1), encoding="utf-8") n_code = sum(1 for c in nb["cells"] if c["cell_type"] == "code") print(f"✅ {OUT} 생성 (셀 {len(nb['cells'])}개, 코드 {n_code}개)") if __name__ == "__main__": main()