wehub-resource-sync
a7d6d88f6f
CI / changes (push) Has been cancelled
CI / cd libs/checkpoint (push) Has been cancelled
CI / cd libs/checkpoint-conformance (push) Has been cancelled
CI / cd libs/checkpoint-postgres (push) Has been cancelled
CI / cd libs/checkpoint-sqlite (push) Has been cancelled
CI / cd libs/cli (push) Has been cancelled
CI / cd libs/prebuilt (push) Has been cancelled
CI / cd libs/sdk-py (push) Has been cancelled
CI / cd libs/langgraph (push) Has been cancelled
CI / Check SDK methods matching (push) Has been cancelled
CI / Check CLI schema hasn't changed #3.13 (push) Has been cancelled
CI / CLI integration test (push) Has been cancelled
CI / sdk-py integration test (push) Has been cancelled
CI / CI Success (push) Has been cancelled
baseline / benchmark (push) Has been cancelled
Deploy Redirects to GitHub Pages / deploy (push) Has been cancelled
54 lines
1.7 KiB
Python
54 lines
1.7 KiB
Python
from __future__ import annotations
|
|
|
|
from typing_extensions import TypedDict
|
|
|
|
from langgraph.graph import END, START, StateGraph
|
|
from langgraph.types import Command
|
|
|
|
|
|
def test_parent_command_from_nested_subgraph() -> None:
|
|
class ParentState(TypedDict):
|
|
jump_from_idx: int
|
|
|
|
class ChildState(TypedDict):
|
|
jump: bool
|
|
|
|
child_builder: StateGraph[ChildState] = StateGraph(ChildState)
|
|
|
|
def child_node(state: ChildState) -> Command | ChildState:
|
|
if state["jump"]:
|
|
return Command(graph=Command.PARENT, goto="parent_second")
|
|
return state
|
|
|
|
child_builder.add_node("node", child_node)
|
|
child_builder.add_edge(START, "node")
|
|
|
|
child_0 = child_builder.compile()
|
|
child_1 = child_builder.compile()
|
|
|
|
parent_builder: StateGraph[ParentState] = StateGraph(ParentState)
|
|
|
|
def parent_first(state: ParentState) -> ParentState:
|
|
child_0.invoke({"jump": state["jump_from_idx"] == 1})
|
|
if state["jump_from_idx"] == 1:
|
|
raise AssertionError("Shouldn't be here")
|
|
|
|
child_1.invoke({"jump": state["jump_from_idx"] == 2})
|
|
if state["jump_from_idx"] == 2:
|
|
raise AssertionError("Shouldn't be here")
|
|
|
|
return state
|
|
|
|
def parent_second(state: ParentState) -> ParentState:
|
|
return state
|
|
|
|
parent_builder.add_node("parent_first", parent_first)
|
|
parent_builder.add_node("parent_second", parent_second)
|
|
parent_builder.add_edge(START, "parent_first")
|
|
parent_builder.add_edge("parent_second", END)
|
|
|
|
graph = parent_builder.compile()
|
|
|
|
assert graph.invoke({"jump_from_idx": 1}) == {"jump_from_idx": 1}
|
|
assert graph.invoke({"jump_from_idx": 2}) == {"jump_from_idx": 2}
|