SK네트웍스 Family AI 캠프 2기 : 24th week (10월 4주차 + 10월 회고)

이번 주는 제 개인적인 일로 정신없이 시간이 흘렀습니다.

그동안 시간이 너무 빨리 흘러간다고 투정 부렸던 것과 차원이 다를 정도로요.

 

할아버지는 잘 보내드리고 왔습니다.

그동안 감사했다고, 사랑한다고 말씀드리고 왔습니다.

마지막 순간에 편안한 얼굴을 하고 떠나셔서 다행히 마음이 놓였습니다.

 

그리고 지난주에 눈물이 안 나왔다는 것은 거짓말이었습니다.

전화로 통보 받았을 때, 너무 갑작스러워서 눈물이 안 났던 것뿐이었습니다.

생각해 보니 이전에 비슷한 경험을 했을 때도 그랬던 것 같네요.

너무나도 당연한 인생사 중 하나지만, 사람을 마지막으로 배웅하는 것은 늘 어렵게만 느껴집니다.

 

이런 상황을 겪을 때마다 '삶이란 무엇일까'라는 고민을 마주하게 됩니다.

부, 명예, 찬란하게 빛나는 젊음, 매력적인 이성, 사랑하는 가족과 친구들... 이외의 수많은 가치들...

인생이란 끝이 존재하기에 이 모든 것은 재와 같이 사라지기 마련입니다.

 

그럼 인간은 무엇을 위해 열심히 살아야 하는 것일까요?

10대 때부터 줄곧 고민했던 생각들 중 하나로, 이제는 어느 정도 답을 할 수 있게 된 것 같습니다.

 

삶은 공허합니다. 

空에서 출발한 모든 존재는 다시 空으로 돌아갑니다.

잔인할 정도로 허무함만 존재합니다.

 

사실 제가 열심히 사는 이유 중 하나는 삶의 허무함을 극복하려는 게 가장 큽니다.

저도 사람이기에 돈/명예 등이 필요한 때가 있다는 것을 압니다. 

먹고살아야 하는 문제가 걸려 있는데 아예 필요 없다고 말하는 것은 가식이지요.

하지만 단순히 그것 때문에 열심히 하는 것은 아닙니다.

 

시지프스처럼 제게 주어진 임무에 집중하다 보면 그 순간만큼은 공허한 감정이 사라지거든요.

 

허무한 삶 속에서 제가 최대한으로 사람들에게 표현할 수 있는 애정은 아래와 같습니다.

 주어진 임무에 함께 집중함으로써 삶의 공허함을 같이 이겨내 보자!

 

저를 포함한 모든 이들에게 전하는 일종의 애정 표현입니다.

시지프스는 타의적으로 돌을 미는 형벌을 받았지만, 인간은 공허함을 해결하기 위해 능동적으로 움직일 수 있으니 나름대로 축복이라고 느껴집니다.

 

이러한 사고관을 모토로 삼아서 그런지 결과보다 과정에 더 초점을 두는 사람으로 성장한 것 같습니다.

SK네트웍스 Family AI 캠프 역시 마지막까지 열심히 임해보겠습니다.

 

- P.S -

혼자 시무룩하게 있지 말라고 걱정해 주던 동갑 라인 친구들 '재민, 정현, 주희, 종호' 고마워!

과정 끝까지 잘 마쳐보자!

 

 

● 성취

이번 주에는 4일이나 자리를 비우게 되어서 팀 프로젝트 면에서 걱정이 많았습니다.

목요일에 잠시 복귀해서 상태를 파악했을 때, 다른 팀원들이 정말 열심히 노력한 흔적이 보여서 감동했습니다.

누구보다 많이 고뇌하고, 고심한 내용들을 구현하기 위해 고군분투한 모습들이 눈에 보였습니다.

 

다른 팀에 비해 초반에 우여곡절이 있어서 걱정이 많았는데, 방향을 잡고 앞을 향해 나아가는 모습이 인상 깊었습니다.

능동적으로 프로젝트에 임하는 모습이 우리 2기 2팀의 가장 큰 장점이라고 생각합니다.

 

제 개인 이슈 때문에 자리를 비우게 되어서 다들 불안했을 텐데, 저 없는 동안에도 열심히 잘해줘서 정말 고마웠어요.

저도 끝까지 최선을 다해서 마무리 짓도록 하겠습니다.

 

유빈이, 인교 언니, 영재 오빠, 서연 언니! 앞으로 좀만 더 힘내보자, 화이팅!

 

 

● 학습 & 개선

목요일에 잠시 복귀했을 때, 프로젝트에서 사용할 모델 고도화를 위해 self-rag에 대해 찾아봤습니다.

오늘은 Self-RAG를 개량한 코드에 대해 설명하겠습니다.

(논문 원본과 다른 부분이 존재합니다. 논문 원본은 링크를 걸어 놓았습니다.)

 

Self-RAG는 자아 성찰 기능을 탑재한 발전된 형태의 RAG 시스템이라고 생각하면 좋아요.

2023년 10월 17일에 Arxiv에 게시된 따끈따끈한 논문이랍니다.

 

자, 이제부터 LangChain과 LangGraph로 구현한 Self-RAG 개량 코드를 소개하겠습니다.

 

# 라이브러리 설치

!pip install -qU kiwipiepy konlpy langchain-teddynote
!pip install -U langchain_community tiktoken langchain-openai langchainhub langchain langgraph pypdf rank_bm25

 

먼저 필요한 라이브러리들을 설치해 줍니다.

 

 

# 라이브러리 및 환경 설정

from dotenv import load_dotenv
import os
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import TextLoader
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain_community.document_loaders import PyPDFLoader
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_openai import ChatOpenAI
from langchain import hub
from langchain_core.output_parsers import StrOutputParser
from langchain_teddynote.retrievers import konlpy_bm25
from langchain import hub
from langchain_core.output_parsers import StrOutputParser
from pprint import pprint
from langgraph.graph import END, StateGraph, START

 

코드 동작에 필요한 라이브러리들을 불러와 주세요.

 

 

# API 설정

load_dotenv()
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")

 

'.env' 파일에 기록해 놓은 OPENAI_API_KEY를 가져옵니다.

 

'.env' 파일은 메모장으로도 쉽게 만들 수 있답니다.

메모장 안에 OPENAI_API_KEY='여러분의 OPEN API 키'를 입력해 주신 후, '.env'라는 명칭으로 저장해 주세요.

 

 

# Retriever 생성

documents = TextLoader("./data/data_file.txt").load()
text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=20)
doc_splits = text_splitter.split_documents(documents)

retriever = konlpy_bm25.KkmaBM25Retriever.from_documents(
    documents=doc_splits,
    collection_name="rag-bm25",
    embedding=OpenAIEmbeddings(api_key=OPENAI_API_KEY),
    normalize_method='tmm',
    top_k=3,
    weight=0.89,
)

 

텍스트 파일을 불러온 후, Splitter을 사용하여 내용을 분할합니다.

이렇게 쪼갠 내용들은 문서들로 취급합니다.

저는 쪼갠 문서들을 BM25 기반 검색 시스템에 넣었습니다. 

BM25는 키워드 기반의 검색 시스템으로, 유사도 기반의 VectorDB와 자주 비교되곤 합니다.

 

 

# Retriever Grader 생성 

# Data model
class GradeDocuments(BaseModel):
    binary_score: str = Field(
        description="Documents are relevant to the question, 'yes' or 'no'"
    )


# LLM with function call
llm = ChatOpenAI(model_name="gpt-4o-mini",
                 temperature=0.1,
                 openai_api_key=OPENAI_API_KEY,
                 )
structured_llm_grader = llm.with_structured_output(GradeDocuments)

# Prompt
system = """You are a grader assessing relevance of a retrieved document to a user question. \n
    It does not need to be a stringent test. The goal is to filter out erroneous retrievals. \n
    If the document contains keyword(s) or semantic meaning related to the user question, grade it as relevant. \n
    Give a binary score 'yes' or 'no' score to indicate whether the document is relevant to the question."""
grade_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", system),
        ("human", "Retrieved document: \n\n {document} \n\n User question: {question}"),
    ]
)

retrieval_grader = grade_prompt | structured_llm_grader

 

'문서 내용 적절성 체크용 chain'을  retrieval_grader이라는 이름으로 정의합니다.

chain 결과는 yes/no로 대답합니다.

 

 

# 문장 생성 파트

# Prompt
prompt = hub.pull("rlm/rag-prompt")

# Post-processing
def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

# Chain
rag_chain = prompt | llm | StrOutputParser()

# Run
generation = rag_chain.invoke({"context": docs, "question": question})

 

'문장 생성용 chain'을 generation이라는 이름으로 정의합니다.

 

 

# Hallucination Grader

# Data model
class GradeHallucinations(BaseModel):
    binary_score: str = Field(
        description="Answer is grounded in the facts, 'yes' or 'no'"
    )


# LLM grader
structured_llm_grader = llm.with_structured_output(GradeHallucinations)

# Prompt
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 = hallucination_prompt | structured_llm_grader

 

'할루시네이션 체크용 chain'을 hallucination_grader라는 이름으로 정의합니다.

chain 결과는 yes/no로 대답합니다.

 

 

# Answer Grader

# Data model
class GradeAnswer(BaseModel):
    binary_score: str = Field(
        description="Answer addresses the question, 'yes' or 'no'"
    )


# LLM grader
structured_llm_grader = llm.with_structured_output(GradeAnswer)

# Prompt
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 = answer_prompt | structured_llm_grader

 

'답변 적절성 체크용 chain'을 answer_grader라는 이름으로 정의합니다.

chain 결과는 yes/no로 대답합니다.

 

 

# Question Re-writer

# Prompt
system = """You a question re-writer that converts an input question to a better version that is optimized \n
     for vectorstore retrieval. Look at the input and try to reason about the underlying semantic intent / meaning.
     Please, write in korean language."""
re_write_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", system),
        (
            "human",
            "Here is the initial question: \n\n {question} \n Formulate an improved question.",
        ),
    ]
)

question_rewriter = re_write_prompt | llm | StrOutputParser()
question_rewriter.invoke({"question": question})

 

'사용자가 입력한 질문을 더 정교한 형태로 바꿔주는 chain'을 question_rewriter라는 이름으로 정의합니다.

 

 

# Graph

from typing import List

from typing_extensions import TypedDict


class GraphState(TypedDict):
    """
    그래프 상태 표현.

    Attributes:
        question: 질문
        generation: LLM 생성 내용
        documents: 문서 리스트
    """

    question: str
    generation: str
    documents: List[str]

 

그래프 생성에 필요한 클래스를 정의해 줍니다.

여기서는 GraphState라는 클래스를 정의함으로써 그래프 상태를 표현할 것입니다.

 

### Nodes


def retrieve(state):
    """
    문서 Retrieve 적용하기

    Args:
        state (dict): 현재 그래프 상태

    Returns:
        state (dict): retrieved 문서를 포함한 새로운 그래프 및 문서 상태
    """
    print("---RETRIEVE---")
    question = state["question"]

    # Retrieval
    documents = retriever.get_relevant_documents(question)
    return {"documents": documents, "question": question}


def generate(state):
    """
    Generate answer

    Args:
        state (dict): 현재 그래프 상태

    Returns:
        state (dict): LLM 생성 내용을 포함한 새로운 그래프 및 생성 상태
    """
    print("---GENERATE---")
    question = state["question"]
    documents = state["documents"]

    # RAG generation
    generation = rag_chain.invoke({"context": documents, "question": question})
    return {"documents": documents, "question": question, "generation": generation}


def grade_documents(state):
    """
    retrieved 문서가 질문과 연관성이 있는지 결정

    Args:
        state (dict): 현재 그래프 상태

    Returns:
        state (dict): 연관성 있는 문서만 업데이트된 상태
    """

    print("---CHECK DOCUMENT RELEVANCE TO QUESTION---")
    question = state["question"]
    documents = state["documents"]

    # 문서별 점수 산정하기
    filtered_docs = []
    for d in documents:
        score = retrieval_grader.invoke(
            {"question": question, "document": d.page_content}
        )
        grade = score.binary_score
        if grade == "yes":
            print("---GRADE: DOCUMENT RELEVANT---")
            filtered_docs.append(d)
        else:
            print("---GRADE: DOCUMENT NOT RELEVANT---")
            continue
    return {"documents": filtered_docs, "question": question}


def transform_query(state):
    """
    더 나은 질문으로 변경하기

    Args:
        state (dict): 현재 그래프 상태

    Returns:
        state (dict): 더 나은 질문으로 업데이트된 상태
    """

    print("---TRANSFORM QUERY---")
    question = state["question"]
    documents = state["documents"]

    # Re-write question
    better_question = question_rewriter.invoke({"question": question})
    return {"documents": documents, "question": better_question}


### Edges


def decide_to_generate(state):
    """
    답변을 생성할 것인지, 새로운 질문을 받을 것인지 결정하기

    Args:
        state (dict): 현재 그래프 상태

    Returns:
        str: 다음 노드 행동 결정
    """

    print("---ASSESS GRADED DOCUMENTS---")
    state["question"]
    filtered_documents = state["documents"]

    if not filtered_documents:
        # All documents have been filtered check_relevance
        # We will re-generate a new query
        print(
            "---DECISION: ALL DOCUMENTS ARE NOT RELEVANT TO QUESTION, TRANSFORM QUERY---"
        )
        return "transform_query"
    else:
        # We have relevant documents, so generate answer
        print("---DECISION: GENERATE---")
        return "generate"


def grade_generation_v_documents_and_question(state):
    """
    생성된 내용이 문서와 연관되어 있는지 확인하기

    Args:
        state (dict): 현재 그래프 상태

    Returns:
        str: 다음 노드 행동 결정
    """

    print("---CHECK HALLUCINATIONS---")
    question = state["question"]
    documents = state["documents"]
    generation = state["generation"]

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

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

 

위에서 정의한 chain들을 사용해서 진행 상태를 진단할 수 있는 함수들을 만들어 줍니다.

 

 

# Graph 만들기

workflow = StateGraph(GraphState)

# Define the nodes
workflow.add_node("retrieve", retrieve)  # retrieve
workflow.add_node("grade_documents", grade_documents)  # grade documents
workflow.add_node("generate", generate)  # generatae
workflow.add_node("transform_query", transform_query)  # transform_query

# Build graph
workflow.add_edge(START, "retrieve")
workflow.add_edge("retrieve", "grade_documents")
workflow.add_conditional_edges(
    "grade_documents",
    decide_to_generate,
    {
        "transform_query": "transform_query",
        "generate": "generate",
    },
)
workflow.add_edge("transform_query", "retrieve")
workflow.add_conditional_edges(
    "generate",
    grade_generation_v_documents_and_question,
    {
        "not supported": "generate",
        "useful": END,
        "not useful": "transform_query",
    },
)

# Compile
app = workflow.compile()

 

위에서 정의한 함수들과 클래스를 그래프 형태로 함께 엮어줍니다.

 

 

# Graph 동작

# Run
inputs = {"question": "여러분의 질문을 입력하세요. (ex. OO에 대해 알고 싶어요.)"}
for output in app.stream(inputs):
    for key, value in output.items():
      # Node
      pprint(f"Node '{key}':")
      # Optional: print full state at each node
      # pprint.pprint(value["keys"], indent=2, width=80, depth=None)
    pprint("\n---\n")

# Final generation
pprint(value["generation"])

 

자, 이제 graph를 동작할 수 있게 되었습니다.

questions에 여러분이 궁금한 질문들을 넣어 주세요.

 

논문에 비해 약식으로 구현되었지만, 전반적인 Self-RAG 구조를 알아보기에는 큰 무리가 없다고 생각합니다.

Self-RAG의 핵심이 '자기 평가/자아 성찰'이라는 점이 잘 와닿죠?

 

 

● 10월 회고

이번 달에는 여러 의미로 일이 정말 많았네요.

덕분에 더 강해질 수 있었습니다.

일희일비하지 않고 끝까지 잘 마무리할 수 있도록 최선을 다하겠습니다.

 

마지막까지 과정에 충실한 태도로 임하겠습니다.

 

 

+) 부족한 부분이 있으면 댓글로 말씀해 주세요! 겸허한 마음으로 더 공부하겠습니다.