LangGraph 공식문서를 번역한 내용입니다. 필요한 경우 부연 설명을 추가하였고 이해하기 쉽게 예제를 일부 변경하였습니다. 문제가 되면 삭제하겠습니다.
https://langchain-ai.github.io/langgraph/how-tos/subgraph-persistence/
이 가이드는 서브그래프를 사용하는 그래프에 스레드 수준 영속성을 추가하는 방법을 보여준다.
준비
필요한 패키지를 설치한다.
pip install --quiet -U langgraph영속성이 있는 그래프 정의
서브그래프가 있는 그래프에 영속성을 추가하려면, 부모 그래프를 컴파일할 때 체크포인터(checkpointer)를 전달하기만 하면 된다. LangGraph는 체크포인터를 자동으로 자식 서브그래프에 전달한다.
Note: 서브그래프를 컴파일할 때 체크포인터를 제공하면 안 된다. 대신, 부모 그래프의
compile()에 전달할 단일 체크포인터를 정의해야 하며, LangGraph는 체크포인터를 자동으로 자식 서브그래프에 전달한다. 서브그래프의compile()에 체크포인터를 전달하면 단순히 무시된다. 이 원칙은 서브그래프를 호출하는 노드 함수가 추가될 때도 적용된다.
동작을 확인하기 위해 단일 서브그래프 노드가 있는 간단한 그래프를 정의해 보자.
from langgraph.graph import START, StateGraph
from langgraph.checkpoint.memory import MemorySaver
from typing import TypedDict
# 서브 그래프
class SubgraphState(TypedDict):
foo: str # 이 키는 부모 그래프 상태와 공유되어 있다.
bar: str
def subgraph_node_1(state: SubgraphState):
return {"bar": "bar"}
def subgraph_node_2(state: SubgraphState):
# 이 노드는 서브그래프에서만 사용 가능한 상태 키 ('bar')를 사용하고 있으며, 공유 상태 키 ('foo')에 대한 업데이트를 보내고 있다.
return {"foo": state["foo"] + state["bar"]}
subgraph_builder = StateGraph(SubgraphState)
subgraph_builder.add_node(subgraph_node_1)
subgraph_builder.add_node(subgraph_node_2)
subgraph_builder.add_edge(START, "subgraph_node_1")
subgraph_builder.add_edge("subgraph_node_1", "subgraph_node_2")
subgraph = subgraph_builder.compile()
# 부모 그래프
class State(TypedDict):
foo: str
def node_1(state: State):
return {"foo": "hi! " + state["foo"]}
builder = StateGraph(State)
builder.add_node("node_1", node_1)
# 서브그래프를 부모 그래프에 노드로 추가하는 것을 볼 수 있다.
builder.add_node("node_2", subgraph)
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")이제 인메모리 체크포인터(MemorySaver)를 사용하여 그래프를 컴파일할 수 있다.
checkpointer = MemorySaver()
# 부모 그래프를 컴파일할 때만 checkpointer를 전달해야 한다.
# LangGraph는 자동으로 checkpointer를 하위 서브그래프로 전파한다.
graph = builder.compile(checkpointer=checkpointer)영속성 동작 확인
이제 그래프를 실행하고 부모 그래프와 서브그래프의 영속된 상태를 확인하여 영속성이 제대로 작동하는지 검증해 보자. state.values에서 부모 그래프와 서브그래프 모두에 대한 최종 실행 결과를 볼 수 있어야 한다.
config = {"configurable": {"thread_id": "1"}}for _, chunk in graph.stream({"foo": "foo"}, config, subgraphs=True):
print(chunk){'node_1': {'foo': 'hi! foo'}}
{'subgraph_node_1': {'bar': 'bar'}}
{'subgraph_node_2': {'foo': 'hi! foobar'}}
{'node_2': {'foo': 'hi! foobar'}}이제 그래프를 호출할 때 사용한 것과 동일한 설정으로 graph.get_state()를 호출하여 부모 그래프의 상태를 볼 수 있다.
print(graph.get_state(config).values){'foo': 'hi! foobar'}서브그래프 상태를 보려면 두 가지 작업을 해야 한다.
- 서브그래프에 대한 가장 최근의 설정 값을 찾기
graph.get_state()를 사용하여 가장 최근 서브그래프 설정에 대한 값을 가져오기
올바른 설정을 찾기 위해, 부모 그래프의 상태 기록을 살펴보고 node_2(서브그래프가 있는 노드)에서 결과를 반환하기 전의 상태 스냅샷을 찾을 수 있다.
state_with_subgraph = [
s for s in graph.get_state_history(config) if s.next == ("node_2",)
][0]상태 스냅샷에는 다음에 실행될 작업 목록이 포함된다. 서브그래프를 사용할 때, 작업에는 서브그래프 상태를 가져오는 데 사용할 수 있는 설정이 포함된다.
subgraph_config = state_with_subgraph.tasks[0].state
print(subgraph_config){'configurable': {'thread_id': '1',
'checkpoint_ns': 'node_2:6ef111a6-f290-7376-0dfc-a4152307bc5b'}}print(graph.get_state(subgraph_config).values){'foo': 'hi! foobar', 'bar': 'bar'}사람이 개입하는 워크플로우에서 서브그래프 상태를 수정하는 방법에 대해 더 알고 싶다면, 이 가이드에서 확인해 보자.
