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 Classic | LangGraph |
|---|---|
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/else | add_conditional_edges | 분기가 그래프의 일급 구성 요소 |
| 반복 | AgentExecutor의 내부 루프 | 노드 → 조건부 엣지 → 자기 자신 | 반복 횟수/조건이 그래프에 명시됨 |
| 상태 보관 | ConversationBufferMemory 등 종류별 클래스 | State 스키마 + 리듀서 | 임의 상태를 한 자리에 정의 |
| 휴먼 개입 | 외부에서 우회 구현 | interrupt_before/after로 일급 지원 | 그래프 흐름의 한 노드로 표현 |
| 다중 에이전트 | 한 객체 안에 어색하게 끼워넣음 | 슈퍼바이저/스웜 패턴(11~13장) | 노드가 곧 에이전트, 엣지가 곧 메시지 흐름 |
| 호출 규약 | 표준 | 컴파일된 그래프도 | invoke/stream/batch 그대로 |
이 표가 머리에 있으면 다음부터의 모든 슬라이드가 가벼워진다. 그래프, 상태, 노드, 엣지 — 네 단어가 LangChain의 어떤 자리들을 흡수했는지가 보인다.
왜 그래프인가
체인 한 가닥으로는 못 푸는 문제가 있다. 분기·반복·재시도가 섞이면 가 자연스러운 표현이다.
는 와 로 를 표현한다. 한 사이클이 곧 한 바퀴 순회다.
그래프는 흐름을, 상태는 기억을 책임진다.
핵심 4요소
, , , . 이 넷이면 거의 모든 를 그릴 수 있다.
| 요소 | 책임 |
|---|---|
| State | 그래프 전체가 공유하는 데이터 |
| Node | 한 단계의 계산. 보통 함수 |
| Edge | 다음 노드로의 전이 |
| Graph | 위 셋을 묶어 실행 가능한 컴파일 단위 |
가장 단순한 그래프
상태 정의, 추가, 연결, 컴파일. 이 네 줄이 의 본질이다. START → 처리 → END 흐름을 그대로 코드로 옮긴다. 는 TypedDict나 Annotation으로 미리 선언한다.
# 가장 단순한 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은 TypedDict나 Pydantic, 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 같은 메서드가 노출된다.
이 단계에서 ·인터럽트·서브그래프 같은 부가 옵션을 함께 끼워 넣는다. 실행은 결국 invoke나 stream 한 줄이다.
# 그래프 실행 — 동기 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.prebuilt의 create_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장에서 본다. 빌더의 진짜 깊이는 그쪽이다.