---

목차
1. 작성 개요
2. RAG의 기본적인 형태
3. 관련성 체크 노드
4. 웹 검색 노드
5. 쿼리 재작성 노드

---

1. 작성 개요

Agent를 중심으로 LangGraph를 학습했는데, LangGraph를 통해 RAG도 잘 다룰 수 있는 방법을 습득하여, 두 가지 기술을 적절히 조합하는 방법에 대해서도 학습할 필요가 있다. 또한 이전에 올린 게시물에 flow engineering이라는 게시물을 올렸는데, 그 게시물에 이어 구조적인 부분을 이해하기 쉽도록 기본 형태와 노드를 기준으로 설명하려고 한다.

다음 게시물은 테디노트님의 유료 강의를 듣고 복습을 위해 작성한 것으로 게시물이 아쉬우시다면 유료 강의 구매해서 들으시면 된다.

2. RAG의 기본적인 형태

1) 상태 정의

from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages


# GraphState 상태 정의
class GraphState(TypedDict):
    question: Annotated[str, "Question"]  # 질문
    context: Annotated[str, "Context"]  # 문서의 검색 결과
    answer: Annotated[str, "Answer"]  # 답변
    messages: Annotated[list, add_messages]  # 메시지(누적되는 list)

주목 할 만한 부분은 context다 Annotated를 보면 Str로 타입이 설정되어 있는데, 보통 Context를 아무런 조치없이 가져오면, Document들이 리스트에 담겨오는 경우가 많다. 이는 검색한 결과를 Str로 일종의 파싱을 진행한다는 것을 의미한다.

Str로 파싱하는 작업은 보통 xml형태나 LLM이 이해하기 좋은 방식으로 파싱하는 경우가 더 좋은 LLM 성능을 만드는데 도움이 될 수 있다. 그렇게 진행하는 이유는 토큰 효율성과 Document 간 overlap 되는 부분이 생기지 않도록 방지해 LLM이 이해하기 더 쉽기 때문이다.

2) 노드 정의

# 문서 검색 노드
def retrieve_document(state: GraphState) -> GraphState:
    # 질문을 상태에서 가져옵니다.
    latest_question = state["question"]

    # 문서에서 검색하여 관련성 있는 문서를 찾습니다.
    retrieved_docs = pdf_retriever.invoke(latest_question)

    # 검색된 문서를 형식화합니다.(프롬프트 입력으로 넣어주기 위함)
    retrieved_docs = format_docs(retrieved_docs)

    # 검색된 문서를 context 키에 저장합니다.
    return {"context": retrieved_docs}


# 답변 생성 노드
def llm_answer(state: GraphState) -> GraphState:
    # 질문을 상태에서 가져옵니다.
    latest_question = state["question"]

    # 검색된 문서를 상태에서 가져옵니다.
    context = state["context"]

    # 체인을 호출하여 답변을 생성합니다.
    response = pdf_chain.invoke(
        {
            "question": latest_question,
            "context": context,
            "chat_history": messages_to_history(state["messages"]),
        }
    )
    # 생성된 답변, (유저의 질문, 답변) 메시지를 상태에 저장합니다.
    return {
        "answer": response,
        "messages": [("user", latest_question), ("assistant", response)],
    }

기본적인 형태의 RAG를 수행하기 때문에 문서 검색 노드와 답변 생성 노드 2개로 구성된다. 문서 검색 노드에서 작성된 모듈들은 테디노트님이 해당 과업을 위해 만든 파일로 실제 작업에서는 개인의 필요에 따라 더 구체적으로 알아서 만들 필요가 있다.

3) 그래프 생성

from langgraph.graph import END, StateGraph
from langgraph.checkpoint.memory import MemorySaver

# 그래프 생성
workflow = StateGraph(GraphState)

# 노드 정의
workflow.add_node("retrieve", retrieve_document)
workflow.add_node("llm_answer", llm_answer)

# 엣지 정의
workflow.add_edge("retrieve", "llm_answer")  # 검색 -> 답변
workflow.add_edge("llm_answer", END)  # 답변 -> 종료

# 그래프 진입점 설정
workflow.set_entry_point("retrieve")

# 체크포인터 설정
memory = MemorySaver()

# 컴파일
app = workflow.compile(checkpointer=memory)

그래프는 그래프를 생성하고 엣지를 연결해 간단하게 만들 수 있다.

3. 관련성 체크 노드

그냥 RAG만 수행하게 되면 적절한 context를 참고하여 결과를 반환했는지 확인할 수 없기 때문에, 관련성 체크 노드를 이용해 결과를 확인해 보아야 한다. 이를 위해 관련성 체크 노드를 만들어 그래프가 문서 검색 노드 -> 관련성 체크 노드 -> 답변 생성 노드로 동작하게 하고, 관련성이 없으면 다시 한번 문서를 검색할 수 있도록 하는 구조로 만들어야 한다.

1) 상태 정의

from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages


# GraphState 상태 정의
class GraphState(TypedDict):
    question: Annotated[str, "Question"]  # 질문
    context: Annotated[str, "Context"]  # 문서의 검색 결과
    answer: Annotated[str, "Answer"]  # 답변
    messages: Annotated[list, add_messages]  # 메시지(누적되는 list)
    relevance: Annotated[str, "Relevance"]  # 관련성

다음 부분에서는 관련 성을 체크를 해야하니 relevance 부분만 추가 되었다.

2) 추가된 노드

from langchain_openai import ChatOpenAI
from langchain_teddynote.evaluator import GroundednessChecker
from langchain_teddynote.messages import messages_to_history
from rag.utils import format_docs

# 관련성 체크 노드
def relevance_check(state: GraphState) -> GraphState:
    # 관련성 평가기를 생성합니다.
    question_retrieval_relevant = GroundednessChecker(
        llm=ChatOpenAI(model="gpt-4o-mini", temperature=0), target="question-retrieval"
    ).create()

    # 관련성 체크를 실행("yes" or "no")
    response = question_retrieval_relevant.invoke(
        {"question": state["question"], "context": state["context"]}
    )

    print("==== [RELEVANCE CHECK] ====")
    print(response.score)

    # 참고: 여기서의 관련성 평가기는 각자의 Prompt 를 사용하여 수정할 수 있습니다. 여러분들의 Groundedness Check 를 만들어 사용해 보세요!
    return {"relevance": response.score}


def is_relevant(state: GraphState) -> GraphState:
    if state["relevance"] == "yes":
        return "relevant"
    else:
        return "not relevant"

여기서는 관련성 체크 노드를 위한 함수와 이를 보조하기 위한 함수 하나를 만들었다. is_relevant는 state에서 관련성 관련 결과를 바탕으로 관련성이 있는지 없는지를 전달하기 위한 함수다.

여기서는 테디 노트님이 만드신 모듈을 이용했지만, 보통은 langchain에 pre-built 되어 있는 groundness checker를 사용할 수 있다. Upstage에서 만든 것이 가장 유명하고 자주 사용되고 있지만 필요에 따라 다른 곳에서 만든것을 사용할 수 있다.

3) 그래프 정의

from langgraph.graph import END, StateGraph
from langgraph.checkpoint.memory import MemorySaver

# 그래프 정의
workflow = StateGraph(GraphState)

# 노드 추가
workflow.add_node("retrieve", retrieve_document)
# 관련성 체크 노드 추가
workflow.add_node("relevance_check", relevance_check)
workflow.add_node("llm_answer", llm_answer)

# 엣지 추가
workflow.add_edge("retrieve", "relevance_check")  # 검색 -> 관련성 체크


# 조건부 엣지를 추가합니다.
workflow.add_conditional_edges(
    "relevance_check",  # 관련성 체크 노드에서 나온 결과를 is_relevant 함수에 전달합니다.
    is_relevant,
    {
        "relevant": "llm_answer",  # 관련성이 있으면 답변을 생성합니다.
        "not relevant": "retrieve",  # 관련성이 없으면 다시 검색합니다.
    },
)

workflow.add_edge("llm_answer", END)

# 그래프 진입점 설정
workflow.set_entry_point("retrieve")

# 체크포인터 설정
memory = MemorySaver()

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

여기서는 조건부 엣지와 is_relevant 함수를 통해 분기하는 부분을 만들어줄 수 있다.

4) 관련성 있는 부분이 없어 무한 루프에 빠진다면?

from langgraph.errors import GraphRecursionError
from langchain_core.runnables import RunnableConfig

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

# 질문 입력
inputs = GraphState(question="테디노트의 랭체인 튜토리얼에 대한 정보를 알려주세요.")

try:
    # 그래프 실행
    stream_graph(app, inputs, config, ["retrieve", "relevance_check", "llm_answer"])
except GraphRecursionError as recursion_error:
    print(f"GraphRecursionError: {recursion_error}")

관련 없는 내용만 계속 검색된다면, RunnableConfig를 통해 최대 재귀 횟수를 정하고, 그 이 상 돌았을 때 적절한 오류를 반환하게 해, 서비스가 죽어버리는 결과가 나오는 것을 막아야 한다.

4. 웹 검색 노드

혹시 RAG를 회사나 개인이 소유하고 있는 문서를 기반으로 진행하게 되면, 가끔 문서에 없는 내용을 질문받았을 때 적절한 검색 증강이 이루어지지 않을 수 있다. 이를 보완하기 위해 웹 검색 노드가 필요 할 수 있다.

1) 상태는 3과 동일하다.

2) 추가된 도구와 노드

from langchain_teddynote.tools.tavily import TavilySearch

# 검색 도구 생성
tavily_tool = TavilySearch()

search_query = "2024년 노벨 문학상 수상자는?"

# 다양한 파라미터를 사용한 검색 예제
search_result = tavily_tool.search(
    query=search_query,  # 검색 쿼리
    max_results=3,  # 최대 검색 결과
    format_output=True,  # 결과 포맷팅
)

# Web Search 노드
def web_search(state: GraphState) -> GraphState:
    # 검색 도구 생성
    tavily_tool = TavilySearch()

    search_query = state["question"]

    # 다양한 파라미터를 사용한 검색 예제
    search_result = tavily_tool.search(
        query=search_query,  # 검색 쿼리
        topic="general",  # 일반 주제
        max_results=6,  # 최대 검색 결과
        format_output=True,  # 결과 포맷팅
    )

    return {"context": search_result}

3) 그래프 정의

from langgraph.graph import END, StateGraph
from langgraph.checkpoint.memory import MemorySaver

# 그래프 정의
workflow = StateGraph(GraphState)

# 노드 추가
workflow.add_node("retrieve", retrieve_document)
workflow.add_node("relevance_check", relevance_check)
workflow.add_node("llm_answer", llm_answer)

# Web Search 노드 추가
workflow.add_node("web_search", web_search)

# 엣지 추가
workflow.add_edge("retrieve", "relevance_check")  # 검색 -> 관련성 체크


# 조건부 엣지를 추가합니다.
workflow.add_conditional_edges(
    "relevance_check",  # 관련성 체크 노드에서 나온 결과를 is_relevant 함수에 전달합니다.
    is_relevant,
    {
        "relevant": "llm_answer",  # 관련성이 있으면 답변을 생성합니다.
        "not relevant": "web_search",  # 관련성이 없으면 웹 검색을 수행합니다.
    },
)

workflow.add_edge("web_search", "llm_answer")  # 검색 -> 답변
workflow.add_edge("llm_answer", END)  # 답변 -> 종료

# 그래프 진입점 설정
workflow.set_entry_point("retrieve")

# 체크포인터 설정
memory = MemorySaver()

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

다음과 같이 그래프를 정의하면 문서 검색 노드 -> 관련성 체크 노드 -> 관련성이 있다면 답변 생성 노드 없다면 웹 검색 노드를 거쳐 답변 생성 노드로 간다.

5. 쿼리 재작성 노드

RAG를 잘하기 위해서는 어떤 쿼리를 만들어서 검색기들에 전달하는지가 마찬가지로 중요하다. 그러나 유저들은 보통 적절한 쿼리를 제공하지 않을 가능성이 높으니 쿼리 재작성 노드를 만들어 추가할 필요가 있다.

1) 상태 정의

from typing import Annotated, TypedDict, List
from langgraph.graph.message import add_messages


# GraphState 상태 정의
class GraphState(TypedDict):
    question: Annotated[List[str], add_messages]  # 질문(누적되는 list)
    context: Annotated[str, "Context"]  # 문서의 검색 결과
    answer: Annotated[str, "Answer"]  # 답변
    messages: Annotated[list, add_messages]  # 메시지(누적되는 list)
    relevance: Annotated[str, "Relevance"]  # 관련성

대부분의 내용이 앞에서 정의된 것과 비슷하다. 다만 바뀐 부분은 question으로 Annotated에 정의된 데이터타입이 List[str]로 바뀌었다. 이유는 query가 바뀌면서 누적되고 추가되기 때문에 reducer와 함께 사용되어야 하기 때문이다.

2) 추가된 노드 정의

from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser

# Query Rewrite 프롬프트 정의
re_write_prompt = PromptTemplate(
    template="""Reformulate the given question to enhance its effectiveness for vectorstore retrieval.

- Analyze the initial question to identify areas for improvement such as specificity, clarity, and relevance.
- Consider the context and potential keywords that would optimize retrieval.
- Maintain the intent of the original question while enhancing its structure and vocabulary.

# Steps

1. **Understand the Original Question**: Identify the core intent and any keywords.
2. **Enhance Clarity**: Simplify language and ensure the question is direct and to the point.
3. **Optimize for Retrieval**: Add or rearrange keywords for better alignment with vectorstore indexing.
4. **Review**: Ensure the improved question accurately reflects the original intent and is free of ambiguity.

# Output Format

- Provide a single, improved question.
- Do not include any introductory or explanatory text; only the reformulated question.

# Examples

**Input**: 
"What are the benefits of using renewable energy sources over fossil fuels?"

**Output**: 
"How do renewable energy sources compare to fossil fuels in terms of benefits?"

**Input**: 
"How does climate change impact polar bear populations?"

**Output**: 
"What effects does climate change have on polar bear populations?"

# Notes

- Ensure the improved question is concise and contextually relevant.
- Avoid altering the fundamental intent or meaning of the original question.


[REMEMBER] Re-written question should be in the same language as the original question.

# Here is the original question that needs to be rewritten:
{question}
""",
    input_variables=["generation", "question"],
)

question_rewriter = (
    re_write_prompt | ChatOpenAI(model="gpt-4o-mini", temperature=0) | StrOutputParser()
)

# Query Rewrite 노드
def query_rewrite(state: GraphState) -> GraphState:
    latest_question = state["question"][-1].content
    question_rewritten = question_rewriter.invoke({"question": latest_question})
    return {"question": question_rewritten}

추가된 노드는 LLM을 사용하여 만든 노드다. 다음과 같은 형식의 노드는 계속해서 좋은 프롬프트가 없을지 더 고민해 볼 필요가 있을 것 같다.

3) 그래프 정의

from langgraph.graph import END, StateGraph
from langgraph.checkpoint.memory import MemorySaver

# 그래프 정의
workflow = StateGraph(GraphState)

# 노드 추가
workflow.add_node("retrieve", retrieve_document)
workflow.add_node("relevance_check", relevance_check)
workflow.add_node("llm_answer", llm_answer)
workflow.add_node("web_search", web_search)

# Query Rewrite 노드 추가
workflow.add_node("query_rewrite", query_rewrite)

# 엣지 추가
workflow.add_edge("query_rewrite", "retrieve")  # 질문 재작성 -> 검색
workflow.add_edge("retrieve", "relevance_check")  # 검색 -> 관련성 체크

# 조건부 엣지를 추가합니다.
workflow.add_conditional_edges(
    "relevance_check",  # 관련성 체크 노드에서 나온 결과를 is_relevant 함수에 전달합니다.
    is_relevant,
    {
        "relevant": "llm_answer",  # 관련성이 있으면 답변을 생성합니다.
        "not relevant": "web_search",  # 관련성이 없으면 다시 검색합니다.
    },
)

workflow.add_edge("web_search", "llm_answer")  # 검색 -> 답변
workflow.add_edge("llm_answer", END)  # 답변 -> 종료

# 그래프 진입점 설정
workflow.set_entry_point("query_rewrite")

# 체크포인터 설정
memory = MemorySaver()

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

구조는 맨 처음에 쿼리를 재작성 해주는 부분이 추가된 것이다.