LangGraph로 쉽게 구현하는 RAG 파이프라인: LLM 정확도를 높이는 비결

최근 자연어 처리 분야에서 대규모 언어 모델(LLM)의 활용이 증가하면서, Retrieval-Augmented Generation(RAG)이 주목받고 있습니다. 그러나 RAG를 단순히 단방향으로 한 번만 처리하면 여러 제약 사항이 발생하며, LLM의 정확도도 떨어질 수 있습니다. 이러한 문제를 해결하기 위해 순환 처리 및 재처리가 중요하며, 이를 손쉽게 구현할 수 있는 도구로 LangGraph가 등장했습니다. 이번 블로그에서는 LangGraph의 핵심 개념과 이를 활용한 RAG 파이프라인 구현 방법을 자세히 살펴보겠습니다.

---

왜 순환 처리와 재처리가 필요한가?

단방향 RAG 처리에서는 사용자 입력에 대해 한 번의 검색과 생성만 이루어집니다. 하지만 복잡한 질문이나 문맥이 필요한 경우, 단순한 한 번의 처리로는 충분한 답변을 제공하기 어렵습니다. 예를 들어, 사용자의 질문이 모호하거나 추가 정보가 필요한 경우, 모델은 부정확하거나 불완전한 답변을 생성할 수 있습니다.

순환 처리와 재처리를 통해 모델은 이전의 출력이나 상태를 기반으로 추가적인 검색이나 생성을 수행할 수 있습니다. 이는 LLM의 이해도를 높이고, 사용자에게 더욱 정확하고 유용한 정보를 제공할 수 있게 합니다.

반응형

---

LangGraph란 무엇인가?

LangGraph는 복잡한 LLM 애플리케이션을 구축하기 위한 그래프 기반의 프레임워크입니다. 노드(Node)와 엣지(Edge)를 사용하여 데이터 흐름을 시각화하고 제어할 수 있으며, 조건부 엣지(Conditional Edge), Human-in-the-loop, Checkpointer, State 등의 기능을 통해 유연하고 강력한 파이프라인을 구성할 수 있습니다.

LangGraph의 주요 구성 요소

1. 노드(Node): 데이터 처리를 담당하는 기본 단위입니다. 각 노드는 특정 기능을 수행하며, 예를 들어 텍스트 생성, 검색, 데이터 변환 등이 가능합니다.
2. 엣지(Edge): 노드 간의 연결을 나타내며, 데이터 흐름을 정의합니다.
3. 조건부 엣지(Conditional Edge): 특정 조건에 따라 노드 간의 연결을 동적으로 변경할 수 있습니다. 이를 통해 복잡한 로직을 구현할 수 있습니다.
4. Human-in-the-loop: 사람의 판단이나 입력을 받을 수 있는 지점을 설정하여, 모델의 한계를 보완하고 정확도를 높일 수 있습니다.
5. Checkpointer: 특정 지점에서 상태를 저장하여, 필요할 경우 이전 상태로 복원하거나 재처리를 수행할 수 있습니다.
6. State: 노드 간에 공유되는 데이터로, 컨텍스트나 중간 결과를 저장하고 전달하는 역할을 합니다.

---

LangGraph로 RAG 파이프라인 구현하기

이제 LangGraph를 사용하여 순환 처리와 재처리가 가능한 RAG 파이프라인을 구축하는 방법을 살펴보겠습니다.

1. LangGraph 설치

pip install langgraph

2. 기본 노드 정의

from langgraph import Node, Edge, Graph, State

# 검색 노드
class RetrieverNode(Node):
    def run(self, state: State):
        query = state['query']
        # 검색 로직 구현
        documents = retrieve_documents(query)
        state['documents'] = documents

# 생성 노드
class GeneratorNode(Node):
    def run(self, state: State):
        documents = state['documents']
        # 생성 로직 구현
        answer = generate_answer(documents)
        state['answer'] = answer

3. 조건부 엣지와 재처리 구현

# 평가 노드
class EvaluatorNode(Node):
    def run(self, state: State):
        answer = state['answer']
        # 평가 로직 구현 (예: 정확도 판단)
        if is_answer_satisfactory(answer):
            state['satisfied'] = True
        else:
            state['satisfied'] = False

# 그래프 구성
graph = Graph()

retriever = RetrieverNode()
generator = GeneratorNode()
evaluator = EvaluatorNode()

graph.add_nodes([retriever, generator, evaluator])

# 노드 연결
graph.add_edge(Edge(retriever, generator))
graph.add_edge(Edge(generator, evaluator))

# 조건부 엣지로 재처리 구현
def feedback_condition(state):
    return not state['satisfied']

graph.add_edge(Edge(evaluator, retriever, condition=feedback_condition))

4. Human-in-the-loop 및 Checkpointer 적용

# Human-in-the-loop 노드
class HumanFeedbackNode(Node):
    def run(self, state: State):
        answer = state['answer']
        # 사용자로부터 피드백 받기
        feedback = get_user_feedback(answer)
        state['feedback'] = feedback

# 체크포인터 노드
class CheckpointNode(Node):
    def run(self, state: State):
        # 현재 상태 저장
        save_state(state)

# 그래프에 추가
human_feedback = HumanFeedbackNode()
checkpoint = CheckpointNode()

graph.add_node(human_feedback)
graph.add_node(checkpoint)

graph.add_edge(Edge(evaluator, human_feedback, condition=lambda state: not state['satisfied']))
graph.add_edge(Edge(human_feedback, retriever))
graph.add_edge(Edge(generator, checkpoint))

5. 그래프 실행

# 초기 상태 설정
initial_state = State({'query': '사용자의 질문'})

# 그래프 실행
graph.run(initial_state)

728x90

---

LangGraph를 활용하면 복잡한 RAG 파이프라인을 손쉽게 구현할 수 있습니다. 순환 처리와 재처리를 통해 LLM의 한계를 극복하고, 더욱 정확하고 신뢰성 있는 응답을 제공할 수 있습니다. 또한 Human-in-the-loop와 Checkpointer를 통해 인간의 판단을 결합하고 상태를 관리함으로써, 실무에서 필요한 유연성과 안정성을 확보할 수 있습니다.