9. MeetingGPT - 회의록 분석 및 요약 AI 만들기

작성일

안녕하세요! 이번 챕터에서는 회의나 강의 영상으로부터 자동으로 회의록을 만들고, 내용을 요약하고, 궁금한 점을 질문할 수 있는 “MeetingGPT” 애플리케이션을 만들어 보겠습니다. 비디오 파일 처리, 음성 인식, 그리고 LangChain을 이용한 자연어 처리까지 다양한 기술을 경험하게 될 것입니다.

🎯 이번 챕터에서 배울 것

  • FFmpeg: 동영상에서 오디오를 추출하는 방법을 배웁니다.
  • Pydub: 긴 오디오 파일을 작은 조각으로 나누는 방법을 익힙니다.
  • OpenAI Whisper: 음성을 텍스트로 변환하는 STT(Speech-to-Text) 기술을 사용합니다.
  • Streamlit: 파일 업로드, 탭, 상태 메시지 등 다양한 UI 컴포넌트를 활용합니다.
  • LangChain: Refine Chain을 이용해 긴 문서를 요약하고, RAG(Retrieval-Augmented Generation) 기술로 문서 기반 Q&A 챗봇을 구축합니다.
  • Caching: Streamlit의 캐싱 기능으로 앱의 성능을 최적화합니다.

비디오에서 오디오 추출하기

🎯 이번 단계에서 배울 것

  • MeetingGPT 앱의 기본 구조 설정하기
  • ffmpeg을 사용하여 비디오 파일에서 오디오를 추출하는 방법 배우기
  • subprocess 모듈을 사용하여 외부 커맨드 실행하기

📝 1단계: MeetingGPT 페이지 생성

가장 먼저, 새로운 Streamlit 페이지를 만듭니다. pages 폴더 안에 05_MeetingGPT.py 파일을 생성하고 기본적인 설정을 추가합니다.

전체 코드 (pages/05_MeetingGPT.py):

1
2
3
4
5
6
import streamlit as st

st.set_page_config(
    page_title="MeetingGPT",
    page_icon="💼",
)

🔍 코드 상세 설명

  • st.set_page_config: Streamlit 앱의 제목과 아이콘을 설정합니다. 이 코드는 각 페이지 파일에서 가장 먼저 실행되는 것이 좋습니다.

📝 2단계: 오디오 추출 기능 구현

Jupyter Notebook에서 비디오를 오디오로 변환하는 기능을 먼저 실험해 보겠습니다. 이 기능은 ffmpeg이라는 아주 강력하고 유명한 오픈소스 도구를 사용합니다.

전체 코드 (notebook.ipynb):

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
import subprocess

def extract_audio_from_video(video_path, audio_path):
    command = [
        "ffmpeg",
        "-i",
        video_path,
        "-vn",
        audio_path,
    ]
    subprocess.run(command)

extract_audio_from_video(
    "./files/podcast.mp4",
    "./files/podcast.mp3",
)

🔍 코드 상세 설명

1. ffmpeg이란?

  • ffmpeg은 비디오와 오디오 파일을 처리하기 위한 무료 오픈소스 소프트웨어입니다. 변환, 스트리밍, 녹화 등 거의 모든 종류의 미디어 파일 처리가 가능합니다. 우리는 이것을 사용해 비디오 파일(podcast.mp4)에서 오디오 트랙만 추출하여 MP3 파일(podcast.mp3)로 저장할 것입니다.
  • -i: 입력(input) 파일 지정
  • -vn: 비디오(video)를 포함하지 않음 (오디오만 추출)

2. subprocess.run()이란?

  • Python 코드 내에서 외부 커맨드(명령어)를 실행하고 싶을 때 사용하는 모듈입니다. 우리는 ffmpeg 명령어를 Python을 통해 실행하기 위해 이 함수를 사용합니다.

📝 3단계: 홈페이지 및 의존성 업데이트

마지막으로, 홈페이지에 MeetingGPT 링크를 활성화하고, 필요한 라이브러리를 requirements.txt에 추가합니다.

수정 코드 (Home.py):

1
2
3
4
5
# Before
# - [ ] [MeetingGPT](/MeetingGPT)

# After
# - [ ] [💼 MeetingGPT](/MeetingGPT)
  • 이모지를 추가하여 좀 더 보기 좋게 만들었습니다.

✅ 체크리스트

  • pages/05_MeetingGPT.py 파일을 생성하고 set_page_config를 추가했나요?
  • ffmpeg이 무엇인지, 왜 사용하는지 이해했나요?
  • subprocess.run()으로 Python에서 외부 명령어를 실행하는 방법을 이해했나요?

💡 연습 과제

  1. 다른 포맷으로 변환: extract_audio_from_video 함수를 수정하여 오디오를 mp3가 아닌 wav 포맷으로 저장해 보세요. (ffmpeg 명령어 옵션을 찾아보세요!)
  2. 에러 처리: subprocess.run()에 check=True 인자를 추가하면 명령어 실행 실패 시 에러를 발생시킵니다. 이것을 추가하고, 존재하지 않는 비디오 파일 경로를 입력하여 어떻게 동작하는지 확인해 보세요.

긴 오디오 파일 분할하기

🎯 이번 단계에서 배울 것

  • pydub 라이브러리를 사용하여 오디오 파일 로드 및 조작하기
  • 긴 오디오를 작은 조각(chunk)으로 나누는 방법 배우기
  • 오디오 분할을 위한 시간 계산 및 반복문 활용하기

📝 1단계: pydub 라이브러리 소개

pydub은 Python으로 오디오를 조작할 수 있게 해주는 간단하고 편리한 라이브러리입니다. 자르기, 붙이기, 볼륨 조절 등 다양한 작업을 쉽게 할 수 있습니다. 우리는 이 라이브러리를 사용해 1시간이 넘는 긴 오디오 파일을 10분 단위의 작은 파일로 나눌 것입니다.

왜 나눠야 할까요? OpenAI의 Whisper API 같은 음성 인식 API들은 한 번에 처리할 수 있는 파일 크기나 길이에 제한이 있는 경우가 많습니다. 따라서 큰 파일을 작은 조각으로 나눠서 처리하는 것은 매우 일반적인 전략입니다.

📝 2단계: 오디오 파일 로드 및 분할 로직 구현

전체 코드 (notebook.ipynb):

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from pydub import AudioSegment
import math

# 이전 단계에서 추출한 오디오 파일을 로드합니다.
track = AudioSegment.from_mp3("./files/podcast.mp3")

# 10분을 밀리초(ms) 단위로 변환합니다. pydub은 시간을 ms로 다룹니다.
ten_minutes = 10 * 60 * 1000

# 전체 오디오 길이를 10분으로 나누어 몇 개의 조각이 필요한지 계산합니다.
# math.ceil은 소수점을 올림하여 정수로 만듭니다. (예: 7.3 -> 8)
chunks = math.ceil(len(track) / ten_minutes)

# 계산된 조각 수만큼 반복합니다.
for i in range(chunks):
    start_time = i * ten_minutes
    end_time = (i + 1) * ten_minutes

    # track[start:end]와 같이 Python 리스트처럼 오디오를 자를 수 있습니다.
    chunk = track[start_time:end_time]

    # 자른 오디오 조각을 별도의 파일로 저장합니다.
    chunk.export(f"./files/chunks/chunk_{i}.mp3", format="mp3")

🔍 코드 상세 설명

  • AudioSegment.from_mp3(): MP3 파일을 pydub이 다룰 수 있는 AudioSegment 객체로 불러옵니다.
  • len(track): 오디오의 전체 길이를 밀리초(ms) 단위로 반환합니다.
  • track[start_time:end_time]: 파이썬 리스트를 슬라이싱하는 것과 똑같은 방식으로 오디오의 특정 구간을 잘라낼 수 있습니다. 매우 직관적이죠!
  • chunk.export(): AudioSegment 객체를 실제 오디오 파일로 저장합니다. format 인자로 포맷을 지정할 수 있습니다.

✅ 체크리스트

  • pydub을 왜 사용하는지 이해했나요?
  • 밀리초(ms) 단위로 시간을 계산하는 방법을 이해했나요?
  • AudioSegment 객체를 자르고(slicing) 저장하는(export) 방법을 익혔나요?

💡 연습 과제

  1. 다른 길이로 자르기: 코드를 수정하여 오디오를 10분 대신 5분 단위로 잘라보세요. 몇 개의 파일이 생성되나요?
  2. 마지막 조각 길이 확인: 마지막으로 생성된 오디오 조각의 길이를 확인하는 코드를 추가해 보세요. (힌트: len(chunk)) 다른 조각들과 길이가 다를 수 있습니다. 왜 그럴까요?

오디오를 텍스트로 변환하기 (Whisper API)

🎯 이번 단계에서 배울 것

  • OpenAI의 Whisper API를 사용하여 오디오를 텍스트로 변환하는 방법 배우기
  • glob 라이브러리를 사용하여 특정 패턴의 파일 목록을 가져오는 방법 익히기
  • 여러 개의 텍스트 조각을 하나로 합치고 파일로 저장하는 방법 배우기
  • 코드를 재사용 가능한 함수로 리팩토링하기

📝 1단계: 코드 리팩토링

먼저, 이전 단계에서 작성한 오디오 분할 코드를 재사용하기 쉽도록 함수로 만들어 보겠습니다.

수정 코드 (notebook.ipynb):

1
2
3
4
5
6
7
8
9
def cut_audio_in_chunks(audio_path, chunk_size, chunks_folder):
    track = AudioSegment.from_mp3(audio_path)
    chunk_len = chunk_size * 60 * 1000
    chunks = math.ceil(len(track) / chunk_len)
    for i in range(chunks):
        start_time = i * chunk_len
        end_time = (i + 1) * chunk_len
        chunk = track[start_time:end_time]
        chunk.export(f"./{chunks_folder}/chunk_{i}.mp3", format="mp3")

📝 2단계: Whisper API로 오디오 변환

이제 잘라진 오디오 조각들을 하나씩 Whisper API로 보내 텍스트로 변환하는 함수를 만들겠습니다.

전체 코드 (notebook.ipynb):

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
import openai
import glob

def transcribe_chunks(chunk_folder, destination):
    # chunk_folder에 있는 모든 .mp3 파일의 목록을 가져옵니다.
    files = glob.glob(f"{chunk_folder}/*.mp3")
    
    final_transcript = ""
    
    # 각 오디오 파일을 순서대로 처리합니다.
    for file in files:
        # 파일을 'rb'(read binary) 모드로 엽니다. 오디오 파일은 바이너리 파일입니다.
        with open(file, "rb") as audio_file:
            # openai.Audio.transcribe를 호출하여 음성을 텍스트로 변환합니다.
            transcript = openai.Audio.transcribe(
                "whisper-1",  # 사용할 모델 이름
                audio_file,
            )
            # 변환된 텍스트를 final_transcript 변수에 계속 추가합니다.
            final_transcript += transcript["text"]
            
    # 합쳐진 전체 텍스트를 destination 파일에 씁니다.
    with open(destination, "w") as file:
        file.write(final_transcript)

🔍 코드 상세 설명

  • glob.glob("*.mp3"): glob은 파일 경로를 찾을 때 와일드카드(*, ? 등)를 사용할 수 있게 해주는 라이브러리입니다. *.mp3는 “이름이 무엇이든 상관없이 .mp3로 끝나는 모든 파일”이라는 뜻입니다.
  • with open(file, "rb"): 파일을 열 때 사용하는 구문입니다. with를 사용하면 파일을 다 쓴 후에 자동으로 닫아주어 편리합니다. 오디오, 이미지 같은 미디어 파일은 텍스트가 아니므로 ‘바이너리(binary)’ 모드인 "rb"로 열어야 합니다.
  • openai.Audio.transcribe(): OpenAI 라이브러리에서 음성-텍스트 변환을 담당하는 함수입니다. 사용할 모델(whisper-1)과 바이너리 모드로 열린 오디오 파일을 넘겨주면, 변환된 텍스트가 담긴 객체를 반환합니다.

✅ 체크리스트

  • glob을 사용하여 여러 파일을 한 번에 선택하는 방법을 이해했나요?
  • 오디오 파일을 열 때 왜 "rb" 모드를 사용해야 하는지 이해했나요?
  • openai.Audio.transcribe 함수를 사용하여 음성을 텍스트로 변환할 수 있나요?

💡 연습 과제

  1. 개별 파일로 저장: transcribe_chunks 함수를 수정하여, 각 오디오 조각의 텍스트 변환 결과를 하나의 큰 파일이 아닌, 각자 별도의 .txt 파일(예: chunk_0.txt, chunk_1.txt)로 저장하도록 만들어 보세요.
  2. 처리 순서 정렬: glob.glob이 반환하는 파일 목록은 운영체제에 따라 순서가 보장되지 않을 수 있습니다. 파일 목록을 이름순으로 정렬(files.sort())하는 코드를 추가하여 항상 chunk_0, chunk_1, chunk_2… 순서로 처리되도록 보장해 보세요.

메모리 효율적인 텍스트 변환

🎯 이번 단계에서 배울 것

  • 파일을 추가 모드(append mode)로 열어 내용을 이어 쓰는 방법 배우기
  • 대용량 데이터를 처리할 때 메모리를 효율적으로 사용하는 코드 작성법 익히기
  • with 구문을 사용하여 여러 파일을 동시에 안전하게 다루는 방법 배우기

📝 1단계: 기존 코드의 문제점

이전 단계의 transcribe_chunks 함수는 잘 동작하지만, 한 가지 잠재적인 문제가 있습니다.

1
2
3
4
5
6
final_transcript = ""
for file in files:
    # ...
    final_transcript += transcript["text"] # 이 부분!
with open(destination, "w") as file:
    file.write(final_transcript)

만약 오디오 파일이 매우 길어서 변환된 텍스트의 양이 엄청나게 크다면(수십~수백 MB), final_transcript라는 변수 하나가 모든 텍스트를 메모리에 저장하고 있어야 합니다. 이는 메모리를 비효율적으로 사용하는 방식이며, 시스템에 부담을 줄 수 있습니다.

📝 2단계: 코드 개선

이 문제를 해결하기 위해, 텍스트를 변수에 계속 쌓아두는 대신, 변환될 때마다 결과 파일에 바로바로 이어 쓰도록 코드를 개선해 보겠습니다.

비교 예시:

Before:

1
2
3
4
5
6
7
8
9
10
11
12
def transcribe_chunks(chunk_folder, destination):
    files = glob.glob(f"{chunk_folder}/*.mp3")
    final_transcript = "" # 텍스트를 저장할 변수
    for file in files:
        with open(file, "rb") as audio_file:
            transcript = openai.Audio.transcribe(
                "whisper-1",
                audio_file,
            )
            final_transcript += transcript["text"] # 메모리에 계속 추가
    with open(destination, "w") as file: # 마지막에 한 번에 쓰기
        file.write(final_transcript)

After:

1
2
3
4
5
6
7
8
9
10
11
def transcribe_chunks(chunk_folder, destination):
    files = glob.glob(f"{chunk_folder}/*.mp3")
    for file in files:
        # 오디오 파일과 텍스트 파일을 동시에 엽니다.
        with open(file, "rb") as audio_file, open(destination, "a") as text_file:
            transcript = openai.Audio.transcribe(
                "whisper-1",
                audio_file,
            )
            # 변환된 텍스트를 바로 파일에 이어 씁니다.
            text_file.write(transcript["text"])

🔍 코드 상세 설명

  • open(destination, "a"): 파일을 열 때 두 번째 인자로 "a"를 주면 ‘추가(append) 모드’가 됩니다.
    • "w" (write): 파일 전체를 새로 씀 (기존 내용 삭제)
    • "a" (append): 파일의 맨 끝에 내용을 이어 씀 (기존 내용 유지)
  • with open(...) as f1, open(...) as f2:: with 구문은 콤마(,)를 사용하여 여러 파일을 동시에 열고 안전하게 관리할 수 있습니다.
  • 이 방식은 변환된 텍스트를 메모리에 저장하지 않고 바로 파일에 쓰기 때문에, 아무리 큰 파일이라도 메모리 사용량이 거의 늘어나지 않는다는 큰 장점이 있습니다.

✅ 체크리스트

  • 파일 열기 모드 "w"와 "a"의 차이점을 설명할 수 있나요?
  • 왜 추가(append) 모드를 사용하는 것이 메모리 효율적인지 이해했나요?
  • with 구문으로 여러 파일을 동시에 여는 방법을 알고 있나요?

💡 연습 과제

  1. 로그 파일 만들기: transcribe_chunks 함수를 수정하여, 텍스트 전문 파일 외에 별도의 log.txt 파일을 만들어 보세요. 이 로그 파일에는 “Transcribing chunk_0.mp3…”, “Transcribing chunk_1.mp3…” 와 같이 현재 어떤 파일이 처리되고 있는지 기록되도록 해보세요. (힌트: log.txt는 추가 모드로 열어야 합니다.)

사용자 인터페이스(UI) 및 캐싱 구현

🎯 이번 단계에서 배울 것

  • Streamlit의 st.file_uploader를 사용하여 파일 업로드 UI 만들기
  • st.status를 사용하여 오래 걸리는 작업의 진행 상태를 사용자에게 보여주기
  • @st.cache_data 데코레이터를 사용하여 함수의 결과를 캐싱하고 앱 성능 최적화하기
  • Jupyter Notebook의 실험 코드를 실제 웹 애플리케이션으로 통합하기

📝 1단계: 함수를 앱으로 이동 및 캐싱 적용

Jupyter Notebook에서 만들었던 extract_audio_from_video, cut_audio_in_chunks, transcribe_chunks 함수들을 05_MeetingGPT.py 파일로 옮겨옵니다. 그리고 아주 중요한 최적화 작업을 추가합니다: 바로 캐싱(Caching) 입니다.

전체 코드 (pages/05_MeetingGPT.py):

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
import os

# .cache 폴더에 최종 결과물인 팟캐스트 텍스트 파일이 있는지 확인
has_transcript = os.path.exists("./.cache/podcast.txt")

@st.cache_data()
def transcribe_chunks(chunk_folder, destination):
    if has_transcript:
        return
    # ... (기존 함수 내용) ...

@st.cache_data()
def extract_audio_from_video(video_path):
    if has_transcript:
        return
    # ... (기존 함수 내용) ...

@st.cache_data()
def cut_audio_in_chunks(audio_path, chunk_size, chunks_folder):
    if has_transcript:
        return
    # ... (기존 함수 내용) ...

🔍 코드 상세 설명

  • @st.cache_data(): Streamlit의 마법 같은 기능인 ‘캐싱 데코레이터’입니다. 함수 위에 이 코드를 붙여주면, Streamlit은 함수를 어떤 인자(argument)로 호출했는지, 그리고 그 결과가 무엇이었는지를 기억합니다. 만약 다음에 똑같은 인자로 함수가 다시 호출되면, 함수를 또 실행하는 대신 기억해 둔 결과를 즉시 반환합니다.
  • 오디오 추출, 분할, 변환 작업은 매우 오래 걸리고 비용(API 사용료)이 발생할 수 있습니다. 캐싱을 사용하면 사용자가 페이지를 새로고침하거나 다른 행동을 해도 이미 완료된 작업은 다시 실행하지 않으므로, 앱의 반응성이 엄청나게 향상되고 비용도 절약됩니다.
  • if has_transcript: return 코드를 추가하여, 최종 결과물이 이미 존재하면 이 비싼 함수들이 즉시 종료되도록 한번 더 방어해 줍니다.

📝 2단계: 파일 업로드 UI 및 처리 파이프라인 구축

이제 사용자가 직접 비디오 파일을 올릴 수 있는 UI를 만들고, 파일이 업로드되면 우리가 만든 함수들을 순서대로 실행하는 파이프라인을 구축합니다.

전체 코드 (pages/05_MeetingGPT.py):

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
st.markdown("""
# MeetingGPT
Welcome to MeetingGPT...
""")

with st.sidebar:
    video = st.file_uploader(
        "Video",
        type=["mp4", "avi", "mkv", "mov"],
    )

if video:
    chunks_folder = "./.cache/chunks"
    with st.status("Loading video...") as status:
        video_content = video.read()
        video_path = f"./.cache/{video.name}"
        audio_path = video_path.replace("mp4", "mp3")
        transcript_path = video_path.replace("mp4", "txt")
        
        with open(video_path, "wb") as f:
            f.write(video_content)
            
        status.update(label="Extracting audio...")
        extract_audio_from_video(video_path)
        
        status.update(label="Cutting audio segments...")
        cut_audio_in_chunks(audio_path, 10, chunks_folder)
        
        status.update(label="Transcribing audio...")
        transcribe_chunks(chunks_folder, transcript_path)

🔍 코드 상세 설명

  • st.sidebar: with st.sidebar: 블록 안에 있는 모든 UI 요소는 화면 왼쪽의 사이드바에 나타납니다.
  • st.file_uploader: 파일을 업로드할 수 있는 위젯을 만듭니다. type 인자로 허용할 파일 확장자를 지정할 수 있습니다.
  • if video:: file_uploader는 파일이 업로드되면 UploadedFile 객체를, 아니면 None을 반환합니다. 이 if문은 사용자가 파일을 업로드했을 때만 아래 코드가 실행되도록 합니다.
  • st.status("..."): 오래 걸리는 작업의 진행 상태를 사용자에게 보여주는 위젯입니다. with 블록이 끝나면 자동으로 “Completed” 메시지로 바뀝니다.
  • status.update(label="..."): st.status의 메시지를 동적으로 변경합니다. 이를 통해 사용자에게 현재 어떤 단계가 진행 중인지 명확하게 알려줄 수 있습니다.

✅ 체크리스트

  • @st.cache_data가 무엇이고 왜 사용해야 하는지 설명할 수 있나요?
  • st.file_uploader를 사용해 파일 업로드 기능을 만들 수 있나요?
  • st.status와 status.update를 사용해 사용자에게 진행 상황을 알려줄 수 있나요?

💡 연습 과제

  1. 업로드 파일 정보 표시: video 객체는 name, size, type 같은 유용한 속성들을 가지고 있습니다. 파일이 업로드되면 사이드바에 이 정보들을 st.write로 표시해 보세요.
  2. 캐시 지우기 버튼: Streamlit에는 캐시를 수동으로 지우는 기능이 없습니다. 하지만 st.button을 만들고, 버튼이 클릭되면 .cache 폴더 안의 파일들을 os.remove나 shutil.rmtree를 사용해 직접 삭제하는 기능을 구현해 보세요. (주의: 파일 시스템을 직접 조작하는 것은 위험할 수 있으니, 어떤 파일/폴더를 지우는지 명확히 확인하고 실행하세요!)

결과 표시를 위한 UI 개선

🎯 이번 단계에서 배울 것

  • Streamlit의 st.tabs를 사용하여 콘텐츠를 탭으로 구성하는 방법 배우기
  • st.status의 update 메서드를 사용하여 동적으로 상태 메시지를 변경하는 방법 익히기 (복습)
  • 파일을 읽고 그 내용을 화면에 표시하는 방법 배우기

📝 1단계: 동적인 상태 업데이트 (개선)

이전 단계에서는 st.status를 여러 번 사용하여 각 단계를 표시했습니다. 이것을 하나의 st.status 블록 안에서 status.update를 사용하는 방식으로 개선하여 사용자 경험을 더 좋게 만들 수 있습니다. (이 내용은 9.5 단계에서 이미 After 코드로 반영되었습니다. 여기서는 개념을 다시 한번 명확히 짚고 넘어갑니다.)

비교 예시:

Before:

1
2
3
4
5
with st.status("Extracting audio..."):
    extract_audio_from_video(video_path)
with st.status("Cutting audio segments..."):
    cut_audio_in_chunks(audio_path, 10, chunks_folder)
# ...

After:

1
2
3
4
5
6
7
8
with st.status("Loading video...") as status:
    # ...
    status.update(label="Extracting audio...")
    extract_audio_from_video(video_path)
    
    status.update(label="Cutting audio segments...")
    cut_audio_in_chunks(audio_path, 10, chunks_folder)
    # ...

After 방식은 사용자에게 하나의 진행 표시줄 안에서 상태 메시지만 바뀌는 것처럼 보여 훨씬 깔끔합니다.

📝 2단계: 탭(Tab)으로 결과 구성하기

이제 모든 처리가 끝나고 생성된 결과물(텍스트 전문, 요약, Q&A)을 보여줄 공간을 만들겠습니다. st.tabs를 사용하면 여러 콘텐츠를 깔끔한 탭 인터페이스로 정리할 수 있습니다.

전체 코드 (pages/05_MeetingGPT.py):

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
if video:
    # ... (파일 처리 파이프라인) ...

    # 탭을 생성합니다. 탭 이름이 담긴 리스트를 전달합니다.
    transcript_tab, summary_tab, qa_tab = st.tabs(
        [
            "Transcript",
            "Summary",
            "Q&A",
        ]
    )

    # "Transcript" 탭에 들어갈 내용을 정의합니다.
    with transcript_tab:
        # 생성된 텍스트 파일을 'r'(read) 모드로 엽니다.
        with open(transcript_path, "r") as file:
            # 파일 내용을 읽어서 화면에 표시합니다.
            st.write(file.read())

🔍 코드 상세 설명

  • st.tabs([...]): 탭 위젯을 생성합니다. 인자로 전달된 리스트의 각 문자열이 탭의 이름이 됩니다. 이 함수는 각 탭에 해당하는 객체들을 튜플 형태로 반환합니다.
  • with transcript_tab:: st.tabs가 반환한 탭 객체를 with 구문과 함께 사용하면, 해당 with 블록 안에 있는 내용이 그 탭 안에 표시됩니다.
  • open(transcript_path, "r"): 텍스트 파일을 읽기 위해 ‘읽기(read) 모드’인 "r"로 엽니다.
  • st.write(file.read()): file.read()로 파일의 전체 내용을 문자열로 읽어온 뒤, st.write를 사용해 화면에 보여줍니다. st.write는 텍스트, 데이터프레임, 마크다운 등 다양한 것을 “알아서 잘” 표시해주는 편리한 함수입니다.

✅ 체크리스트

  • st.tabs를 사용하여 탭 UI를 만들 수 있나요?
  • with 구문을 사용하여 특정 탭에 콘텐츠를 추가하는 방법을 이해했나요?
  • 텍스트 파일을 읽어서 st.write로 화면에 표시할 수 있나요?

💡 연습 과제

  1. 탭 이름과 아이콘 변경: st.tabs는 탭 이름에 이모지를 추가하는 것을 지원합니다. 탭 이름들을 ["📜 Transcript", "📝 Summary", "❓ Q&A"] 와 같이 변경해 보세요.
  2. 마크다운 사용: st.write 대신 st.markdown을 사용하여 텍스트 전문을 표시해 보세요. 어떤 차이가 있나요? (힌트: st.markdown은 텍스트를 마크다운으로 해석합니다.)

Refine Chain을 이용한 텍스트 요약

🎯 이번 단계에서 배울 것

  • LangChain의 Refine Chain 패턴을 이해하고 구현하는 방법 배우기
  • TextLoader와 RecursiveCharacterTextSplitter를 사용하여 긴 텍스트를 처리 가능한 조각으로 나누기
  • 점진적으로 요약을 개선해나가는 프롬프트 엔지니어링 기법 배우기
  • st.button을 사용하여 사용자 액션을 트리거하는 방법 익히기

📝 1단계: LangChain 설정 및 문서 분할

요약 기능을 구현하기 위해 필요한 LangChain 모듈들을 가져오고, 긴 텍스트 전문을 작은 조각으로 나누는 준비를 합니다.

추가된 코드 (pages/05_MeetingGPT.py):

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from langchain.chat_models import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.document_loaders import TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.schema import StrOutputParser

llm = ChatOpenAI(temperature=0.1)

# ...

with summary_tab:
    start = st.button("Generate summary")

    if start:
        # 1. 문서 로드
        loader = TextLoader(transcript_path)
        
        # 2. 문서 분할
        splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
            chunk_size=800,
            chunk_overlap=100,
        )
        docs = loader.load_and_split(text_splitter=splitter)

🔍 코드 상세 설명

  • TextLoader: 텍스트 파일을 LangChain이 다룰 수 있는 Document 형식으로 불러옵니다.
  • RecursiveCharacterTextSplitter: LLM이 한 번에 처리할 수 있는 토큰 양(Context Window)에는 한계가 있습니다. 이 스플리터는 긴 텍스트를 지정된 chunk_size에 맞춰 의미적으로 최대한 연결되는 작은 조각(Document 객체들의 리스트)으로 나누어 줍니다. chunk_overlap은 조각 간에 겹치는 부분을 만들어 문맥이 끊어지는 것을 방지합니다.

📝 2단계: 초기 요약 및 Refine 로직 구현

이제 “Refine Chain” 패턴을 구현합니다. 이 패턴의 핵심 아이디어는 다음과 같습니다.

  1. 첫 번째 문서 조각으로 초기 요약을 만든다.
  2. 다음 문서 조각과 기존 요약을 함께 LLM에게 주면서, “이 새로운 내용을 참고해서 기존 요약을 더 좋게 다듬어줘” 라고 요청한다.
  3. 모든 문서 조각에 대해 2번 과정을 반복한다.

추가된 코드 (pages/05_MeetingGPT.py):

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
# 3. 초기 요약 생성
first_summary_prompt = ChatPromptTemplate.from_template(
    """Write a concise summary of the following:
    "{text}"
    CONCISE SUMMARY:"""
)
first_summary_chain = first_summary_prompt | llm | StrOutputParser()
summary = first_summary_chain.invoke({"text": docs[0].page_content})

# 4. 점진적 요약 개선 (Refine)
refine_prompt = ChatPromptTemplate.from_template(
    """
    Your job is to produce a final summary.
    We have provided an existing summary up to a certain point: {existing_summary}
    We have the opportunity to refine the existing summary (only if needed) with some more context below.
    ------------
    {context}
    ------------
    Given the new context, refine the original summary.
    If the context isn't useful, RETURN the original summary.
    """
)
refine_chain = refine_prompt | llm | StrOutputParser()

with st.status("Summarizing...") as status:
    for i, doc in enumerate(docs[1:]):
        status.update(label=f"Processing document {i+1}/{len(docs)-1} ")
        summary = refine_chain.invoke({
            "existing_summary": summary,
            "context": doc.page_content,
        })
st.write(summary)

🔍 코드 상세 설명

  • st.button("..."): 클릭할 수 있는 버튼을 만듭니다. 버튼이 클릭되면 True를, 아니면 False를 반환합니다. if start: 구문을 통해 버튼이 클릭되었을 때만 요약 프로세스가 시작되도록 합니다.
  • first_summary_prompt: 첫 번째 문서 조각(docs[0])을 요약하기 위한 간단한 프롬프트입니다.
  • refine_prompt: Refine Chain의 핵심입니다. {existing_summary}와 {context}라는 두 개의 변수를 받아, 기존 요약을 새로운 문맥으로 개선하도록 LLM에게 지시합니다.
  • for i, doc in enumerate(docs[1:]):: 두 번째 문서 조각부터 마지막까지 순회하면서 refine_chain을 반복적으로 호출합니다. 매번 summary 변수가 더 정교한 내용으로 업데이트됩니다.

✅ 체크리스트

  • 긴 문서를 처리하기 위해 왜 TextSplitter가 필요한지 이해했나요?
  • “Refine Chain”이 어떤 원리로 동작하는지 설명할 수 있나요?
  • st.button을 사용하여 특정 기능을 실행시키는 방법을 알고 있나요?

💡 연습 과제

  1. 프롬프트 수정: refine_prompt를 수정하여 요약의 스타일을 바꿔보세요. 예를 들어, “Summarize in bullet points.” (불렛 포인트로 요약해줘) 또는 “Focus on action items and decisions made.” (결정된 사항과 해야 할 일 위주로 요약해줘) 같은 지시사항을 추가해 보세요.
  2. 진행 상황 표시: st.status 블록 안에서 st.write(summary)를 호출하여, 각 단계를 거칠 때마다 요약이 어떻게 변해가는지 실시간으로 화면에 표시해 보세요.

RAG를 이용한 Q&A 기능 구현

🎯 이번 단계에서 배울 것

  • RAG(Retrieval-Augmented Generation)의 기본 개념 이해하기
  • FAISS를 사용하여 텍스트 문서로부터 벡터 스토어(vector store)를 구축하는 방법 배우기
  • OpenAIEmbeddings와 CacheBackedEmbeddings를 사용하여 임베딩을 생성하고 캐시하는 방법 익히기
  • retriever를 사용하여 질문과 관련된 문서를 검색하는 방법 배우기

📝 1단계: RAG의 개념과 임베딩

**RAG(Retrieval-Augmented Generation, 검색 증강 생성)**는 LLM이 질문에 답변할 때, 관련된 정보를 외부 문서에서 ‘검색(Retrieval)’하여 그 내용을 ‘참고(Augmented)’해서 답변을 ‘생성(Generation)’하는 기술입니다. 이를 통해 LLM이 알지 못하는 최신 정보나 특정 문서의 내용에 대해서도 정확한 답변을 할 수 있게 됩니다.

이 과정의 핵심은 임베딩(Embedding) 입니다. 임베딩은 텍스트(단어, 문장, 문서)를 의미를 담은 숫자들의 벡터(vector)로 변환하는 과정입니다. 이렇게 변환하면 컴퓨터가 “의미적으로 비슷한” 텍스트들을 수학적으로 계산(벡터 간의 거리 측정)하여 찾을 수 있게 됩니다.

📝 2단계: 임베딩 및 벡터 스토어 생성 (embed_file 함수)

이제 텍스트 전문을 임베딩하고, 검색이 가능하도록 벡터 스토어(Vector Store)에 저장하는 함수를 만들어 보겠습니다.

추가된 코드 (pages/05_MeetingGPT.py):

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
from langchain.storage import LocalFileStore
from langchain.vectorstores.faiss import FAISS
from langchain.embeddings import CacheBackedEmbeddings, OpenAIEmbeddings

@st.cache_data()
def embed_file(file_path):
    # 임베딩을 저장할 캐시 디렉토리를 설정합니다.
    cache_dir = LocalFileStore(f"./.cache/embeddings/{os.path.basename(file_path)}")
    
    loader = TextLoader(file_path)
    docs = loader.load_and_split(text_splitter=splitter)
    
    # OpenAI 임베딩 모델을 정의합니다.
    embeddings = OpenAIEmbeddings()
    
    # 디스크에 임베딩을 캐싱하는 기능을 추가합니다.
    cached_embeddings = CacheBackedEmbeddings.from_bytes_store(embeddings, cache_dir)
    
    # 문서들을 임베딩하고 FAISS 벡터 스토어에 저장합니다.
    vectorstore = FAISS.from_documents(docs, cached_embeddings)
    
    # 벡터 스토어를 retriever로 변환하여 반환합니다.
    retriever = vectorstore.as_retriever()
    return retriever

🔍 코드 상세 설명

  • OpenAIEmbeddings: OpenAI의 임베딩 모델(예: text-embedding-ada-002)을 사용하여 텍스트를 벡터로 변환합니다.
  • LocalFileStore: 지정된 경로의 로컬 디스크에 데이터를 저장하는 간단한 저장소입니다.
  • CacheBackedEmbeddings: 아주 유용한 기능입니다. embeddings 모델을 cache_dir 저장소로 감싸줍니다. 어떤 텍스트에 대한 임베딩을 요청받으면, 먼저 캐시에 해당 임베딩이 있는지 확인합니다. 있으면 API를 호출하지 않고 캐시에서 바로 가져오고, 없으면 API를 호출하여 임베딩을 생성한 뒤 캐시에 저장합니다. API 호출 비용과 시간을 크게 절약해 줍니다.
  • FAISS: Facebook AI에서 만든, 매우 빠른 유사도 검색 라이브러리입니다. FAISS.from_documents는 문서들과 임베딩을 받아 FAISS 벡터 스토어를 구축합니다.
  • vectorstore.as_retriever(): 벡터 스토어를 ‘검색기(retriever)’ 객체로 변환합니다. 이 retriever는 “질문(query)을 받으면 가장 관련성 높은 문서들을 찾아주는” 역할을 합니다.

📝 3단계: Q&A 탭 구현

이제 Q&A 탭에서 retriever를 사용하여 질문과 관련된 문서 조각을 검색하는 기능을 구현합니다.

추가된 코드 (pages/05_MeetingGPT.py):

1
2
3
4
5
6
7
8
9
with qa_tab:
    # 1. 벡터 스토어/retriever 생성
    retriever = embed_file(transcript_path)

    # 2. retriever를 사용하여 문서 검색 (테스트)
    docs = retriever.invoke("do they talk about marcus aurelius?")

    # 3. 검색된 결과 표시
    st.write(docs)

🔍 코드 상세 설명

  • retriever = embed_file(...): 위에서 만든 함수를 호출하여 retriever를 가져옵니다. @st.cache_data 덕분에 이 비싼 과정은 파일 당 한 번만 실행됩니다.
  • retriever.invoke("..."): retriever의 가장 중요한 기능입니다. 여기에 질문을 던지면, 벡터 스토어에서 질문과 의미적으로 가장 유사한(관련성이 높은) 문서 조각들을 찾아 리스트 형태로 반환합니다.
  • st.write(docs)는 현재 검색된 문서 조각들을 그대로 보여줍니다. 실제 챗봇을 만들려면 이 docs와 사용자의 질문을 함께 LLM에게 보내 “이 문서들을 참고해서 질문에 답해줘” 라는 프롬프트를 구성해야 합니다. (이 부분은 다음 챕터에서 더 자세히 다룹니다.)

✅ 체크리스트

  • RAG가 무엇인지, 왜 필요한지 설명할 수 있나요?
  • 임베딩과 벡터 스토어의 역할을 이해했나요?
  • CacheBackedEmbeddings가 어떻게 시간과 비용을 절약해 주는지 이해했나요?
  • retriever가 어떤 역할을 하는지 알고 있나요?

💡 연습 과제

  1. 질문 입력 UI 만들기: 현재는 질문이 하드코딩되어 있습니다. st.text_input을 사용하여 사용자가 직접 질문을 입력할 수 있는 UI를 만들고, 사용자가 입력한 질문으로 문서를 검색하도록 코드를 수정해 보세요.
  2. 검색 결과 개수 조절: vectorstore.as_retriever()를 호출할 때 search_kwargs={"k": 5} 와 같은 인자를 추가하면 검색 결과의 개수를 조절할 수 있습니다. k 값을 1, 3, 5로 바꿔보면서 결과가 어떻게 달라지는지 확인해 보세요.

🎓 요약

축하합니다! MeetingGPT를 완성하며 정말 많은 것을 배웠습니다.

  • 비디오에서 오디오를 추출하고(ffmpeg), 작은 조각으로 나누고(pydub), 텍스트로 변환하는(Whisper) 미디어 처리 파이프라인을 구축했습니다.
  • 사용자가 파일을 업로드하고(st.file_uploader), 처리 과정을 지켜보고(st.status), 결과를 탭으로(st.tabs) 확인할 수 있는 Streamlit 웹 애플리케이션을 만들었습니다.
  • 비싼 계산 결과를 저장하여 재사용하는 캐싱(@st.cache_data, CacheBackedEmbeddings)의 중요성을 배웠습니다.
  • LangChain을 사용하여 긴 문서를 점진적으로 요약하는 Refine Chain과, 문서 기반 Q&A를 가능하게 하는 RAG 아키텍처를 직접 구현해 보았습니다.

출처 : https://nomadcoders.co/fullstack-gpt