目标:创建一个图graph,在这个图graph中,你需要传入一个整数列表、一个名称以及一个操作符。如果操作符是 “+”,就对列表中的元素进行求和;如果操作符是 “*”,就对列表中的元素进行求积 —— 所有这些操作都要在同一个节点内完成。

# 导入类型提示工具,用于定义状态结构和函数参数类型
from typing import TypedDict, List
# 导入LangGraph的核心组件,用于构建状态图
from langgraph.graph import StateGraph, START, END

# 定义状态结构(图中流转的数据载体)
# 使用TypedDict声明一个结构化字典,明确状态中包含的字段及类型
class State(TypedDict):
    values: List[int]  # 存储待计算的整数列表(如[1,2,3,4])
    result: str        # 存储计算结果的字符串(如"HI, Jim, 计算结果=24")
    name: str          # 存储用户名(用于在结果中显示)
    op: str            # 存储操作符("+"表示加法,"*"表示乘法)

# 定义乘法函数:计算整数列表中所有元素的乘积
def multiply(numbers: List[int]) -> int:
    """计算列表中所有整数的乘积"""
    if not numbers:  # 处理空列表的边缘情况(避免乘法运算出错)
        return 0
    product = 1  # 初始化乘积为1(乘法的单位元)
    for num in numbers:
        product *= num  # 依次乘以列表中的每个元素
    return product

# 定义核心处理节点:根据操作符对整数列表进行计算
def process_values(x: State) -> State:
    """处理多个不同的输入信息:根据操作符计算结果并更新状态"""
    print(x)  # 打印当前输入的状态(调试用,查看处理前的数据)
    
    # 根据操作符(op)执行不同的计算
    if x["op"] == "+":
        # 若操作符为"+",计算列表元素的总和,并将结果存入状态的result字段
        x["result"] = f'HI, {x["name"]}, 计算结果={sum(x["values"])}'
    elif x["op"] == '*':
        # 若操作符为"*",调用multiply函数计算乘积,并将结果存入result字段
        x["result"] = f'HI, {x["name"]}, 计算结果={multiply(x["values"])}'
    else:
        # 处理不支持的操作符,返回错误提示
        x["result"] = f'HI, 不支持的操作符: {x["op"]}'
    
    print(x)  # 打印处理后的状态(调试用,查看更新后的结果)
    return x  # 返回更新后的状态,供后续流程使用

# 3. 构建图(初始化状态图构建器,关联我们定义的State结构)
graph_builder = StateGraph(State)

# 4. 添加节点和边(定义图的结构和流程)
# 向图中添加一个名为"a"的节点,关联处理函数process_values
graph_builder.add_node("a", process_values)
# 定义流程起点:从START(系统内置的起始点)流向节点"a"
graph_builder.add_edge(START, "a")
# 定义流程终点:从节点"a"流向END(系统内置的结束点)
graph_builder.add_edge("a", END)

# 5. 编译图(将定义的结构转换为可执行的图对象)
graph = graph_builder.compile()

# 调用图进行计算:传入初始状态(整数列表、用户名、操作符)
# 这里的输入是:计算[1,2,3,4]的乘积,用户名为Jim
res = graph.invoke({"values": [1,2,3,4], "name": "Jim", "op": "*"})
# 打印最终的计算结果(从状态的result字段中提取)
print(res["result"])  # 输出:HI, Jim, 计算结果=24

# 6. 显示图结构(可视化流程)
# 导入IPython的图片显示工具(需在Jupyter环境中运行)
from IPython.display import Image, display
# 生成图的Mermaid格式流程图(PNG图片)并显示
# 该图会展示:START → "a"节点 → END 的简单流程
display(Image(graph.get_graph().draw_mermaid_png()))

代码核心逻辑总结:

  1. 状态定义:通过State(TypedDict)明确图中流转的数据结构(包含待计算的列表、用户名、操作符和结果)。
  2. 工具函数multiply函数实现列表元素的乘法(加法直接使用 Python 内置的sum函数)。
  3. 节点逻辑process_values节点根据操作符(op)判断是执行加法还是乘法,计算后更新状态中的result字段。
  4. 图结构:构建了一个简单的线性流程(START → "a"节点 → END),所有计算逻辑都在 "a" 节点中完成。
  5. 执行与可视化:调用graph.invoke()传入初始数据触发计算,并通过draw_mermaid_png()可视化图的结构。

整个代码展示了 LangGraph 的核心用法:用状态承载数据,用节点处理逻辑,用图定义流程,最终实现一个简单的数值计算功能。

Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐