LLM을 활용하는 서비스를 개발하는 건 좋지만, LLM이 준 답변을 어떻게 코드에서 사용하면 좋을까?
모아보카의 서비스를 위해서는 채점 결과를 데이터베이스에 저장해야 하고, 점수에 따라 다른 UI를 보여줘야 하며, 피드백 내용에 기반한 새로운 문제 출제를 고려하는 등 여러 대응이 필요하다. 그리고, 이 모든 작업을 위해선 LLM의 응답이 예측 가능한 형식을 따라야 한다.
따라서, LLM 응답을 안정적으로 파싱하기 위해 필요한 것은 JSON 구조화이고, 이것을 처리하는 법을 공부할 필요성이 있다. 단순히 프롬프트에서 "JSON으로 응답해줘"라고 보내는 것만으로는 부족하다. 자연어 모델은 유연한 응답이 장점이지만, 그만큼 다양한 답변이 돌아올 수 있고, json.loads(response)를 실행하면 에러가 발생하는 경우도 종종(아니 제법 자주) 생긴다. 마크다운 코드 블록으로 답변을 준다든가, 마지막 항목 뒤에 쉼표를 붙인다든가, 큰따옴표 대신 작은 따옴표를 사용해 문법을 틀린다든가... 어느 쪽이든 표준 JSON에는 허용되지 않는 문법들이다. 심지어 필드의 이름이 바뀌거나 누락되는 경우도 생긴다.
OpenAI API의 1) JSON 모드 활용법, 2) 프롬프트로 출력 형식을 제어하는 기법, 3) 파싱 실패 시 대응법에 대해 정리해 보겠다.
OpenAI JSON모드란?
response_format 파라미터를 통한 옵션으로 {"type": "json_object"}로 설정하면 출력이 JSON이 된다.
response = client.chat.completions.create( model="gpt-4-turbo-preview", response_format={"type": "json_object"}, messages=[...] )
JSON 모드 사용 시 시스템 프롬프트나 사용자 프롬프트에 무조건 "JSON으로 응답하라"는 지시가 포함되어야 한다! 더하여, 스트리밍 응답을 사용하는 경우, 응답이 끝나기 전까진 부분적인 JSON만 받게 된다. 따라서, 파싱은 전체 응답을 받은 후에 해야 한다. (고려할 부분이 많다.)
설명 텍스트나 마크다운 코드 블록, 문법 오류 문제는 이 선에서 해결 가능하다.
BUT 원하는 스키마는 보장하지 않는다!
프롬프트에서 스키마 강제하는 방법
JSON 모드만으로는 스키마를 보장할 수 없으므로 프롬프트 엔지니어링을 통해 원하는 형식으로 다듬어야 한다.
1. 명시적으로 스키마를 정의할 것.
가장 기본적인 방법으로 원하는 JSON 구조를 프롬프트에 직접 보여주면 된다.
각 필드의 이름, 타입, 의미, 제약조건을 명시할 것.
다음 형식의 JSON으로 응답하세요: { "grammar_score": (0-25 사이의 정수), "grammar_feedback": (문법에 대한 피드백, 한국어 문자열), "vocabulary_score": (0-25 사이의 정수), "vocabulary_feedback": (어휘에 대한 피드백, 한국어 문자열), "total_score": (위 점수들의 합, 0-100 사이의 정수) }
2) 예시를 추가해 프롬프트 강화 가능(few-shot)
스키마 정의로 부족하다면 예시를 몇 개 보여주면 효과가 좋다. 이때, 최대한 다양한 케이스와 엣지 케이스를 보여주면 좋다. 예를 들어, 필드가 필요 없을 때, null을 사용해야 하는지, 아예 필드를 생략해도 되는지 예시로 보여주면 혼란을 줄일 수 있다.예시 1 - 오류가 있는 문장: 입력: "I go to school yesterday." 출력: {"grammar_score": 15, "feedback": "과거 시제를 사용해야 합니다.", "corrected": "I went to school yesterday."} 예시 2 - 완벽한 문장: 입력: "She has been studying for three hours." 출력: {"grammar_score": 25, "feedback": "완벽합니다!", "corrected": null}
3) 부정 지시 (네거티브 프롬프팅)
하지 말아햐 하는 것들을 알려주는 것도 효과적인데, 주로 프롬프트의 마지막 부분에 모아두면 편하다. LLM은 프롬프트 끝 부분에 있는 지시를 더 잘 따르는 경향이 있다고 한다(!)
4) Temperature 설정
온도(...)는 LLM 응답의 무작위성(창의성)을 조절하는 파라미터다. 높을 수록 다양하고 창의적인 응답이 나온다.
구조화된 출력이 필요한 경우에는 낮은 temperature을 사용하면 된다. 예를 들어 채점이나 분류 같이 일관성이 중요한 작업에서는 예측 가능한 응답이 필요하다. 반대로, 피드백 문구 생성 등의 자연스러운 표현이 필요한 작업에는 temperature을 높이면 좋다!
방어적 파싱 '시스템'을 구축
아무리 프롬프트를 잘 짜도, LLM은 예상치 못한 응답을 할 수 있다. 따라서, 파싱 코드는 이런 예외 상황, 즉 실패를 가정하고 작성할 필요가 있다. 그리고, 가장 효과적인 방법은 여러 파싱 전략을 순차적으로 시도하는 것으로 이것을 '다단계 파싱 전략'이라고 부른다.
가장 먼저 시도하는 것은 직접 파싱이다. 응답 텍스트를 그대로 json.loads()에 넣고 실패하면 마크다운 제거를 시도하거나, JSON 추출을 시도하는 등, 여러 방법을 시도해 보자.
import json
import re
from typing import Optional, Dict, Any
class RobustJSONParser:
"""여러 전략을 순차적으로 시도하는 JSON 파서"""
def parse(self, text: str) -> Optional[Dict[str, Any]]:
"""다양한 방법으로 JSON 파싱 시도"""
strategies = [
self._direct_parse,
self._strip_markdown,
self._extract_json_object,
self._fix_common_errors,
]
for strategy in strategies:
result = strategy(text)
if result is not None:
return result
return None
def _direct_parse(self, text: str) -> Optional[Dict]:
"""직접 파싱 시도"""
try:
return json.loads(text.strip())
except json.JSONDecodeError:
return None
def _strip_markdown(self, text: str) -> Optional[Dict]:
"""마크다운 코드 블록 제거 후 파싱"""
# ```json ... ``` 제거
cleaned = re.sub(r'^```(?:json)?\s*', '', text.strip())
cleaned = re.sub(r'\s*```$', '', cleaned)
try:
return json.loads(cleaned)
except json.JSONDecodeError:
return None
def _extract_json_object(self, text: str) -> Optional[Dict]:
"""텍스트에서 JSON 객체 추출"""
# 첫 번째 { 부터 마지막 } 까지 추출
match = re.search(r'\{[\s\S]*\}', text)
if match:
try:
return json.loads(match.group())
except json.JSONDecodeError:
pass
return None
def _fix_common_errors(self, text: str) -> Optional[Dict]:
"""흔한 JSON 오류 수정 후 파싱"""
# JSON 객체 추출
match = re.search(r'\{[\s\S]*\}', text)
if not match:
return None
json_str = match.group()
# 수정 시도
fixes = [
# 작은따옴표를 큰따옴표로
(r"'([^']*)':", r'"\1":'),
(r":\s*'([^']*)'", r': "\1"'),
# 후행 쉼표 제거
(r',\s*}', '}'),
(r',\s*]', ']'),
# 따옴표 없는 키
(r'(\{|,)\s*([a-zA-Z_][a-zA-Z0-9_]*)\s*:', r'\1"\2":'),
]
fixed = json_str
for pattern, replacement in fixes:
fixed = re.sub(pattern, replacement, fixed)
try:
return json.loads(fixed)
except json.JSONDecodeError:
return None
# 사용 예시
parser = RobustJSONParser()
test_cases = [
'{"score": 85}',
'```json\n{"score": 85}\n```',
"Here's the result: {\"score\": 85}",
"{score: 85}", # 따옴표 누락
]
for test in test_cases:
result = parser.parse(test)
print(f"Input: {test[:30]}... → Parsed: {result}")
스키마 검증
파싱에 성공했다면, 파싱된 딕셔너리가 원하는 필드를 가지고 있는지 확인할 필요가 있다. 크게 세 가지를 확인한다.
1) 필수 필드 존재 여부 - 핵심 필드가 빠져 있진 않은지 체크
2) 타입 검증 - 점수 필드에 문자열이 들어가 있으면 안 된다
3) 값 범위 검증 - 0~25 범위에 50이 있으면 오류
def validate_evaluation(data):
required = ['grammar_score', 'vocabulary_score', 'feedback']
for field in required:
if field not in data:
return False, f"필드 누락: {field}"
if not isinstance(data['grammar_score'], int):
return False, "grammar_score는 정수여야 합니다"
if not 0 <= data['grammar_score'] <= 25:
return False, "grammar_score는 0-25 범위여야 합니다"
return True, None
검증이 실패하면 재시도를 요청할 수 있다. 이때, 최대 횟수 등을 정해야 한다. 무한히 재시도하면 토큰 비용으로 빚이 생길 것이다. 더하여, 연속 요청은 API 서버에 부담을 주므로 지수 백오프를 적용하는 것이 좋다. 1초, 2초, 4초...와 같이 딜레이를 주자.
import time
from tenacity import retry, stop_after_attempt, wait_exponential
class EvaluationService:
def __init__(self, client, parser, validator):
self.client = client
self.parser = parser
self.validator = validator
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=1, max=10)
)
def evaluate_with_retry(self, sentence: str) -> EvaluationResult:
"""재시도 로직이 포함된 평가"""
response = self.client.chat.completions.create(
model="gpt-4-turbo-preview",
response_format={"type": "json_object"},
messages=self._build_messages(sentence),
temperature=0.3
)
content = response.choices[0].message.content
parsed = self.parser.parse(content)
if parsed is None:
raise ValueError("Failed to parse JSON response")
result = self.validator.to_dataclass(parsed)
if result is None:
raise ValueError("Schema validation failed")
return result
그리고 만약, 재시도마저 실패한 경우 폴백 응답이 필요하다. 사용자에게 에러 메시지를 보여주는 것보단 무언가 응답을 제공하는 게 중요하다. 폴백이 발생할 때마다 로그를 남기고 모니터링하자....
def create_fallback_response(sentence: str) -> EvaluationResult:
"""파싱 실패 시 기본 응답 생성"""
return EvaluationResult(
grammar_score=15,
grammar_feedback="평가를 처리하는 중 문제가 발생했습니다.",
vocabulary_score=15,
vocabulary_feedback="잠시 후 다시 시도해 주세요.",
naturalness_score=15,
naturalness_feedback="",
task_score=15,
task_feedback="",
total_score=60,
overall_feedback="일시적인 오류가 발생했습니다. 다시 시도해 주세요.",
corrected_sentence=None
)
def evaluate_safely(service, sentence: str) -> EvaluationResult:
"""안전한 평가 (항상 결과 반환)"""
try:
return service.evaluate_with_retry(sentence)
except Exception as e:
print(f"All retries failed: {e}")
return create_fallback_response(sentence)
Structured Output
OpenAI의 기능 중 하나인 Structured Outputs를 이용하면 JSON 스키마를 API에 직접 전달할 수 있고, LLM은 반드시 그 스키마를 따르는 응답만을 생성한다(!) 파이썬에서는 Pydantic 모델로 스키마를 정의한다.
다만, 지원하는 모델이 지원적이므로 어떤 모델이 사용 가능한지 확인할 필요는 있어 보인다. 스키마가 복잡하면 비용도 증가한다. 동적인 스키마에는 매번 새로운 스키마를 정의해야 하는 번거로움이 있으므로 사용을 권장하지 않는다.
from pydantic import BaseModel, Field
from typing import Optional
class EvaluationSchema(BaseModel):
"""평가 결과 스키마 (Structured Outputs용)"""
grammar_score: int = Field(ge=0, le=25, description="문법 점수")
grammar_feedback: str = Field(description="문법 피드백 (한국어)")
vocabulary_score: int = Field(ge=0, le=25, description="어휘 점수")
vocabulary_feedback: str = Field(description="어휘 피드백 (한국어)")
naturalness_score: int = Field(ge=0, le=25, description="자연스러움 점수")
naturalness_feedback: str = Field(description="자연스러움 피드백 (한국어)")
task_score: int = Field(ge=0, le=25, description="과제 완수 점수")
task_feedback: str = Field(description="과제 피드백 (한국어)")
total_score: int = Field(ge=0, le=100, description="총점")
corrected_sentence: Optional[str] = Field(description="수정된 문장")
overall_feedback: str = Field(description="종합 피드백 (한국어)")
from openai import OpenAI
client = OpenAI()
def evaluate_with_structured_output(sentence: str) -> EvaluationSchema:
"""Structured Outputs를 사용한 평가"""
response = client.beta.chat.completions.parse(
model="gpt-4o-2024-08-06", # Structured Outputs 지원 모델
messages=[
{
"role": "system",
"content": "You are a language tutor evaluating learner sentences."
},
{
"role": "user",
"content": f"Evaluate: '{sentence}'"
}
],
response_format=EvaluationSchema # Pydantic 모델 직접 전달
)
# 자동으로 파싱된 객체 반환
return response.choices[0].message.parsed