[langgraph] human-in-the-loop 추가하기
원문 출처: https://langchain-ai.github.io/langgraph/tutorials/get-started/4-human-in-the-loop/
에이전트는 신뢰성이 떨어질 수 있으며, 작업을 성공적으로 수행하려면 인간의 입력이 필요할 때가 있습니다. 또한, 일부 작업은 실행 전에 반드시 인간의 승인을 받아야 할 수도 있습니다.
LangGraph의 영속성 계층은 human-in-the-loop 워크플로우를 지원하여, 사용자 피드백에 따라 실행을 일시 중지하고 다시 시작할 수 있습니다. 이 기능의 핵심 인터페이스는 interrupt 함수입니다. 노드 내에서 interrupt를 호출하면 실행이 일시 중지되고, Command를 전달해 새로운 입력과 함께 실행을 재개할 수 있습니다. interrupt는 Python의 내장 input()과 유사하게 동작하지만 몇 가지 차이점이 있습니다.
이 튜토리얼은 메모리 추가하기를 기반으로 합니다.
1. human_assistance 도구 추가
이전 튜토리얼의 코드를 바탕으로, 챗봇에 human_assistance 도구를 추가합니다. 이 도구는 interrupt를 사용해 인간의 정보를 받아옵니다.
LLM 선택 예시
OpenAI
pip install -U "langchain[openai]"import os
from langchain.chat_models import init_chat_model
os.environ["OPENAI_API_KEY"] = "sk-..."
llm = init_chat_model("openai:gpt-4.1")Anthropic
pip install -U "langchain[anthropic]"import os
from langchain.chat_models import init_chat_model
os.environ["ANTHROPIC_API_KEY"] = "sk-..."
llm = init_chat_model("anthropic:claude-3-5-sonnet-latest")Azure
pip install -U "langchain[openai]"import os
from langchain.chat_models import init_chat_model
os.environ["AZURE_OPENAI_API_KEY"] = "..."
os.environ["AZURE_OPENAI_ENDPOINT"] = "..."
os.environ["OPENAI_API_VERSION"] = "2025-03-01-preview"
llm = init_chat_model(
"azure_openai:gpt-4.1",
azure_deployment=os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"],
)Google Gemini
pip install -U "langchain[google-genai]"import os
from langchain.chat_models import init_chat_model
os.environ["GOOGLE_API_KEY"] = "..."
llm = init_chat_model("google_genai:gemini-2.0-flash")AWS Bedrock
pip install -U "langchain[aws]"from langchain.chat_models import init_chat_model
# 자격 증명 구성: https://docs.aws.amazon.com/bedrock/latest/userguide/getting-started.html
llm = init_chat_model(
"anthropic.claude-3-5-sonnet-20240620-v1:0",
model_provider="bedrock_converse",
)StateGraph에 human_assistance 도구 추가
from typing import Annotated
from langchain_tavily import TavilySearch
from langchain_core.tools import tool
from typing_extensions import TypedDict
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.types import Command, interrupt
class State(TypedDict):
messages: Annotated[list, add_messages]
graph_builder = StateGraph(State)
@tool
def human_assistance(query: str) -> str:
"""Request assistance from a human."""
human_response = interrupt({"query": query})
return human_response["data"]
tool = TavilySearch(max_results=2)
tools = [tool, human_assistance]
llm_with_tools = llm.bind_tools(tools)
def chatbot(state: State):
message = llm_with_tools.invoke(state["messages"])
# human-in-the-loop 도구 실행 중에는 병렬 호출을 비활성화해야 함
assert len(message.tool_calls) <= 1
return {"messages": [message]}
graph_builder.add_node("chatbot", chatbot)
tool_node = ToolNode(tools=tools)
graph_builder.add_node("tools", tool_node)
graph_builder.add_conditional_edges(
"chatbot",
tools_condition,
)
graph_builder.add_edge("tools", "chatbot")
graph_builder.add_edge(START, "chatbot")
human-in-the-loop 워크플로우에 대한 더 많은 예시는 공식 문서를 참고하세요.
2. 그래프 컴파일
이전과 마찬가지로 체크포인터를 지정해 그래프를 컴파일합니다:
memory = MemorySaver()
graph = graph_builder.compile(checkpointer=memory)3. 그래프 시각화 (선택 사항)
그래프를 시각화하면 human_assistance 도구가 추가된 구조를 확인할 수 있습니다.
from IPython.display import Image, display
try:
display(Image(graph.get_graph().draw_mermaid_png()))
except Exception:
# 추가 종속성이 필요하며 선택 사항입니다
pass4. 챗봇에 프롬프트 입력하기
이제 human_assistance 도구가 동작하는 질문을 해봅니다.
user_input = "I need some expert guidance for building an AI agent. Could you request assistance for me?"
config = {"configurable": {"thread_id": "1"}}
events = graph.stream(
{"messages": [{"role": "user", "content": user_input}]},
config,
stream_mode="values",
)
for event in events:
if "messages" in event:
event["messages"][-1].pretty_print()실행 결과 예시:
================================ Human Message =================================
I need some expert guidance for building an AI agent. Could you request assistance for me?
================================== Ai Message ==================================
[{'text': "Certainly! I'd be happy to request expert assistance for you regarding building an AI agent. To do this, I'll use the human_assistance function to relay your request. Let me do that for you now.", 'type': 'text'}, {'id': 'toolu_01ABUqneqnuHNuo1vhfDFQCW', 'input': {'query': 'A user is requesting expert guidance for building an AI agent. Could you please provide some expert advice or resources on this topic?'}, 'name': 'human_assistance', 'type': 'tool_use'}]
Tool Calls:
human_assistance (toolu_01ABUqneqnuHNuo1vhfDFQCW)
Call ID: toolu_01ABUqneqnuHNuo1vhfDFQCW
Args:
query: A user is requesting expert guidance for building an AI agent. Could you please provide some expert advice or resources on this topic?interrupt가 호출되면 실행이 일시 중지됩니다. 체크포인터가 DB라면 언제든 복구할 수 있습니다. 여기서는 인메모리 체크포인터이므로 파이썬 커널이 살아있는 한 언제든 재개할 수 있습니다.
5. 실행 재개하기
실행을 재개하려면, 도구가 기대하는 데이터를 담은 Command 객체를 전달합니다. 예시에서는 "data" 키를 사용합니다.
human_response = (
"We, the experts are here to help! We'd recommend you check out LangGraph to build your agent. "
"It's much more reliable and extensible than simple autonomous agents."
)
from langgraph.types import Command
human_command = Command(resume={"data": human_response})
events = graph.stream(human_command, config, stream_mode="values")
for event in events:
if "messages" in event:
event["messages"][-1].pretty_print()실행 결과 예시:
================================== Ai Message ==================================
Thank you for your patience. I've received some expert advice regarding your request for guidance on building an AI agent. Here's what the experts have suggested:
The experts recommend that you look into LangGraph for building your AI agent. They mention that LangGraph is a more reliable and extensible option compared to simple autonomous agents.
...입력이 도구 메시지로 처리되어, 챗봇이 이어서 답변을 제공합니다.
축하합니다! 이제 interrupt를 활용해 human-in-the-loop 기능을 챗봇에 추가했습니다. 이로써 필요한 경우 인간의 개입과 검증이 가능한 AI 시스템을 만들 수 있습니다. 체크포인터가 있다면, 그래프는 언제든 중단/재개가 가능합니다.
아래는 튜토리얼 전체 코드 예시입니다:
from typing import Annotated
from langchain_tavily import TavilySearch
from langchain_core.tools import tool
from typing_extensions import TypedDict
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.types import Command, interrupt
class State(TypedDict):
messages: Annotated[list, add_messages]
graph_builder = StateGraph(State)
@tool
def human_assistance(query: str) -> str:
human_response = interrupt({"query": query})
return human_response["data"]
tool = TavilySearch(max_results=2)
tools = [tool, human_assistance]
llm_with_tools = llm.bind_tools(tools)
def chatbot(state: State):
message = llm_with_tools.invoke(state["messages"])
assert(len(message.tool_calls) <= 1)
return {"messages": [message]}
graph_builder.add_node("chatbot", chatbot)
tool_node = ToolNode(tools=tools)
graph_builder.add_node("tools", tool_node)
graph_builder.add_conditional_edges(
"chatbot",
tools_condition,
)
graph_builder.add_edge("tools", "chatbot")
graph_builder.add_edge(START, "chatbot")
memory = MemorySaver()
graph = graph_builder.compile(checkpointer=memory)지금까지는 메시지 리스트 하나만을 상태로 사용했지만, 더 복잡한 동작을 원한다면 상태에 추가 필드를 정의할 수 있습니다. 다음 튜토리얼에서는 상태 커스터마이징을 다룹니다.
반응형
