LangGraph 通过专注于四个主要组件来简化 AI Agent 开发:
- 图形(Graph):这是 LangGraph 的核心架构,如同一张复杂的路线图,涵盖了所有节点以及它们之间的连接关系。它定义了整个系统的结构和信息流动路径。
- 状态(State):状态可以被看作是一个共享的数据中心,它记录了整个流程中的关键信息。每个节点都可以读取和写入这个数据中心,从而实现信息的共享和交互。
- 节点(Node):节点是图形中的基本单元,类似于一个个独立的站点。每个节点都代表一个特定的任务、操作或检查点,是信息处理和任务执行的具体位置。
- 边(Edge):边是连接节点的通道,决定了信息如何从一个节点流向另一个节点。它定义了节点之间的关系和信息传递的方向,是信息流动的路径。
State 状态:消息的共享中心
在 LangGraph 中,状态是应用程序当前快照的共享数据结构,用来存放程序当前数据。它可以是任何类型,通常是 TypedDict 或 Pydantic BaseModel。
主要是给节点之间传递数据的作用。每个节点都可以访问状态和进行读写的操作。当一个节点修改状态时,它会把更新的信息广播到图中的所有其他节点。这种机制使得其他节点对状态的变化而产生不同的输出。
定义一个最简单的状态,当状态中的属性发生变化时,默认会直接覆盖式更新对应的Key。
from typing_extensions import TypedDict
class State(TypedDict):
# 存储需求描述
requirement: str
# 存储测试用例
test_case: str
# 存储测试结果
test_result: str
当状态中的属性发生变化时,还有另外一种模式。在下面这个例子中,我们使用 TypedDict 定义了一个包含 messages 键的字典。 该键预期持有一个字符串。它指定 messages 是一个Annotated字符串列表。可以有方法 operator.add,表示可以通过将新消息与现有数据append的操作来更新列表。
class State(TypedDict):
messages: Annotated[List[str], operator.add]
节点(Node)
节点通常是一个Python函数,第一个位置参数是如上所定义的状态。
所以每个节点都可以使用 state["requirement"] 获取状态中 requirement 属性的值。
在 LangGraph 中,节点就像流程中的节点,每个节点都有特定的任务。它们接收输入,进行处理,然后产出输出结果。这些节点能执行各种任务,从简单的数据转换到复杂的决策。
为了更直观地理解节点的工作机制,让我们通过一个简洁的例子来一探究竟:
假设有一个节点负责接收功能需求,经需求分析产出测试用例、再经过另一个节点,负责执行用例、产出测试报告或打回测试。
def case_design(state):
test_case = "测试用例"
print(state["test_case"], "完成")
return {"test_case": test_case}
def run_case(state):
test_execute = "测试执行"
print(state["test_case"], "完成")
return {"test_execute": test_execute}
def test_back(state):
back = "测试打回"
print(state["test_result"], "完成")
return {"test_result": test_back}
def test_report(state):
report = "测试通过"
print(state["test_result"], "完成")
return {"test_result": report}
现在,我们来看看如何将这些节点添加到我们的 LangGraph「图」中:
graph.add_node("case_design_node", case_design)
通过 add_node 方法,我们将这些功能集成到我们的工作流程图中。 每个节点都有一个唯一的名称(如 run_case_node),这样我们就可以在后续的工作流程中明确地引用它们。
另外,LangGraph 内置了 图的两个特殊节点:开始节点(START)、结束节点(END)。
from langgraph.graph import START, END
边(Edge):连接节点的桥梁
在 LangGraph 中,边就如同工作流程中的传送带或决策枢纽。它们规定了信息在节点之间的传递路径以及传递条件。想象一下测试流程:
在测试流程中,边决定着数据或状态应该被传递到哪一个节点。有时候,这是一条直接的传输通道;而有时候,则需要根据数据或状态的具体属性来做出相应的决策。
普通边(显式控制流)
LangGraph允许您通过普通图边显式定义应用程序的控制流(直接连接两个节点)。
最基本的边,直接连接两个节点。 使用 add_edge 方法来添加:
builder.add_edge("case_design", "run_case")
这行代码告诉系统,'case_design' 完成处理后,应该执行 'run_case',同时相关数据或状态也将直接传递给 'run_case' 。
如果要总是 从 node_1 转到 node_2 ,则使用普通边缘。
条件边(动态控制流)
在LangGraph中,也允许您通过LLM或数据状态决定应用程序控制流的部分。这可以通过使用条件边实现。
条件边允许我们根据特定条件选择下一个执行的节点。 使用 add_conditional_edges 方法来添加。
如果您想在节点之间选择路由,则使用条件边。
条件边是作为函数实现的,它根据某些逻辑返回下一个要访问的节点。
import random
from typing import Literal
def decide_mood(state) -> Literal["test_back", "test_report"]:
# 通常,我们将使用state中某个或某些数据的值来决定下一个要访问的节点
user_input = state['requirement']
# Here, let's just do a 50 / 50 split between nodes test_back, test_report
if random.random() < 0.5:
# 50% of the time, we return Node 2
return "test_back"
# 50% of the time, we return Node 3
return "test_report"
图构建
现在,我们从上面定义的组件,通过StateGraph 构建图形。
首先,我们使用上面定义的 State 类初始化了一个状态图。
然后,我们添加节点和边。
我们使用 START节点,该节点是将用户输入发送到图的特殊节点,以指示在哪里开始我们的图形。
from IPython.display import Image, display
from langgraph.graph import StateGraph, START, END
# Build graph
builder = StateGraph(State)
builder.add_node("case_design", case_design)
builder.add_node("run_case", run_case)
builder.add_node("test_back", test_back)
builder.add_node("test_report", test_report)
# Logic
builder.add_edge(START, "case_design")
builder.add_edge("case_design", "run_case")
builder.add_conditional_edges("run_case", decide_mood)
builder.add_edge("test_back", END)
builder.add_edge("test_report", END)
# Add
graph = builder.compile()
# View
display(Image(graph.get_graph().draw_mermaid_png()))
所构建的图形如下,至此,我们完成了一个简单的Langgraph 图的构建。
下一章节预告:Langgraph中的 Chain、Router。