LangGraph 공식문서를 번역한 내용입니다. 필요한 경우 부연 설명을 추가하였고 이해하기 쉽게 예제를 일부 변경하였습니다. 문제가 되면 삭제하겠습니다.
https://langchain-ai.github.io/langgraph/how-tos/human_in_the_loop/breakpoints/
사람이 개입하는(Human-in-the-loop, HIL) 상호작용은 에이전트 시스템에서 매우 중요하다. 중단 지점은 일반적인 HIL 상호작용 패턴으로, 그래프가 특정 단계에서 멈추고 진행하기 전에 사람의 승인을 요청하도록 한다(예: 민감한 작업의 경우).
중단 지점은 LangGraph 체크포인트를 기반으로 만들어진다. 체크포인트는 각 노드 실행 후 그래프의 상태를 저장하며, 그래프 상태를 보존하는 스레드에 저장된다. 이를 통해 그래프 실행을 특정 지점에서 일시 정지하고, 사람의 승인을 기다린 후 마지막 체크포인트에서 실행을 재개할 수 있다.
준비
우선, 필요한 패키지를 설치하자.
pip install langgraph langchain_openai기본적인 사용법
아주 기본적인 사용법을 살펴보자.
아래에서 두 가지 작업을 수행한다.
interrupt_before를 사용하여 지정된 단계에서 브레이크포인트를 설정한다.- 체크포인터를 설정하여 그래프의 상태를 저장한다.
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_3"])
display(
Image(
graph.get_graph().draw_mermaid_png(output_file_path="how-to-add-breakpoint.png")
)
)
체크포인터에 사용할 스레드 ID를 생성한다.
interrupt_before로 정의된 대로 3단계까지 실행한다.
사용자의 입력 또는 승인이 이루어진 후, None을 사용하여 그래프를 호출함으로써 실행을 재개한다.
initial_input = {"input": "안녕~"}
thread = {"configurable": {"thread_id": "1"}}
# 첫 번째 중단까지 그래프 실행
for event in graph.stream(initial_input, thread, stream_mode="values"):
print(event)
try:
user_approval = input("3단계로 가시겠습니까? (yes/no): ")
except:
user_approval = "yes"
if user_approval.lower() == "yes":
# input에 None 입력 시 이전 상태에서 그래프 실행 계속
for event in graph.stream(None, thread, stream_mode="values"):
print(event)
else:
print("Operation cancelled by user."){'input': '안녕~'}
---Step 1---
---Step 2---
3단계로 가시겠습니까? (yes/no): yes
{'input': '안녕~'}
---Step 3---에이전트
에이전트 컨텍스트에서 중단 지점은 특정 에이전트 동작을 수동으로 승인하는 데 유용하다.
이를 보여주기 위해 도구 호출을 수행하는 비교적 간단한 ReAct 스타일 에이전트를 만들어 보자.
액션 노드가 호출되기 전에 중단 지점을 추가한다.
from IPython.display import display, Image
from dotenv import load_dotenv
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END, StateGraph
from langgraph.graph import MessagesState, START
from langgraph.prebuilt import ToolNode
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)
# 시작점을 'agent'로 설정한다.
# 이것은 이 노드가 처음으로 호출되는 것을 의미한다.
workflow.add_edge(START, "agent")
# 조건부 엣지를 추가한다.
workflow.add_conditional_edges(
# 'agent' 노드가 호출된 후에 호출되는 엣지를 의미한다.
"agent",
# 다음으로 다음에 호출될 노드를 결정할 함수를 전달한다.
should_continue,
# 마지막으로 매핑을 전달한다.
# 키는 문자열이고, 값은 다른 노드들이다.
# END는 그래프가 끝나야 한다는 것을 표시하는 특별한 노드이다.
# 이후 `should_continue`를 호출하고, 그 출력이 이 매핑의 키들과 일치하게 된다.
# 매칭이 된다면 노드가 호출된다.
{
# 만일 `tools`이라면, 도구 노드를 호출한다.
"continue": "action",
# 그렇지 않다면 끝낸다.
"end": END,
},
)
# tools에서 agent로의 일반 엣지를 추가한다.
# 이는 tools가 호출된 후에 agent 노드가 호출된다는 것을 의미한다.
workflow.add_edge("action", "agent")
# 메모리 설정
memory = MemorySaver()
# interrupt_before=["action"]를 추가한다.
# 이것은 `action` 노드가 호출되기 전에 중단점을 추가한다.
app = workflow.compile(checkpointer=memory, interrupt_before=["action"])
display(
Image(
app.get_graph().draw_mermaid_png(output_file_path="how-to-add-breakpoint2.png")
)
)
에이전트와 상호작용
이제 에이전트와 대화할 수 있다.
중단점이 액션 노드 이전에 설정된 interrupt_before로 인해 도구를 호출하기 전에 실행이 중단되는 것을 볼 수 있다.
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_2eSdHqSqMEnrmbEo691zBaXx)
Call ID: call_2eSdHqSqMEnrmbEo691zBaXx
Args:
query: 서울 날씨다시 시작하기 (resume)
이제 입력 없이 에이전트를 다시 호출하여 계속 진행할 수 있다.
이렇게 하면 요청된 대로 도구가 실행된다.
입력에 None을 사용해 중단된 그래프를 실행하면, 중단이 없었던 것처럼 진행된다.
for event in app.stream(None, thread, stream_mode="values"):
event["messages"][-1].pretty_print()================================= Tool Message =================================
Name: search
["서울 날씨는 맑아~ 😈."]
================================== Ai Message ==================================
현재 서울의 날씨는 맑습니다.반응형
