cert_study_app/chains/question_parser_chain.py

import json
import time
from typing import List, Optional

from pydantic import BaseModel, Field
from langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import PydanticOutputParser


class QuestionParsedResult(BaseModel):
    stem: str = Field(description="문제 본문", default="")
    options: List[str] = Field(description="보기 목록 (객관식일 경우)", default_factory=list)
    answer: List[str] = Field(description="정답 또는 선택결과", default_factory=list)
    explanation: str = Field(description="간단한 해설", default="")
    question_type: str = Field(description="문제 유형 (mcq, yes_no, sequence, code, scenario)")
    code: Optional[str] = Field(description="코드가 있다면 포함", default="")
    sequence: Optional[List[str]] = Field(description="순서형일 경우 항목", default_factory=list)

parser = PydanticOutputParser(pydantic_object=QuestionParsedResult)

QUESTION_PARSE_PROMPT = ChatPromptTemplate.from_template(
    """
너는 OCR로 인식된 시험 문제를 분석하여 구조화하는 전문 파서이다.
문제 유형은 다음 중 하나일 수 있다:
- mcq: 객관식
- yes_no: 참/거짓 또는 예/아니오
- sequence: 순서형 문제
- code: 코드 기반 문제
- scenario: 시나리오 기반 문제

이 문제는 주로 '{question_type}' 유형으로 추정된다.

{format_instructions}

문제 원문:
{ocr_text}
"""
)


def detect_question_type(text: str) -> str:
    lowered = text.lower()
    if any(k in lowered for k in ["예", "아니오", "yes", "no", "true", "false"]):
        return "yes_no"
    if any(k in lowered for k in ["순서", "정렬", "drag", "drop", "순서대로"]):
        return "sequence"
    if any(k in lowered for k in ["json", "{", "}", "az ", "set-az", "cli", "powershell", "bash", "cmd", "코드"]):
        return "code"
    if any(k in lowered for k in ["시나리오", "case", "contoso", "litware", "fabrikam", "조건", "상황"]):
        return "scenario"
    return "mcq"


def build_question_parser_chain(llm):
    # LCEL: Prompt -> LLM -> Pydantic Parser
    return QUESTION_PARSE_PROMPT | llm | parser


def parse_question_with_chain(llm, page: int, text: str) -> dict:
    question_type = detect_question_type(text)
    chain = build_question_parser_chain(llm)

    try:
        started_at = time.time()
        # parser.get_format_instructions()를 통해 프롬프트에 JSON 스키마를 주입합니다.
        parsed_result = chain.invoke(
            {
                "question_type": question_type,
                "ocr_text": text[:1800],
                "format_instructions": parser.get_format_instructions(),
            }
        )
        
        # Pydantic 모델을 dict로 변환 (Pydantic v1/v2 호환성을 위해 dict() 사용)
        parsed = parsed_result.dict()
        parsed["page"] = page
        
        print(f"[LLM] p{page} ({question_type}) parsed in {time.time() - started_at:.1f}s")
        return parsed
        
    except Exception as exc:
        print(f"[WARN] LLM parse failed (p{page}): {exc}")
        return {
            "page": page,
            "stem": text[:400],
            "options": [],
            "answer": [],
            "explanation": f"오류: {exc}",
            "question_type": question_type,
            "code": "",
            "sequence": [],
        }