[langgraph] 사용자 입력을 대기하는 방법

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

https://langchain-ai.github.io/langgraph/how-tos/human_in_the_loop/wait-user-input/

사람이 개입하는(Human-in-the-loop, HIL) 상호작용은 에이전트 시스템에서 매우 중요하다. 사용자 입력을 기다리는 것은 일반적인 HIL 상호작용 패턴으로, 에이전트가 사용자에게 명확한 질문을 하고 입력을 기다린 후 계속 진행할 수 있게 한다.

LangGraph에서는 이를 중단점을 사용하여 구현할 수 있다. 중단점을 사용하면 그래프 실행을 특정 단계에서 중지시킬 수 있다. 이 중단점에서 사용자 입력을 기다리고, 입력을 받으면 그래프 상태에 추가한 후 계속 진행할 수 있다.

준비

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

pip install langgraph

기본적인 사용법

기본적인 사용 예를 살펴보자. 직관적인 접근 방식은 단순히 사용자 피드백을 받을 수 있는 human_feedback 노드를 생성하는 것이다. 이렇게 하면 그래프에서 특정 지점에 피드백 수집을 배치할 수 있다.

1. human_feedback 노드 앞에서 실행을 중단하도록 interrupt_before를 사용하여 중단점을 지정한다.
2. 이 노드까지 그래프의 상태를 저장할 수 있도록 체크포인터를 설정한다.
3. update_state를 사용하여 사용자 응답을 그래프 상태에 업데이트한다.

as_node 매개변수를 사용하여 이 상태 업데이트가 지정된 human_feedback 노드로 적용되도록 한다. 그러면 그래프는 human_feedback 노드가 실행된 것처럼 계속 실행된다.

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
    user_feedback: str


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


def human_feedback(state):
    print("---human_feedback---")
    pass


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


builder = StateGraph(State)
builder.add_node("step_1", step_1)
builder.add_node("human_feedback", human_feedback)
builder.add_node("step_3", step_3)

builder.add_edge(START, "step_1")
builder.add_edge("step_1", "human_feedback")
builder.add_edge("human_feedback", "step_3")
builder.add_edge("step_3", END)

memory = MemorySaver()

graph = builder.compile(checkpointer=memory, interrupt_before=["human_feedback"])

display(
    Image(
        graph.get_graph().draw_mermaid_png(
            output_file_path="how-to-wait-for-user-input.png"
        )
    )
)

중단점까지 실행한다. (human_feedback)

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

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

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

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

이제 사용자 입력으로 그래프 상태를 수동으로 업데이트할 수 있다.

try:
    user_input = input("피드백 주세요: ")
except:
    user_input = "go to step 3!"

# human_feedback 노드에 있는 것처럼 상태를 업데이트한다.
graph.update_state(thread, {"user_feedback": user_input}, as_node="human_feedback")

# 상태 확인
print("--State after update--")
print(graph.get_state(thread))

# human_feedback 다음에 3번 노드가 있는지 확인하기 위해 다음 노드를 확인한다.
print(graph.get_state(thread).next)

--State after update--
StateSnapshot(values={'input': '안녕~', 'user_feedback': 'good'}, next=('step_3',), config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1efae325-7eee-69d0-8002-6174a9f247e7'}}, metadata={'source': 'update', 'step': 2, 'writes': {'human_feedback': {'user_feedback': 'good'}}, 'parents': {}}, created_at='2024-11-29T09:14:31.864744+00:00', parent_config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1efae325-6993-6cfc-8001-f834af25dd2e'}}, tasks=(PregelTask(id='598c2074-d071-1ced-8280-6480e0131cc3', name='step_3', path=('__pregel_pull', 'step_3'), error=None, interrupts=(), state=None, result=None),))

('step_3',)

이제 중단점 후에 진행할 수 있습니다.

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

---Step 3---

피드백이 상태에 추가된 것을 확인할 수 있습니다.

print(graph.get_state(thread).values)

{'input': 'hello world', 'user_feedback': 'go to step 3!'}

에이전트

에이전트 컨텍스트에서 사용자 피드백을 기다리는 것은 특정 질문을 하기 위한 용도이다.

이를 위해, 도구 호출을 하는 간단한 ReAct 스타일의 에이전트를 만들어보자.

여기에서는 OpenAI의 모델과 가짜 도구(데모용)를 사용할 것이다.

from IPython.display import Image, display
from dotenv import load_dotenv
from langchain_core.tools import tool
from langgraph.graph import MessagesState, START
from langgraph.prebuilt import ToolNode

load_dotenv()


@tool
def search(query: str):
    """Call to surf the web."""
    return f"찾아봤습니다: {query}. 결과: 서울 날씨는 좋아요~ 😈."


tools = [search]
tool_node = ToolNode(tools)

from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o-mini")

from pydantic import BaseModel


class AskHuman(BaseModel):
    """Ask the human a question"""

    question: str


model = model.bind_tools(tools + [AskHuman])


def should_continue(state):
    messages = state["messages"]
    last_message = messages[-1]
    if not last_message.tool_calls:
        return "end"
    # 도구 호출이 사람에게 물어보는 것이면 해당 노드를 반환한다.
    # 여기에 로직을 추가하여 사람이 입력해야 하는 것이 있다는 것을 시스템에 알릴 수도 있다.
    # 예를 들어, 슬랙 메시지를 보내거나 기타 등등
    elif last_message.tool_calls[0]["name"] == "AskHuman":
        return "ask_human"
    else:
        return "continue"


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


# 사람에게 물어보는 가짜 노드를 정의한다.
def ask_human(state):
    pass


from langgraph.graph import END, StateGraph

workflow = StateGraph(MessagesState)

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

workflow.add_edge(START, "agent")

workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        # 'tools' 이라면 도구 노드를 호출한다.
        "continue": "action",
        # 사람에게 물어볼 수 있다.
        "ask_human": "ask_human",
        # 그렇지 않다면 종료한다.
        "end": END,
    },
)

# tools에서 agent로 가는 일반적인 엣지를 추가한다.
# 이것은 tools가 호출된 후 agent 노드가 다음에 호출된다는 것을 의미한다.
workflow.add_edge("action", "agent")

# 사람의 응답을 받은 후에는 다시 agent로 돌아간다.
workflow.add_edge("ask_human", "agent")

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()

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

display(
    Image(
        app.get_graph().draw_mermaid_png(
            output_file_path="how-to-wait-graph-state-with-agent.png"
        )
    )
)

에이전트와 상호작업

이제 에이전트와 대화할 수 있다. 사용자에게 위치를 물어본 후 날씨를 알려 달라고 요청하자.

이렇게 하면 먼저 ask_human 도구를 사용한 후, 정상적인 도구를 사용할 것이다.

from langchain_core.messages import HumanMessage

config = {"configurable": {"thread_id": "2"}}
input_message = HumanMessage(
    content="그들이 어디에 있는지 묻는 도구를 사용하고, 그곳의 날씨를 찾아줘."
)
for event in app.stream({"messages": [input_message]}, config, stream_mode="values"):
    event["messages"][-1].pretty_print()

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

그들이 어디에 있는지 묻는 도구를 사용하고, 그곳의 날씨를 찾아줘.
================================== Ai Message ==================================
Tool Calls:
  AskHuman (call_mXDWwekXUEk3Qg0ZJABiRERk)
 Call ID: call_mXDWwekXUEk3Qg0ZJABiRERk
  Args:
    question: 그들이 누구인지 알려줄 수 있나요?
('agent',)

이제 사용자로부터 응답을 받은 후 이 스레드를 업데이트하고, 다시 실행을 한다.

이것을 도구 호출로 처리하고 있기 때문에, 도구 호출의 응답처럼 상태를 업데이트해야 한다. 이를 위해서는 상태를 확인하여 도구 호출의 ID를 가져와야 한다.

tool_call_id = app.get_state(config).values["messages"][-1].tool_calls[0]["id"]

# id를 사용하여 도구 호출을 만들고 원하는 응답을 추가한다.
tool_message = [
    {"tool_call_id": tool_call_id, "type": "tool", "content": "서울"}
]

# 위의 코드는 아래와 동일하다.
# from langchain_core.messages import ToolMessage
# tool_message = [ToolMessage(tool_call_id=tool_call_id, content="서울")]

# 이제 상태를 업데이트한다.
# 우리는 `as_node="ask_human"`을 지정하고 있다.
# 이것은 이 노드로 이 업데이트를 적용하게 만들 것이다.
# 이후에 계속 정상적으로 진행되도록 만들 것이다.

app.update_state(config, {"messages": tool_message}, as_node="ask_human")

# 상태를 확인할 수 있다.
# 상태에는 현재 `agent` 노드가 다음에 있음을 볼 수 있다.
# 이것은 그래프를 어떻게 정의했는지에 따라 결정된다.
# 우리는 방금 트리거한 `ask_human` 노드 이후에 `agent` 노드로 가는 엣지가 있다.
print(app.get_state(config).next)

('agent',)

이제 에이전트에게 계속 진행하도록 지시할 수 있다. 추가 입력이 필요 없기 때문에, 그래프에 None을 입력으로 전달하면 된다.

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

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

찾아봤습니다: {query}. 결과: 서울 날씨는 좋아요~ 😈.
================================== Ai Message ==================================

서울의 날씨는 좋습니다! 다른 도움이 필요하시면 말씀해 주세요.