# app.py from flask import Flask, render_template, request, session, redirect, url_for from flask_session import Session import torch from transformers import AutoTokenizer, AutoModelForCausalLM import nltk from rouge_score import rouge_scorer from sacrebleu.metrics import BLEU from datetime import datetime import os import math import logging import gc import time import re print("AI 모델과 평가 지표를 로딩합니다...") try: nltk_data_path = '/tmp/nltk_data' nltk.download('punkt', download_dir=nltk_data_path, quiet=True) nltk.data.path.append(nltk_data_path) model_name = "EleutherAI/polyglot-ko-1.3b" print(f"모델 로딩 중: {model_name}") tokenizer = AutoTokenizer.from_pretrained( model_name, trust_remote_code=True ) if tokenizer.pad_token is None: tokenizer.pad_token = tokenizer.eos_token device = torch.device("cuda" if torch.cuda.is_available() else "cpu") model = AutoModelForCausalLM.from_pretrained( model_name, torch_dtype=torch.float16 if torch.cuda.is_available() else torch.float32, low_cpu_mem_usage=True, trust_remote_code=True ) model.to(device) # 모델 최적화 model.eval() if torch.cuda.is_available(): model.half() scorer = rouge_scorer.RougeScorer(['rouge1', 'rouge2', 'rougeL'], use_stemmer=True) bleu = BLEU() print("AI 모델 로딩 및 최적화 완료.") model_loaded = True if torch.cuda.is_available(): print(f"GPU 메모리 사용량: {torch.cuda.memory_allocated() / 1024**3:.2f} GB") except Exception as e: print(f"모델 로딩 중 심각한 오류 발생: {e}") model_loaded = False app = Flask(__name__) app.config["SESSION_PERMANENT"] = False app.config["SESSION_TYPE"] = "filesystem" app.config['SECRET_KEY'] = os.environ.get('SECRET_KEY', os.urandom(24)) Session(app) log_handler = logging.FileHandler('report_log.txt', encoding='utf-8') log_handler.setLevel(logging.INFO) log_formatter = logging.Formatter('%(asctime)s - %(message)s', '%Y-%m-%d %H:%M:%S') log_handler.setFormatter(log_formatter) app.logger.addHandler(log_handler) app.logger.setLevel(logging.INFO) def is_structured_text(text): """문서가 구조화된 형식인지 판단""" lines = text.split('\n') # 구조화 패턴 체크 structure_indicators = 0 # 번호 매기기 패턴 (1., 1), (1), 가., 가), ①, ㉠ 등) numbering_patterns = [ r'^\s*\d+[\.\)]\s+', # 1. or 1) r'^\s*\(\d+\)\s+', # (1) r'^\s*[가-힣][\.\)]\s+', # 가. or 가) r'^\s*[①-⑳]\s+', # ① r'^\s*[ⅰ-ⅹ][\.\)]\s+', # ⅰ. or ⅰ) r'^\s*[a-zA-Z][\.\)]\s+', # a. or A) r'^\s*[-•‣⁃]\s+', # bullet points ] # 제목/헤더 패턴 header_patterns = [ r'^#{1,6}\s+', # Markdown headers r'^[제第]\s*\d+\s*[장절관조항]', # 제1장, 제2절 등 r'^\d+\.\s+\w+', # 1. 서론 r'^[ⅠⅡⅢⅣⅤⅥⅦⅧⅨⅩⅪⅫⅩⅢⅩⅣⅩⅤⅩⅥⅩⅦⅩⅧⅩⅨⅩⅩ][\.\s]', # 로마 숫자 ] total_lines = len([l for l in lines if l.strip()]) if total_lines == 0: return False numbered_lines = 0 header_lines = 0 for line in lines: if not line.strip(): continue # 번호 매기기 체크 for pattern in numbering_patterns: if re.match(pattern, line): numbered_lines += 1 break # 헤더 체크 for pattern in header_patterns: if re.match(pattern, line): header_lines += 1 break # 구조화 비율 계산 structure_ratio = (numbered_lines + header_lines) / total_lines # 20% 이상이 구조화되어 있으면 구조화된 문서로 판단 return structure_ratio > 0.2 def validate_ppl_text(text): text_len = len(text) if text_len < 2000: return {"valid": False, "message": f"텍스트가 너무 짧습니다. 현재 {text_len}자, 최소 2000자 이상 입력해주세요."} # 구조화된 텍스트는 반복 검사 완화 is_structured = is_structured_text(text) # 문자 수준 반복 패턴 검사 (연속된 동일 문자) char_repetitions = 0 max_consecutive = 0 current_consecutive = 1 for i in range(1, len(text)): if text[i] == text[i-1] and text[i] not in ' \n\t': # 공백 제외 current_consecutive += 1 max_consecutive = max(max_consecutive, current_consecutive) else: if current_consecutive > 10: # 10자 이상 연속되면 반복으로 간주 char_repetitions += current_consecutive current_consecutive = 1 char_repetition_ratio = char_repetitions / text_len # 구조화된 문서는 기준 완화 repetition_threshold = 0.5 if is_structured else 0.3 if char_repetition_ratio > repetition_threshold: return {"valid": False, "message": f"반복 문자가 너무 많습니다 ({char_repetition_ratio*100:.1f}%). 정상적인 텍스트를 입력해주세요."} # 단어 수준 반복 검사 words = text.split() if len(words) > 0: # 구조화된 문서의 경우 짧은 문구(번호, 제목 등) 제외 if not is_structured: bigrams = [' '.join(words[i:i+2]) for i in range(len(words) - 1)] trigrams = [' '.join(words[i:i+3]) for i in range(len(words) - 2)] bigram_unique_ratio = len(set(bigrams)) / len(bigrams) if bigrams else 1 trigram_unique_ratio = len(set(trigrams)) / len(trigrams) if trigrams else 1 if bigram_unique_ratio < 0.5 or trigram_unique_ratio < 0.6: return {"valid": False, "message": "반복되는 단어 패턴이 너무 많습니다. 다양한 내용의 텍스트를 입력해주세요."} # 토큰 수준 반복 검사 (기존 코드 유지 및 강화) tokens = tokenizer.convert_ids_to_tokens(tokenizer(text, max_length=1024, truncation=True).input_ids) # 구조화된 문서는 n-gram 검사 기준 완화 for n in range(2, 7): if len(tokens) >= n: ngrams = [tuple(tokens[i:i+n]) for i in range(len(tokens) - n + 1)] if ngrams: unique_ratio = len(set(ngrams)) / len(ngrams) # 구조화된 문서는 기준 완화 if is_structured: threshold = 0.2 + (n - 2) * 0.1 # 더 낮은 기준 else: threshold = 0.3 + (n - 2) * 0.1 if unique_ratio < threshold: return {"valid": False, "message": f"반복되는 {n}-gram 패턴이 너무 많습니다. 다양한 내용의 텍스트를 입력해주세요."} word_count = len(words) structure_msg = " (구조화된 문서로 감지됨)" if is_structured else "" return {"valid": True, "message": f"✅ 검증 완료: {text_len}자, {word_count}단어{structure_msg}"} def calculate_perplexity_logic(text, max_tokens=512, use_sliding_window=False): encodings = tokenizer(text, return_tensors="pt", max_length=max_tokens, truncation=True) input_ids = encodings.input_ids[0].to(device) if len(input_ids) < 10: raise ValueError("토큰 수가 너무 적습니다 (최소 10개)") tokens = tokenizer.convert_ids_to_tokens(input_ids) # 구조화된 텍스트 여부 확인 is_structured = is_structured_text(text) # GPT 스타일 텍스트 감지 (고품질 텍스트 특징) is_high_quality = detect_high_quality_text(text) # 반복 페널티 계산 (개선) repetition_penalties = {} # 의미있는 반복만 체크 (무의미한 반복 vs 의미있는 반복 구분) char_repetitions = 0 # 연속된 동일 문자만 체크 (아아아아 같은 패턴) for i in range(1, min(len(text), 1000)): if i < len(text) and text[i] == text[i-1] and text[i] not in ' \n\t.,!?;:': char_repetitions += 1 char_penalty = min(char_repetitions / 1000, 0.5) # 최대 0.5로 제한 # 토큰 수준 반복 체크 (완화) for n in range(3, 6): # 2-gram 제외, 3-gram부터 체크 if len(tokens) >= n: ngrams = [tuple(tokens[i:i+n]) for i in range(len(tokens) - n + 1)] if ngrams: unique_ratio = len(set(ngrams)) / len(ngrams) # 한글 텍스트는 기준 완화 repetition_penalties[f'{n}gram'] = max(0, (1 - unique_ratio) * 0.5) # 어휘 다양성 체크 (한글 특성 고려) unique_tokens = len(set(tokens)) total_tokens = len(tokens) vocabulary_diversity = unique_tokens / total_tokens # 한글은 토큰화 시 더 잘게 쪼개지므로 다양성 기준 완화 expected_diversity = 0.3 if any(ord(c) > 0x3130 for c in text[:100]) else 0.5 diversity_penalty = max(0, expected_diversity - vocabulary_diversity) # 고품질/구조화 텍스트 보너스 quality_bonus = 1.0 if is_high_quality: quality_bonus = 0.5 # 고품질 텍스트는 페널티 50% 감소 elif is_structured: quality_bonus = 0.7 # 구조화 텍스트는 페널티 30% 감소 # 종합 페널티 계산 (크게 완화) avg_repetition = sum(repetition_penalties.values()) / max(len(repetition_penalties), 1) # 실제 무의미한 반복이 있을 때만 강한 페널티 if char_penalty > 0.3: # 연속 반복이 많을 때 total_penalty = (avg_repetition * 2 + char_penalty * 3 + diversity_penalty) penalty_factor = math.exp(total_penalty * 3.0) else: # 일반적인 경우 total_penalty = (avg_repetition * 0.3 + char_penalty * 0.5 + diversity_penalty * 0.5) * quality_bonus penalty_factor = math.exp(total_penalty * 1.5) # 지수 크게 완화 seq_len = input_ids.size(0) with torch.no_grad(): if not use_sliding_window or seq_len <= 256: outputs = model(input_ids.unsqueeze(0), labels=input_ids.unsqueeze(0)) ppl = torch.exp(outputs.loss).item() else: max_length = 256 stride = 128 nlls = [] for begin_loc in range(0, seq_len, stride): end_loc = min(begin_loc + max_length, seq_len) input_chunk = input_ids[begin_loc:end_loc].unsqueeze(0) try: outputs = model(input_chunk, labels=input_chunk) if outputs.loss is not None and torch.isfinite(outputs.loss): nlls.append(outputs.loss) except Exception as chunk_error: print(f"청크 처리 오류: {chunk_error}") continue if not nlls: raise RuntimeError("유효한 loss 값을 계산할 수 없습니다") ppl = torch.exp(torch.mean(torch.stack(nlls))).item() # 기본 PPL 보정 (고품질 텍스트) if is_high_quality and ppl > 20: ppl = ppl * 0.6 # 고품질 텍스트는 기본 PPL 40% 감소 elif is_structured and ppl > 30: ppl = ppl * 0.8 # 구조화 텍스트는 20% 감소 adjusted_ppl = ppl * penalty_factor # 극단적인 값 제한 (상한선 설정) if adjusted_ppl > 200 and char_penalty < 0.2: # 반복이 적은데 PPL이 너무 높으면 조정 adjusted_ppl = min(adjusted_ppl, 150) return { 'base_ppl': ppl, 'adjusted_ppl': adjusted_ppl, 'penalty_factor': penalty_factor, 'token_count': len(input_ids), 'vocabulary_diversity': vocabulary_diversity, 'char_repetition': char_penalty, 'is_structured': is_structured, 'is_high_quality': is_high_quality } def detect_high_quality_text(text): """고품질 텍스트(GPT 생성 등) 감지""" indicators = 0 # 1. 문단 구조 체크 paragraphs = text.split('\n\n') if len(paragraphs) >= 3: indicators += 1 # 2. 문장 종결 일관성 sentences = re.split(r'[.!?]\s', text) if len(sentences) > 5: # 대부분 온전한 문장으로 끝나는지 complete_sentences = sum(1 for s in sentences if len(s.strip()) > 10) if complete_sentences / len(sentences) > 0.8: indicators += 1 # 3. 접속사/전환어 사용 transition_words = ['첫째', '둘째', '셋째', '따라서', '그러나', '또한', '예를 들어', '결론적으로', '무엇보다', '나아가', '더불어', '특히', '이에 따라'] transition_count = sum(1 for word in transition_words if word in text) if transition_count >= 3: indicators += 1 # 4. 전문 용어 밀도 professional_terms = ['체계', '구축', '기반', '활용', '분석', '모델', '시스템', '프로세스', '전략', '방안', '효과', '개선', '고도화'] prof_count = sum(text.count(term) for term in professional_terms) if prof_count > len(text.split()) * 0.02: # 2% 이상 indicators += 1 # 5. 균일한 문단 길이 if len(paragraphs) > 2: lengths = [len(p) for p in paragraphs if p.strip()] if lengths: avg_length = sum(lengths) / len(lengths) variance = sum((l - avg_length) ** 2 for l in lengths) / len(lengths) if variance < (avg_length * 0.5) ** 2: # 변동이 적음 indicators += 1 # 3개 이상 지표 충족 시 고품질 텍스트로 판단 return indicators >= 3 def get_ppl_calculation_mode(text_length): if text_length > 2000: return "ultra_fast" elif text_length > 1000: return "fast" else: return "accurate" def get_ppl_score(adjusted_ppl): # 3점 만점, 5단계 (20% 차등) if adjusted_ppl < 15: return 3.0 # 100% elif adjusted_ppl < 30: return 2.4 # 80% elif adjusted_ppl < 50: return 1.8 # 60% elif adjusted_ppl < 100: return 1.2 # 40% else: return 0.6 # 20% def get_rouge_score(final_rouge_score): # 3점 만점, 5단계 (20% 차등) if final_rouge_score >= 0.60: return 3.0 # 100% elif final_rouge_score >= 0.50: return 2.4 # 80% elif final_rouge_score >= 0.40: return 1.8 # 60% elif final_rouge_score >= 0.30: return 1.2 # 40% else: return 0.6 # 20% def get_bleu_score(bleu_score): # 2점 만점, 5단계 (20% 차등) if bleu_score >= 0.50: return 2.0 # 100% elif bleu_score >= 0.40: return 1.6 # 80% elif bleu_score >= 0.30: return 1.2 # 60% elif bleu_score >= 0.20: return 0.8 # 40% else: return 0.4 # 20% def cleanup_memory(): if torch.cuda.is_available(): torch.cuda.empty_cache() gc.collect() @app.route('/', methods=['GET']) def index(): all_results = session.get('all_results', {}) input_texts = session.get('input_texts', {}) return render_template('index.html', model_loaded=model_loaded, all_results=all_results, input_texts=input_texts) @app.route('/evaluate', methods=['POST']) def evaluate_text(): if 'all_results' not in session: session['all_results'] = {} if 'input_texts' not in session: session['input_texts'] = {} target_url = request.form.get('target_url') if target_url: session['all_results']['target_url'] = target_url metric = request.form.get('metric') results_to_store = {'metric': metric} try: if metric == 'perplexity': text = request.form.get('ppl_text', '').strip() session['input_texts']['ppl_text'] = text validation_result = validate_ppl_text(text) if not validation_result["valid"]: results_to_store['error'] = validation_result["message"] elif not model_loaded: results_to_store['error'] = "모델이 로딩되지 않았습니다." else: try: cleanup_memory() calc_mode = get_ppl_calculation_mode(len(text)) start_time = time.time() if calc_mode == "ultra_fast": ppl_result = calculate_perplexity_logic(text, max_tokens=256, use_sliding_window=False) elif calc_mode == "fast": ppl_result = calculate_perplexity_logic(text, max_tokens=384, use_sliding_window=False) else: ppl_result = calculate_perplexity_logic(text, max_tokens=512, use_sliding_window=True) calc_time = time.time() - start_time adjusted_ppl = ppl_result['adjusted_ppl'] results_to_store['score_value'] = adjusted_ppl results_to_store['score_display'] = f"{adjusted_ppl:.4f}" results_to_store['details'] = { 'base_ppl': f"{ppl_result['base_ppl']:.4f}", 'penalty_factor': f"{ppl_result['penalty_factor']:.4f}", 'token_count': ppl_result['token_count'], 'calc_time': f"{calc_time:.2f}s", 'calc_mode': calc_mode, 'is_structured': ppl_result.get('is_structured', False) } results_to_store['final_score'] = get_ppl_score(adjusted_ppl) cleanup_memory() except Exception as ppl_error: results_to_store['error'] = f"PPL 계산 중 오류: {ppl_error}" session['all_results']['perplexity'] = results_to_store elif metric == 'rouge': gen_text = request.form.get('rouge_generated', '').strip() ref_text = request.form.get('rouge_reference', '').strip() session['input_texts']['rouge_generated'] = gen_text session['input_texts']['rouge_reference'] = ref_text if not gen_text or not ref_text: results_to_store['error'] = "생성된 요약문과 참조 요약문을 모두 입력해주세요." else: scores = scorer.score(ref_text, gen_text) r1, r2, rL = scores['rouge1'].fmeasure, scores['rouge2'].fmeasure, scores['rougeL'].fmeasure weighted_avg = (r1 * 0.3 + r2 * 0.3 + rL * 0.4) len_gen = len(gen_text.split()); len_ref = len(ref_text.split()) length_ratio = len_gen / len_ref if len_ref > 0 else 0 if 0.8 <= length_ratio <= 1.2: length_penalty = 1.0 elif length_ratio < 0.5 or length_ratio > 2.0: length_penalty = 0.8 else: length_penalty = 0.9 final_rouge_score = weighted_avg * length_penalty results_to_store['score_value'] = final_rouge_score results_to_store['score_display'] = f"{final_rouge_score:.4f}" results_to_store['details'] = { 'rouge1': f"{r1:.4f}", 'rouge2': f"{r2:.4f}", 'rougeL': f"{rL:.4f}", 'weighted_avg': f"{weighted_avg:.4f}", 'length_penalty': f"{length_penalty:.2f}" } results_to_store['final_score'] = get_rouge_score(final_rouge_score) session['all_results']['rouge'] = results_to_store elif metric == 'bleu': gen_text = request.form.get('bleu_generated', '').strip() ref_text1 = request.form.get('bleu_reference1', '').strip() ref_text2 = request.form.get('bleu_reference2', '').strip() session['input_texts']['bleu_generated'] = gen_text session['input_texts']['bleu_reference1'] = ref_text1 session['input_texts']['bleu_reference2'] = ref_text2 if not gen_text or not ref_text1 or not ref_text2: results_to_store['error'] = "생성된 번역문과 2개의 참조 번역문을 모두 입력해주세요." else: try: # sacrebleu 버전별 호환성 처리 references = [[ref_text1, ref_text2]] # 참조 번역문을 리스트로 감싸기 # 새로운 API 시도 try: # sacrebleu 2.x 버전 bleu_score = bleu.sentence_score(gen_text, references).score / 100 except Exception: # 구버전 또는 다른 방식 시도 try: # corpus_score 사용 bleu_score = bleu.corpus_score([gen_text], [references]).score / 100 except Exception: # 가장 기본적인 방식 from sacrebleu import sentence_bleu bleu_score = sentence_bleu(gen_text, references).score / 100 results_to_store['score_value'] = bleu_score results_to_store['score_display'] = f"{bleu_score:.4f}" results_to_store['final_score'] = get_bleu_score(bleu_score) except Exception as bleu_error: # BLEU 계산 실패 시 대체 방법 try: # nltk BLEU 사용 from nltk.translate.bleu_score import sentence_bleu as nltk_bleu from nltk.translate.bleu_score import SmoothingFunction gen_tokens = gen_text.split() ref_tokens1 = ref_text1.split() ref_tokens2 = ref_text2.split() smoothing = SmoothingFunction().method1 bleu_score = nltk_bleu([ref_tokens1, ref_tokens2], gen_tokens, smoothing_function=smoothing) results_to_store['score_value'] = bleu_score results_to_store['score_display'] = f"{bleu_score:.4f}" results_to_store['final_score'] = get_bleu_score(bleu_score) except Exception as nltk_error: results_to_store['error'] = f"BLEU 계산 중 오류: {str(bleu_error)[:100]}" session['all_results']['bleu'] = results_to_store elif metric in ['mmlu', 'truthfulqa', 'drop', 'mbpp_humaneval']: generated_text = request.form.get(f'{metric}_generated', '') reference_text = request.form.get(f'{metric}_reference', '') grade = request.form.get(f'{metric}_grade', '') session['input_texts'][f'{metric}_generated'] = generated_text session['input_texts'][f'{metric}_reference'] = reference_text # 조정된 배점: 모두 3점 만점 max_score = 3 # 5단계 (20% 차등) 적용 score_map = { '수': 1.0, # 100% '우': 0.8, # 80% '미': 0.6, # 60% '양': 0.4, # 40% '가': 0.2 # 20% } if grade and grade in score_map: final_score = max_score * score_map[grade] results_to_store['grade'] = grade results_to_store['final_score'] = final_score else: results_to_store['grade'] = None results_to_store['final_score'] = 0 if not grade: results_to_store['error'] = "평가 등급을 선택해주세요." session['all_results'][metric] = results_to_store except Exception as e: results_to_store['error'] = f"계산 중 오류 발생: {e}" session['all_results'][metric] = results_to_store app.logger.error(f"평가 중 오류 - 메트릭: {metric}, 오류: {e}") session.modified = True return redirect(url_for('index', _anchor=metric)) @app.route('/report') def report(): all_results = session.get('all_results', {}) input_texts = session.get('input_texts', {}) try: target_url = all_results.get('target_url', 'N/A') total_score = sum(res.get('final_score', 0) for res in all_results.values() if isinstance(res, dict)) log_message = f"보고서 생성 - 대상: {target_url}, 총점: {total_score:.2f}/20" app.logger.info(log_message) except Exception as e: app.logger.error(f"로그 기록 중 오류 발생: {e}") return render_template('report.html', all_results=all_results, input_texts=input_texts) @app.route('/reset') def reset(): session.pop('all_results', None) session.pop('input_texts', None) cleanup_memory() return redirect(url_for('index')) @app.route('/memory_status') def memory_status(): status = {} if torch.cuda.is_available(): status['gpu_allocated'] = f"{torch.cuda.memory_allocated() / 1024**3:.2f} GB" status['gpu_reserved'] = f"{torch.cuda.memory_reserved() / 1024**3:.2f} GB" import psutil process = psutil.Process() status['ram_usage'] = f"{process.memory_info().rss / 1024**3:.2f} GB" return status if __name__ == '__main__': app.run(host='0.0.0.0', port=7860)