[LangGraph] Node 노드의 개념과 사용법

노드 Node

1. 노드의 개념

• 특정한 작업을 수행하는 “작업 함수” 역할, 기본 실행 단위
• 그래프의 상태(State)를 입력으로 받아 지정된 작업을 수행한 뒤, 상태의 변경 부분을 반환한다.
• 상태 외로, 설정(RunnableConfig)과 런타임(Runtime) 객체도 입력으로 받을 수 있다.
• 동기 노드, 비동기 노드가 존재한다.

2. 입력 파라미터

• state : 그래프의 상태
• config : 구성 정보 등을 포함하는 RunnableConfig 객체
• runtime : 런타임 컨텍스트와 기타 정보를 포함하는 Runtime 객체

3. 노드의 기본 구조

(1) 노드 선언과 정의

• 기본적으로 함수의 형태를 가진다.
• State(필수), Runtime, RunnableConfig 세 가지 메타데이터를 받을 수 있다.

• 입력으로 State를 받고, 출력으로 State의 업데이트 된 부분을 반환한다.

• 기본적인 노드

def plain_node(state: State):
    result = some_working()
    return {"work_result":result}

• 런타임을 포함하는 노드

def node_with_runtime(state: State, runtime: Runtime[Context]):
    print("In node: ", runtime.context.user_id)
    return {"results": f"Hello, {state['input']}!"}

• 메타데이터를 포함하는 노드

def inspect_metadata(state: dict, config: RunnableConfig) -> dict:
    metadata = config["metadata"]
    print(f"Step: {metadata['langgraph_step']}")
    print(f"Node: {metadata['langgraph_node']}")
    print(f"Triggers: {metadata['langgraph_triggers']}")
    return state

(2) 그래프에 노드 추가

• 그래프의 add_node 메서드를 사용해 그래프에 노드를 추가한다.
• add_node 는 (1) 노드 별칭 (2) 노드 함수 두 가지 파라미터를 받는다.
• 노드 별칭을 지정하지 않으면, 함수 이름이 노드 별칭으로 지정된다.

builder = StateGraph(State)
...
builder.add_node("plain_node", plain_node)
builder.add_node("node_with_runtime", node_with_runtime)
builder.add_node("node_with_execution_info", node_with_execution_info)

(3) 시작 노드와 종료 노드 지정

• add_edge 메서드와 START 또는 END 문자열 객체로 시작노드와 종료노드를 지정한다.
• START : 그래프의 시작을 알리는 가상 노드
• END : 그래프의 종료를 알리는 가상 노드

graph.add_edge(START, "node_a")
graph.add_edge("node_f", END)

• 참고로 아래 방식으로도 시작 노드와 종료 노드를 지정할 수 있다.
• set_entry_point 메서드로 시작 노드를 지정하며
• set_finish_point 메서드로 종료 노드를 지정한다.

graph.set_entry_point("node_a")
graph.set_finish_point("node_f")

4. 예시 코드

#### (1) 전체 코드

```python
# 출처 : https://docs.langchain.com/oss/python/langgraph/graph-api#nodes
from dataclasses import dataclass
from typing_extensions import TypedDict

from langgraph.graph import StateGraph
from langgraph.runtime import Runtime

class State(TypedDict):
    input: str
    results: str

@dataclass
class Context:
    user_id: str

builder = StateGraph(State)

def plain_node(state: State):
    return state

def node_with_runtime(state: State, runtime: Runtime[Context]):
    print("In node: ", runtime.context.user_id)
    return {"results": f"Hello, {state['input']}!"}

def node_with_execution_info(state: State, runtime: Runtime):
    print("In node with thread_id: ", runtime.execution_info.thread_id)
    return {"results": f"Hello, {state['input']}!"}

def inspect_metadata(state: dict, config: RunnableConfig) -> dict:
    metadata = config["metadata"]
    print(f"Step: {metadata['langgraph_step']}")
    print(f"Node: {metadata['langgraph_node']}")
    print(f"Triggers: {metadata['langgraph_triggers']}")
    return state

builder.add_node("plain_node", plain_node)
builder.add_node("node_with_runtime", node_with_runtime)
builder.add_node("node_with_execution_info", node_with_execution_info)
...

```

Reference

Graph API overview - Docs by LangChain