[LangGraph] Edge 엣지의 개념과 종류

1. 엣지 Edge

(1) 엣지의 개념

• 엣지는 로직(애플리케이션의 흐름)이 어떻게 라우팅되는지, 그리고 그래프의 종료를 어떻게 결정할지 정의한다.
• 라우팅이란, 입력 또는 현재 상태(State)에 따라 “다음에 어떤 노드로 보낼지”를 정하는 것을 의미한다.
• 그래프의 구성 요소 관점에서 엣지는 하나의 노드와 다른 노드간의 연결을 의미한다.

(2) 엣지의 종류

엣지의 종류	설명
일반 엣지 Normal Edge	현재 노드에서 정해진 다음 노드로 바로 이동하는 경우
조건부 엣지 Conditional Edge	조건을 판단하는 함수를 호출한 뒤, 그 결과에 따라 이동할 다음 노드를 정하는 경우
진입점 Entry Point	사용자 입력이 들어왔을 때 가장 먼저 실행될 노드를 지정
조건부 진입점 Conditional Entry Point	사용자 입력이 들어왔을 때, 어떤 노드를 가장 먼저 실행할지 결정

• 각 엣지에 대해서는 아래에서 자세히 살펴본다.

(3) 병렬 실행 엣지

• 하나의 노드에 여러 개의 다음 경로(엣지)가 연결될 수 있다. 이 경우, 다음 노드들은 Super Step에서 병렬로 함께 실행된다.
• 이는 추후 심화 포스팅에서 살펴본다.

2. 엣지 Edge의 종류

(1) Normal edges

• 하나의 노드에서 작업이 완료된 뒤, 항상 같은 노드로 이동하는 엣지
• add_edge(노드, 다음노드) 메서드를 이용하여 그래프에 엣지를 추가한다.
• 별도의 판단 조건 없이 A 작업이 끝나면 항상 B 작업이 수행되는 가장 기본적인 연결 방식

graph.add_edge("node_a", "node_b")

(2) Conditional edges

• (1) 상황에 따라 실행할 다음 노드를 결정하는 경우
• (2) 다음에 여러 노드를 병렬로 동시에 실행하는 경우
• (3) 그래프를 종료시키고 싶은 경우
• add_conditional_edge(노드, 라우팅함수) 메서드를 이용하여 그래프에 엣지를 추가한다.
• 라우팅 함수(routing function)는 현재 그래프 상태를 입력으로 받아, 다음에 실행할 노드를 나타내는 값을 함수이다.

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
import random

class AppState(TypedDict):
    input:int
    value:int
    count:int
    response:str

def generate_node(state:AppState):
    return {"value" : random.randint(1, 100), "count" : state["count"] + 1}

def routing_function(state:AppState):
    if state["input"] > state["value"]:
        return "terminate"
    else:
        return "generate"

def terminate_node(state:AppState):
    return {"response" : f"{state['count']} 번 반복 실행됐습니다."}

graph = StateGraph(AppState)
graph.add_node("generate", generate_node)
graph.add_node("terminate", terminate_node)

graph.add_edge(START, "generate")
graph.add_conditional_edges("generate", routing_function) # !! 여기 !!
graph.add_edge("terminate", END)

app = graph.compile()
app.invoke({"input":5, "count":0})

{'input': 5, 'value': 4, 'count': 26, 'response': '26 번 반복 실행됐습니다.'}

• 라우팅 함수의 반환값과 이에 따른 다음에 실행할 노드를 dictionary 형태로 직접 지정해줄 수도 있다.

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
import random

class AppState(TypedDict):
    input:int
    value:int
    count:int
    response:str

def generate_node(state:AppState):
    return {"value" : random.randint(1, 100), "count" : state["count"] + 1}

def routing_function(state:AppState):
    return state["input"] > state["value"]

def terminate_node(state:AppState):
    return {"response" : f"{state['count']} 번 반복 실행됐습니다."}

graph = StateGraph(AppState)
graph.add_node("generate", generate_node)
graph.add_node("terminate", terminate_node)

graph.add_edge(START, "generate")
graph.add_conditional_edges("generate", routing_function,
                           {True:"terminate", False:"generate"}) # !! 여기 !!
graph.add_edge("terminate", END)

app = graph.compile()
app.invoke({"input":5, "count":0})

{'input': 5, 'value': 4, 'count': 26, 'response': '26 번 반복 실행됐습니다.'}

• 그래프를 시각화하면 아래와 같다.

• 그 외 병렬처리 방법 등은 추후 심화 포스팅에서 다룬다.

(3) Entry Point 진입점

• 그래프가 시작될 때 가장 먼저 실행되는 노드를 정하는 것
• 가상의 노드인 langgraph.graph.START 노드에서 첫 노드로 엣지를 연결해줌으로써 진입점을 지정할 수 있다.

from langgraph.graph import StateGraph, START, END

graph = StateGraph(AppState)
graph.add_node("generate", generate_node)

graph.add_edge(START, "generate")

(4) Conditional Entry Point 조건부 진입점

• 시작 지점에서 조건에 따라 처음 실행될 노드를 정할 수도 있다.
• add_conditional_edges 와 START 노드, 그리고 라우팅 함수를 사용해 구현할 수 있다.

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
import random

class AppState(TypedDict):
    input:int
    value:int
    count:int
    response:str

def input_routing_function(state:AppState):
    return state["input"] < 0

def generate_node(state:AppState):
    return {"value" : random.randint(1, 100), "count" : state["count"] + 1}

def routing_function(state:AppState):
    return state["input"] > state["value"]

def terminate_node(state:AppState):
    if "value" not in state.keys():
        return {"response" : f"사용자가 입력한 {state['input']}은(는) 0보다 작습니다."}
    return {"response" : f"{state['count']} 번 반복 실행됐습니다."}

graph = StateGraph(AppState)
graph.add_node("generate", generate_node)
graph.add_node("terminate", terminate_node)

graph.add_conditional_edges(START, input_routing_function, {True:"terminate", False:"generate"})
graph.add_conditional_edges("generate", routing_function, {True:"terminate", False:"generate"})
graph.add_edge("terminate", END)

app = graph.compile()
app.invoke({"input":-30, "count":0})

{'input': -30, 'count': 0, 'response': '사용자가 입력한 -30은(는) 0보다 작습니다.'}

• 그래프를 시각화하면 아래와 같다.

Reference

Graph API overview - Docs by LangChain