langchain / / 2024. 11. 5. 07:54

[langgraph] Self RAG 실습 예제

udemy 강좌 내용[LangGraph- Develop LLM powered AI agents with LangGraph]의 일부를 정리한 내용입니다. 예제의 일부 내용을 수정하였습니다. 문제가 되면 삭제하겠습니다.

https://www.udemy.com/course/langgraph/learn/lecture/43978374#overview

Self RAG 개념

이번에는 Self RAG를 구현해볼 예정이다. Self RAG 논문에서 유래되었고 모델이 생성한 답변을 평가하는 것을 의미한다. generation을 받은 다음 document와 비교를 해서 모델의 hallucination이 있는지를 확인한다. 그리고 답변이 document를 근거로 나타내는지를 체크한다.

답변이 사용자의 질문에 대한 대답이라면 사용자에게 답변을 리턴하고 질문에 대한 대답이 아니라면 vector store에는 더 이상의 추가 정보를 찾을 수 없다는 것을 나타내므로 웹 검색이 필요하다는 것을 의미한다. 그리고 document에 근거한 답변이 아닌 hallucination이 있다면 답변에 근거한 답변을 만들기 위해 재생성한다.

Self RAG 구현

이전에 살펴본 Corrective RAG에서 generate 노드 후에 추가 작업을 진행할 것이다. generate 노드에서 답변을 생성하는 대신 reflection 레이어를 추가한다. 여기서 hallucination grader와 answer_grader chain을 구현한다. 두 체인은 answer를 평가하고 모델이 hallucination이 있는지 결정한다. 답변이 document를 기반으로 만들어졌는지 확인하는 기능을 한다.

[hallucination_grader.py]

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableSequence
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field

llm = ChatOpenAI(temperature=0)


class GradeHallucinations(BaseModel):
    """Binary score for hallucination present in generation answer."""

    binary_score: bool = Field(
        description="Answer is grounded in the facts, 'yes' or 'no'"
    )


structured_llm_grader = llm.with_structured_output(GradeHallucinations)

system = """You are a grader assessing whether an LLM generation is grounded in / supported by a set of retrieved facts. \n 
     Give a binary score 'yes' or 'no'. 'Yes' means that the answer is grounded in / supported by the set of facts."""
hallucination_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", system),
        ("human", "Set of facts: \n\n {documents} \n\n LLM generation: {generation}"),
    ]
)

hallucination_grader: RunnableSequence = hallucination_prompt | structured_llm_grader

여기서 LLM의 답변이 document를 기반으로 하는지 체크한다.

GradeHallucinations 클래스를 만들어 hallucination을 체크하여 응답을 받도록 하자.
llm은 with_structured_output으로 pydantic model(GradeHallucinations)을 리턴값으로 취한다.

ChatPromptTemplate 튜플에서 첫 번째 메시지는 system 템플릿이 된다. 그리고 두 번째 메시지는 human이며 Fact 정보가 된다. 여기에 문서 정보와 LLM이 생성한 generation 답변을 추가할 것이다.

이제 hallucination prompt를 가지고 있는 hallucination_grader chain을 만들어 structured_llm_grader로 연결하자.

여기서 일어나는 일은 yes or no 응답을 받는 것이다. 답변이 document 기반이라면 적용할 것이다.

이제 답변을 생성할 chain을 구현하자.

[answer_grader.py]

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableSequence
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field


class GradeAnswer(BaseModel):

    binary_score: bool = Field(
        description="Answer addresses the question, 'yes' or 'no'"
    )


llm = ChatOpenAI(temperature=0)
structured_llm_grader = llm.with_structured_output(GradeAnswer)

system = """You are a grader assessing whether an answer addresses / resolves a question \n 
     Give a binary score 'yes' or 'no'. Yes' means that the answer resolves the question."""
answer_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", system),
        ("human", "User question: \n\n {question} \n\n LLM generation: {generation}"),
    ]
)

answer_grader: RunnableSequence = answer_prompt | structured_llm_grader

GradeAnswer 클래스를 만들자. 속성은 binary_score 하나만 가지고 있다. answer가 question을 나타내는지 여부를 설명하고 있다.

다음은 system 프롬프트이다. answer가 question의 문제를 해결하는지 여부를 나타낸다. ChatPromptTemplate에서 question과 LLM이 생성한 generation을 추가한다.

answer_grader의 응답은 True or False가 된다.

[graph.py]

graph에서 위에서 만든 hallucination_grader를 import 하자. 그리고 함수를 하나 만든다. 이것은 conditional_edge 함수이다.

def grade_generation_grounded_in_documents_and_question(state: GraphState) -> str:
    print("---CHECK HALLUCINATIONS---")
    question = state["question"]
    documents = state["documents"]
    generation = state["generation"]

    score = hallucination_grader.invoke(
        {"documents": documents, "generation": generation}
    )

    if hallucination_grade := score.binary_score:
        print("---DECISION: GENERATION IS GROUNDED IN DOCUMENTS---")
        print("---GRADE GENERATION vs QUESTION---")
        score = answer_grader.invoke({"question": question, "generation": generation})
        if answer_grade := score.binary_score:
            print("---DECISION: GENERATION ADDRESSES QUESTION---")
            return "useful"
        else:
            print("---DECISION: GENERATION DOES NOT ADDRESS QUESTION---")
            return "not useful"
    else:
        print("---DECISION: GENERATION IS NOT GROUNDED IN DOCUMENTS, RE-TRY---")
        return "not supported"

      ...

workflow.add_conditional_edges(
    GENERATE,
    grade_generation_grounded_in_documents_and_question,
    {
        "not supported": GENERATE,
        "useful": END,
        "not useful": WEBSEARCH,
    },
)

hallucination_grader를 통해 hallucination이 있는지 체크하여 응답결과로 받는다.

hallucination_grade가 True라면 hallucination이 없다는 것을 의미한다. 그리고 answer_grade가 True라면 질문에 대한 답변이 정상적으로 생성된 것을 의미한다. 이 경우는 useful로 리턴한다.

반면에, document를 기반으로 답변을 만들었지만 질문에 대한 답변을 나타내는 것이 아니라면 not useful로 리턴한다. 이 경우는 vector store에 정보가 충분하지 않다는 것을 의미한다. 그래서 외부 검색을 활용한다.

그리고 답변이 document 기반이 아니라면 not_supported로 리턴한다. 이 경우는 document에서 재생성할 필요가 있다.

이제 main.py를 실행해보자.

from dotenv import load_dotenv

load_dotenv()

from graphs.graph import app

if __name__ == "__main__":
    print(app.invoke(input={"question": "야구에서 홈런이 뭐야?"}))

실행결과는 아래와 같다.

---RETRIEVE---
---CHECK DOCUMENT RELEVANCE TO QUESTION---
---GRADE: DOCUMENT NOT RELEVANT---
---GRADE: DOCUMENT NOT RELEVANT---
---GRADE: DOCUMENT RELEVANT---
---GRADE: DOCUMENT NOT RELEVANT---
---ASSESS GRADED DOCUMENTS---
---DECISION: GENERATE---
---GENERATE---
---CHECK HALLUCINATIONS---
---DECISION: GENERATION IS GROUNDED IN DOCUMENTS---
---GRADE GENERATION vs QUESTION---
---DECISION: GENERATION ADDRESSES QUESTION---
{'question': '야구에서 홈런이 뭐야?', 'generation': '홈런은 타자가 수비수의 실책 없이 홈 베이스를 밟을 수 있게 공을 치는 것이다. 현대 야구에서는 홈런을 전자의 경우로 기록하고, 인사이드 파크 홈런을 후자의 경우로 기록한다. 만루홈런을 치면 최대 4점을 득점할 수 있다.', 'web_search': False, 'documents': [Document(metadata={'language': 'ko', 'source': 'https://namu.wiki/w/%EC%95%BC%EA%B5%AC/%EA%B2%BD%EA%B8%B0%20%EB%B0%A9%EC%8B%9D', 'title': '야구/경기 방식 - 나무위키'}, page_content='살아서 올 세이프가 된 경우에는 안타가 아니라 야수선택으로 기록된다. 반대로 주자가 죽었더라도 주자가 한 개의 루를 지나서 추가 진루를 하려다가 죽은 경우(예를 들어 1사 1루 상황에서 우전 안타 - 주자가 2루를 지나 3루까지 가려다가 3루에서 죽은 경우)에는 타자에게는 안타를 인정한다.홈런(Home Run)수비수의 실책 없이 타자가 홈 베이스를 밟을 수 있게 공을 치는 것. 타자가 홈 베이스를 밟을 시간만큼 공을 치려면 수비가 못 잡도록  바운드가 되기 전에 아예 담장 밖으로 날려버리거나[25], 담장 안에 공이 떨어질 경우엔 수비수가 그동안 공을 처리하지 못할 정도로 멀리 오랫동안 처리할 타구를 보내야 한다. 현대 야구에서는 전자의 경우를 홈런으로 기록하고, 후자의 경우는 인사이드 파크 홈런으로 기록한다. 루상에 주자가 있으면 순서대로 진루하여 홈 플레이트를 밟으면 득점한다. 그렇기 때문에 만루홈런을[26] 치면 최대 4점을 득점할 수 있다. 다만 공의 착지점을 확인하기 애매할 때가 많은 만큼 이를 판정하기 위해 파울 라인과 페어 지역의 경계 지점에 파울 폴이 세워져 있는데, 일단은 당연히 이 안쪽으로 공이 들어가야 하고, 폴 위를 정확히 지나가거나 폴을 직접 맞추는 것도 홈런으로 인정된다. 또한 공이 펜스를 넘었다가 그물이나 관중에 의해 다시 공이 그라운드로 들어가는 경우에도 홈런으로 인정된다. 이를 확인하기 위해서 보통 단층 펜스 상단에 노란 줄을 긋고 줄을 넘어가면 홈런으로 인정하는 룰을 많이 쓴다.[27] 이중으로 펜스가 쳐진 경우는 예외. 획일적 구조로 지어진 한국과는 달리 미국은 구장마다 펜스의 생김새와 규정이 각각 달라서')]}

langsmith의 실행결과는 아래와 같다.

반응형
  • 네이버 블로그 공유
  • 네이버 밴드 공유
  • 페이스북 공유
  • 카카오스토리 공유