랭그래프(LangGraph) 도구(tools), 조건부 엣지, Human-in-the-Loop 사용법과 예제

포스팅 개요

이번 포스팅에서는 이전 글에서 다루었던 LangGraph의 기본 개념을 넘어, 한층 더 지능적이고 유연한 LLM 에이전트를 구축하는 방법을 알아봅니다. LangGraph의 강력한 기능인 도구(Tool) 사용, 조건부 엣지(Conditional Edge), 그리고 사용자의 개입을 허용하는 사람의 개입(Human-in-the-Loop) 메커니즘을 집중적으로 다룹니다.

LangGraph를 사용하여 에이전트가 상황에 따라 동적으로 행동을 결정하고, 스스로 해결할 수 없는 문제에 대해서는 사람에게 도움을 요청하여 작업을 일시 중단했다가 피드백을 받아 재개하는 전체 과정을 상세한 코드 예제와 함께 살펴보겠습니다.

 

본 포스팅을 작성하면서 참고한 자료는 다음과 같습니다.

• https://github.com/langchain-ai/langgraph?tab=readme-ov-file
• https://langchain-ai.github.io/langgraph/concepts/tools/
• https://langchain-ai.github.io/langgraph/concepts/human_in_the_loop/

Overview

---

포스팅 본문

지난 포스팅( https://lsjsj92.tistory.com/696 )에서는 LangGraph의 State, Node, Edge라는 세 가지 핵심 요소를 이용해 간단한 챗봇을 만드는 방법을 알아보았습니다. 하지만 실제 세상의 문제를 해결하기 위해서는 LLM이 단순히 대답만 하는 것을 넘어, 외부 도구를 사용해 정보를 가져오거나, 특정 조건에 따라 다른 작업을 수행하고, 때로는 사람의 판단을 구하는 등 훨씬 복잡한 상호작용이 필요합니다.

이번 포스팅에서는 바로 이러한 고급 기능을 구현하는 방법을 단계별로 알아보겠습니다.

1. LangGraph 에이전트의 핵심 기능 3가지
2. LangGraph 도구, 조건부 엣지, Human-in-the-Loop 예제 코드

---

1. LangGraph 에이전트의 핵심 기능 3가지

오늘 만들 에이전트는 다음 세 가지 핵심 기능을 갖추고 있습니다.

• 도구(Tool) 사용: LLM이 대화뿐만 아니라, 특정 작업을 수행하는 함수를 호출할 수 있는 능력입니다. 예를 들어 '오늘 날씨 알려줘'라는 요청에 웹 검색 도구를 사용하여 최신 정보를 가져올 수 있습니다.
• 조건부 엣지(Conditional Edges): 에이전트의 '두뇌'와 같은 역할을 합니다. LLM의 판단에 따라 다음에 실행할 노드를 동적으로 결정하는 경로입니다. "LLM이 도구를 사용하겠다고 판단했는가?"라는 조건에 따라 '도구 실행 노드'로 가거나, '워크플로우 종료'로 분기할 수 있습니다.
• 사람의 개입(Human-in-the-Loop): AI가 스스로 해결하기 어려운 문제에 직면했을 때, 워크플로우를 일시 중지하고 사람에게 도움을 요청하는 기능입니다. 사용자는 피드백을 제공하고, 에이전트는 그 피드백을 바탕으로 중단된 지점부터 다시 작업을 이어 나갑니다.

---

2. LangGraph 도구, 조건부 엣지, Human-in-the-Loop 예제 코드

이론적인 것보다 코드를 보면 더욱 이해가 빠르실 겁니다. 실제 코드를 통해 위 기능들이 어떻게 구현되는지 상세히 살펴보겠습니다.

이 코드는 LLM이 웹 검색을 하거나, 필요시 사람의 도움을 요청하는 에이전트인데요. 실제 동작되는 함수를 만들지는 않았고, 예제를 위한 fake function을 구성하였습니다.

2-1. 도구(Tool) 정의: fake_web_search와 human_assistance

에이전트가 사용할 두 가지 도구를 정의합니다. 하나는 일반적인 정보 검색용, 다른 하나는 사람의 개입을 위한 특별한 도구입니다.

from langchain_core.tools import tool
from langgraph.types import interrupt

@tool
def fake_web_search(query: str) -> str:
    """주어진 쿼리에 대해 웹 검색을 수행합니다. (가상)"""
    print(f"--- 가상 웹 검색 수행: {query} ---")
    if "langgraph" in query.lower():
        return "LangGraph는 복잡한 AI 에이전트를 만들기 위한 LangChain의 라이브러리입니다."
    if "날씨" in query.lower():
        return "서울의 오늘 날씨는 맑고, 최고 기온은 28도입니다."
    return "검색 결과가 없습니다."

@tool
def human_assistance(query: str) -> str:
    """AI가 스스로 해결할 수 없는 복잡한 문제에 대해 사람에게 도움을 요청합니다."""
    print(f"--- 사람의 참여 요청: {query} ---")
    # interrupt를 호출하여 그래프 실행을 멈추고, 사용자 입력을 기다립니다.
    human_response = interrupt(value={"query": query})
    return human_response

tools = [fake_web_search, human_assistance]
llm_with_tools = llm.bind_tools(tools)

• fake_web_search: 날씨나 특정 키워드에 대한 정보를 반환하는 가상 검색 도구입니다. real-application에서는 여기에 실제 동작되는 API 코드 등을 구축하면 됩니다. 
• human_assistance: 이 도구가 바로 Human-in-the-Loop의 핵심입니다. 내부적으로 LangGraph의 interrupt() 함수를 호출합니다.

2-2. 시스템 프롬프트와 에이전트 노드 정의

LLM이 언제 어떤 도구를 사용해야 할지 명확하게 알려주기 위해 시스템 프롬프트를 사용합니다.

SYSTEM_PROMPT = """당신은 유능한 AI 어시스턴트입니다. 사용자의 질문에 답하기 위해 다음 규칙을 따르세요.

- 일반적인 정보나 최신 정보(날씨 등)가 필요하면 `fake_web_search` 도구를 사용하세요.
- 스스로 답할 수 없거나, 사용자가 명시적으로 '전문가', '사람', '도움' 등을 요청하며 복잡한 문제를 문의하면, 반드시 `human_assistance` 도구를 사용하여 사람에게 도움을 요청하세요.
- 그 외의 일반적인 대화는 도구 없이 직접 답변하세요.
"""

def chatbot_node(state: AgentState):
    """LLM을 호출하여 다음 행동(응답 또는 도구 호출)을 결정합니다."""
    print("--- 에이전트 노드 실행 ---")
    messages_with_prompt = [SystemMessage(content=SYSTEM_PROMPT)] + state["messages"]
    response = llm_with_tools.invoke(messages_with_prompt)
    return {"messages": [response]}

• SYSTEM_PROMPT: LLM의 역할과 도구 사용 규칙을 명확하게 정의합니다. 이 지침 덕분에 LLM은 "날씨" 질문에는 fake_web_search를, "전문가 조언" 요청에는 human_assistance를 호출해야겠다고 판단할 수 있습니다.
• chatbot_node: 이 노드는 대화 기록에 시스템 프롬프트를 추가하여 LLM에게 전달하고, LLM의 결정(일반 답변 또는 도구 호출)을 받아 상태를 업데이트합니다.

반응형

2-3. 그래프 구성: 조건부 엣지와 체크포인터

이제 노드들을 연결하여 실제 워크플로우를 구성합니다. 여기서 조건부 엣지와 체크포인터가 등장합니다.

from langgraph.graph import StateGraph, START
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.checkpoint.memory import InMemorySaver

graph_builder = StateGraph(AgentState)

graph_builder.add_node("chatbot", chatbot_node)
graph_builder.add_node("tools", ToolNode(tools))

graph_builder.add_edge(START, "chatbot") # 시작은 무조건 chatbot 노드

# 조건부 엣지 설정
graph_builder.add_conditional_edges(
    "chatbot",
    tools_condition, # LLM의 응답에 tool_calls가 있으면 "tools", 없으면 END로 분기
)
graph_builder.add_edge("tools", "chatbot") # 도구 실행 후, 결과를 가지고 다시 chatbot으로

# 체크포인터 설정
memory = InMemorySaver()
agent_app = graph_builder.compile(checkpointer=memory)

• graph_builder.add_conditional_edges("chatbot", tools_condition): 이 부분이 바로 조건부 엣지입니다.
• checkpointer=memory: compile() 함수에 체크포인터를 지정했습니다. 이는 interrupt()로 워크플로우가 중단될 때, 현재까지의 모든 대화 상태(State)를 저장하는 역할을 합니다. 상태가 저장되어야 나중에 사용자가 피드백을 주었을 때 중단된 지점부터 완벽하게 이어서 실행할 수 있습니다.

2-4. 에이전트 실행: 사람의 개입(Human-in-the-Loop) 처리

이제 human_assistance 도구가 호출되어 워크플로우가 중단되었을 때, 어떻게 처리하고 재개하는지 살펴보겠습니다.

def run_agent(app: CompiledGraph, user_input: str, thread_id: str):
    """사용자 입력으로 에이전트를 실행하고, 인간 참여를 처리하는 테스트 함수"""
    config = {"configurable": {"thread_id": thread_id}}
    
    # 초기 입력을 HumanMessage로 설정
    state = {"messages": [HumanMessage(content=user_input)]}
    
    # stream()을 사용하여 에이전트 실행
    events = app.stream(state, config=config, stream_mode="values")
    
    interrupted_tool_call_id = None

    for event in events:
        # AI의 응답 출력
        if "messages" in event:
            last_message = event["messages"][-1]
            if isinstance(last_message, AIMessage):
                print(f"AI 응답: {last_message.content}")
                if last_message.tool_calls:
                    # human_assistance 도구 호출 ID 저장
                    if last_message.tool_calls[0]['name'] == 'human_assistance':
                        interrupted_tool_call_id = last_message.tool_calls[0]['id']
                    print(f"도구 호출: {last_message.tool_calls[0]['name']}({last_message.tool_calls[0]['args']})")

    # 스트림이 끝난 후, 중단된 상태인지 확인
    snapshot = app.get_state(config)
    if snapshot.next: # 다음 실행할 노드가 남아있다면 (즉, 중단되었다면)
        print("\n--- 사람의 도움이 필요합니다! ---")
        human_feedback = input("피드백을 입력해주세요: ")
        
        # ToolMessage를 사용하여 중단된 지점부터 실행 재개
        # 이전에 저장해둔 tool_call_id를 사용합니다.
        resumed_events = app.stream(
            {"messages": [ToolMessage(content=human_feedback, tool_call_id=interrupted_tool_call_id)]},
            config=config,
            stream_mode="values"
        )
        for event in resumed_events:
            if "messages" in event:
                last_message = event["messages"][-1]
                if isinstance(last_message, AIMessage):
                    print(f"AI 응답 (피드백 반영): {last_message.content}")
    
    print("\n--- 워크플로우 종료 ---")

1. app.stream(...): 에이전트를 실행합니다. 만약 human_assistance가 호출되면, interrupt에 의해 이 스트림은 중단 지점에서 멈춥니다.
2. snapshot = app.get_state(config): 스트림이 끝난 후, 현재 대화(thread_id)의 상태를 가져옵니다. snapshot.next에 다음 실행할 노드 이름이 남아있다면, 이는 워크플로우가 중단되었음을 의미합니다.
3. app.stream({"messages": [ToolMessage(...)]}): 사용자에게 피드백을 입력받은 후, 이 피드백을 ToolMessage 형태로 만들어 다시 stream을 호출합니다.

 

이 모든 과정을 거쳐 만들어진 에이전트의 구조는 아래와 같이 시각화할 수 있습니다.

 

위 그림을 보면 chatbot 노드에서 tools_condition에 따라 tools 노드로 가거나 __end__로 가는 분기점을 명확히 확인할 수 있으며, tools 노드가 다시 chatbot으로 돌아오는 순환 구조를 가지고 있음을 알 수 있습니다.

 

아래 화면은 실제 실행한 결과입니다.

 

 

의도했던 대로 도구를 사용하거나, Human-in-the-loop가 동작됨을 확인할 수 있습니다.

---

마무리

이번 포스팅에서는 LangGraph를 사용하여 도구를 사용하고, 조건에 따라 행동을 결정하며, 필요할 때는 사람에게 도움을 요청하는 Langgraph 예제를 알아보았습니다.

도움이 되시길 바랍니다.