KUGEHS/backend_estuary
0
1#!/usr/bin/env python3
2# -*- coding: utf-8 -*-
3# json_parser_fix.py - JSON 파싱 오류 해결 및 응답 정규화
4# LLM 응답의 다양한 형태를 안전하게 처리
5
6# UTF-8 인코딩 강제 설정 (독립 실행 시 대비)
7import sys
8import os
9if sys.platform.startswith('win'):
10 os.environ['PYTHONIOENCODING'] = 'utf-8'
11 if hasattr(sys.stdout, 'buffer') and sys.stdout.encoding.lower() != 'utf-8':
12 import io
13 sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
14 if hasattr(sys.stderr, 'buffer') and sys.stderr.encoding.lower() != 'utf-8':
15 import io
16 sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8')
17
18import json
19import re
20from typing import Dict, Any, Optional, Tuple, List
21import time
22import logging
23
24# ========================================
25# 로깅 설정
26# ========================================
27def setup_json_logger():
28 """JSON 파싱 전용 로거 설정"""
29 logger = logging.getLogger('json_parser')
30 logger.setLevel(logging.DEBUG)
31
32 if not logger.handlers:
33 # 콘솔 핸들러
34 handler = logging.StreamHandler()
35 formatter = logging.Formatter('[JSON_PARSER] %(asctime)s - %(levelname)s - %(message)s')
36 handler.setFormatter(formatter)
37 logger.addHandler(handler)
38
39 # 파일 핸들러 (디버깅용)
40 try:
41 os.makedirs('./logs', exist_ok=True)
42 file_handler = logging.FileHandler('./logs/json_parser_debug.log', encoding='utf-8')
43 file_handler.setLevel(logging.DEBUG)
44 file_handler.setFormatter(formatter)
45 logger.addHandler(file_handler)
46 except Exception as e:
47 print(f"파일 로깅 설정 실패: {e}")
48
49 return logger
50
51# 전역 로거 인스턴스
52json_logger = setup_json_logger()
53
54def create_emergency_fallback_result():
55 """
56 응급 폴백 결과 생성 함수
57 """
58 return {
59 "scores": {
60 "academic": 3.0,
61 "career": 3.0,
62 "community": 3.0,
63 "overall": 3.0
64 },
65 "strengths": ["LLM 응답 형식 불일치로 기본값 적용"],
66 "improvements": ["응답 구조 개선 필요"],
67 "detailed_feedback": "AI 응답의 형식이 예상과 달라 기본 분석 결과를 제공합니다. 보다 정확한 분석을 위해 다시 시도해주세요."
68 }
69
70# ========================================
71# 스키마 검증 및 데이터 유효성 검사
72# ========================================
73
74def validate_evaluation_schema(data: Dict[str, Any]) -> Tuple[bool, List[str]]:
75 """평가 결과 스키마 검증"""
76 errors = []
77
78 # 필수 필드 검증
79 required_fields = {
80 'scores': dict,
81 'strengths': list,
82 'improvements': list,
83 'detailed_feedback': str
84 }
85
86 for field, expected_type in required_fields.items():
87 if field not in data:
88 errors.append(f"필수 필드 누락: {field}")
89 elif not isinstance(data[field], expected_type):
90 errors.append(f"필드 타입 오류: {field} (예상: {expected_type.__name__}, 실제: {type(data[field]).__name__})")
91
92 # scores 세부 검증
93 if 'scores' in data and isinstance(data['scores'], dict):
94 required_score_fields = ['academic', 'career', 'community', 'overall']
95 for score_field in required_score_fields:
96 if score_field not in data['scores']:
97 errors.append(f"점수 필드 누락: scores.{score_field}")
98 else:
99 score_value = data['scores'][score_field]
100 if not isinstance(score_value, (int, float)):
101 errors.append(f"점수 타입 오류: scores.{score_field} (숫자가 아님)")
102 elif not (1.0 <= score_value <= 5.0):
103 errors.append(f"점수 범위 오류: scores.{score_field}={score_value} (1.0-5.0 범위 벗어남)")
104
105 # 리스트 필드 검증
106 for list_field in ['strengths', 'improvements']:
107 if list_field in data and isinstance(data[list_field], list):
108 if len(data[list_field]) == 0:
109 errors.append(f"빈 리스트: {list_field}")
110 elif not all(isinstance(item, str) for item in data[list_field]):
111 errors.append(f"리스트 요소 타입 오류: {list_field} (모든 요소가 문자열이어야 함)")
112
113 return len(errors) == 0, errors
114
115def auto_fix_data_types(data: Dict[str, Any]) -> Dict[str, Any]:
116 """데이터 타입 자동 수정"""
117 if not isinstance(data, dict):
118 return data
119
120 fixed_data = data.copy()
121
122 # scores 필드 수정
123 if 'scores' in fixed_data:
124 if not isinstance(fixed_data['scores'], dict):
125 fixed_data['scores'] = {}
126
127 score_fields = ['academic', 'career', 'community', 'overall']
128 for field in score_fields:
129 if field not in fixed_data['scores']:
130 fixed_data['scores'][field] = 3.0
131 else:
132 try:
133 value = float(fixed_data['scores'][field])
134 fixed_data['scores'][field] = max(1.0, min(5.0, value))
135 except (ValueError, TypeError):
136 fixed_data['scores'][field] = 3.0
137
138 # 리스트 필드 수정
139 for list_field in ['strengths', 'improvements']:
140 if list_field not in fixed_data:
141 fixed_data[list_field] = [f"기본 {list_field}"]
142 elif not isinstance(fixed_data[list_field], list):
143 fixed_data[list_field] = [str(fixed_data[list_field])]
144 elif len(fixed_data[list_field]) == 0:
145 fixed_data[list_field] = [f"기본 {list_field}"]
146 else:
147 # 모든 요소를 문자열로 변환
148 fixed_data[list_field] = [str(item) for item in fixed_data[list_field]]
149
150 # detailed_feedback 수정
151 if 'detailed_feedback' not in fixed_data:
152 fixed_data['detailed_feedback'] = "자동 생성된 피드백"
153 elif not isinstance(fixed_data['detailed_feedback'], str):
154 fixed_data['detailed_feedback'] = str(fixed_data['detailed_feedback'])
155
156 return fixed_data
157
158# ========================================
159# 한글 인코딩 안전 처리 함수
160# ========================================
161
162def safe_encode_korean(text: Any) -> str:
163 """한글이 포함된 텍스트를 안전하게 처리 (간단한 문자열 변환만)"""
164 if text is None:
165 return ""
166
167 try:
168 # 문자열로만 변환, 인코딩 조작 없음
169 return str(text)
170 except Exception:
171 return ""
172
173def safe_process_korean_dict(data: Dict[str, Any]) -> Dict[str, Any]:
174 """딕셔너리의 모든 값에 대해 안전한 문자열 변환만 수행"""
175 if not isinstance(data, dict):
176 return data
177
178 # 딕셔너리는 그대로 반환 (인코딩 변환하지 않음)
179 return data
180
181# ========================================
182# 1. JSON 정리 및 수정 함수들
183# ========================================
184
185def fix_specific_gemma_errors(text: str) -> str:
186 """
187 Gemma 모델의 특정 오류 패턴을 우선적으로 수정 - 정상 JSON은 그대로 유지
188 """
189 if not text or not isinstance(text, str):
190 return '{"scores": {"academic": 3.0, "career": 3.0, "community": 3.0, "overall": 3.0}, "strengths": ["기본값"], "improvements": ["기본값"], "detailed_feedback": "파싱 실패로 기본값 반환"}'
191
192 # 먼저 마크다운 코드 블록에서 JSON 추출 시도
193 try:
194 import json
195
196 # 강화된 마크다운 처리
197 markdown_json = extract_json_from_markdown_blocks(text)
198 if markdown_json:
199 # 추출된 JSON이 유효한지 확인
200 json.loads(markdown_json)
201 return markdown_json
202
203 # 기본 마크다운 제거 후 테스트
204 clean_text = text.strip()
205 if clean_text.startswith('```json'):
206 clean_text = clean_text[7:]
207 if clean_text.endswith('```'):
208 clean_text = clean_text[:-3]
209 clean_text = clean_text.strip()
210
211 # 정상적인 JSON이면 그대로 반환
212 json.loads(clean_text)
213 return clean_text
214 except:
215 pass
216
217 # 정상 JSON이 아닌 경우에만 수정 작업 수행
218
219 # 1. 초고도화 개별과목분석 패턴 감지 및 처리
220 if any(keyword in text for keyword in ["과목명", "과목분류", "학년별_성장", "역량_분석", "전공연관성"]):
221 json_logger.info("초고도화 개별과목분석 패턴 감지")
222 return fix_academic_enhanced_structure(text)
223
224 # 2. 기본 scores 구조 처리
225 if '"scores"' in text or 'scores' in text:
226 # 숫자 패턴 찾기
227 numbers = re.findall(r'\d+\.?\d*', text)
228 if len(numbers) >= 4:
229 try:
230 academic = float(numbers[0]) if numbers[0] else 3.0
231 career = float(numbers[1]) if numbers[1] else 3.0
232 community = float(numbers[2]) if numbers[2] else 3.0
233 overall = float(numbers[3]) if numbers[3] else 3.0
234
235 # 점수 범위 검증
236 academic = max(1.0, min(5.0, academic))
237 career = max(1.0, min(5.0, career))
238 community = max(1.0, min(5.0, community))
239 overall = max(1.0, min(5.0, overall))
240
241 return f'{{"scores": {{"academic": {academic}, "career": {career}, "community": {community}, "overall": {overall}}}, "strengths": ["분석에서 추출한 강점"], "improvements": ["분석에서 추출한 개선점"], "detailed_feedback": "Gemma 모델 응답을 기반으로 한 분석"}}'
242 except:
243 pass
244
245 # 3. 원본 텍스트 극단적 정리
246 # 모든 문제 패턴을 한 번에 제거
247 original_text = text
248
249 # A. 오류 패턴 완전 제거 - 특히 '\n "scores"' 패턴 처리
250 # 특별 처리: 정확히 '\n "scores"' 패턴
251 text = re.sub(r".*?'\\n\s*\"scores\"", '{"scores"', text)
252 text = re.sub(r".*?'\s*\\n\s*\"scores\"", '{"scores"', text)
253 text = re.sub(r".*?'\s*\n\s*\"scores\"", '{"scores"', text)
254 text = re.sub(r".*?\\n\s*\"scores\"", '{"scores"', text)
255 text = re.sub(r".*?\n\s*\"scores\"", '{"scores"', text)
256
257 # 매우 구체적인 패턴: '\n "scores"'
258 text = re.sub(r".*?'\\n\s+\"scores\"", '{"scores"', text)
259 text = re.sub(r".*?'\s*\n\s+\"scores\"", '{"scores"', text)
260
261 # B. 시작 부분이 { 가 아니면 강제로 추가
262 if not text.strip().startswith('{'):
263 if '"scores"' in text:
264 idx = text.find('"scores"')
265 text = '{' + text[idx:]
266 else:
267 # 완전 실패 시 기본 구조 반환
268 return '{"scores": {"academic": 3.0, "career": 3.0, "community": 3.0, "overall": 3.0}, "strengths": ["파싱 실패"], "improvements": ["파싱 실패"], "detailed_feedback": "JSON 파싱에 실패했습니다"}'
269
270 # C. 끝 부분 확인
271 if not text.rstrip().endswith('}'):
272 text += '}'
273
274 # D. 기본 키들이 없으면 추가
275 if '"strengths"' not in text:
276 text = text.replace('}', ', "strengths": ["기본 강점"]}')
277 if '"improvements"' not in text:
278 text = text.replace('}', ', "improvements": ["기본 개선점"]}')
279 if '"detailed_feedback"' not in text:
280 text = text.replace('}', ', "detailed_feedback": "기본 피드백"}')
281
282 return text
283
284def clean_json_string(text: str) -> str:
285 """
286 JSON 문자열 정리 - 파싱 전 사전 처리 (google/gemma-3n-e4b 대응 강화)
287 """
288 if not text or not isinstance(text, str):
289 return "{}"
290
291 # 0. Gemma 모델 특화 오류 우선 처리
292 text = fix_specific_gemma_errors(text)
293
294 # 1. 앞뒤 공백 제거
295 text = text.strip()
296
297 # 2. 마크다운 코드 블록 제거 - 강화된 처리
298 # 먼저 마크다운 블록에서 JSON 추출 시도
299 markdown_json = extract_json_from_markdown_blocks(text)
300 if markdown_json:
301 return markdown_json
302
303 # 기본 마크다운 제거
304 text = re.sub(r'```json\s*', '', text)
305 text = re.sub(r'```\s*$', '', text)
306 text = re.sub(r'```', '', text)
307
308 # 3. <think> 태그 제거 (o1 모델 대응)
309 text = re.sub(r'<think>.*?</think>', '', text, flags=re.DOTALL)
310
311 # 4. gemma-3n-e4b 모델 특정 패턴 수정 - 강화판
312
313 # A. 가장 문제가 되는 '\n "scores"' 패턴 우선 처리
314 text = re.sub(r"'\s*\\n\s*\"scores\"", '{"scores"', text)
315 text = re.sub(r"\\n\s*\"scores\"", '"scores"', text)
316 text = re.sub(r'\n\s*"scores"', '"scores"', text)
317 text = re.sub(r"'\s*\n\s*\"scores\"", '{"scores"', text)
318
319 # B. 일반적인 줄바꿈 패턴들
320 text = re.sub(r'\n\s*"([a-zA-Z_][a-zA-Z0-9_]*)"', r', "\1"', text)
321 text = re.sub(r',\s*\n\s*"([a-zA-Z_][a-zA-Z0-9_]*)"', r', "\1"', text)
322
323 # C. 시작 부분 문제 패턴 (아예 { 없이 시작하는 경우)
324 if not text.strip().startswith('{') and '"scores"' in text:
325 # scores로 시작하는 경우 { 추가
326 text = re.sub(r'^[^{]*"scores"', '{"scores"', text.strip())
327
328 # D. 백슬래시 이스케이프 문제
329 text = re.sub(r'\\"scores\\"', '"scores"', text)
330 text = re.sub(r'\\n\s*', ' ', text)
331 text = re.sub(r'\\t', ' ', text)
332
333 # 5. JSON 키-값 구조 정규화
334 # "key" \n : value -> "key": value
335 text = re.sub(r'"\s*\n\s*:', '":', text)
336
337 # 6. 설명 텍스트 제거 (JSON 앞뒤 불필요한 설명)
338 lines = text.split('\n')
339 json_lines = []
340 in_json = False
341 brace_count = 0
342
343 for line in lines:
344 stripped = line.strip()
345
346 # JSON 시작 감지
347 if stripped.startswith('{') and not in_json:
348 in_json = True
349 brace_count = stripped.count('{') - stripped.count('}')
350 json_lines.append(line)
351 elif in_json:
352 brace_count += stripped.count('{') - stripped.count('}')
353 json_lines.append(line)
354
355 # JSON 종료 감지
356 if brace_count == 0:
357 break
358
359 if json_lines:
360 text = '\n'.join(json_lines)
361
362 # 5. 잘못된 문자 수정
363 replacements = [
364 ('True', 'true'),
365 ('False', 'false'),
366 ('None', 'null'),
367 ("'", '"'), # 작은따옴표를 큰따옴표로
368 ('\\n', '\\\\n'), # 줄바꿈 이스케이프
369 ('\\t', '\\\\t'), # 탭 이스케이프
370 ]
371
372 for old, new in replacements:
373 text = text.replace(old, new)
374
375 # 6. 후행 쉼표 제거
376 text = re.sub(r',(\s*[}\]])', r'\1', text)
377
378 # 7. 중복 쉼표 제거
379 text = re.sub(r',\s*,', ',', text)
380
381 return text.strip()
382
383def fix_academic_enhanced_structure(text: str) -> str:
384 """
385 초고도화 개별과목분석용 JSON 구조 수정 - 2025-01-18 추가
386 """
387 if not text or not isinstance(text, str):
388 return '{"error": "빈 텍스트로 수정 불가"}'
389
390 json_logger.info("초고도화 개별과목분석 구조 수정 시작")
391
392 # 1. 기본 정리만 수행 (clean_json_string 호출하지 않음 - 재귀 방지)
393 text = text.strip()
394
395 # 마크다운 코드 블록 제거
396 text = re.sub(r'```json\s*', '', text)
397 text = re.sub(r'```\s*$', '', text)
398 text = re.sub(r'```', '', text)
399
400 # 2. 과목분석 특화 패턴 처리
401 # 과목명이 누락된 경우 추가
402 if "과목명" not in text and "과목분류" in text:
403 text = text.replace('"과목분류"', '"과목명": "분석대상과목", "과목분류"')
404
405 # 3. 필수 필드가 누락된 경우 기본 구조 추가
406 required_fields = {
407 "과목명": "분석대상과목",
408 "과목분류": "기타",
409 "총점": 0.0,
410 "학년별_성장": {
411 "1학년": "해당없음",
412 "2학년": "해당없음",
413 "3학년": "해당없음"
414 },
415 "핵심기록_원문": ["기록 없음"],
416 "역량_분석": {
417 "지식이해": "분석 없음",
418 "과정기능": "분석 없음",
419 "가치태도": "분석 없음"
420 },
421 "전공연관성": "분석 없음",
422 "성장궤적": "분석 없음",
423 "차별화요소": "분석 없음"
424 }
425
426 # 4. 기본 구조가 전혀 없는 경우 생성
427 if not any(field in text for field in ["과목명", "과목분류", "학년별_성장"]):
428 import json
429 return json.dumps(required_fields, ensure_ascii=False, indent=2)
430
431 # 5. 부분적으로 있는 경우 보완
432 try:
433 import json
434 # 기존 JSON 파싱 시도
435 parsed = json.loads(text)
436
437 # 누락된 필드 보완
438 for field, default_value in required_fields.items():
439 if field not in parsed:
440 parsed[field] = default_value
441
442 return json.dumps(parsed, ensure_ascii=False, indent=2)
443
444 except json.JSONDecodeError:
445 # 파싱 실패 시 기본 구조 반환
446 json_logger.warning("초고도화 구조 수정 실패, 기본 구조 반환")
447 import json
448 return json.dumps(required_fields, ensure_ascii=False, indent=2)
449
450def fix_common_json_errors(text: str) -> str:
451 """
452 일반적인 JSON 오류 수정
453 """
454 # 1. 키 따옴표 누락 수정
455 text = re.sub(r'(\w+)(\s*:)', r'"\1"\2', text)
456
457 # 2. 문자열 값 따옴표 확인
458 text = re.sub(r':\s*([^"\d\[\{][^,\n\}]*)', r': "\1"', text)
459
460 # 3. 숫자가 아닌 값에 따옴표 추가
461 text = re.sub(r':\s*([가-힣a-zA-Z][^",\n\}]*)', r': "\1"', text)
462
463 # 4. 빈 값 처리
464 text = re.sub(r':\s*,', ': "",', text)
465 text = re.sub(r':\s*}', ': ""}', text)
466
467 return text
468
469def extract_json_from_mixed_content(text: str) -> Optional[str]:
470 """
471 혼합 콘텐츠에서 JSON 부분만 추출 - 마크다운 코드 블록 지원 강화
472 """
473 if not text or not isinstance(text, str):
474 return None
475
476 # 1단계: 마크다운 코드 블록에서 JSON 추출 (가장 우선)
477 markdown_json = extract_json_from_markdown_blocks(text)
478 if markdown_json:
479 return markdown_json
480
481 # 2단계: 기존 JSON 패턴 찾기
482 json_patterns = [
483 r'\{[^{}]*(?:\{[^{}]*\}[^{}]*)*\}', # 중첩된 객체
484 r'\{.*?\}', # 단순 객체
485 ]
486
487 for pattern in json_patterns:
488 matches = re.findall(pattern, text, re.DOTALL)
489 for match in matches:
490 # 유효한 JSON인지 테스트
491 try:
492 json.loads(match)
493 return match
494 except:
495 continue
496
497 return None
498
499def extract_json_from_markdown_blocks(text: str) -> Optional[str]:
500 """
501 마크다운 코드 블록에서 JSON 추출 - Gemma 모델 특화
502 """
503 if not text:
504 return None
505
506 # 다양한 마크다운 코드 블록 패턴
507 markdown_patterns = [
508 # 표준 JSON 코드 블록
509 r'```json\s*\n(.*?)\n```',
510 r'```json\s*(.*?)```',
511
512 # 백틱 없이 json 키워드만 있는 경우
513 r'json\s*\n(.*?)(?=\n\*\*|$)',
514 r'json\s*(.*?)(?=\n\*\*|$)',
515
516 # 백틱만 있는 경우 (json 키워드 없음)
517 r'```\s*\n(\{.*?\})\s*\n```',
518 r'```(\{.*?\})```',
519
520 # Gemma 특화: ## 제목 다음에 오는 JSON
521 r'## [^#\n]*JSON[^#\n]*\n\s*```json\s*\n(.*?)\n```',
522 r'## [^#\n]*JSON[^#\n]*\n\s*```\s*\n(.*?)\n```',
523
524 # 더 관대한 패턴: JSON이라는 단어 근처의 코드 블록
525 r'JSON[^`]*```json\s*\n(.*?)\n```',
526 r'JSON[^`]*```\s*\n(.*?)\n```',
527 ]
528
529 for pattern in markdown_patterns:
530 try:
531 matches = re.findall(pattern, text, re.DOTALL | re.IGNORECASE)
532 for match in matches:
533 # 매치된 내용 정리
534 cleaned_match = match.strip()
535
536 # JSON 시작/끝 확인
537 if cleaned_match.startswith('{') and cleaned_match.endswith('}'):
538 try:
539 # JSON 유효성 검증
540 json.loads(cleaned_match)
541 json_logger.info(f"마크다운에서 JSON 추출 성공: 패턴={pattern[:20]}...")
542 return cleaned_match
543 except json.JSONDecodeError:
544 # JSON이 깨졌을 수도 있으니 정리 후 재시도
545 try:
546 fixed_json = clean_json_string(cleaned_match)
547 json.loads(fixed_json)
548 json_logger.info(f"마크다운에서 JSON 추출 후 수정 성공: 패턴={pattern[:20]}...")
549 return fixed_json
550 except:
551 continue
552 except re.error as e:
553 json_logger.warning(f"정규표현식 오류 in markdown extraction: {e}")
554 continue
555
556 json_logger.warning("마크다운 블록에서 JSON 추출 실패")
557 return None
558
559# ========================================
560# 2. 다단계 JSON 파싱 함수
561# ========================================
562
563def safe_json_parse(text: str) -> Tuple[bool, Optional[Dict], str]:
564 """
565 안전한 JSON 파싱 - 개선된 다단계 시도 + 스키마 검증
566
567 Returns:
568 (성공여부, 파싱된_데이터, 오류_메시지)
569 """
570 if not text or not isinstance(text, str):
571 json_logger.warning("빈 텍스트 또는 잘못된 타입 입력")
572 return False, None, "빈 텍스트 또는 잘못된 타입"
573
574 json_logger.info(f"JSON 파싱 시작: 텍스트 길이={len(text)}")
575
576 # 🔍 디버그: 실제 LLM 응답 전체 로깅 (처음 1000자와 마지막 1000자)
577 if len(text) > 2000:
578 json_logger.debug(f"LLM 응답 시작 1000자: {text[:1000]}")
579 json_logger.debug(f"LLM 응답 끝 1000자: {text[-1000:]}")
580 else:
581 json_logger.debug(f"LLM 전체 응답: {text}")
582
583 parsing_attempts = []
584
585 # 1단계: 원본 그대로 파싱 시도
586 try:
587 result = json.loads(text)
588 json_logger.info("1단계: 원본 파싱 성공")
589
590 # 스키마 검증
591 is_valid, errors = validate_evaluation_schema(result)
592 if is_valid:
593 safe_result = safe_process_korean_dict(result)
594 json_logger.info("스키마 검증 통과")
595 return True, safe_result, ""
596 else:
597 json_logger.warning(f"스키마 검증 실패: {errors}")
598 # 자동 수정 시도
599 fixed_result = auto_fix_data_types(result)
600 is_valid_after_fix, _ = validate_evaluation_schema(fixed_result)
601 if is_valid_after_fix:
602 safe_result = safe_process_korean_dict(fixed_result)
603 json_logger.info("자동 수정 후 스키마 검증 통과")
604 return True, safe_result, "자동 수정됨"
605 else:
606 parsing_attempts.append(f"스키마 검증 실패: {', '.join(errors[:3])}")
607
608 except json.JSONDecodeError as e:
609 json_logger.warning(f"1단계 파싱 실패: {str(e)}")
610 parsing_attempts.append(f"원본 파싱 실패: {str(e)}")
611
612 # 2단계: 기본 정리 후 파싱
613 try:
614 cleaned = clean_json_string(text)
615 result = json.loads(cleaned)
616 json_logger.info("2단계: 정리 후 파싱 성공")
617
618 # 스키마 검증 및 자동 수정
619 is_valid, errors = validate_evaluation_schema(result)
620 if not is_valid:
621 result = auto_fix_data_types(result)
622
623 safe_result = safe_process_korean_dict(result)
624 return True, safe_result, "정리 후 성공"
625
626 except json.JSONDecodeError as e:
627 json_logger.warning(f"2단계 파싱 실패: {str(e)}")
628 parsing_attempts.append(f"정리 후 파싱 실패: {str(e)}")
629
630 # 3단계: Gemma 특화 오류 수정
631 try:
632 gemma_fixed = fix_specific_gemma_errors(text)
633 result = json.loads(gemma_fixed)
634 json_logger.info("3단계: Gemma 특화 수정 성공")
635
636 # 스키마 검증 및 자동 수정
637 is_valid, errors = validate_evaluation_schema(result)
638 if not is_valid:
639 result = auto_fix_data_types(result)
640
641 safe_result = safe_process_korean_dict(result)
642 return True, safe_result, "Gemma 특화 수정 성공"
643
644 except json.JSONDecodeError as e:
645 json_logger.warning(f"3단계 파싱 실패: {str(e)}")
646 parsing_attempts.append(f"Gemma 특화 수정 실패: {str(e)}")
647
648 # 4단계: JSON 부분만 추출하여 파싱
649 try:
650 extracted = extract_json_from_mixed_content(text)
651 if extracted:
652 result = json.loads(extracted)
653 json_logger.info("4단계: JSON 추출 후 파싱 성공")
654
655 # 스키마 검증 및 자동 수정
656 is_valid, errors = validate_evaluation_schema(result)
657 if not is_valid:
658 result = auto_fix_data_types(result)
659
660 safe_result = safe_process_korean_dict(result)
661 return True, safe_result, "JSON 추출 성공"
662 else:
663 parsing_attempts.append("JSON 부분 추출 실패")
664 except json.JSONDecodeError as e:
665 json_logger.warning(f"4단계 파싱 실패: {str(e)}")
666 parsing_attempts.append(f"추출 후 파싱 실패: {str(e)}")
667
668 # 5단계: 부분 복구 시도
669 try:
670 partial_result = attempt_partial_recovery(text)
671 if partial_result:
672 json_logger.info("5단계: 부분 복구 성공")
673
674 # 스키마 검증 및 자동 수정
675 is_valid, errors = validate_evaluation_schema(partial_result)
676 if not is_valid:
677 partial_result = auto_fix_data_types(partial_result)
678
679 safe_result = safe_process_korean_dict(partial_result)
680 return True, safe_result, "부분 복구 성공"
681 except Exception as e:
682 json_logger.warning(f"5단계 복구 실패: {str(e)}")
683 parsing_attempts.append(f"부분 복구 실패: {str(e)}")
684
685 # 모든 시도 실패 - 응급 폴백 생성
686 json_logger.error("모든 파싱 시도 실패, 응급 폴백 생성")
687 emergency_result = create_emergency_fallback_result()
688 error_summary = " | ".join(parsing_attempts)
689
690 return True, emergency_result, f"응급 폴백 사용: {error_summary}"
691
692def estimate_scores_from_text(text: str) -> Dict[str, int]:
693 """
694 텍스트 내용 분석을 통한 지능적 점수 추정
695 """
696 estimated_scores = {}
697
698 # 평가 항목별 키워드 패턴 정의 (긍정/부정 키워드)
699 evaluation_keywords = {
700 '적극성': {
701 'positive': ['적극적', '주도적', '자발적', '능동적', '참여', '발표', '질문', '의욕', '열정', '솔선수범'],
702 'negative': ['소극적', '수동적', '참여부족', '의욕부족', '무관심']
703 },
704 '탐구정신': {
705 'positive': ['탐구', '연구', '실험', '분석', '창의', '혁신', '발견', '궁금', '호기심', '깊이'],
706 'negative': ['탐구부족', '단순', '피상적', '호기심부족']
707 },
708 '전공진로탐색': {
709 'positive': ['진로', '전공', '관련', '계획', '목표', '비전', '꿈', '미래', '준비', '탐색'],
710 'negative': ['진로미정', '계획부족', '목표부재']
711 },
712 '협력성': {
713 'positive': ['협력', '팀워크', '협동', '소통', '배려', '존중', '화합', '조율', '공동', '함께'],
714 'negative': ['갈등', '독단적', '비협조적', '소통부족']
715 },
716 '기록의충실성': {
717 'positive': ['상세', '구체적', '충실', '완성도', '꼼꼼', '정확', '체계적'],
718 'negative': ['부족', '미흡', '단편적', '불충분']
719 },
720 '학업역량': {
721 'positive': ['성취', '향상', '우수', '뛰어남', '성장', '발전', '능력', '실력', '노력'],
722 'negative': ['부진', '저조', '미흡', '부족']
723 },
724 '진로역량': {
725 'positive': ['진로탐색', '체험', '전문성', '역량강화', '준비', '계획'],
726 'negative': ['진로고민', '방향성부족', '준비부족']
727 },
728 '공동체역량': {
729 'positive': ['봉사', '나눔', '배려', '공동체', '사회참여', '리더십', '책임감'],
730 'negative': ['개인주의', '참여부족', '무관심']
731 }
732 }
733
734 # 점수 범위 키워드 (명시적 점수 표현)
735 score_indicators = {
736 5: ['매우우수', '탁월', '뛰어남', '월등', '최고', '완벽', '뛰어난', '우수함'],
737 4: ['우수', '좋음', '양호', '높음', '상당', '잘함', '좋은'],
738 3: ['보통', '평균', '적당', '기본', '일반적', '표준'],
739 2: ['미흡', '부족', '개선필요', '낮음', '아쉬움'],
740 1: ['매우부족', '부진', '저조', '미달', '현저히부족']
741 }
742
743 # 각 평가 항목별 점수 추정
744 for item_name, keywords in evaluation_keywords.items():
745 # 1. 명시적 점수 확인
746 explicit_score = extract_explicit_score(text, item_name)
747 if explicit_score:
748 estimated_scores[item_name] = explicit_score
749 continue
750
751 # 2. 키워드 기반 점수 추정
752 positive_count = sum(1 for keyword in keywords['positive'] if keyword in text)
753 negative_count = sum(1 for keyword in keywords['negative'] if keyword in text)
754
755 # 3. 점수 범위 키워드 확인
756 range_score = None
757 for score, indicators in score_indicators.items():
758 if any(indicator in text for indicator in indicators):
759 range_score = score
760 break
761
762 # 4. 최종 점수 결정
763 if range_score:
764 estimated_scores[item_name] = range_score
765 elif positive_count > negative_count:
766 if positive_count >= 3:
767 estimated_scores[item_name] = 4
768 elif positive_count >= 2:
769 estimated_scores[item_name] = 4
770 else:
771 estimated_scores[item_name] = 3
772 elif negative_count > positive_count:
773 if negative_count >= 2:
774 estimated_scores[item_name] = 2
775 else:
776 estimated_scores[item_name] = 3
777 else:
778 # 키워드가 없거나 동점인 경우 텍스트 길이로 판단
779 text_length = len(text)
780 if text_length > 1000: # 충분한 내용
781 estimated_scores[item_name] = 3
782 elif text_length > 500: # 보통 내용
783 estimated_scores[item_name] = 3
784 else: # 내용 부족
785 estimated_scores[item_name] = 2
786
787 return estimated_scores
788
789def extract_explicit_score(text: str, item_name: str) -> Optional[int]:
790 """
791 텍스트에서 특정 항목의 명시적 점수 추출
792 """
793 # 다양한 점수 표현 패턴
794 patterns = [
795 f'{item_name}[:\s]*([1-5])점?',
796 f'{item_name}[:\s]*점수[:\s]*([1-5])',
797 f'{item_name}[:\s]*평가[:\s]*([1-5])',
798 f'{item_name}.*?([1-5])점',
799 f'{item_name}.*?([1-5])/5'
800 ]
801
802 for pattern in patterns:
803 matches = re.findall(pattern, text, re.IGNORECASE)
804 if matches:
805 try:
806 score = int(matches[0])
807 if 1 <= score <= 5:
808 return score
809 except ValueError:
810 continue
811
812 return None
813
814def attempt_partial_recovery(text: str) -> Optional[Dict]:
815 """
816 부분적 JSON 복구 시도 (google/gemma-3n-e4b 대응 강화) - 지능적 추정 통합
817 """
818 # 지능적 점수 추정
819 estimated_scores = estimate_scores_from_text(text)
820
821 # 기본 구조 생성 (추정 점수 적용)
822 base_structure = {
823 "scores": {
824 "academic": estimated_scores.get('학업역량', 3.0),
825 "career": estimated_scores.get('진로역량', 3.0),
826 "community": estimated_scores.get('공동체역량', 3.0),
827 "overall": sum(estimated_scores.values()) / len(estimated_scores) if estimated_scores else 3.0
828 },
829 "evaluation_results": {
830 "세부능력및특기사항": {},
831 "창의적체험활동": {}
832 },
833 "recommended_majors": "분석 실패로 인한 추정값",
834 "overall_feedback": "JSON 파싱 오류로 인한 텍스트 분석 기반 응답",
835 "intelligent_recovery": True
836 }
837
838 # gemma-3n-e4b 모델의 특정 점수 패턴 찾기 - 더 관대한 패턴
839 score_patterns = [
840 # 표준 scores 구조 (더 관대하게)
841 r'academic["\s:]*(\d+\.?\d*)',
842 r'career["\s:]*(\d+\.?\d*)',
843 r'community["\s:]*(\d+\.?\d*)',
844 r'overall["\s:]*(\d+\.?\d*)',
845
846 # JSON 형태에서 직접 추출
847 r'"academic"\s*:\s*(\d+\.?\d*)',
848 r'"career"\s*:\s*(\d+\.?\d*)',
849 r'"community"\s*:\s*(\d+\.?\d*)',
850 r'"overall"\s*:\s*(\d+\.?\d*)',
851
852 # 숫자만 있는 패턴 (순서대로 academic, career, community, overall)
853 r'(\d+\.?\d*)\s*,\s*(\d+\.?\d*)\s*,\s*(\d+\.?\d*)\s*,\s*(\d+\.?\d*)',
854
855 # 한글 패턴
856 r'학업[역량]*[:\-\s]*(\d+\.?\d*)',
857 r'진로[역량]*[:\-\s]*(\d+\.?\d*)',
858 r'공동체[역량]*[:\-\s]*(\d+\.?\d*)',
859 r'종합[평가]*[:\-\s]*(\d+\.?\d*)',
860 ]
861
862 # 점수 복구 시도
863 found_scores = {}
864 score_keys = ["academic", "career", "community", "overall"]
865 korean_keys = ["학업역량", "진로역량", "공동체역량", "종합평가"]
866
867 for i, pattern in enumerate(score_patterns):
868 matches = re.findall(pattern, text, re.IGNORECASE)
869 if matches:
870 for match in matches:
871 try:
872 score_val = float(match)
873 if 0.0 <= score_val <= 5.0:
874 if i < 4: # 표준 scores 구조
875 key = score_keys[i % 4]
876 elif i < 8: # 개별 점수
877 key = score_keys[i - 4]
878 else: # 한글 패턴
879 key = score_keys[(i - 8) % 4]
880
881 found_scores[key] = score_val
882 break # 첫 번째 매치만 사용
883 except ValueError:
884 continue
885
886 # 복구된 점수로 구조 업데이트
887 if found_scores:
888 for key, value in found_scores.items():
889 if key in base_structure["scores"]:
890 base_structure["scores"][key] = value
891
892 # 기본값이라도 유효한 구조 반환
893 return base_structure
894
895# ========================================
896# 3. LLM 응답 처리 메인 함수
897# ========================================
898
899def process_llm_evaluation_response(response_content: str) -> Dict:
900 """
901 LLM 평가 응답 처리 - 메인 엔트리 포인트
902
903 Args:
904 response_content: LLM에서 받은 원본 응답 텍스트
905
906 Returns:
907 처리된 평가 결과 딕셔너리
908 """
909 if not response_content:
910 return create_error_response("빈 응답")
911
912 # JSON 파싱 시도
913 success, parsed_data, error_msg = safe_json_parse(response_content)
914
915 if success and parsed_data:
916 # 파싱 성공 - 데이터 검증 및 정규화
917 return validate_and_normalize_evaluation(parsed_data)
918 else:
919 # 파싱 실패 - 텍스트 분석으로 폴백
920 return fallback_text_analysis(response_content, error_msg)
921
922def validate_and_normalize_evaluation(data: Dict) -> Dict:
923 """
924 파싱된 평가 데이터 검증 및 정규화
925 """
926 try:
927 # 기본 구조 확인
928 if "evaluation_results" not in data:
929 return create_error_response("evaluation_results 키 누락")
930
931 evaluation_results = data["evaluation_results"]
932
933 # 필수 섹션 확인
934 required_sections = ["세부능력및특기사항", "창의적체험활동"]
935 for section in required_sections:
936 if section not in evaluation_results:
937 evaluation_results[section] = {}
938
939 # 필수 항목 확인 및 기본값 설정
940 required_items = {
941 "세부능력및특기사항": ["적극성", "탐구정신", "전공진로탐색", "협력성"],
942 "창의적체험활동": ["기록의충실성", "학업역량", "진로역량", "공동체역량"]
943 }
944
945 total_score = 0
946 total_count = 0
947
948 for section, items in required_items.items():
949 section_data = evaluation_results[section]
950
951 for item in items:
952 if item not in section_data:
953 section_data[item] = {
954 "score": 3,
955 "reason": f"{item} 데이터 누락으로 기본값 적용"
956 }
957
958 item_data = section_data[item]
959
960 # 점수 정규화
961 if "score" in item_data:
962 score = item_data["score"]
963 if isinstance(score, (int, float)) and 1 <= score <= 5:
964 total_score += score
965 total_count += 1
966 else:
967 item_data["score"] = 3
968 total_score += 3
969 total_count += 1
970 else:
971 item_data["score"] = 3
972 total_score += 3
973 total_count += 1
974
975 # 평가 근거 확인
976 if "reason" not in item_data or not item_data["reason"]:
977 item_data["reason"] = f"{item} 평가 근거 없음"
978
979 # 평균 점수 계산
980 average_score = round(total_score / total_count, 1) if total_count > 0 else 3.0
981
982 # 결과 구조 정규화
983 normalized_result = {
984 "success": True,
985 "evaluation_results": evaluation_results,
986 "recommended_majors": data.get("recommended_majors", "전공 추천 정보 없음"),
987 "overall_feedback": data.get("overall_feedback", "종합 평가 의견 없음"),
988 "average_score": average_score,
989 "validation_status": "정규화 완료"
990 }
991
992 return normalized_result
993
994 except Exception as e:
995 return create_error_response(f"데이터 검증 중 오류: {str(e)}")
996
997def fallback_text_analysis(text: str, error_msg: str) -> Dict:
998 """
999 JSON 파싱 실패 시 텍스트 분석으로 폴백 - 지능적 점수 추정
1000 """
1001 try:
1002 # 지능적 기본 점수 추정
1003 estimated_scores = estimate_scores_from_text(text)
1004
1005 # 기본 결과 구조 생성 (추정 점수 적용)
1006 fallback_result = {
1007 "success": False,
1008 "evaluation_results": {
1009 "세부능력및특기사항": {
1010 "적극성": {"score": estimated_scores.get('적극성', 3), "reason": f"텍스트 분석 기반 추정 점수 (파싱 실패로 인한 폴백)"},
1011 "탐구정신": {"score": estimated_scores.get('탐구정신', 3), "reason": "텍스트 분석 기반 추정 점수 (파싱 실패로 인한 폴백)"},
1012 "전공진로탐색": {"score": estimated_scores.get('전공진로탐색', 3), "reason": "텍스트 분석 기반 추정 점수 (파싱 실패로 인한 폴백)"},
1013 "협력성": {"score": estimated_scores.get('협력성', 3), "reason": "텍스트 분석 기반 추정 점수 (파싱 실패로 인한 폴백)"}
1014 },
1015 "창의적체험활동": {
1016 "기록의충실성": {"score": estimated_scores.get('기록의충실성', 3), "reason": "텍스트 분석 기반 추정 점수 (파싱 실패로 인한 폴백)"},
1017 "학업역량": {"score": estimated_scores.get('학업역량', 3), "reason": "텍스트 분석 기반 추정 점수 (파싱 실패로 인한 폴백)"},
1018 "진로역량": {"score": estimated_scores.get('진로역량', 3), "reason": "텍스트 분석 기반 추정 점수 (파싱 실패로 인한 폴백)"},
1019 "공동체역량": {"score": estimated_scores.get('공동체역량', 3), "reason": "텍스트 분석 기반 추정 점수 (파싱 실패로 인한 폴백)"}
1020 }
1021 },
1022 "recommended_majors": "JSON 파싱 실패로 추천 불가",
1023 "overall_feedback": f"JSON 파싱 오류 발생하여 텍스트 분석으로 대체 분석 수행: {error_msg}",
1024 "average_score": sum(estimated_scores.values()) / len(estimated_scores) if estimated_scores else 3.0,
1025 "parsing_error": error_msg,
1026 "fallback_mode": True,
1027 "intelligent_estimation": True
1028 }
1029
1030 # 텍스트에서 점수 정보 추출 시도
1031 score_extraction_patterns = [
1032 r'적극성[:\s]*(\d+)점?',
1033 r'탐구정신[:\s]*(\d+)점?',
1034 r'전공진로탐색[:\s]*(\d+)점?',
1035 r'협력성[:\s]*(\d+)점?',
1036 r'기록의충실성[:\s]*(\d+)점?',
1037 r'학업역량[:\s]*(\d+)점?',
1038 r'진로역량[:\s]*(\d+)점?',
1039 r'공동체역량[:\s]*(\d+)점?'
1040 ]
1041
1042 item_names = [
1043 '적극성', '탐구정신', '전공진로탐색', '협력성',
1044 '기록의충실성', '학업역량', '진로역량', '공동체역량'
1045 ]
1046
1047 sections = {
1048 '적극성': '세부능력및특기사항',
1049 '탐구정신': '세부능력및특기사항',
1050 '전공진로탐색': '세부능력및특기사항',
1051 '협력성': '세부능력및특기사항',
1052 '기록의충실성': '창의적체험활동',
1053 '학업역량': '창의적체험활동',
1054 '진로역량': '창의적체험활동',
1055 '공동체역량': '창의적체험활동'
1056 }
1057
1058 # 점수 추출 및 적용
1059 extracted_scores = []
1060 for i, pattern in enumerate(score_extraction_patterns):
1061 matches = re.findall(pattern, text, re.IGNORECASE)
1062 if matches:
1063 try:
1064 score = int(matches[0])
1065 if 1 <= score <= 5:
1066 item_name = item_names[i]
1067 section_name = sections[item_name]
1068 fallback_result["evaluation_results"][section_name][item_name]["score"] = score
1069 fallback_result["evaluation_results"][section_name][item_name]["reason"] = f"텍스트에서 추출된 점수: {score}점"
1070 extracted_scores.append(score)
1071 except (ValueError, IndexError):
1072 continue
1073
1074 # 평균 점수 재계산
1075 if extracted_scores:
1076 avg_extracted = sum(extracted_scores) / len(extracted_scores)
1077 # 추출된 점수와 기본값의 가중 평균
1078 total_items = 8
1079 extracted_count = len(extracted_scores)
1080 default_count = total_items - extracted_count
1081
1082 total_score = sum(extracted_scores) + (default_count * 3)
1083 fallback_result["average_score"] = round(total_score / total_items, 1)
1084
1085 # 전공 추천 추출 시도
1086 major_patterns = [
1087 r'추천.*?전공[:\s]*([가-힣\s,]+)',
1088 r'전공.*?추천[:\s]*([가-힣\s,]+)',
1089 r'적합.*?분야[:\s]*([가-힣\s,]+)'
1090 ]
1091
1092 for pattern in major_patterns:
1093 matches = re.findall(pattern, text, re.IGNORECASE)
1094 if matches:
1095 major_text = matches[0].strip()
1096 if len(major_text) > 3: # 의미있는 내용인 경우
1097 fallback_result["recommended_majors"] = major_text
1098 break
1099
1100 # 전체 피드백 추출 시도
1101 feedback_patterns = [
1102 r'종합.*?평가[:\s]*([^\.]{50,200})',
1103 r'전체.*?의견[:\s]*([^\.]{50,200})',
1104 r'종합.*?피드백[:\s]*([^\.]{50,200})'
1105 ]
1106
1107 for pattern in feedback_patterns:
1108 matches = re.findall(pattern, text, re.IGNORECASE | re.DOTALL)
1109 if matches:
1110 feedback_text = matches[0].strip()
1111 if len(feedback_text) > 20:
1112 fallback_result["overall_feedback"] = feedback_text
1113 break
1114
1115 return fallback_result
1116
1117 except Exception as e:
1118 return create_error_response(f"폴백 분석 중 오류: {str(e)}")
1119
1120def create_error_response(error_message: str) -> Dict:
1121 """
1122 오류 응답 생성
1123 """
1124 return {
1125 "success": False,
1126 "error": error_message,
1127 "evaluation_results": {
1128 "세부능력및특기사항": {
1129 "적극성": {"score": 3, "reason": f"오류로 인한 기본값: {error_message}"},
1130 "탐구정신": {"score": 3, "reason": "오류로 인한 기본값"},
1131 "전공진로탐색": {"score": 3, "reason": "오류로 인한 기본값"},
1132 "협력성": {"score": 3, "reason": "오류로 인한 기본값"}
1133 },
1134 "창의적체험활동": {
1135 "기록의충실성": {"score": 3, "reason": "오류로 인한 기본값"},
1136 "학업역량": {"score": 3, "reason": "오류로 인한 기본값"},
1137 "진로역량": {"score": 3, "reason": "오류로 인한 기본값"},
1138 "공동체역량": {"score": 3, "reason": "오류로 인한 기본값"}
1139 }
1140 },
1141 "recommended_majors": "오류로 인한 추천 불가",
1142 "overall_feedback": f"분석 중 오류가 발생했습니다: {error_message}",
1143 "average_score": 3.0
1144 }
1145
1146# ========================================
1147# 4. 응답 품질 검증 함수들
1148# ========================================
1149
1150def check_response_quality(data: Dict) -> Dict:
1151 """
1152 응답 품질 검증 및 품질 점수 계산
1153 """
1154 quality_score = 0
1155 max_score = 100
1156 issues = []
1157
1158 # 1. 기본 구조 확인 (20점)
1159 if "evaluation_results" in data:
1160 quality_score += 20
1161 else:
1162 issues.append("evaluation_results 구조 누락")
1163
1164 # 2. 필수 항목 완성도 확인 (40점)
1165 required_items = 8 # 총 8개 평가 항목
1166 completed_items = 0
1167
1168 if "evaluation_results" in data:
1169 for section in ["세부능력및특기사항", "창의적체험활동"]:
1170 if section in data["evaluation_results"]:
1171 section_data = data["evaluation_results"][section]
1172 for item_name, item_data in section_data.items():
1173 if isinstance(item_data, dict) and "score" in item_data and "reason" in item_data:
1174 if isinstance(item_data["score"], (int, float)) and 1 <= item_data["score"] <= 5:
1175 if len(str(item_data["reason"])) >= 10: # 최소 10자 이상의 근거
1176 completed_items += 1
1177
1178 quality_score += (completed_items / required_items) * 40
1179
1180 # 3. 평가 근거 품질 확인 (20점)
1181 reason_quality = 0
1182 total_reasons = 0
1183
1184 if "evaluation_results" in data:
1185 for section in data["evaluation_results"].values():
1186 if isinstance(section, dict):
1187 for item_data in section.values():
1188 if isinstance(item_data, dict) and "reason" in item_data:
1189 reason_text = str(item_data["reason"])
1190 total_reasons += 1
1191
1192 # 근거 품질 평가
1193 if len(reason_text) >= 50: # 충분한 길이
1194 reason_quality += 1
1195 elif len(reason_text) >= 20: # 최소 길이
1196 reason_quality += 0.5
1197
1198 if total_reasons > 0:
1199 quality_score += (reason_quality / total_reasons) * 20
1200