2026-05-15 강의 정리

---

목차

1. 문제와 접근
2. 전날 Self-RAG 루프 복습
3. generate_with_counter
4. hallucination_router
5. 재생성 전략
6. usefulness_grader
7. usefulness_router
8. safe router 통합
9. app.stream과 그래프 시각화
10. CRAG 평가
11. knowledge_refine_node
12. 핵심 개념 총정리
13. 자주 나오는 실수/주의사항
14. 종료 이유를 설명할 수 있는 루프

---

1. 문제와 접근

재검색과 재생성을 함께 사용하면 실패 시 어떤 작업을 다시 할지 결정하는 규칙이 복잡해진다. Self-RAG의 관련성·충실도·유용성 평가를 연결하면서 반복 횟수와 종료 이유를 명확히 한다. CRAG의 지식 정제도 어떤 근거가 남고 빠졌는지 추적하는 관점에서 살펴본다.

retrieve
  ↓
grade_relevance
  ↓
generate
  ↓
grade_hallucination
  ↓ 환각이면 재생성
grade_usefulness
  ↓
점수에 따라 END / regenerate / reretrieve

---

2. 전날 Self-RAG 루프 복습

핵심 state는 검색 문서, 필터링 문서, 생성 답변, 평가 점수, 재시도 횟수다.

class SelfRAGState(TypedDict):
    question: str
    rewritten_question: str
    documents: List[Document]
    filtered_documents: List[Document]
    generation: str
    relevance_grades: List[str]
    hallucination_score: str
    usefulness_score: int
    relevance_retries: int
    generation_retries: int
    max_tries: int
    max_retries: int

---

3. generate_with_counter

재생성 횟수를 추적하기 위해 답변 생성 시 generation_retries를 증가시킨다.

def generate_with_counter(state: SelfRAGState) -> dict:
    docs = state["filtered_documents"] or state["documents"]
    ctx = "\n".join(f"- {d.page_content}" for d in docs)
    msg = RAG_PROMPT.format_messages(question=state["question"], context=ctx)
    return {
        "generation": llm.invoke(msg).content,
        "generation_retries": state["generation_retries"] + 1,
    }

---

4. hallucination_router

환각이면 다시 생성하고, 근거 있음이면 유용성 평가로 넘긴다.

def hallucination_router(state: SelfRAGState) -> str:
    if state["generation_retries"] >= 2:
        return "usefulness"
    return "generate" if state["hallucination_score"] == "yes" else "usefulness"

builder.add_conditional_edges(
    "grade_hallucination",
    hallucination_router,
    {"generate": "generate", "usefulness": "grade_usefulness"},
)

---

5. 재생성 전략

환각이 발생했을 때 사용할 수 있는 두 가지 방식이다.

1. temperature를 높인 별도 LLM으로 다시 생성
2. "이전 답변은 환각이었다"는 지시를 넣고 문서 근거 답변을 재생성

llm_temp = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)


def generate_1_temp(question, docs):
    ctx = "\n".join(f"- {d.page_content}" for d in docs)
    return llm_temp.invoke(
        RAG_PROMPT.format_messages(question=question, context=ctx)
    ).content

HALLUCINATION_PROMPT = ChatPromptTemplate.from_messages([
    ("system", "이전 답변은 환각이었습니다. 문서에 의존해서 다시 답변하세요."),
    ("human", "질문 : {question}\n\n문서 :\n{context}"),
])

---

6. usefulness_grader

답변이 질문에 얼마나 잘 답하는지 1~5점으로 평가한다.

class _Usefulness(BaseModel):
    score: int = Field(description="답변의 유용성 1~5점", ge=1, le=5)


usefulness_grader = llm.with_structured_output(_Usefulness)

USE_PROMPT = ChatPromptTemplate.from_messages([
    ("system", "답변이 질문에 얼마나 잘 답하는지 1~5점으로 평가. 1=관련 없음, 5=완벽"),
    ("human", "질문: {question}\n답변: {answer}"),
])

def grade_usefulness_node(state: SelfRAGState) -> dict:
    msg = USE_PROMPT.format_messages(
        question=state["question"],
        answer=state["generation"],
    )
    return {"usefulness_score": usefulness_grader.invoke(msg).score}

---

7. usefulness_router

유용성 점수에 따라 종료, 재생성, 재검색을 결정한다.

def usefulness_router(state: SelfRAGState) -> str:
    s = state["usefulness_score"]
    if state["generation_retries"] >= 3 or state["relevance_retries"] >= 2:
        return "end"
    if s >= 5:
        return "end"
    if s >= 3:
        return "regenerate"
    return "reretrieve"

5점   -> END
3~4점 -> generate
1~2점 -> retrieve

---

8. safe router 통합

재시도 한도를 공통으로 둔다고 모든 경로가 자동으로 제한되지는 않는다. 각 경로에서 어떤 카운터를 증가시키고 어떤 라우터가 읽는지 연결해서 확인한다. 한도 도달에 따른 종료와 평가를 통과한 종료도 결과에서 구분할 필요가 있다.

다음 예제는 공통 재시도 한도를 사용하는 라우터 구성을 보여 준다. 실제 그래프에 연결할 때는 각 카운터를 갱신하는 노드와 종료 경로도 함께 확인한다.

def safe_relevance_router(state: SelfRAGState) -> str:
    if state["relevance_retries"] >= state["max_retries"]:
        return "generate"
    if state["filtered_documents"]:
        return "generate"
    return "rewrite_query"


def safe_hallucination_router(state: SelfRAGState) -> str:
    if state["generation_retries"] >= state["max_retries"]:
        return "usefulness"
    return "generate" if state["hallucination_score"] == "yes" else "usefulness"


def safe_usefulness_router(state: SelfRAGState) -> str:
    s = state["usefulness_score"]
    if (
        state["generation_retries"] >= state["max_retries"]
        or state["relevance_retries"] >= state["max_retries"]
    ):
        return "end"
    if s >= 5:
        return "end"
    if s >= 3:
        return "regenerate"
    return "reretrieve"

최종 그래프

builder = StateGraph(SelfRAGState)
builder.add_node("retrieve", retrieve_aware_node)
builder.add_node("grade_relevance", grade_and_filter_node)
builder.add_node("rewrite_query", rewrite_query_counter)
builder.add_node("generate", generate_with_counter)
builder.add_node("grade_hallucination", grade_hallucination_node)
builder.add_node("grade_usefulness", grade_usefulness_node)

builder.add_edge(START, "retrieve")
builder.add_edge("retrieve", "grade_relevance")
builder.add_conditional_edges("grade_relevance", safe_relevance_router, {
    "generate": "generate",
    "rewrite_query": "rewrite_query",
})
builder.add_edge("rewrite_query", "retrieve")
builder.add_edge("generate", "grade_hallucination")
builder.add_conditional_edges("grade_hallucination", safe_hallucination_router, {
    "generate": "generate",
    "usefulness": "grade_usefulness",
})
builder.add_conditional_edges("grade_usefulness", safe_usefulness_router, {
    "end": END,
    "regenerate": "generate",
    "reretrieve": "retrieve",
})
app = builder.compile()

---

9. app.stream과 그래프 시각화

그래프 실행 중 어떤 노드가 어떤 값을 반환하는지 stream으로 확인한다.

for chunk in app.stream({**INIT, "question": "self-rag의 핵심은?", "max_retries": 4}):
    for node_name, payload in chunk.items():
        print(node_name, payload)

from IPython.display import Image, display
display(Image(app.get_graph().draw_mermaid_png()))

---

10. CRAG 평가

CRAG는 검색 결과를 correct / ambiguous / incorrect로 평가한다.

class _CragGrade(BaseModel):
    """검색 결과의 신뢰도를 3단계로 평가."""

    grade: Literal["correct", "ambiguous", "incorrect"] = Field(
        description="correct=자신있음, ambiguous=부분적, incorrect=무관"
    )
    score: float = Field(ge=0.0, le=1.0, description="0~1 신뢰도 점수")


crag_grader = llm.with_structured_output(_CragGrade)

def crag_grade_node(state: SelfRAGState) -> dict:
    docs = state.get("filtered_documents") or state["documents"]
    if not docs:
        return {"crag_grade": "incorrect", "crag_score": 0.0}

    ctx = "\n---\n".join(d.page_content for d in docs[:5])
    r = crag_grader.invoke(
        CRAG_PROMPT.format_messages(question=state["question"], docs=ctx)
    )
    return {"crag_grade": r.grade, "crag_score": r.score}

---

11. knowledge_refine_node

지식 정제 후에는 남은 텍스트만 보지 않고 원래 검색 문서와 비교한다. 질문의 핵심 조건이나 예외가 제거되지 않았는지 확인하고, 정제 결과가 빈 경우에 다음 노드가 어떻게 동작할지도 정한다.

검색 문서를 문장 단위로 쪼개고, 질문에 유용한 문장만 남긴다.

class _StripGrade(BaseModel):
    relevant: bool = Field(description="이 문장이 질문 답변하는데 유용한가요?")


strip_grader = llm.with_structured_output(_StripGrade)

def knowledge_refine_node(state: SelfRAGState) -> dict:
    refined = []
    docs = state.get("filtered_documents") or state["documents"]

    for doc in docs:
        strips = [s.strip() for s in doc.page_content.replace("!", ".").replace("?", ".").split(".")]
        kept = []
        for s in strips:
            if len(s) < 10:
                continue
            try:
                r = strip_grader.invoke(
                    STRIP_PROMPT.format_messages(question=state["question"], strip=s)
                )
                if r.relevant:
                    kept.append(s)
            except Exception:
                kept.append(s)

        if kept:
            refined.append(Document(page_content=". ".join(kept), metadata=doc.metadata))

    return {"refined_documents": refined}

---

12. 핵심 개념 총정리

개념	의미
hallucination_router	답변이 문서 밖 정보를 쓰면 재생성
usefulness_grader	답변이 질문에 잘 답하는지 점수화
safe router	재시도 한도로 무한 루프 방지
CRAG	검색 결과를 correct/ambiguous/incorrect로 평가
knowledge refine	문장 단위로 관련 지식만 정제
app.stream	노드별 실행 결과 관찰

---

13. 자주 나오는 실수/주의사항

실수 1: 리스트와 문자열 비교

# 틀린 코드
return "generate" if ["hallucination_score"] == "yes" else "end"

# 올바른 코드
return "generate" if state["hallucination_score"] == "yes" else "end"

실수 2: .context 속성 사용

# 틀린 코드
return llm_temp.invoke(...).context

# 올바른 코드
return llm_temp.invoke(...).content

실수 3: prompt 변수명 불일치

# prompt에는 {context}가 있는데 ctx로 넘기면 안 됨
HALLUCINATION_PROMPT.format_messages(question=question, context=ctx)

실수 4: retry 카운터 증가 누락

return {
    "rewritten_question": new_q,
    "relevance_retries": state["relevance_retries"] + 1,
}

---

14. 종료 이유를 설명할 수 있는 루프

루프의 품질은 반복 횟수보다 재시도 이유와 결과의 연결로 판단한다. 관련성 부족으로 재검색한 요청과 생성 오류로 다시 답한 요청을 나누어 기록하면 불필요한 호출이 어느 경로에서 늘어나는지 알 수 있다.

실행 기록에는 평가값, 이동한 노드, 남은 시도 횟수를 함께 남긴다. 같은 상태로 같은 경로를 반복하는 사례를 확인하고, 한도 소진 시에는 확인 가능한 근거의 범위 안에서 답하거나 정보 부족을 드러내는 정책을 정한다.