LangGraph

1. LangGraph가 필요한 이유

1.1. 기술의 발전은 갈증에서 시작된다

기술의 발전은 어떤 갈증으로부터 시작된다.
“이런 걸 할 수 있었으면 좋겠는데, 지금 방식으로는 불편하다.”
이 문제의식이 새로운 기술을 만든다.

RAG 기술이 강력해지면서 개발자들은 다음과 같은 고민을 하게 되었다:

• 생성된 답변이 정확한가?
• 문서에 없는 정보는 답변하지 않게 만들 수 없을까?
• 불충분한 정보는 외부 검색으로 보충할 수 없을까?
• 답변의 신뢰성을 평가할 방법이 있을까?

이 갈증을 해결하기 위해 LangGraph가 등장했다.

2. Naive RAG의 한계

2.1. 단방향 파이프라인의 구조적 문제

기존 RAG 구조는 다음과 같이 단방향 파이프라인이다:

질문 → 문서 로더 → 텍스트 분할 → 임베딩 생성 → 벡터 저장소
문서 검색 → LLM 호출 → 답변 생성 → 종료

문제가 되는 이유:

• 고정된 설정: 실행 중에 청크 크기/검색 방식 등 동적 변경 불가.
• 신뢰할 수 없는 LLM: 한 번의 실행에서 모든 것이 정확히 맞아야 함.
• 역방향 불가능: 중간 단계에서 오류가 나도 되돌릴 수 없음.
• 복잡성 증가: 문제 해결하려고 체인을 덧붙이면 구조가 기형적으로 늘어남.
• 디버깅 어려움: 어디서 문제가 발생했는지 추적이 어려움.

2.2. 실제 사례

질문:
“생성형 AI 가우스를 만든 회사의 2023년 매출액은?”

1. Naive RAG
   문서에는 “가우스는 삼성전자가 만들었다”만 존재 → 매출액 없음
   결과: “매출액 정보를 찾을 수 없습니다.”
2. 웹 검색 체인 추가
   웹 검색: “삼성전자 2023년 매출액 250조원”
   결과: 정상 답변
3. 검색 결과 오염
   웹 검색: “삼성SDS 매출액 20조원”
   결과: 잘못된 답변 생성

기존 RAG에서는 이런 오류를 자동으로 제어하거나 재수정하기 어렵다.

2.3. 기존 RAG의 본질적 문제 요약

 

항목	설명
단방향 구조	한 번 실행되면 되돌릴 수 없음
고정 설정	동적 변경 불가
한 번에 성공 필요	하나라도 실패하면 전체 실패
복잡성 증가	체인이 길어지고 관리 어려움
디버깅 어려움	문제 지점 추적 어려움

3. LangGraph 핵심 개념

3.1. LangGraph의 철학

기존 RAG 사고방식:
“체인을 잘 설계해서 한 번에 성공하게 하자.”

LangGraph 사고방식:
“모든 단계를 독립 노드로 쪼개고, 실행 중 필요에 따라 분기하고 재실행하자.”

3.2. LangGraph의 구성 요소

1. Node(노드)
   각 작업 단위를 함수로 관리
   예: 검색 노드, 평가 노드, 생성 노드
2. Edge(엣지)
   노드 간 흐름을 연결

• 일반 엣지
• 조건부 엣지(가장 중요)

1. State(상태)
   노드 간 데이터를 전달하는 컨테이너
   각 노드는 이전 로직을 몰라도 State를 통해 필요한 정보를 읽음

4. State

4.1. State란?

State는 노드 간 데이터를 전달하는 택배 상자와 같다.

A 노드 → 상태 저장 → B 노드
B 노드는 상태를 읽고 다음 작업을 수행한다.

4.2. State 정의(TypedDict)

 

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

class GraphState(TypedDict):
    question: str
    context: str
    answer: str
    score: str
    messages: Annotated[List, add_messages]

4.3. Reducer(add_messages)

List 값을 새로 덮어쓰지 않고 자동으로 이어 붙여준다.

5. Node

5.1. Node는 함수

def retrieve(state):
    docs = retriever.get(state["question"])
    return {"context": docs}

5.2. Node 내부는 자유

Node 내부는 파이썬 코드, API 호출, DB 조회 등 어떤 작업이든 가능하다.
LangChain을 반드시 쓸 필요도 없다.

6. Edge

6.1. 일반 Edge

 

graph.add_edge("A", "B")

6.2. 조건부 Edge

def route(state):
    if state["score"] == "GOOD":
        return "generate"
    return "rewrite"

graph.add_conditional_edges(
    "grade",
    route,
    {
        "generate": "generate",
        "rewrite": "rewrite"
    }
)

7. RAG 예시

7.1. 기본 RAG

질문 → Retrieval → Generation → 출력

7.2. Agentic RAG(순환형)

질문
→ 검색
→ 평가
→ (부족함) → 질문 재작성 → 다시 검색
→ 생성
→ 환각 검사
→ (부족함) → 웹 검색
→ 재생성
→ 최종 출력

8. LangGraph의 핵심: 상태 전이와 순환

평가가 BAD면 다음 단계:

1. Query Rewrite
2. 검색 재실행
3. 생성 재실행
4. 재평가 → GOOD이면 종료

이 모든 과정이 자동 순환된다.

9. LangGraph 주요 기능

9.1. Conditional Edge

동적 라우팅의 핵심.

9.2. Human-in-the-Loop

중간에 사람이 승인 여부를 판단하도록 구성할 수 있음.

9.3. Checkpointer

각 단계의 상태를 저장 → 특정 시점으로 돌아가 재실행(Replaying) 가능.

10. LangGraph vs LangChain

10.1. LangChain

• LLM, 프롬프트, Retriever 등을 연결하는 체인 기반 프레임워크
• 단방향 구조 중심

10.2. LangGraph

• 복잡한 워크플로우를 그래프 기반으로 설계
• 상태 기반 제어
• 조건부 분기, 순환, 재실행 최적화

10.3. 비교 요약

항목	LangChain	LangGraph
목적	LLM 통합	워크플로우 제어
구조	선형 체인	그래프
조건부 분기	어려움	매우 쉬움
순환 구조	거의 불가능	기본 지원
상태 관리	단순 메모리	TypedDict 기반
복잡 RAG	제한적	최적화됨

10.4. 실제 코드 비교

LangChain RAG

 

from langchain.chains import RetrievalQA
from langchain.llms import OpenAI

class SimpleRAG:
    def __init__(self):
        self.llm = OpenAI()
        self.qa_chain = RetrievalQA.from_chain_type(
            llm=self.llm,
            retriever=self.retriever,
            chain_type="stuff"
        )

    def ask(self, question):
        return self.qa_chain.run(question)

LangGraph RAG

 

from langgraph.graph import StateGraph, END
from typing import TypedDict

class RAGState(TypedDict):
    question: str
    documents: str
    answer: str
    score: str

class AdaptiveRAG:
    def __init__(self):
        self.graph = StateGraph(RAGState)
        self._build_graph()

    def _build_graph(self):
        self.graph.add_node("retrieve", self.retrieve)
        self.graph.add_node("grade", self.grade)
        self.graph.add_node("generate", self.generate)
        self.graph.add_node("rewrite", self.rewrite)

        self.graph.add_edge("retrieve", "grade")

        self.graph.add_conditional_edges(
            "grade",
            self.should_continue,
            {
                "generate": "generate",
                "rewrite": "rewrite"
            }
        )

        self.graph.add_edge("rewrite", "retrieve")
        self.graph.add_edge("generate", END)

        self.graph.set_entry_point("retrieve")
        self.compiled = self.graph.compile()

    def should_continue(self, state):
        if state["score"] == "GOOD":
            return "generate"
        return "rewrite"