Kwangmin Kim - LangGraph 개요

1 LangGraph의 탄생배경

RAG를 도입하면서 현실에서 다음과 같은 문제에 반복적으로 봉착한다.
- LLM이 생성한 답변이 실제 근거가 없는 hallucination인지 판단하기 어렵다.
- RAG에서 얻은 답변이 문서 기반 검색에서 얻은 정보가 아닌 외부(LLM) 사전지식에 의존해 잘못된 답변을 만들 수 있다.
- 문서 검색으로 원하는 정보를 찾지 못하면 외부(웹, 논문 등)를 추가 탐색해야 하는데, 이 과정에서 신뢰할 수 없는 정보가 포함되면 최종 답변이 오염된다. (예, 삼성의 매출액을 묻는 질문에 웹검색을 통해 삼성 SDS 매출액을 가져올 수 있다.)
가상의 예시: 기업 소개 문서로 QA를 수행할 때 매출액을 묻는 경우를 생각한다.
1. 기업 소개 문서에는 매출액 정보가 없다.
2. 시스템은 먼저 문서 검색을 수행하고 결과가 없으면 웹 검색으로 보강하려 시도한다.
3. 웹 검색에서 나온 정보가 출처가 불분명하거나 잘못된 경우, LLM은 그 정보를 근거로 답변을 만들어 hallucination을 발생시킬 수 있다.
4. 추가 검색-재검증 과정을 무한 반복하면 chain 구조가 복잡해지고 토큰 비용이 급증하고 응답 지연이 커진다.
이처럼 단순한 action chain이 여러 도구와 의사결정 루프를 거치며 복잡해지면 파이프라인에 action chain들이 덕지 덕지 붙게 되어 유지와 디버깅이 어려워진다.
- 반복적 검색과 재작성 루프는 토큰 소모와 응답 지연을 키운다.
- 중간 결정(예: 결과 신뢰도 기준을 만족하지 않으면 검색을 중단) 로직이 늘어나면 파이프라인이 비선형적으로 복잡해진다.
- 각 단계의 작은 변화가 전체 답변 품질에 큰 영향을 끼쳐 예측 불가능성이 커진다.
- 결국, 코드가 점점 길어지고 복잡해진다.
- 또한, LLM의 일관되지 않은 답변이 나비효과로 이어져 답변 품질 저하로 이어진다.

2 Conventional RAG의 한계: 단방향의 Pipeline

사전에 정의된 데이터 소스(PDF, DB, 테이블 등)에 의존한다.
사전에 정의된 고정된 청크 크기와 색인 전략을 사용한다.
사전에 정의된 쿼리 입력 형식과 검색 방법이 고정되어 유연성 부족하다.
신뢰하기 어려운 LLM/Agent의 불확실성
사전에 정의된 고정된 프롬프트 템플릿으로 다양한 상황 대처가 어렵다.
문서와의 관련성/신뢰성이 부족한 LLM의 답변 결과
이전 단계로 되돌아가 결과를 수정하거나 재검증하기 어렵다.
즉, 이 일련의 단반향의 파이프라인을 한번의 시도로 좋은 성능을 이끌어내야 하는데, LLM의 불확실성과 외부 정보의 신뢰성 문제로 인해 답변 품질이 크게 저하될 수 있다.
예를 들어, 보고서를 한번에 일필휘지로 완벽한 답변을 만들어내는건 힘들고 상사의 feedback을 반영하여 답변을 개선하는 iterative한 과정이 필요한데, 기존 RAG에서는 이 과정이 매우 어렵다.

3 LangGraph 란?

각 세부 과정을 Node로 분리하여 독립적으로 개발, 테스트, 재사용이 가능한 모듈형 RAG 프레임워크이다.
문법이 생소할 수 있지만 쉬운편이라 빠르게 익힐 수 있다.
LangChain은 자유도가 좀 떨어지기 때문에 호불호가 갈리는 반면, LangGraph는 RAG 파이프라인을 그래프 기반 워크플로우로 모델링하여 유연성과 투명성을 높이는 프레임워크이다.
LangChain처럼 도구 생태계를 활용할 수 있으면서도 LangChain에 종속되지 않는 경량 그래프 추상화를 제공한다.
Modular RAG를 위한 플로우 엔지니어링 프레임워크로, RAG 파이프라인을 노드와 엣지로 구성된 그래프로 모델링하여 각 단계의 독립성과 상호작용을 명확히 표현한다.
- Modular RAG: RAG을 모듈화하여 각 단계(문서 적재, 청크 분할, 임베딩, 검색, 평가, 답변 생성 등)를 독립적인 컴포넌트로 구성하는 접근 방식.
핵심 목표는 다음과 같다.
- 각 처리 단위를 독립적인 노드로 분리해 재사용성과 테스트 가능한 유닛으로 만든다.
- 노드 간의 흐름을 엣지로 정의해 조건부 분기와 반복, 병렬 처리를 자연스럽게 표현한다.
- 상태(state)를 명시적으로 관리해 중간 결과를 검증하거나 과거 실행을 재생(replay)할 수 있게 만든다.
예시
- 기존 단방향 파이프 라인: 질문 → 청크 분할 → 임베딩 → 문서 검색 → 답변 생성 → 최종 답변
- LangGraph의 그래프 모델링
  - STEP1: 질문 → 청크 분할 → 임베딩 → 문서 검색 → 평가자1 → 답변 생성 → 최종 답변 (질문과 문서 관련성 판단)
  - STEP2-1: 평가자1 → 웹 검색(문서 보강) → 문서 검색 → 평가자1 → 답변 생성 → 최종 답변 (문서의 품질이 낮을때)
  - STEP2-2: 평가자1 → 질문 재작성 → 문서 검색 → 평가자1 → 답변 생성 → 최종 답변 (질문이 모호할 때)
  - STEP3: 평가자1 → 답변 생성 → 평가자2 → 최종 답변 (LLM 답변의 신뢰성 판단)
  - STEP3-1: 평가자1 → 답변 생성 → 평가자2 → STEP2-1 → 최종 답변
  - STEP3-2: 평가자1 → 답변 생성 → 평가자2 → STEP2-2 → 최종 답변
  - STEP4: 평가자1 → 답변 생성 → 평가자2 → Human-in-the-loop → 최종 답변 (자동화가 불확실할 때 사람이 판단하도록 하는 경우)

3.1 LangGraph의 구성 성분

Node, Edge, State, Conditional Edge, Human-in-the-loop, Checkpointer 등을 구성하여 LLM을 활용한 워크플로우에 순환 연산 기능을 추가하여 흐름을 제어할 수 있다.
Node: 특정 작업(task)을 수행하는 단위 함수 또는 컴포넌트로 Python 함수로 자유롭게 정의한다.
- 입력과 출력을 명확히 정의하고 상태를 읽고 쓸 수 있다.
- LangChain 문법을 사용하지 않아도 된다.
Edge: 노드 간의 실행 흐름을 정의하는 연결선.
- 단순 순차 실행뿐 아니라 조건부 분기, 반복, 병렬화를 표현한다.
State: 워크플로우 전반에서 공유되거나 노드별로 보존되는 값으로 일종의 메세지 전달자 역할을 한다.
- 예: 쿼리 변형 히스토리, 검색 결과, LLM 응답, 평가 메타데이터 등.
Conditional Edge: 조건에 따라 다른 경로로 분기시키는 엣지.
Human-in-the-loop: 분기나 검증 단계에서 사람이 개입해 결정하는 지점.
Checkpointer: 과거 실행의 스냅샷을 저장하고 특정 시점으로 되돌려 재실행하거나 재검증하는 수정 & 재실행 기능.
- 과거에 했던 대화내용을 기억해 멀티턴을 가능하게 한다.

3.2 설계 아이디어와 장점

모듈화: 데이터 적재, 전처리, 임베딩, 검색, 평가, 응답 생성 등 각 단계를 독립 노드로 구성해 조합 가능하게 한다.
가시성: 그래프 구조로 흐름을 표현하면 어느 단계에서 문제가 발생했는지 추적하기 쉬워진다.
유연성: 조건부 엣지로 다양한 분기 전략을 구현하고, human-in-the-loop을 손쉽게 끼워 넣을 수 있다.
복구/재현성: 체크포인터를 통해 특정 시점으로 돌아가 중간 상태를 수정하고 재실행할 수 있다.

4 구체적 가상 예시: 기업 매출 질문 시나리오

상황: 내부 company_profile.pdf에 기업 개요와 연혁은 있지만 최신 매출액 정보는 없다.
목표: 사용자 질문에 근거 기반 답변을 제공하되, 문서에 없을 경우 안전하게 외부 근거를 보강한다.
권장 플로우(요약):
1. document_loader → split → embedding → store로 문서 색인 준비.
2. 사용자 질의 도착 → query_rewrite로 명확화(예: 기간, 통화 단위 추가).
3. retriever로 관련 청크 검색 → relevance_evaluator로 신뢰도 판단.
4. 관련도가 기준 이상이면 LLM 응답 생성 → quality_evaluator로 검증 후 반환.
5. 관련도가 낮거나 정보가 부족하면 web_surfing으로 외부 검색 시도 → source_verifier로 출처 신뢰성 점검.
6. 외부 출처가 신뢰할 수 있으면 LLM이 출처를 명시해 답변 생성, 그렇지 않으면 사용자에게 불확실성을 알리고 human-in-the-loop로 판단 요청.

4.1 세부 동작과 안전장치

신뢰도 기준: 검색 점수와 출처 메타데이터(도메인, 발행일, 공식성)를 결합해 임계값을 설정한다.
토큰/비용 제어: 외부 검색은 최대 N번으로 제한하고, 재검색 루프는 체크포인터로 중단/재개할 수 있게 설계한다.
출처 투명성: LLM 응답에 반드시 참조 출처를 병기하도록 강제한다.
Human fallback: 자동화가 불확실할 때는 검증 요청을 생성해 사람이 판단하도록 한다.

4.2 간단한 pseudocode (개념 설명용)

# flow: document indexing
def index_documents(docs):  
    chunks = split(docs)  
    embeds = embed(chunks)  
    store(embeds)  

# flow: query handling
def handle_query(query):  
    query = query_rewrite(query)  
    results = retriever(query)  
    if relevance_evaluator(results):  
        answer = llm_generate(results, query)  
        if quality_evaluator(answer):  
            return answer_with_sources(answer, results)  
    external = web_surfing(query, max_attempts=2)  
    verified = source_verifier(external)  
    if verified:  
        return answer_with_sources(llm_generate(results+external, query), results+external)  
    return request_human_review(query, results, external)

위 pseudocode는 개념을 설명하기 위한 간단한 예이며, 실제 구현에서는 비동기 처리, 에러 핸들링, 체크포인터 관리, 로깅 등이 필요하다.

5 구성 성분의 세부기능

5.1 상태 (State)

노드와 노드 간에 정보를 전달할 때 상태 객체에 담아 전달한다.
TypedDict: 일반 파이썬 dict에 type hinting을 추가한 개념이지만 쉽게 dictionary로 생각해도 좋다.
- key:value에서 value의 데이터 타입을 지정해주는 것
- 예: question: Annoated[list, add_messages]
question, context, answer, messages, relevance 등 모든 값을 다 채우지 않아도 된다.
새로운 노드에서 값을 덮어쓰기(Overwrite)방식으로 채운다.
reducer (add_message 또는 operator.add): 기존의 list에 새로운 메시지를 추가하는 방식으로 채운다.
- append 방식과 유사하지만, 반환값에 annotation과 함께 list로 감싸기만 하면 자동으로 메시지가 쌓이게 된다.
- 예를 들어, 기존에 messages에 [message1, message2] 가 있는데 messages: message3이 입력이 되면 messages에는 [message3]만 남게 된다.
- 반면, messages: Annoated[list, add_messages]가 입력이 되면 messages에는 [message1, message2, message3]이 되게 된다.


from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages # reducer로 사용할 add_messages 함수를 가져온다.

# GraphState 상태를 저장하는 용도로 사용합니다.

class GraphState(TypedDict):
    question: Annotated[list, add_messages] # 질문(Query Rewrite 누적) - 메세지 자동 추가
    context: Annotated[str, 'Context'] # 문서의 검색 결과 - overwrite
    answer: Annotated[str, 'Answer'] # 답변 - overwrite
    messages: Annotated[list, add_messages] # 메세지 - 메세지 자동 추가
    relevance: Annotated[str, 'Relevance'] # 관련성 - overwrite

Redcuer (add_message 혹은 operator.add): 자동으로 list에 메세지를 추가해주는 기능
- list안에 chatbot의 경우 질문과 답변이 누적해서 쌓이게 되는데 append 메서드를 사용하지 않고 반환값에 annoatation과 함께 list로 감싸기만 하면 자동으로 메시지가 쌓이게 된다.
- reducer (add_messages 혹은 operator.add): 자동으로 list에 메시지를 추가
  - left (Messages): 기본 메시지 리스트
  - right (Messages): 병합할 메시지 리스트 또는 단일 메시지
예를 들어, GraphState 클래스의 messages: Annoated[list, add_messages] # 메세지 - 메세지 자동 추가 가 입력이 되면 다음의 코드가 자동으로 실행이 된다.

from langchain_core.messages import AIMessage, HumanMessage
from langgraph.graph import add_messages

# 기본 사용 예시

msgs1 = [HumanMessage(content="hello?", id = "1")]
msgs2 = [AIMessage(content="glad to see you~", id = "2")]

result1 = add_messages(msgs1, msgs2)
print(result1)

5.1.1 노드별 상태 값의 변화 원리

간단한 예시: 각 노드는 다른 노드에 대한 정보가 없기 때문에 상태 객체에 상태값을 담아서 필요한 정보만을 전달(덮어쓰기 방식)한다.

Key	1. 노드1 (Question)	2. 노드2 (Retrieve)	3. 노드3 (Answer)	4. 노드4 (Evaluate)
time	1	2	3	4
name		David	David	Tom
llm	GPT	GPT	GPT	GPT

각 노드에서 새롭게 업데이트 하는 값은 기존 Key의 값이 덮어쓰기(overwrite) 방식으로 업데이트
노드에서 필요한 상태값을 조회하여 동작에 활용
최종적으로 노드4에서 state값을 조회해보면 노드 1에서 반영된 llm값(GPT)이 그대로 상태전달되어 조회 가능
- 즉, 각 노드에서 필요한 상태값을 조회하여 동작에 활용할 수 있고, 최근 노드의 상태값에서 시계열적으로 누적된 정보가 상태전달되어 조회 가능하다.
RAG 좀 더 복잡한 예시:
노드4에서 ’문서’에 대하여 답변 관련성 점수를 부여
score가 bad인 경우 (선택할 수 있는 행동 3가지)
- 노드1: 질문을 재작성 요청
- 노드2: 문서를 다시 검색 및 검색을 통한 정보 보완
- 노드3: 답변을 재작성 요청

Key	노드1 (Question)	노드2 (Retrieve)	노드3 (Answer)	노드4 (Evaluate)
time	1	2	3	4
context		문서1	문서1	문서1
question	질문1	질문1	질문1	질문1
answer			답변1	답변1
score				BAD

time1: 노드1에서 질문1이 입력
time2: 노드2에서 문서1이 검색, 상태값을 열어보면 질문1이 그대로 상태전달되어 조회 가능
time3: 노드3에서 답변1이 생성, 상태값을 열어보면 질문1과 문서1이 그대로 상태전달되어 조회 가능
time4: 노드4에서 답변1에 대한 점수 BAD가 부여, 상태값을 열어보면 질문1, 문서1, 답변1이 그대로 상태전달되어 조회 가능
노드4에서 score가 BAD로 평가되면 다음의 옵션을 선택할 수 있는 조건부 엣지(Conditional Edge)가 만들어질 수 있다.
- 옵션1: 노드1에서 질문을 재작성 요청
- 옵션2: 노드2에서 문서를 다시 검색 및 검색을 통한 정보 보완
- 옵션3: 노드3에서 답변을 재작성 요청
조건부 엣지 옵션1: 노드1로 상태전이하여 질문을 재작성 요청 하는 경우 (노드4 → 노드1)
- time5: 질문이 재작성되어 질문2가 입력
- time6: 질문이 바뀌었으니 문서를 재검색해야함 (문서2 업데이트)
- time7: 노드3에서 질문2와 문서2를 바탕으로 답변2가 생성됨
- time8: 노드4에서 답변2에 대한 평가가 이루어짐

Key	노드1 (Question)	노드2 (Retrieve)	노드3 (Answer)	노드4 (Evaluate)
time	5	6	7	8
context	문서1	문서2	문서2	문서2
question	질문2	질문2	질문2	질문2
answer	답변1	답변1	답변2	답변2
score	BAD1	BAD1	BAD1	GOOD

조건부 엣지 옵션2: 노드2로 상태전이하여 문서 검색을 재요청 하는 경우 (노드4 → 노드2)

Key	노드1 (Question)	노드2 (Retrieve)	노드3 (Answer)	노드4 (Evaluate)
context		문서2	문서2	문서2
question	질문1	질문1	질문1	질문1
answer		답변1	답변2	답변2
score		BAD	BAD	GOOD

조건부 엣지 옵션3: 노드3로 상태전이하여 답변을 재작성 요청 하는 경우 (노드4 → 노드3)

Key	1. 노드1 (Question)	2. 노드2 (Retrieve)	5. 노드3 (Answer)	6. 노드4 (Evaluate)
context		문서1	문서1	문서1
question	질문1	질문1	질문1	질문1
answer			답변2	답변2
score			BAD	GOOD

5.2 노드 (Node)

python 함수로 정의
- 입력 파라미터1: 상태 객체 (state: GraphState)
- 반환값: 상태 객체 (context = format_docs(retrieved_docs)) 또는 Conditional Edge의 경우 dict
노드 작명: 동사(또는 동사+명사)로 짓는 게 표준이다
- 노드는 상태를 변환하는 행위이기 때문이다. 명사로 지으면 그게 데이터인지 행위인지 모호해진다.
- graph에서 주체의 정보는 상태 객체에 담겨 전달되기 때문에 노드 이름에는 주체의 정보가 들어갈 필요가 없다.
- 따라서, 의미적으로 상태 객체가 주어, 노드 객체가 동사 역할을 하는 형태로 노드 이름을 짓는 것이 좋다.
- 예시: retrieve_document, generate_answer, evaluate_relevance 등
문서 검색 노드 예시

# 노드1: 문서에서 검색하여 관련성 있는 split를 검색
def retrieve_document(state: GraphState) -> GraphState
    # Question 에 대한 문서 검색을 retriever로 수행
    retrieved_docs = pdf_retriever.invoke(state["qeustion"])
    # 검색된 문서를 context 키에 저장
    return GraphState(context = format_docs(retrieved_docs))  

# 노드2: Chain을 사용하여 답변 생성
def llm_answer(state: GraphState) -> GraphState:    
    return GraphState(
        answer=pdf_chain.invoke({"question": state["question"], "context":state['context']})
    )

노드 정교하게 세분화하여 만들수록 Agent의 정교한 흐름을 만들어 성능이 좋아진다.
따라서, Agent의 성능은 좋은 Node를 얼마나 잘 만들었는지와 각 노드들간의 상태를 주고받아 정교한 통제를 할 수 있는지에 달려있다.
핵심 아이디어: 단일 책임 + 명확한 상태 변환
- 입력 상태에서 출력 상태로의 변환이 예측 가능하고 단일한 책임을 가질 것.
- 각 노드는 하나의 작업에 집중하고, 입력과 출력이 명확히 정의된 상태 객체를 사용해 다른 노드와 통신한다.
- 예시:
  - 문서 검색 노드: 질문을 입력으로 받아 관련 문서를 상태에 저장
  - 답변 생성 노드: 질문과 문서를 입력으로 받아 답변을 상태에 저장
- 이렇게 하면 각 노드를 독립적으로 개발, 테스트, 재사용할 수 있고, 전체 워크플로우의 흐름을 명확하게 제어할 수 있다.
나쁜 노드 vs 좋은 노드

# 나쁜 노드 - 너무 많은 책임
def process_query(state: GraphState) -> GraphState:
    query = state["query"]
    # 검색도 하고
    docs = retriever.search(query)
    # 재랭킹도 하고
    reranked = reranker.rerank(docs)
    # LLM 호출도 하고
    answer = llm.invoke(reranked)
    # 검증도 하고
    is_valid = validator.check(answer)
    return {"answer": answer, "is_valid": is_valid}

# 좋은 노드 - 단일 책임
def retrieve_documents(state: GraphState) -> GraphState:
    docs = retriever.search(state["query"])
    return {"retrieved_docs": docs}

def rerank_documents(state: GraphState) -> GraphState:
    reranked = reranker.rerank(state["retrieved_docs"])
    return {"reranked_docs": reranked}

def generate_answer(state: GraphState) -> GraphState:
    answer = llm.invoke(state["reranked_docs"])
    return {"answer": answer}

def validate_answer(state: GraphState) -> GraphState:
    is_valid = validator.check(state["answer"])
    return {"is_valid": is_valid, "retry_count": state.get("retry_count", 0)}

결국 Agent의 성능이 좋으려면 세부 노드를 정교하게 만들어놓고 각 노드들간의 상태를 주고받아 정교한 통제를 할 수 있어야한다.
graph 생성 후 노드 추가하는 방식
- 이전에 정의한 함수를 graph에 추가
- add_node ("노드이름", 함수)

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

# langgraph.graph에서 StateGraph와 END를 가져온다.
workflow = StateGraph(GraphState)

# 노드들을 정의한다.
workflow.add_node("retrieve", retrieve_document) # 에이전트 노드 추가
workflow.add_node("llm_answer", llm_answer) # 정보 검색 노드를 추가

5.3 엣지(Edge)

노드에서 노드간의 연결
add_edge(“from 노드이름”, “to 노드이름”)

# 노드연결
workflow.add_edge("retreive", "llm_answer") # 검색 -> 답변
workflow.add_edge("llm_answer", "relevance_check") # 답변 -> 관련성 체크

5.4 조건부 엣지 (Conditional Edge)

노드에 조건부 엣지를 추가하여 분기를 수행할 수 있다.
add_conditional_edge(“form-노드이름”,to-조건부 판단 함수, dict로 다음 단계 결정)
흐름 예시
- “relevance_check”노드에서 나온결과를 is_relevant함수에 입력
- path map: is_relevant함수의 반환 값은 “grounded”, “notGrounded”, “notSure” 중 하나
  - value에 해당하는 값이 END 노드면 Graph종료
  - “llm_answer”와 같이 다른 노드이름이면 해당 노드로 연결

#조건부 엣지를 추가

workflow.add_conditional_edge(
    "relevance_check", # 관련성 체크 노드에서 나온 결과를 is_relevant 함수에 전달
    is_relevant,
    {
        "grounded": END # END노드: 관련성이 있으면 종료합니다.
        "notGrounded": "llm_answer" # 답변 생성 노드: 관련성이 없으면 다시 답변 생성
        "notSure": "llm_answer" # 답변 생성 노드: 관련성 체크 결과가 모호하다면 다시 답변을 생성
    }
)

시작점 지정: set_entry_point("노드이름") 지정한 시작점부터 graph가 시작
- 노드 설계를 한 후 만약 시작점 지점을 맨 마지막 노드로 하면 에러가 발생

workflow.set_entry_point("retrieve")

5.5 CheckPointer (Memory): graph의 생성 및 시각화

checkpointer: 각 노드간 실행결과를 추적하기 위한 메모리 (대화에 대한 기록과 유사)
그래프 생성을 하는데 단순하게 그래프 생성하는 것은 compile만 하면 되는데 체크포인터는 메모리 기능이 있음
각 노드간의 실행결과를 추적하기 위한 메모리(대화에 대한 기록과 유사)
체크포인터를 활용하여 특정 시점 (snapshot)으로 되돌리기 기능도 가능
app = workflow.compile()로 그래프 생성. 메모리 없이 바로 실행도 가능

# 기록을 위한 메모리 저장소 설정
memory = MemorySaver()

app = workflow.compile(checkpointer=memory) # 메모리가 있는 그래프 컴파일
app = workflow.compile() # 메모리 없이 그래프 컴파일

memory = MemorySaver()로 메모리 저장소 설정하여 기억 저장 용도로 사용한다.
compile(checkpointer=memory) 지정하여 그래프 생성

5.6 Graph Visualization

생성한 그래프 시각화: get_graph(xray=True).draw_mermaid_png()
실무에선 그래프가 복잡해지면 대부분의 경우 시각화해서 보는게 좋다. 코드로 보면 헷갈린다.

display(
    Image(app.get_graph(xray=True).draw_mermaid_png()) # mermaid로 그래프 시각화하여 보여주는데 인터넷이 반드시 연결되어 있어야 한다.
)

5.7 Graph Execution

RunnableConfig
- recursion_limit: 최대 노드 실행 개수를 지정 (10인 경우: 총 13개의 노드까지 실행), evaluator의 무한 루프 방지용
- thread_id: 그래프 실행 ID를 기록하고, 추후 추적하기 위한 목적으로 활용
  - 멀티턴의 경우: thread_id별로 대화내용을 따로 저장해 관리한다.
  - 카톡의 채팅방과 유사한 개념으로 생각하면 된다.
상태로 시작
- 여기서 question에 질문만 입력하고 상태를 첫 번째 노드에게 전달
app.invoke(state, config)로 그래프 실행

6 노드 설계 전략

6.1 상태 스키마를 먼저 설계하고 노드를 나중에 설계한다

노드가 아니라 상태가 플랫폼의 척추다.
상태 스키마가 명확하면 노드 분리가 자연스럽게 따라온다.

from typing import TypedDict, List, Optional
from langchain_core.documents import Document

class GraphState(TypedDict):
    # 입력
    query: str
    user_type: str  # "researcher" | "developer"
    
    # 중간 상태
    retrieved_docs: List[Document]
    reranked_docs: List[Document]
    retrieved_standards: List[dict]  # 데이터 표준화 컨텍스트
    
    # 출력
    answer: str
    confidence_score: float
    is_valid: bool
    
    # 제어 흐름
    retry_count: int
    route_decision: str  # conditional edge용
    error_message: Optional[str]

6.2 노드 분리 기준 3가지

IO 경계: 외부 시스템 호출(DB, API, LLM)마다 노드를 분리. 실패 지점이 명확해지고 재시도 로직 적용이 쉬워진다.
책임 경계: 검색/재랭킹/생성/검증은 각각 다른 노드. 하나가 바뀌어도 다른 노드에 영향 없음.
관찰 경계: 로깅/모니터링이 필요한 지점마다 노드로 분리. 블랙박스 구간을 없앤다.

6.3 Conditional Edge 설계가 정교함을 만든다

노드 자체보다 노드 간 라우팅 로직이 Agent 성능을 결정한다.

def route_after_validation(state: GraphState) -> str:
    if state["is_valid"]:
        return "end"
    elif state["retry_count"] < 3:
        return "retry_retrieve"  # 검색부터 다시
    elif state["confidence_score"] < 0.5:
        return "escalate_to_human"  # human-in-the-loop
    else:
        return "fallback_answer"

# 그래프에 등록
graph.add_conditional_edges(
    "validate_answer",
    route_after_validation,
    {
        "end": END,
        "retry_retrieve": "retrieve_documents",
        "escalate_to_human": "human_review",
        "fallback_answer": "generate_fallback"
    }
)

6.4 프로젝트 적용 예시

데이터 표준화 Agent의 하이브리드 엔진(Rule → ML → RAG)을 노드로 표현하면:

[입력 노드] parse_input
      ↓
[Rule 엔진 노드] apply_abbreviation_rules
      ↓
[라우팅] route_after_rule
      ├── 신뢰도 높음 → [검증 노드] validate_with_rag → END
      ├── 신뢰도 낮음 → [ML 노드] apply_domain_ml_model
      │                      ↓
      │               [라우팅] route_after_ml
      │                      ├── 신뢰도 높음 → validate_with_rag
      │                      └── 신뢰도 낮음 → [RAG 노드] generate_with_llm
      └── 실패 → [RAG 노드] generate_with_llm
                      ↓
              [검증 노드] validate_with_rag
                      ↓
              [라우팅] route_after_validation
                      ├── 통과 → [추천 노드] format_metadata_output → END
                      └── 실패, retry < 3 → parse_input (재시도)
                      └── 실패, retry >= 3 → human_review

이 구조의 핵심은 각 엔진의 신뢰도 점수가 상태로 전달되어 라우팅 결정의 근거가 된다는 것이다. 노드가 세분화될수록 어느 단계에서 실패했는지 추적이 가능해지고, 개별 노드만 교체해서 성능을 개선할 수 있다.

7 LangGraph의 가치가 발휘되는 지점

단순한 선형 파이프라인이나 소규모 에이전트라면, 함수 잘 정의하고 호출 관계 정리하는 거랑 실질적으로 차이 없음.
LangGraph가 boilerplate만 추가하는 꼴이 되는 경우도 많음.
하지만, Agent이 복잡해질수록 LangGraph의 가치가 발휘됨
- State 공유 메커니즘 — 여러 노드가 동일한 state object를 read/write할 때, 함수 체인으로 구현하면 state를 파라미터로 계속 pass해줘야 하거나 전역변수 써야 함. LangGraph는 이걸 typed state schema로 중앙 관리.
- Conditional Edge + Loop — 에이전트가 “아직 답이 부족하면 다시 검색”처럼 동적으로 분기하고 순환할 때, 함수로 짜면 재귀나 while loop으로 구현해야 하는데 흐름 파악이 어려워짐. LangGraph는 이게 그래프에 명시적으로 드러남.
- Human-in-the-loop / Interrupt — interrupt_before, interrupt_after로 특정 노드 실행 전후에 사람이 개입할 수 있는 구조가 프레임워크 레벨로 지원됨. 함수 체인으로 이거 구현하려면 꽤 복잡해.
- Checkpointing / 재개 — 실행 중간 상태를 저장하고 나중에 이어서 실행하는 게 내장돼 있어. 장시간 실행되는 에이전트에서 중요함.
- Observability — LangSmith랑 붙이면 어느 노드에서 얼마나 걸렸는지, state가 어떻게 변했는지 트레이싱이 됨.
- Modularity & Reusability — 노드가 독립적으로 설계되고 상태로 통신하니까, 특정 노드만 떼어내서 다른 그래프에서 재사용하기 쉬움. 함수 체인으로 짜면 함수 간 의존성이 높아져서 재사용이 어려워짐.