STT & AI 요약 API 시스템 (YAP)

멀티미디어 처리 및 AI 기반 요약 API 서버

📋 프로젝트 개요

**YAP (Yet Another Processor)**는 영상/음성 파일을 텍스트로 변환하고 AI로 요약하는 범용 API 서버입니다. Flask 기반으로 구축되어 다양한 클라이언트 애플리케이션에 STT(Speech-to-Text), 번역, LLM 요약 서비스를 제공합니다.

시스템 역할

┌─────────────────────────────────────┐
│   클라이언트 애플리케이션            │
│   ├─ YouTube 자동화 시스템 ⭐       │
│   ├─ 강의 녹화 시스템               │
│   └─ 회의록 자동화 도구             │
└─────────────────────────────────────┘
            ↓ (REST API)
┌─────────────────────────────────────┐
│   YAP API 서버 (본 시스템)          │
│   ├─ STT API (Whisper)              │
│   ├─ 번역 API (Google, DeepL)       │
│   └─ LLM 요약 (GPT-4, Claude)       │
└─────────────────────────────────────┘

핵심 기능

멀티미디어 처리:

영상 포맷: MP4, AVI, MKV, MOV, WMV
음성 포맷: MP3, WAV, M4A, FLAC, OGG
텍스트 직접 업로드: TXT, SRT, VTT

AI 파이프라인:

음성 추출: FFmpeg로 영상 → 음성
STT 변환: Whisper (로컬) 또는 Google STT
번역: Google Translation, DeepL (선택)
요약: GPT-4, Claude, Gemini (선택)

YouTube 통합:

yt-dlp로 YouTube 영상 자동 다운로드
메타데이터 자동 추출
자막 파일 생성 (SRT, VTT)

🎯 해결한 문제

API 서버의 필요성

기존: 각 프로젝트마다 STT/LLM 코드 중복

YouTube 자동화 → Whisper 직접 호출
강의 녹화 → Whisper 직접 호출
회의록 → Whisper 직접 호출

YAP 도입 후: 중앙화된 API 서버

모든 프로젝트 → YAP API → Whisper/LLM

해결한 기술적 과제

비동기 처리
- 긴 영상(2시간+) 처리 시간 문제
- Celery로 백그라운드 작업 큐 처리
- 실시간 진행률 업데이트 (Socket.IO)
대용량 파일 처리
- 2GB+ 파일 업로드
- 청크 업로드 + 재개 가능 업로드
- 디스크 공간 자동 정리
Multi-LLM 지원
- GPT-4, Claude, Gemini 통합
- 토큰 제한 대응 (청크 분할)
- API 비용 최적화
안정성
- 실패 시 자동 재시도
- 부분 실패 허용 (청크별 처리)
- 에러 로깅 및 알림

🛠 기술 스택

Backend

Framework: Flask 2.3+
Async Queue: Celery + Redis
Database: PostgreSQL (작업 메타데이터)
Cache: Redis (세션, 임시 데이터)

AI/ML

STT:
- OpenAI Whisper (로컬, large-v3)
- Google Cloud Speech-to-Text (API)
- Azure Speech Service (옵션)
LLM:
- OpenAI GPT-4 / GPT-4-turbo
- Anthropic Claude 3 (Opus, Sonnet)
- Google Gemini Pro
Translation:
- Google Cloud Translation
- DeepL API

Media Processing

FFmpeg: 영상/음성 변환
yt-dlp: YouTube 다운로드
moviepy: 영상 편집 (옵션)
pydub: 음성 처리

Infrastructure

Storage: S3 / MinIO (self-hosted)
Monitoring: Prometheus + Grafana
Logging: ELK Stack (Elasticsearch, Logstash, Kibana)

💡 주요 구현 내용

1. RESTful API 설계

파일 업로드 API

# app/routes/upload.py
@app.route('/api/upload', methods=['POST'])
def upload_files():
    """
    멀티파트 파일 업로드
    
    Request:
        - files: File[] (최대 10개)
        - language: str (ko, en, ja, zh, es)
        - translate: bool (옵션)
        - target_language: str (옵션)
        - summarize: bool (옵션)
        - prompt: str (커스텀 프롬프트)
    
    Response:
        {
            "task_id": "uuid",
            "estimated_time": 420,
            "files": [...]
        }
    """
    files = request.files.getlist('files')
    options = {
        'language': request.form.get('language', 'ko'),
        'translate': request.form.get('translate', 'false') == 'true',
        'target_language': request.form.get('target_language'),
        'summarize': request.form.get('summarize', 'false') == 'true',
        'prompt': request.form.get('prompt')
    }
    
    # 파일 검증
    for file in files:
        if not allowed_file(file.filename):
            return jsonify({
                'error': f'지원하지 않는 파일: {file.filename}'
            }), 400
    
    # 파일 저장
    file_paths = []
    for file in files:
        path = save_upload_file(file)
        file_paths.append(path)
    
    # Celery 작업 시작
    task = process_media_task.delay(
        file_paths=file_paths,
        options=options
    )
    
    return jsonify({
        'status': 'success',
        'task_id': task.id,
        'estimated_time': estimate_processing_time(file_paths)
    })

YouTube 처리 API

# app/routes/youtube.py
@app.route('/api/youtube', methods=['POST'])
def process_youtube():
    """
    YouTube URL 처리
    
    Request:
        {
            "url": "https://youtube.com/watch?v=...",
            "language": "ko",
            "options": {
                "translate": true,
                "target_language": "en",
                "summarize": true
            }
        }
    
    Response:
        {
            "task_id": "uuid",
            "video_info": {...}
        }
    """
    data = request.json
    url = data.get('url')
    
    # YouTube 메타데이터 추출 (빠른 응답)
    video_info = extract_youtube_info(url)
    
    # 백그라운드 다운로드 + 처리
    task = process_youtube_task.delay(
        url=url,
        language=data.get('language', 'ko'),
        options=data.get('options', {})
    )
    
    return jsonify({
        'task_id': task.id,
        'video_info': video_info
    })

작업 상태 조회 API

# app/routes/task.py
@app.route('/api/task/<task_id>')
def get_task_status(task_id):
    """
    작업 진행 상황 조회
    
    Response (처리 중):
        {
            "status": "processing",
            "progress": 65,
            "current_step": "stt_conversion",
            "steps": {...}
        }
    
    Response (완료):
        {
            "status": "completed",
            "result": {
                "transcript": {...},
                "translation": {...},
                "summary": {...},
                "download_urls": {...}
            }
        }
    """
    task = AsyncResult(task_id)
    
    if task.state == 'PENDING':
        return jsonify({
            'status': 'pending',
            'message': '작업 대기 중'
        })
    
    elif task.state == 'PROGRESS':
        return jsonify({
            'status': 'processing',
            'progress': task.info.get('progress', 0),
            'current_step': task.info.get('current_step'),
            'message': task.info.get('message')
        })
    
    elif task.state == 'SUCCESS':
        result = task.result
        return jsonify({
            'status': 'completed',
            'result': result
        })
    
    else:  # FAILURE
        return jsonify({
            'status': 'failed',
            'error': str(task.info)
        }), 500

2. Celery 비동기 작업 처리

메인 처리 태스크

# app/tasks/process_task.py
from celery import Task
 
class ProcessMediaTask(Task):
    """
    미디어 처리 Base Task
    """
    def on_failure(self, exc, task_id, args, kwargs, einfo):
        """실패 시 처리"""
        logger.error(f'Task {task_id} failed: {exc}')
        # DB 상태 업데이트
        update_task_status(task_id, 'failed', error=str(exc))
 
@celery_app.task(
    bind=True,
    base=ProcessMediaTask,
    max_retries=3,
    time_limit=7200  # 2시간
)
def process_media_task(self, file_paths, options):
    """
    멀티미디어 처리 메인 태스크
    
    단계:
        1. [0-30%] 음성 추출
        2. [30-60%] STT 변환
        3. [60-80%] 번역 (옵션)
        4. [80-95%] 요약 (옵션)
        5. [95-100%] 결과 저장
    """
    task_id = self.request.id
    
    try:
        # 1단계: 음성 추출 (0-30%)
        self.update_state(
            state='PROGRESS',
            meta={
                'progress': 10,
                'current_step': 'audio_extraction',
                'message': '음성을 추출하고 있습니다...'
            }
        )
        
        audio_paths = []
        for file_path in file_paths:
            if is_video_file(file_path):
                audio_path = extract_audio(file_path)
            else:
                audio_path = file_path
            audio_paths.append(audio_path)
        
        # 2단계: STT 변환 (30-60%)
        self.update_state(
            state='PROGRESS',
            meta={
                'progress': 40,
                'current_step': 'stt_conversion',
                'message': '음성을 텍스트로 변환 중...'
            }
        )
        
        transcripts = []
        for audio_path in audio_paths:
            transcript = stt_service.transcribe(
                audio_path,
                language=options['language']
            )
            transcripts.append(transcript)
        
        # 전체 텍스트 통합
        full_text = "\n\n".join([t['text'] for t in transcripts])
        
        # 3단계: 번역 (60-80%)
        translation = None
        if options.get('translate'):
            self.update_state(
                state='PROGRESS',
                meta={
                    'progress': 70,
                    'current_step': 'translation',
                    'message': '번역 중...'
                }
            )
            
            translation = translation_service.translate(
                full_text,
                source=options['language'],
                target=options['target_language']
            )
        
        # 4단계: AI 요약 (80-95%)
        summary = None
        if options.get('summarize'):
            self.update_state(
                state='PROGRESS',
                meta={
                    'progress': 85,
                    'current_step': 'summarization',
                    'message': 'AI 요약 생성 중...'
                }
            )
            
            summary = llm_service.summarize(
                full_text,
                prompt=options.get('prompt')
            )
        
        # 5단계: 결과 저장 (95-100%)
        self.update_state(
            state='PROGRESS',
            meta={
                'progress': 95,
                'current_step': 'saving',
                'message': '결과를 저장하고 있습니다...'
            }
        )
        
        result_urls = save_results(
            task_id=task_id,
            transcripts=transcripts,
            translation=translation,
            summary=summary
        )
        
        return {
            'status': 'completed',
            'result': result_urls,
            'processing_time': time.time() - start_time
        }
        
    except Exception as exc:
        logger.error(f'Task failed: {exc}')
        self.retry(exc=exc, countdown=60)

3. Whisper STT 서비스

# app/services/stt_service.py
import whisper
 
class STTService:
    """
    Whisper 기반 STT 서비스
    """
    def __init__(self, model_name='large-v3', device='cuda'):
        self.model = whisper.load_model(
            model_name,
            device=device
        )
    
    def transcribe(self, audio_path, language='ko'):
        """
        음성 → 텍스트 변환
        
        Args:
            audio_path: 음성 파일 경로
            language: 언어 코드
        
        Returns:
            {
                "text": "전체 전사 텍스트",
                "segments": [
                    {
                        "id": 0,
                        "start": 0.0,
                        "end": 5.24,
                        "text": "안녕하세요.",
                        "confidence": 0.95
                    }
                ],
                "language": "ko",
                "duration": 3600.5
            }
        """
        # Whisper 실행
        result = self.model.transcribe(
            audio_path,
            language=language,
            task='transcribe',
            verbose=False,
            word_timestamps=True
        )
        
        # 자막 파일 생성 (SRT)
        srt_content = self._generate_srt(result['segments'])
        
        # VTT 파일 생성
        vtt_content = self._generate_vtt(result['segments'])
        
        return {
            'text': result['text'],
            'segments': result['segments'],
            'language': result['language'],
            'duration': result.get('duration'),
            'srt': srt_content,
            'vtt': vtt_content
        }
    
    def _generate_srt(self, segments):
        """
        SRT 자막 생성
        """
        srt_lines = []
        for i, seg in enumerate(segments):
            start = self._format_time(seg['start'])
            end = self._format_time(seg['end'])
            text = seg['text'].strip()
            
            srt_lines.append(f"{i+1}")
            srt_lines.append(f"{start} --> {end}")
            srt_lines.append(text)
            srt_lines.append("")
        
        return "\n".join(srt_lines)
    
    def _format_time(self, seconds):
        """
        초 → SRT 시간 형식
        """
        hours = int(seconds // 3600)
        minutes = int((seconds % 3600) // 60)
        secs = int(seconds % 60)
        millis = int((seconds % 1) * 1000)
        
        return f"{hours:02d}:{minutes:02d}:{secs:02d},{millis:03d}"

4. Multi-LLM 요약 서비스

# app/services/llm_service.py
class LLMService:
    """
    Multi-LLM 요약 서비스
    """
    def __init__(self, provider='openai', api_key=None):
        self.provider = provider
        self.api_key = api_key
        
        if provider == 'openai':
            import openai
            openai.api_key = api_key
            self.client = openai
        elif provider == 'anthropic':
            from anthropic import Anthropic
            self.client = Anthropic(api_key=api_key)
        elif provider == 'google':
            import google.generativeai as genai
            genai.configure(api_key=api_key)
            self.client = genai
    
    def summarize(self, text, prompt=None, model=None):
        """
        LLM 기반 요약
        
        Args:
            text: 전사 텍스트
            prompt: 커스텀 프롬프트
            model: gpt-4-turbo, claude-3-opus, gemini-pro
        
        Returns:
            {
                "summary": "요약 텍스트",
                "key_points": ["핵심1", "핵심2"],
                "structure": {...},
                "tokens_used": {...}
            }
        """
        # 기본 프롬프트
        if not prompt:
            prompt = """
다음 텍스트를 분석하고 구조화된 요약을 작성해주세요:
 
1. 전체 요약 (3-5문장)
2. 핵심 포인트 (5개)
3. 주요 주제/섹션별 정리
 
텍스트:
{text}
"""
        
        prompt = prompt.format(text=text)
        
        # 토큰 제한 대응 (청크 분할)
        if len(text) > 30000:
            return self._summarize_long_text(text, prompt, model)
        
        # 단일 요청
        if self.provider == 'openai':
            response = self.client.chat.completions.create(
                model=model or 'gpt-4-turbo',
                messages=[
                    {"role": "system", "content": "당신은 전문 요약 AI입니다."},
                    {"role": "user", "content": prompt}
                ],
                temperature=0.7,
                max_tokens=2000
            )
            
            summary_text = response.choices[0].message.content
            
        elif self.provider == 'anthropic':
            response = self.client.messages.create(
                model=model or 'claude-3-opus-20240229',
                messages=[
                    {"role": "user", "content": prompt}
                ],
                max_tokens=2000
            )
            
            summary_text = response.content[0].text
        
        # 구조화 파싱
        return self._parse_summary(summary_text)
    
    def _summarize_long_text(self, text, prompt, model):
        """
        긴 텍스트 Map-Reduce 요약
        """
        CHUNK_SIZE = 30000
        
        # 1단계: 청크 분할
        chunks = [text[i:i+CHUNK_SIZE] 
                  for i in range(0, len(text), CHUNK_SIZE)]
        
        # 2단계: 각 청크 요약 (Map)
        chunk_summaries = []
        for i, chunk in enumerate(chunks):
            chunk_prompt = f"다음 텍스트의 핵심을 요약:\n\n{chunk}"
            summary = self.summarize(chunk, chunk_prompt, model)
            chunk_summaries.append(summary['summary'])
        
        # 3단계: 통합 요약 (Reduce)
        combined = "\n\n".join([
            f"## 파트 {i+1}\n{s}"
            for i, s in enumerate(chunk_summaries)
        ])
        
        final_prompt = f"""
다음은 긴 텍스트의 파트별 요약입니다.
전체를 종합하여 최종 요약을 작성해주세요:
 
{combined}
"""
        
        return self.summarize(combined, final_prompt, model)

5. YouTube 통합

# app/services/youtube_downloader.py
import yt_dlp
 
class YouTubeDownloader:
    """
    YouTube 다운로드 서비스
    """
    def download(self, url, output_dir):
        """
        YouTube 영상 다운로드
        """
        ydl_opts = {
            'format': 'bestaudio',
            'outtmpl': f'{output_dir}/%(title)s.%(ext)s',
            'postprocessors': [{
                'key': 'FFmpegExtractAudio',
                'preferredcodec': 'wav',
                'preferredquality': '192',
            }],
            'quiet': False
        }
        
        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
            info = ydl.extract_info(url, download=True)
            
            return {
                'title': info.get('title'),
                'duration': info.get('duration'),
                'channel': info.get('uploader'),
                'file_path': ydl.prepare_filename(info)
            }

6. 데이터베이스 설계

-- tasks 테이블
CREATE TABLE tasks (
    id UUID PRIMARY KEY,
    type VARCHAR(20) CHECK (type IN ('upload', 'youtube')),
    status VARCHAR(20) DEFAULT 'pending',
    source_language VARCHAR(10),
    target_language VARCHAR(10),
    options JSONB,
    progress INTEGER DEFAULT 0,
    current_step VARCHAR(50),
    created_at TIMESTAMP DEFAULT NOW(),
    completed_at TIMESTAMP
);
 
-- files 테이블
CREATE TABLE files (
    id UUID PRIMARY KEY,
    task_id UUID REFERENCES tasks(id),
    filename VARCHAR(255),
    file_type VARCHAR(20),
    file_size BIGINT,
    duration INTEGER,
    storage_path TEXT,
    created_at TIMESTAMP DEFAULT NOW()
);
 
-- results 테이블
CREATE TABLE results (
    id UUID PRIMARY KEY,
    task_id UUID REFERENCES tasks(id),
    transcript TEXT,
    transcript_segments JSONB,
    translation TEXT,
    summary TEXT,
    key_points JSONB,
    processing_time INTEGER,
    created_at TIMESTAMP DEFAULT NOW()
);

🏗 시스템 아키텍처

전체 구조

[클라이언트]
    ↓ (REST API)
[Flask App]
    ├─ API Routes
    └─ Celery Task Queue
        ↓
[Redis Queue]
    ↓
[Celery Workers]
    ├─ FFmpeg (음성 추출)
    ├─ Whisper (STT)
    ├─ Translation API
    └─ LLM API
        ↓
[PostgreSQL]
[S3/MinIO]

디렉토리 구조

flask-video-transcription/
├── app/
│   ├── routes/          # API 엔드포인트
│   ├── services/        # 핵심 서비스
│   │   ├── media_processor.py
│   │   ├── stt_service.py
│   │   ├── translation.py
│   │   └── llm_service.py
│   ├── tasks/           # Celery 작업
│   ├── models/          # DB 모델
│   └── utils/           # 유틸리티
├── tests/
├── docker/
└── k8s/

📊 주요 기능

API 엔드포인트

엔드포인트	메서드	설명
`/api/upload`	POST	파일 업로드
`/api/youtube`	POST	YouTube 처리
`/api/task/<id>`	GET	작업 상태 조회
`/api/download/<id>/<type>`	GET	결과 다운로드

지원 파일 형식

영상: MP4, AVI, MKV, MOV, WMV, FLV 음성: MP3, WAV, M4A, FLAC, OGG, AAC 텍스트: TXT, SRT, VTT

처리 옵션

STT: Whisper (large-v3), Google STT
번역: Google Translation, DeepL
LLM: GPT-4, Claude 3, Gemini Pro

📈 성과

처리 성능

STT 속도: 실시간의 0.3배 (GPU)
2시간 영상: 약 36분 처리
동시 작업: 최대 10개 병렬 처리

비용 최적화

Whisper 로컬: API 비용 0원
청크 처리: LLM 토큰 30% 절감
캐싱: 중복 처리 0건

API 사용

YouTube 자동화: 일일 평균 5건 요청
성공률: 98%+
평균 응답 시간: < 3초 (작업 시작)

🔗 연동 프로젝트

YouTube 자동화 시스템 연동

# YouTube 자동화에서 YAP API 호출
import requests
 
def process_video_with_yap(video_path):
    """
    YAP API로 영상 전송 및 처리
    """
    # 1. 파일 업로드
    with open(video_path, 'rb') as f:
        response = requests.post(
            'http://yap-server:5000/api/upload',
            files={'files': f},
            data={
                'language': 'ko',
                'summarize': 'true',
                'prompt': '강의 내용을 5개 핵심으로 요약'
            }
        )
    
    task_id = response.json()['task_id']
    
    # 2. 작업 완료 대기
    while True:
        status_response = requests.get(
            f'http://yap-server:5000/api/task/{task_id}'
        )
        
        status = status_response.json()
        
        if status['status'] == 'completed':
            return status['result']
        elif status['status'] == 'failed':
            raise Exception(status['error'])
        
        time.sleep(10)

📝 배운 점

Flask API 서버 설계

RESTful 원칙
- 리소스 중심 URL 설계
- HTTP 메서드 적절한 사용
- 상태 코드 명확한 반환
비동기 처리 패턴
- Celery로 긴 작업 백그라운드 처리
- Redis 메시지 브로커
- 진행률 실시간 업데이트
에러 핸들링
- 재시도 로직 (max_retries=3)
- 부분 실패 허용 (청크별 처리)
- 상세한 에러 로깅

AI/ML 파이프라인 구축

Whisper 최적화
- GPU 가속으로 10배 속도 향상
- large-v3 모델로 정확도 95%+
- 타임스탬프로 정확한 자막 생성
Multi-LLM 통합
- GPT-4: 가장 정확하지만 비용 높음
- Claude: 긴 텍스트 처리 우수
- Gemini: 비용 대비 성능 우수
토큰 제한 극복
- Map-Reduce 방식 청크 처리
- 30K 글자 단위 분할
- 재병합으로 전체 요약

운영 경험

모니터링: Prometheus + Grafana
로깅: ELK Stack 중앙화
알림: 실패 시 Slack 알림

🔮 향후 계획

단기

화자 분리 (Diarization)
감정 분석 추가
키워드 자동 추출

중기

실시간 스트리밍 STT
다국어 자막 동시 생성
WebSocket 진행률 스트리밍

장기

모바일 앱 (React Native)
SaaS 전환 (과금 시스템)
Enterprise 기능 (팀, 권한 관리)

프로젝트 기간: 2024.12 - 현재 현재 상태: 완료됨 (운영 중) 역할: 범용 STT/AI API 서버 연동 프로젝트: YouTube 자동화 시스템 최종 업데이트: 2025-12-21

Quartz 4

Explorer

STT & AI 요약 API 시스템 (YAP)

STT & AI 요약 API 시스템 (YAP)

📋 프로젝트 개요

시스템 역할

핵심 기능

🎯 해결한 문제

API 서버의 필요성

해결한 기술적 과제

🛠 기술 스택

Backend

AI/ML

Media Processing

Infrastructure

💡 주요 구현 내용

1. RESTful API 설계

파일 업로드 API

YouTube 처리 API

작업 상태 조회 API

2. Celery 비동기 작업 처리

메인 처리 태스크

3. Whisper STT 서비스

4. Multi-LLM 요약 서비스

5. YouTube 통합

6. 데이터베이스 설계

🏗 시스템 아키텍처

전체 구조

디렉토리 구조

📊 주요 기능

API 엔드포인트

지원 파일 형식

처리 옵션

📈 성과

처리 성능

비용 최적화

API 사용

🔗 연동 프로젝트

YouTube 자동화 시스템 연동

📝 배운 점

Flask API 서버 설계

AI/ML 파이프라인 구축

운영 경험

🔮 향후 계획

단기

중기

장기

Graph View

Table of Contents