[5월 3주차-5/20(1)]AI 기반 문서 요약 및 회의록 자동화 시스템

AI 기술을 활용한 문서 요약과 회의록 자동화는 반복적이고 시간이 오래 걸리는 작업을 효율적으로 처리할 수 있도록 도와줍니다. 이 문서에서는 논문, 문서, 음성 파일을 기반으로 정보를 추출하고 요약하는 두 가지 주요 시스템을 설명합니다.

---

🧪 PART 1. 논문과 문서를 요약해주는 AI 연구원

✅ 시스템 개요

이 시스템은 PDF 파일로 제공되는 논문이나 보고서를 자동으로 분석하여, 문제 인식, 저자 소개, 주요 내용을 요약합니다. OpenAI GPT 모델을 활용하며, PyMuPDF를 통해 PDF에서 텍스트를 추출합니다.

📦 필수 라이브러리 설치

pip install PyMuPDF
pip install openai
pip install python-dotenv

🛠️ 주요 구현 기능

1. .env에서 API 키 불러오기

from dotenv import load_dotenv
import os
load_dotenv()
api_key = os.getenv("OPENAI_API_KEY")

2. PDF에서 텍스트 추출 (pdf_to_text 함수)

• 페이지별로 헤더, 푸터를 제거하고 본문만 추출
• fitz (PyMuPDF)를 이용하여 clip 영역을 지정함
• 결과는 텍스트 파일로 저장됨

3. 텍스트 요약 요청 (summarize_txt 함수)

• 추출된 텍스트를 system prompt로 GPT-4o 모델에 전달
• 저자의 문제 인식, 주장, 구조 등을 분석한 요약 생성

4. 전체 파이프라인 (summarize_pdf 함수)

• PDF → 텍스트 → 요약 → txt 저장까지 자동 처리

🧾 출력 (요약 텍스트)

# 웹 기반 밀 재배관리 의사결정 지원시스템 설계 및 구축

## 저자의 문제 인식 및 주장 (15문장 이내)

저자는 한국의 밀 소비량이 증가하고 있지만 자급률이 매우 낮아 안정적인 생산 기반이 필요하다고 인식하고 있다. 이를 해결하기 위해 정부는 밀 산업 발전 계획을 수립했으나, 밀의 자급률을 높이기 위해서는 가격과 품질 경쟁력을 확보해야 한다고 주장한다. 특히 노지에서 재배되는 밀은 기상환경의 영향을 크게 받기 때문에, 이를 극복하기 위한 체계적인 재배 및 관리가 필요하다. 저자는 디지털 기술을 활용한 스마트 농업이 이러한 문제를 해결할 수 있는 방안이라고 제안한다. 이를 위해 과정기반 작물모형인 APSIM을 활용한 웹 기반 의사결정 지원시스템을 설계하고 구축하였다. 이 시스템은 기상 데이터를 기반으로 밀의 잠재 수확량을 예측하고, 사용자에게 실시간으로 정보를 제공하여 농가의 의사결정을 지원한다. 특히, 과거 40년의 기상 데이터를 활용하여 미래 기상 시나리오를 분석하고, 잠재 수확량을 예측함으로써 농가가 기상 재해에 대비할 수 있도록 돕는다. 저자는 이 시스템이 전국의 기상 데이터와 토양 환경 정보를 활용하여 다른 지역에도 적용 가능하다고 주장한다. 또한, 향후 질소 함량 등 추가적인 영향 인자를 포함하여 모델을 고도화함으로써, 국내 밀 생산량 및 품질 증대에 기여할 수 있을 것으로 기대한다.

## 저자 소개

김솔희, 석승원, 청리광, 장태일, 김태곤은 전북대학교 스마트팜학과 및 농촌건설공학과에 소속된 연구자들로, 디지털 농업 및 작물 모델링 분야에서 연구를 진행하고 있다. 김태곤은 전북대학교 스마트팜학과의 조교수로, 본 연구의 교신저자이다. 이들은 디지털 농업 기술을 활용하여 농업 생산성을 향상시키기 위한 다양한 연구를 수행하고 있다.

 

전체 코드

from openai import OpenAI
from dotenv import load_dotenv
import os
import pymupdf

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

def pdf_to_text(pdf_file_path: str):
    pdf_file_path = "data/과정기반 작물모형을 이용한 웹 기반 밀 재배관리 의사결정 지원시스템 설계 및 구축.pdf"
    doc = pymupdf.open(pdf_file_path)

    header_height = 80
    footer_height = 80

    full_text = ''

    for page in doc:
        rect = page.rect # 페이지 크기 가져오기
        
        header = page.get_text(clip=(0, 0, rect.width , header_height))
        footer = page.get_text(clip=(0, rect.height - footer_height, rect.width , rect.height))
        text = page.get_text(clip=(0, header_height, rect.width , rect.height - footer_height))

        full_text += text + '\n------------------------------------\n'

    # 파일명만 추출
    pdf_file_name = os.path.basename(pdf_file_path)
    pdf_file_name = os.path.splitext(pdf_file_name)[0] # 확장자 제거

    txt_file_path = f'data/{pdf_file_name}_with_preprocessing.txt'

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

    return txt_file_path


def summarize_txt(file_path: str): # ①
    client = OpenAI(api_key=api_key)

    # 주어진 텍스트 파일을 읽어들인다.
    with open(file_path, 'r', encoding='utf-8') as f:
        txt = f.read()

    # 요약을 위한 시스템 프롬프트를 생성한다.
    system_prompt = f'''
    너는 다음 글을 요약하는 봇이다. 아래 글을 읽고, 저자의 문제 인식과 주장을 파악하고, 주요 내용을 요약하라. 

    작성해야 하는 포맷은 다음과 같다. 
    
    # 제목

    ## 저자의 문제 인식 및 주장 (15문장 이내)
    
    ## 저자 소개

    
    =============== 이하 텍스트 ===============

    { txt }
    '''

    print(system_prompt)
    print('=========================================')

    # OpenAI API를 사용하여 요약을 생성한다.
    response = client.chat.completions.create(
        model="gpt-4o",
        temperature=0.1,
        messages=[
            {"role": "system", "content": system_prompt},
        ]
    )

    return response.choices[0].message.content

def summarize_pdf(pdf_file_path: str, output_file_path: str):
    txt_file_path = pdf_to_text(pdf_file_path)
    summary = summarize_txt(txt_file_path)

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


if __name__ == '__main__':
    pdf_file_path = "data/과정기반 작물모형을 이용한 웹 기반 밀 재배관리 의사결정 지원시스템 설계 및 구축.pdf"
    summarize_pdf(pdf_file_path, 'output/crop_model_summary2.txt')

---

 

 

🔊 PART 2. 회의록을 정리하는 AI 서기

✅ 시스템 개요

회의나 강의 등 음성 파일(mp3)을 받아 텍스트로 변환하고, 화자 분리(diarization)를 통해 누가 언제 무엇을 말했는지 자동으로 구분합니다.

📦 필수 패키지 설치

pip install openai
pip install torch torchvision torchaudio
pip install --upgrade transformers datasets[audio] accelerate
pip install pyannote.audio
pip install numpy==1.26.4

⚙️ 전제 조건

• ffmpeg 설치 필요: 오디오 디코딩을 위해 사용

os.environ["PATH"] += os.pathsep + r"[ffmpeg 설치 경로]/bin"

🛠️ 주요 기능 구성

1. Whisper 기반 STT (whisper_stt)

• 모델: openai/whisper-large-v3-turbo
• 청크 단위 오디오 처리 (10초 단위 + 2초 stride)
• 결과를 pandas DataFrame으로 저장하고 .csv로 출력

2. 화자 분리 (speaker_diarization)

• 모델: pyannote/speaker-diarization-3.1
• RTTM 포맷으로 저장하고 pandas로 변환
• 화자별 발화 구간을 식별하고 groupby로 병합

3. 텍스트-화자 매칭 (stt_to_rttm)

• Whisper로 받아쓴 텍스트의 구간을 diarization 결과와 매칭
• 겹치는 시간이 가장 긴 화자에게 텍스트를 배정
• 최종적으로 화자별 회의록 .csv 파일 생성

전체 코드

import os
import pandas as pd
from pyannote.audio import Pipeline



os.environ["PATH"] += os.pathsep + r"C:\Users\Admin\PycharmProjects\20250520\ffmpeg-2025-05-15-git-12b853530a-full_build\bin"

import torch
from transformers import AutoModelForSpeechSeq2Seq, AutoProcessor, pipeline

### 받아쓰기하는 함수 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

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="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"
    )

    # cuda가 사용 가능한 경우 cuda를 사용하도록 설정
    if torch.cuda.is_available():
        pipeline.to(torch.device("cuda"))
        print('cuda is available')
    else:
        print('cuda is not available')

    diarization_pipeline = pipeline(audio_file_path)

    # dump the diarization output to disk using RTTM format
    with open(output_rttm_file_path, "w", encoding='utf-8') as rttm:
        diarization_pipeline.write_rttm(rttm)

    # pandas dataframe으로 변환
    df_rttm = pd.read_csv(
        output_rttm_file_path,  # rttm 파일 경로
        sep=' ',  # 구분자는 띄어쓰기
        header=None,  # 헤더는 없음
        names=['type', 'file', 'chnl', 'start', 'duration', 'C1', 'C2', 'speaker_id', 'C3', 'C4']
    )

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

    # speaker_id를 기반으로 화자별로 구간을 나누기
    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"]

    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_rttm에
    "text"라는 빈 열을 추가하고 Whisper 모델로 받아온 텍스트를 채운다.
    """
    df_rttm["text"] = ""

    """
    중요한 부분은 for 문을 사용해 위스퍼 모델로 받아쓴 결과인 df_stt의
    'start', 'end', 'text' 정보를 이용해 df_rttm의 적절한 행을 찾는다
    """
    for i_stt, row_stt in df_stt.iterrows():
        overlap_dict = {}

        '''
        df_stt 행과 df_rttm 행이 겹치는 시간을 각각 계산해서 가장 많이 겹치는 행에 텍스트를 추가.
        '''
        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"]))
            overlap_dict[i_rttm] = overlap

        max_overlap = max(overlap_dict.values())
        max_overlap_idx = max(overlap_dict, key=overlap_dict.get)

        '''
        겹치는 시간이 0초라면 해당 문장을 건너 뛴다.
        '''
        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 = "audio/싼기타_비싼기타.mp3"              # 원본 오디오 파일
    stt_output_file_path = "audio/싼기타_비싼기타.csv"         # STT 결과 파일
    rttm_file_path = "audio/싼기타_비싼기타.rttm"              # 화자 분리 원본 파일
    rttm_csv_file_path = "audio/싼기타_비싼기타_rttm.csv"     # 화자 분리 CSV 파일
    final_csv_file_path = "audio/싼기타_비싼기타_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
    )

    print(df_rttm)

---

🔐 보안 및 인증

• .env에 반드시 OPENAI_API_KEY 포함
• pyannote.audio 사용 시 HuggingFace access token 필요 (use_auth_token="hf_xxx")

---