LangGraph 입문

를 그래프로 본다. 생태계에서 갈라져 나와, 상태와 분기를 한곳에 묶는 전용 라이브러리로 자리 잡았다.

이 라이브러리를 이해하려면 먼저 그 모태인 이 어떤 문제를 풀려 했고 어디서 벽에 부딪혔는지를 짚는 편이 빠르다. 이어지는 일곱 슬라이드에서 LangChain의 핵심 부품(··)을 훑고, 왜 가 만들어졌는지를 본다. 그 다음에 그래프 자체로 들어간다.


① LangChain은 무엇이었나 — 컴포넌트 박스의 시대

은 처음에 LLM 앱을 만들 때마다 반복되는 부품을 표준화한 라이브러리였다. 호출할 모델, 변수 자리가 있는 , 응답을 구조화 데이터로 바꾸는 , 대화 히스토리를 이어 붙이는 , 외부 시스템을 호출하는 . 다섯 종류의 박스를 미리 만들어 두고, 사용자는 그것들을 조립만 한다.

이 시기의 핵심 단어는 체인이었다. LLMChain, ConversationChain, RetrievalQA 같은 클래스가 가장 자주 쓰였고, 모두 “프롬프트 → 모델 → 파서”라는 한 가닥 흐름을 갖고 있었다. 입력 한 번 → 출력 한 번. 분기도 반복도 없다.

한 줄의 체인이 한 번의 LLM 호출에 정확히 대응했다. 그래서 디버깅이 쉬웠다.

박스가 일관된 인터페이스를 가져야 조립이 가능하다. 그 인터페이스가 이다. 모든 LangChain 부품이 이 한 가지 추상을 구현한다. 다음 슬라이드에서 본다.


② LCEL — 파이프로 잇는 체인

(LangChain Expression Language)은 Unix 파이프 비유를 그대로 가져온 컴포지션 문법이다. prompt | llm | parser 한 줄이 곧 체인이다. | 연산자는 내부적으로 를 만든다.

핵심 약속은 인터페이스 통일이다. 모든 invoke(동기 한 번 호출), stream(토큰 단위 스트리밍), batch(여러 입력 일괄 처리), 그리고 그 비동기 짝꿍 ainvoke / astream / abatch를 노출한다. 어떤 부품이든 이 메서드 이름만 알면 호출 방법이 같다.

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

prompt = ChatPromptTemplate.from_messages([
  ("system", "너는 한 줄 요약 봇이다."),
  ("user", "{text}"),
])
model = ChatOpenAI(model="gpt-4o-mini")
parser = StrOutputParser()

# | 연산자가 RunnableSequence를 만든다
chain = prompt | model | parser

print(chain.invoke({"text": "에이전트는 도구를 호출한다."}))

# 같은 체인을 그대로 스트리밍/배치로도 호출 가능
for tok in chain.stream({"text": "..."}):
  print(tok, end="")

# 여러 입력 일괄
chain.batch([{"text": "a"}, {"text": "b"}])

병렬 호출이 필요할 때는 을 쓴다. 같은 입력을 여러 체인에 동시에 흘려보내고 결과를 dict로 모아준다. 분기·합류·전처리 같은 흔한 패턴이 이 두 컨테이너 위에서 표현된다.


③ AgentExecutor의 시대 — 한 객체에 담은 루프

체인은 한 가닥 흐름에는 잘 맞았지만 도구 호출 루프에는 부족했다. 도구를 호출하면 결과(observation)가 다시 LLM에 들어가서 다음 행동을 결정해야 하고, 그게 끝날 때까지 같은 사이클을 반복해야 한다. 그래서 등장한 것이 다.

AgentExecutor(Reason → Act → Observe → Repeat)를 한 객체로 감쌌다. 사용자는 모델, 도구 목록, 프롬프트만 주면 되고, 멈춤 조건은 max_iterations(최대 반복 횟수)와 early_stopping_method(중단 방식) 같은 설정으로 조절한다. 중간 추론 흔적은 intermediate_steps에 리스트로 누적된다.

# LangChain Classic 시기의 전형적 패턴
# (v1에서는 AgentExecutor 계열이 langchain_classic.agents 로 이동했다)
from langchain_classic.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

@tool
def search(query: str) -> str:
  """웹을 검색한다."""
  return f"결과: {query}"

prompt = ChatPromptTemplate.from_messages([
  ("system", "도구를 활용해 사용자 질문에 답한다."),
  ("user", "{input}"),
  ("placeholder", "{agent_scratchpad}"),
])
model = ChatOpenAI(model="gpt-4o-mini")

agent = create_tool_calling_agent(model, [search], prompt)
executor = AgentExecutor(
  agent=agent,
  tools=[search],
  max_iterations=8,        # 무한 루프 방지
  return_intermediate_steps=True,
)
result = executor.invoke({"input": "에이전트 책을 검색해줘"})

이 모델의 강점은 캡슐화였다. 한 객체에 모델·도구·루프가 다 들어 있어서 호출 한 줄로 끝났다. 약점도 같은 자리에서 생겼다. 어디서 멈출지, 어떻게 중간 검수할지, 사람을 어떻게 끼워 넣을지 — 흐름에 손대고 싶을 때는 객체 안쪽을 열어야 했다.

참고: 위 import 경로가 langchain.agents가 아니라 langchain_classic.agents인 이유가 그것이다. 최신 LangChain(v1)에서 langchain.agents는 새로운 create_agent 팩토리만 노출하고, AgentExecutor 계열은 classic 패키지로 옮겨졌다. 신규 코드의 권장 경로는 다음 슬라이드에서 본다.


④ 벽에 부딪힌 지점 — 체인의 다섯 가지 한계

체인 한 가닥과 한 객체짜리 루프로는 표현이 어색한 흐름이 있다. 다섯 가지가 반복해서 나타났다.

한계무엇이 어려웠나
분기”답이 충분하면 멈추고, 부족하면 검색을 더 한다”를 체인 안에 표현하려면 RunnableBranch 같은 우회로가 필요했다
반복”조건이 맞을 때까지 N회 반복”이 한 가닥 체인에 자연스럽게 안 들어갔다. AgentExecutor는 했지만 그 내부를 들여다보기 어려웠다
상태 보관메모리는 대화 히스토리 한 종류에 최적화돼 있었고, “지금까지 수집한 근거 5개”처럼 임의 상태를 깔끔히 다루기 어려웠다
휴먼-인-더-루프도구 호출 직전에 사람에게 보여 주고 승인받는 동작은 AgentExecutor의 intermediate_steps만으로는 멈춰서 사용자 입력을 기다린다는 의미를 표현하기 부족했다
다중 에이전트한 객체에 모델·도구·루프를 다 담는 구조에서, 여러 에이전트가 메시지를 주고받게 하려면 또 다른 추상이 필요했다

공통 원인은 자료구조에 있었다. 흐름은 본질적으로 그래프인데(분기와 합류와 반복이 있는 방향 그래프) 표현은 리스트(체인 한 가닥) 또는 블랙박스(AgentExecutor 한 객체) 둘 중 하나였다. 자료구조와 도메인이 어긋나면 표현은 자꾸 어색해진다.

해법은 도메인 그대로의 자료구조를 쓰는 것이었다. 그래프를 그래프로 쓰면 된다.


⑤ 그래서 LangGraph가 나왔다

대체하지 않는다. LangChain의 부품(·모델·도구·파서)을 그대로 노드 함수 안에서 호출하고, 흐름만 그래프로 끌어올린다. 노드는 한 단계의 계산이고, 엣지는 다음 단계로의 전이이며, 상태는 그래프 전체가 공유하는 메모리다.

이렇게 보면 LangChain에서 익숙했던 추상들이 LangGraph에서는 다음과 같이 자리를 옮긴다.

LangChain ClassicLangGraph
Runnable (모델·프롬프트·파서)노드 함수의 내부에서 그대로 호출
AgentExecutor 루프 +
intermediate_steps 리스트State 스키마의 messages 필드 (+ 리듀서)
max_iterations 노브그래프 레벨의 recursion_limit
ConversationBufferMemory상태 스키마 + 체크포인터
RunnableBranch 우회로add_conditional_edges로 분기 일급화

오늘날 신규 코드의 시작점은 두 곳이다. 단순 ReAct 에이전트는 from langgraph.prebuilt import create_react_agent 한 줄이면 컴파일된 그래프()가 나온다. 더 복잡한 흐름은 StateGraph를 직접 짠다. 두 경우 모두 결과물은 이라서 invoke / stream / batch로 호출하는 호출 규약은 그대로 유지된다.

결론은 단순하다. LangChain을 안다는 건 LangGraph를 거의 다 안다는 뜻이다. 부품은 같고, 다만 흐름의 자료구조가 그래프로 승격됐을 뿐이다.


⑥ 두 세계가 만나는 코드

실제 프로젝트에서는 노드 함수 안에서 체인을 그대로 호출하는 패턴이 가장 흔하다. 그래프는 흐름을 담당하고, 노드 내부는 여전히 이 담당한다.

from typing import TypedDict, Annotated
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

# 1) LangChain Runnable — 그대로 LCEL 체인
prompt = ChatPromptTemplate.from_messages([
  ("system", "너는 한 줄 요약 봇이다."),
  ("user", "{text}"),
])
model = ChatOpenAI(model="gpt-4o-mini")
summarize_chain = prompt | model | StrOutputParser()

# 2) LangGraph 상태 + 노드
class State(TypedDict):
  text: str
  summary: str

def summarize(state: State) -> dict:
  # 노드 안에서 LCEL 체인을 그대로 호출
  return {"summary": summarize_chain.invoke({"text": state["text"]})}

# 3) 그래프 조립
graph = (
  StateGraph(State)
  .add_node("summarize", summarize)
  .add_edge(START, "summarize")
  .add_edge("summarize", END)
  .compile()
)

print(graph.invoke({"text": "...", "summary": ""}))

여기서 주목할 점은 어디서 흐름이 보이고 어디서 호출 규약이 유지되는가다. 그래프는 add_node / add_edge / add_conditional_edges로 명시적으로 그려지고, 노드 안쪽은 여전히 chain.invoke(...) 한 줄이다. 두 추상이 충돌하지 않고 서로의 자리를 지킨다.


⑦ 한눈에 정리하는 매핑

지금까지의 비교를 한 장에 다시 모아 둔다.

필요한 것LangChain Classic 방식LangGraph 방식왜 LangGraph가 자연스러운가
순차 흐름prompt | model | parser ()한 노드 안에서 그대로 LCEL 호출차이 없음, 그대로 재사용
분기RunnableBranch 또는 코드 안 if/elseadd_conditional_edges분기가 그래프의 일급 구성 요소
반복AgentExecutor의 내부 루프노드 → 조건부 엣지 → 자기 자신반복 횟수/조건이 그래프에 명시됨
상태 보관ConversationBufferMemory 등 종류별 클래스State 스키마 + 리듀서임의 상태를 한 자리에 정의
휴먼 개입외부에서 우회 구현interrupt_before/after로 일급 지원그래프 흐름의 한 노드로 표현
다중 에이전트한 객체 안에 어색하게 끼워넣음슈퍼바이저/스웜 패턴(11~13장)노드가 곧 에이전트, 엣지가 곧 메시지 흐름
호출 규약 표준컴파일된 그래프도 invoke/stream/batch 그대로

이 표가 머리에 있으면 다음부터의 모든 슬라이드가 가벼워진다. 그래프, 상태, 노드, 엣지 — 네 단어가 LangChain의 어떤 자리들을 흡수했는지가 보인다.


왜 그래프인가

체인 한 가닥으로는 못 푸는 문제가 있다. 분기·반복·재시도가 섞이면 가 자연스러운 표현이다.

를 표현한다. 한 사이클이 곧 한 바퀴 순회다.

그래프는 흐름을, 상태는 기억을 책임진다.

다이어그램 로딩…

핵심 4요소

, , , . 이 넷이면 거의 모든 를 그릴 수 있다.

요소책임
State그래프 전체가 공유하는 데이터
Node한 단계의 계산. 보통 함수
Edge다음 노드로의 전이
Graph위 셋을 묶어 실행 가능한 컴파일 단위

가장 단순한 그래프

상태 정의, 추가, 연결, 컴파일. 이 네 줄이 의 본질이다. START → 처리 → END 흐름을 그대로 코드로 옮긴다. TypedDictAnnotation으로 미리 선언한다.

# 가장 단순한 LangGraph 예제
from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
  text: str

def upper(state: State) -> dict:
  return {"text": state["text"].upper()}

graph = StateGraph(State)
graph.add_node("upper", upper)
graph.add_edge(START, "upper")
graph.add_edge("upper", END)
app = graph.compile()

print(app.invoke({"text": "hello"}))  # {'text': 'HELLO'}

상태 스키마

의 메모리다. Python은 TypedDictPydantic, TypeScript는 Annotation.Root로 정의한다.

리듀서를 지정하면 가 반환한 값이 덮어쓰기가 아닌 병합으로 적용된다. 메시지 누적 리스트가 대표 사례다.

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

# add_messages 리듀서: 새 메시지를 append
class ChatState(TypedDict):
  messages: Annotated[list, add_messages]

# 노드는 부분 상태만 반환 — 리듀서가 합쳐 준다
def echo(state: ChatState) -> dict:
  last = state["messages"][-1]
  return {"messages": [{"role": "assistant", "content": last["content"]}]}

graph = StateGraph(ChatState).add_node("echo", echo)

ReAct를 그래프로

LLM 와 도구 노드가 번갈아 도는 사이클이 다. 두 노드 사이에 를 두면 “도구 호출이 더 필요한가”를 동적으로 판단한다. 의 의사결정 지점이 바로 여기다.

다이어그램 로딩…

ToolNode와 조건부 엣지

langgraph.prebuilt에 있는 ToolNode 전담 를 한 줄로 만들어 준다. tools_condition은 LLM 결과에 도구 호출이 있으면 tools 노드로, 없으면 END로 보내는 분기 함수다. 이 둘이 의 표준 조합이다.

from langgraph.graph import StateGraph, START
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
from typing import Annotated, TypedDict
from langchain_anthropic import ChatAnthropic

class State(TypedDict):
  messages: Annotated[list, add_messages]

def add(a: int, b: int) -> int:
  """두 수를 더한다."""
  return a + b

llm = ChatAnthropic(model="claude-sonnet-4-5").bind_tools([add])

def call_llm(state: State) -> dict:
  return {"messages": [llm.invoke(state["messages"])]}

g = StateGraph(State)
g.add_node("llm", call_llm)
g.add_node("tools", ToolNode([add]))
g.add_edge(START, "llm")
g.add_conditional_edges("llm", tools_condition, {"tools": "tools", "__end__": "__end__"})
g.add_edge("tools", "llm")
app = g.compile()

컴파일과 실행

compile() 정의를 실행 가능한 CompiledGraph로 굳힌다. 이후엔 invoke, stream, batch 같은 메서드가 노출된다.

이 단계에서 ·인터럽트·서브그래프 같은 부가 옵션을 함께 끼워 넣는다. 실행은 결국 invokestream 한 줄이다.

# 그래프 실행 — 동기 vs 스트리밍
result = app.invoke({"messages": [{"role": "user", "content": "2 + 3은?"}]})

# 노드 단위 진행 상황 스트림
for chunk in app.stream({"messages": [{"role": "user", "content": "2 + 3은?"}]}):
  print(chunk)  # {"llm": {...}}, {"tools": {...}} ...

체크포인터로 영속성 켜기

는 매 실행 직후 스냅샷을 저장한다. 프로세스가 죽었다 살아나도 같은 자리에서 재개되고, 과거 어느 지점으로 돌아갈 수도 있다.

학습용엔 InMemorySaver(구 이름 MemorySaver), 프로덕션은 Postgres/SQLite 백엔드 어댑터를 쓴다.

from langgraph.checkpoint.memory import InMemorySaver

# 컴파일할 때 checkpointer를 주입
app = g.compile(checkpointer=InMemorySaver())

# 스레드 ID로 한 대화를 묶는다
config = {"configurable": {"thread_id": "user-42"}}
app.invoke({"messages": [{"role": "user", "content": "안녕"}]}, config)
app.invoke({"messages": [{"role": "user", "content": "방금 뭐라 했지?"}]}, config)

스레드 — 대화의 키

는 한 사용자·한 세션을 식별하는 키다. thread_id가 같으면 는 동일 대화의 연속으로 본다.

여러 사용자가 한 인스턴스를 공유해도 스레드만 다르면 가 안 섞인다.


상태 조회와 시간여행 맛보기

가 있으면 get_state로 현재 를, get_state_history로 과거 스냅샷 목록을 가져올 수 있다. 17장의 시간여행 디버깅으로 이어지는 기능이다. 단위로 묶여서 반환된다.

# 현재 상태
snapshot = app.get_state(config)
print(snapshot.values, snapshot.next)

# 과거 스냅샷 목록 (최신 → 오래된)
for s in app.get_state_history(config):
  print(s.config["configurable"]["checkpoint_id"], s.values.get("messages", []))

정적 vs 조건부 엣지

는 두 종류다. add_edge(a, b)는 무조건 a → b. add_conditional_edges(a, fn, mapping)는 함수가 반환한 키로 다음 를 고른다.

가 곧 의 의사결정 지점이다. 여기에 LLM 출력·도구 호출·외부 신호가 모두 들어온다.

def route(state: State) -> str:
  # 상태를 보고 다음 라벨을 반환
  return "approve" if state["score"] >= 0.8 else "reject"

g.add_conditional_edges("classify", route, {
  "approve": "send_email",
  "reject": "human_review",
})

최소 ReAct 에이전트 한 줄

langgraph.prebuiltcreate_react_agent는 위 모든 조각을 한 줄로 묶는다. 학습엔 풀버전이, 실전엔 prebuilt가 답이다. 를 만들고 를 끼우고 설정만 주면 끝이다.

from langgraph.prebuilt import create_react_agent
from langgraph.checkpoint.memory import InMemorySaver
from langchain_anthropic import ChatAnthropic

agent = create_react_agent(
  ChatAnthropic(model="claude-sonnet-4-5"),
  tools=[add],
  checkpointer=InMemorySaver(),
)
agent.invoke(
  {"messages": [{"role": "user", "content": "2 더하기 3"}]},
  {"configurable": {"thread_id": "demo"}},
)

다음 장으로

기본기를 익혔으니 의 진짜 무기 — 서브그래프, 인터럽트, 시간여행, 스트리밍 — 으로 넘어가자. 큰 를 모듈화하고, 사람을 끼워 넣고, 과거로 되감는 방법을 17장에서 본다. 빌더의 진짜 깊이는 그쪽이다.