4,629
💬 LangGraph는 다른 프레임워크보다 작고 가벼우며, 개발자가 원하는 만큼 커스텀(컨트롤)할 수 있다는 장점이 있다. 이름 그대로 "agent = graph"라는 아이디어로 만들어진 프레임워크로, 그래프를 실행할 수 있는 코드를 제공한다.
에이전트를 설계할 때 "이 노드에서 저 노드로 흐른다"는 그림을 떠올려 보면, 결국 에이전트는 하나의 그래프로 표현할 수 있다. LangGraph는 바로 그 그래프를 코드로 만들 수 있게 도와준다.
💡 예를 들어 LangGraph는 node를 만들 수 있게 도와준다. node는 _무엇이든 할 수 있는 일종의 함수_다.
| 구성요소 | 역할 |
|---|---|
| State | 그래프를 통해 이동하는 데이터. 유저 입력 등 그래프가 다루는 값이 여기에 담긴다. TypedDict로 어떤 프로퍼티가 올 수 있는지 미리 정의한다. |
| Node | 실질적인 작업이 이루어지는 함수. state를 인자로 받고, 정해진 규칙 없이 무엇이든 할 수 있다. |
| Edge | 노드를 연결하는 화살표. add_edge(A, B)는 A에서 B로 흐름을 잇는다. |
에이전트를 설계할 때 그래프로 그리는 것을 생각해 보면 Agent = Graph라고 할 수 있다!
아래에서 TypedDict는 그냥 딕셔너리인데, 어떤 프로퍼티들이 올 수 있는지를 미리 지정해 놓은 딕셔너리다.
from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict
# State: 그래프의 데이터
class State(TypedDict):
hello: str
# 그래프 빌더 생성
graph_builder = StateGraph(State)
# Node = function (state를 받아서 무엇이든 할 수 있다)
def node_one(state: State):
print("node_one")
def node_two(state: State):
print("node_two")
def node_three(state: State):
print("node_three")
# 그래프에 node 할당
graph_builder.add_node("node_1", node_one)
graph_builder.add_node("node_2", node_two)
graph_builder.add_node("node_3", node_three)
# Edge로 node 연결 — add_edge(A, B): A → B
graph_builder.add_edge(START, "node_1")
graph_builder.add_edge("node_1", "node_2")
graph_builder.add_edge("node_2", "node_3")
graph_builder.add_edge("node_3", END)
# compile() 시 LangGraph가 그래프의 유효성을 검사한다
graph = graph_builder.compile()
graph💡
graph_builder.compile()을 호출하면 LangGraph가 이 그래프가 유효한지 검사한 뒤 실행 가능한 그래프 객체를 돌려준다.
Graph에는 State가 있고, 여기에 데이터를 넣고 바꾸는 흐름은 다음 3단계로 정리된다.
graph.invoke(...)의 input으로 넘긴다. 이를 위해 State(TypedDict)에 프로퍼티를 미리 정의해 두어야 한다.return**한다.💡 State를 수정할 수 있는 건 오직 node다. node는 state를 받고, 업데이트된 값을
return하기만 하면 된다. state를 명시하거나 별도의 함수·API를 알 필요가 없다. 그리고 다음 node에게 직접 넘기는 게 아니라 — 다음 node는 변경된 state를 "쳐다본다".
from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict
class State(TypedDict):
hello: str
a: bool
graph_builder = StateGraph(State)
def node_one(state: State):
print("node_one", state) # 초기 state (예: 사용자 input)
return {"hello": "from node one", "a": True}
def node_two(state: State):
print("node_two", state) # node_one이 return한 state를 받음
return {"hello": "from node two", "a": False}
def node_three(state: State):
print("node_three", state) # node_two가 return한 state를 받음
return {"hello": "from node three"}
graph_builder.add_node("node_1", node_one)
graph_builder.add_node("node_2", node_two)
graph_builder.add_node("node_3", node_three)
graph_builder.add_edge(START, "node_1")
graph_builder.add_edge("node_1", "node_2")
graph_builder.add_edge("node_2", "node_3")
graph_builder.add_edge("node_3", END)
graph = graph_builder.compile()
# state 초기화 — invoke 할 때 넘긴다
result = graph.invoke({"hello": "world"})
print(result)각 node의 print를 따라가 보면 state가 어떻게 흘러가는지 보인다.
node_one {'hello': 'world', ...}
node_two {'hello': 'from node one', 'a': True}
node_three {'hello': 'from node two', 'a': False}💡
hello는 매 node마다 덮어써지고,a는node_three가 건드리지 않았으므로 직전 값(False)이 그대로 유지된다. 이렇게 State는 여러 필드를 가질 수 있고, 각 node는 자신이 바꾸고 싶은 필드만 골라서 return하면 된다. 마지막node_three가 return한 state가 곧final_state로 반환된다.
state는 여러 개의 스키마를 가질 수 있다. 대표적으로 다음 3가지다.
✅ 이렇게 스키마를 구분하면 state를 divide and conquer(분할과 정복) 할 수 있다. 예를 들어 사용자가 내부 state에 접근하지 못하게 하거나, 최종 output에서 특정 키를 숨기고 싶을 때 유용하다.
from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict
# private state: node 내부에서만 쓰는 state
class PrivateState(TypedDict):
a: int
b: int
# input state: 사용자가 제공하는 state
class InputState(TypedDict):
hello: str
# output state: agent의 output을 담는 state
class OutputState(TypedDict):
bye: str
# input은 input_schema에, output은 output_schema에,
# private(overall) state는 첫 번째 인자로 등록
graph_builder = StateGraph(
PrivateState,
input_schema=InputState,
output_schema=OutputState,
)
# 각 node는 원하는 스키마를 선택해 state: ... 로 명시하면 된다
def node_one(state: InputState) -> InputState:
print("node_one", state)
return {"hello": "world"}
def node_two(state: PrivateState) -> PrivateState:
print("node_two", state)
return {"a": 1}
def node_three(state: PrivateState) -> PrivateState:
print("node_three", state)
return {"b": 1}
# private state를 받지만 output state를 업데이트
def node_four(state: PrivateState) -> OutputState:
print("node_four", state)
return {"bye": "world"}
graph_builder.add_node("node_1", node_one)
graph_builder.add_node("node_2", node_two)
graph_builder.add_node("node_3", node_three)
graph_builder.add_node("node_4", node_four)
graph_builder.add_edge(START, "node_1")
graph_builder.add_edge("node_1", "node_2")
graph_builder.add_edge("node_2", "node_3")
graph_builder.add_edge("node_3", "node_4")
graph_builder.add_edge("node_4", END)
graph = graph_builder.compile()
graph✅ 각 node는
state: InputState,state: PrivateState처럼 자기가 다룰 스키마만 골라서 타입으로 명시하면 된다. LangGraph가 알맞은 채널을 연결해 준다.
node가 무언가를 return하면 그게 state를 업데이트한다. 그렇다면 그 업데이트가 정확히 어떻게 일어나는가?
⚠️ 기본 동작은 "덮어쓰기(override)"다. 예를 들어
messagesstate에서 마지막 메시지를 LLM에 보내고, 응답으로 state를 업데이트한다고 하자. 해당 필드를 그냥 return하면 이전 messages가 통째로 덮어써진다.
이전 값을 보존하고 싶다면 두 가지 방법이 있다.
return {"messages": state["messages"] + ["Hello!"]}💡 reducer function은 말 그대로 함수다.
typing의Annotated를 이용해 필드에 붙인다.messages: Annotated[list[str], update_function]— 첫 번째 인자는 필드의 타입, 두 번째 인자는 어떻게 업데이트할지 정하는 함수다. 이 함수는old(이전 state),new(새 state)를 인자로 받는다.
from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict
from typing import Annotated
import operator
# reducer 예시: 직접 정의하면 old + new 형태
def update_function(old, new):
return old + new
class State(TypedDict):
# messages: Annotated[list[str], update_function]
# 리스트를 이어 붙이는 reducer는 operator.add로 간단히 대체 가능
messages: Annotated[list[str], operator.add]
graph_builder = StateGraph(State)
def node_one(state: State):
return {"messages": ["Hello, nice to meet you!"]}
def node_two(state: State):
return {}
def node_three(state: State):
return {}
def node_four(state: State):
return {}
graph_builder.add_node("node_1", node_one)
graph_builder.add_node("node_2", node_two)
graph_builder.add_node("node_3", node_three)
graph_builder.add_node("node_4", node_four)
graph_builder.add_edge(START, "node_1")
graph_builder.add_edge("node_1", "node_2")
graph_builder.add_edge("node_2", "node_3")
graph_builder.add_edge("node_3", "node_4")
graph_builder.add_edge("node_4", END)
graph = graph_builder.compile()
graph.invoke({"messages": ["Hello!"]})
# → {'messages': ['Hello!', 'Hello, nice to meet you!']}✅
operator.add를 reducer로 지정했기 때문에,node_one이 return한["Hello, nice to meet you!"]는 초기값["Hello!"]를 덮어쓰지 않고 뒤에 이어 붙는다. 채팅 기록처럼 누적되어야 하는 state에 딱 맞는 패턴이다.
LangGraph에서는 node의 결과를 캐싱할 수 있다. 비용이 큰 연산을 반복 실행하지 않도록 아껴 준다.
💡 적용은 두 군데에서 이뤄진다.
컴파일 시:
graph_builder.compile(cache=InMemoryCache())로 캐시 저장소를 넘긴다.노드 추가 시:
add_node(..., cache_policy=CachePolicy(ttl=20))로 만료 시간(TTL)을 설정한다.
ttl=20이면, 그 node가 최초 실행된 뒤 20초가 지나기 전에 같은 입력으로 다시 실행되면 node를 실행하지 않고 캐시된 결과를 사용한다.
from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict
from langgraph.types import CachePolicy
from langgraph.cache.memory import InMemoryCache
from datetime import datetime
import time
class State(TypedDict):
time: str
graph_builder = StateGraph(State)
def node_one(state: State):
return {}
def node_two(state: State):
# 캐시가 살아 있으면 이 값은 갱신되지 않는다
return {"time": f"{datetime.now()}"}
def node_three(state: State):
return {}
def node_four(state: State):
return {}
graph_builder.add_node("node_1", node_one)
# node_2에만 20초 TTL 캐시 정책 적용
graph_builder.add_node("node_2", node_two, cache_policy=CachePolicy(ttl=20))
graph_builder.add_node("node_3", node_three)
graph_builder.add_node("node_4", node_four)
graph_builder.add_edge(START, "node_1")
graph_builder.add_edge("node_1", "node_2")
graph_builder.add_edge("node_2", "node_3")
graph_builder.add_edge("node_3", "node_4")
graph_builder.add_edge("node_4", END)
# 컴파일 시 캐시 저장소를 넘긴다
graph = graph_builder.compile(cache=InMemoryCache())
# 5초 간격으로 반복 호출 — 20초 안에는 time 값이 그대로 유지된다
for _ in range(7):
print(graph.invoke({}))
time.sleep(5)💡 실행해 보면, 처음 20초 동안은
node_2가 만든time값이 동일하게 찍힌다. TTL이 지난 뒤에야 node가 다시 실행되어 새 시각으로 갱신된다.
add_conditional_edges는 라우팅 함수의 반환값에 따라 다음 노드를 동적으로 결정하는 갈림길을 만든다.
.add_conditional_edges("시작점 노드", 라우팅_함수)라우팅 함수도 state를 받으며, edge의 이름(=다음 노드)을 return해야 한다. 가장 직관적인 형태는 노드 이름을 직접 반환하는 것이다.
def decide_path(state: State) -> Literal["node_3", "node_4"]:
if state["seed"] % 2 == 0:
return "node_3"
else:
return "node_4"✅ 하지만 노드 이름을 직접 반환하지 않는 방법도 있다.
bool이나 DB·API에서 받아온 값을 반환하고,add_conditional_edges의 세 번째 인자에 반환값 → 노드 맵을 넘기는 방식이다.pythongraph_builder.add_conditional_edges( "node_2", decide_path, {True: "node_3", False: "node_4"} )이렇게 하면 라우팅 함수는 노드 이름을 몰라도 되고, 라우팅 규칙을 한 군데(맵)에서 관리할 수 있다.
from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict
class State(TypedDict):
seed: int
graph_builder = StateGraph(State)
def node_one(state: State):
return {}
def node_two(state: State):
return {}
def node_three(state: State):
return {}
def node_four(state: State):
return {}
graph_builder.add_node("node_1", node_one)
graph_builder.add_node("node_2", node_two)
graph_builder.add_node("node_3", node_three)
graph_builder.add_node("node_4", node_four)
# 노드 이름 대신 bool을 반환하고, 맵으로 라우팅
def decide_path(state: State):
return state["seed"] % 2 == 0
graph_builder.add_edge(START, "node_1")
graph_builder.add_edge("node_1", "node_2")
graph_builder.add_conditional_edges(
"node_2", decide_path, {True: "node_3", False: "node_4", "hello": END}
)
graph_builder.add_edge("node_4", END)
# node_3는 END에 명시적으로 연결하지 않았다 (아래 콜아웃 참고)
graph = graph_builder.compile()
graph💡 **
node_3**를 **END**에 연결하지 않았는데도 그래프가 정상 종료된다. 이유는 — outgoing edge가 없는 node는 기본적으로 END로 연결되기 때문이다. 실제로seed=2(짝수 →node_3)로 실행해 보면,node_3가 실행된 뒤 그래프가 깔끔하게 종료되고 결과가 반환된다. 맵에 있는"hello": END처럼 라우팅 함수가 실제로 반환하지 않는 키는 그냥 무시될 뿐 에러를 일으키지 않는다.
Send 클래스는 custom state를 담아 node를 동적으로 invoke할 수 있게 해 준다.
💡 왜 동적으로 호출해야 할까? node를 몇 번 invoke할지 미리 알 수 없는 경우가 있기 때문이다. 예를 들어 문서 요약 그래프에서 문서가 하나면 node를 한 번만 호출하면 되지만, 개수가 정해지지 않은 파일들을 요약해야 한다면 Send API로 필요한 만큼 node를 만들어낼 수 있다.
✅ 가장 큰 장점: LangGraph에서 Send로 여러 node를 동시에 실행하면 이들이 병렬로 실행된다.
from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict, Annotated
from typing import Union
from langgraph.types import Send
import operator
class State(TypedDict):
words: list[str] # input 형태
output: Annotated[list[dict[str, Union[str, int]]], operator.add] # output 형태 (누적)
graph_builder = StateGraph(State)
def node_one(state: State):
return {}
# Send로 넘긴 payload(여기서는 word 문자열)를 그대로 받는다
def node_two(word: str):
count = len(word)
return {"output": [{"word": word, "letters": count}]}
graph_builder.add_node("node_1", node_one)
graph_builder.add_node("node_2", node_two)
# 각 word를 Send로 감싸 "node_2"에 개별 전달 → word 개수만큼 node_2가 생성된다
def dispatcher(state: State):
return [Send("node_2", word) for word in state["words"]]
graph_builder.add_edge(START, "node_1")
# node_1 → node_2로 가되, node_2가 몇 개 생길지 모르므로 conditional edge 사용
# 세 번째 인자로 도달 가능한 노드 목록을 명시
graph_builder.add_conditional_edges("node_1", dispatcher, ["node_2"])
graph_builder.add_edge("node_2", END)
graph = graph_builder.compile()
graph.invoke({"words": ["hello", "world", "how", "are", "you", "doing"]})💡 **
Send("node_2", word)**의 두 번째 인자는 대상 node에 전달할 payload다. 여기서는 문자열word를 그대로 넘겼고,node_two(word: str)가 이를 받아len(word)로 글자 수를 센다. 이 repo 환경에서 실제로 실행하면 각 단어가 병렬로 처리되어output에 누적된다:[{'word': 'hello', 'letters': 5}, {'word': 'world', 'letters': 5}, {'word': 'how', 'letters': 3}, ...]
(공식 예제에서는Send("node", {"subject": s})처럼 dict를 넘기고 node가state["subject"]로 읽기도 한다. dict든 문자열이든 payload는 그대로 전달되므로 node가 받는 형태에 맞추기만 하면 된다.)
triage(분류) 노드에서 각 담당 노드로 조건부 엣지를 연결할 수도 있다. 하지만 "다른 node로 넘기면서(transfer) 그 이유(state)도 함께 업데이트" 하고 싶다면 — 이때 쓰는 것이 Command다.
💡
Command는 한 node가 다른 node로 transfer를 시작하면서 동시에 state를 업데이트하게 해 준다. edge로 연결되어 있지 않아도 그래프/노드 어디로든 점프할 수 있고, 자식 그래프에서 부모 그래프로도 이동할 수 있다. edge가 "노드 → 노드" 연결이라면, Command는 일종의 치트키인 셈이다. (그래서 별칭이 HandOff다.)
from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict
from typing import Literal
from langgraph.types import Command
class State(TypedDict):
transfer_reason: str
graph_builder = StateGraph(State)
# 반환 타입 힌트로 goto 대상 노드를 명시하면 그래프 시각화·검증에 도움이 된다
def triage_node(state: State) -> Command[Literal["account_support"]]:
return Command(
goto="account_support", # 원하는 노드로 goto
update={ # 이동과 동시에 state 업데이트
"transfer_reason": "The user wants to change password"
},
)
def tech_support(state: State):
return {}
def account_support(state: State):
return {}
graph_builder.add_node("triage_node", triage_node)
graph_builder.add_node("tech_support", tech_support)
graph_builder.add_node("account_support", account_support)
graph_builder.add_edge(START, "triage_node")
graph_builder.add_edge("tech_support", END)
graph_builder.add_edge("account_support", END)
graph = graph_builder.compile()
graph.invoke({})✅ 조건부 엣지 vs. Command의 차이
조건부 엣지: node가 실행을 끝낸 뒤, 프레임워크가 라우팅 함수를 호출해 어디로 갈지 정한다.
Command: node 스스로 다른 node로 transfer를 시작하고, 그 과정에서 state까지 업데이트한다.
🧠 한눈에 정리 — LangGraph의 핵심은 State(데이터) · Node(함수) · Edge(연결) 세 가지다. 여기에 Multiple Schemas(스키마 분리), Reducer(덮어쓸까 더할까), Node Caching(결과 재사용), Conditional Edges(갈림길), Send API(동적·병렬 실행), Command(이동+업데이트)를 얹으면, 복잡한 에이전트도 하나의 그래프로 설계할 수 있다.