Jongya

Jonghyuk Park

1. State 상태 업데이트 메커니즘

(1) State 의 channel (채널)

  • State 안의 “각각의 키 하나 하나를 그래프 런타임(Pregel)이 관리하는 저장 슬롯” 이다.
  • 쉽게 말해, State 안에 있는 각각의 필드가 하나 하나의 채널이다.
  • 각 채널은 “현재 값” 과 “업데이트 규칙”을 가진다.
  • 별도의 업데이트 규칙을 정해주지 않으면, 기본적으러 덮어쓰기를 수행한다.
1
2
3
4

class GraphState(TypedDict):
    user_input: str                  # user_input 채널
    intent: str                      # intent 채널
    logs: Annotated[list[str], add]  # result 채널

(2) State 업데이트 예시

  • 예시를 들기 전, node 에 대해 잠깐 설명해야 한다.
  • node는, LangGraph에서 특정한 작업을 수행하는 단위, 함수에 해당한다.
  • 또한 node는 State를 입력으로 받아, 업데이트한 뒤 State를 반환한다.
  • 사용자의 발화에서 의도를 추출하고 답변을 생성하는 간단한 애플리케이션을 예로 들어본다.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

from typing import TypedDict, Annotated
from operator import add
from langgraph.graph import StateGraph

class GraphState(TypedDict):
    user_input: str
    intent: str
    result: str

# 사용자의 의도를 판단해, intent 채널에 업데이트
def classify_node(state: GraphState) -> dict:
    if "환불" in state["user_input"]:
        return {"intent": "refund"}
    elif "구매" in state["user_input"]:
        return {"intent": "purchase"}
    return {"intent": "general"}

# 사용자 의도에 따라 result 채널에 업데이트
def response_node(state: GraphState) -> dict:
    if state["intent"] == "refund":
        return {"result": "환불 절차를 안내합니다."}
    elif state["intent"] == "purchase":
        return {"result": "구메 절차를 안내합니다."}
    return {"result": "일반 문의로 처리합니다."}
  • 알아둘 점은, 노드는 전체 State를 수정하는 게 아니라, 업데이트 되는 채널만 반환하며
  • 반환된 값을 참고하여 랭그래프의 런타임이 그 채널의 업데이트 규칙에 따라 기존 값을 수정한다는 것이다.
  • 위 랭그래프에 대해 환불받고 싶어요 라는 user_input이 들어왔다고 해보자.
1
2
3
4
5

{
    "user_input": "환불받고 싶어요",
    "intent": "",
    "result": ""
}
  • classify_node 노드 실행 후에는 State가 아래와 같이 업데이트 된다.
1
2
3
4
5

{
    "user_input": "환불받고 싶어요",
    "intent": "refund",
    "result": ""
}
  • 그 다음 response_node 가 실행되면 State 는 아래와 같이 바뀔 것이다.
1
2
3
4
5

{
    "user_input": "환불받고 싶어요",
    "intent": "refund",
    "result": "환불 절차를 안내합니다."
}

(3) State 상태 업데이트 메커니즘

  • 상태 업데이트 메커니즘에 대해 간단히 정리해보면 아래와 같다.
  • (1) 그래프가 현재 State 를 노드에 전달한다.
  • (2) 노드는 State를 읽고, 작업을 수행한 뒤, 변경분(채널)만 dict 형태로 반환한다.
  • (3) LangGraph 런타임이 반환값을 받아 채널의 업데이트 규칙에 맞게 State에 반영한다.
  • (4) 다음 노드는 업데이트된 State를 받아 작업을 수행한다.


2. 리듀서 Reducer

(1) 채널의 업데이트 규칙

  • 앞서 소개했듯, State 의 각 채널은 “현재 값”과 함께 ”업데이트 규칙”을 가진다.
  • 그렇다면 채널의 업데이트 규칙에는 어떤 종류들이 있을까?
  • 기본값은 덮어쓰기고, “리스트에 추가”하는 규칙도 자주 사용된다.
  • 그리고 자유롭게 업데이트 규칙을 정할 수도 있다.
업데이트 규칙 설명
덮어쓰기 • 채널의 현재 값을 덮어쓴다.
• 기본적인 업데이트 규칙
• 별도 규칙을 정하지 않으면 덮어쓰기가 적용됨
리스트에 추가 • 채널이 list[] 자료형인 경우
• 현재값(리스트)의 요소를 추가하는 방식으로 업데이트됨
• Message 또는 Log 등에 자주 사용됨
사용자 정의 • 그 외로도 다양한 방식의 업데이트 규칙이 가능하다.
• 예를 들어 int 형 채널에 대해 값을 1씩 늘린다던가 하는 규칙 등

(2) 리듀서(Reducer)의 정의

  • State의 특정 키(채널)에 새 값이 들어왔을 때, 기존 값과 새 값을 어떻게 합칠지 정의하는 함수
  • 각 키(채널)들은 자신만의 reducer를 가질 수 있다.
  • reducer를 지정하지 않은 경우, 기본 업데이트 규칙은 덮어쓰기.

(3) 리듀서의 예시

  • 아래는 삼세판 가위바위보로 승자를 가리는 랭그래프다.
  • (1) count : 가위바위보 횟수
  • (2) win_logs : 각 판의 승리자를 누적한 리스트
  • (3) winner : 최종 승리자
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

from typing import Annotated
from operator import add
from typing_extensions import TypedDict
from collections import Counter
from langgraph.graph import StateGraph

class State(TypedDict):
    count: Annotated[int, add]
    win_logs: Annotated[list[str], add]
    winner: str

def first_round_node(state: State) -> State:
    return {
        "count": 1,
        "win_logs": ["철수"]
    }

def second_round_node(state: State) -> State:
    return {
        "count": 1,
        "win_logs": ["민수"]
    }

def third_round_node(state: State) -> State:
    return {
        "count": 1,
        "win_logs": ["철수"]
    }

def judge_node(state: State) -> State:
    c = Counter(state["win_logs"])
    return {
        "winner": c.most_common()[0][0]
    }
    
builder = StateGraph(State)

builder.add_node("first", first_round_node)
builder.add_node("second", second_round_node)
builder.add_node("third", third_round_node)
builder.add_node("judgement", judge_node)

builder.set_entry_point("first")
builder.add_edge("first", "second")
builder.add_edge("second", "third")
builder.add_edge("third", "judgement")

app = builder.compile()

state = State({"count":0})
app.invoke(state)
  • 이 그래프는 첫 노드를 거치면, 상태가 아래와 같이 업데이트된다.
1

{"count":1, "win_logs":["철수"]}
  • 두 번째 판에서는 민수가 이겼고, 상태는 아래와 같이 업데이트된다.
1

{"count":2, "win_logs":["철수", "민수"]}
  • 마지막 판에서는 철수가 이겼고, 상태는 아래와 같이 업데이트된다.
1

{"count":3, "win_logs":["철수", "민수", "철수"]}
  • 최종 판결 노드를 거친 최종 상태는 아래와 같다.
1

{"count":3, "win_logs":["철수", "민수", "철수"], winner:"철수"}

(4) 예시 작동 방식 살펴보기

  • count 와 win_logs 두 채널에서 operator.add 라는 함수를 리듀서로 사용했다.
  • 이 함수는 더하기를 수행하는 함수로, + 와 같은 작동방식을 가진다.
  • 동일한 함수임에도count 와 win_logs 각 채널에서의 작동방식이 다름을 볼 수 있다. 이는 자료형의 차이 때문.
  • count 는 int 형이기 때문에 add 를 하면 두 숫자가 더해진 결과가 반환된다.
  • win_logs 는 list 형이기 때문에 add 가 extend 와 같은 동작을 수행한다.
  • 리듀서의 작동 방식을 함수 선언문으로 표현해보면 아래와 같다.
1

def reducer(a:SOMETYPE, b:SOMETYPE) -> SOMETYPE

(5) 주의할 사항

  • (1) 업데이트 되는 자료형과, 리듀서가 리턴하는 자료형이 같아야 한다.
  • (2) list 형태인 채널에 대해서는 list 로 리턴해야 한다.
  • 이 정도를 주의하면 좋을 것이다.
  • 추후 추가

3. messages 에 대한 리듀서

(1) messages

  • LangGraph 에서는 대화형 애플리케이션을 만들 때 messages 라는 채널을 자주 사용한다.
  • messages 채널은 사용자 메시지, AI 응답 메시지, 시스템 메시지, 툴 호출 메시지 등을 순서대로 저장하는 리스트다.
  • 즉, 챗봇이나 에이전트가 지금까지 어떤 대화를 주고받았는지 저장하는 대화 기록 역할을 한다.

(2) add_messages 리듀서

1
2
3
4
5
6
7

from typing import Annotated
from typing_extensions import TypedDict
from langchain_core.messages import AnyMessage
from langgraph.graph.message import add_messages

class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]
  • 여기서 messages 채널에는 add_messages 라는 리듀서를 적용했다.
  • 앞서 operator.add 를 사용하면 리스트가 단순히 이어붙여졌다면, add_messages 는 메시지 목록을 다루기 위해 LangGraph 에서 제공하는 전용 리듀서다.
  • 기본적인 역할은 기존 messages 리스트에 새로 들어온 메시지를 추가하는 것이다.

(3) add_message 리듀서 작동방식

예를 들어 현재 State 가 아래와 같다고 해보자.

1
2
3
4
5

{
    "messages": [
        HumanMessage(content="안녕하세요")
    ]
}

이후 어떤 노드가 아래와 같이 AI 응답 메시지만 반환한다고 해보자.

1
2
3
4
5
6

def chatbot_node(state: State) -> dict:
    return {
        "messages": [
            AIMessage(content="안녕하세요. 무엇을 도와드릴까요?")
        ]
    }

그러면 LangGraph 런타임은 messages 채널의 리듀서인 add_messages 를 사용해 기존 메시지 목록과 새 메시지를 합치고, 최종 State 는 아래와 같이 된다.

1
2
3
4
5
6

{
    "messages": [
        HumanMessage(content="안녕하세요"),
        AIMessage(content="안녕하세요. 무엇을 도와드릴까요?")
    ]
}

(4) add_message만의 장점

  • add_messages 는 단순히 리스트를 더하는 것과 비슷해 보이지만, 메시지 객체를 다룰 때 더 적합하다.
  • HumanMessage, AIMessage, SystemMessage, ToolMessage 같은 LangChain 메시지 객체들을 대화 흐름에 맞게 누적하는 데 용이하다.
  • 또한 입력으로 메시지 객체뿐 아니라 딕셔너리 형태의 메시지가 들어와도 LangChain 메시지 객체로 역직렬화해서 다룰 수 있다.

예를 들어 아래 두 방식 모두 사용 가능하다.

1
2
3
4
5

{
    "messages": [
        HumanMessage(content="안녕하세요")
    ]
}

1
2
3
4
5

{
    "messages": [
        {"role": "user", "content": "안녕하세요"}
    ]
}

(5) message 내용을 꺼낼 때

1
2

last_message = state["messages"][-1]
content = last_message.content

(6) MessageState

  • MessageState란, LangGraph에서 제공하는 State의 한 종류
  • 내부적으로 messages 채널을 가지고 있고, 이 채널에 add_messages 리듀서가 적용되어 있다.
  • 대화형 그래프에서는 messages 채널을 정의하는 일이 많기 때문이, 이를 준비해둔 템플릿이라고 보면 된다.
1
2
3
4
5

from langgraph.graph import MessagesState

class State(MessagesState):
    user_id: str
    summary: str
  • 위 코드는 아래와 비슷한 구조라고 이해하면 된다.
1
2
3
4
5
6
7
8

from typing import TypedDict
from langchain_core.messages import AnyMessage
from langgraph.graph.message import add_messages

class MessagesState(TypedDict):
    user_id: str
    summary: str
    messages: Annotated[list[AnyMessage], add_messages]

Reference

https://docs.langchain.com/oss/python/langgraph/graph-api

Do it! LLM을 활용한 AI 에이전트 개발 입문 (이성용 저)

https://wikidocs.net/261579

https://www.youtube.com/watch?v=W_uwR_yx4-c

Comments

You May Also Enjoy