[LangGraph] 그래프의 생성, 컴파일(연결), 실행 (with invoke/stream)

그래프 Graph

1. 그래프의 개념

• 그래프란 Node(노드)와 Edge(엣지)로 이루어진 작업 흐름도를 뜻한다.
• 개념적으로는 “어떤 작업을 어떤 순서와 조건으로 실행할지 정의한 구조”라고 할 수 있다.
• node : 작업을 실제로 수행하는 함수
• edge : 다음에 어떤 노드로 이동할지 연결하는 규칙
• graph : node 와 edge로 이루어진 일련의 작업 흐름을 하나로 묶은 구조

# 가장 기본적인 상태그래프  

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

class AppState(TypedDict):
    value:int

...

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)

2. 컴파일 Compile

• 앞서 살펴본 그래프(graph)는 작업의 “구조도”일 뿐이므로, 실행할 수 없다.
• 따라서 그래프에 정의된 작업을 실행하기 위해서는 그래프 객체를 컴파일(.compile())을 하여 실행 가능한 객체를 만들어야 한다.
• 컴파일 단계에서는 그래프 구조에 기본적인 문제가 없는지 검사한다. (어디에도 연결되지 않은 노드가 있는지.. 등)
• 또한 체크포인터(checkpointer)나 브레이크포인트(breakpoint)처럼 실행시 사용할 설정도 이 단계에서 지정할 수 있다.

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

class AppState(TypedDict):
    value:int

...

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 # --> 컴파일된 실행 가능한 객체

3. 컴파일된 그래프 실행 방식

• invoke() : 한 번 실행하고 최종 상태를 한 번 반환받는 실행 방식.
• stream() : 실행 중간 과정을 단계별로 보면서 반환받는 실행 방식.
• 최종 결과만 필요할 경우 → invoke / 중간 흐름도 보고 싶은 경우 → stream
• 이외로 ainvoke() 와 astream() 이라는 비동기 실행 방법도 있으며, 이는 추후 심화 포스팅에서 다룬다.

(1) invoke

# invoke
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":10, "count":0}) # invoke

{'input': 10, 'value': 3, 'count': 7, 'response': '7 번 반복 실행됐습니다.'}

(2) stream

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()
for step in app.stream({"input":10, "count":0}): # stream
    print(step)

{'generate': {'value': 27, 'count': 1}}
{'generate': {'value': 32, 'count': 2}}
{'generate': {'value': 73, 'count': 3}}
{'generate': {'value': 28, 'count': 4}}
{'generate': {'value': 93, 'count': 5}}
{'generate': {'value': 32, 'count': 6}}
{'generate': {'value': 2, 'count': 7}}
{'terminate': {'response': '7 번 반복 실행됐습니다.'}}

Reference

Graph API overview - Docs by LangChain