[langgraph] 그래프 상태를 수정하는 방법

LangGraph 공식문서를 번역한 내용입니다. 필요한 경우 부연 설명을 추가하였고 이해하기 쉽게 예제를 일부 변경하였습니다. 문제가 되면 삭제하겠습니다.

https://langchain-ai.github.io/langgraph/how-tos/human_in_the_loop/edit-graph-state/

사람이 개입하는 (HIL) 상호작용은 에이전트 시스템에서 매우 중요하다. 그래프 상태를 수동으로 업데이트하는 것은 일반적인 HIL 상호작용 패턴으로, 사람이 행동을 수정할 수 있도록 한다 (예: 어떤 도구가 호출되는지 또는 어떻게 호출되는지).

우리는 LangGraph에서 이를 중단점을 사용하여 구현할 수 있다. 중단점은 특정 단계 전에 그래프 실행을 중단할 수 있게 해주며, 이 중단점에서 그래프 상태를 수동으로 업데이트한 후 그 지점에서 실행을 재개하여 계속 진행할 수 있다.

준비

우선, 필요한 패키지를 설치하자.

pip install langgraph langchain_openai

기본적인 사용법

매우 기본적인 사용법을 살펴보자.

아래에서 우리는 세 가지 작업을 수행한다.

1. 지정된 단계(노드) 전에 interrupt_before를 사용하여 중단점을 지정한다.
2. 이 노드까지 그래프의 상태를 저장할 수 있도록 체크포인터를 설정한다.
3. .update_state를 사용하여 그래프의 상태를 업데이트한다.

from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from IPython.display import Image, display


class State(TypedDict):
    input: str


def step_1(state):
    print("---Step 1---")
    pass


def step_2(state):
    print("---Step 2---")
    pass


def step_3(state):
    print("---Step 3---")
    pass


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, interrupt_before=["step_2"])

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

initial_input = {"input": "안녕~"}

thread = {"configurable": {"thread_id": "1"}}

for event in graph.stream(initial_input, thread, stream_mode="values"):
    print(event)

{'input': '안녕~'}
---Step 1---

이제 그래프 상태를 수동으로 업데이트 한다.

print("Current state!")
print(graph.get_state(thread).values)

graph.update_state(thread, {"input": "안녕~ 수정했어!"})

print("---\n---\nUpdated state!")
print(graph.get_state(thread).values)

Current state!
{'input': 'hello world'}
---
---
Updated state!
{'input': '안녕~ 수정했어!'}

for event in graph.stream(None, thread, stream_mode="values"):
    print(event)

---Step 2---
---Step 3---

에이전트

에이전트의 컨텍스트에서 상태 업데이트는 도구 호출을 수정하는 데 유용하다.

이를 보여주기 위해 도구 호출을 수행하는 상대적으로 간단한 ReAct 스타일의 에이전트를 만든다.

우리는 Openai의 모델과 가짜 도구(데모 목적으로만 사용)를 사용할 것이다.

from dotenv import load_dotenv
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.graph import MessagesState, START, END, StateGraph
from langgraph.prebuilt import ToolNode
from langgraph.checkpoint.memory import MemorySaver

load_dotenv()


@tool
def search(query: str):
    """Call to surf the web."""
    return ["서울 날씨 좋아요~ 😈."]


tools = [search]
tool_node = ToolNode(tools)


model = ChatOpenAI(model="gpt-4o-mini")
model = model.bind_tools(tools)


def should_continue(state):
    messages = state["messages"]
    last_message = messages[-1]
    if not last_message.tool_calls:
        return "end"
    else:
        return "continue"


def call_model(state):
    messages = state["messages"]
    response = model.invoke(messages)
    return {"messages": [response]}


workflow = StateGraph(MessagesState)

workflow.add_node("agent", call_model)
workflow.add_node("action", tool_node)

workflow.add_edge(START, "agent")

workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        "continue": "action",
        "end": END,
    },
)

workflow.add_edge("action", "agent")

memory = MemorySaver()

app = workflow.compile(checkpointer=memory, interrupt_before=["action"])

에이전트와 상호작업

이제 에이전트와 대화할 수 있으며, 도구를 호출하기 전에 멈추는 것을 볼 수 있다.

from langchain_core.messages import HumanMessage

thread = {"configurable": {"thread_id": "3"}}
inputs = [HumanMessage(content="지금 서울 날씨 알려줘")]
for event in app.stream({"messages": inputs}, thread, stream_mode="values"):
    event["messages"][-1].pretty_print()

================================ Human Message =================================

지금 서울 날씨 알려줘
================================== Ai Message ==================================
Tool Calls:
  search (call_m5wLfqULa66KPkXFgfvFqnkH)
 Call ID: call_m5wLfqULa66KPkXFgfvFqnkH
  Args:
    query: 서울 날씨

수정

이제 상태를 적절하게 업데이트할 수 있다. 도구 호출을 "현재 서울 날씨"로 수정해보자.

current_state = app.get_state(thread)

last_message = current_state.values["messages"][-1]

last_message.tool_calls[0]["args"] = {"query": "현재 서울 날씨"}

result = app.update_state(thread, {"messages": last_message})
print(result)

{'configurable': {'thread_id': '3', 'checkpoint_ns': '', 'checkpoint_id': '1efae312-7efd-67e0-8002-916a45e4e732'}}

이제 애플리케이션의 현재 상태를 확인하여 제대로 업데이트되었는지 확인해보자.

current_state = app.get_state(thread).values["messages"][-1].tool_calls
print(current_state)

[{'name': 'search', 'args': {'query': '현재 서울 날씨'}, 'id': 'call_hKkMN0FTbvuXIIWd2nSmVi2Q', 'type': 'tool_call'}]

다시 시작

이제 입력 없이 에이전트를 다시 호출하여 계속 진행할 수 있다. 즉, 요청된 대로 도구를 실행한다. 로그에서 도구에 업데이트된 인자가 전달되는 것을 확인할 수 있다.

for event in app.stream(None, thread, stream_mode="values"):
    event["messages"][-1].pretty_print()

================================== Ai Message ==================================
Tool Calls:
  search (call_BE46Ju4sRRVDakN0sgYTzFFL)
 Call ID: call_BE46Ju4sRRVDakN0sgYTzFFL
  Args:
    query: 현재 서울 날씨
================================= Tool Message =================================
Name: search

["서울 날씨 좋아요~ 😈."]
================================== Ai Message ==================================

현재 서울의 날씨는 좋습니다! 추가적인 정보가 필요하시면 말씀해 주세요.