分支控制

import operator
from typing import Any,Annotated
from typing_extensions import TypedDict
from IPython.display import Image,display
from langgraph.graph import START, END, StateGraph
#Annotated 允许为类型提供额外的元数据，而不影响类型检查器对类型本身的理解
class State(TypedDict):
    aggregate:Annotated[list,operator.add]
   
def a(state:State):
    return {"aggregate": ["A"]}
def b(state:State):
    return {"aggregate": ["B"]}
def c(state:State):
    return {"aggregate": ["C"]}
def d(state:State):
    return {"aggregate": ["D"]}
graph_builder = StateGraph(State)
#添加节点
graph_builder.add_node(a)
graph_builder.add_node(b)
graph_builder.add_node(c)
graph_builder.add_node(d)
# 添加边
graph_builder.add_edge(START, "a")
graph_builder.add_edge("a", "b")
graph_builder.add_edge("a", "c")
graph_builder.add_edge("b", "d")
graph_builder.add_edge("c", "d")
graph = graph_builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))

分支与循环：

from typing import Annotated,Literal
from typing_extensions import TypedDict
from IPython.display import Image,display
from langgraph.graph import START, END, StateGraph
from langgraph.errors import GraphRecursionError

#Annotated 允许为类型提供额外的元数据，而不影响类型检查器对类型本身的理解
class State(TypedDict):
    aggregate:Annotated[list,operator.add]
   
def a(state:State):
    return {"aggregate": ["A"]}

def b(state:State):
    return {"aggregate": ["B"]}

#条件路由函数
def route(state:State) -> Literal["b", END]:
    if len(state["aggregate"]) < 5:
        return "b"
    else:
        return END

graph_builder = StateGraph(State)

#添加节点
graph_builder.add_node(a)
graph_builder.add_node(b)

# 添加边
graph_builder.add_edge(START, "a")
graph_builder.add_conditional_edges("a", route)
graph_builder.add_edge("b", "a")

graph = graph_builder.compile()

display(Image(graph.get_graph().draw_mermaid_png()))

#限制递归调用次数，防止异常情况下的大量无用调用
try:
    graph.invoke({"aggregate": []},{"recursion_limit":3})
except GraphRecursionError:
    print("Recursion error occurred")

条件与循环复杂例子

from typing import Annotated,Literal
from typing_extensions import TypedDict
from IPython.display import Image,display
from langgraph.graph import START, END, StateGraph
from langgraph.errors import GraphRecursionError
#Annotated 允许为类型提供额外的元数据，而不影响类型检查器对类型本身的理解
class State(TypedDict):
    aggregate:Annotated[list,operator.add]
   
def a(state:State):
    return {"aggregate": ["A"]}
def b(state:State):
    return {"aggregate": ["B"]}
def c(state:State):
    return {"aggregate": ["C"]}
def d(state:State):
    return {"aggregate": ["D"]}
#条件路由函数
def route(state:State) -> Literal["b", END]:
    if len(state["aggregate"]) < 5:
        return "b"
    else:
        return END
graph_builder = StateGraph(State)
#添加节点
graph_builder.add_node(a)
graph_builder.add_node(b)
graph_builder.add_node(c)
graph_builder.add_node(d)
# 添加边
graph_builder.add_edge(START, "a")
graph_builder.add_conditional_edges("a", route)
graph_builder.add_edge("b", "c")
graph_builder.add_edge("b", "d")
graph_builder.add_edge(["c","d"], "a")
graph = graph_builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))
#限制递归调用次数，防止异常情况下的大量无用调用
graph.invoke({"aggregate": []})