RAG- LangGraph 시작하기 (state+ node + edge = graph)

LangGraph를 처음 접하는 사람들을 위해 정리해보았다.
요즘 RAG기반 시스템을 구축하다 보면 "LangChain 만으로는 복잡한 워크플로우를 제어하기 어렵다"는 문제를 자주 마주치게 된다. 이때 유용하게 등장한 것이 LangGraph이다. 

---

LangGraph란 무엇인가?

한 줄로 요약하면 LangChain 위에서 동작하는 "상태 기계(State Machine)" 기반 LLM 워크플로우 프레임워크이다.
즉, LLM을 이용한 파이프라인을 그래프 구조로 모델링하고, 각 단계의 상태와 전이를 명확히 정의할 수 있다.

보통 LangChain은 chain이나 agent 같은 흐름 단위로 로직을 묶는다. 하지만 현실의 애플리케이션에서는 다음과 같은 시나리오가 자주 등장한다:

• 질문 → 검색 → 요약 → 답변 생성
• 사용자 입력에 따라 다른 경로로 분기
• 실패 시 재시도 / 보정 루프 필요
• 여러 개의 Agent가 상호작용

이런 경우 단순한 직선형 체인보다 상태 기반 그래프 구조가 훨씬 직관적이고 관리하기 쉽다. LangGraph는 바로 이런 복잡한 흐름을 다루기 위한 도구이다.

 

 

LangChain VS LangGraph

구분	LangChain	LangGraph
흐름 구조	직선형 체인	그래프 기반 (분기·루프 지원)
분기/루프	직접 코드 구현	조건부 엣지, 루프 지원
상태 관리	단계별 입력/출력 전달	전체 컨텍스트(State) 공유
적합한 상황	간단한 직선형 태스크	복잡한 워크플로우, 프로덕션

 

---

핵심 개념

LangGraph를 이해하기 위해 알아야 할 주요 요소는 다음과 같다.

1. State

• 현재 워크플로우에서 저장해야 할 정보를 정의.
• 예: user_question, retrieved_docs, draft_answer, retry_count 등.

2. Node

• 그래프의 실행 단위.
• 특정 작업을 수행하고 상태를 갱신한다.
• 예: LLM 호출 노드, 검색 노드, 요약 노드 등.

3. Edge

• 노드 간 전이(transition) 규칙.
• 실행 결과나 조건에 따라 다음 노드를 결정.
• 예: 답변 품질이 낮으면 ReRank 노드로 이동, 충분하면 END로 종료.

4. Graph

• 전체 파이프라인을 그래프로 정의.
• START에서 시작해서 END까지 도달하는 구조.

---

예시 코드 

간단한 RAG 플로우를 LangGraph로 표현하면 다음과 같다

[START] --> (retrieve) --> (generate) --> [END]

 

 

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

# 1) 상태 스키마: 노드들이 공유/갱신하는 컨텍스트 구조
class RAGState(TypedDict):
    question: str        # 사용자 질문
    docs: List[str]      # 검색 결과(간단히 문자열 목록으로 표현)
    answer: str          # 최종 생성 답변

# 2) 노드: 현재 상태를 입력받아 '부분 상태'를 반환(병합 대상)
def retrieve(state: RAGState):
    # 실제로는 BM25/벡터검색/하이브리드 검색이 들어갈 자리
    q = state["question"]
    docs = ["doc1", "doc2"]  # 데모용 더미 값
    return {"docs": docs}    # 부분 상태 델타(merge)

def generate(state: RAGState):
    # 직전 노드가 채운 docs를 사용해 답변을 생성
    docs = state["docs"]
    answer = f"Answer based on {docs}"
    return {"answer": answer}  # 상태의 answer 키를 갱신

# 3) 그래프 정의: 노드 등록 + 엣지(실행 경로) 연결
graph = StateGraph(RAGState)
graph.add_node("retrieve", retrieve)
graph.add_node("generate", generate)

graph.add_edge(START, "retrieve")   # 시작 -> 검색
graph.add_edge("retrieve", "generate")
graph.add_edge("generate", END)     # 생성 -> 종료

# 4) 컴파일 & 실행
app = graph.compile()
result = app.invoke({"question": "LangGraph란?"})
print(result["answer"])

 

이렇게 정의하면 각 노드가 상태를 받아서 결과를 내고, 그래프가 전이를 관리해준다.
복잡한 분기나 재시도 로직도 add_conditional_edges로 쉽게 표현할 수 있다. 하지만 현재 이코드는 단순한 직선 파이프 라인이고 조건부 분기를 추가하면 조금 더 복잡해진다

 

[START] --> (retrieve) --> (generate) --> (validate) --> (ok) --> [END]
                                \                              /           \
                                  \<---------------- (retry)       (fallback) --> (fallback) -->[END]

 

from typing import TypedDict, List, Optional
from langgraph.graph import StateGraph, END, START

# ===== 설정 =====
MAX_RETRIES = 2                 # 재시도 최대 횟수
QUALITY_THRESHOLD = 0.6         # 품질 합격 기준 (데모용)

# ===== 상태 스키마 =====
class RAGState(TypedDict, total=False):
    question: str               # 사용자 질문 (입력)
    docs: List[str]             # 검색 결과
    answer: str                 # 생성 답변
    quality: float              # 답변 품질 점수(0~1)
    retry_count: int            # 현재 재시도 횟수

# ===== 노드 구현 =====
def retrieve(state: RAGState) -> RAGState:
    # 실제로는 BM25/벡터/하이브리드 검색이 들어가는 자리
    q = state["question"]
    # 데모: 질문 키워드에 따라 문서 다양화된 척…
    docs = [f"doc about {q}", "generic background doc"]
    return {"docs": docs}

def generate(state: RAGState) -> RAGState:
    docs = state["docs"]
    # 데모: 문서 길이에 비례해 답변 길이 구성
    answer = f"Answer based on {len(docs)} docs: {docs}"
    return {"answer": answer}

def validate(state: RAGState) -> RAGState:
    """
    간단한 품질 점수 산출:
    - 규칙 기반 더미 스코어: 답변 길이와 'about' 키워드 존재 여부를 반영
    실제로는 LLM 평가/규칙/스코어러를 붙이면 됨.
    """
    ans = state.get("answer", "")
    has_about = ("about" in ans)
    score = min(1.0, 0.3 + 0.05 * len(ans) + (0.3 if has_about else 0.0))
    return {"quality": score}

def fallback(state: RAGState) -> RAGState:
    """
    재시도 한도를 넘겼을 때의 마지막 안전망.
    - 보수적인 템플릿/FAQ/관리자 연결 안내 등으로 구성 가능
    """
    safe = (
        "죄송합니다. 현재 질문에 대해 신뢰할 수 있는 답변을 확신하기 어려움. "
        "추가 키워드로 다시 물어보거나, 관련 문서를 더 제공해주면 좋음."
    )
    return {"answer": safe}

# ===== 분기 라우터 =====
def route_after_validate(state: RAGState) -> str:
    """
    validate 실행 뒤 다음 경로를 결정:
    - 품질 미달 & 재시도 여유 있음 → 'retry' (다시 retrieve로)
    - 품질 미달 & 재시도 한도 초과 → 'fallback'
    - 품질 합격 → 'ok'
    """
    quality = state.get("quality", 0.0)
    retry_count = state.get("retry_count", 0)

    if quality >= QUALITY_THRESHOLD:
        return "ok"
    else:
        if retry_count < MAX_RETRIES:
            # 다음 루프에서 retry_count를 올리기 위해 증가시켜 반환
            state["retry_count"] = retry_count + 1
            return "retry"
        else:
            return "fallback"

# ===== 그래프 정의 =====
graph = StateGraph(RAGState)

graph.add_node("retrieve", retrieve)
graph.add_node("generate", generate)
graph.add_node("validate", validate)
graph.add_node("fallback", fallback)

graph.add_edge(START, "retrieve")
graph.add_edge("retrieve", "generate")
graph.add_edge("generate", "validate")

# validate 이후 조건부 분기: retry / ok / fallback
graph.add_conditional_edges(
    "validate",
    route_after_validate,
    {
        "retry": "retrieve",   # 다시 검색→생성 루프
        "ok": END,             # 종료
        "fallback": "fallback" # 폴백 응답 후 종료
    },
)

graph.add_edge("fallback", END)

# ===== 컴파일 & 실행 =====
app = graph.compile()

# 초기 상태에 retry_count를 0으로 넣어 시작
result = app.invoke({"question": "LangGraph란?", "retry_count": 0})
print("[answer]", result["answer"])
print("[quality]", result.get("quality"), "[retry_count]", result.get("retry_count"))

핵심은 validate 노드에서 품질 지표를 계산하고, add_conditional_edges로 retry / ok / fallback 경로를 나누는 것 이다.

즉 여기서 (validate) 노드가 바로 라우터 분기점이다.

• retrieve → generate → validate 까지 직선 실행이다.
• validate 가 quality 를 기록하고, route_after_validate 가
  • quality ≥ 0.6 → END
  • quality < 0.6 & retry_count < MAX_RETRIES → retrieve 로 루프
  • quality < 0.6 & 재시도 한도 초과 → fallback → END
• 무한 루프 방지를 위해 retry_count 를 상태에 저장·증가시킴.

필요 시, retrieve 단계에서 쿼리 확장/검색 파라미터 변경, generate 단계에서 프롬프트 강화, validate 단계에서 LLM 평가지표/규칙 결합 등으로 점진 개선하면 됨

---

 

오늘은 간단하게 langGraph에 대해 알아보았다. 기존 LangChain은 직선형 파이프라인 형태로만 동작해 복잡한 흐름을 표현하기 어렵지만, LangGraph는 라우터 분기점이나 루프를 활용해 다양한 경로를 설계할 수 있다.

• 복잡한 워크플로우 시각화: 체인보다 훨씬 구조적임.
• 조건부 분기 / 루프: 단순 직선형 체인이 아닌 복잡한 플로우를 구현 가능.
• 상태(State) 기반 관리: 전체 컨텍스트를 상태로 공유.
• LangChain과 호환: 기존에 쓰던 LangChain 컴포넌트 그대로 활용 가능.

 

결국 LangGraph는 LangChain을 확장해주는 역할을 한다.
단순한 데모 수준을 넘어서, 여러 모듈이 얽히고 조건부 분기·재시도가 많은 실제 서비스에서는 거의 필수적으로 사용하게 된다.

LangChain으로 Rag 구축 프로젝트 했던게 작년인데 AI 분야는 정말 빠르게 발전한다. 다음글에서는 LangGraph에 좀 더 자세히 알아보려고 한다.