2조 NLP 자유주제 프로젝트(MRC)

2조 : 박세연, 오현정, 임규민, 최준헌 

1. 주제 : 주제: 행정 문서에 대한 기계독해 모델 개발

1-1. 주제 선정 배경

행정 문서는 복잡한 용어와 형식 때문에 일반 시민들이 이해하기 어려운 경우가 많습니다. 정부와 시민 간의 원활한 의사소통은 행정 문서의 이해를 기반으로 하며, 이를 통해 복지 혜택 신청, 민원 처리, 세금 신고 등의 행정 서비스 이용이 원활해집니다. 하지만 행정 문서의 내용이 시민들에게 친숙하지 않으면 행정 서비스의 이용이 어렵고, 불필요한 시간과 비용이 발생하게 됩니다. 이 프로젝트에서는 행정 문서 데이터를 기반으로 MRC 모델을 학습시키고, 사용자가 행정 문서에 대해 보다 정확하고 빠르게 이해할 수 있도록 돕는 솔루션을 개발하고자 합니다.

 

2. 분석 과정

import json
import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split
from datasets import load_dataset
from transformers import AutoTokenizer

2-1. 파일 불러오기 

# Step 1: Train과 Validation 데이터 로드
train_file_path = '/content/drive/MyDrive/CODE/TL_span_extraction_how.json'  # Train 데이터 JSON 파일 경로
valid_file_path = '/content/drive/MyDrive/CODE/VL_span_extraction_how.json'  # Validation 데이터 JSON 파일 경로

# JSON 데이터 로드
with open(train_file_path, 'r', encoding='utf-8') as f: # 읽기 모드로 파일을 열고 f 변수에 할당 
    train_data = json.load(f)['data']  # 파일 내부의 data 키 값 추출  

with open(valid_file_path, 'r', encoding='utf-8') as f:
    valid_data = json.load(f)['data']

print("Train 데이터 및 Validation 데이터 로드 완료!")

 

2-2. 데이터 형식 검증 

# Step 2: 데이터 구조 검증
def validate_data_structure(data):
    """데이터 구조를 검증합니다."""
    required_keys = ['doc_id', 'doc_title', 'doc_class', 'paragraphs']
    for entry in data:
        for key in required_keys:
            if key not in entry:
                raise ValueError(f"필수 키 '{key}'가 누락되었습니다: {entry}")
        for para in entry['paragraphs']:
            if 'context' not in para or 'qas' not in para:
                raise ValueError(f"'context' 또는 'qas' 키가 누락되었습니다: {para}")
            for qa in para['qas']:
                if 'question' not in qa or 'answers' not in qa:
                    raise ValueError(f"'question' 또는 'answers' 키가 누락되었습니다: {qa}")
                if isinstance(qa['answers'], dict):  # answers가 dict 형태일 경우 경고
                    print(f"answers가 dict로 되어 있습니다. 자동으로 리스트로 변환합니다: {qa}")
    print("데이터 구조 검증 완료!")

# 함수 적용 
validate_data_structure(train_data)
validate_data_structure(valid_data)

---

 

validate_data_structure(data)

required_keys = ['doc_id', 'doc_title', 'doc_class', 'paragraphs']

 

필수 키 정의: 각 데이터 항목에서 반드시 포함되어야 하는 키 나열

• doc_id: 문서 ID
• doc_title: 문서 제목
• doc_class: 문서의 분류
• paragraphs: 문단 데이터(리스트 형태)

for entry in data:
    for key in required_keys:
        if key not in entry:
            raise ValueError(f"필수 키 '{key}'가 누락되었습니다: {entry}")

 

 

최상위 키 검증 

• for entry in data: data 리스트의 각 항목(entry)을 순회
• for key in required_keys: 각 항목에서 모든 필수 키가 포함되어 있는지 확인
• if key not in entry: entry에 필수 키가 없으면 오류(ValueError)를 발생
  • 오류 메시지에는 누락된 항목의 세부 정보(entry)를 포함해 디버깅을 쉽게 함 

 

for para in entry['paragraphs']:

 

문단(paragraphs) 검증

• 각 문서(entry)의 paragraphs 항목을 순회
  • 문단 데이터(para)가 반드시 context와 qas 키를 포함해야 함 
  • context: 문단의 텍스트 내용
  • qas: 질문-답변 데이터(리스트 형태)

for qa in para['qas']:
    if 'question' not in qa or 'answers' not in qa:
        raise ValueError(f"'question' 또는 'answers' 키가 누락되었습니다: {qa}")

 

질문-답변(qas) 검증

• 각 문단의 qas(질문-답변 데이터)를 순회 
  • qa 항목이 반드시 question과 answers 키를 포함해야 함 
  • question: 질문 텍스트
  • answers: 정답 데이터(리스트 또는 딕셔너리 형태)

if isinstance(qa['answers'], dict):  # answers가 dict 형태일 경우 경고
    print(f"answers가 dict로 되어 있습니다. 자동으로 리스트로 변환합니다: {qa}")

 

answer 데이터 검증 

 

• answers가 딕셔너리(dict)로 되어 있는지 확인
• 딕셔너리로 되어 있는 경우, 이를 경고 메시지로 출력
• 자동 변환 관련 메시지는 단순한 로깅이며 실제 변환은 별도의 단계에서 수행될 수 있음 

print("데이터 구조 검증 완료!")

모든 검증을 통과하면 성공 메시지를 출력

 

---

2-3. SQuAD 형식으로 변환 

# Step 3: SQuAD 형식으로 변환
def convert_to_squad_format(data):
    """
    데이터를 SQuAD 형식으로 변환합니다.
    Args:
        data (list): 입력 데이터
    Returns:
        dict: SQuAD 형식 데이터
    """
    squad_format = {"data": []}

    for entry in data:
        title = entry['doc_title']
        paragraphs = []

        for para in entry['paragraphs']:
            context = para['context']
            qas = []

            for qa in para['qas']:
                # answers를 SQuAD 형식으로 변환
                if isinstance(qa['answers'], dict):  # answers가 dict 형태일 경우 변환
                    answers = [
                        {
                            "text": qa['answers']['text'],
                            "answer_start": qa['answers']['answer_start']
                        }
                    ]
                elif isinstance(qa['answers'], list):  # answers가 리스트 형태일 경우 그대로 사용
                    answers = qa['answers']
                else:
                    print(f"answers 필드가 비정상입니다. 건너뜁니다: {qa}")
                    continue

                # context와 answer_start가 일치하는지 검증
                for answer in answers:
                    if not context[answer['answer_start']:answer['answer_start'] + len(answer['text'])] == answer['text']:
                        print(f"answer_start와 context가 일치하지 않습니다. 건너뜁니다: {qa}")
                        continue

                # Q&A 추가
                qas.append({
                    "id": qa['question_id'],  # 유니크 ID
                    "question": qa['question'],
                    "answers": answers,
                    "is_impossible": qa.get('is_impossible', False)  # 기본값 False
                })

            # 문단 추가
            paragraphs.append({
                "context": context,
                "qas": qas
            })

        # 문서 추가
        squad_format["data"].append({
            "title": title,
            "paragraphs": paragraphs
        })

    return squad_format

# Train 데이터 변환
train_squad = convert_to_squad_format(train_data)
train_output_file = '/content/drive/MyDrive/CODE/new_train_data_squad.json'
with open(train_output_file, 'w', encoding='utf-8') as f:
    json.dump(train_squad, f, ensure_ascii=False, indent=4)

print(f"Train 데이터 변환 및 저장 완료: {train_output_file}")

# Validation 데이터 변환
valid_squad = convert_to_squad_format(valid_data)
valid_output_file = '/content/drive/MyDrive/CODE/new_valid_data_squad.json'
with open(valid_output_file, 'w', encoding='utf-8') as f:
    json.dump(valid_squad, f, ensure_ascii=False, indent=4)

print(f"Validation 데이터 변환 및 저장 완료: {valid_output_file}")

▶ SQuAD(SQuAD: Stanford Question Answering Dataset) 형식 기본 구조 

{
  "data": [
    {
      "title": "문서 제목",
      "paragraphs": [
        {
          "context": "문단 텍스트",
          "qas": [
            {
              "id": "질문 ID",
              "question": "질문 텍스트",
              "answers": [
                {"text": "정답 텍스트", "answer_start": "정답 시작 위치"}
              ],
              "is_impossible": false
            }
          ]
        }
      ]
    }
  ]
}

 

 

---

convert_to_squad_format

문서 변환

for entry in data:
    title = entry['doc_title']
    paragraphs = []

▷ 각 문서(entry)를 SQuAD 형식의 "title"과 "paragraphs"로 변환

 

문단 변환

for para in entry['paragraphs']:
    context = para['context']
    qas = []

▷ 문단 데이터(paragraphs)를 변환하여 SQuAD 형식의 "context"와 "qas"로 구성

 

질문-답변(QA) 변환

# answer 변환
if isinstance(qa['answers'], dict):  # answers가 dict 형태일 경우 변환
    answers = [
        {
            "text": qa['answers']['text'],
            "answer_start": qa['answers']['answer_start']
        }
    ]
elif isinstance(qa['answers'], list):  # answers가 리스트 형태일 경우 그대로 사용
    answers = qa['answers']
else:
    print(f"answers 필드가 비정상입니다. 건너뜁니다: {qa}")
    continue

▷  질문(qa)의 정답(answers) 데이터를 SQuAD 형식으로 변환

• qa['answers']가 딕셔너리(dict)라면: 딕셔너리를 리스트로 변환
  • 변환된 형식: [{"text": "정답 텍스트", "answer_start": "시작 위치"}

•  qa['answers']가 리스트라면: 그대로 사용
•  answers가 이상한 형태라면: 경고 메시지를 출력하고 해당 QA를 무시

•  

정답과 문맥 검증

for answer in answers:
    if not context[answer['answer_start']:answer['answer_start'] + len(answer['text'])] == answer['text']:
        print(f"answer_start와 context가 일치하지 않습니다. 건너뜁니다: {qa}")
        continue

▷ 정답(answers)의 시작 위치(answer_start)와 텍스트(text)가 실제 문맥(context)과 일치하는지 확인

 

QA 추가 

qas.append({
    "id": qa['question_id'],  # 유니크 ID
    "question": qa['question'],
    "answers": answers,
    "is_impossible": qa.get('is_impossible', False)  # 기본값 False
})

▷  변환된 질문(qas) 데이터를 SQuAD 형식으로 추가

• 구성:
  • "id": 질문의 고유 ID
  • "question": 질문 텍스트
  • "answers": 변환된 정답 리스트
  • "is_impossible": 정답이 존재하지 않는 질문인지 여부 

문서 추가 

squad_format["data"].append({
    "title": title,
    "paragraphs": paragraphs
})

▷  변환된 문서 데이터를 SQuAD 형식의 "data" 리스트에 추가

 

결과 반환(SQuAD 형식) 

return squad_format

2-4. Test 데이터 분리 

# Step 4: Test 데이터 분리 및 저장
def convert_to_dataframe(squad_data):
    """
    SQuAD 형식의 데이터를 Pandas DataFrame으로 변환합니다.

    Args:
        squad_data (dict): SQuAD 형식의 데이터

    Returns:
        pandas.DataFrame: 변환된 DataFrame
    """
    data_list = []
    for data_item in squad_data['data']:
        for paragraph in data_item['paragraphs']:
            for qa in paragraph['qas']:
                for answer in qa['answers']:
                    data_list.append({
                        'doc_title': data_item['title'],
                        'doc_id': qa['id'],
                        'context': paragraph['context'],
                        'question': qa['question'],
                        'answer_text': answer['text'],
                        'answer_start': answer['answer_start'],
                        'is_impossible': qa['is_impossible']
                    })
    return pd.DataFrame(data_list)

# Train 데이터를 DataFrame으로 변환
with open(train_output_file, 'r', encoding='utf-8') as f:
    train_squad_loaded = json.load(f)

train_df = convert_to_dataframe(train_squad_loaded)

# Train 데이터를 Test와 New Train으로 분리
test_df, new_train_df = train_test_split(
    train_df,
    test_size=0.2,
    random_state=42,
    stratify=train_df['is_impossible']  # 데이터 균형 유지
)

# Test 데이터를 SQuAD 형식으로 변환
# test_squad = convert_to_dataframe(test_df.to_dict(orient='records'))  # Error line
# The following code block converts the test_df back to SQuAD format
test_squad = {
    "data": [
        {
            "title": row["doc_title"],
            "paragraphs": [
                {
                    "context": row["context"],
                    "qas": [
                        {
                            "id": row["doc_id"],
                            "question": row["question"],
                            "answers": [
                                {
                                    "text": row["answer_text"],
                                    "answer_start": row["answer_start"]
                                }
                            ],
                            "is_impossible": row["is_impossible"]
                        }
                    ]
                }
            ]
        }
        for index, row in test_df.iterrows()
    ]
}

test_output_file = '/content/drive/MyDrive/CODE/test_data_squad.json'

with open(test_output_file, 'w', encoding='utf-8') as f:
    json.dump(test_squad, f, ensure_ascii=False, indent=4)

print(f"Test 데이터 변환 및 저장 완료: {test_output_file}")

---

convert_to_dataframe

1. squad_data['data']를 순회하며 문서(data_item)를 처리
2. 각 문서의 paragraphs를 순회하며 문단(paragraph)를 처리
3. 각 문단의 qas를 순회하며 질문(qa)를 처리
4. 각 질문의 answers를 순회하며 정답 데이터를 추출
5. data_list 리스트에 각 데이터 항목을 딕셔너리로 추가
6. data_list를 Pandas DataFrame으로 변환해 반환

---

Train 데이터를 DataFrame으로 변환

- SQuAD 형식으로 저장된 JSON 파일(train_output_file)을 읽어 Pandas DataFrame으로 변환

 

Train 데이터를 Test(20%)와 New Train(80%)으로 분리

 

• test_df: 테스트 데이터로 사용할 DataFrame
• new_train_df: 새로운 훈련 데이터로 사용할 DataFrame

Test 데이터를 SQuAD 형식으로 변환 및 JSON 파일로 저장 

 

- test_df를 다시 SQuAD 형식으로 변환하여 저장

2-5. SQuAD 형식의 데이터 가져오기

from datasets import load_dataset

# JSON 파일 경로
train_output_file = '/content/drive/MyDrive/CODE/new_train_data_squad.json'
valid_output_file = '/content/drive/MyDrive/CODE/new_valid_data_squad.json'
test_output_file = '/content/drive/MyDrive/CODE/test_data_squad.json'

# 데이터 로드
dataset = load_dataset("json", data_files={
    "train": train_output_file,
    "validation": valid_output_file,
    "test": test_output_file
})

print("데이터 로드 완료!")

2-6. 데이터 평탄화 

def flatten_squad_data(data):
    """
    중첩된 SQuAD 데이터 구조를 평탄화하여 context, question, answers 필드를 생성합니다.
    Args:
        data (dict): SQuAD 형식의 중첩 데이터
    Returns:
        list: 평탄화된 데이터 리스트
    """
    flattened_data = []
    for doc in data["data"]:
        title = doc.get("title", "")
        for paragraph in doc["paragraphs"]:
            context = paragraph["context"]
            for qa in paragraph["qas"]:
                flattened_data.append({
                    "title": title,
                    "context": context,
                    "question": qa["question"],
                    "answers": qa["answers"]
                })
    return flattened_data
    
# Train 데이터 평탄화
train_flattened = flatten_squad_data(train_squad)

# Validation 데이터 평탄화
valid_flattened = flatten_squad_data(valid_squad)

# Test 데이터 평탄화
test_flattened = flatten_squad_data(test_squad)

평탄화(Flattening)는 계층적이고 중첩된 데이터 구조를 단순화하여 1차원 또는 더 낮은 차원의 구조로 변환하는 과정을 의미한다. 

SQuAD 데이터의 중첩 구조를 가진다.

 

• 최상위 레벨(data)에는 여러 문서(title)가 포함되어 있다.
• 각 문서에는 여러 문단(paragraphs)이 있고, 각 문단에는 여러 질문-답변 세트(qas)가 중첩되어 있다.
• 질문-답변(qas)은 다시 정답(answers)을 포함하는 리스트 형태로 중첩되어 있다.

원본 데이터 

{
  "data": [
    {
      "title": "History of Rome",
      "paragraphs": [
        {
          "context": "Rome was not built in a day.",
          "qas": [
            {
              "question": "What is this about?",
              "answers": [{"text": "Rome", "answer_start": 0}]
            }
          ]
        }
      ]
    }
  ]
}

평탄화 결과

[
    {
        "title": "Document Title",
        "context": "Paragraph text here.",
        "question": "What is this?",
        "answers": [
            {
                "text": "An example.",
                "answer_start": 18
            }
        ]
    }
]

 

2-7. 토크나이저 

SQuAD 형식의 데이터를 Hugging Face Transformers 라이브러리를 활용해 모델 학습 및 평가에 적합한 형태로 전처리

 

토크나이저 로드 

model_name = "bert-base-uncased"
tokenizer = AutoTokenizer.from_pretrained(model_name)

 

데이터를 Hugging Face Dataset 형식으로 변환

train_dataset = Dataset.from_list(train_flattened)
valid_dataset = Dataset.from_list(valid_flattened)
test_dataset = Dataset.from_list(test_flattened)

dataset = {"train": train_dataset, "validation": valid_dataset, "test": test_dataset}

 

전처리 함수 정의 

def preprocess_function(examples):
    tokenized_examples = tokenizer(
        examples["question"],
        examples["context"],
        truncation=True,
        max_length=384,
        padding="max_length",
        return_offsets_mapping=True,
    )

 

정답 위치(start/end positions) 매핑

    start_positions = []
    end_positions = []

    for i, offsets in enumerate(tokenized_examples["offset_mapping"]):
        answers = examples["answers"][i]
        if len(answers) > 0:
            start_char = answers[0]["answer_start"]
            end_char = start_char + len(answers[0]["text"])

            token_start_index = 0
            token_end_index = len(offsets) - 1
            while token_start_index < len(offsets) and offsets[token_start_index][0] <= start_char:
                token_start_index += 1
            while token_end_index >= 0 and offsets[token_end_index][1] >= end_char:
                token_end_index -= 1

            start_positions.append(token_start_index - 1)
            end_positions.append(token_end_index + 1)
        else:
            start_positions.append(0)
            end_positions.append(0)

 

결과 구성

    tokenized_examples["start_positions"] = start_positions
    tokenized_examples["end_positions"] = end_positions
    tokenized_examples.pop("offset_mapping")
    return tokenized_examples

 

데이터셋 전처리 

tokenized_datasets = {
    split: dataset[split].map(preprocess_function, batched=True)
    for split in ["train", "validation", "test"]
}

 

결과 출력 

print(tokenized_datasets)

 

 

3. 결과

 

Hugging Face Transformers 라이브러리를 사용하여 질문 응답(Q&A) 모델을 훈련 및 평가

 

모델 훈련 및 평가

from transformers import TrainingArguments, Trainer, DefaultDataCollator,AutoModelForQuestionAnswering
import evaluate  # metrics 계산에 필요

# 평가 함수 정의
def compute_metrics(eval_pred):
    metric = evaluate.load("squad")
    start_logits, end_logits = eval_pred.predictions
    # Assuming eval_pred.label_ids contains the start and end positions
    start_positions, end_positions = eval_pred.label_ids

    predictions = []

    # Accessing the input_ids from the eval_dataset
    # Here, I assume you have access to the original tokenized_datasets
    # Replace "eval" with the actual split name if different
    for i in range(len(start_logits)):
        input_ids = tokenized_datasets["validation"][i]["input_ids"]
        start_idx = np.argmax(start_logits[i])
        end_idx = np.argmax(end_logits[i])

        # Ensure start_idx and end_idx are within bounds
        start_idx = min(start_idx, len(input_ids) - 1)
        end_idx = min(end_idx, len(input_ids) - 1)

        predictions.append({
            "id": str(i),
            "prediction_text": tokenizer.decode(
                input_ids[start_idx : end_idx + 1]
            )
        })

    references = [
        {
            "id": str(i),
            "answers": {
                'text': [tokenizer.decode(tokenized_datasets["validation"][i]["input_ids"][start_positions[i] : end_positions[i] + 1])],
                'answer_start': [start_positions[i]]
            }
        }
        for i in range(len(start_positions))
    ]

    return metric.compute(predictions=predictions, references=references)

# 학습 설정
training_args = TrainingArguments(
    output_dir="./results",
    eval_strategy="steps",
    eval_steps=500,
    save_strategy="steps",
    save_steps=1000,
    learning_rate=3e-5,
    per_device_train_batch_size=4,
    per_device_eval_batch_size=4,
    num_train_epochs=1,
    weight_decay=0.01,
    logging_dir="./logs",
    logging_steps=500,
    save_total_limit=2,
    load_best_model_at_end=True,
    fp16=True,
    report_to="none",
)

# Data Collator
data_collator = DefaultDataCollator(return_tensors="pt")

# Before initializing the Trainer, load your pre-trained model
model = AutoModelForQuestionAnswering.from_pretrained(model_name) # This line is added

# Trainer 설정
trainer = Trainer(
    model=model,
    args=training_args,
    train_dataset=tokenized_datasets["train"],
    eval_dataset=tokenized_datasets["validation"],
    data_collator=data_collator,
    compute_metrics=compute_metrics,
)

 

 

모델 훈련 결과 출력

trainer.train()

 

평가 데이터셋의 성능 확인

results = trainer.evaluate()
print(results)

1. eval_loss : 모델이 평가 데이터에 대해 예측한 결과와 실제 정답 간의 손실(loss)
   eval_loss: NaN은 모델이 평가 데이터를 처리하는 동안 비정상적인 계산이 발생했음.
2. eval_exact_match : 모델이 예측한 정답 텍스트와 실제 정답 텍스트가 완전히 일치(Exact Match) 하는 비율
   eval_exact_match: 약 62.05%의 정확도를 기록. 초기 모델로는 나쁘지 않은 성과지만, 추가 학습이나 데이터 정제를 통해 개선 가능.
   
   
3. eval_f1 : 모델이 예측한 정답 텍스트와 실제 정답 텍스트 간의 부분적인 일치 정도(F1 Score)를 나타냄
   eval_f1: 약 67.02점을 기록. 모델이 대체로 정확히 예측하지만, 일부 경우는 완전히 일치하지 않음. 데이터 품질이나 훈련 시간 증가로 개선 가능.

 

훈련된 모델과 토크나이저를 저장

model.save_pretrained("./saved_model")
tokenizer.save_pretrained("./saved_model")

 

4. 해석

 

모델 해석

• Exact Match는 약 62%, F1 Score는 약 67%로 초기 모델로는 괜찮은 성능.
• 평가 손실: NaN으로 손실 계산 문제 있음. 데이터셋 또는 학습 과정 점검 필요.
• 훈련: 1 에포크만 진행되어 학습 부족 상태.

---

개선 방향

1. 데이터 품질:
   • 정답 레이블 확인 및 데이터 다양성 개선.
   • 데이터 증강으로 훈련 데이터 확장.
2. 모델 및 학습 설정:
   • 훈련 에포크 증가.
   • 손실 계산 문제 해결 및 학습률 조정.
3. 평가 및 테스트:
   • 오류 사례 분석 및 실제 환경에서 성능 점검.
   • 도메인별 데이터로 일반화 성능 테스트.

---

해당 결과를 바탕으로 향후에 데이터 개선과 학습 설정 튜닝이 이루어진다면 모델 성능이 향상될 것이라 생각됨.