LangGraph 입문 — LCEL로 안 되던 순환·기억·사람 개입을 그래프로 풀기

LANGCHAIN 실전 1.0 · 03

LangGraph 입문 — 상태·영속성·스트리밍·HITL로 에이전트 런타임 익히기

LCEL로 안 되는 순환·분기를 StateGraph로, 대화 기억을 checkpointer로, 사람 승인을 interrupt()로

🟡 중급 ⏱️ 약 14분 🔧 LangChain 1.x 🗓️ 최종 검토 2026-06-21

TL;DR — LCEL은 선형 파이프라(DAG)라 순환·동적 분기·다중 액터가 필요하면 막힙니다. 이때 쓰는 게 LangGraph의 StateGraph입니다. 상태(State)는 reducer로 누적하고 단기 기억은 checkpointer + thread_id, 장기 기억은 Store로 다룹니다. 응답은 stream_mode로 흘려보내고 민감한 작업은 interrupt()로 사람 승인을 받습니다. 1.0의 create_agent도 결국 이 LangGraph 위에서 컴파일됩니다.

이 글은 LCEL까지는 익혔지만 에이전트·상태 관리는 처음인 개발자가 LangGraph의 핵심 구성요소(상태·영속성·스트리밍·HITL)를 이해하고 대화를 기억하고 사람이 개입하는 그래프를 직접 조립할 기준을 얻도록 돕습니다.

이 글에서 다루는 것

• StateGraph 기본 골격(State·node·edge·compile)
• reducer로 상태를 덮어쓸지 누적할지 정하기
• 조건부 엣지로 동적 라우팅
• 영속성 ① 단기 checkpointer + thread_id, ② 장기 Store
• stream_mode 4종과 interrupt()/Command 기반 HITL

이 글에서 다루지 않는 것

• create_agent·미들웨어 상세 (4편에서)
• RAG·검색 파이프라인 (5편에서)
• LangSmith 관측·평가 (6편에서)

목 차

1. 왜 LangGraph인가 2. StateGraph 기본 3. State와 reducer 4. 조건부 엣지 5. 영속성 ① 단기 6. 영속성 ② 장기 Store 7. 스트리밍 8. HITL interrupt() 9. 정리 · 다음 편

1. 왜 LangGraph인가 — LCEL이 막히는 지점

2편에서 본 LCEL은 prompt | model | parser처럼 데이터가 한 방향으로 흐르는 선형 파이프(DAG)입니다. 단순~중간 오케스트레이션에는 더없이 깔끔합니다. 하지만 한계가 분명합니다. 같은 노드로 되돌아오는 순환(모델이 도구를 부르고 결과를 보고 다시 판단하는 에이전트 루프), 조건에 따라 길이 갈라지는 동적 분기, 그리고 여러 액터가 한 상태를 공유하는 구조는 파이프(|) 하나로는 표현할 수 없습니다.

LangGraph는 이걸 상태 그래프(state graph)로 풉니다. 흐름을 직선 파이프가 아니라 노드(작업) + 엣지(이동)의 그래프로 정의하고 그 그래프를 흐르는 공유 상태(State)를 두는 방식입니다. 순환도 분기도 그래프에서는 그냥 엣지 하나입니다. 그래서 LangGraph는 1.0 이후 현대 에이전트의 표준 저수준 런타임이 됐습니다. 다음 편에서 볼 create_agent도 내부적으로 LangGraph 그래프로 컴파일됩니다. 즉 고수준 에이전트를 제대로 쓰려면 그 아래 런타임을 먼저 알아야 합니다.

비유하면 LCEL은 컨베이어 벨트입니다. 물건이 한 줄로 흘러갑니다. 각 공정이 차례로 손을 댑니다. 반면 LangGraph는 역할이 나뉜 작업실들과 그 사이의 통로에 가깝습니다. 어떤 통로로 갈지(분기), 같은 방을 다시 들를지(순환)를 그때그때 정합니다. 작업물(State)은 방을 옮겨 다니며 누적됩니다. 둘은 경쟁 관계가 아니라 노드 안에서 LCEL을 쓰는 포함 관계입니다.

	LCEL	LangGraph
형태	선형 파이프(DAG)	상태 그래프(순환·분기 가능)
적합	프롬프트→모델→파서, 단순 RAG	에이전트 루프, 다중 액터, HITL, 영속성
관계	Runnable 합성	상태 런타임(노드 내부에서 LCEL 사용)

표 1. LCEL vs LangGraph — 언제 무엇을 쓰나

2. StateGraph 기본 — 가장 작은 그래프

LangGraph 그래프는 네 가지 재료로 만듭니다. ① 상태(State) 정의 → ② 노드(node) 추가 → ③ 엣지(edge) 연결 → ④ 컴파일(compile)입니다. 본격적으로 들어가기 전에, 코드를 돌릴 환경을 먼저 맞춥니다.

환경

• 언어/런타임: Python 3.10 이상(1.x 요구)
• 핵심: langgraph · langchain · langchain-core · 파트너 패키지(예: langchain-openai)
• 설치: pip install langgraph langchain langchain-openai

PYTHON · 가장 작은 챗봇 그래프

from typing import Annotated, TypedDict from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langchain.chat_models import init_chat_model model = init_chat_model("openai:gpt-5.5", temperature=0) class State(TypedDict): messages: Annotated[list, add_messages] # reducer: 덮어쓰기 대신 누적 def chatbot(state: State): return {"messages": [model.invoke(state["messages"])]} builder = StateGraph(State) builder.add_node("chatbot", chatbot) builder.add_edge(START, "chatbot") builder.add_edge("chatbot", END) graph = builder.compile() # ★ compile 후에만 invoke/stream 가능

노드는 현재 상태를 받아 상태의 일부를 갱신해 반환하는 함수입니다. 위 chatbot은 누적된 메시지를 모델에 넣고 응답을 다시 messages 키로 돌려줍니다. 엣지는 START(진입점)와 END(종료점) 상수로 흐름을 잇습니다.

여기서 놓치기 쉬운 한 가지가 .compile()입니다. StateGraph는 설계도일 뿐입니다. 컴파일을 거쳐야만 실행 가능한 그래프(Runnable)가 됩니다. 컴파일된 그래프는 LCEL과 동일하게 invoke·stream을 노출하므로, 앞 편에서 익힌 Runnable 인터페이스가 그대로 통합니다.

3. State와 reducer — 덮어쓸까, 누적할까

State는 그래프를 흐르는 공유 상태입니다. 노드가 반환한 값을 기존 상태에 어떻게 합칠지가 LangGraph의 핵심 결정입니다. 기본 규칙은 키별 last-write-wins(덮어쓰기) — 같은 키를 두 번 쓰면 나중 값이 이깁니다. 그런데 대화 메시지처럼 쌓여야 하는 값은 덮어쓰면 안 됩니다. 이때 키에 reducer를 지정하면 동작이 "누적"으로 바뀝니다.

기본 — last-write-wins (덮어쓰기)

step: int

노드가 {"step": 2}를 반환하면 이전 값은 사라지고 2가 됩니다. 카운터·플래그·최종 결과처럼 최신값만 의미 있는 키에 적합.

reducer — 누적

messages: Annotated[list, add_messages]

노드가 새 메시지 한 개를 반환해도 기존 리스트 뒤에 이어붙습니다. 대화 이력·로그·수집 결과처럼 쌓여야 하는 키에 적합.

그림 1. 같은 키, reducer 유무에 따라 갈리는 상태 갱신

reducer는 Annotated[타입, 합치는_함수]로 선언합니다. 메시지 누적에는 전용 add_messages를 쓰는데, 단순 리스트 연결을 넘어 같은 id 메시지 갱신은 물론 dict→메시지 객체 자동 변환과 RemoveMessage 기반 이력 정리(트리밍)까지 해 줍니다. 일반 리스트 누적이면 operator.add도 가능합니다. 핵심은 "이 키는 덮어쓸 값인가, 쌓을 값인가"를 State를 설계할 때 미리 정하는 것입니다.

4. 조건부 엣지 — 길을 동적으로 가른다

고정 엣지(add_edge)는 항상 같은 다음 노드로 갑니다. 하지만 에이전트는 "지금 도구를 더 불러야 하나, 아니면 답을 내야 하나"처럼 상태를 보고 다음 길을 골라야 합니다. 이게 LCEL 파이프로는 안 되던 동적 분기입니다. LangGraph에서는 조건부 엣지로 표현합니다.

PYTHON · 조건부 엣지로 동적 라우팅

def route(state: State): last = state["messages"][-1] # 모델이 도구 호출을 요청했으면 tools로, 아니면 종료 return "tools" if getattr(last, "tool_calls", None) else END builder.add_conditional_edges("chatbot", route) # 라우팅 함수 반환값 = 다음 노드 builder.add_edge("tools", "chatbot") # 도구 실행 후 다시 모델로 (순환!)

add_conditional_edges("node", route_fn)는 라우팅 함수가 반환한 문자열을 다음 노드 이름으로 해석합니다. 위 코드에서 모델이 도구를 부르면 tools 노드로 갔다가 다시 chatbot으로 돌아오는 순환이 생깁니다. 이 "모델 → 도구 → 다시 모델" 루프가 바로 에이전트의 본체입니다. 4편에서 볼 create_agent가 이 패턴을 내부에서 컴파일해 줍니다. 지금은 "분기·순환이 엣지 한 줄로 표현된다"는 점만 잡으면 충분합니다.

5. 영속성 ① 단기 — checkpointer + thread_id

앞의 그래프는 한 번 호출하고 나면 상태를 잊습니다. 대화를 이어가려면 지난 턴의 상태를 저장·복원해야 합니다. LangGraph는 이걸 checkpointer로 처리합니다. 컴파일할 때 checkpointer를 끼웁니다. 호출할 때 thread_id를 주면 그 스레드의 메시지·상태가 자동으로 저장되고 다음 호출에서 복원됩니다.

PYTHON · 단기 메모리(스레드별 자동 저장·복원)

from langgraph.checkpoint.memory import InMemorySaver # 운영: SqliteSaver/PostgresSaver/RedisSaver graph = builder.compile(checkpointer=InMemorySaver()) # thread_id = 대화(사용자) 단위 분리 → 멀티유저 안전 graph.invoke({"messages": [{"role": "user", "content": "내 이름은 정식이야"}]}, config={"configurable": {"thread_id": "user-1"}}) graph.invoke({"messages": [{"role": "user", "content": "내 이름 뭐였지?"}]}, config={"configurable": {"thread_id": "user-1"}}) # 같은 스레드 → 기억함

thread_id가 핵심입니다. 이게 대화(사용자) 단위 구분선이라, "user-1"과 "user-2"는 서로의 대화를 보지 못합니다. 멀티유저 서비스가 한 그래프 인스턴스로 안전하게 돌아가는 이유입니다. InMemorySaver는 학습·프로토타입용(프로세스가 죽으면 사라짐)입니다. 운영에서는 SqliteSaver·PostgresSaver·RedisSaver로 교체합니다.

📌 checkpointer가 여는 두 가지 — ① time travel: get_state·update_state·get_state_history로 과거 상태를 조회·수정·분기할 수 있습니다. ② durability: invoke(..., durability="exit"|"async"|"sync")로 내결함성과 성능을 저울질합니다. 다음 절의 장기 기억과 HITL도 모두 이 checkpointer가 전제입니다.

6. 영속성 ② 장기 — Store(시맨틱 검색)

checkpointer가 "한 대화 안의 기억"이라면, Store는 대화·세션을 가로지르는 기억입니다. 사용자의 이름·선호·과거 사실처럼 스레드가 달라져도 유지돼야 하는 정보는 thread 단위 checkpointer로는 부족합니다. Store는 namespace 튜플로 데이터를 묶습니다. put/search로 저장·조회하며, 임베딩 인덱스를 주면 시맨틱 검색까지 됩니다.

PYTHON · 장기 메모리(스레드를 가로지르는 사용자 사실)

from langgraph.checkpoint.memory import InMemorySaver from langgraph.store.memory import InMemoryStore # 운영: PostgresStore # index를 주면 시맨틱 검색 활성화: index={"embed": emb, "dims": 1536} graph = builder.compile(checkpointer=InMemorySaver(), store=InMemoryStore()) # namespace 튜플로 사용자별 사실을 분리 저장 → search로 의미 기반 조회

정리하면 단기 = checkpointer + thread_id(한 대화의 메시지/상태), 장기 = Store + namespace(여러 대화에 걸친 사실·선호)로 역할이 갈립니다. 컴파일 시 두 인자를 함께 넘기면 한 그래프가 두 종류의 기억을 모두 갖습니다. 운영에서는 둘 다 영속 백엔드(PostgresSaver·PostgresStore)로 교체하는 게 표준입니다.

⚠️ 이름 충돌 주의 — 두 개의 InMemoryStore

LangGraph 장기 메모리의 langgraph.store.memory.InMemoryStore와, RAG에서 부모 문서 docstore로 쓰는 langchain.storage.InMemoryStore는 이름만 같고 전혀 다른 개념입니다(후자는 5편 ParentDocumentRetriever에서 등장). import 경로를 반드시 확인하세요 — 같은 클래스명에 속아 엉뚱한 걸 끼우면 디버깅이 길어집니다.

7. 스트리밍 — stream_mode로 무엇을 흘릴까

에이전트는 한 번 호출에 여러 노드를 거치므로, 최종 답만 기다리면 사용자는 한참 빈 화면을 봅니다. LangGraph는 graph.stream(...)의 stream_mode로 "무엇을 흘려보낼지"를 고릅니다. 용도에 따라 네 가지가 있습니다. 리스트로 여러 개를 동시에 받을 수도 있습니다.

stream_mode	내용	언제
values	각 스텝 후 전체 상태	현재 상태 스냅샷이 필요할 때
updates	스텝별 변경분만	"어느 노드가 무엇을 바꿨나" 추적
messages	LLM 토큰 + 메타 (chunk, metadata)	타자기 효과(토큰 단위 출력)
custom	get_stream_writer()로 발행한 사용자 이벤트	진행률·중간 알림 등 커스텀 UX

표 2. stream_mode 4종 — 흘려보낼 단위 선택

PYTHON · 토큰 단위 스트리밍

for chunk in graph.stream(inp, config, stream_mode="messages"): # 리스트 다중 지정 가능 ... # (chunk, metadata) — chunk.content를 이어붙여 타자기 효과

채팅 UI라면 messages로 토큰을 흘려 타자기 효과를 내고, 디버깅·관제 화면이라면 updates로 노드별 변경을 추적하는 식으로 골라 씁니다. 두 관심사가 동시에 필요하면 stream_mode=["messages", "updates"]처럼 리스트로 받으면 됩니다.

8. Human-in-the-Loop — interrupt() / Command(resume)

DB 쓰기·결제·메일 발송처럼 되돌리기 어려운 작업은 모델 혼자 실행하게 두면 위험합니다. LangGraph는 interrupt()로 그래프를 일시정지하고 사람의 결정을 받은 뒤, Command(resume=...)로 중단 지점부터 재개합니다. 그래프를 멈췄다 이어 가려면 상태를 저장해 둬야 하므로, HITL은 checkpointer + thread_id가 필수입니다.

PYTHON · 실행 전 사람 승인(저수준 HITL)

from langgraph.types import interrupt, Command def approve_node(state): decision = interrupt({"action": "send_email", "to": state["to"]}) # 일시정지 + payload return {"approved": decision} # 재개 (checkpointer + thread_id 필수) graph.invoke(Command(resume="approve"), config={"configurable": {"thread_id": "u1"}})

interrupt(payload)는 그래프를 멈추면서 payload(예: 보낼 메일 정보)를 바깥으로 내보냅니다. 사람이 검토 후 결정을 내리면 같은 thread_id로 Command(resume=...)를 호출해 그 값을 다시 그래프에 주입합니다. 이게 저수준 방식입니다. 4편에서 볼 HumanInTheLoopMiddleware는 이 패턴을 미들웨어로 감싼 고수준 버전입니다.

🕰️ 메모리 폐기 연쇄 — 결국 checkpointer로 수렴한다

옛 자료의 대화 기억 도구들은 줄줄이 폐기됐습니다. ConversationBufferMemory류(0.3.1 폐기)·ConversationChain·메모리 용도의 ChatMessageHistory는 폐기/langchain-classic으로 이동했고 1차 대체였던 RunnableWithMessageHistory마저 core 1.3.3에서 폐기됐습니다. 즉 옛 메모리 클래스를 찾고 있다면 답은 하나 — 이 글의 checkpointer/Store입니다. 긴 대화 압축은 trim_messages와 4편의 SummarizationMiddleware로 다룹니다.

시리즈: 01 LangChain 1.0 시작하기 → 02 LCEL & Runnable → 03 LangGraph 입문 → 04 create_agent → 05 현대 RAG → 06 LangSmith

9. 정리 · 다음 편

✅ LangGraph 핵심 5문장

순환·분기·다중 액터가 필요하면 LCEL이 아니라 StateGraph

상태는 키별로 덮어쓸지(기본) 누적할지(reducer) 설계 단계에서 정한다

단기 기억 = checkpointer + thread_id / 장기 기억 = Store + namespace

응답은 stream_mode(values/updates/messages/custom)로 흘린다

민감 작업은 interrupt() → Command(resume) (checkpointer 필수)

★ 마치며

LangGraph는 LCEL이 막히는 곳에서 시작합니다. 상태를 흐르게 하고(State·reducer), 길을 동적으로 가르고(조건부 엣지), 대화를 기억하고(checkpointer·Store), 사람을 끼워 넣는(interrupt) 네 가지가 이 런타임의 뼈대입니다. 이 뼈대를 손에 쥐면, 다음 편에서 볼 create_agent가 "왜 LangGraph 위에서 model → tool → final 루프를 컴파일하는지"가 자연스럽게 보입니다. 즉 04편은 이 그래프를 직접 짜는 대신, 같은 일을 선언적으로 해 주는 1.0 에이전트 표준과 미들웨어입니다.

#LangGraph #StateGraph #checkpointer #영속성 #스트리밍 #HumanInTheLoop #LangChain #기술블로그

📚 참고 자료 · 공식 문서

• langchain-ai/langchain — 공식 저장소(소스·릴리스·CHANGELOG)
• LangGraph — Graph API(StateGraph·노드·엣지)
• LangGraph — Persistence(checkpointer·Store·time travel)

본문 버전·API는 공식 문서 기반이며 LangChain API는 버전에 따라 변동될 수 있습니다(검증: LangChain 1.x, 2026-06).

안내 · 본 글은 LangChain 공식 문서·저장소를 기준으로 정리한 교육용 콘텐츠입니다. 코드는 개념 설명용 예시이며 LangChain API는 버전에 따라 달라질 수 있습니다.