LangGraph 강의노트 AI Agent LangChain

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

Posted by Euisuk's Dev Log on October 16, 2024

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

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

  • Lesson 1: Streaming
  • Lesson 2: Breakpoints
  • Lesson 3: Editing State and Human Feedback
  • Lesson 4: Dynamic Breakpoints
  • Lesson 5: Time Travel

Lesson 1: Streaming

  • 본 강의에서도 지난번에 사용했던 요약 LLM Agent를 사용하여 실습을 수행합니다.
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
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90

from IPython.display import Image, display

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

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END
from langgraph.graph import MessagesState

# LLM
model = ChatOpenAI(model="gpt-4o", temperature=0) 

# 상태 
class State(MessagesState):
    summary: str

# 모델을 호출하는 로직 정의
def call_model(state: State):
    
    # 요약이 있으면 가져옴
    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):
    
    # 먼저 기존 요약을 가져옴
    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):
    
    """다음에 실행할 노드를 반환합니다."""
    
    messages = state["messages"]
    
    # 메시지가 6개 이상이면 대화를 요약
    if len(messages) > 6:
        return "summarize_conversation"
    
    # 그렇지 않으면 종료
    return END

# 새 그래프 정의
workflow = StateGraph(State)
workflow.add_node("conversation", call_model)
workflow.add_node(summarize_conversation)

# 시작점을 대화로 설정
workflow.add_edge(START, "conversation")
workflow.add_conditional_edges("conversation", should_continue)
workflow.add_edge("summarize_conversation", END)

# 컴파일
memory = MemorySaver()
graph = workflow.compile(checkpointer=memory)
display(Image(graph.get_graph().draw_mermaid_png()))

개요

Langraph의 Streaming 기능은 에이전트가 작업 중 발생하는 이벤트와 상태를 실시간으로 확인할 수 있는 기능입니다.

주요 개념

스트리밍 매서드

  • LangGraph의 .stream()과 .astream() 메서드는 그래프 실행 중 상태를 스트리밍하는 두 가지 주요 방법입니다. 이 두 메서드의 차이점과 사용 방법을 설명해드리겠습니다:

  1. .stream() - 동기 방식 스트리밍:

    • .stream(): Stream은 동기식(Synchronous) 방식으로, 각 노드가 호출될 때마다 상태 업데이트를 스트리밍합니다.

    • 동기적으로 작동한다는 것은, 작업을 하나씩 순차적으로 처리하는 것을 의미합니다. 앞의 작업이 완료되지 않으면 뒤의 작업이 실행되지 않습니다.

      • 예를 들어, 친구에게 전화를 걸어서 친구가 전화를 받기 전까지 기다리는 상황을 생각해 보세요. 전화를 받기 전까지 다른 일을 할 수 없는 것처럼, 동기 방식은 하나의 작업이 끝날 때까지 대기한 후 다음 작업을 진행하는 방식입니다.

    • LangGraph에서 .stream() 메서드는 동기 방식으로 작동합니다. 즉, 각 노드가 완료될 때까지 기다린 후 그 결과를 받아오는 방식입니다. 예를 들어, 그래프의 첫 번째 노드가 실행을 완료해야 두 번째 노드가 실행될 수 있습니다.

      1
2

      for chunk in graph.stream(inputs, stream_mode="values"):
    print(chunk)  # 각 노드 실행 후 출력
      • 이 코드에서는 첫 번째 노드의 실행이 끝나야 그 결과(chunk)가 출력되고, 그 후에야 두 번째 노드가 실행됩니다.
      • 모든 작업이 순서대로 이루어지기 때문에 코드의 흐름이 예측 가능하고 직관적입니다. 그러나 작업이 오래 걸릴 경우, 그 작업이 끝날 때까지 아무것도 할 수 없습니다.

  2. .astream() - 비동기 방식 스트리밍:

    • .astream(): A-Stream은 비동기식(Asynchronous) 방식으로 상태 업데이트를 스트리밍합니다.

    • 비동기적으로 작동한다는 것은, 작업을 동시에 처리할 수 있다는 것을 의미합니다. 앞의 작업이 끝날 때까지 기다리지 않고, 다른 작업을 진행할 수 있습니다.

      • 다시 전화를 거는 예시로 돌아가면, 친구가 전화를 받을 때까지 기다리지 않고 다른 일을 할 수 있는 상황을 상상해 보세요. 친구가 전화를 받으면 그때 다시 통화로 돌아올 수 있습니다.

    • LangGraph에서 .astream() 메서드는 비동기 방식으로 작동합니다. 즉, 한 작업이 끝날 때까지 기다리지 않고 다른 작업을 동시에 처리할 수 있습니다. 이 방식은 특히 많은 시간이 걸리는 작업(예: 네트워크 요청, 데이터베이스 접근 등)을 효율적으로 처리하는 데 유리합니다.

      1
2

      async for chunk in graph.astream(inputs, stream_mode="values"):
    print(chunk)  # 각 노드 실행이 끝날 때까지 기다리지 않고 처리
      • 이 코드에서는 첫 번째 노드의 실행이 완료되기 전에 두 번째 노드가 실행될 수 있습니다.
      • 비동기 방식 덕분에 여러 작업을 동시에 처리할 수 있지만, 그로 인해 코드의 실행 순서가 복잡해질 수 있습니다.

🤔 언제 동기/비동기를 사용해야 할까?

  • 동기: 작업이 비교적 빠르게 끝나고 순차적으로 실행해야 할 때 사용합니다.
    • 예를 들어, 작은 규모의 프로그램이나 한 번에 하나씩 처리해야 하는 작업에서 동기 방식을 사용하는 것이 좋습니다.
  • 비동기: 비동기는 여러 작업을 동시에 처리해야 할 때 특히 유용합니다.
    • 예를 들어, 네트워크 요청, 파일 읽기/쓰기, 데이터베이스 접근과 같이 시간이 오래 걸리는 작업에서 비동기 방식을 사용하는 것이 더 효율적입니다.

스트리밍 모드

  • 스트림 모드는 LangGraph가 실행되는 동안 그래프의 상태를 어떤 방식으로 모니터링하고 확인할 것인지를 결정하는 옵션입니다.
  • 각 모드는 그래프가 실행될 때 생성되는 데이터를 어떻게 스트리밍할지에 대한 방식을 정의합니다. 쉽게 말하면, 스트리밍 모드는 실행 중 발생하는 정보(상태나 결과물)를 실시간으로 어떻게 보고할 것인지를 나타냅니다.

주요 스트림 모드

  1. “values” 모드:

    • 그래프의 각 노드가 실행된 후, 전체 상태를 스트리밍합니다.
    • 즉, 노드가 실행된 후 현재까지의 모든 상태를 확인할 수 있습니다.
    • 예를 들어, 노드가 실행될 때마다 전체 작업 흐름의 스냅샷을 제공받는 것과 같습니다. 이 방법은 각 노드 실행 후 전체 상태를 검토할 때 유용합니다.

    사용 예시:

    1
2

    for chunk in graph.stream(inputs, stream_mode="values"):
    print(chunk)  # 각 노드 실행 후 전체 상태 출력

  2. “updates” 모드:

    • 각 노드 실행 후 변경된 부분만 스트리밍합니다.
    • 즉, 상태가 변한 부분만 실시간으로 볼 수 있어, 불필요한 데이터를 줄이고 관심 있는 부분만 확인하는 데 유리합니다.
    • 예를 들어, 대규모 데이터를 처리하는 그래프에서 변화가 있는 부분만 확인하고 싶을 때 유용합니다.

    사용 예시:

    1
2

    for chunk in graph.stream(inputs, stream_mode="updates"):
    print(chunk)  # 변경된 상태만 출력

  1. LLM 토큰 스트리밍: LLM 토큰 스트리밍은 LangGraph에서 대형 언어 모델(LLM)이 생성하는 텍스트를 실시간으로 모니터링하는 기능입니다.
    • LLM은 주로 긴 텍스트를 생성하는데, 모든 결과가 한 번에 반환되는 대신 토큰이라는 작은 단위로 결과를 점진적으로 제공할 수 있습니다.
    • 이때 스트리밍을 통해 모델이 생성하는 텍스트를 한 번에 기다리지 않고 실시간으로 확인할 수 있는 장점이 있습니다. 이 기능은 대화형 시스템, 긴 텍스트 생성 작업에서 특히 유용합니다.

💡 LLM 토큰 스트리밍의 목적

  1. 실시간 응답 제공: LLM이 응답을 생성하는 동안 즉각적인 피드백을 사용자에게 제공하여 기다리는 시간을 줄이고, 사용자 경험을 향상시킬 수 있습니다.

  2. 세밀한 모니터링: LLM이 중간에 생성하는 텍스트를 관찰하여 시스템의 성능을 평가하거나 문제를 디버깅할 수 있습니다.

  3. 유연한 제어: 특정 노드에서 발생하는 토큰만을 실시간으로 확인하고, 필요한 경우 중간에 개입하거나 제어할 수 있습니다.

스트리밍 토큰의 주요 개념

  1. 이벤트 기반 스트리밍:
    • LLM이 텍스트를 생성하는 동안 이벤트가 발생하며, 이러한 이벤트를 기반으로 토큰이 스트리밍됩니다.
    • LangGraph에서 .astream_events() 메서드를 사용해 이러한 이벤트를 비동기적으로 처리할 수 있습니다.
1
2
3
4

config = {"configurable": {"thread_id": "3"}}
input_message = HumanMessage(content="피파에서 대한민국은 어느정도 실력인가요?")
async for event in graph.astream_events({"messages": [input_message]}, config, version="v2"):
    print(f"노드: {event['metadata'].get('langgraph_node','')}. 유형: {event['event']}. 이름: {event['name']}")

  • 출력 예시:

    1
2
3
4
5
6
7
8

      
노드: . 유형: on_chain_start. 이름: LangGraph
노드: __start__. 유형: on_chain_start. 이름: __start__
노드: __start__. 유형: on_chain_end. 이름: __start__
노드: conversation. 유형: on_chain_start. 이름: conversation
노드: conversation. 유형: on_chat_model_start. 이름: ChatOpenAI
노드: conversation. 유형: on_chat_model_stream. 이름: ChatOpenAI
(중략)
  • 이 코드는 그래프의 특정 노드에서 발생하는 이벤트를 실시간으로 추적하여, 각 노드가 어떤 작업을 하고 있는지 상태 정보를 얻습니다.

  • 해당 코드 및 출력 결과에서 event['metadata'], event['event'], event['name'] 등은 LangGraph에서 발생하는 이벤트의 구성 요소입니다.

    • 이러한 구성 요소들은 그래프 실행 중 발생하는 이벤트에 대한 메타데이터를 제공합니다. 이를 통해 어떤 노드에서 어떤 유형의 이벤트가 발생했는지, 그 이벤트의 이름이 무엇인지를 실시간으로 추적할 수 있습니다.
  • 예를 들어, 위 출력 마지막에 대화형 AI에서 현재 LLM이 “on_chat_model_stream” 이벤트를 통해 현재 해당 노드에서 응답을 생성하는 중임을 확인할 수 있습니다.

  1. 이벤트 구조:

    • event: 발생한 이벤트의 유형을 나타냅니다. 예를 들어, LLM이 스트리밍하는 이벤트는 "on_chat_model_stream" 혹은 "on_chain_stream" 같은 유형으로 표시됩니다.

      • on_chat_model_stream: LLM(대형 언어 모델)이 생성하는 토큰을 실시간으로 스트리밍하는 이벤트입니다. LLM이 입력을 받아 응답을 생성하는 과정에서 생성된 부분적인 텍스트(토큰)들이 실시간으로 전송됩니다.
      • on_chain_stream: 그래프의 특정 노드에서 발생하는 일반적인 이벤트 스트리밍을 나타냅니다. 모든 노드에서 발생할 수 있는 이벤트이며, 각 노드가 실행되면서 일어나는 작업의 상태를 스트리밍합니다.
    • name: 이벤트의 이름으로, 어떤 노드에서 발생한 이벤트인지 명시합니다.
    • data: 이벤트와 관련된 데이터를 담고 있습니다. LLM이 생성하는 실제 텍스트 데이터(토큰)를 포함합니다. 이 데이터는 AIMessageChunk와 같은 형태로 제공됩니다.
    • metadata: 이 필드는 추가적인 메타 정보를 포함합니다. 여기에는 이벤트가 발생한 노드(예: langgraph_node)에 대한 정보도 포함됩니다.

스트리밍 방법

  1. 특정 노드의 토큰 스트리밍:
    • LLM이 여러 노드에서 호출될 수 있기 때문에, 특정 노드에서 발생하는 토큰만을 스트리밍하고 싶을 때가 있습니다.
    • 이를 위해 metadata['langgraph_node'] 값을 이용해 특정 노드의 출력을 필터링할 수 있습니다.

on_chain_stream:

  • 이 코드는 특정 노드(예: conversation)의 일반적인 상태 변화나 그래프 실행 상태를 추적하기 위한 용도로 사용됩니다.
  • LLM의 토큰 생성과는 직접적인 관련이 없으며, 그래프의 노드 상태 변화를 추적하는 방식입니다.
1
2
3
4
5
6
7
8

# on_chain_stream
node_to_stream = 'conversation'
config = {"configurable": {"thread_id": "1"}}
input_message = HumanMessage(content="피파에서 대한민국은 어느정도 실력인가요?")
async for event in graph.astream_events({"messages": [input_message]}, config, version="v2"):
    # 특정 노드의 상태 변화 추적하기
    if event["event"] == "on_chain_stream" and event['metadata'].get('langgraph_node','') == node_to_stream:
        print(event["data"])
  • on_chain_stream: 그래프의 상태 변화나 노드 실행 상태가 바뀔 때 발생하는 이벤트입니다. LLM의 응답 생성이 끝나고 그래프의 상태가 변하면 이 이벤트가 발생하며, 이때 전체 응답이 나온 이유는 LLM이 답변을 완료했기 때문이 아니라, 그래프 내 노드의 상태가 변화했기 때문입니다.
1
2
3
4
5
6
7
8
9

node_to_stream = 'conversation'
config = {"configurable": {"thread_id": "2"}}
input_message = HumanMessage(content="피파에서 대한민국은 어느정도 실력인가요?")
async for event in graph.astream_events({"messages": [input_message]}, config, version="v2"):
    # 특정 노드에서 채팅 모델 토큰 가져오기 
    if event["event"] == "on_chat_model_stream" and event['metadata'].get('langgraph_node','') == node_to_stream:
    # if event["event"] == "on_chain_stream" and event['metadata'].get('langgraph_node','') == node_to_stream:    
        print(event["data"])
(중략)

on_chat_model_stream:

  • 이 코드는 LLM의 응답이 점진적으로 생성될 때마다 발생하는 이벤트를 스트리밍합니다.
  • on_chat_model_stream 이벤트는 LLM이 응답을 생성할 때 부분적으로 생성된 텍스트(토큰)를 실시간으로 추적하는 데 사용됩니다.
1
2
3
4
5
6
7
8

# on_chat_model_stream
node_to_stream = 'conversation'
config = {"configurable": {"thread_id": "2"}}
input_message = HumanMessage(content="피파에서 대한민국은 어느정도 실력인가요?")
async for event in graph.astream_events({"messages": [input_message]}, config, version="v2"):
    # 특정 노드에서 채팅 모델 토큰 추적하기
    if event["event"] == "on_chat_model_stream" and event['metadata'].get('langgraph_node','') == node_to_stream:
        print(event["data"])
  • on_chat_model_stream: LLM이 토큰 단위로 점진적으로 응답을 생성할 때마다 발생하는 이벤트입니다. LLM이 긴 응답을 생성할 때 각각의 토큰(작은 텍스트 단위)이 실시간으로 스트리밍되어 결과가 나오는 것입니다. 이 방식은 실시간으로 토큰을 추적하며, 긴 응답을 기다리지 않고 중간 결과를 빠르게 확인할 수 있습니다.
1
2
3
4
5
6
7
8
9
10
11
12
13
14

{'chunk': AIMessageChunk(content='대한', additional_kwargs={}, response_metadata={}, id='run-f5d6eb65-2633-4bfe-bb73-82cc16348735')}
{'chunk': AIMessageChunk(content='민국', additional_kwargs={}, response_metadata={}, id='run-f5d6eb65-2633-4bfe-bb73-82cc16348735')}
{'chunk': AIMessageChunk(content=' 축', additional_kwargs={}, response_metadata={}, id='run-f5d6eb65-2633-4bfe-bb73-82cc16348735')}
{'chunk': AIMessageChunk(content='구', additional_kwargs={}, response_metadata={}, id='run-f5d6eb65-2633-4bfe-bb73-82cc16348735')}
{'chunk': AIMessageChunk(content=' 국가', additional_kwargs={}, response_metadata={}, id='run-f5d6eb65-2633-4bfe-bb73-82cc16348735')}
{'chunk': AIMessageChunk(content='대표', additional_kwargs={}, response_metadata={}, id='run-f5d6eb65-2633-4bfe-bb73-82cc16348735')}
{'chunk': AIMessageChunk(content='팀', additional_kwargs={}, response_metadata={}, id='run-f5d6eb65-2633-4bfe-bb73-82cc16348735')}
{'chunk': AIMessageChunk(content='은', additional_kwargs={}, response_metadata={}, id='run-f5d6eb65-2633-4bfe-bb73-82cc16348735')}
{'chunk': AIMessageChunk(content=' 아', additional_kwargs={}, response_metadata={}, id='run-f5d6eb65-2633-4bfe-bb73-82cc16348735')}
{'chunk': AIMessageChunk(content='시아', additional_kwargs={}, response_metadata={}, id='run-f5d6eb65-2633-4bfe-bb73-82cc16348735')}
{'chunk': AIMessageChunk(content='에서', additional_kwargs={}, response_metadata={}, id='run-f5d6eb65-2633-4bfe-bb73-82cc16348735')}
{'chunk': AIMessageChunk(content=' 강', additional_kwargs={}, response_metadata={}, id='run-f5d6eb65-2633-4bfe-bb73-82cc16348735')}
{'chunk': AIMessageChunk(content='력', additional_kwargs={}, response_metadata={}, id='run-f5d6eb65-2633-4bfe-bb73-82cc16348735')}
(중략)
  1. 토큰 데이터 추출:
    • 이벤트에서 event['data'] 값이 AIMessageChunk 형태로 제공되며, 이를 사용해 실제 토큰 데이터를 추출합니다.
    • chunk 키를 사용하여 스트리밍된 텍스트 데이터를 얻고, 이를 실시간으로 확인할 수 있습니다.
1
2
3
4
5
6
7
8

config = {"configurable": {"thread_id": "222"}}
input_message = HumanMessage(content="피파에서 대한민국은 어느정도 실력인가요?")
async for event in graph.astream_events({"messages": [input_message]}, config, version="v2"):
    # 특정 노드에서 채팅 모델 토큰 가져오기 
    # if event["event"] == "on_chat_model_stream" and event['metadata'].get('langgraph_node','') == node_to_stream:
    if event["event"] == "on_chain_stream" and event['metadata'].get('langgraph_node','') == node_to_stream:
        data = event["data"]
        print(data["chunk"]['messages'].content)

Lesson 2: Breakpoints

개요

Breakpoints는 작업 흐름에서 특정 지점에서 그래프의 실행을 멈추고, 사용자가 검토하거나 작업을 중단할 수 있게 해주는 기능입니다.

  • 이는 Human-in-the-Loop 워크플로우에서 중요한 요소로, 민감한 작업(예: 데이터베이스 쓰기, 외부 시스템에 쓰기)에 대해 사람이 승인을 할 수 있도록 해줍니다.

주요 개념

  1. Human-in-the-loop (인간 개입) 개념:

    • AI 시스템의 작동 과정에 인간이 개입하여 감독, 수정, 승인 등을 수행하는 방식입니다.
    • 이는 AI의 결정이나 행동을 검토하고 필요시 조정할 수 있게 해줍니다.

  2. LangGraph에서의 Human-in-the-loop 동기:

    a) 승인 (Approval):

    • AI 에이전트의 특정 행동이나 결정을 사용자가 승인할 수 있습니다.
    • 중요한 결정이나 위험한 작업 전에 인간의 검토를 받을 수 있습니다.

    b) 디버깅 (Debugging):

    • 그래프 실행을 되돌려 문제를 재현하거나 피할 수 있습니다.
    • 오류가 발생한 지점을 정확히 파악하고 수정할 수 있습니다.

    c) 편집 (Editing):

    • 그래프의 상태를 수정할 수 있습니다.
    • 실행 중 필요한 변경사항을 적용할 수 있습니다.

  3. 브레이크포인트 (Breakpoints):

    • 그래프의 특정 단계에서 실행을 중단하는 기능입니다.
    • 사용자가 그래프의 상태를 검사하고 필요한 조치를 취할 수 있게 합니다.
    • 주로 승인 프로세스에 사용되며, 특정 노드 실행 전에 중단점을 설정할 수 있습니다.

브레이크포인트 (Breakpoints)

  • Interrupt Before 또는 Interrupt After를 통해 브래이크포인트를 설정할 수 있습니다.
    1. Interrupt Before: 특정 노드가 실행되기 전에 그래프 실행을 중단합니다.
    2. Interrupt After: 특정 노드 실행 후에 그래프를 멈춥니다.
  1. 스트리밍과의 연계:
    • 스트리밍을 통해 그래프 실행 중 출력을 실시간으로 볼 수 있습니다.
    • 브레이크포인트와 결합하여 더 세밀한 제어와 모니터링이 가능합니다.

예제

  • 예전에 모듈1에서 수행했던 실습 예제를 응용해보겠습니다:
    • multiply, add, divide 함수가 정의되어 있습니다. 이들은 간단한 산술 연산을 수행합니다.
    • 이 함수들은 tools 리스트에 포함되어 bind_tools를 통해 LLM에 바인딩됩니다.
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 langchain_openai import ChatOpenAI

def multiply(a: int, b: int) -> int:
    """a와 b를 곱합니다.

    인자:
        a: 첫 번째 정수
        b: 두 번째 정수
    """
    return a * b

# 이것은 도구가 될 것입니다
def add(a: int, b: int) -> int:
    """a와 b를 더합니다.

    인자:
        a: 첫 번째 정수
        b: 두 번째 정수
    """
    return a + b

def divide(a: int, b: int) -> float:
    """a를 b로 나눕니다.

    인자:
        a: 첫 번째 정수
        b: 두 번째 정수
    """
    return a / b

tools = [add, multiply, divide]
llm =  ChatOpenAI(model="gpt-4")
llm_with_tools = llm.bind_tools(tools)

  • 그래프 구조 및 컴파일 정의:
    • assistant 노드: LLM을 사용하여 사용자 입력에 응답합니다.
    • tools 노드: 실제 산술 연산을 수행합니다.
      • add_conditional_edges 메서드에 tools_condition을 사용하면, 자동으로 도구 호출 여부에 따른 조건부 라우팅이 설정됩니다.
      • 도구 호출이 있으면 “tools” 노드로 라우팅합니다.
      • 도구 호출이 없으면 “END”로 라우팅합니다.
    • interrupt_before=["tools"]: tools 노드 실행 전에 그래프가 중단됩니다.
    • MemorySaver를 사용하여 상태를 저장하고 관리합니다.
  • 그래프 실행 및 상태 확인:
    • “2와 3을 곱하세요”라는 초기 입력으로 그래프를 실행합니다.
    • graph.stream()을 사용하여 실행 과정을 스트리밍합니다.
    • 각 이벤트마다 메시지를 출력합니다.
    • graph.get_state()를 사용하여 현재 그래프의 상태를 가져옵니다.
    • state.next를 통해 다음에 실행될 노드를 확인할 수 있습니다.

  • 중단점(breakpoint): 중단점을 설정하면 그래프가 특정 노드에서 실행을 멈추고, 그 시점의 상태를 저장합니다. 사용자는 이 시점에서 그래프의 상태를 조회하거나, 입력을 받아 그래프의 실행을 계속할지 결정할 수 있습니다.

  • 재실행 (None을 사용): 중단된 상태에서 그래프를 None으로 호출하면, 가장 최근 체크포인트에서 그래프를 이어서 실행합니다. 이는 그래프를 중단점에서 다시 시작하는 매우 간단하고 강력한 방법입니다.

예제

  • 사용자 승인과 조건부 실행: 예를 들어, 사용자가 도구 호출을 승인하면, 그래프는 중단된 상태에서 도구 노드를 실행하고, 그 결과를 대화 모델에 전달하여 최종 응답을 생성합니다.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

# 입력
initial_input = {"messages": HumanMessage(content="2와 3을 곱하세요")}

# 스레드
thread = {"configurable": {"thread_id": "3"}}

# 첫 번째 중단까지 그래프 실행
for event in graph.stream(initial_input, thread, stream_mode="values"):
    event['messages'][-1].pretty_print()

# 사용자 피드백 받기
user_approval = input("도구를 호출하시겠습니까? (예/아니오): ")

# 승인 확인
if user_approval.lower() == "예":
    
    # 승인된 경우, 그래프 실행 계속
    for event in graph.stream(None, thread, stream_mode="values"):
        event['messages'][-1].pretty_print()
        
else:
    print("사용자에 의해 작업이 취소되었습니다.")

Lesson 3: Editing State and Human Feedback

개요

LangGraph는 중단점(Breakpoints)을 활용하여 그래프의 상태를 관리하고, 특히 사람이 개입하는 흐름에서 상태를 수정하거나 확인할 수 있는 과정을 보여줍니다.

주요 개념

중단점(Breakpoints) 설정

  • 중단점의 개념: 그래프의 특정 노드에서 실행을 일시 중단하고, 사용자의 개입을 기다리거나 상태를 수정하는 기회를 제공하는 메커니즘입니다. 아래의 목적으로 사용됩니다.

    1. 승인(Approval): 사용자가 특정 도구 호출을 승인하거나 거절할 수 있습니다. (Lesson2 소개)
    2. 디버깅(Debugging): 그래프가 중단된 상태에서 현재 상태를 검토하고, 문제를 해결할 수 있습니다. (Lesson2 소개)
    3. 편집(Editing): 그래프가 중단된 시점에서 상태를 수정하여, 원하는 대로 진행되도록 조정할 수 있습니다.

코드 예시

  • 위 코드를 수행하면 assistant를 수행하기 이전에 interupption이 발생하게 됩니다. (interrupt_before=["assistant"])

상태 업데이트 (graph.update_state)

1
2
3
4

graph.update_state(
    thread,
    {"messages": [HumanMessage(content="아니요, 실제로 3과 3을 곱하세요!")]}
)

  • 상태 업데이트: 이 코드는 현재 그래프의 상태를 업데이트하는 예시입니다.

    • thread는 해당 그래프 실행의 고유 ID로, 어떤 상태를 업데이트할 것인지를 식별하는 데 사용됩니다.

  • MessagesState: 메시지 상태를 관리하는 LangGraph의 내장된 클래스입니다. 이 클래스는 메시지들의 상태를 추적하고 업데이트할 때 사용됩니다.

    • Graph.update_state() 호출할 때, MessagesState는 메시지를 추가하거나 업데이트하는 작업을 내부적으로 처리하며, 이 과정에서 리듀서(reducer) 역할을 수행합니다.

      • 메시지 저장: messages 키를 통해 메시지 상태를 유지 및 관리합니다.
      • 상태 업데이트: update_state() 메서드를 사용하여 새로운 메시지를 추가하거나 기존 메시지를 수정할 때, MessagesState 클래스는 이를 자동으로 처리합니다.
      • 추가/덮어쓰기 로직: 메시지가 추가될 때 id가 없으면 새로운 메시지를 리스트에 추가하고, id가 있으면 해당 메시지를 덮어쓰는 방식으로 동작합니다.

💡(참고) graph.update_state() 메서드를 호출하면 기본적으로 다음과 같은 동작이 수행됩니다:

  • “messages” 키에 대한 기본 리듀서 함수가 실행됩니다. 이 리듀서는 다음과 같이 동작합니다:
    • 새 메시지의 ID가 기존 메시지와 일치하면 해당 메시지를 대체합니다.
    • ID가 일치하지 않으면 새 메시지를 기존 메시지 목록의 끝에 추가합니다.

상태 업데이트 후 그래프 재실행

  • graph.stream(None, thread, stream_mode=”values”): 그래프를 재실행하면서, 중단되었던 지점에서 상태를 확인하고, 상태가 업데이트된 후 남은 노드를 실행합니다.

  • 재실행 과정:

    1. 그래프 상태를 업데이트하고, None을 전달하여 중단된 지점에서 다시 실행을 계속합니다.
    2. 그래프가 진행되면서 최종 도구 호출 및 AI 모델 응답을 수행하게 됩니다.

Dummy Node와 피드백 적용

  • 이 기법은 실시간 사용자 개입을 통해, LLM과의 상호작용을 보다 동적으로 만들 수 있는 방법입니다.
  • 사용자는 특정 노드에서 실행 중단 후 상태를 수정할 수 있으며, 그 이후로도 그래프가 자연스럽게 계속 실행됩니다.

  • 이를 통해 더 복잡한 작업 흐름에서 인간 피드백을 반영할 수 있습니다.

    • Dummy Node란?: Dummy Node는 실제로는 아무 작업도 하지 않지만, 특정 시점에서 사용자의 피드백을 받을 수 있도록 중단점을 만들어 주는 가상의 노드입니다.

      • 이 노드는 실질적인 연산이나 상태 변경을 하지 않고, 그래프를 일시 정지하여 사용자가 개입할 수 있는 지점을 만들어 주는 역할을 합니다.

    • 피드백 적용: Dummy Node는 사용자의 피드백을 받을 수 있는 구간을 제공하며, 그래프 실행을 잠시 중단합니다. 이후 사용자가 상태 업데이트를 요청하면, 그 피드백을 받아 그래프 상태에 반영합니다.

      • 사용자의 입력(피드백)은 그래프 상태에 직접 주입되어, 그 후에 이어지는 노드의 실행에 영향을 미칩니다. 이를 통해 사용자가 그래프 실행 과정에 동적으로 개입할 수 있습니다.

  • 그래프 정의 : 위 그래프에는 3개의 주요 노드가 있습니다:

    1. human_feedback: 사용자가 피드백을 제공할 수 있는 가상의 노드(Dummy Node)입니다.

      • 이 노드에서는 아무 작업도 하지 않지만, 사용자 피드백을 받을 지점을 만듭니다.
    2. assistant: 사용자 입력과 LLM을 이용해 도구 호출 등의 연산을 수행하는 노드입니다.

    3. tools: 도구 호출을 담당하는 노드로, 실제 연산(예: 덧셈, 곱셈)을 수행합니다.

      1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

      # 시스템 메시지
sys_msg = SystemMessage(content="당신은 일련의 입력에 대해 산술 연산을 수행하는 도움이 되는 어시스턴트입니다.")
     
# 중단되어야 하는 무작동 노드
def human_feedback(state: MessagesState):
    pass
     
# 어시스턴트 노드
def assistant(state: MessagesState):
   return {"messages": [llm_with_tools.invoke([sys_msg] + state["messages"])]}
     
# 그래프 빌드
builder = StateGraph(MessagesState)
     
builder.add_node("assistant", assistant)
builder.add_node("tools", ToolNode(tools))
builder.add_node("human_feedback", human_feedback)

  1. 중단점 정의

    • 이 코드에서는 중단점(interrupt)을 human_feedback 노드에 지정했습니다. 즉, 이 노드에 도달할 때 그래프 실행이 멈추고 사용자의 피드백을 기다리게 됩니다.
    • interrupt_before=["human_feedback"] 부분이 이 역할을 담당합니다. 
      1
2

      # 그래프를 human_feedback 노드 전에 중단
graph = builder.compile(interrupt_before=["human_feedback"], checkpointer=memory)

  2. 실행 중 피드백 받기

    • 그래프는 처음에 사용자 입력을 받아 human_feedback 노드까지 실행된 후 멈춥니다.
    • 그 후 사용자의 피드백을 기다리며, input()을 통해 피드백을 받습니다.
    • 사용자가 입력한 피드백은 graph.update_state() 메서드를 통해 human_feedback 노드의 상태에 반영됩니다. ```

      사용자로부터 피드백 받기

      user_input = input(“상태를 어떻게 업데이트하고 싶은지 말씀해 주세요: “)

    human_feedback 노드인 것처럼 상태를 업데이트

    graph.update_state(thread, {“messages”: user_input}, as_node=”human_feedback”) ```

  1. 상태 업데이트 및 실행 재개

    • 피드백을 받은 후, graph.update_state()를 통해 해당 피드백을 상태에 반영한 뒤 그래프 실행을 계속합니다.
    • 사용자의 피드백에 따라 이후의 도구 호출이나 어시스턴트 응답이 달라질 수 있습니다. 
      1
2
3

      # 그래프 실행 계속
for event in graph.stream(None, thread, stream_mode="values"):
  event["messages"][-1].pretty_print()

Lesson 4: Dynamic Breakpoints

개요

  • 동적 중단점(Dynamic Breakpoints)은 그래프 실행 중 특정 조건에 따라 자동으로 발생하는 중단점입니다.
    • 즉, 실행되는 중간에 특정 상태나 조건을 감지하여 자동으로 멈추고, 이후 상태를 수정하거나 추가적인 입력을 받는 형태로 이어질 수 있습니다.
    • 동적 중단점은 특히 입력이 예상치 못한 값을 가질 때, 상태 업데이트가 필요할 때, 또는 특정 조건이 충족될 때 매우 유용합니다.

주요 개념

동적 중단점(Dynamic Breakpoints)의 목표

  • 일반적인 중단점은 그래프가 특정 노드에 도달할 때 수동으로 설정되지만, 동적 중단점은 그래프가 스스로 특정 조건에 따라 중단하도록 할 수 있습니다.
  • 이를 통해 그래프는 내부에서 스스로 상태를 점검하고 필요시 중단하여, 상태 업데이트 또는 수정이 가능하게 만듭니다.

동적 중단점(Dynamic Breakpoints)의 이점

  • 조건부로 중단: 개발자가 정의한 특정 로직에 따라 조건부로 중단이 가능합니다.

    • 예를 들어, 입력값이 너무 길거나 특정 기준을 초과할 경우 자동으로 중단됩니다.

  • 중단 이유 전달: 중단점이 발생한 이유를 사용자에게 명확히 전달할 수 있습니다.

    • NodeInterrupt에 전달할 정보를 포함시켜 왜 중단이 발생했는지를 설명할 수 있습니다.

코드 실습

1. 그래프 설정

  • 먼저, StateGraph를 사용하여 간단한 3단계 그래프를 설정합니다.
    • 이 그래프는 step_1, step_2, step_3으로 구성되며, step_2에서 입력값의 길이가 5자 이상일 경우 자동으로 중단시키는 NodeInterrupt를 사용합니다.
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

from typing_extensions import TypedDict
from langgraph.checkpoint.memory import MemorySaver
from langgraph.errors import NodeInterrupt
from langgraph.graph import START, END, StateGraph

# 상태 정의
class State(TypedDict):
    input: str

# 1단계: 상태를 그대로 출력
def step_1(state: State) -> State:
    print("---1단계---")
    return state

# 2단계: 입력의 길이가 5자를 넘으면 중단 발생
def step_2(state: State) -> State:
    if len(state['input']) > 5:
        raise NodeInterrupt(f"5자보다 긴 입력을 받았습니다: {state['input']}")
    print("---2단계---")
    return state

# 3단계: 상태를 그대로 출력
def step_3(state: State) -> State:
    print("---3단계---")
    return state

# 그래프 설정
builder = StateGraph(State)
builder.add_node("step_1", step_1)
builder.add_node("step_2", step_2)
builder.add_node("step_3", step_3)
builder.add_edge(START, "step_1")
builder.add_edge("step_1", "step_2")
builder.add_edge("step_2", "step_3")
builder.add_edge("step_3", END)

# 메모리 설정
memory = MemorySaver()

# 그래프 컴파일
graph = builder.compile(checkpointer=memory)

2. 그래프 실행 및 중단점 발생

입력값이 5자보다 길면 그래프는 step_2에서 중단됩니다.

1
2
3
4
5
6

initial_input = {"input": "hello world"}
thread_config = {"configurable": {"thread_id": "1"}}

# Run the graph until the first interruption
for event in graph.stream(initial_input, thread_config, stream_mode="values"):
    print(event)

실행 결과:

1
2
3

{'input': 'hello world'}
---1단계---
{'input': 'hello world'}

이 시점에서 그래프는 step_2에서 멈추게 되고, 상태를 확인할 수 있습니다.

현재 상태를 조금 더 명확하게 설명하자면, NodeInterrupt가 발생했기 때문에 노드와 엣지 간의 정확한 위치를 구분하는 것이 중요합니다.

  • 노드 위치 관점에서: 실행 흐름은 step_1을 완료하고 step_2를 실행하는 도중에 중단되었습니다. 이는 아직 step_2의 실행이 완료되지 않았음을 의미하므로, 현재는 step_2에 도달한 상태입니다.
  • 엣지 위치 관점에서: step_1에서 step_2로의 전이가 이루어졌으나, step_2가 완료되지 않았으므로 1-2 엣지“에서 step_2로 넘어가는 중에 멈춘 상태라고도 볼 수 있습니다.

3. 상태 확인 및 수정

그래프가 중단된 상태에서 다음 실행할 노드가 step_2임을 확인합니다.

1
2

state = graph.get_state(thread_config)
print(state.next)  # ('step_2',)

그리고 중단점이 발생한 이유도 확인할 수 있습니다.

1

print(state.tasks)

1

(PregelTask(id='ec5598b4-c9cc-ce9e-8387-5f27e35742a5', name='step_2', path=('__pregel_pull', 'step_2'), error=None, interrupts=(Interrupt(value='5자보다 긴 입력을 받았습니다: hello world', when='during'),), state=None),)

4. 상태 업데이트 및 그래프 재실행

상태를 변경하지 않으면 같은 노드에서 계속 멈춰있게 됩니다.

상태를 “Hello”으로 변경하여 그래프를 재실행합니다. 변경 후에는 그래프의 끝까지 정상적으로 실행되는 것을 확인할 수 있습니다.

1
2
3
4
5
6
7
8
9

# 상태 업데이트: 입력을 "Hello"으로 변경
graph.update_state(
    thread_config,
    {"input": "Hello"},
)

# 그래프 재실행
for event in graph.stream(None, thread_config, stream_mode="values"):
    print(event)

업데이트 후 실행 결과:

1
2
3
4
5

{'input': 'Hello'}
---2단계---
{'input': 'Hello'}
---3단계---
{'input': 'Hello'}

✨ 동적 중단점의 핵심 개념

  1. 자동 중단점 설정: 그래프는 특정 조건(예: 입력 길이 초과)을 만족할 때 자동으로 멈추고, 중단점을 발생시킵니다.

  2. 상태 확인 및 업데이트: 그래프가 중단되면 현재 상태를 확인하고, 필요시 상태를 업데이트할 수 있습니다.

  3. 그래프 재실행: 상태가 업데이트된 후, 그래프는 중단된 시점부터 다시 실행됩니다.

  4. 유연한 워크플로우 구성: 동적 중단점을 사용하면, 사용자의 승인, 디버깅, 상태 수정 등을 유연하게 처리할 수 있습니다.

Lesson 5: Time Travel

개요

Time Travel은 그래프의 이전 상태로 돌아가거나, 그 상태를 재생하여 과거의 실행 흐름을 살펴보고 문제를 분석하는 데 유용한 기능입니다.

  • 이 기능은 그래프가 실행된 모든 상태를 기록하고, 그 중 어느 지점에서든 다시 시작할 수 있도록 해줍니다.
  • Time Travel은 디버깅, 성능 최적화, 여러 경로 탐색 등에 도움을 줍니다.
    • 재생(Replay)과 분기(Fork)의 두 가지 핵심 개념을 통해 그래프의 실행 흐름을 수정하거나 다양한 시나리오를 실험할 수 있습니다.

주요 개념

  1. Replay(재생):

    • 특정 체크포인트에서 그래프를 다시 재생하여 과거의 실행을 복원합니다.
    • 그래프의 모든 상태가 저장되기 때문에 이전에 완료된 작업을 반복 실행할 필요 없이 그 상태를 그대로 재생할 수 있습니다.
    • 이는 문제를 재현하거나 분석할 때 매우 유용합니다.

  2. Fork(분기):

    • 특정 체크포인트에서 상태를 수정하여 새로운 실행 흐름을 탐색할 수 있습니다.
    • 이 기능은 그래프의 특정 지점에서 다른 경로로 진행하도록 상태를 업데이트한 후, 그 경로에 따른 실행 결과를 확인하는 방식으로 활용됩니다.
    • Fork는 새로운 시나리오를 테스트하거나, 예상치 못한 입력에 따른 흐름을 실험할 때 유용합니다.

Time Travel의 주요 단계

0. Agent Define

  • 이전과 마찬가지로 agent를 정의해보겠습니다. 계속해서 사용하던 agent이므로 자세한 설명은 생략하겠습니다.
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
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72

from langchain_openai import ChatOpenAI

def multiply(a: int, b: int) -> int:
    """a와 b를 곱합니다.

    인자:
        a: 첫 번째 정수
        b: 두 번째 정수
    """
    return a * b

# 이것은 도구가 될 것입니다
def add(a: int, b: int) -> int:
    """a와 b를 더합니다.

    인자:
        a: 첫 번째 정수
        b: 두 번째 정수
    """
    return a + b

def divide(a: int, b: int) -> float:
    """a를 b로 나눕니다.

    인자:
        a: 첫 번째 정수
        b: 두 번째 정수
    """
    return a / b

tools = [add, multiply, divide]
llm = ChatOpenAI(model="gpt-4")
llm_with_tools = llm.bind_tools(tools)

from IPython.display import Image, display

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import MessagesState
from langgraph.graph import START, END, StateGraph
from langgraph.prebuilt import tools_condition, ToolNode

from langchain_core.messages import AIMessage, HumanMessage, SystemMessage

# 시스템 메시지
sys_msg = SystemMessage(content="당신은 일련의 입력에 대해 산술 연산을 수행하는 도움이 되는 어시스턴트입니다.")

# 노드
def assistant(state: MessagesState):
   return {"messages": [llm_with_tools.invoke([sys_msg] + state["messages"])]}

# 그래프
builder = StateGraph(MessagesState)

# 노드 정의: 이들이 작업을 수행합니다
builder.add_node("assistant", assistant)
builder.add_node("tools", ToolNode(tools))

# 엣지 정의: 이들이 제어 흐름을 결정합니다
builder.add_edge(START, "assistant")
builder.add_conditional_edges(
    "assistant",
    # 어시스턴트의 최신 메시지(결과)가 도구 호출이면 -> tools_condition이 tools로 라우팅합니다
    # 어시스턴트의 최신 메시지(결과)가 도구 호출이 아니면 -> tools_condition이 END로 라우팅합니다
    tools_condition,
)
builder.add_edge("tools", "assistant")

memory = MemorySaver()
graph = builder.compile(checkpointer=MemorySaver())

# 표시
display(Image(graph.get_graph(xray=True).draw_mermaid_png()))

1. 현재 상태 조회

  • LangGraph에서 get_state 메서드를 호출하여 그래프의 현재 상태를 확인할 수 있습니다. 이는 현재 실행 중이거나 완료된 그래프의 상태를 가져옵니다.
1
2

state = graph.get_state({'configurable': {'thread_id': '1'}})
print(state)
  • 이를 통해 현재 그래프의 모든 메시지, 메타데이터, 체크포인트 ID 등을 확인할 수 있습니다.

2. 상태 히스토리 조회

  • get_state_history 메서드를 사용하면 그래프 실행 동안 발생한 모든 상태의 히스토리를 조회할 수 있습니다. 이 히스토리는 각 단계에서 그래프가 어떤 상태를 가졌는지 보여줍니다.
1
2
3

history = graph.get_state_history(thread)
for state in history:
    print(state)
  • 이 히스토리는 각 상태의 체크포인트를 기록하며, 개발자는 이 히스토리를 통해 이전 상태를 재생하거나 수정할 수 있습니다.
1
2
3
4

history = graph.get_state_history(thread)
for i, state in enumerate(history,1):
    print(i)
    print(state)

(참고) 중간 정리

  • 체크포인트는 여행 중 찍은 사진 한 장이고, 스레드는 그 여행의 전체 사진 앨범이라고 할 수 있습니다.

  • Graph: 그래프는 노드와 엣지로 구성되어 있는 컨트롤 플로우입니다.

    • 각 노드는 어시스턴트, 도구 사용 등의 특정 단계(상태, state)를 나타냅니다.

  • Checkpoints: 체크포인터(checkpointer)가 각 노드의 상태를 체크포인트로 기록합니다.

    • 체크포인트에는 상태 정보와 함께 ‘다음에 어디로 갈지‘와 같은 메타데이터도 포함됩니다.
  • Thread: 스레드(thread)는 체크포인트의 모음입니다.
  • StateSnapshot: 상태 스냅샷(StateSnapshot)은 체크포인트의 데이터 타입니다.
  • Graph.get_state(): 현재(가장 최근의) 체크포인트를 반환하는 함수입니다.
  • Graph.get_state_history(): 모든 체크포인트의 리스트를 반환하여, 상태 변화의 히스토리를 제공합니다.

3. Replay (다시 재생)

  • Replay는 그래프 실행이 완료된 후, 특정 체크포인트에서 그래프를 다시 실행하지 않고 그 상태를 그대로 재생하는 기능입니다.
    • 이는 과거 상태에서 문제가 발생한 지점을 재현하거나, 분석을 위한 디버깅 목적으로 사용됩니다.

  1. Graph.stream(None,{thread_id}): 이 함수는 현재 상태에서 그래프를 실행합니다.
  • 결과값으로 StateSnapshot()을 반환합니다.
  1. Graph.stream(None,{checkpoint_id, thread_id}): 이 함수는 특정 체크포인트에서 그래프를 다시 실행(re-play)합니다.
  • 주목할 점은 실제로 노드들이 실행되지 않는다는 것입니다.
1
2
3

# 특정 체크포인트 ID를 기반으로 재생
checkpoint_id = history[-2]['config']['checkpoint_id']
graph.stream(None, {'configurable': {'thread_id': '1', 'checkpoint_id': checkpoint_id}}, stream_mode="values")
  • Replay는 이미 실행된 체크포인트를 재생하기 때문에, 그래프의 노드를 다시 실행하지 않고 기록된 상태를 빠르게 복원할 수 있습니다.

4. Forking (분기)

  • Forking은 특정 체크포인트에서 상태를 수정하고 새로운 실행 흐름을 만드는 기능입니다.
    • 이를 통해 사용자는 그래프의 상태를 변경하여 다른 경로를 탐색하거나 실험할 수 있습니다.
    • 예를 들어, 특정 단계에서 새로운 입력을 주어 다른 결과가 나오는지 확인할 수 있습니다.

  • ① 현재 상태에서 포킹

    • 우리는 LangGraph의 Graph.update_state 메서드를 사용하여 현재 상태를 포킹할 수 있습니다.

    • 예를 들어, 다음과 같은 코드를 통해 상태를 업데이트할 수 있습니다:

      1

      Graph.update_state({thread_id}, {state: "I like cold brew"})
    • 위 코드를 실행하면 thread_id의 현재 그래프 상태가 업데이트되면서 새로운 체크포인트가 생성됩니다. 이를 “현재 상태에서 포킹”한다고 표현할 수 있습니다.
    • 상태가 기록되고 새로운 체크포인트가 생성되면, 포크된 새로운 상태 스냅샷(StateSnapshot)이 만들어집니다.
    • 위 그림의 첫번째 예시에서는 “I heart Langchain” ⟶ “I like cold brew“라는 상태로 새로운 체크포인트가 만들어집니다.

  • ② 과거 체크포인트에서 포킹

    • 포킹의 또 다른 방식은 과거의 특정 체크포인트에서 새로운 상태를 포킹하는 것입니다.
    • 이를 위해, 우리는 특정 checkpoint_id를 제공하여 그 시점의 상태를 기반으로 포크할 수 있습니다.

    • 예를 들어, 다음과 같은 코드로 특정 체크포인트에서 상태를 포킹할 수 있습니다:

      1

      Graph.update_state({checkpoint_id}, thread_id, {state: "I like"})
    • 이 경우, 이전에 생성된 특정 체크포인트에서 상태가 업데이트되고 새로운 체크포인트가 생성됩니다.
    • 과거의 상태를 기반으로 새로운 상태를 만들어 포크하는 과정이라고 할 수 있습니다.
    • 이를 통해 특정 시점으로 돌아가 새로운 실행 흐름을 만들 수 있습니다.

💬 Re-play vs Re-Execute

  • 재생(Replaying): 이미 실행된 체크포인트를 다시 실행할 때, 그래프는 그 상태가 이미 실행되었음을 인지하고 재생합니다. 즉, 같은 상태를 그대로 유지한 채 실행이 반복됩니다. 예를 들어, 이전에 실행된 상태 스냅샷을 전달하면, 그 상태에 따라 그래프가 이미 실행된 것처럼 동작하게 됩니다.
  • 재실행(Re-executing): 반면, 새로운 체크포인트를 통해 그래프를 실행하면, 그래프는 이전에 이 상태가 실행된 적이 없다는 것을 인식하고 새롭게 실행을 시작합니다. 이를 재실행이라고 부릅니다. 포크된 체크포인트가 처음 실행되는 경우, 새로운 상태를 바탕으로 그래프가 다시 실행됩니다.

  • 상태 덧붙임(Appending)과 덮어쓰기(Overwriting)

    • 상태 업데이트 시 중요한 부분은 덧붙임(Appending)과 덮어쓰기(Overwriting)입니다.
    • 예를 들어, 메시지와 같은 상태는 기본적으로 덧붙여지지만, 특정 id를 전달하면 해당 메시지가 덮어쓰기 됩니다.

    • 이는 add messages reducer와 같은 방식에서 발생합니다.

      📋 주의

      messages에 대한 리듀서가 어떻게 작동하는지 기억하세요!

      • 메시지 ID를 제공하지 않으면 뒤에 추가(append)합니다.
      • 상태에 추가하는 대신 메시지를 덮어쓰기(overwrite) 위해 메시지 ID를 제공해야합니다!
      1

      Graph.update_state({checkpoint_id}, thread_id, {state: "I like", id: "message_id"})
    • 위 코드를 수행하면 새로운 메시지가 추가되지 않고 기존 메시지가 덮어씌워지며 업데이트됩니다.
    • 메시지 상태를 어떻게 관리할지에 따라 덧붙일지 또는 덮어쓸지를 결정할 수 있습니다.

  • 상태 히스토리 관리

    • 각 상태 업데이트는 그래프의 히스토리에서 새로운 체크포인트를 생성합니다. 예를 들어, “I like cold brew”라는 상태를 업데이트하고 새로운 체크포인트가 생성되었다면, 이 체크포인트는 히스토리에 기록됩니다.
    • 여러 번 상태를 업데이트하면, 각 상태는 고유한 체크포인트로 관리되고, 필요한 경우 과거의 특정 체크포인트로 돌아가 새로운 포크를 만들 수 있습니다.
    • 히스토리에서 새로운 체크포인트를 추가하는 방식으로 상태를 관리할 수 있으며, 그래프의 각 스레드는 여러 체크포인트로 구성됩니다. 이렇게 하면 매우 유연하게 상태를 관리할 수 있습니다.

  • 포크된 체크포인트로 재실행

    • 포크된 체크포인트를 통해 그래프를 다시 실행할 때, LangGraph는 해당 체크포인트가 이전에 실행된 적이 없는지 확인하고 재실행합니다.
    • 예를 들어, 새로운 checkpoint_id로 재실행을 하면, 과거 상태에서 새로운 실행 흐름이 시작됩니다.

    • 다음과 같이 새로운 포크된 체크포인트를 제공하여 재실행할 수 있습니다:

      1

      Graph.stream(None, {checkpoint_id}, thread_id)
    • 이 코드를 통해 포크된 체크포인트로 그래프가 재실행되고, 그에 따라 새로운 상태가 생성됩니다.

이 강의 내용은 Langraph의 다양한 기능과 Human-in-the-Loop의 활용 방안을 중심으로 구성되어 있으며, 각각의 기능은 실제 워크플로우에서 매우 유용하게 사용됩니다. 각 레슨의 세부 내용을 철저히 이해하면 Langraph를 이용한 고급 작업 흐름을 쉽게 구현할 수 있을 것입니다.

CATALOG
    FEATURED TAGS
    개념정리 딥러닝 AI Agent OpenAI Python Paper Review NLP 머신러닝 Conference 꿀팁 강의노트 Tools ChatGPT 환경설정 Prompt Engineering Computer Vision IT지식 트렌드 Transformer Linux RAG LangChain Google 시계열 Vibe Coding NVIDIA PyTorch 수학 비지도학습 이상탐지 CES Claude Gemini Lovable Anthropic 차원축소 Git CIO CNN EXAONE Pandas Visualization LangGraph LG State Space Model Graph Repo Review SK AI Summit Snippet Sora 도서리뷰 Distillation Network Qwen 알고리즘 데이터분석 MCP Ollama NumPy 강화학습 Splunk Velog
    FRIEND SITES