MarketSync / utils /parser_utils.py
hyeonjoo's picture
Initial project commit with LFS
9b1e3db
# utils/parser_utils.py
import re
import json
from typing import List, Dict, Any, Union
def extract_json_from_llm_response(response_text: str) -> Union[List[Dict[str, Any]], Dict[str, Any]]:
"""
LLM 응닡 ν…μŠ€νŠΈμ—μ„œ ```json ... ``` λ˜λŠ” [...] λ˜λŠ” {...} 블둝을
μ•ˆμ „ν•˜κ²Œ μΆ”μΆœν•˜κ³  νŒŒμ‹±ν•©λ‹ˆλ‹€.
μ‹€νŒ¨ μ‹œ ValueErrorλ₯Ό λ°œμƒμ‹œν‚΅λ‹ˆλ‹€.
"""
json_str = None
# 1. ```json [...] ``` λ§ˆν¬λ‹€μš΄ 블둝 검색 (κ°€μž₯ μš°μ„ )
# re.DOTALL (s) ν”Œλž˜κ·Έ: μ€„λ°”κΏˆ 문자λ₯Ό ν¬ν•¨ν•˜μ—¬ λ§€μΉ­
# re.MULTILINE (m) ν”Œλž˜κ·Έ: ^, $κ°€ 각 μ€„μ˜ μ‹œμž‘/끝에 λ§€μΉ­
json_match = re.search(
r'```json\s*([\s\S]*?)\s*```',
response_text,
re.DOTALL | re.IGNORECASE
)
if json_match:
json_str = json_match.group(1).strip()
else:
# 2. λ§ˆν¬λ‹€μš΄μ΄ μ—†λ‹€λ©΄, 첫 번째 { λ˜λŠ” [ λ₯Ό 찾음
first_bracket_match = re.search(r'[{|\[]', response_text)
if first_bracket_match:
start_index = first_bracket_match.start()
# 응닡이 리슀트([])둜 μ‹œμž‘ν•˜λŠ” 경우
if response_text[start_index] == '[':
list_match = re.search(r'(\[[\s\S]*\])', response_text[start_index:], re.DOTALL)
if list_match:
json_str = list_match.group(0)
# 응닡이 λ”•μ…”λ„ˆλ¦¬({})둜 μ‹œμž‘ν•˜λŠ” 경우
elif response_text[start_index] == '{':
dict_match = re.search(r'(\{[\s\S]*\})', response_text[start_index:], re.DOTALL)
if dict_match:
json_str = dict_match.group(0)
if json_str is None:
raise ValueError(f"μ‘λ‹΅μ—μ„œ JSON 블둝을 μ°Ύμ§€ λͺ»ν–ˆμŠ΅λ‹ˆλ‹€. (응닡 μ‹œμž‘: {response_text[:150]}...)")
try:
# (디버깅) μΆ”μΆœλœ λ¬Έμžμ—΄ λ‘œκΉ…
# print(f"--- [Parser DEBUG] Extracted JSON String: {json_str[:200]}... ---")
return json.loads(json_str)
except json.JSONDecodeError as e:
raise ValueError(f"JSON νŒŒμ‹±μ— μ‹€νŒ¨ν–ˆμŠ΅λ‹ˆλ‹€: {e}. (μΆ”μΆœλœ λ¬Έμžμ—΄: {json_str[:150]}...)")