langgraph의 공식문서를 번역해 놓은 자료입니다. 이해하기 쉽게 예제는 변경하였습니다. 또한 필요한 경우 부연 설명을 추가하였습니다. 문제가 되면 삭제하겠습니다.
https://langchain-ai.github.io/langgraph/tutorials/introduction/
Part 5: 상태(State)를 수동으로 업데이트 하기
이전 섹션에서는 그래프를 중단하여 사람이 그 작업을 확인할 수 있는 방법을 보여주었다. 이를 통해 사람은 상태를 읽을(read) 수 있지만, 에이전트의 경로를 변경하려면 쓰기(write) 권한이 필요하다.
다행히도 LangGraph를 사용하면 상태를 수동으로 업데이트할 수 있다. 상태를 업데이트하면 에이전트의 행동을 수정하여 에이전트의 궤적을 제어할 수 있다(심지어 과거를 수정할 수도 있다). 이 기능은 에이전트의 실수를 수정하거나 대안 경로를 탐색하거나 에이전트를 특정 목표로 안내하려고 할 때 특히 유용하다.
아래에서 체크포인트 상태를 업데이트하는 방법을 보여준다. 이전과 마찬가지로 먼저 그래프를 정의한다. 이전과 동일한 그래프를 재사용할 것이다.
from typing import Annotated
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_openai import ChatOpenAI
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
from typing_extensions import TypedDict
class State(TypedDict):
messages: Annotated[list, add_messages]
graph_builder = StateGraph(State)
tool = TavilySearchResults(max_results=2)
tools = [tool]
llm = ChatOpenAI(model="gpt-3.5-turbo")
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,
interrupt_before=["tools"],
)
user_input = "지금 LangGraph를 공부하고 있어. LangGraph에 대해 찾아줄 수 있어?"
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()snapshot = graph.get_state(config)
existing_message = snapshot.values["messages"][-1]
existing_message.pretty_print()================================== Ai Message ==================================
Tool Calls:
tavily_search_results_json (call_koW8yErRm93Fh3WzobNsr6Cf)
Call ID: call_koW8yErRm93Fh3WzobNsr6Cf
Args:
query: LangGraph지금까지 모든 내용은 이전 섹션에서 했던 것과 동일하다. LLM이 검색 엔진 도구를 사용하겠다고 요청했으며, 그래프가 중단되었다. 이전과 같이 진행하면 도구가 호출되어 웹을 검색하게 된다.
하지만 사용자가 개입하고 싶어 한다면 어떻게 될까? 챗봇이 도구를 사용할 필요가 없다고 생각한다면?
이제 직접 올바른 응답을 제공해 보자.
from langchain_core.messages import AIMessage, ToolMessage
answer = "LangGraph는 아주 아주 좋은 라이브러리야~~~"
new_messages = [
# LLM API는 도구 호출과 일치하는 ToolMessage를 기대한다. 여기서 그것을 충족시킬 것이다.
ToolMessage(content=answer, tool_call_id=existing_message.tool_calls[0]["id"]),
# 이후에는 LLM의 응답에 직접 추가한다.
AIMessage(content=answer),
]
new_messages[-1].pretty_print()
graph.update_state(
config,
# 업데이트 값. 우리의 `State`의 메시지는 "append-only"이다. 이것은 기존 상태에 추가될 것이다. 다음 섹션에서 기존 메시지를 업데이트하는 방법을 알아볼 것이다!
{"messages": new_messages},
)
print("\n\nLast 2 messages;")
print(graph.get_state(config).values["messages"][-2:])================================== Ai Message ==================================
LangGraph는 아주 아주 좋은 라이브러리야~~~
Last 2 messages;
[ToolMessage(content='LangGraph는 아주 아주 좋은 라이브러리야~~~', id='d144bfbc-72ae-49fa-b909-288686757d6d', tool_call_id='call_Fq6XQggxWxHkueg44MHiYFAF'), AIMessage(content='LangGraph는 아주 아주 좋은 라이브러리야~~~', additional_kwargs={}, response_metadata={}, id='84824ee0-2abb-4a1a-967c-ed991197dee8')]이제 최종 응답 메시지를 제공했으므로 그래프가 완성되었다. 상태 업데이트는 그래프 단계를 시뮬레이션하기 때문에 해당하는 트레이스를 생성한다. 위의 update_state 호출의 LangSmith 트레이스를 검사하여 어떤 일이 일어나는지 확인하자.
새로운 메시지가 상태에 이미 있는 메시지에 추가된 것을 확인하자. 상태 유형을 어떻게 정의했는지 기억하나?
class State(TypedDict):
messages: Annotated[list, add_messages]우리는 메시지를 add_messages 함수로 사용했다. 이는 그래프에 기존 목록에 값을 항상 추가하며, 목록을 직접 덮어쓰지 않도록 한다. 동일한 논리가 여기에도 적용되므로, update_state에 전달된 메시지는 같은 방식으로 추가된다.
update_state 함수는 마치 그래프의 노드 중 하나처럼 작동한다. 기본적으로 업데이트 작업은 마지막으로 실행된 노드를 사용하지만, 아래에서 수동으로 지정할 수 있다. 업데이트를 추가하고 그래프에 이를 "챗봇"에서 온 것처럼 처리하도록 지시해 보자.
graph.update_state(
config,
{"messages": [AIMessage(content="나는 AI 전문가야")]},
# 이 기능이 어느 노드에서 동작할 것인가?
# 이것은 마치 이 노드가 방금 실행된 것처럼 처리될 것이다.
as_node="chatbot",
){'configurable': {'thread_id': '1',
'checkpoint_ns': '',
'checkpoint_id': '1ef7d134-3958-6412-8002-3f4b4112062f'}}제공된 링크에서 이 업데이트 호출에 대한 LangSmith 트레이스를 확인하자. 트레이스에서 그래프가 tools_condition 엣지로 계속 진행되는 것을 확인하자. 우리는 업데이트를 as_node="chatbot"으로 처리하도록 그래프에 지시했다. 아래 다이어그램에서 챗봇 노드에서 시작하면 자연스럽게 tools_condition 엣지로 이어지고, 업데이트된 메시지에 도구 호출이 없기 때문에 결국 __end__에 도달하게 된다.
from IPython.display import Image, display
try:
display(
Image(
graph.get_graph().draw_mermaid_png(
output_file_path="./manually_updating_the_state.png"
)
)
)
except Exception:
# This requires some extra dependencies and is optional
pass
이전과 같이 현재 상태를 검사하여 체크포인트가 우리의 수동 업데이트를 반영하는지 확인하자.
snapshot = graph.get_state(config)
print(snapshot.values["messages"][-3:])
print(snapshot.next)[
ToolMessage(content='LangGraph는 아주 아주 좋은 라이브러리야~~~', id='cae96ef9-85f8-407e-98be-296776a84c07', tool_call_id='call_kLnImi2UNZXMnr67FtjoH6yX'),
AIMessage(content='LangGraph는 아주 아주 좋은 라이브러리야~~~', additional_kwargs={}, response_metadata={}, id='3db3472e-de0e-44c0-a7d9-50db19714c01'),
AIMessage(content='나는 AI 전문가야', additional_kwargs={}, response_metadata={}, id='6dadc695-d1e5-486b-993c-3a4cdb86ca58')
]Notice: 상태에 AI 메시지를 계속 추가했다. 우리가 챗봇 역할을 하여 도구 호출이 포함되지 않은 AIMessage로 응답하기 때문에, 그래프는 완료된 상태에 있다(다음은 비어 있다).
기존 메시지를 덮어쓰고 싶다면 어떻게 할까?
위에서 그래프의 상태에서 사용한 add_messages 함수는 메시지 키에 대한 업데이트가 어떻게 이루어지는지를 제어한다. 이 함수는 새로운 메시지 목록에 있는 메시지 ID를 확인한다. 만약 ID가 기존 상태의 메시지와 일치하면, add_messages는 기존 메시지를 새로운 내용으로 덮어쓴다.
예를 들어, 도구 호출을 업데이트하여 검색 엔진에서 좋은 결과를 얻도록 해보자. 먼저 새로운 스레드를 시작한다.
user_input = "지금 LangGraph를 공부하고 있어. LangGraph에 대해 찾아줄 수 있어?"
config = {"configurable": {"thread_id": "2"}} # 여기서는 thread_id = 2 를 사용한다
events = graph.stream(
{"messages": [("user", user_input)]}, config, stream_mode="values"
)
for event in events:
if "messages" in event:
event["messages"][-1].pretty_print()================================ Human Message =================================
지금 LangGraph를 공부하고 있어. LangGraph에 대해 찾아줄 수 있어?
================================== Ai Message ==================================
Tool Calls:
tavily_search_results_json (call_HR1XZ4UfuMgUrpm7pzgXdXnR)
Call ID: call_HR1XZ4UfuMgUrpm7pzgXdXnR
Args:
query: LangGraph다음으로, 우리의 에이전트를 위한 도구 호출을 업데이트해 보자. 여기서 "LangGraph에서 StateGraph를 사용하는 방법"을 검색해보자.
from langchain_core.messages import AIMessage
snapshot = graph.get_state(config)
existing_message = snapshot.values["messages"][-1]
print("Original")
print("Message ID", existing_message.id)
print(existing_message.tool_calls[0])
new_tool_call = existing_message.tool_calls[0].copy()
new_tool_call["args"]["query"] = "LangGraph에서 StateGraph를 사용하는 방법"
new_message = AIMessage(
content=existing_message.content,
tool_calls=[new_tool_call],
# 중요! ID는 LangGraph가 이 메시지를 상태에 추가하는 것이 아니라 교체하는 방법으로 사용된다.
id=existing_message.id,
)
print("Updated")
print(new_message.tool_calls[0])
print("Message ID", new_message.id)
graph.update_state(config, {"messages": [new_message]})
print("\n\nTool calls")
graph.get_state(config).values["messages"][-1].tool_callsOriginal
Message ID run-342f3f54-356b-4cc1-b747-573f6aa31054-0
{'name': 'tavily_search_results_json', 'args': {'query': 'LangGraph'}, 'id': 'toolu_01TfAeisrpx4ddgJpoAxqrVh', 'type': 'tool_call'}
Updated
{'name': 'tavily_search_results_json', 'args': {'query': 'LangGraph에서 StateGraph를 사용하는 방법'}, 'id': 'toolu_01TfAeisrpx4ddgJpoAxqrVh', 'type': 'tool_call'}
Message ID run-342f3f54-356b-4cc1-b747-573f6aa31054-0
Tool calls
[{'name': 'tavily_search_results_json',
'args': {'query': 'LangGraph에서 StateGraph를 사용하는 방법'},
'id': 'toolu_01TfAeisrpx4ddgJpoAxqrVh',
'type': 'tool_call'}]AI의 도구 호출을 간단한 "LangGraph" 대신 "LangGraph에서 StateGraph를 사용하는 방법"를 검색하도록 수정한 것을 확인하자.
LangSmith 트레이스를 확인하여 상태 업데이트 호출을 살펴보자. 새로운 메시지가 이전 AI 메시지를 성공적으로 업데이트한 것을 볼 수 있다.
입력으로 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:
tavily_search_results_json (toolu_01TfAeisrpx4ddgJpoAxqrVh)
Call ID: toolu_01TfAeisrpx4ddgJpoAxqrVh
Args:
query: LangGraph에서 StateGraph를 사용하는 방법
================================= Tool Message =================================
Name: tavily_search_results_json
[{"url": "https://www.studywithgpt.com/ko/tutorial/d5ktrh", "content": "LangGraph에서 스트리밍의 가장 일반적인 용도는 LLM(대형 언어 모델) 토큰을 스트리밍하는 것입니다. ... 그래프의 .stream 또는 .astream 메서드를 사용하는 방법 (여기서 stream_mode=\"custom\" 설정) ... StateGraph를 사용하여 새로운 그래프를 정의하고,"}, {"url": "https://medium.com/@kbdhunga/beginners-guide-to-langgraph-understanding-state-nodes-and-edges-part-1-897e6114fa48", "content": "StateGraph: StateGraph in LangGraph is a class that allows us to create graphs whose nodes communicate by reading and writing to a shared state. The StateGraph class is parameterized by a user"}]
================================== Ai Message ==================================
LangGraph에서 StateGraph를 사용하는 방법에 대한 정보를 찾았어요. StateGraph는 LangGraph에서 노드가 공유 상태를 읽고 쓰도록 허용하는 클래스입니다. StateGraph를 사용하여 새로운 그래프를 정의할 수 있습니다. 더 자세한 정보를 원하시면 아래 링크를 참고하세요:
1. [LangGraph에서 StateGraph 사용 방법](https://www.studywithgpt.com/ko/tutorial/d5ktrh)
2. [LangGraph에서 StateGraph 이해하기](https://medium.com/@kbdhunga/beginners-guide-to-langgraph-understanding-state-nodes-and-edges-part-1-897e6114fa48)트레이스를 확인하여 도구 호출과 이후 LLM 응답을 살펴보자. 이제 그래프가 업데이트된 쿼리 용어를 사용하여 검색 엔진에 쿼리를 전송하는 것을 확인할 수 있다. 여기서 LLM의 검색을 수동으로 재정의할 수 있었다.
이 모든 것은 그래프의 체크포인트 메모리에 반영되어 있으므로, 대화를 계속하면 수정된 모든 상태를 기억할 것이다.
events = graph.stream(
{
"messages": (
"user",
"내가 배운 것을 기억하니?",
)
},
config,
stream_mode="values",
)
for event in events:
if "messages" in event:
event["messages"][-1].pretty_print()================================ Human Message =================================
내가 배운 것을 기억하니?
================================== Ai Message ==================================
당연히요! 당신이 LangGraph에서 StateGraph를 사용하여 새로운 그래프를 정의하는 방법을 배웠습니다. StateGraph는 노드가 공유 상태를 읽고 쓸 수 있도록 하는 LangGraph의 클래스입니다. 이것이 당신이 배운 내용입니다.interrupt_before와 update_state를 사용하여 인간 개입 워크플로의 일환으로 상태를 수동으로 수정했다. 인터럽트와 상태 수정은 에이전트의 동작 방식을 제어할 수 있게 해준다. 지속적인 체크포인팅과 결합하면, 액션을 일시 중지하고 언제든지 재개할 수 있다. 사용자가 그래프가 인터럽트될 때 반드시 자리에 있을 필요는 없다.
이 섹션의 그래프 코드는 이전과 동일하다. 기억해야 할 핵심 스니펫은 특정 노드에 도달할 때마다 그래프를 명시적으로 일시 중지하려면 .compile(..., interrupt_before=[...]) (또는 interrupt_after)를 추가하는 것이다. 그런 다음 update_state를 사용하여 체크포인트를 수정하고 그래프가 어떻게 진행될지를 제어할 수 있다.
