mirror of
https://github.com/langchain-ai/langgraph.git
synced 2026-09-10 11:47:51 +02:00
 --------- Co-authored-by: Vadym Barda <vadym@langchain.dev>
44 KiB
44 KiB
In [1]:
%%capture --no-stderr
%pip install -U langgraphIn [1]:
import operator
from typing import Annotated, Literal
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
class State(TypedDict):
# The operator.add reducer fn makes this append-only
aggregate: Annotated[list, operator.add]
def a(state: State):
print(f'Node A sees {state["aggregate"]}')
return {"aggregate": ["A"]}
def b(state: State):
print(f'Node B sees {state["aggregate"]}')
return {"aggregate": ["B"]}
# Define nodes
builder = StateGraph(State)
builder.add_node(a)
builder.add_node(b)
# Define edges
def route(state: State) -> Literal["b", END]:
if len(state["aggregate"]) < 7:
return "b"
else:
return END
builder.add_edge(START, "a")
builder.add_conditional_edges("a", route)
builder.add_edge("b", "a")
graph = builder.compile()In [2]:
from IPython.display import Image, display
display(Image(graph.get_graph().draw_mermaid_png()))In [3]:
graph.invoke({"aggregate": []})Out [3]:
Node A sees [] Node B sees ['A'] Node A sees ['A', 'B'] Node B sees ['A', 'B', 'A'] Node A sees ['A', 'B', 'A', 'B'] Node B sees ['A', 'B', 'A', 'B', 'A'] Node A sees ['A', 'B', 'A', 'B', 'A', 'B']
{'aggregate': ['A', 'B', 'A', 'B', 'A', 'B', 'A']}In [4]:
from langgraph.errors import GraphRecursionError
try:
graph.invoke({"aggregate": []}, {"recursion_limit": 4})
except GraphRecursionError:
print("Recursion Error")Node A sees [] Node B sees ['A'] Node A sees ['A', 'B'] Node B sees ['A', 'B', 'A'] Recursion Error
In [5]:
import operator
from typing import Annotated, Literal
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
class State(TypedDict):
aggregate: Annotated[list, operator.add]
def a(state: State):
print(f'Node A sees {state["aggregate"]}')
return {"aggregate": ["A"]}
def b(state: State):
print(f'Node B sees {state["aggregate"]}')
return {"aggregate": ["B"]}
def c(state: State):
print(f'Node C sees {state["aggregate"]}')
return {"aggregate": ["C"]}
def d(state: State):
print(f'Node D sees {state["aggregate"]}')
return {"aggregate": ["D"]}
# Define nodes
builder = StateGraph(State)
builder.add_node(a)
builder.add_node(b)
builder.add_node(c)
builder.add_node(d)
# Define edges
def route(state: State) -> Literal["b", END]:
if len(state["aggregate"]) < 7:
return "b"
else:
return END
builder.add_edge(START, "a")
builder.add_conditional_edges("a", route)
builder.add_edge("b", "c")
builder.add_edge("b", "d")
builder.add_edge(["c", "d"], "a")
graph = builder.compile()In [6]:
from IPython.display import Image, display
display(Image(graph.get_graph().draw_mermaid_png()))In [7]:
result = graph.invoke({"aggregate": []})Node A sees [] Node B sees ['A'] Node D sees ['A', 'B'] Node C sees ['A', 'B'] Node A sees ['A', 'B', 'C', 'D'] Node B sees ['A', 'B', 'C', 'D', 'A'] Node D sees ['A', 'B', 'C', 'D', 'A', 'B'] Node C sees ['A', 'B', 'C', 'D', 'A', 'B'] Node A sees ['A', 'B', 'C', 'D', 'A', 'B', 'C', 'D']
In [8]:
from langgraph.errors import GraphRecursionError
try:
result = graph.invoke({"aggregate": []}, {"recursion_limit": 4})
except GraphRecursionError:
print("Recursion Error")Node A sees [] Node B sees ['A'] Node C sees ['A', 'B'] Node D sees ['A', 'B'] Node A sees ['A', 'B', 'C', 'D'] Recursion Error