LangGraph : UseCase (CRAG)

김지우·2025년 1월 29일
langgraphrag

목차

  1. 작성 개요
  2. 기본 PDF 기반 Retrieval Chain 생성
  3. 검색된 문서의 관련성 평가
  4. 답변 생성 체인
  5. 쿼리 재작성
  6. 웹 검색 도구
  7. 상태 정의
  8. 노드 정의
  9. 조건부 엣지
  10. 그래프 정의 및 실행

1. 작성 개요

다음 게시물은 LangGraph를 활용하여 다양한 구조의 Advanced RAG를 학습하기 위한 게시물이다. 다음 게시물에서는 CRAG(Corrective RAG)를 담을 예정이다. Corrective RAG는 우선적으로 PDF와 같은 문서에서 정보를 검색하고, 문서들과 쿼리의 관련성이 임계치보다 낮을 때, 웹검색을 진행해 데이터를 보강하여 답변을 생성하는 RAG 테크닉을 말한다.

자율적으로 도구를 선택하는 것이나, 결과를 바탕으로 쿼리를 수정하는 것에 초점이 맞춰져 있는 것이 아니라, 실제 결과를 필요하다면 다양한 소스에서 가져올 수 있도록 하는 증강하는 문서의 질을 충분히 보장하는 것에 초점을 맞춘 방법론이라고 생각한다.

다음 글의 경우 테디노트남의 유료 강의를 듣고 학습한 게시물로 게시물의 내용이 부족하다고 생각한다면 테디노트님의 유료강의를 결제하는 것을 권장한다.

2. 기본 PDF 기반 Retrieval Chain 생성

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_teddynote.models import get_model_name, LLMs
from pydantic import BaseModel, Field

# 모델 이름 가져오기
MODEL_NAME = get_model_name(LLMs.GPT4)


# 검색된 문서의 관련성 여부를 이진 점수로 평가하는 데이터 모델
class GradeDocuments(BaseModel):
    """A binary score to determine the relevance of the retrieved document."""

    # 문서가 질문과 관련이 있는지 여부를 'yes' 또는 'no'로 나타내는 필드
    binary_score: str = Field(
        description="Documents are relevant to the question, 'yes' or 'no'"
    )


# LLM 초기화
llm = ChatOpenAI(model=MODEL_NAME, temperature=0)

# GradeDocuments 데이터 모델을 사용하여 구조화된 출력을 생성하는 LLM
structured_llm_grader = llm.with_structured_output(GradeDocuments)

# 시스템 프롬프트 정의
system = """You are a grader assessing relevance of a retrieved document to a user question. \n 
    If the document contains keyword(s) or semantic meaning related to the 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 평가기 초기화
retrieval_grader = grade_prompt | structured_llm_grader

우선 문서가 관련이 있는 지 없는지를 평가하는 함수 혹은 객체는 기본적으로 LLM으로 만드는 것을 전제로 한다. 체인을 만들어 yes 혹은 no 라는 답변을 받을 수 있도록 한다. 여기서는 문서의 집합이 아닌 1개의 단일 문서에 대한 평가를 수행한다.

관련성은 여기서 yes or no로 반환하지만, 필요에 따라 특정한 스케일 안의(0~1 혹은 0 ~100) 수치로 바꾸거나 응용할 수 도 있다. 

question = "삼성전자가 개발한 생성AI 에 대해 설명하세요."

# 문서 검색
docs = pdf_retriever.invoke(question)

# 검색된 문서 중 1번 index 문서의 페이지 내용을 추출
doc_txt = docs[1].page_content

# 검색된 문서와 질문을 사용하여 관련성 평가를 실행하고 결과 출력
print(retrieval_grader.invoke({"question": question, "document": doc_txt}))

그럼 다음과 같은 방식으로 문서에 대한 평가를 진행할 수 있다.

4. 답변 생성 체인

from langchain import hub
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI


# LangChain Hub에서 RAG 프롬프트를 가져와 사용
prompt = hub.pull("teddynote/rag-prompt")

# LLM 초기화
llm = ChatOpenAI(model_name=MODEL_NAME, temperature=0)


# 문서 포맷팅
def format_docs(docs):
    return "\n\n".join(
        [
            f'<document><content>{doc.page_content}</content><source>{doc.metadata["source"]}</source><page>{doc.metadata["page"]+1}</page></document>'
            for doc in docs
        ]
    )


# 체인 생성
rag_chain = prompt | llm | StrOutputParser()


# 체인 실행 및 결과 출력
generation = rag_chain.invoke({"context": format_docs(docs), "question": question})
print(generation)

나중에 노드에 답변하는 노드를 만들어야 하기 때문에, 문서 기반으로 답변하는 체인을 만든다. 구조는 가장 기본적인 naive rag다.

5. 쿼리 재작성

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

# LLM 설정
llm = ChatOpenAI(model=MODEL_NAME, temperature=0)

# Query Rewrite 시스템 프롬프트
system = """You a question re-writer that converts an input question to a better version that is optimized 
for web search. Look at the input and try to reason about the underlying semantic intent / meaning."""

# 프롬프트 정의
re_write_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", system),
        (
            "human",
            "Here is the initial question: \n\n {question} \n Formulate an improved question.",
        ),
    ]
)

# Question Re-writer 체인 초기화
question_rewriter = re_write_prompt | llm | StrOutputParser()

다음 구조에서는 쿼리를 재작성 하는 노드를 위한 체인을 만든다. 쿼리를 왜 재작성해야하나? LLM에 좋은 답변을 유도하기 위한 쿼리와 웹 검색 도구에 적절한 결과를 반환하기 위한 쿼리는 다소 다르다. 어떤 웹 검색 도구를 이용하는지에 따라서도 다르지만, 어떤 형태, 문장형인지 , 단답형인지에 따른 결과도 다소 달랐던 경험이 있다. 쿼리를 재작성해서 질문을 구체화 시킬 수 도 있을거라고 본다. 따라서 그 검색기의 특징을 파악한다면 상기 프롬프트를 구체화 시킬 필요도 있다고 생각한다.

6. 웹 검색 도구

# 웹 검색 도구 초기화
from langchain_teddynote.tools.tavily import TavilySearch

# 최대 검색 결과를 3으로 설정
web_search_tool = TavilySearch(max_results=3)

웹 검색 도구용으로 호출해둔다.

7. 상태 정의

from typing import Annotated, List
from typing_extensions import TypedDict


# 상태 정의
class GraphState(TypedDict):
    question: Annotated[str, "The question to answer"]
    generation: Annotated[str, "The generation from the LLM"]
    web_search: Annotated[str, "Whether to add search"]
    documents: Annotated[List[str], "The documents retrieved"]

LangGraph니까 상태를 정의해 둔다. 당연히 상태를 정의 한다.

8. 노드 정의

from langchain.schema import Document


# 문서 검색 노드
def retrieve(state: GraphState):
    print("\n==== RETRIEVE ====\n")
    question = state["question"]

    # 문서 검색 수행
    documents = pdf_retriever.invoke(question)
    return {"documents": documents}


# 답변 생성 노드
def generate(state: GraphState):
    print("\n==== GENERATE ====\n")
    question = state["question"]
    documents = state["documents"]

    # RAG를 사용한 답변 생성
    generation = rag_chain.invoke({"context": documents, "question": question})
    return {"generation": generation}


# 문서 평가 노드
def grade_documents(state: GraphState):
    print("\n==== [CHECK DOCUMENT RELEVANCE TO QUESTION] ====\n")
    question = state["question"]
    documents = state["documents"]

    # 필터링된 문서
    filtered_docs = []
    relevant_doc_count = 0

    for d in documents:
        # Question-Document 의 관련성 평가
        score = retrieval_grader.invoke(
            {"question": question, "document": d.page_content}
        )
        grade = score.binary_score

        if grade == "yes":
            print("==== [GRADE: DOCUMENT RELEVANT] ====")
            # 관련 있는 문서를 filtered_docs 에 추가
            filtered_docs.append(d)
            relevant_doc_count += 1
        else:
            print("==== [GRADE: DOCUMENT NOT RELEVANT] ====")
            continue

    # 관련 문서가 없으면 웹 검색 수행
    web_search = "Yes" if relevant_doc_count == 0 else "No"
    return {"documents": filtered_docs, "web_search": web_search}


# 쿼리 재작성 노드
def query_rewrite(state: GraphState):
    print("\n==== [REWRITE QUERY] ====\n")
    question = state["question"]

    # 질문 재작성
    better_question = question_rewriter.invoke({"question": question})
    return {"question": better_question}


# 웹 검색 노드
def web_search(state: GraphState):
    print("\n==== [WEB SEARCH] ====\n")
    question = state["question"]
    documents = state["documents"]

    # 웹 검색 수행
    docs = web_search_tool.invoke({"query": question})
    # 검색 결과를 문서 형식으로 변환
    web_results = "\n".join([d["content"] for d in docs])
    web_results = Document(page_content=web_results)
    documents.append(web_results)

    return {"documents": documents}

노드는 총 5개다. 다음 노드는 우선 문서 검색 노드로 시작하여 문서 평가 노드로 이동한다. 문서 평가에서 지정해 놓은 개수의 문서만큼 관련성을 확보하지 못한다면 웹검색을 위해 쿼리 재작성을 위한 노드로 이동한다. 쿼리를 재작성한다면 웹 검색 노드로 이동하고 이후 결과를 반환하는 노드로 이동한다.

반명 지정해 놓은 개수의 문서만큼 관련성을 확보한다면 바로 결과를 반환하는 노드로 이동해 결과를 반환한다.

9. 조건부 엣지

def decide_to_generate(state: GraphState):
    # 평가된 문서를 기반으로 다음 단계 결정
    print("==== [ASSESS GRADED DOCUMENTS] ====")
    # 웹 검색 필요 여부
    web_search = state["web_search"]

    if web_search == "Yes":
        # 웹 검색으로 정보 보강이 필요한 경우
        print(
            "==== [DECISION: ALL DOCUMENTS ARE NOT RELEVANT TO QUESTION, QUERY REWRITE] ===="
        )
        # 쿼리 재작성 노드로 라우팅
        return "query_rewrite"
    else:
        # 관련 문서가 존재하므로 답변 생성 단계(generate) 로 진행
        print("==== [DECISION: GENERATE] ====")
        return "generate"

노드를 연결하기 전 관련성이 없어 정보 보강이 필요한지 그렇지 않은지를 판단하기 위한 조건부 엣지를 만들어야 하고, 이를 위한 함수를 정의해야 한다.

10. 그래프 정의 및 실행

from langgraph.graph import END, StateGraph, START

# 그래프 상태 초기화
workflow = StateGraph(GraphState)

# 노드 정의
workflow.add_node("retrieve", retrieve)
workflow.add_node("grade_documents", grade_documents)
workflow.add_node("generate", generate)
workflow.add_node("query_rewrite", query_rewrite)
workflow.add_node("web_search_node", web_search)

# 엣지 연결
workflow.add_edge(START, "retrieve")
workflow.add_edge("retrieve", "grade_documents")

# 문서 평가 노드에서 조건부 엣지 추가
workflow.add_conditional_edges(
    "grade_documents",
    decide_to_generate,
    {
        "query_rewrite": "query_rewrite",
        "generate": "generate",
    },
)

# 엣지 연결
workflow.add_edge("query_rewrite", "web_search_node")
workflow.add_edge("web_search_node", "generate")
workflow.add_edge("generate", END)

# 그래프 컴파일
app = workflow.compile()

from langchain_core.runnables import RunnableConfig
from langchain_teddynote.messages import stream_graph, invoke_graph, random_uuid

# config 설정(재귀 최대 횟수, thread_id)
config = RunnableConfig(recursion_limit=20, configurable={"thread_id": random_uuid()})

# 질문 입력
inputs = {
    "question": "삼성전자가 개발한 생성형 AI 의 이름은?",
}

# 스트리밍 형식으로 그래프 실행
stream_graph(
    app,
    inputs,
    config,
    ["retrieve", "grade_documents", "query_rewrite", "web_search_node", "generate"],
)

다음과 같이 엣지를 연결하고 그래프를 실행하면 완성!

profile
김지우
프로그래밍 기록 + 공부 기록
이전 포스트

LangGraph : UseCase (Simulation)

다음 포스트

LangGraph: UseCase (Self RAG)

0개의 댓글