1 Human-in-the-loop

에이전트는 신뢰할 수 없으며 작업을 성공적으로 수행하기 위해 인간의 입력이 필요할 수 있다.
마찬가지로, 일부 작업에 대해서는 모든 것이 의도한 대로 실행되고 있는지 확인하기 위해 실행 전에 사람이 직접 개입하여 “승인” 을 요구하고 싶을 수 있다.
LangChain을 이용한 Naive RAG 구현시 Human-in-the-loop을 구현하기 불편했다.
LangGraph는 여러 가지 방법으로 human-in-the-loop 워크플로를 지원한다. 이번에는 interrupt_before 기능을 통해 도구 노드 실행 전에 항상 중단하도록 설정할 수 있다.

# API 키를 환경변수로 관리하기 위한 설정 파일
from dotenv import load_dotenv

# API 키 정보 로드
load_dotenv()

# LangSmith 추적을 설정합니다. https://smith.langchain.com
# !pip install -qU langchain-teddynote
from langchain_teddynote import logging

# 프로젝트 이름을 입력합니다.
logging.langsmith("CH17-LangGraph-Modules")

from typing import Annotated, List, Dict
from typing_extensions import TypedDict

from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
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 langchain_teddynote.graphs import visualize_graph
from langchain_teddynote.tools import GoogleNews


########## 1. 상태 정의 ##########
# 상태 정의
class State(TypedDict):
    # 메시지 목록 주석 추가
    messages: Annotated[list, add_messages]


########## 2. 도구 정의 및 바인딩 ##########
# 도구 초기화
# 키워드로 뉴스 검색하는 도구 생성
news_tool = GoogleNews()


@tool
def search_keyword(query: str) -> List[Dict[str, str]]:
    """Look up news by keyword"""
    news_tool = GoogleNews()
    return news_tool.search_by_keyword(query, k=5)


tools = [search_keyword]

# LLM 초기화
llm = ChatOpenAI(model="gpt-4o-mini")

# 도구와 LLM 결합
llm_with_tools = llm.bind_tools(tools)


########## 3. 노드 추가 ##########
# 챗봇 함수 정의
def chatbot(state: State):
    # 메시지 호출 및 반환
    return {"messages": [llm_with_tools.invoke(state["messages"])]}


# 상태 그래프 생성
graph_builder = StateGraph(State)

# 챗봇 노드 추가
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,
)

########## 4. 엣지 추가 ##########

# tools > chatbot
graph_builder.add_edge("tools", "chatbot")

# START > chatbot
graph_builder.add_edge(START, "chatbot")

# chatbot > END
graph_builder.add_edge("chatbot", END)

########## 5. MemorySaver 추가 ##########

# 메모리 저장소 초기화
memory = MemorySaver()

이제 그래프를 컴파일하고, tools 노드 전에 interrupt_before를 지정하십시오.

########## 6. interrupt_before 추가 ##########

# 그래프 빌더 컴파일
graph = graph_builder.compile(checkpointer=memory)

########## 7. 그래프 시각화 ##########
# 그래프 시각화
visualize_graph(graph)

from langchain_teddynote.messages import pretty_print_messages
from langchain_core.runnables import RunnableConfig

# 질문
question = "AI 관련 최신 뉴스를 알려주세요."

# 초기 입력 State 를 정의
input = State(messages=[("user", question)])

# config 설정
config = RunnableConfig(
    recursion_limit=10,  # 최대 10개의 노드까지 방문. 그 이상은 RecursionError 발생
    configurable={"thread_id": "1"},  # 스레드 ID 설정
    tags=["my-rag"],  # Tag
)

for event in graph.stream(
    input=input,
    config=config,
    stream_mode="values",
    interrupt_before=["tools"],  # tools 실행 전 interrupt(tools 노드 실행 전 중단)
):
    for key, value in event.items():
        # key 는 노드 이름
        print(f"\n[{key}]\n")

        # value 는 노드의 출력값
        # print(value)
        pretty_print_messages(value)

        # value 에는 state 가 dict 형태로 저장(values 의 key 값)
        if "messages" in value:
            print(f"메시지 개수: {len(value['messages'])}")

[messages]

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

AI 관련 최신 뉴스를 알려주세요.

[messages]

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

AI 관련 최신 뉴스를 알려주세요.
================================== Ai Message ==================================
Tool Calls:
  search_keyword (call_Axca1RLkFvDATzG0WtrS7T04)
 Call ID: call_Axca1RLkFvDATzG0WtrS7T04
  Args:
    query: AI

Chatbot 노드가 실행된 후, tools 노드가 실행되기 전에 중단된 것을 볼 수 있다.
그래프 상태를 확인하여 제대로 작동했는지 확인해 본다.

# 그래프 상태 스냅샷 생성
snapshot = graph.get_state(config)

# 다음 스냅샷 상태
snapshot.next

('tools',)

(이전 블로그에서는) __END__ 도달했기 때문에 .next 가 존재하지 않았다.
하지만, 지금은 .next 가 tools 로 지정되어 있다.
도구 호출을 확인

from langchain_teddynote.messages import display_message_tree

# 메시지 스냅샷에서 마지막 메시지 추출
existing_message = snapshot.values["messages"][-1]

# 메시지 트리 표시
display_message_tree(existing_message.tool_calls)

    index [0]
        name: "search_keyword"
        args: {"query": "AI"}
        id: "call_Axca1RLkFvDATzG0WtrS7T04"
        type: "tool_call"

Chatbot 노드가 질문에서 AI 라는 키워드를 인식하여 search_keyword 도구에 검색 키워드 정보를 주고 있다.
다음으로는 이전에 종료된 지점 이후부터 이어서 그래프를 진행 해 본다.
중단 후 재개: LangGraph 는 계속 그래프를 진행하는 것 쉽게 할 수 있다.
- 단지 입력에 None을 전달하면 된다.
- 처음 부터 다시 시작하는 것이 아니라, 마지막으로 중단된 지점에서 이어서 실행이 된다.

# `None`는 현재 상태에 아무것도 추가하지 않음
events = graph.stream(None, config, stream_mode="values")

# 이벤트 반복 처리
for event in events:
    # 메시지가 이벤트에 포함된 경우
    if "messages" in event:
        # 마지막 메시지의 예쁜 출력
        event["messages"][-1].pretty_print()

================================== Ai Message ==================================
Tool Calls:
  search_keyword (call_Axca1RLkFvDATzG0WtrS7T04)
 Call ID: call_Axca1RLkFvDATzG0WtrS7T04
  Args:
    query: AI
================================= Tool Message =================================
Name: search_keyword

[{"url": "https://news.google.com/rss/articles/CBMidkFVX3", "content": "바다 속에 데이터센터를?… 발열과 사투 벌이는 AI 기업 - 동아일보"}, {"url": "https://news.google.com/rss/articles/CBMiakFVX3", "content": "오픈AI, '챗GPT'에 검색 기능 정식 출시...구글은 '개발자용 검색 지원'으로 맞불 - AI타임스"}, {"url": "https://news.google.com/rss/articles/CBMiswJBVV", "content": "칼럼 | 악화하는 오픈AI-마이크로소프트의 관계 - CIO.com"}, {"url": "https://news.google.com/rss/articles/CBMigw", "content": "“AI 활용한 무기 가공할 위력… 오용되면 끝장” - 조선일보"}, {"url": "https://news.google.com/rss/articles/CBMiYEFVX", "content": "네이버·이통3사 이어 카카오도 AI戰 합류…AI, 韓 IT 성장동력 될까[돈 버는 AI 下] - 뉴시스"}]
================================== Ai Message ==================================

최근 AI 관련 뉴스는 다음과 같습니다:

1. **[바다 속에 데이터센터를?… 발열과 사투 벌이는 AI 기업](https://news.google.com/rss/articles/CBMidkFVX)** - AI 기업들이 발열 문제를 해결하기 위해 바다 속에 데이터센터를 설치하는 방안에 대해 다루고 있습니다.

2. **[오픈AI, '챗GPT'에 검색 기능 정식 출시...구글은 '개발자용 검색 지원'으로 맞불](https://news.google.com/rss/articles/CBMiakF)** - 오픈AI가 챗GPT에 새로운 검색 기능을 추가한 반면, 구글은 개발자용 검색 지원을 강화하고 있다는 소식입니다.

3. **[칼럼 | 악화하는 오픈AI-마이크로소프트의 관계](https://news.google.com/rss/articles/CBMiswJBV)** - 오픈AI와 마이크로소프트 간의 관계가 악화되고 있다는 내용을 다룬 칼럼입니다.

4. **[“AI 활용한 무기 가공할 위력… 오용되면 끝장”](https://news.google.com/rss/articles/CBMigwFBV)** - AI의 힘을 무기로 가공할 수 있지만, 잘못 사용될 경우 심각한 결과를 초래할 수 있다는 경고가 담긴 기사입니다.

5. **[네이버·이통3사 이어 카카오도 AI戰 합류…AI, 韓 IT 성장동력 될까[돈 버는 AI 下]](https://news.google.com/rss/articles/CBMiYEFVX)** - 카카오가 AI 전쟁에 합류하면서 AI가 한국 IT의 성장 동력이 될 수 있을지에 대한 논의입니다.

이 외에도 다양한 AI 관련 뉴스가 보도되고 있습니다. 관심 있는 주제에 대해 더 깊이 알아보시기 바랍니다!

이제, interrupt를 사용하여 챗봇에 인간이 개입할 수 있는 실행을 추가하여 필요할 때 인간의 감독과 개입을 가능하게 했다. 이는 추후에 시스템으로 구현할때, 잠재적인 UI를 제공할 수 있다.
이미 checkpointer를 추가했기 때문에, 그래프는 무기한 일시 중지되고 언제든지 다시 시작할 수 있다.
get_state_history 메서드를 사용하여 그래프의 상태 기록을 가져오고, 원하는 지점에서 다시 시작할 수 있다.
상태 기록을 통해 원하는 상태를 지정하여 해당 지점에서 다시 시작 할 수 있다.

to_replay = None

# 상태 기록 가져오기
for state in graph.get_state_history(config):
    # 메시지 수 및 다음 상태 출력
    print("메시지 수: ", len(state.values["messages"]), "다음 노드: ", state.next)
    print("-" * 80)
    # 특정 상태 선택 기준: 채팅 메시지 수
    if len(state.values["messages"]) == 3:
        to_replay = state

메시지 수:  4 다음 노드:  ()
--------------------------------------------------------------------------------
메시지 수:  3 다음 노드:  ('chatbot',)
--------------------------------------------------------------------------------
메시지 수:  2 다음 노드:  ('tools',)
--------------------------------------------------------------------------------
메시지 수:  1 다음 노드:  ('chatbot',)
--------------------------------------------------------------------------------
메시지 수:  0 다음 노드:  ('__start__',)
--------------------------------------------------------------------------------

그래프의 모든 단계에 대해 체크포인트가 저장된다는 점에 주목 할 필요가 있다.
원하는 지점은 to_replay 변수에 저장된다. 이를 활용하여 다시 시작할 수 있는 지점을 지정할 수 있다.

# 다음 항목의 다음 요소 출력
print(to_replay.next)

# 다음 항목의 설정 정보 출력
print(to_replay.config)

('chatbot',)
{'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1ef9951b-f2ef-6676-8002-fc3f0866298b'}}

to_replay.config 에 checkpoint_id 가 포함되어 있다.

to_replay.config

{'configurable': {'thread_id': '1',  'checkpoint_ns': '',  'checkpoint_id': '1ef9951b-f2ef-6676-8002-fc3f0866298b'}}

이 checkpoint_id 값을 제공하면 LangGraph의 체크포인터가 그 시점의 상태를 로드 할 수 있다.
- 단, 이때는 입력값을 None으로 전달해야 한다.

# `to_replay.config`는 `checkpoint_id`는 체크포인터에 저장된 상태에 해당
for event in graph.stream(None, to_replay.config, stream_mode="values"):
    # 메시지가 이벤트에 포함된 경우
    if "messages" in event:
        # 마지막 메시지 출력
        event["messages"][-1].pretty_print()

================================= Tool Message =================================
Name: search_keyword

[{"url": "https://news.google.com/rss/articles/CBMidkFV", "content": "바다 속에 데이터센터를?… 발열과 사투 벌이는 AI 기업 - 동아일보"}, {"url": "https://news.google.com/rss/articles/CBMiakFVX", "content": "오픈AI, '챗GPT'에 검색 기능 정식 출시...구글은 '개발자용 검색 지원'으로 맞불 - AI타임스"}, {"url": "https://news.google.com/rss/articles/CBMiswJB", "content": "칼럼 | 악화하는 오픈AI-마이크로소프트의 관계 - CIO.com"}, {"url": "https://news.google.com/rss/articles/CBMigwFBV", "content": "“AI 활용한 무기 가공할 위력… 오용되면 끝장” - 조선일보"}, {"url": "https://news.google.com/rss/articles/CBMiYEFV", "content": "네이버·이통3사 이어 카카오도 AI戰 합류…AI, 韓 IT 성장동력 될까[돈 버는 AI 下] - 뉴시스"}]
================================== Ai Message ==================================

다음은 AI 관련 최신 뉴스입니다:

1. [바다 속에 데이터센터를?… 발열과 사투 벌이는 AI 기업 - 동아일보](https://news.google.com/rss/articles/CBMidkFVX)

2. [오픈AI, '챗GPT'에 검색 기능 정식 출시...구글은 '개발자용 검색 지원'으로 맞불 - AI타임스](https://news.google.com/rss/articles/CBMiakFV)

3. [칼럼 | 악화하는 오픈AI-마이크로소프트의 관계 - CIO.com](https://news.google.com/rss/articles/CBMiswJB)

4. [“AI 활용한 무기 가공할 위력… 오용되면 끝장” - 조선일보](https://news.google.com/rss/articles/CBMigwFBV)

5. [네이버·이통3사 이어 카카오도 AI戰 합류…AI, 韓 IT 성장동력 될까[돈 버는 AI 下] - 뉴시스](https://news.google.com/rss/articles/CBMiYEFV)

이 뉴스들은 최근 AI와 관련된 다양한 이슈를 다루고 있습니다. 관심 있는 기사를 클릭하여 더 자세한 내용을 확인해 보세요.

to_replay 지점: 검색이 끝나고 검색 결과가 chatbot 노드에 반환된 시점
checkpoint_id를 사용하여 해당 시점에서 그래프를 다시 시작할 수 있다.