2026-05-14 강의 정리

---

목차

1. 문제와 접근
2. SelfRAGState 설계
3. 샘플 문서와 FAISS retriever
4. retrieve_node
5. generate_node
6. 관련성 평가 노드
7. 문서 필터링과 relevance_router
8. 쿼리 재작성 루프
9. 환각 평가 노드
10. 핵심 개념 총정리
11. 자주 나오는 실수/주의사항
12. 평가 루프가 필요한 실패를 구분하기

---

1. 문제와 접근

Self-RAG는 검색과 생성을 다시 평가하는 루프를 추가한다. 평가가 늘어나는 만큼 실패 원인을 더 잘 구분할 수 있어야 하며, 반복 자체를 품질 개선으로 간주하지 않는다. 관련성 평가와 근거 충실도 평가를 나누고 재검색이 필요한 상황을 명시하는 데 초점을 둔다.

retrieve
  ↓
grade_relevance
  ↓ 관련 문서 있음        ↓ 관련 문서 없음
generate                 rewrite_query
  ↓                         ↓
grade_hallucination      retrieve
  ↓
END

from typing import TypedDict, List, Literal
from dotenv import load_dotenv

load_dotenv()

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.documents import Document
from langchain_core.prompts import ChatPromptTemplate
from langchain_community.vectorstores import FAISS
from langgraph.graph import StateGraph, START, END
from pydantic import BaseModel, Field

llm = ChatOpenAI(model="gpt-4o-mini")
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

---

2. SelfRAGState 설계

검색 문서, 필터링 문서, 생성 결과, 평가 점수, 재시도 횟수를 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

INIT = {
    "question": "",
    "rewritten_question": "",
    "documents": [],
    "filtered_documents": [],
    "generation": "",
    "relevance_grades": [],
    "hallucination_score": "",
    "usefulness_score": 0,
    "relevance_retries": 0,
    "generation_retries": 0,
    "max_retries": 4,
}

---

3. 샘플 문서와 FAISS retriever

Self-RAG, reflection token, 관련성 평가, 환각 평가, 유용성 평가 문서를 만든다.

DOCS = [
    Document(page_content="Self-RAG는 검색·생성·평가를 LLM이 스스로 판단하는 자가 평가 RAG입니다.",
             metadata={"src": "selfrag-intro"}),
    Document(page_content="원논문 Self-RAG는 reflection token을 학습시켜 검색 여부와 답변 품질을 표시합니다.",
             metadata={"src": "selfrag-paper"}),
    Document(page_content="구현체는 LLM-as-judge 패턴(with_structured_output)으로 reflection token을 대신합니다.",
             metadata={"src": "selfrag-impl"}),
    Document(page_content="관련성 평가에서 yes 문서만 컨텍스트로 추리고, 모두 no면 쿼리 재작성 또는 재검색.",
             metadata={"src": "selfrag-relevance"}),
    Document(page_content="환각 평가는 답변이 문서에 근거하는지 확인합니다. yes(환각)면 다시 생성합니다.",
             metadata={"src": "selfrag-hallucination"}),
    Document(page_content="유용성 평가는 답변이 질문에 정말 답하는지 1-5점으로 매깁니다.",
             metadata={"src": "selfrag-usefulness"}),
]

vectorstore = FAISS.from_documents(DOCS, embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

---

4. retrieve_node

질문으로 문서를 검색한다.

def retrieve_node(state: SelfRAGState) -> dict:
    docs = retriever.invoke(state["question"])
    return {"documents": docs}

builder = StateGraph(SelfRAGState)
builder.add_node("retrieve", retrieve_node)
builder.add_edge(START, "retrieve")
builder.add_edge("retrieve", END)
app = builder.compile()

---

5. generate_node

검색된 문서를 context로 넣어 답변을 생성한다.

RAG_PROMPT = ChatPromptTemplate.from_messages([
    ("system", "다음 문서를 근거로 한국어로 간단히 답하세요. 문서에 없으면 '모름'이라고 답하세요"),
    ("human", "질문 : {question}\n\n문서:\n{context}"),
])


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

START -> retrieve -> generate -> END

---

6. 관련성 평가 노드

검색된 각 문서가 질문과 관련 있는지 yes/no로 평가한다.

class _Relevance(BaseModel):
    binary: Literal["yes", "no"] = Field(description="문서가 질문과 관련이 있는지")


relevance_grader = llm.with_structured_output(_Relevance)

REL_PROMPT = ChatPromptTemplate.from_messages([
    ("system", "당신은 문서가 질문과 관련 있는지 평가하는 전문가입니다. yes/no 로만 반환해주세요"),
    ("human", "질문: {question}\n문서: {doc}"),
])

def grade_relevance_node(state: SelfRAGState) -> dict:
    grades = []
    for d in state["documents"]:
        msg = REL_PROMPT.format_messages(question=state["question"], doc=d.page_content)
        grades.append(relevance_grader.invoke(msg).binary)
    return {"relevance_grades": grades}

---

7. 문서 필터링과 relevance_router

yes 문서만 남기고, 남은 문서가 없으면 다시 검색하거나 쿼리를 재작성한다.

def grade_and_filter_node(state: SelfRAGState) -> dict:
    grades, keep = [], []
    for d in state["documents"]:
        msg = REL_PROMPT.format_messages(question=state["question"], doc=d.page_content)
        g = relevance_grader.invoke(msg).binary
        grades.append(g)
        if g == "yes":
            keep.append(d)
    return {"relevance_grades": grades, "filtered_documents": keep}

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


def relevance_router(state: SelfRAGState) -> str:
    return "generate" if state["filtered_documents"] else "retrieve"

builder.add_conditional_edges(
    "grade_relevance",
    relevance_router,
    {"generate": "generate", "retrieve": "retrieve"},
)

---

8. 쿼리 재작성 루프

이 절은 재작성한 질문을 저장하고 다음 검색에 사용하는 흐름에 집중한다. 반복을 제한하려면 재작성 횟수를 실제로 증가시키는 노드와 한도 초과 시의 분기를 함께 연결해야 한다. 원래 질문도 보존해 재작성 중 의도가 달라졌는지 비교한다.

검색 실패 시 질문을 검색에 더 적합한 표현으로 바꾼다.

REWRITE_PROMPT = ChatPromptTemplate.from_messages([
    ("system", "다음 질문을 의미는 보존하되 검색이 더 잘 되도록 동의어, 구체화로 재작성해줘. 한 줄로만."),
    ("human", "원래 질문: {question}"),
])


def rewrite_query_node(state: SelfRAGState) -> dict:
    msg = REWRITE_PROMPT.format_messages(question=state["question"])
    new_q = llm.invoke(msg).content.strip()
    return {"rewritten_question": new_q}


def retrieve_aware_node(state: SelfRAGState) -> dict:
    q = state["rewritten_question"] or state["question"]
    return {"documents": retriever.invoke(q)}

def relevance_router(state: SelfRAGState) -> str:
    if state["filtered_documents"]:
        return "generate"
    return "rewrite_query"

retrieve -> grade_relevance
              ├─ generate
              └─ rewrite_query -> retrieve

---

9. 환각 평가 노드

답변이 검색 문서에 근거하는지와 질문에 충분히 답하는지는 다른 기준이다. 근거에는 맞지만 질문과 무관한 답변도 있을 수 있으므로, 평가 결과의 이름과 yes/no의 의미를 라우터 조건까지 일관되게 유지한다.

생성된 답변이 문서에 근거하는지 다시 평가한다.

class _Hallucination(BaseModel):
    binary: Literal["yes", "no"] = Field(
        description="답변이 환각인지 (yes=문서 밖 정보, no=근거 있음)"
    )


hallucination_grader = llm.with_structured_output(_Hallucination)

HAL_PROMPT = ChatPromptTemplate.from_messages([
    ("system", "답변의 모든 사실이 문서에 명시적으로 있으면 no, 문서 밖 정보가 있으면 yes"),
    ("human", "문서:\n{context}\n\n답변: {answer}"),
])

def grade_hallucination_node(state: SelfRAGState) -> dict:
    docs = state["filtered_documents"] or state["documents"]
    ctx = "\n".join(d.page_content for d in docs)
    msg = HAL_PROMPT.format_messages(context=ctx, answer=state["generation"])
    return {"hallucination_score": hallucination_grader.invoke(msg).binary}

---

10. 핵심 개념 총정리

단계	역할
retrieve	질문으로 문서 검색
grade_relevance	검색 문서가 질문과 관련 있는지 평가
filtered_documents	관련 문서만 남긴 결과
rewrite_query	검색 실패 시 질문 재작성
generate	문서 기반 답변 생성
grade_hallucination	답변이 문서에 근거하는지 평가

---

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

실수 1: 무한 루프 방지 카운터 없음

def bounded_router(state: SelfRAGState) -> str:
    if state["filtered_documents"]:
        return "generate"
    if state["relevance_retries"] < 1:
        return "rewrite_query"
    return "generate"

실수 2: state 키 오타

# 틀린 코드
state["docments"]

# 올바른 코드
state["documents"]

실수 3: max_tries와 max_retries 혼용

State 정의와 INIT에서 같은 키를 사용해야 한다.

실수 4: 조건부 라우터 반환 타입

def relevance_router(state: SelfRAGState) -> str:
    return "generate"

조건부 엣지 함수는 다음 노드 이름 문자열을 반환한다.

---

12. 평가 루프가 필요한 실패를 구분하기

검색 문서가 없거나 관련성이 부족한 경우에는 검색 입력과 코퍼스를 살펴본다. 충분한 근거가 있는데 생성 답변이 벗어나는 경우에는 생성 단계와 프롬프트를 확인한다. 두 문제를 모두 재생성으로 처리하지 않도록 경로를 나눈다.

루프를 연결한 뒤에는 최초 답변 대비 실제 개선과 추가 호출 수를 함께 기록한다. 한도까지 실패한 경우의 응답도 정의해 두어야 반복이 끝났다는 이유만으로 낮은 품질의 답변이 정상 결과가 되지 않는다.