In [ ]:
Copied!
%%capture --no-stderr
%pip install --quiet -U langgraph
%%capture --no-stderr
%pip install --quiet -U langgraph
In [2]:
Copied!
from typing_extensions import TypedDict
class State(TypedDict):
graph_state: str
from typing_extensions import TypedDict
class State(TypedDict):
graph_state: str
In [4]:
Copied!
def node_1(state):
print("---Node 1---")
return {"graph_state": state['graph_state'] +" I am"}
def node_2(state):
print("---Node 2---")
return {"graph_state": state['graph_state'] +" happy!"}
def node_3(state):
print("---Node 3---")
return {"graph_state": state['graph_state'] +" sad!"}
def node_1(state):
print("---Node 1---")
return {"graph_state": state['graph_state'] +" I am"}
def node_2(state):
print("---Node 2---")
return {"graph_state": state['graph_state'] +" happy!"}
def node_3(state):
print("---Node 3---")
return {"graph_state": state['graph_state'] +" sad!"}
In [5]:
Copied!
import random
from typing import Literal
def decide_mood(state) -> Literal["node_2", "node_3"]:
# Often, we will use state to decide on the next node to visit
user_input = state['graph_state']
# Here, let's just do a 50 / 50 split between nodes 2, 3
if random.random() < 0.5:
# 50% of the time, we return Node 2
return "node_2"
# 50% of the time, we return Node 3
return "node_3"
import random
from typing import Literal
def decide_mood(state) -> Literal["node_2", "node_3"]:
# Often, we will use state to decide on the next node to visit
user_input = state['graph_state']
# Here, let's just do a 50 / 50 split between nodes 2, 3
if random.random() < 0.5:
# 50% of the time, we return Node 2
return "node_2"
# 50% of the time, we return Node 3
return "node_3"
图结构构建¶
现在，我们将基于上述定义的组件构建图结构。
StateGraph类是我们将使用的基础图类。
首先，使用之前定义的State类初始化一个StateGraph。
接着，添加节点和边。
我们使用START节点（特殊节点）来标识图的起点，该节点负责将用户输入传递至图中。
END节点是表示终止节点的特殊节点。
最后，我们编译图结构以执行基本的图结构检查。
该图结构可被可视化为Mermaid流程图。
In [6]:
Copied!
from IPython.display import Image, display
from langgraph.graph import StateGraph, START, END
# Build graph
builder = StateGraph(State)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_node("node_3", node_3)
# Logic
builder.add_edge(START, "node_1")
builder.add_conditional_edges("node_1", decide_mood)
builder.add_edge("node_2", END)
builder.add_edge("node_3", END)
# Add
graph = builder.compile()
# View
display(Image(graph.get_graph().draw_mermaid_png()))
from IPython.display import Image, display
from langgraph.graph import StateGraph, START, END
# Build graph
builder = StateGraph(State)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_node("node_3", node_3)
# Logic
builder.add_edge(START, "node_1")
builder.add_conditional_edges("node_1", decide_mood)
builder.add_edge("node_2", END)
builder.add_edge("node_3", END)
# Add
graph = builder.compile()
# View
display(Image(graph.get_graph().draw_mermaid_png()))
In [7]:
Copied!
graph.invoke({"graph_state" : "Hi, this is Lance."})
graph.invoke({"graph_state" : "Hi, this is Lance."})
---Node 1--- ---Node 3---
Out[7]:
{'graph_state': 'Hi, this is Lance. I am sad!'}
invoke 会同步运行整个图结构。
它会等待每一步骤完成后再继续执行下一步。
在所有节点执行完毕后，该方法会返回图的最终状态。
在本例中，它返回的是 node_3 完成后的状态：
{'graph_state': 'Hi, this is Lance. I am sad!'}
In [ ]:
Copied!