4. langgraph_chat_with_human_in_the_loop

LangGraph 챗봇에 사람의 개입 통합하기

본 강의에서는 LangGraph를 활용하여 챗봇 워크플로우에 인간의 개입(Human-in-the-loop)을 통합하는 방법을 단계별로 안내합니다. 이러한 통합은 복잡한 질문이나 모델의 불확실한 응답 시, 인간이 직접 개입하여 대화의 품질과 정확성을 향상시키는 데 도움이 됩니다.

학습 목표

• 챗봇 워크플로우에 인간 검토 노드 추가하기
• LangGraph의 interrupt 기능을 이용해 인간 개입 요청하기
• 인간 개입 후 워크플로우 중단 및 재개하기

학습 내용 요약

1. 인간 검토 노드 추가:
   • 챗봇이 처리하기 어려운 질문을 받았을 때, 이를 인간 검토 노드로 라우팅하여 사람이 직접 응답할 수 있도록 합니다.
2. LangGraph의 interrupt 기능 활용:
   • interrupt 함수를 사용하여 그래프 실행을 일시 중지하고, 인간의 입력을 기다릴 수 있습니다. 이 기능을 통해 특정 지점에서 인간의 개입을 요청하고, 입력을 받은 후 실행을 재개할 수 있습니다.
3. 체크포인팅을 통한 상태 관리:
   • 인간의 입력을 기다리는 동안 현재 상태를 저장하여, 이후 실행을 원활하게 재개할 수 있도록 합니다.

이러한 방식을 통해, 챗봇은 자동화된 응답과 인간의 개입을 효과적으로 결합하여 보다 정확하고 신뢰성 있는 서비스를 제공할 수 있습니다.

1. 환경 설정

라이브러리 설치

다음 명령어를 실행하여 필요한 라이브러리를 설치합니다.

pip install -U python-dotenv notebook langgraph langchain_openai tavily-python langchain_community

환경변수 설정

환경변수 파일 .env를 생성하여 다음의 내용을 설정합니다.

OPENAI_API_KEY=본인의_OpenAI_API키
OPENAI_MODEL=gpt-4o-mini
TAVILY_API_KEY=본인의_tavily_api_key

환경변수를 로드하기 위해 Python의 python-dotenv 라이브러리를 사용합니다.

2. 코드 설명

라이브러리 임포트

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from typing_extensions import TypedDict
from typing import Annotated, List
from dotenv import load_dotenv
import os

• ChatOpenAI: OpenAI의 Chat Completion API를 호출하기 위한 클래스입니다.
• StateGraph, START, END: LangGraph의 상태 기반 그래프 구조를 구성하는 데 사용됩니다.
• add_messages: LangGraph의 메시지 관리 기능을 자동화하는 역할을 수행합니다.
• TypedDict, Annotated, List: 상태 정의를 위한 타입 힌트 도구입니다.
• load_dotenv: 환경변수를 로드합니다.

환경변수 로딩

load_dotenv()

openai_model = os.getenv("OPENAI_MODEL", "gpt-4o-mini")

도구 정의

Tavily 검색 도구를 설정합니다.

from langchain_community.tools.tavily_search import TavilySearchResults

search_tool = TavilySearchResults(max_results=2)
search_tool.invoke("LangGraph가 무엇인가요?")

• TavilySearchResults: 사용자가 입력한 질의에 대해 최대 2개의 결과를 가져오는 검색 도구를 만들었습니다.

Human-in-the-loop 도구를 설정합니다.

• LangGraph에서 Human-in-the-loop 기능을 구현할 때 LLM의 요청에 따라 사람이 직접 개입하여 응답을 제공할 수 있도록 하는 **도구(tool)**를 정의합니다.
• 이 도구는 사용자가 입력한 질문에 대해 LLM이 직접 응답할 수 없는 경우, 사람에게 질문을 전달하고 답변을 받을 수 있도록 합니다.

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

@tool
def human_assist(query):
    """Human assist tool"""
    human_response = interrupt({"query": query})
    return human_response["data"]

① @tool 데코레이터

@tool

• LangChain의 도구(tool)로 해당 함수를 자동 등록합니다.
• LLM이 해당 도구를 호출할 수 있도록 합니다.
• 데코레이터를 통해 명시적 name 및 description을 제공하지 않으면, 함수 이름(human_assist)과 Docstring("Human assist tool")이 자동으로 사용됩니다.

② 함수 정의 (human_assist)

def human_assist(query):

• 함수 이름: human_assist
• 입력 인자:
  • query (필수): 사용자가 입력한 질문 또는 LLM이 추가 정보가 필요하다고 판단한 내용을 전달합니다.

③ 함수의 Docstring (설명 문자열)

"""Human assist tool"""

• 함수가 수행하는 역할을 간략하게 설명합니다.
• LLM이 이 도구의 기능을 쉽게 이해할 수 있도록 도와줍니다.

④ interrupt 함수 호출

human_response = interrupt({"query": query})

• interrupt 함수의 역할:
  • LangGraph에서 워크플로우의 실행을 일시적으로 중단하고, 외부(인간)의 입력을 기다리는 데 사용됩니다.
  • 이 함수를 호출하면 LangGraph는 즉시 실행을 중단하고, 호출된 곳에서 지정한 데이터를 외부로 전달하여 인간의 개입을 요청합니다.
  • 외부에서 제공한 입력이 도착할 때까지 상태를 저장하고 대기합니다.
• 입력 형태:이 형태로 인간에게 전달됩니다.
• { "query": "LLM이 물어본 질문 또는 요청 내용" }

⑤ 인간 입력의 반환 및 처리

return human_response["data"]

• 인간이 제공한 입력(응답)을 받으면, LangGraph는 해당 응답을 interrupt 호출의 반환값으로 전달합니다.
• human_response는 다음과 같은 구조를 가집니다:

  {
      "data": "인간이 입력한 실제 응답 내용"
  }
  

• 따라서 이 코드는 인간이 제공한 실제 데이터를 반환하여 LLM의 최종 응답에 사용할 수 있도록 합니다.

상태(State) 정의

class State(TypedDict):
    messages: Annotated[list, add_messages]

• 이 코드는 LangGraph에서 챗봇의 상태(State)를 정의하는 부분입니다.
• 상태는 LangGraph의 핵심 개념입니다. 챗봇이 메시지를 주고받으며 관리할 데이터를 명확히 정의합니다.
• LangGraph는 상태 기반 그래프(StateGraph)를 사용하는 프레임워크로, 각 단계의 상태를 명확히 정의하고, 상태 간에 데이터를 전달하는 방식으로 작동합니다.
• messages는 메시지를 관리하는 리스트로, 자동 메시지 추가 및 관리가 가능합니다.

구성 요소 설명

• class State(TypedDict)
  • Python의 TypedDict를 사용해 명시적인 타입이 있는 딕셔너리를 정의합니다.
  • 이를 통해 상태(State)의 구조와 타입을 명확히 지정하여 코드의 가독성과 안정성을 높입니다.
• messages
  • 상태에 저장되는 주요 데이터입니다. 여기서는 대화 중 주고받은 메시지를 담고 있는 리스트입니다.
  • LangGraph는 대화의 문맥을 유지하기 위해 메시지의 리스트를 관리합니다.
• Annotated[List, add_messages]
  • Annotated 타입 힌트를 사용하여 메시지 리스트가 특정한 규칙(add_messages)을 따르도록 합니다.
  • 여기서 add_messages는 LangGraph에서 제공하는 특별한 주석(annotation)으로, 메시지 리스트를 다룰 때 자동으로 관리(추가)를 돕는 기능을 수행합니다.

이 상태 정의를 통해 챗봇은:

• 대화의 각 단계를 명확히 관리할 수 있습니다.
• 사용자의 메시지와 모델의 응답 메시지를 체계적으로 기록하고 관리합니다.
• 명시적인 타입과 자동 관리 기능을 활용하여 코드의 오류를 최소화하고 유지보수를 쉽게 합니다.

이러한 방식을 통해, LangGraph는 상태 기반의 복잡한 챗봇 애플리케이션에서도 효율적이고 명확한 데이터 관리와 흐름 제어를 가능하게 합니다.

LLM 모델 설정 및 도구 바인딩

llm = ChatOpenAI(model=openai_model)
tools = [search_tool, human_assist]
llm_with_tools = llm.bind_tools(tools)

① 도구 목록 준비 (tools)

tools = [search_tool, human_assist]

• LLM이 활용할 수 있는 도구들을 리스트 형태로 정의합니다.
• 이 리스트는 이전에 정의된 함수나 클래스 기반 도구를 포함할 수 있습니다.
• search_tool: 웹 검색 등의 자동화 도구
• human_assist: 챗봇이 답하기 어려울 때, 인간의 판단을 요청하는 도구

도구는 보통 @tool 데코레이터로 정의하거나 클래스 형태로 정의할 수 있습니다.

챗봇 노드 정의

챗봇이 도구를 이용하여 사용자 메시지에 응답하도록 설정합니다.

def chatbot(state: State):
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}

• chatbot(state: State): 챗봇 노드 함수로, 상태에서 받은 메시지를 기반으로 도구를 활용하여 응답을 생성하고 상태를 업데이트합니다.

체크포인터 설정

체크포인팅을 위한 메모리 체크포인터를 생성합니다.

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()

• MemorySaver: 메모리 기반의 체크포인터로, 각 대화의 상태를 메모리에 임시로 저장하고 관리합니다. 이를 통해 챗봇은 이전 대화 내용을 기억하고 다음 번 상호작용 시에도 맥락을 유지한 상태로 대화를 진행할 수 있습니다. 실제 운영 환경에서는 더 영구적인 상태 관리를 위해 데이터베이스 기반 체크포인터(예: SqliteSaver 또는 PostgresSaver)를 사용하는 것이 권장됩니다.

그래프 구성 및 컴파일

LangGraph의 ToolNode와 tools_condition을 사용하여 조건부로 도구 노드를 호출합니다.

from langgraph.prebuilt import ToolNode, tools_condition

tool_node = ToolNode(tools)

workflow = StateGraph(State)

workflow.add_node("chatbot", chatbot)
workflow.add_node("tools", tool_node)
workflow.add_conditional_edges("chatbot", tools_condition)
workflow.add_edge("tools", "chatbot")
workflow.add_edge(START, "chatbot")

graph = workflow.compile(checkpointer=memory)

ToolNode는 LangGraph에서 미리 만들어져 제공되는 도구 실행 전용 노드입니다.

• LLM이 도구(tool)를 호출하겠다고 요청할 때, 실제 도구를 실행하는 역할을 합니다.
• 도구 호출 요청 메시지를 입력으로 받아 도구를 실행하고, 그 결과를 다시 LLM에 전달할 수 있는 형태로 반환합니다.
• 사용자가 별도의 복잡한 로직을 구현하지 않고, 간단하게 도구를 처리할 수 있도록 미리 만들어진 클래스입니다.
• ToolNode 생성시 도구를 리스트 형태로 전달합니다.

작동 원리

ToolNode의 내부 구조는 다음과 같은 순서로 동작합니다:

1. 챗봇(LLM) 노드가 도구 호출을 요청한 메시지를 생성합니다.
2. ToolNode가 이 메시지를 입력으로 받아 메시지에 포함된 **도구 호출(tool_calls)**을 실행합니다.
3. 도구 실행 결과를 ToolMessage 형태로 변환하여 다시 챗봇(LLM) 노드가 이해할 수 있게 반환합니다.

tools_condition은 LangGraph에서 미리 만들어 제공하는 조건부 엣지(conditional edge)를 위한 조건 함수입니다.

• 챗봇 노드의 출력 메시지에 도구 호출 요청이 포함되어 있는지 확인합니다.
• 도구 호출 요청이 있다면, 도구 노드(ToolNode)를 실행하도록 조건을 설정해주는 기능입니다.
• 별도로 조건 로직을 구현하지 않아도, 도구 호출의 필요성을 쉽게 판단할 수 있도록 미리 만들어진 편의 함수입니다.

작동 원리

tools_condition 함수는 다음 로직을 수행합니다:

1. 입력 상태에서 가장 최근 메시지를 확인합니다.
2. 최근 메시지에 tool_calls가 포함되어 있는지 확인합니다.
3. 만약 메시지에 하나 이상의 도구 호출 요청이 존재하면(tool_calls가 존재하면), 다음 노드를 'tools'로 설정합니다.
4. 도구 호출 요청이 없다면 자동으로 종료(END) 노드로 설정하여 그래프를 종료합니다.

그래프 시각화

컴파일된 그래프를 이용해 시각화봅니다.

from IPython.display import Image, display

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

3. 챗봇 실행

동일한 thread_id를 사용하여 이전 대화 맥락을 유지하는 예시입니다. 이 예시에서는 두 개의 서로 다른 thread_id를 사용하여 두 개의 독립된 대화를 관리하는 방법을 보여줍니다.

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

config는 LangGraph의 그래프 실행 시 설정을 정의하는 딕셔너리입니다. 여기서 configurable 키는 실행 시 동적으로 설정할 수 있는 옵션을 포함합니다.

• thread_id: 대화의 고유 식별자 역할을 합니다.
• LangGraph는 상태 기반 그래프(StateGraph)를 사용하여 대화를 관리합니다.
• thread_id는 각 대화의 상태를 구분하는 데 사용됩니다.
• 동일한 thread_id를 사용하면 이전 대화의 맥락을 유지하며 대화를 이어갈 수 있습니다.
• 서로 다른 thread_id를 사용하면 독립된 대화를 관리할 수 있습니다.

예를 들어:

• thread_id가 "user123"인 경우, 해당 대화의 상태를 기반으로 응답을 생성합니다.
• 새로운 thread_id를 지정하면 이전 대화와는 별개의 새로운 대화가 시작됩니다.

이 설정은 LangGraph의 체크포인팅(checkpointing) 기능과 결합하여 다중 턴 대화에서 맥락을 유지하거나 독립적인 대화를 관리하는 데 유용합니다.

from pprint import pprint
snapshot = graph.get_state(config)
if 'messages' in snapshot.values:
	pprint(snapshot.values['messages'])
else:
	print("No messages found in the snapshot.")
print(snapshot.next)

user_input1 = "AI 에이전트 개발을 위한 LangGraph의 특징에 대해 설명해주세요."
state1 = {"messages": [HumanMessage(content=user_input1)]}
response1 = graph.invoke(state1, config)

print(response1["messages"][-1].content)

snapshot = graph.get_state(config)
if 'messages' in snapshot.values:
	pprint(snapshot.values['messages'])
else:
	print("No messages found in the snapshot.")
print(snapshot.next)

user_input2 = "AI 에이전트 개발을 위한 기술 선택에 대한 전문가의 지원이 필요해요. 지원 요청을 해도 될까요?"
state2 = {"messages": [HumanMessage(content=user_input2)]}
response2 = graph.invoke(state2, config)

print(response2["messages"][-1].content)

snapshot = graph.get_state(config)
if 'messages' in snapshot.values:
	pprint(snapshot.values['messages'])
else:
	print("No messages found in the snapshot.")
print(snapshot.next)

from langgraph.types import Command

human_response = (
    "네, 물론입니다. AI 에이전트 개발을 위한 기술 선택에 대한 지원을 해드리겠습니다. "
    "우선 LangGraph를 사용하는 것에 대해 어떻게 생각하시나요? "
    "LangGraph는 AI 에이전트를 개발하는 데 매우 유용한 도구입니다. "
)

human_command = Command(resume={"data": human_response})
response = graph.invoke(human_command, config)
print(response["messages"][-1].content)

snapshot = graph.get_state(config)
pprint(snapshot.values['messages'])
print(snapshot.next)

user_input3 = "앞서 추천해주신 기술의 시장성은 어떤가요?"
state3 = {"messages": [HumanMessage(content=user_input3)]}
response3 = graph.invoke(state3, config)

print(response3["messages"][-1].content)

snapshot = graph.get_state(config)
pprint(snapshot.values['messages'])
print(snapshot.next)

user_input4 = "LangGraph의 메모리 기능 추가에 대한 전문가의 지원이 필요해요."
state4 = {"messages": [HumanMessage(content=user_input4)]}
response4 = graph.invoke(state4, config)

print(response4["messages"][-1].content)

snapshot = graph.get_state(config)
if 'messages' in snapshot.values:
	pprint(snapshot.values['messages'])
else:
	print("No messages found in the snapshot.")
print(snapshot.next)

human_response = (
    "MemorySaver는 메모리 기반의 체크포인터로, 각 대화의 상태를 메모리에 임시로 저장하고 관리합니다. "
    "이를 통해 챗봇은 이전 대화 내용을 기억하고 다음 번 상호작용 시에도 맥락을 유지한 상태로 대화를 진행할 수 있습니다. "
    "실제 운영 환경에서는 더 영구적인 상태 관리를 위해 데이터베이스 기반 체크포인터(예: SqliteSaver 또는 PostgresSaver)를 사용하는 것이 권장됩니다."
)

human_command = Command(resume={"data": human_response})
response = graph.invoke(human_command, config)
print(response["messages"][-1].content)

snapshot = graph.get_state(config)
pprint(snapshot.values['messages'])
print(snapshot.next)