[강의노트] LangChain Academy : Introduction to LangGraph (Module 2)

원본 게시글: https://velog.io/@euisuk-chung/LangChain-Academy-Introduction-to-LangGraph-Module-2

랭체인(LangChain)과 랭그래프(LangGraph)는 대규모 언어 모델(LLM)을 활용한 애플리케이션 개발을 위한 도구들입니다. 위 강의는 LangChain에서 운영하는 LangChain Academy에서 제작한 “Introduction to LangGraph” 강의의 내용을 정리 및 추가 설명한 내용입니다.

• 강의 링크 : https://youtu.be/29XE10U6ooc
• 랭체인 : https://www.langchain.com/

이번 포스트는 “Module2”내용을 다룹니다:

목차

• Lesson 1: State Schema
• Lesson 2: State Reducers
• Lesson 3: Multiple Schemas
• Lesson 4: Trim and Filter Messages
• Lesson 5: Chatbot w/ Summarizing Messages and Memory
• Lesson 6: Chatbot w/ Summarizing Messages and External Memory

Lesson 1: State Schema

LangGraph의 주요 개념과 구성 요소

1. 그래프 구조:

   • LangGraph는 에이전트 워크플로우를 그래프로 모델링합니다.
   • 주요 구성 요소: State(상태), Nodes(노드), Edges(엣지)

2. StateGraph와 MessageGraph:

   • StateGraph: 사용자 정의 State 객체로 매개변수화된 주요 그래프 클래스
   • MessageGraph: State가 메시지 리스트로만 구성된 특수한 그래프 유형

3. 상태 관리:

   • TypedDict나 Pydantic 모델을 사용하여 상태 스키마 정의
   • Reducer 함수를 통한 상태 업데이트 방식 정의
   • 다중 스키마 지원: 내부 노드 통신, 입출력 스키마 분리 등

4. 노드와 엣지:

   • 노드: 그래프의 로직을 구현하는 Python 함수
   • 엣지: 노드 간 연결과 실행 흐름을 정의
   • 조건부 엣지를 통한 동적 라우팅

5. 메시지 처리:

   • add_messages 함수를 사용한 효율적인 메시지 관리
   • MessagesState를 통한 간편한 메시지 상태 관리

6. 그래프 실행:

   • 컴파일 과정을 통한 그래프 구조 검증
   • “super-steps”를 통한 이산적 실행 방식

7. 유연성과 확장성:

   • 다양한 상태 스키마 및 업데이트 방식 지원
   • 복잡한 워크플로우 구현 가능

State Schema는 Langraph에서 에이전트가 처리하는 데이터의 구조와 타입을 정의하는 방식입니다. 이 스키마는 데이터의 형식을 지정하여, 그래프 내에서 데이터가 어떻게 처리되고 업데이트되는지를 제어합니다.

• Langraph에서는 Typedict, Python Data Classes, Pydantic 등 다양한 방법으로 스키마를 정의할 수 있습니다. 각각의 방법에 대해서 살펴보겠습니다.

  1. Typedict: 주로 사용되는 방식으로, Python의 딕셔너리 형식을 기반으로 하여 키(key)와 해당 값의 타입을 지정합니다.
     • 하지만 Typedict는 런타임에서 타입 검사를 하지 않아, 잘못된 값이 할당되어도 오류가 발생하지 않는 문제가 있습니다.

  

  • 앞에 Module 1에서 나왔던 예시로 아래와 같은 예시가 있었습니다:

    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
    # Module 1 - TypeDict예시
    import random
    from IPython.display import Image, display
    from langgraph.graph import StateGraph, START, END
        
    def node_1(state):
        print("---Node 1---")
        return {"name": state['name'] + " is ... "}
        
    def node_2(state):
        print("---Node 2---")
        return {"mood": "happy"}
        
    def node_3(state):
        print("---Node 3---")
        return {"mood": "sad"}
        
    def decide_mood(state) -> Literal["node_2", "node_3"]:
        
        # Here, let's just do a 50 / 50 split between nodes 2, 3
        if random.random() < 0.5:
        
            # 50% of the time, we return Node 2
            return "node_2"
        
        # 50% of the time, we return Node 3
        return "node_3"
        
    # Build graph
    builder = StateGraph(TypedDictState)
    builder.add_node("node_1", node_1)
    builder.add_node("node_2", node_2)
    builder.add_node("node_3", node_3)
        
    # Logic
    builder.add_edge(START, "node_1")
    builder.add_conditional_edges("node_1", decide_mood)
    builder.add_edge("node_2", END)
    builder.add_edge("node_3", END)
        
    # Add
    graph = builder.compile()
        
    # View
    display(Image(graph.get_graph().draw_mermaid_png()))
    

    

    (참고) 일반 클래스

    • 가장 기본적인 객체 지향 프로그래밍의 구조입니다.
    • 속성과 메서드를 자유롭게 정의할 수 있습니다.
    • init() 등의 메서드를 직접 구현해야 합니다.
    • 캡슐화, 상속, 다형성 등의 OOP 개념을 완전히 활용할 수 있습니다.

  (참고) TypedDict

  • Python 3.8부터 정식으로 도입되었습니다

    (이전 버전에서는 typing_extensions 모듈을 통해 사용 가능).

  • 딕셔너리의 키와 값에 대한 타입을 지정할 수 있게 해줍니다.
  • 런타임에는 일반 딕셔너리처럼 동작합니다.
  • 키에 대한 접근은 딕셔너리 문법을 사용합니다 (예: obj["key"]).
  • 주로 정적 타입 검사를 위해 사용됩니다.

  (참고) Dataclass

  • Python 3.7부터 도입된 기능입니다.
  • 데이터를 저장하기 위한 클래스를 쉽게 만들 수 있게 해줍니다.
  • 자동으로 init(), repr(), eq() 등의 메서드를 생성합니다.
  • 타입 힌팅을 지원합니다.
  • 속성에 직접 접근 가능합니다 (예: obj.attribute).
  • 불변성(immutability)을 쉽게 구현할 수 있습니다.

  (참고) TypedDict vs Dataclass

  • 주요 차이점으로는 속성 접근 방식이 다릅니다.

    

    • TypedDict: 딕셔너리 스타일로 접근 (state[“name”])
    • dataclass: 객체 속성 스타일로 접근 (state.name)

  1. Python Data Classes: 데이터 클래스는 더 간결한 구문을 제공하며, 속성에 접근할 때 dict 대신 dot(.) 연산자를 사용할 수 있습니다.
     • 하지만 역시 타입 힌트가 런타임에 강제되지 않아 잘못된 값을 할당해도 오류가 발생하지 않습니다.

  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16
  17
  18
  19
  20
  21
      def node_1(state):
        print("---Node 1---")
        return {"name": state.name + " is ... "}
      
      # Build graph
      builder = StateGraph(DataclassState)
      builder.add_node("node_1", node_1)
      builder.add_node("node_2", node_2)
      builder.add_node("node_3", node_3)
      
      # Logic
      builder.add_edge(START, "node_1")
      builder.add_conditional_edges("node_1", decide_mood)
      builder.add_edge("node_2", END)
      builder.add_edge("node_3", END)
      
      # Add
      graph = builder.compile()
      
      # View
      display(Image(graph.get_graph().draw_mermaid_png()))
  

  1. Pydantic: 런타임에서 데이터 유효성을 검사하는 기능을 제공합니다.
     • Pydantic은 Python의 타입 힌팅을 사용하여 데이터 검증과 설정 관리를 제공하는 라이브러리입니다. LangGraph에서 상태 스키마를 정의할 때 특히 유용합니다.
     • Pydantic 주요 특징
  • 런타임 데이터 검증
  • 복잡한 데이터 구조 지원
  • 자동 문서화
  • 설정 관리
  • JSON 스키마 생성

  • 아래는 실제 Pydantic 클래스를 선언하고 validation을 수행하는 코드입니다:

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    from pydantic import BaseModel, field_validator, ValidationError
    # BaseModel 상속
    class PydanticState(BaseModel):
        name: str
        mood: str # "happy" or "sad" 
        
        @field_validator('mood')
        @classmethod
        def validate_mood(cls, value):
            # Ensure the mood is either "happy" or "sad"
            if value not in ["happy", "sad"]:
                raise ValueError("Each mood must be either 'happy' or 'sad'")
            return value
        
    try:
        state = PydanticState(name="John Doe", mood="mad")
    except ValidationError as e:
        print("Validation Error:", e)
    

    a. BaseModel 상속:

    • PydanticState(BaseModel): PydanticState 클래스는 Pydantic의 BaseModel을 상속받아 정의됩니다.
    • 이를 통해 자동 검증, 직렬화, 역직렬화 기능을 사용할 수 있습니다.

    b. 필드 정의:

    • name과 mood 필드를 str 타입으로 정의합니다.

    c. @field_validator 데코레이터:

    • @field_validator 데코레이터 : Pydantic 라이브러리에서 제공하는 기능입니다. Pydantic은 파이썬에서 데이터 모델을 정의하고 검증하는 데 사용됩니다.
    • @field_validator는 특정 필드의 값이 맞는지 확인하는 로직을 직접 만들 수 있게 해줍니다.
    • 즉, 어떤 필드에 값이 들어왔을 때 그 값이 우리가 원하는 조건에 맞는지 자동으로 검사해주는 기능이에요.
    • 예를 들어, @field_validator('mood')라면, “mood“라는 필드에 값이 들어올 때, 그 값이 “happy”나 “sad” 중 하나인지 확인하고자 하는 목적으로 사용되었다고 보면 됩니다.

    d. @classmethod:

    • 클래스 자체에 대해 작업을 할 수 있게 만들어주는 데코레이터입니다.
    • 일반적으로 메서드를 만들면 인스턴스(객체)를 사용해서 호출해야 하는데, 클래스 메서드는 인스턴스가 아닌 클래스 자체에 연결되어 실행됩니다.
    • Pydantic에서 필드 검증을 할 때, 검증 메서드를 클래스 레벨에서 사용해야 하기 때문에 @classmethod가 필요로하게 됩니다.
    • 즉, 인스턴스를 만들기 전에 필드 값을 확인해야 하기 때문에 클래스 전체에서 그 검증 로직에 접근할 수 있도록 @classmethod로 메서드를 정의하는 것이죠.

    e. 예외 처리:

    • 유효하지 않은 데이터로 인스턴스를 생성하려고 하면 ValidationError가 발생합니다.

    • 아래 예시에서는 “mad”라는 유효하지 않은 mood 값을 사용하여 오류를 발생시킵니다.

      

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# Build graph
builder = StateGraph(PydanticState)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_node("node_3", node_3)

# Logic
builder.add_edge(START, "node_1")
builder.add_conditional_edges("node_1", decide_mood)
builder.add_edge("node_2", END)
builder.add_edge("node_3", END)

# Add
graph = builder.compile()

# View
display(Image(graph.get_graph().draw_mermaid_png()))

Lesson 2: State Reducers

앞에 Module 1에서 설명했던 State Reducer에 대해서 더욱 자세하게 설명합니다. 아래 예시는 TypeDict를 사용할 때 발생할 수 있는 상태 업데이트 충돌 문제입니다.

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
class State(TypedDict):
    foo: int

def node_1(state):
    print("---Node 1---")
    return {"foo": state['foo'] + 1}

def node_2(state):
    print("---Node 2---")
    return {"foo": state['foo'] + 1}

def node_3(state):
    print("---Node 3---")
    return {"foo": state['foo'] + 1}

# Build graph
builder = StateGraph(State)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_node("node_3", node_3)

# Logic
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
builder.add_edge("node_1", "node_3")
builder.add_edge("node_2", END)
builder.add_edge("node_3", END)

# Add
graph = builder.compile()

# View
display(Image(graph.get_graph().draw_mermaid_png()))

위 그림를 해석해보면 node_1은 node_2와 node_3으로 분기됩니다. node_2와 node_3는 병렬로 실행됩니다. 아래 코드를 실행해보면 다음과 같은 에러가 발생합니다.

1
2
3
4
5
from langgraph.errors import InvalidUpdateError
try:
    graph.invoke({"foo" : 1})
except InvalidUpdateError as e:
    print(f"InvalidUpdateError occurred: {e}")

상태 업데이트 문제:

• 각 노드는 ‘foo’ 값을 1씩 증가시키려 합니다.
• node_2와 node_3가 동시에 실행되면서 둘 다 ‘foo’ 값을 업데이트하려 합니다.

충돌 발생:

• 병렬 실행 중 두 노드가 같은 키(‘foo’)를 동시에 업데이트하려 합니다.
• LangGraph는 이런 상황에서 어떤 업데이트를 적용해야 할지 결정할 수 없습니다.

• 이러한 충돌 상황에서 LangGraph는 InvalidUpdateError를 발생시킵니다.

  1
  2
  3
  4
  5
  6
    
  We see a problem! 
  Node 1 branches to nodes 2 and 3.
  Nodes 2 and 3 run in parallel, which means they run in the same step of the graph.
  They both attempt to overwrite the state *within the same step*. 
  This is ambiguous for the graph! Which state should it keep? 
  

• 이러한 충돌을 해결하기 위해서는 상태 업데이트 방식을 명확히 정의해야 합니다.

이때 사용되는 것이 바로 기본 Reducer 함수입니다.

• Reducer는 LangGraph에서 상태(State) 업데이트를 처리하는 함수입니다.

  • 업데이트 명시적으로 정의: 노드에서 반환된 업데이트를 현재 상태에 어떻게 적용할지 정의합니다.
  • 발생할 수 있는 충돌을 해결: 여러 노드에서 동시에 상태를 업데이트할 때 발생할 수 있는 충돌을 해결합니다.

• Python의 typing 모듈에서 제공하는 Annotated를 사용하여 reducer 함수를 지정합니다.

  • Annotated는 Python의 typing 모듈에서 제공하는 특별한 타입 힌트입니다.

    Annotated[Type, metadata1, metadata2, ...]

• Python의 내장 operator 모듈의 add 함수를 reducer로 사용합니다.

  • LangGraph에서 operator.add를 reducer로 사용할 때는 주로 리스트 연결 기능을 활용합니다. 이를 통해 새로운 상태 업데이트를 기존 리스트에 추가할 수 있습니다.

    ➕ operator.add

    • Python의 내장 operator 모듈에서 제공하는 일반적인 덧셈 연산 함수입니다. 하지만 이 함수는 단순히 숫자를 더하는 것 이상의 기능을 합니다:
      1. 숫자 덧셈:
         • 두 숫자를 더합니다. 예: operator.add(1, 2) 는 3을 반환합니다.
      2. 문자열 연결:
         • 두 문자열을 연결합니다. 예: operator.add("Hello", "World") 는 "HelloWorld"를 반환합니다.
      3. 리스트 연결:
         • 두 리스트를 연결합니다. 예: operator.add([1, 2], [3, 4]) 는 [1, 2, 3, 4]를 반환합니다.

    1
         
    

• 그렇다면 이제 기본 Reducer를 사용한 코드로 업데이트를 수행해보도록 하겠습니다:

  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
    
  from operator import add
  from typing import Annotated
    
  class State(TypedDict):
      foo: Annotated[list[int], add]
    
  def node_1(state):
      print("---Node 1---")
      return {"foo": [state['foo'][-1] + 1]}
    
  def node_2(state):
      print("---Node 2---")
      return {"foo": [state['foo'][-1] + 1]}
    
  def node_3(state):
      print("---Node 3---")
      return {"foo": [state['foo'][-1] + 1]}
    
  # Build graph
  builder = StateGraph(State)
  builder.add_node("node_1", node_1)
  builder.add_node("node_2", node_2)
  builder.add_node("node_3", node_3)
    
  # Logic
  builder.add_edge(START, "node_1")
  builder.add_edge("node_1", "node_2")
  builder.add_edge("node_1", "node_3")
  builder.add_edge("node_2", END)
  builder.add_edge("node_3", END)
    
  # Add
  graph = builder.compile()
    
  # View
  display(Image(graph.get_graph().draw_mermaid_png()))
  

  

  • 코드를 변경했더니 에러 없이 돌아가는 것을 확인할 수 있습니다. 하지만 그래프의 노드들이 None 값을 처리하도록 설계되지 않았다면, TypeError가 발생할 가능성이 높습니다.

    

이때 사용되는 것이 바로 커스텀 Reducer 함수입니다.

• 이 reduce_list 함수는 사용자 정의 리듀서(custom reducer)로, 두 리스트를 안전하게 결합하는 역할을 합니다.
  • 이 함수의 주요 특징과 None 처리 방식은 다음과 같습니다:
    • 두 개의 리스트를 입력받아 하나의 리스트로 결합합니다.
    • None 값을 포함한 다양한 입력 상황을 안전하게 처리합니다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
def reduce_list(left: list | None, right: list | None) -> list:
    """Safely combine two lists, handling cases where either or both inputs might be None.

    Args:
        left (list | None): The first list to combine, or None.
        right (list | None): The second list to combine, or None.

    Returns:
        list: A new list containing all elements from both input lists.
               If an input is None, it's treated as an empty list.
    """
    if not left:
        left = []
    if not right:
        right = []
    return left + right

• 위와 같이 구현함으로써 None이 들어와도 정상적으로 값을 반환하는 것을 확인할 수 있습니다.

  

SUMMARY

• 기본 Reducer: 상태를 덮어쓰는 방식입니다. 예를 들어, 그래프의 두 노드가 동일한 키를 동시에 업데이트하면, 마지막 업데이트된 값이 최종 값으로 저장됩니다.
• 커스텀 Reducer: 상태를 덮어쓰지 않고 값을 병합하거나 리스트에 추가하는 방식으로 동작합니다. 예를 들어, 키가 리스트로 정의된 경우 Reducer를 사용해 새로운 값을 리스트에 추가할 수 있습니다.
  • 에러 처리: 상태 업데이트 시 값이 None과 같은 유효하지 않은 값이 입력될 수 있습니다. 이를 방지하기 위해 커스텀 Reducer를 정의하여 이러한 경우에도 안전하게 값을 처리할 수 있습니다.

Messages

• CustomMessagesState: TypedDict를 상속받아 커스텀 딕셔너리 타입을 정의합니다.

  messages 키를 명시적으로 정의하고, Annotated를 사용하여 add_messages 리듀서를 연결합니다.

  • 추가적인 키들(added_key_1, added_key_2 등)을 직접 정의합니다.

• ExtendedMessagesState: MessagesState를 상속받습니다. 이 클래스는 이미 messages 키와 add_messages 리듀서가 구현되어 있습니다.

  • 추가적인 키들(added_key_1, added_key_2 등)만 정의합니다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from typing import Annotated
from langgraph.graph import MessagesState
from langchain_core.messages import AnyMessage
from langgraph.graph.message import add_messages

# Define a custom TypedDict that includes a list of messages with add_messages reducer
class CustomMessagesState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]
    added_key_1: str
    added_key_2: str
    # etc

# Use MessagesState, which includes the messages key with add_messages reducer
class ExtendedMessagesState(MessagesState):
    # Add any keys needed beyond messages, which is pre-built 
    added_key_1: str
    added_key_2: str
    # etc

• MessagesState은 내장된 messages key가 있는 것을 확인할 수 있습니다. CustomMessagesState는 별도로 messages key를 정의해줘야 합니다.

LangGraph에서는 메시지를 효과적으로 처리/관리하는 방법

LangGraph에서는 메시지를 효과적으로 처리/관리하기 위한 두 가지 주요 방법을 제공합니다.

• add_messages 리듀서 : add_messages 리듀서는 상태의 messages 키에 새 메시지를 추가하는 기능을 합니다. 이를 사용하면 메시지 목록을 쉽게 업데이트할 수 있습니다.
• MessagesState 클래스: MessagesState는 messages 키와 add_messages 리듀서가 미리 구현된 편리한 클래스입니다. 이를 상속하여 추가 키를 포함하는 커스텀 상태 클래스를 쉽게 만들 수 있습니다.

add_messages 리듀서의 다양한 기능

• 메시지 덮어쓰기 기능
  • 동일한 ID를 가진 메시지를 추가하면 기존 메시지를 덮어씁니다.
  • 이를 통해 특정 메시지를 업데이트할 수 있습니다.

    1
    2
    3
    4
    5
    6
    7
    8
      # Initial state
      initial_messages = [AIMessage(content="Hello! How can I assist you?", name="Model", id="1"),
                          HumanMessage(content="I'm looking for information on marine biology.", name="HUMAN", id="2")
                         ]
      # New message to add
      new_message = HumanMessage(content="I'm looking for information on whales, specifically", name="HUMAN", id="2")
      # Test
      add_messages(initial_messages , new_message)
    

    

• 메시지 제거 기능
  • RemoveMessage 객체를 사용하여 특정 ID의 메시지를 제거할 수 있습니다1
  • 이 기능은 더 이상 필요하지 않은 메시지를 상태에서 제거할 때 유용합니다.

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
      # Message list
      messages = [AIMessage("Hello how can I help you.", name="MODEL", id="1")]
      messages.append(HumanMessage("Hi. I am researching ocean mammals", name="HUMAN", id="2"))
      messages.append(AIMessage("That's great! Ocean mammals, also known as marine mammals. How can I help you?", name="MODEL", id="3"))
      messages.append(HumanMessage("Yes, I know about whales. But what others should I learn about?", name="HUMAN", id="4"))
      print("ORIGINAL MESSAGES:")
      print(messages)
      print('---------------')
      # Isolate messages to delete
      delete_messages = [RemoveMessage(id=m.id) for m in messages[:-2]]
      print("DELETED MESSAGES:")
      print(delete_messages)
      print('---------------')
      # Apply the deletion
      updated_messages = add_messages(messages, delete_messages)
      print("UPDATED MESSAGES:")
      print(updated_messages)
      print('---------------')
    

    

Lesson 3: Multiple Schemas

• Langraph에서 그래프의 노드 간에는 여러 스키마를 사용할 수 있습니다.

  • 이는 입력(Input) 스키마와 출력(Output) 스키마를 구분하여 그래프 내에서 각기 다른 형식으로 데이터를 처리할 수 있게 합니다.
  • 이러한 다중 스키마 접근 방식은 복잡한 워크플로우를 더 효과적으로 관리하고, 데이터의 흐름을 정밀하게 제어할 수 있게 해줍니다.

• 주요 스키마 유형:

  1. InputState (입력 스키마): 그래프에 전달되는 초기 데이터 구조를 정의합니다.
  2. OutputState (출력 스키마): 그래프가 최종적으로 반환하는 데이터 구조를 정의합니다.
  3. OverallState (전체 상태 스키마): 그래프 내부에서 사용되는 모든 데이터를 포함하는 포괄적인 스키마입니다.
  4. PrivateState (비공개 상태 스키마): 특정 노드 간에만 공유되는 임시 또는 중간 데이터를 정의합니다.
  5. User-defined schema (사용자 정의 스키마): 사용자가 특정 요구사항에 맞춰 직접 정의하는 스키마입니다

🤖 (추가) 예를 들어, 고객 서비스 챗봇을 만든다고 가정해 보겠습니다.

🤖 이 챗봇은 고객의 질문을 받아 적절한 답변을 제공하고, 필요한 경우 내부 데이터베이스를 검색합니다. 이 경우 다음과 같은 여러 스키마를 사용할 수 있습니다:

1. InputState (입력 스키마): 그래프에 전달되는 초기 데이터 구조를 정의

   1
   2
   3
   class InputState(TypedDict):
       customer_query: str
       customer_id: str
   

2. OutputState (출력 스키마): 그래프가 최종적으로 반환하는 데이터 구조를 정의

   1
   2
   3
   class OutputState(TypedDict):
       response: str
       satisfaction_score: int
   

3. OverallState (전체 상태 스키마): 그래프 내부에서 사용되는 모든 데이터를 포함하는 포괄적인 스키마

   1
   2
   3
   4
   5
   6
   7
   class OverallState(TypedDict):
       customer_query: str
       customer_id: str
       response: str
       satisfaction_score: int
       internal_notes: str
       database_results: dict
   

4. PrivateState (비공개 상태 스키마): 특정 노드 간에만 공유되는 임시 또는 중간 데이터를 정의

   1
   2
   3
   class PrivateState(TypedDict):
       sentiment_analysis: str
       priority_level: int
   

5. 이제 이 스키마들을 사용하는 그래프를 구성해 보겠습니다:

   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
   from langgraph.graph import StateGraph, START, END
      
   def query_analyzer(state: InputState) -> PrivateState:
       # 고객 질문 분석
       return {"sentiment_analysis": "positive", "priority_level": 2}
      
   def database_searcher(state: OverallState) -> OverallState:
       # 데이터베이스 검색
       return {"database_results": {"product_info": "..."}}
      
   def response_generator(state: OverallState) -> OutputState:
       # 응답 생성
       return {"response": "Here's the information you requested...", "satisfaction_score": 8}
      
   graph = StateGraph(OverallState, input=InputState, output=OutputState)
      
   graph.add_node("query_analyzer", query_analyzer)
   graph.add_node("database_searcher", database_searcher)
   graph.add_node("response_generator", response_generator)
      
   graph.add_edge(START, "query_analyzer")
   graph.add_edge("query_analyzer", "database_searcher")
   graph.add_edge("database_searcher", "response_generator")
   graph.add_edge("response_generator", END)
      
   compiled_graph = graph.compile()
   

이 예시에서:

1. 그래프는 InputState로 시작하여 고객의 질문을 받습니다.
2. query_analyzer 노드는 PrivateState를 생성하여 내부적으로 사용합니다.
3. database_searcher와 response_generator 노드는 OverallState를 사용하여 작업합니다.
4. 최종적으로 그래프는 OutputState를 반환하여 고객에게 응답을 제공합니다.

이렇게 함으로써:

• 입력과 출력을 명확히 제한할 수 있습니다.
• 내부 작업에 필요한 추가 데이터를 관리할 수 있습니다.
• 특정 노드 간에만 공유되는 비공개 데이터를 처리할 수 있습니다.

이 구조를 통해 복잡한 워크플로우를 관리하면서도 각 단계에서 필요한 정보만을 노출하고 처리할 수 있습니다.

각각의 State별로 상세하게 살펴보도록 하겠습니다:

PrivateState 예시

• OverallState (전체 상태 스키마): 그래프 내부에서 사용되는 모든 데이터를 포함하는 포괄적인 데이터 구조를 정의합니다.
• PrivateState (비공개 상태 스키마): 특정 노드 간에만 공유되는 임시 또는 중간 데이터를 정의합니다.
  • 특정 노드 간에만 공유되는 임시 또는 중간 데이터를 정의할때 사용합니다.
  • 아래 코드에서는 중간에 있는 Private 노드에서 사용된 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
25
26
27
28
29
30
31
32
33
from typing_extensions import TypedDict
from IPython.display import Image, display
from langgraph.graph import StateGraph, START, END

class OverallState(TypedDict):
    foo: int

class PrivateState(TypedDict):
    baz: int

def node_1(state: OverallState) -> PrivateState:
    print("---Node 1---")
    return {"baz": state['foo'] + 1}

def node_2(state: PrivateState) -> OverallState:
    print("---Node 2---")
    return {"foo": state['baz'] + 1}

# Build graph
builder = StateGraph(OverallState)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)

# Logic
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
builder.add_edge("node_2", END)

# Add
graph = builder.compile()

# View
display(Image(graph.get_graph().draw_mermaid_png()))

• baz is only included in PrivateState.
• node_2 uses PrivateState as input, but writes out to OverallState.

Input / Output Schema

• InputState (입력 스키마, 질문만 포함): 그래프에 전달되는 초기 데이터 구조를 정의합니다.
• OutputState (출력 스키마, 답변만 포함): 그래프가 최종적으로 반환하는 데이터 구조를 정의합니다.
• OverallState (전체 상태 스키마, 질문/답변/노트를 모두 포함): 그래프 내부에서 사용되는 모든 데이터를 포함하는 포괄적인 데이터 구조를 정의합니다.

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
class InputState(TypedDict):
    question: str

class OutputState(TypedDict):
    answer: str

class OverallState(TypedDict):
    question: str
    answer: str
    notes: str

def thinking_node(state: InputState):
    return {"answer": "bye", "notes": "... his is name is Chung"}

def answer_node(state: OverallState) -> OutputState:
    return {"answer": "bye Chung"}

graph = StateGraph(OverallState, input=InputState, output=OutputState)
graph.add_node("answer_node", answer_node)
graph.add_node("thinking_node", thinking_node)
graph.add_edge(START, "thinking_node")
graph.add_edge("thinking_node", "answer_node")
graph.add_edge("answer_node", END)

graph = graph.compile()

# View
display(Image(graph.get_graph().draw_mermaid_png()))

graph.invoke({"question":"hi"})

Lesson 4: Trim and Filter Messages

• 복습:

  • from langchain_core.messages import AIMessage, HumanMessage: 우리는 Langchain의 Messages 기능을 사용하여 AIMessage, HumanMessage를 정의하고 이를 LLM에 전달해줄 수 있습니다.

    1
    2
    3
    4
    5
    6
    from langchain_core.messages import AIMessage, HumanMessage
    messages = [AIMessage(f"당신이 해양 포유류를 연구하고 있다고 하셨나요?", name="AI")]
    messages.append(HumanMessage(f"네, 저는 고래에 대해 알고 있어요. 하지만 다른 어떤 동물에 대해 배워야 할까요?", name="HUMAN"))
        
    for m in messages:
        m.pretty_print()
    

    

  • 이전에 나눴던 대화들을 바탕으로 LLM이 답변을 하는 것을 확인할 수 있습니다.

    1
    2
    3
    4
    5
    6
    from langchain_openai import ChatOpenAI
    llm = ChatOpenAI(model="gpt-4o")
    messages.append(llm.invoke(messages))
        
    for m in messages:
        m.pretty_print()
    

    

  • 이러한 기능은 from langgraph.graph import MessagesState에 정의되어 있습니다.

    • ✨ 앞에서 다뤘던 것처럼 MessagesState에는 내장된 ‘messages’ 키가 존재합니다. 또한 이 키에는 내장된 add_messages 리듀서가 존재합니다.

      1
      2
      3
      4
      from typing import Annotated
      from langgraph.graph import MessagesState
      from langchain_core.messages import AnyMessage
      from langgraph.graph.message import add_messages
      

      1
      2
      3
      4
      5
      6
      # add_messages 리듀서로 메시지 리스트를 포함하는 사용자 정의 TypedDict 정의
      class CustomMessagesState(TypedDict):
        messages: Annotated[list[AnyMessage], add_messages]
        added_key_1: str
        added_key_2: str
        # 등
      

      1
      2
      3
      4
      5
      6
      # MessagesState 사용, 이미 add_messages 리듀서가 있는 messages 키 포함
      class ExtendedMessagesState(MessagesState):
        # messages 외에 필요한 키 추가, messages는 미리 구축되어 있음 
        added_key_1: str
        added_key_2: str
        # etc
      

  • messages 키: AnyMessage 객체의 리스트를 저장합니다. (list[AnyMessage])
  • AnyMessage는 HumanMessage, AIMessage, SystemMessage 등 다양한 메시지 타입을 포함할 수 있습니다
  • add_messages 리듀서가 기본으로 포함되어 있습니다. 이 리듀서는 새 메시지를 기존 메시지 리스트에 추가합니다.

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    from IPython.display import Image, display
    from langgraph.graph import MessagesState
    from langgraph.graph import StateGraph, START, END
      
    # Node
    def chat_model_node(state: MessagesState):
        return {"messages": llm.invoke(state["messages"])}
      
    # Build graph
    builder = StateGraph(MessagesState)
    builder.add_node("chat_model", chat_model_node)
    builder.add_edge(START, "chat_model")
    builder.add_edge("chat_model", END)
    graph = builder.compile()
      
    # View
    display(Image(graph.get_graph().draw_mermaid_png()))
    

    (참고) add_messages 리듀서 Output 예시

    • add_messages를 사용하면 그냥 str으로 메세지를 전달해줘도 HumanMessage로 전달되는 것을 확인할 수 있습니다. 반먄에 그냥 append를 할 경우, 일반 list append가 수행되는 것을 확인할 수 있습니다.

      

    • 그렇다면 append로 HumanMessage나 AIMessage를 넣어줄 수 있는 방법은 없을까요? 물론 있습니다!

      

장기 대화에서 메시지 히스토리가 길어질 경우, 토큰 사용량이 급증할 수 있습니다.

• 이를 방지하기 위해 Langraph에서는 메시지 히스토리를 관리하는 아래와 같은 방법들을 제공합니다. 각 기법에 대해서 살펴보겠습니다.

  1. 필터링: 메시지 히스토리에서 필요한 최근 메시지만 남기고 이전 메시지를 삭제하는 방식입니다.

     • (복습) 이전 시간에 배운 remove_messages 리듀서 함수는 특정 메시지 ID를 기준으로 메시지를 삭제하는 기능을 제공합니다. -> 이는 그래프 state에서도 해당 정보가 사라지게 됩니다
     • 하지만 이번에 배울 필터링의 개념은 이와는 조금 다른 개념입니다. 왜냐하면 모델에 들어가게되는 정보는 한정적이지만, 그래프의 state에는 모든 정보가 들어가있을 것이기 때문입니다. ```

     Node

     def chat_model_node(state: MessagesState): return {“messages”: [llm.invoke(state[“messages”][-1:])]} ```

     

     • 아래 그림 보면 message에는 모든 정보가 다 담겨 있는 것을 확인할 수 있습니다.

       

     • 반대로 실제 llm에 넘어간 질의문은 “해마에 대해서 알려알려줘”라는 질문만이 넘어간 것을 볼 수 있습니다.

       

  2. 트리밍: 메시지를 토큰 수 기준으로 잘라내는 방법(trim_messages)입니다. 위의 필터링이 에이전트 간의 메시지의 사후 하위 집합만 반환하는 반면, 트리밍은 채팅 모델이 응답하는 데 사용할 수 있는 토큰 수를 제한합니다.

     1
     2
     3
     4
     5
     6
     7
     8
     9
     10
     11
     12
     13
          
      from langchain_core.messages import trim_messages
          
      # Node
      def chat_model_node(state: MessagesState):
          messages = trim_messages(
                  state["messages"],
                  max_tokens=100,
                  strategy="last",
                  token_counter=ChatOpenAI(model="gpt-4o"),
                  allow_partial=False,
              )
          return {"messages": [llm.invoke(messages)]}
     

     • state["messages"]: 트리밍할 메시지 목록입니다.
     • max_tokens=100: 결과 메시지의 최대 토큰 수를 100으로 제한합니다.
     • strategy="last": 마지막 메시지부터 시작해 역순으로 메시지를 포함시킵니다.
     • token_counter=ChatOpenAI(model="gpt-4"): 토큰 수를 계산하는 데 사용할 모델을 지정합니다.

     • allow_partial=False: 메시지를 부분적으로 자르는 것을 허용하지 않습니다.

       

     • 여기도 위와 마찬가지로 LLM 모델에 들어가는 것은 token 조건에 해당하는 질의문만 넘어가게 됩니다.

       

     • 그리고 langraph 상에는 전체의 message history는 그대로 유지되게 됩니다.

       

Lesson 5: Chatbot w/ Summarizing Messages and Memory

이번 내용은 대화형 챗봇에서 대화를 요약하는 방법을 설명합니다. 요약 기능은 대화를 오래 유지하면서도 메시지의 전체 기록을 남기지 않고 대화를 압축하는 기술로, 앞에서 다룬 필터링이나 메시지 삭제보다 더 많은 정보를 보존하려는 시도입니다.

• 이 기능은 LLMs(Large Language Models)를 사용하여 대화의 주요 내용을 실시간으로 요약하는 방법을 설명합니다.

1. 기본 설정 및 메시지 상태 구성

• message state에는 기본적으로 메시지 목록을 저장하는 messages 키가 있지만, 여기에 대화 요약을 저장할 수 있는 새로운 summary 키를 추가합니다.

1
2
3
from langgraph.graph import MessagesState
class State(MessagesState):
    summary: str

• call model이라는 노드에서는 기존 요약이 있으면 메시지 목록에 추가하고 모델을 호출합니다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
from langchain_core.messages import SystemMessage, HumanMessage, RemoveMessage

# Define the logic to call the model
def call_model(state: State):
    
    # Get summary if it exists
    summary = state.get("summary", "")

    # If there is summary, then we add it
    if summary:
        
        # Add summary to system message
        system_message = f"Summary of conversation earlier: {summary}"

        # Append summary to any newer messages
        messages = [SystemMessage(content=system_message)] + state["messages"]
    
    else:
        messages = state["messages"]
    
    response = model.invoke(messages)
    return {"messages": response}

2. 요약 노드 구성

• 대화를 요약하는 summarize_conversation 노드를 추가하여 대화 상태를 받아 요약을 생성합니다.
  • 만약 기존 요약이 있다면 이를 새로운 요약에 포함하여 업데이트합니다.
  • 기존 요약이 없다면 새로 요약을 생성합니다.
  • 요약을 생성한 후 상태를 필터링하기 위해 RemoveMessage를 사용하여 가장 최근 메시지 두 개만 남기고 나머지 메시지를 삭제합니다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
def summarize_conversation(state: State):
    
    # First, we get any existing summary
    summary = state.get("summary", "")

    # Create our summarization prompt 
    if summary:
        
        # A summary already exists
        summary_message = (
            f"This is summary of the conversation to date: {summary}\n\n"
            "Extend the summary by taking into account the new messages above:"
        )
        
    else:
        summary_message = "Create a summary of the conversation above:"

    # Add prompt to our history
    messages = state["messages"] + [HumanMessage(content=summary_message)]
    response = model.invoke(messages)
    
    # Delete all but the 2 most recent messages
    delete_messages = [RemoveMessage(id=m.id) for m in state["messages"][:-2]]
    return {"summary": response.content, "messages": delete_messages}

3. 메시지 삭제와 요약 업데이트

• 대화 길이에 따라 요약을 생성할지 결정하는 조건부 엣지를 추가합니다.
  • len(messages) > 6: 길이가 6보다 길면 요약을 제공합니다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
from langgraph.graph import END
# Determine whether to end or summarize the conversation
def should_continue(state: State):
    
    """Return the next node to execute."""
    
    messages = state["messages"]
    
    # If there are more than six messages, then we summarize the conversation
    if len(messages) > 6:
        return "summarize_conversation"
    
    # Otherwise we can just end
    return END

4.체크포인터로 상태 유지

• 상태는 일시적이기 때문에 대화가 길어지면 상태를 보존하기 위해 MemorySaver의 checkpointer 기능을 사용합니다.
• 이 체크포인터는 메모리에 저장된 상태를 보존하며, 이를 통해 여러 번의 대화에서 상태를 유지할 수 있습니다.
  • 요약 후 LLM에 대화를 요청하는 것이 아니라, 결과를 얻은 후 output에 산출하기 전에 누적된 대화가 많다면 요약을 수행하는 형태의 그래프
  • memory 기능이 있어 다음 대화를 시작하는 단계에서 누적된 수에 따라 요약 여부 결정함

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
from IPython.display import Image, display
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START

# Define a new graph
workflow = StateGraph(State)
workflow.add_node("conversation", call_model)
workflow.add_node(summarize_conversation)

# Set the entrypoint as conversation
workflow.add_edge(START, "conversation")
workflow.add_conditional_edges("conversation", should_continue)
workflow.add_edge("summarize_conversation", END)

# Compile
memory = MemorySaver()
graph = workflow.compile(checkpointer=memory)
display(Image(graph.get_graph().draw_mermaid_png()))

Threads

체크포인터는 각 단계에서 상태를 체크포인트로 저장합니다. 이렇게 저장된 체크포인트는 대화의 스레드로 그룹화될 수 있습니다.

• 예시: ChatGPT에서 사용자와 AI 사이의 대화는 스레드로 구성됩니다. 각 스레드는 하나의 주제나 대화 흐름을 나타냅니다.

• 스레드의 특징:

  • 각 메시지는 스레드 내의 체크포인트와 같습니다.
  • 새로운 대화를 시작하면 새 스레드가 생성됩니다.
  • 하나의 스레드 안에서 대화의 맥락이 유지됩니다.

• ChatGPT 인터페이스에서는 왼쪽 사이드바에 각 스레드가 표시됩니다. 사용자는 이를 통해:
  • 이전 대화 스레드로 쉽게 돌아갈 수 있습니다.
  • 여러 주제의 대화를 동시에 관리할 수 있습니다.
  • 특정 대화 내용을 빠르게 찾을 수 있습니다.

Langraph에서는 이와 비슷한 개념으로 configurable을 사용하여 스레드 ID를 설정합니다.

아래 그림과 같이 config = {"configurable": {"thread_id": "1"}} 선언 후 invoke 시에 config으로 넣어주게 되면 따로따로 실행해도 message에 기록이 제대로 남고 있는 것을 확인할 수 있습니다.

아직까지 6개의 대화가 넘지 않았기 때문에 요약(summary)가 수행되지 않은 것을 볼 수 있습니다.

• 하지만 하나의 대화를 추가함으로써 6개가 넘어서 요약이 발생하고 should_continue => summarize_conversation이 발생하고, 가장 최근 2개의 대화만이 남는 것을 확인할 수 있습니다.

• 또한, 이전과는 다르게 Summary도 생성이 되어있는 것을 확인할 수 있습니다.

💡 (SUMMARY) 위에 정의한 구조는 다음과 같이 작동합니다:

• 요약이 생성되면 State 객체의 summary 필드에 저장됩니다.
• 새로운 사용자 메시지가 들어오면 call_model 함수가 호출됩니다.
• call_model 함수 내에서:
  • 저장된 요약이 있는지 확인합니다.
  • 요약이 있다면, 이를 시스템 메시지로 변환합니다.
  • 이 시스템 메시지를 대화 히스토리의 맨 앞에 추가합니다.
  • 그 다음에 새로운 사용자 메시지를 포함한 최근 메시지들이 추가됩니다.
• 이렇게 구성된 메시지 리스트(요약 + 최근 메시지들)가 LLM에 전달됩니다.

Lesson 6: Chatbot w/ Summarizing Messages and External Memory

Langraph는 외부 데이터베이스와의 연동을 통해 더 영속적인 메모리 저장 기능을 제공합니다.

1. SQL Lite 사용: Langraph는 SQL Lite와 같은 외부 데이터베이스를 지원하여 메모리를 영구적으로 저장할 수 있습니다. 이 방식은 장기 대화를 지속적으로 유지하고 관리할 수 있는 환경을 제공합니다.

📊 SQLite (https://www.sqlite.org/)

• SQLite는 가볍고 설정이 필요 없는 서버리스 RDBMS로, 다양한 환경에서 손쉽게 사용할 수 있습니다.
• Python에서는 sqlite3 모듈을 통해 데이터베이스 연결부터 쿼리 실행까지 간단하게 구현할 수 있습니다.
• 전체 데이터가 하나의 파일에 저장되어 관리와 배포가 용이합니다.
• 트랜잭션 관리와 SQL 표준 지원으로 안정적인 데이터 처리가 가능합니다.

1. LangChain에서의 SqliteSaver 체크포인터: 언급된 SqliteSaver 체크포인터는 LangChain에서 제공하는 체크포인트 저장 방식 중 하나로, 대화나 상태를 SQLite 데이터베이스에 저장하는 방식입니다. 이를 통해 대화의 상태나 진행 상황을 메모리나 파일에 저장하여 복구할 수 있습니다.

   • 체크포인터 역할: 시스템이 중단되거나 대화가 길어졌을 때, 상태를 보존하거나 관리할 수 있도록 돕습니다. SQLite는 이러한 체크포인터로 적합한 경량형 데이터베이스입니다. 메모리를 SQL Lite에 저장함으로써 세션이 종료되더라도 대화 기록을 유지할 수 있습니다. 새로운 세션이 시작되더라도 이전 대화를 불러와 이어서 진행할 수 있습니다.

     

   • 메모리 내 SQLite 데이터베이스

     1
     2
     3
     4
          
     import sqlite3
     # In-memory database
     conn = sqlite3.connect(":memory:", check_same_thread=False)
     

     • :memory:: SQLite는 "memory" 문자열을 제공하면, 메모리 내에서 데이터베이스를 생성합니다. 이는 데이터베이스가 휘발성 메모리(RAM)에서 동작하며, 애플리케이션이 종료되거나 연결이 끊어지면 데이터가 사라집니다. 따라서 테스트, 임시 데이터 처리, 빠른 연산이 필요한 상황에서 매우 유용합니다.
     • check_same_thread=False: SQLite는 기본적으로 멀티스레드에서의 동시 작업을 허용하지 않지만, 이 옵션을 설정하면 다중 스레드 환경에서도 하나의 연결을 공유할 수 있습니다.

   • 로컬 SQLite 데이터베이스 저장

     1
     2
     3
     4
          
     # 로컬 db에 저장
     db_path = "state_db/example.db"
     conn = sqlite3.connect(db_path, check_same_thread=False)
     

     • 로컬 파일 기반 SQLite: 위 예시에서는 db_path = "state_db/example.db"와 같이, 로컬 파일에 데이터베이스를 저장하는 방식을 보여줍니다. SQLite는 파일 기반 데이터베이스로, 하나의 파일에 모든 데이터를 저장합니다. 이 파일은 다른 환경으로 쉽게 이동하거나 백업할 수 있습니다.

• SQLite 데이터베이스 설정 및 체크포인터 생성:

  1
  2
  3
    
  from langgraph.checkpoint.sqlite import SqliteSaver
  memory = SqliteSaver(conn)
  

  • SqliteSaver를 사용하여 SQLite 데이터베이스를 체크포인터로 설정합니다. 이를 통해 대화 상태가 데이터베이스에 영구적으로 저장되며, 나중에 상태를 불러올 수 있습니다.

• 이전 챕터에서 만든 Agent 재-생성

  • (복습) call_model, summarize_conversation, should_continue 기반 내용 요약 및 질의를 수행하는 LLM agent => Agent 설명
  • (목적) 이전에는 in-memory로 작동했지만, 이번에스는 external DB와 연동을 하고 싶음. ```

  from langchain_openai import ChatOpenAI from langchain_core.messages import SystemMessage, HumanMessage, RemoveMessage

  from langgraph.graph import END from langgraph.graph import MessagesState

  model = ChatOpenAI(model=”gpt-4o”,temperature=0)

  class State(MessagesState): summary: str

  모델을 호출하는 로직 정의

  def call_model(state: State):

  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16
  17
  # 요약이 있다면 가져옵니다
  summary = state.get("summary", "")
    
  # 요약이 있다면 추가합니다
  if summary:
    
      # 시스템 메시지에 요약 추가
      system_message = f"이전 대화의 요약: {summary}"
    
      # 새로운 메시지에 요약 추가
      messages = [SystemMessage(content=system_message)] + state["messages"]
    
  else:
      messages = state["messages"]
    
  response = model.invoke(messages)
  return {"messages": response}
  

  def summarize_conversation(state: State):

  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16
  17
  18
  19
  20
  21
  22
  # 먼저 기존 요약을 가져옵니다
  summary = state.get("summary", "")
    
  # 요약 프롬프트 생성 
  if summary:
    
      # 요약이 이미 존재함
      summary_message = (
          f"지금까지의 대화 요약입니다: {summary}\n\n"
          "위의 새로운 메시지를 고려하여 요약을 확장하세요:"
      )
    
  else:
      summary_message = "위의 대화를 요약하세요:"
    
  # 프롬프트를 히스토리에 추가
  messages = state["messages"] + [HumanMessage(content=summary_message)]
  response = model.invoke(messages)
    
  # 가장 최근 2개의 메시지만 남기고 모두 삭제
  delete_messages = [RemoveMessage(id=m.id) for m in state["messages"][:-2]]
  return {"summary": response.content, "messages": delete_messages}
  

  대화를 종료할지 요약할지 결정

  def should_continue(state: State):

  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  """다음에 실행할 노드를 반환합니다."""
    
  messages = state["messages"]
    
  # 메시지가 6개 이상이면 대화를 요약합니다
  if len(messages) > 6:
      return "summarize_conversation"
    
  # 그렇지 않으면 종료합니다
  return END ```
  

  

  • 차이점 :
    • In-Memory: from langgraph.checkpoint.memory import MemorySaver
    • External-DB: from langgraph.checkpoint.sqlite import SqliteSaver 그 외에는 공식 지원되는 외부 데이터베이스 체크포인터로 from langgraph.checkpoint.postgres import PostgresSaver이 있습니다.

• 대화 상태 로드/복구: SQLite를 사용하면 대화 상태가 데이터베이스에 저장되므로, 대화가 길어지거나 세션이 종료되더라도 데이터를 복구할 수 있습니다.

  • 예를 들어, 노트북 커널을 재시작하더라도 동일한 SQLite DB를 사용하여 상태를 다시 불러올 수 있습니다. ```

  config = {“configurable”: {“thread_id”: “1”}} graph_state = graph.get_state(config) graph_state

  1
  2
  3
  4
  + `config`: 대화의 스레드 ID를 지정하는 설정입니다. 여기서는 "thread\_id": "1"이라는 스레드에 해당하는 대화 상태를 가져오는 설정을 하고 있습니다.
  + `graph.get_state(config)`: 이 메서드는 주어진 config에 맞는 스레드의 대화 상태를 SQLite 데이터베이스에서 불러오는 역할을 합니다.  
      
    ![](https://velog.velcdn.com/images/euisuk-chung/post/40ba3039-f77d-451e-9f18-30c731e4327b/image.png)
  

  ```

1. 유용한 CODE-SNIPPETS: SQLite DB를 호출하고 나서, 아래와 같은 함수들을 통해 정보를 호출할 수 있습니다:

1
2
3
4
5
6
7
8
# StateSnapshot에서 메시지 부분을 추출하여 보기 좋게 출력
for message in graph_state.values['messages']:
    print(f"Message ID: {message.id}")
    print(f"Message Type: {'Human' if isinstance(message, HumanMessage) else 'AI'}")
    print(f"Content: {message.content}")
    print('-' * 40)

• 예시 출력:

  

1
2
3
4
# 요약 출력
summary = graph_state.values.get('summary', 'No summary available')
print(f"Conversation Summary: {summary}")

• 예시 출력:

  

• Previous
  [개념] GLU와 그 변형들: 역사와 주요 개념 정리
• Next
  [강의노트] LangChain Academy : Introduction to LangGraph (Module 3)