LangGraph 튜토리얼 완벽 정리(2)

이번 포스팅에서는 지난 포스팅에 이어 LangGraph 튜토리얼에 대해 정리해보겠습니다. 이 포스팅은 공식 튜토리얼 문서를 참고했습니다.

목차는 다음과 같습니다.

• Manually Updating the State
• Customizing State
• Time Travel

5. Manually Updating the State

이전 섹션에서는 그래프를 중단하여 사람이 에이전트의 상태를 확인하는 방법을 알아보았습니다.
이번에는 update_state 기능을 사용해 상태를 수동으로 수정하여 에이전트의 경로를 변경하거나 원하는 결과로 조정하는 방법을 다룹니다. 이 기능은 에이전트의 실수를 수정하거나, 다른 경로를 탐색하거나, 특정 목표로 유도하는 데 유용합니다.

1. 기존 그래프 설정

먼저, 이전과 동일한 그래프를 정의합니다.

from typing import Annotated

from langchain_anthropic import ChatAnthropic
from langchain_community.tools.tavily_search import TavilySearchResults
from typing_extensions import TypedDict

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition


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


graph_builder = StateGraph(State)


tool = TavilySearchResults(max_results=2)
tools = [tool]
llm = ChatAnthropic(model="claude-3-5-sonnet-20240620")
llm_with_tools = llm.bind_tools(tools)


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


graph_builder.add_node("chatbot", chatbot)

tool_node = ToolNode(tools=[tool])
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,
    # This is new!
    interrupt_before=["tools"],
    # Note: can also interrupt **after** actions, if desired.
    # interrupt_after=["tools"]
)

2. 그래프 실행 및 상태 확인

다음 코드를 통해 사용자 입력을 받습니다. tools 노드에서 작업을 중단하도록 설정되어 있습니다.

user_input = "I'm learning LangGraph. Could you do some research on it for me?"
config = {"configurable": {"thread_id": "1"}}
events = graph.stream({"messages": [("user", user_input)]}, config)
for event in events:
    if "messages" in event:
        event["messages"][-1].pretty_print()

출력 예:

Ai Message: Certainly! I'd be happy to research LangGraph for you...
Tool Calls: tavily_search_results_json

3. 직접 응답 제공으로 상태 업데이트

원하는 답변을 직접 입력해 상태를 업데이트할 수 있습니다.

from langchain_core.messages import AIMessage, ToolMessage

answer = "LangGraph is a library for building stateful, multi-actor applications with LLMs."
new_messages = [
    ToolMessage(content=answer, tool_call_id=existing_message.tool_calls[0]["id"]),
    AIMessage(content=answer),
]

graph.update_state(config, {"messages": new_messages})

4. 업데이트된 메시지 확인

상태를 업데이트한 후, 마지막 메시지를 확인하여 변경 사항이 반영되었는지 검토합니다.

print("\n\nLast 2 messages;")
print(graph.get_state(config).values["messages"][-2:])

출력 예:

[ToolMessage(content='LangGraph is a library...', tool_call_id='toolu_018YcbFR37CG8RRXnavH5fxZ'), AIMessage(content='LangGraph is a library...')]

5. 특정 노드 기반으로 업데이트 실행

as_node 매개변수를 사용해 chatbot 노드에서 발생한 응답처럼 처리할 수 있습니다.

graph.update_state(
    config,
    {"messages": [AIMessage(content="I'm an AI expert!")]},
    as_node="chatbot",
)

6. 기존 메시지 덮어쓰기

특정 메시지를 덮어쓰려면 동일한 ID를 사용하여 새 메시지를 업데이트합니다. 예를 들어, 기존의 tool 호출 내용을 변경하려면 다음과 같이 합니다.

snapshot = graph.get_state(config)
existing_message = snapshot.values["messages"][-1]
new_tool_call = existing_message.tool_calls[0].copy()
new_tool_call["args"]["query"] = "LangGraph human-in-the-loop workflow"
new_message = AIMessage(
    content=existing_message.content,
    tool_calls=[new_tool_call],
    id=existing_message.id,
)

graph.update_state(config, {"messages": [new_message]})

7. 그래프 재개

중단된 상태에서 None을 전달하여 그래프가 이어서 실행되도록 합니다.

events = graph.stream(None, config, stream_mode="values")
for event in events:
    if "messages" in event:
        event["messages"][-1].pretty_print()

6. Customizing State

지금까지는 메시지 목록 형태의 간단한 상태만 사용했습니다. 이 단순한 상태만으로도 여러 작업을 수행할 수 있지만, 메시지 리스트 외의 필드를 추가하여 복잡한 동작을 정의할 수 있습니다.
이번 섹션에서는 상태를 확장해 챗봇이 특정 상황에서 인간에게 도움을 요청할 수 있도록 설정합니다.

1. 새로운 state 정의

기존 상태에 ask_human 플래그를 추가하여, LLM이 “RequestAssistance” 툴을 호출하면 이 플래그를 활성화하도록 합니다.

from typing import Annotated
from langchain_anthropic import ChatAnthropic
from langchain_community.tools.tavily_search import TavilySearchResults
from typing_extensions import TypedDict
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START
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

2. RequestAssistance 모델 정의

RequestAssistance 모델을 정의하여 사용자가 도움을 요청할 때 적절한 메시지를 전달합니다.

from pydantic import BaseModel

class RequestAssistance(BaseModel):
    """지원 요청에 대한 모델 정의."""
    request: str

3. 챗봇 노드 정의 및 ask_human 플래그 정의

챗봇이 RequestAssistance 플래그를 호출할 경우 ask_human 플래그를 활성화하여 상태를 업데이트합니다.

tool = TavilySearchResults(max_results=2)
tools = [tool]
llm = ChatAnthropic(model="claude-3-5-sonnet-20240620")
llm_with_tools = llm.bind_tools(tools + [RequestAssistance])

def chatbot(state: State):
    response = llm_with_tools.invoke(state["messages"])
    ask_human = False
    if response.tool_calls and response.tool_calls[0]["name"] == RequestAssistance.__name__:
        ask_human = True
    return {"messages": [response], "ask_human": ask_human}

4. 그래프 빌더 및 노드 추가

챗봇과 tool 노드를 그래프에 추가합니다.

graph_builder = StateGraph(State)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_node("tools", ToolNode(tools=[tool]))

5. human 노드 추가

human_node는 주로 placeholder로 작동하며, 중단 시 인간이 상태를 업데이트하지 않으면 “응답 없음” 메시지를 삽입하고 ask_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"],
    )

def human_node(state: State):
    new_messages = []
    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,
    }

graph_builder.add_node("human", human_node)

6. 조건부 로직 정의

select_next_node 함수는 ask_human 플래그가 설정된 경우 human 노드로 전환하고, 그렇지 않으면 기번 tools_condition을 사용해 다음 노드를 선택합니다.

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

graph_builder.add_conditional_edges(
    "chatbot",
    select_next_node,
    {"human": "human", "tools": "tools", END: END},
)

엣지 정의 및 그래프 컴파일

간단한 방향 엣지를 추가하고, 그래프를 컴파일할 때 interrupt_before 옵션을 설정해 human 노드에서 중단되도록 합니다.

graph_builder.add_edge("tools", "chatbot")
graph_builder.add_edge("human", "chatbot")
graph_builder.add_edge(START, "chatbot")
memory = MemorySaver()
graph = graph_builder.compile(
    checkpointer=memory,
    interrupt_before=["human"],
)

8. 그래프 실행 예시

이제 사용자가 “전문가 지원 요청”을 요구하는 메시지를 입력하여 예제를 실행합니다.

user_input = "I need some expert guidance for building this AI agent. Could you request assistance for me?"
config = {"configurable": {"thread_id": "1"}}
events = graph.stream({"messages": [("user", user_input)]}, config, stream_mode="values")
for event in events:
    if "messages" in event:
        event["messages"][-1].pretty_print()

출력 예:

Ai Message: Certainly! I understand that you need expert guidance...
Tool Calls: RequestAssistance

9. 상태 확인 및 업데이트

중단된 상태에서 snapshot을 확인하고, RequestAssistance 호출에 대한 응답을 제공합니다.

snapshot = graph.get_state(config)
ai_message = snapshot.values["messages"][-1]
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."
)
tool_message = create_response(human_response, ai_message)
graph.update_state(config, {"messages": [tool_message]})

10. 그래프 재개

중단된 상태에서 None을 입력하여 그래프를 이어서 실행합니다.

events = graph.stream(None, 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 escalated your request to our expert team, and they have provided some initial guidance...

이제 챗봇이 ask_human 플래그에 따라 인간의 지원을 요청하거나, 직접 응답을 저공할 수 있게 되었습니다.
모든 상태가 체크포인트로 저장되기 때문에, 인간이 상태를 수정하는 데 시간이 걸리더라도 그래프의 실행에는 영향을 미치지 않습니다.

7. Time Travel

지금까지는 상태를 체크포인트에 저장하고, 상태를 수동으로 수정하여 미래의 응답을 제어하는 방법을 배웠습니다.
이번에는 LangGraph의 time travel 기능을 통해 이전 상태로 돌아가 다른 결과를 탐색하거나 실수를 수정하는 등의 작업을 해보겠습니다.

time travel 기능을 통해 사용자가 이전 응답 지점에서 시작해 다른 결과를 탐색하거나, 작업을 되돌려 실수를 수정하고 다른 전략을 시도하는 등 다양한 경험을 제공할 수 있습니다.

1. 기존 그래프 로드

기존에 작성행 챗봇 그래프를 그대로 사용합니다.

from typing import Annotated, Literal
from langchain_anthropic import ChatAnthropic
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.messages import AIMessage, ToolMessage
from pydantic import BaseModel
from typing_extensions import TypedDict

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START
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


class RequestAssistance(BaseModel):
    request: str


tool = TavilySearchResults(max_results=2)
tools = [tool]
llm = ChatAnthropic(model="claude-3-5-sonnet-20240620")
llm_with_tools = llm.bind_tools(tools + [RequestAssistance])

def chatbot(state: State):
    response = llm_with_tools.invoke(state["messages"])
    ask_human = False
    if response.tool_calls and response.tool_calls[0]["name"] == RequestAssistance.__name__:
        ask_human = True
    return {"messages": [response], "ask_human": ask_human}


graph_builder = StateGraph(State)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_node("tools", ToolNode(tools=[tool]))


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


def human_node(state: State):
    new_messages = []
    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}

graph_builder.add_node("human", human_node)


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

graph_builder.add_conditional_edges(
    "chatbot",
    select_next_node,
    {"human": "human", "tools": "tools", END: END},
)
graph_builder.add_edge("tools", "chatbot")
graph_builder.add_edge("human", "chatbot")
graph_builder.add_edge(START, "chatbot")
memory = MemorySaver()
graph = graph_builder.compile(checkpointer=memory, interrupt_before=["human"])

2. 그래프 실행 및 체크포인트 생성

여러 단계를 수행하여 그래프가 여러 상태로 체크포인트를 남기도록 합니다.

config = {"configurable": {"thread_id": "1"}}
events = graph.stream(
    {"messages": [("user", "I'm learning LangGraph. Could you do some research on it for me?")]},
    config,
    stream_mode="values",
)
for event in events:
    if "messages" in event:
        event["messages"][-1].pretty_print()

events = graph.stream(
    {"messages": [("user", "Ya that's helpful. Maybe I'll build an autonomous agent with it!")]},
    config,
    stream_mode="values",
)
for event in events:
    if "messages" in event:
        event["messages"][-1].pretty_print()

3. 상태 이력 조회 및 특정 체크포인트 선택

get_state_history 메서드를 사용하여 각 단계의 체크포인트를 조회한 뒤, 특정 지점으로 돌아가 새로 작업을 수행할 수 있습니다.

to_replay = None
for state in graph.get_state_history(config):
    print("Num Messages: ", len(state.values["messages"]), "Next: ", state.next)
    print("-" * 80)
    if len(state.values["messages"]) == 6:
        to_replay = state

여기서 to_replay는 두 번째 그래프 실행 이후 chatbot 노드가 완료된 상태입니다. 다음 노드인 tools에서부터 실행이 이어질 것입니다.

4. 선택한 체크포인트에서 그래프 실행 재개

선택한 to_replay 체크포인트의 설정을 사용해 그래프를 재개합니다.

for event in graph.stream(None, to_replay.config, stream_mode="values"):
    if "messages" in event:
        event["messages"][-1].pretty_print()

출력 예:

Ai Message: Great choice! Building an autonomous agent with LangGraph is an excellent way to dive deep into its capabilities...

이제 특정 시점으로 돌아가 그래프 실행을 이어갈 수 있게 되었습니다. 이를 통해 디버깅, 실험 및 대화형 애플리케이션에서 다양한 시나리오를 시험할 수 있는 기능을 제공받았습니다.

여기까지 LangGraph에서 제공하는 튜토리얼을 모두 마쳤습니다. LangGraph를 통해 기존의 chain만으로 구성했던 챗봇을 좀 더 다이나믹하고 유연하게 수정할 수 있을 것으로 기대됩니다.

긴 글 읽어주셔서 감사합니다.