[langgraph] 상태 커스터마이징
원문 출처: https://langchain-ai.github.io/langgraph/tutorials/get-started/5-customize-state/
이 튜토리얼에서는 메시지 리스트에만 의존하지 않고, 상태에 추가 필드를 정의하여 더 복잡한 동작을 구현하는 방법을 다룹니다. 챗봇은 검색 도구를 사용해 특정 정보를 찾고, 이를 인간에게 검토받는 흐름을 예시로 보여줍니다.
이 튜토리얼은 human-in-the-loop 추가하기를 기반으로 합니다.
1. 상태에 키 추가하기
챗봇이 엔터티의 생일을 조사하도록 name과 birthday 키를 상태에 추가합니다:
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages
class State(TypedDict):
messages: Annotated[list, add_messages]
name: str
birthday: str이렇게 하면 다른 노드나 영속성 계층에서 이 정보를 쉽게 활용할 수 있습니다.
2. 도구 내부에서 상태 업데이트하기
이제 human_assistance 도구 내부에서 상태 키를 채웁니다. 인간이 정보를 검토한 뒤 상태에 반영할 수 있도록 Command를 반환합니다.
from langchain_core.messages import ToolMessage
from langchain_core.tools import InjectedToolCallId, tool
from langgraph.types import Command, interrupt
@tool
def human_assistance(
name: str, birthday: str, tool_call_id: Annotated[str, InjectedToolCallId]
) -> str:
"""Request assistance from a human."""
human_response = interrupt(
{
"question": "Is this correct?",
"name": name,
"birthday": birthday,
},
)
# 정보가 맞으면 그대로 상태에 반영
if human_response.get("correct", "").lower().startswith("y"):
verified_name = name
verified_birthday = birthday
response = "Correct"
# 아니면 인간이 수정한 값을 반영
else:
verified_name = human_response.get("name", name)
verified_birthday = human_response.get("birthday", birthday)
response = f"Made a correction: {human_response}"
state_update = {
"name": verified_name,
"birthday": verified_birthday,
"messages": [ToolMessage(response, tool_call_id=tool_call_id)],
}
return Command(update=state_update)나머지 그래프 구조는 동일합니다.
3. 챗봇에 프롬프트 입력하기
챗봇에게 LangGraph 라이브러리의 "생일"을 조사하고, 답을 얻으면 human_assistance 도구로 검토를 요청하도록 지시합니다. tool 인자에 name과 birthday를 명시하면 챗봇이 이 값을 제안하게 할 수 있습니다.
user_input = (
"Can you look up when LangGraph was released? "
"When you have the answer, use the human_assistance tool for review."
)
config = {"configurable": {"thread_id": "1"}}
events = graph.stream(
{"messages": [{"role": "user", "content": user_input}]},
config,
stream_mode="values",
)
for event in events:
if "messages" in event:
event["messages"][-1].pretty_print()실행 결과 예시:
================================ Human Message =================================
Can you look up when LangGraph was released? When you have the answer, use the human_assistance tool for review.
================================== Ai Message ==================================
[{'text': "Certainly! I'll start by searching for information about LangGraph's release date using the Tavily search function. Then, I'll use the human_assistance tool for review.", 'type': 'text'}, ...]4. human assistance로 검토하기
챗봇이 정확한 날짜를 찾지 못했다면, 정보를 제공해 검토를 도와줄 수 있습니다.
from langgraph.types import Command
human_command = Command(
resume={
"name": "LangGraph",
"birthday": "Jan 17, 2024",
},
)
events = graph.stream(human_command, config, stream_mode="values")
for event in events:
if "messages" in event:
event["messages"][-1].pretty_print()실행 결과 예시:
================================== Ai Message ==================================
Thank you for the human assistance. I can now provide you with the correct information about LangGraph's release date.
LangGraph was initially released on January 17, 2024. This information comes from the human assistance correction, which is more accurate than the search results I initially found.
...이제 상태에 해당 필드가 반영됩니다:
snapshot = graph.get_state(config)
{k: v for k, v in snapshot.values.items() if k in ("name", "birthday")}출력 예시:
{'name': 'LangGraph', 'birthday': 'Jan 17, 2024'}5. 상태 수동 업데이트하기
LangGraph는 애플리케이션 상태를 직접 제어할 수 있습니다. 예를 들어, 언제든(중단 중에도) graph.update_state로 값을 덮어쓸 수 있습니다.
graph.update_state(config, {"name": "LangGraph (library)"})6. 새로운 값 확인하기
graph.get_state로 상태를 확인하면 값이 반영된 것을 볼 수 있습니다:
snapshot = graph.get_state(config)
{k: v for k, v in snapshot.values.items() if k in ("name", "birthday")}출력 예시:
{'name': 'LangGraph (library)', 'birthday': 'Jan 17, 2024'}수동 상태 업데이트는 LangSmith에 트레이스를 남깁니다. human-in-the-loop 워크플로우 제어에도 활용할 수 있지만, 일반적으로는 interrupt를 사용하는 것이 권장됩니다.
축하합니다! 이제 상태에 커스텀 키를 추가해 더 복잡한 워크플로우를 구현할 수 있게 되었고, 도구 내부에서 상태 업데이트를 생성하는 방법도 배웠습니다.
아래는 튜토리얼 전체 코드 예시입니다:
from typing import Annotated
from langchain_tavily import TavilySearch
from langchain_core.messages import ToolMessage
from langchain_core.tools import InjectedToolCallId, tool
from typing_extensions import TypedDict
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 langgraph.types import Command, interrupt
class State(TypedDict):
messages: Annotated[list, add_messages]
name: str
birthday: str
@tool
def human_assistance(
name: str, birthday: str, tool_call_id: Annotated[str, InjectedToolCallId]
) -> str:
human_response = interrupt(
{
"question": "Is this correct?",
"name": name,
"birthday": birthday,
},
)
if human_response.get("correct", "").lower().startswith("y"):
verified_name = name
verified_birthday = birthday
response = "Correct"
else:
verified_name = human_response.get("name", name)
verified_birthday = human_response.get("birthday", birthday)
response = f"Made a correction: {human_response}"
state_update = {
"name": verified_name,
"birthday": verified_birthday,
"messages": [ToolMessage(response, tool_call_id=tool_call_id)],
}
return Command(update=state_update)
tool = TavilySearch(max_results=2)
tools = [tool, human_assistance]
llm_with_tools = llm.bind_tools(tools)
def chatbot(state: State):
message = llm_with_tools.invoke(state["messages"])
assert(len(message.tool_calls) <= 1)
return {"messages": [message]}
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,
)
graph_builder.add_edge("tools", "chatbot")
graph_builder.add_edge(START, "chatbot")
memory = MemorySaver()
graph = graph_builder.compile(checkpointer=memory)이제 LangGraph 기본 튜토리얼의 마지막 개념인 체크포인팅과 상태 업데이트를 활용한 타임트래블로 넘어갈 수 있습니다.
반응형
