LangGraph 예제, 노드의 다양한 역할과 분기 조건의 설정

from typing import Annotated

from typing_extensions import TypedDict
from langgraph.graph.message import add_messages

class State(TypedDict):
    """
    {
      "input":"Hello",
      "result":5
    }
    """
    input: str    # 사용자 input data
    result: int   # Node의 result data

def len_str(state:State) -> State:
    # hello 
    input = state["input"]
    # 5
    result = len(input)
    return {
        "result": result
    }

def add_one(state:State) -> State:
    # 5
    input = state["result"]
    # 6
    result = input + 1
    return {
        "result": result
    }

from langgraph.graph import StateGraph

# graph 객체 생성(선언)
# simple_graph 상 데이터 전달에 사용할 객체 -> State 클래스 
simple_graph = StateGraph(State)

simple_graph.add_node(
    "len_str", len_str
)

simple_graph.add_node(
    "add_one", add_one
)

from langgraph.graph import START, END 

simple_graph.add_edge(
    START, "len_str"
)

simple_graph.add_edge(
    "len_str", "add_one"
)

simple_graph.add_edge(
    "add_one", END
)

compiled_graph = simple_graph.compile()

for chunk in compiled_graph.stream(
    # Start Node에게 전달할 Input data
    {
      "input":"Hello World"
    }
):
    print(chunk)

StateGraph 를 이용해 문자열의 길이를 계산, 그 결과에 1을 더하는 작업의 수행

다시,

클래스 정의 부분

# TypedDict 의 클래스 생성
class State(TypedDict):
	input: str
    result: int

State 는 입력 데이터와 결과 데이터를 담을 딕셔너리 구조

input: str, result: int

 

특정 함수의 정의, State Class 를 입력으로 받음

def len_str(state: State) -> State:
	input = state["input"]
    result = len(input)
    
    return{
    	"result": result
 	}
    
def add_one(state: State) -> State:
	input = state["result"]
    result = result + 1
    
    return{
    	"result": result
    }

이들은 모두 노드로 사용되는 함수들, State 의 입출력을 받음!!!

 

StateGraph 생성 및 노드 추가

from langgraph.graph import StateGraph

simple_graph = StateGraph(State)

simple_graph.add_node("len_str", len_str)
simple_graph.add_node("add_one", add_one)

그래프를 생성, 각 노드의 추가

from langgraph.graph import START, END

simple_graph.add_edge(START, "len_str")
simple_graph.add_edge("len_str", "add_one")
simple_graph.add_edge("add_one", END)

 

그래프 컴파일 및 실행

compiled_graph = simple_graph.compile()

for chunk in compiled_graph.stream({"input": "Hello World"}):
    print(chunk)

그래프 컴파일을 통한 실행 준비 마치기

각 단계에서 반환된 chunk (dict) 결과를 순차적으로 출력

 

이렇게 간단하게 구현해봤음.

 

조건 분기가 추가된 예제

from typing_extensions import TypedDict, Optional

class State(TypedDict):
    input:  Optional[str] = None
    node_ouput: Optional[int] = None 
    is_stop: Optional[bool] = False



def len_str(state:State) -> State:
    input = state["input"]
    result = len(input)
    return {
        **state, # state에 저장된 데이터를 다음 Node 전달 
        "node_ouput": result
    }

def add_one(state:State) -> State:
    input = state["node_ouput"]
    is_stop = state["is_stop"]
    
    result = input + 1
    if result > 10:
        is_stop = True

    return {
        **state, # state에 저장된 데이터를 다음 Node 전달 
        "node_ouput": result,
        "is_stop": is_stop
    }

def add_two(state:State) -> State:
    input = state["node_ouput"]
    result = input + 2
    return {
        **state, # state에 저장된 데이터를 다음 Node 전달 
        "node_ouput": result
    }

from langgraph.graph import StateGraph

# graph 객체 생성(선언)
# simple_graph 상 데이터 전달에 사용할 객체 -> State 클래스 
simple_graph = StateGraph(State)

simple_graph.add_node(
    "len_str", len_str
)

simple_graph.add_node(
    "add_one", add_one
)

simple_graph.add_node(
    "add_two", add_two
)

from langgraph.graph import START, END 

simple_graph.add_edge(
    START , "len_str"
)

simple_graph.add_edge(
    "len_str", "add_one"
)

def is_stop(state: State) -> str:
    is_stop = state["is_stop"]
    if is_stop:
        return "go_stop" 
    else:
        return "go_to_add_two_fnc"  
    
simple_graph.add_conditional_edges(
    "add_one",
    is_stop,
    {
        "go_to_add_two_fnc":"add_two",
        "go_stop":END
    }
)

simple_graph.add_edge(
    "add_two", "add_one"
)

compiled_graph = simple_graph.compile()

from IPython.display import Image, display

try:
    display(
        Image(
            compiled_graph.get_graph().draw_mermaid_png()
        )
    )
except:
    pass 

for chunk in compiled_graph.stream(
    # Start Node에게 전달할 Input data
    {
      "input":"Hello", "is_stop":False
    }
):
    print(chunk)

State 정의

from typing_extensions import TypedDict, Optional

class State(TypedDict):
    input:  Optional[str] = None
    node_ouput: Optional[int] = None 
    is_stop: Optional[bool] = False

이전과는 다른 새로운 필드의 추가,

input, node_output, is_stop

 

def len_str(state: State) -> State:
	input = state["input"]
    result = len(input)
    return {
    	**state,
        "node_output": result
    }

def add_one(state:State) -> State:
    input = state["node_ouput"]
    is_stop = state["is_stop"]
    
    result = input + 1
    if result > 10:
        is_stop = True

    return {
        **state, # state에 저장된 데이터를 다음 Node 전달 
        "node_ouput": result,
        "is_stop": is_stop
    }
    
def add_two(state:State) -> State:
    input = state["node_ouput"]
    result = input + 2
    return {
        **state, # state에 저장된 데이터를 다음 Node 전달 
        "node_ouput": result
    }

문자열 길이를 계산하여 node_output 에 저장, 

기존 state 의 데이터를 유지하며 새로운 node_output 값을 반환한다. 

 

simple_graph.add_node(
    "len_str", len_str
)

simple_graph.add_node(
    "add_one", add_one
)

simple_graph.add_node(
    "add_two", add_two
)

노드 추가

 

from langgraph.graph import START, END 

simple_graph.add_edge(
    START , "len_str"
)

simple_graph.add_edge(
    "len_str", "add_one"
)

초기 엣지 추가

 

조건부 엣지 설정

def is_stop(state: State) -> str:
    is_stop = state["is_stop"]
    if is_stop:
        return "go_stop" 
    else:
        return "go_to_add_two_fnc"  
    
simple_graph.add_conditional_edges(
    "add_one",
    is_stop,
    {
        "go_to_add_two_fnc": "add_two",
        "go_stop": END
    }
)

simple_graph.add_edge(
    "add_two", "add_one"
)

is_stop 값을 확인하여 문자열의 반환

go_stop, go_to_add_two_fnc 문자열 반환

조건부 엣지 설정으로 add_one 이 완료되면 is_stop 의 상태에 따라 "add_two" 또는 END 로 연결된다. 

compiled_graph = simple_graph.compile()

for chunk in compiled_graph.stream({"input": "Hello", "is_stop": False}):
    print(chunk)

그래프 컴파일 및 실행

초기 설정값의 지정, 각 단계의 chunk 결과가 출력된다.