8. 실습 ③ — RAGState 완성

from typing import TypedDict, Annotated, List, Any
import operator

class RAGState(TypedDict, total=False):
    # ── 입력 ──
    question: str            # 사용자 원본 질문 (변경 금지)
    query: str               # 검색용 질문 (재작성 가능)

    # ── 검색 결과 ──
    documents: List[Any]     # 검색된 조각
    scores: List[float]      # 유사도 점수
    retrieval_ok: bool       # 검색 성공 여부

    # ── 생성 결과 ──
    answer: str              # 생성된 답변
    insufficient: bool       # 근거 부족 자가신고
    has_citation: bool       # 출처 번호 포함 여부

    # ── 검증 결과 ──
    grade: str               # "pass" | "retry"
    reason: str              # 판정 이유

    # ── 제어 ──
    retries: int             # 재시도 횟수
    top_k: int               # 검색 개수 (동적 조절용)

    # ── 이력 (누적) ──
    log: Annotated[List[str], operator.add]
    tried_queries: Annotated[List[str], operator.add]

# ============================================================
# 18차시 | State 이해
# 교안: 18차시_State이해.pptx
# 역할: 이후 RAG 그래프가 공유할 State 칸(키)만 정의한다. 실행하지 않는다.
# 입력: 없음 (클래스 정의)
# 출력: 없음
#
#   State  = 아래 RAGState 딕셔너리. 노드 함수는 이 파일에 없다.
#   reducer = log, tried_queries 만 Annotated[..., operator.add].
#            노드가 ["한 줄"] 을 돌려주면 기존 리스트 뒤에 붙는다.
#            그 외 키는 return 한 값으로 그 칸만 덮어쓴다.
#   total=False = 초기 State에 모든 키가 없어도 타입 검사에서 빠진다.
#
# 이 파일은 그래프를 compile 하지 않는다. 칸의 흐름만 적는다.
#
#   question / query
#        |
#        v
#   documents / scores / retrieval_ok
#        |
#        v
#   answer / insufficient / has_citation
#        |
#        v
#   grade / reason
#   retries, top_k 는 반복 제어용 칸
#   log, tried_queries 는 reducer 로 누적
# ============================================================

from typing import TypedDict, Annotated, Any
import operator

# ============================================================
# RAG State 정의
# ============================================================

class RAGState(TypedDict, total=False):
    """RAG 그래프의 모든 노드가 함께 사용하는 데이터 구조입니다."""

    # --------------------------------------------------------
    # 1. 사용자 입력
    # --------------------------------------------------------

    # 사용자가 처음 입력한 질문입니다.
    question: str

    # 실제 검색에 사용할 질문입니다.
    # 검색 실패 시 다른 문장으로 변경할 수 있습니다.
    query: str

    # --------------------------------------------------------
    # 2. 검색 결과
    # --------------------------------------------------------

    # 검색된 문서 조각들을 저장합니다.
    documents: list[Any]

    # 검색된 문서들의 유사도 점수를 저장합니다.
    scores: list[float]

    # 검색 결과를 사용할 수 있는지 저장합니다.
    retrieval_ok: bool

    # --------------------------------------------------------
    # 3. 답변 생성 결과
    # --------------------------------------------------------

    # LLM이 생성한 최종 답변입니다.
    answer: str

    # 근거가 부족한지 저장합니다.
    insufficient: bool

    # 답변에 출처 번호가 포함되었는지 저장합니다.
    has_citation: bool

    # --------------------------------------------------------
    # 4. 답변 검증 결과
    # --------------------------------------------------------

    # 답변 검증 결과를 저장합니다.
    # 예: "pass", "retry"
    grade: str

    # 검증 결과가 나온 이유를 저장합니다.
    reason: str

    # --------------------------------------------------------
    # 5. 그래프 제어 정보
    # --------------------------------------------------------

    # 답변 생성 등을 다시 시도한 횟수입니다.
    retries: int

    # 검색할 문서 개수입니다.
    top_k: int

    # --------------------------------------------------------
    # 6. 실행 이력
    # --------------------------------------------------------

    # 각 노드의 실행 기록을 계속 누적합니다.
    # reducer: 노드가 돌려준 리스트를 기존 log 뒤에 이어 붙인다.
    log: Annotated[list[str], operator.add]

    # 지금까지 사용한 검색어를 계속 누적합니다.
    # reducer: 같은 방식으로 tried_queries 뒤에 이어 붙인다.
    tried_queries: Annotated[list[str], operator.add]

def make_initial_state(question: str) -> dict:
    # invoke에 넣을 초기 State를 만든다
    return {
        "question": question,
        "query": question,      # 처음엔 원본과 동일
        "retries": 0,
        "log": [],
        "tried_queries": [],
    }

# 사용
app.invoke(make_initial_state("환불은 며칠 이내인가요?"))