[LangGraph] Human Node (LLM이 판단)

기존까지는 graph를 stream 메서드로 실행하며, interrupt_before 혹은 interreupt_after 옵션으로 Tool Node가 호출되는지 여부에 따라 항상 중단시켰다. 그런데 만약 LLM이 직접 판단해서 필요한 경우 사람의 도움을 요청하도록 하고 싶다면 어떻게 해야할까?

 

LLM이 사람의 개입이 필요할지에 대한 판단을 직접 내리도록 하는 방법 중 하나는 human 노드를 정의하고 해당 노드에 방문할 때는 항상 중단시키도록 하는 것이다. 대신, 이전 노드에서 다음 스텝으로 human node 로 갈지말지에 대한 판단은 HumanAssistance 도구를 LLM이 호출하는지로 판단하도록 한다.

1. human 노드를 추가하고, 이 노드에서는 항상 중단
2. LLM이 HumanAssistance tool을 호출할 때만 human 노드로 이동
3. conditional_edge의 판단을 위해 ask_human flag를 State에 추가

이번에 만들고자하는 전체적인 그림은 아래와 같다.

• State는 messages 리스트와 ask_human flag를 갖는다.
• 질문이 들어왔을 때, chatbot node가 tool을 binding한 LLM을 호출하여 판단하게 한다. chatbot node의 결과로 3가지가 발생할 수 있다.
  1. HumanAssistance tool을 호출 → ask_human=True, human 노드로 이동
  2. search tool을 호출 → ask_human=False, search_tool로 이동
  3. tool 호출 없이 LLM이 답변

 

1. 상태 정의

먼저 상태를 정의해주자. 앞서 설계한 것처럼 상태에는 메세지의 목록(messages)과 ask_human을 갖도록 한다.

from typing import Annotated, TypedDict
from langchain_openai import ChatOpenAI
from utils.tools.tavily import TavilySearch  # langchain의 tool로 가져와도 됨
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


# 상태 정의
class State(TypedDict):
    messages: Annotated[list, add_messages]
    ask_human: bool  # 사람의 개입 필요 여부 flag

 

2. HumanRequest tool 정의 / Tool Binding

 

human_node로 이동하기 위해, LLM이 사람의 개입이 필요할 때 호출할 수 있는 tool HumanAssistance를 정의한다. LLM이 판단할 수 있도록 docstring을 상세하게 적어주자. (tool binding에는 pydantic 모델 또는 json 스키마도 가능)

from pydantic import BaseModel

class HumanAssistance(BaseModel):
    """Escalate the conversation to an expert. Use this if you are unable to assist directly or if the user requires support beyond your permissions.

    To use this function, relay the user's 'request' so the expert can provide the right guidance.
    """

    request: str

 

앞서 정의한 HumanRequest tool과 함께 웹검색을 위한 tavily tool을 OpenAI 사의 ChatOpenAI 모델이 호출할 수 있도록 binding 해주자.

tool = TavilySearch(max_results=3)
tools = [tool, HumanAssistance]

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

 

3. Node 정의

 

이제 search_tool 노드와 chatbot 노드를 정의하자. 여기서 주의할 점은 HumanAssistance 도구를 호출한 경우, ask_human 플래그를 True로 전환하는 것이다.

# search_tool 노드
search_tool = ToolNode(tools=[tool])   # tavily

# chatbot 노드
def chatbot(state: State):
    ask_human = False  # ask_human 초기화
    response = llm_with_tools.invoke(state["messages"]) # LLM 호출

    # tool 호출이 일어나고, 그 이름이 HumanAssistance 도구인 경우
    if response.tool_calls and response.tool_calls[0]["name"] == HumanAssistance.__name__:
        ask_human = True
    return {"messages": [response], "ask_human": ask_human}

 

다음으로 human 노드를 정의하자. 이 노드는 LLM이 답변 생성을 위해 사람의 개입이 필요하다고 판단할 때 중단(interrupt)을 발생시키는 역할을 한다.

 

human 노드에 도달했다면, 그래프의 작동을 중단하고 사용자에게 상태 업데이트를 요청하게 된다. 만약 사용자가 응답하지 않을 경우에 대비해 create_response 함수를 정의해서 Tool Message를 삽입해 다시 chatbot 노드로 돌아갈 수 있도록 분기 처리를 한다. 또한, ask_human 플래그를 False로 변경해 chatbot 노드가 다시 요청하지 않으면 다시 이 노드에 방문하지 않도록 하자.
(즉, human 노드에서의 로직은 사람이 응답하지 않았을 경우에 대비한 것)

from langchain_core.messages import AIMessage, ToolMessage

def create_response(response: str, ai_message: AIMessage):
    return ToolMessage(
        content=response,
        tool_call_id=ai_message.tool_calls[0]["id"]  # tool_calls id 매칭
    )

# human 노드 정의
def human(state: State):
    new_messages = []
    # human 노드로 오게 된다면, 사람이 직접 ToolMessage를 작성하길 기대함
    # 만약, 사람이 응답하지 않은 경우
    if not isinstance(state["messages"][-1], ToolMessage):
        new_messages.append(
            create_response("No response from human", state["messages"][-1])
        )
    return {
        "messages": new_messages,  # 새 메세지 추가
        "ask_human": False,   # 플래그 해제
    }

 

4. 조건부 엣지 함수 정의

 

이제 필요한 노드는 모두 정의했다. 마지막으로 chatbot 노드에서 각 노드(human, search_tool, END)로 분기를 탈 수 있게 다음 노드를 선택하는 함수를 정의하자.

tools_condition은 langgraph에 사전 정의된 함수로 State의 messages의 가장 최근 메세지가 tool_calls를 포함하는지 여부에 따라 tools | END 분기를 태우는 함수

def select_next_node(state: State):
    if state["ask_human"]:
        return "human"
    return tools_condition(state)

 

5. 그래프 정의

 

이제 모든 노드와 함수가 정의되었으니 그래프를 정의해보자. 앞서 살펴봤던 그림을 참고하자.

flow = StateGraph(State)

# 노드 추가
flow.add_node("chatbot", chatbot)
flow.add_node("human", human)
flow.add_node("search_tool", search_tool)


# 엣지 추가
flow.add_edge(START, "chatbot")
flow.add_edge("human", "chatbot")
flow.add_edge("search_tool", "chatbot")

flow.add_conditional_edges(
    "chatbot",
    select_next_node,
    {
        "human": "human",
        "tools": "search_tool",
        "__end__": END
    }
)

# 그래프 컴파일
memory = MemorySaver()  # 인메모리 저장소
graph = flow.compile(checkpointer=memory, interrupt_before=["human"])

# 시각화
from IPython.display import display, Image

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

 

 

 

이제 완성한 그래프에서 chatbot 노드는 3가지 동작 중 하나를 수행하게 된다.

1. human 노드 호출로 인간에게 도움 요청 (chatbot->select->human)
2. search_tool 노드 호출로 웹 검색 (chatbot->select->search_tool)
3. 직접 응답 (chatbot->select-> end)

그럼, 실제 예상한대로 작동하는지 사람에게 도움을 요청하는 쿼리를 작성해보자.

user_input = "LangGraph로 AI 멀티 에이전트를 만드려는데, 전문가 도움이 필요해. 도움을 요청해줘."

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

events = graph.stream(
    {"messages": [("user", user_input)]}, 
    config, 
    stream_mode="values"    # (key, value) pair of State
    )

for event in events:
    if "messages" in event:
        event["messages"][-1].pretty_print()

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

LangGraph로 AI 멀티 에이전트를 만드려는데, 전문가 도움이 필요해. 도움을 요청해줘.
================================== Ai Message ==================================
Tool Calls:
  HumanAssistance (call_ZryvNi5T7jx516WJsBztFJKW)
 Call ID: call_ZryvNi5T7jx516WJsBztFJKW
  Args:
    request: LangGraph로 AI 멀티 에이전트를 만들고 싶은데, 전문가의 도움이 필요합니다. 구체적으로 어떤 부분에서 지원이 필요한지 안내해주시면 좋겠습니다.

 

결과를 보면, LLM에게 binding한 HumanAssistance 도구를 호출했고, request 값으로 조언을 요청하는 내용을 담고 있다. 현재 상태를 snapshot을 찍어 다음 노드를 찍어보면 human 노드가 나오는데, human 노드를 호출하기 전 그래프가 중단되었단 사실을 알 수 있다.

snapshot = graph.get_state(config)
snapshot.next

('human',)

 

그래프 상태는 실제로 human노드보다 먼저 중단된다.

앞서 human 노드를 정의할 때 내부 로직은 상태를 수동으로 업데이트해주지 않고 진행했을 때를 처리하고 있단 걸 기억하자!

 

그럼 우리가 기대한대로, 응답을 작성하여 새로운 ToolMessage를 추가한 다음, 상태를 수동으로 업데이트하고 그래프를 진행시키고자 한다면 아래처럼 하면 된다.

1. ToolMessage 생성
2. graph.update_state 메서드로 그래프 상태 수동 업데이트
3. 그래프 계속 진행

# 1. ToolMessage 생성
ai_message = snapshot.values["messages"][-1]
human_response = """LangGraph로 AI 멀티 에이전트를 만들기 위해서는 Sean의 블로그 (https://sean-j.tistory.com) 를 참고하면 됩니다!
곧 멀티 에이전트에 대한 내용도 추가될 예정이니 인내심을 가지고 기다려주세요.
"""
tool_message = create_response(human_response, ai_message)

# 2. graph.update_state 메서드로 그래프 상태 수동 업데이트
graph.update_state(
    config,
    {"messages": [tool_message]},
    as_node="human"
)

graph.get_state(config).values

[HumanMessage(content='LangGraph로 AI 멀티 에이전트를 만드려는데, 전문가 도움이 필요해. 도움을 요청해줘.', additional_kwargs={}, response_metadata={}, id='65de7428-480c-4376-9f8d-c503b887a514'),
 AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_ZryvNi5T7jx516WJsBztFJKW', 'function': {'arguments': '{"request":"LangGraph로 AI 멀티 에이전트를 만들고 싶은데, 전문가의 도움이 필요합니다. 구체적으로 어떤 부분에서 지원이 필요한지 안내해주시면 좋겠습니다."}', 'name': 'HumanAssistance'}, 'type': 'function'}], 'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 55, 'prompt_tokens': 168, 'total_tokens': 223, 'completion_tokens_details': {'accepted_prediction_tokens': 0, 'audio_tokens': 0, 'reasoning_tokens': 0, 'rejected_prediction_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-mini-2024-07-18', 'system_fingerprint': 'fp_d02d531b47', 'finish_reason': 'tool_calls', 'logprobs': None}, id='run-17718193-9cef-4555-b5c1-baeb3c8267ab-0', tool_calls=[{'name': 'HumanAssistance', 'args': {'request': 'LangGraph로 AI 멀티 에이전트를 만들고 싶은데, 전문가의 도움이 필요합니다. 구체적으로 어떤 부분에서 지원이 필요한지 안내해주시면 좋겠습니다.'}, 'id': 'call_ZryvNi5T7jx516WJsBztFJKW', 'type': 'tool_call'}], usage_metadata={'input_tokens': 168, 'output_tokens': 55, 'total_tokens': 223, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}}),
 ToolMessage(content='LangGraph로 AI 멀티 에이전트를 만들기 위해서는 Sean의 블로그 (https://sean-j.tistory.com) 를 참고하면 됩니다!\n곧 멀티 에이전트에 대한 내용도 추가될 예정이니 인내심을 가지고 기다려주세요.\n', id='4f47d946-8356-4672-92e2-b49e500f302e', tool_call_id='call_ZryvNi5T7jx516WJsBztFJKW')]

 

위처럼 human 노드 직전에 멈추고, 마치 human 노드의 결과물인 것처럼 ToolMessage를 작성해서 추가해주었다.

마지막으로, 입력값에 None을 주고 그래프를 재개하자.

events = graph.stream(None, config, stream_mode="values")

for event in events:
    if "messages" in event:
        event["messages"][-1].pretty_print()

================================= Tool Message =================================

LangGraph로 AI 멀티 에이전트를 만들기 위해서는 Sean의 블로그 (https://sean-j.tistory.com) 를 참고하면 됩니다!
곧 멀티 에이전트에 대한 내용도 추가될 예정이니 인내심을 가지고 기다려주세요.

================================== Ai Message ==================================

LangGraph로 AI 멀티 에이전트를 만들기 위해서는 Sean의 블로그 [여기](https://sean-j.tistory.com)를 참고하시면 됩니다. 곧 멀티 에이전트에 대한 내용도 추가될 예정이니 인내심을 가지고 기다려주세요. 도움이 필요하신 부분이 있다면 언제든지 말씀해 주세요!')]

 

이제 최종적으로 그래프가 어떤 흐름을 타고 진행되었는지 확인해보자

state = graph.get_state(config)

for message in state.values["messages"]:
    message.pretty_print()

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

LangGraph로 AI 멀티 에이전트를 만드려는데, 전문가 도움이 필요해. 도움을 요청해줘.
================================== Ai Message ==================================
Tool Calls:
  HumanAssistance (call_ZryvNi5T7jx516WJsBztFJKW)
 Call ID: call_ZryvNi5T7jx516WJsBztFJKW
  Args:
    request: LangGraph로 AI 멀티 에이전트를 만들고 싶은데, 전문가의 도움이 필요합니다. 구체적으로 어떤 부분에서 지원이 필요한지 안내해주시면 좋겠습니다.
================================= Tool Message =================================

LangGraph로 AI 멀티 에이전트를 만들기 위해서는 Sean의 블로그 (https://sean-j.tistory.com) 를 참고하면 됩니다!
곧 멀티 에이전트에 대한 내용도 추가될 예정이니 인내심을 가지고 기다려주세요.

================================== Ai Message ==================================

LangGraph로 AI 멀티 에이전트를 만들기 위해서는 Sean의 블로그 [여기](https://sean-j.tistory.com)를 참고하시면 됩니다. 곧 멀티 에이전트에 대한 내용도 추가될 예정이니 인내심을 가지고 기다려주세요. 도움이 필요하신 부분이 있다면 언제든지 말씀해 주세요!

 

위 과정에서 chatbot이 중간에 멈추고, 사람이 개입해서 응답을 생성하는 중간 과정이 있었다. 하지만, 모든 상태가 checkpointer (여기서는 인메모리)에 저장되어 있기 때문에, 사람이 동적으로 개입해도 그래프의 전체 흐름이 영향을 받지 않았다!

 

Cursor AI 등의 AI 툴로 작성할 때, LLM이 코드를 작성하고 사람이 Accept할지 결정하는 과정들도 이러한 개념이 들어간 것이지 않을까 싶다. 추후 LLM 서비스를 고도화할 때 참고하면 좋을 것 같다.

---

출처:

LangGraph. "LangGraph Quick Start". https://langchain-ai.github.io/langgraph/tutorials/introduction/#requirements

위키독스 - <랭체인LangChain 노트> - LangChain 한국어 튜토리얼🇰🇷  (https://wikidocs.net/book/14314)