LangGraph 단계별 학습 예제

langgraph_tutorial.py

"""
═══════════════════════════════════════════════════════════════════════════
  LangGraph 단계별 학습 예제 — 기본 개념부터 심화까지
═══════════════════════════════════════════════════════════════════════════

이 파일은 위에서 아래로 읽으면 LangGraph의 핵심 개념이 순서대로 쌓이도록
설계되었습니다. 각 PART는 독립 실행 가능하며, 앞 PART의 개념 위에 다음 개념을
하나씩 얹습니다.

  PART 1. State        — 그래프가 들고 다니는 "공유 상태"
  PART 2. Node & Edge  — 상태를 변환하는 함수와 그 연결
  PART 3. Reducer      — 상태 필드를 "어떻게 합칠지" 정의
  PART 4. 조건부 엣지   — 런타임 데이터로 분기
  PART 5. 순환(loop)   — 그래프가 되돌아가며 반복 (에이전트의 핵심)
  PART 6. 체크포인터    — 상태 영속화 + 중단/재개
  PART 7. 병렬 실행     — fan-out / fan-in
  PART 8. HITL         — 사람이 중간에 개입 (interrupt)

★ 의도적으로 외부 LLM API를 쓰지 않습니다.
  대신 결정론적인 "가짜 모델 함수"로 로직을 대체해, API 키 없이도 그래프의
  '구조와 흐름'에만 집중할 수 있게 했습니다. 실제로는 이 자리에 LLM 호출이 들어갑니다.

실행:  python langgraph_tutorial.py
필요:  pip install langgraph
"""

from typing import Annotated, TypedDict, Literal
import operator


# ═══════════════════════════════════════════════════════════════════════════
# PART 1. State — 그래프의 심장
# ═══════════════════════════════════════════════════════════════════════════
#
# LangGraph의 모든 것은 "State(상태)"를 중심으로 돕니다.
# State는 그래프가 실행되는 동안 노드들 사이를 흘러다니는 공유 데이터입니다.
#
# 일반 함수 체인과의 결정적 차이:
#   - 함수 체인: A의 리턴값이 B의 인자로 전달 (단방향, 단일 값)
#   - LangGraph: 모든 노드가 같은 State 객체를 읽고/부분 수정 (구조적, 다중 필드)
#
# State는 보통 TypedDict로 정의합니다. (타입 안전 + 가독성)

class BasicState(TypedDict):
    topic: str          # 입력
    draft: str          # 중간 산출물
    result: str         # 최종 산출물


# ═══════════════════════════════════════════════════════════════════════════
# PART 2. Node & Edge — 상태를 변환하고 연결하기
# ═══════════════════════════════════════════════════════════════════════════
#
# Node(노드): State를 받아서, "바뀐 부분만" dict로 리턴하는 함수.
#   - 전체 State를 리턴할 필요 없음. 바꾼 키만 리턴하면 LangGraph가 병합.
# Edge(엣지): 노드 실행 순서를 정의하는 연결선.
#
# 아래는 가장 단순한 선형 그래프: write_draft → polish

from langgraph.graph import StateGraph, START, END

def write_draft(state: BasicState) -> dict:
    # state["topic"]을 읽어서 draft를 만든다 (여기선 가짜 로직)
    topic = state["topic"]
    return {"draft": f"[초안] {topic}에 대한 글"}   # ← 바뀐 키만 리턴

def polish(state: BasicState) -> dict:
    return {"result": state["draft"].replace("[초안]", "[완성]")}

def build_basic_graph():
    # 1) State 스키마를 지정해 그래프 빌더 생성
    g = StateGraph(BasicState)
    # 2) 노드 등록 (이름, 함수)
    g.add_node("write", write_draft)
    g.add_node("polish", polish)
    # 3) 엣지로 흐름 정의: START → write → polish → END
    g.add_edge(START, "write")    # START: 그래프 진입점 (특수 노드)
    g.add_edge("write", "polish")
    g.add_edge("polish", END)     # END: 그래프 종료점 (특수 노드)
    # 4) 컴파일 → 실행 가능한 그래프 객체
    return g.compile()


# ═══════════════════════════════════════════════════════════════════════════
# PART 3. Reducer — "상태를 어떻게 합칠 것인가"
# ═══════════════════════════════════════════════════════════════════════════
#
# 기본 동작: 노드가 어떤 키를 리턴하면, 그 키는 "덮어쓰기"됩니다.
#   {"draft": "A"} 다음에 {"draft": "B"} → draft는 "B" (이전 값 소멸)
#
# 하지만 "누적"하고 싶을 때가 있습니다. 대표적으로 메시지 히스토리.
# 이때 Reducer를 씁니다. Annotated[타입, reducer함수]로 선언하면,
# 그 필드는 덮어쓰기 대신 reducer 로직으로 병합됩니다.
#
# operator.add 를 reducer로 주면: 리스트가 이어붙여집니다(append).

class ReducerState(TypedDict):
    # logs 필드는 노드가 리턴할 때마다 "덮어쓰기"가 아니라 "이어붙이기"
    logs: Annotated[list[str], operator.add]
    counter: int   # reducer 없음 → 덮어쓰기

def step_a(state: ReducerState) -> dict:
    return {"logs": ["A 실행됨"], "counter": 1}

def step_b(state: ReducerState) -> dict:
    # logs는 누적되고(["A 실행됨", "B 실행됨"]), counter는 덮어써짐(1→2)
    return {"logs": ["B 실행됨"], "counter": 2}

def build_reducer_graph():
    g = StateGraph(ReducerState)
    g.add_node("a", step_a)
    g.add_node("b", step_b)
    g.add_edge(START, "a")
    g.add_edge("a", "b")
    g.add_edge("b", END)
    return g.compile()


# ═══════════════════════════════════════════════════════════════════════════
# PART 4. 조건부 엣지 — 런타임 데이터로 분기
# ═══════════════════════════════════════════════════════════════════════════
#
# 지금까지는 흐름이 고정(static)이었습니다. 하지만 진짜 에이전트는
# "상태를 보고 어디로 갈지 결정"해야 합니다. 이것이 조건부 엣지입니다.
#
# add_conditional_edges(출발노드, 라우팅함수, 매핑)
#   - 라우팅함수: State를 받아 "다음 노드 이름(문자열)"을 리턴
#   - 매핑: 리턴된 문자열 → 실제 노드 이름

class RouteState(TypedDict):
    text: str
    sentiment: str
    response: str

def classify(state: RouteState) -> dict:
    # 가짜 감성 분류 (실제로는 LLM)
    s = "positive" if "좋" in state["text"] else "negative"
    return {"sentiment": s}

def handle_positive(state: RouteState) -> dict:
    return {"response": "기쁜 소식이네요!"}

def handle_negative(state: RouteState) -> dict:
    return {"response": "도움을 드리겠습니다."}

def route_by_sentiment(state: RouteState) -> Literal["pos", "neg"]:
    # State를 보고 분기 결정. 리턴 문자열은 아래 매핑의 키.
    return "pos" if state["sentiment"] == "positive" else "neg"

def build_conditional_graph():
    g = StateGraph(RouteState)
    g.add_node("classify", classify)
    g.add_node("pos_handler", handle_positive)
    g.add_node("neg_handler", handle_negative)
    g.add_edge(START, "classify")
    # 조건부 엣지: classify 후, route_by_sentiment의 리턴값으로 분기
    g.add_conditional_edges(
        "classify",
        route_by_sentiment,
        {"pos": "pos_handler", "neg": "neg_handler"},  # 라우팅키 → 노드
    )
    g.add_edge("pos_handler", END)
    g.add_edge("neg_handler", END)
    return g.compile()


# ═══════════════════════════════════════════════════════════════════════════
# PART 5. 순환(Loop) — 에이전트의 본질
# ═══════════════════════════════════════════════════════════════════════════
#
# 조건부 엣지가 "이전 노드"로 돌아가게 만들면 순환이 생깁니다.
# 이것이 에이전트 루프의 핵심입니다:
#   생성 → 평가 → (불충분하면) 다시 생성 → 평가 → ... → (충분하면) 종료
#
# 아래는 Generator/Evaluator 루프. Evaluator가 통과시킬 때까지 반복.
# (태훈님의 Planner/Implementer/Evaluator 패턴의 축소판)
#
# ★ 무한 루프 방지: 시도 횟수를 State에 기록하고 최대치에서 강제 종료.

class LoopState(TypedDict):
    task: str
    attempt: int
    quality: int        # 0~100, 가짜 품질 점수
    output: str

def generate(state: LoopState) -> dict:
    attempt = state.get("attempt", 0) + 1
    # 시도할수록 품질이 오른다고 가정 (실제로는 LLM 재생성)
    quality = min(40 + attempt * 25, 100)
    return {
        "attempt": attempt,
        "quality": quality,
        "output": f"{state['task']} 결과물 (시도 {attempt}, 품질 {quality})",
    }

def should_retry(state: LoopState) -> Literal["retry", "done"]:
    # 평가(Evaluator) 역할: 품질 80 미만이고 5회 미만이면 재시도
    if state["quality"] < 80 and state["attempt"] < 5:
        return "retry"
    return "done"

def build_loop_graph():
    g = StateGraph(LoopState)
    g.add_node("generate", generate)
    g.add_edge(START, "generate")
    # generate 후 평가 → retry면 generate로 되돌아감(순환!), done이면 END
    g.add_conditional_edges(
        "generate",
        should_retry,
        {"retry": "generate", "done": END},
    )
    return g.compile()


# ═══════════════════════════════════════════════════════════════════════════
# PART 6. 체크포인터 — 상태 영속화 + 중단/재개
# ═══════════════════════════════════════════════════════════════════════════
#
# 지금까지 그래프는 실행이 끝나면 상태가 사라졌습니다(메모리상).
# 체크포인터를 붙이면 "매 스텝마다 상태가 저장"되어:
#   - 서버가 죽어도 마지막 지점부터 재개 (durable execution)
#   - 같은 thread_id로 이어서 대화 (멀티턴)
#   - 과거 시점으로 되돌아가기 (time-travel)
#
# 핵심 개념: thread_id = 대화/세션의 경계.
#   같은 thread_id로 invoke하면 이전 상태가 이어집니다.
#
# 여기선 학습용으로 InMemorySaver를 씁니다.
# 프로덕션에선 PostgresSaver(영속) 사용 — 앞서 논의한 그것.

from langgraph.checkpoint.memory import InMemorySaver

class ChatState(TypedDict):
    messages: Annotated[list[str], operator.add]   # 대화 누적 (reducer!)

def chat_turn(state: ChatState) -> dict:
    # 직전 사용자 메시지에 답한다 (가짜)
    last = state["messages"][-1]
    return {"messages": [f"답변: '{last}'에 대한 응답"]}

def build_checkpointed_graph():
    g = StateGraph(ChatState)
    g.add_node("chat", chat_turn)
    g.add_edge(START, "chat")
    g.add_edge("chat", END)
    # ★ compile에 checkpointer를 넘기면 상태가 thread별로 저장됨
    return g.compile(checkpointer=InMemorySaver())


# ═══════════════════════════════════════════════════════════════════════════
# PART 7. 병렬 실행 — fan-out / fan-in
# ═══════════════════════════════════════════════════════════════════════════
#
# 한 노드에서 여러 노드로 엣지를 그으면, 그 노드들이 "병렬"로 실행됩니다.
# (LangGraph의 superstep 모델: 같은 단계의 노드들은 동시에 돈다)
# 그 후 결과를 모으는 노드로 다시 합칩니다(fan-in).
#
# ★ 병렬 노드들이 같은 필드를 쓰면 충돌하므로, reducer로 안전하게 병합해야 함.
#   여기선 operator.add로 결과를 리스트에 누적.

class ParallelState(TypedDict):
    query: str
    findings: Annotated[list[str], operator.add]   # 병렬 결과 누적
    summary: str

def search_web(state: ParallelState) -> dict:
    return {"findings": [f"웹: {state['query']} 검색결과"]}

def search_docs(state: ParallelState) -> dict:
    return {"findings": [f"문서: {state['query']} 검색결과"]}

def search_db(state: ParallelState) -> dict:
    return {"findings": [f"DB: {state['query']} 검색결과"]}

def aggregate(state: ParallelState) -> dict:
    return {"summary": " / ".join(state["findings"])}

def build_parallel_graph():
    g = StateGraph(ParallelState)
    g.add_node("web", search_web)
    g.add_node("docs", search_docs)
    g.add_node("db", search_db)
    g.add_node("aggregate", aggregate)
    # fan-out: START에서 세 노드로 동시 분기
    g.add_edge(START, "web")
    g.add_edge(START, "docs")
    g.add_edge(START, "db")
    # fan-in: 세 노드가 모두 끝나면 aggregate 실행
    g.add_edge("web", "aggregate")
    g.add_edge("docs", "aggregate")
    g.add_edge("db", "aggregate")
    g.add_edge("aggregate", END)
    return g.compile()


# ═══════════════════════════════════════════════════════════════════════════
# PART 8. Human-in-the-Loop (HITL) — 사람이 중간에 개입
# ═══════════════════════════════════════════════════════════════════════════
#
# 심화의 정점. 그래프 실행을 특정 지점에서 "일시정지"하고, 사람의 입력을
# 기다린 뒤 이어서 실행합니다. 승인 워크플로, 검토 단계 등에 필수.
#
# 핵심 도구: interrupt() — 노드 안에서 호출하면 그래프가 멈추고 제어를 반환.
#   사람이 값을 주면 Command(resume=값)로 재개하면, interrupt()가 그 값을 리턴.
#
# ★ HITL은 체크포인터가 반드시 필요합니다 (멈춘 상태를 저장해야 하니까).

from langgraph.types import interrupt, Command

class ApprovalState(TypedDict):
    amount: int
    decision: str

def request_approval(state: ApprovalState) -> dict:
    if state["amount"] >= 1000:
        # 1000 이상이면 사람 승인 필요 → 여기서 그래프가 멈춘다
        human_decision = interrupt(
            {"question": f"{state['amount']}원 지출을 승인하시겠습니까?"}
        )
        # 사람이 Command(resume=...)로 값을 주면 여기로 돌아와 이어짐
        return {"decision": human_decision}
    else:
        # 소액은 자동 승인
        return {"decision": "auto-approved"}

def build_hitl_graph():
    g = StateGraph(ApprovalState)
    g.add_node("approval", request_approval)
    g.add_edge(START, "approval")
    g.add_edge("approval", END)
    return g.compile(checkpointer=InMemorySaver())   # HITL엔 체크포인터 필수


# ═══════════════════════════════════════════════════════════════════════════
# 실행 데모 — 각 PART를 차례로 돌려보며 개념 확인
# ═══════════════════════════════════════════════════════════════════════════

def demo():
    line = "─" * 70

    print(line, "\nPART 1-2. 기본 그래프 (State + Node + Edge)")
    out = build_basic_graph().invoke({"topic": "LangGraph"})
    print("  결과:", out["result"])

    print(line, "\nPART 3. Reducer (누적 vs 덮어쓰기)")
    out = build_reducer_graph().invoke({"logs": [], "counter": 0})
    print("  logs(누적):", out["logs"])       # ['A 실행됨', 'B 실행됨']
    print("  counter(덮어쓰기):", out["counter"])  # 2

    print(line, "\nPART 4. 조건부 엣지 (분기)")
    out = build_conditional_graph().invoke({"text": "오늘 기분이 좋아"})
    print("  긍정 입력 →", out["response"])
    out = build_conditional_graph().invoke({"text": "문제가 생겼어"})
    print("  부정 입력 →", out["response"])

    print(line, "\nPART 5. 순환 루프 (Generate→Evaluate→재시도)")
    out = build_loop_graph().invoke({"task": "보고서", "attempt": 0})
    print(f"  {out['attempt']}회 시도 후 종료, 최종 품질 {out['quality']}")
    print("  최종:", out["output"])

    print(line, "\nPART 6. 체크포인터 (멀티턴 — 같은 thread_id로 상태 유지)")
    graph = build_checkpointed_graph()
    cfg = {"configurable": {"thread_id": "user-123"}}
    graph.invoke({"messages": ["안녕"]}, cfg)
    out = graph.invoke({"messages": ["잘 지내?"]}, cfg)  # 같은 thread → 누적
    print("  누적된 대화:", out["messages"])  # 4개 (안녕,답변,잘지내,답변)

    print(line, "\nPART 7. 병렬 실행 (fan-out/fan-in)")
    out = build_parallel_graph().invoke({"query": "LangGraph", "findings": []})
    print("  병렬 수집 결과 수:", len(out["findings"]))
    print("  요약:", out["summary"])

    print(line, "\nPART 8. Human-in-the-Loop (interrupt/resume)")
    graph = build_hitl_graph()
    cfg = {"configurable": {"thread_id": "approval-1"}}
    # 1000 이상 → interrupt로 멈춤
    result = graph.invoke({"amount": 5000}, cfg)
    print("  실행 멈춤. 대기 중인 질문:", result["__interrupt__"][0].value["question"])
    # 사람이 승인 → Command(resume)로 재개
    final = graph.invoke(Command(resume="manager-approved"), cfg)
    print("  사람 입력 후 재개 →", final["decision"])
    print(line)


if __name__ == "__main__":
    demo()