LangGraph 第5章：第一个 LangGraph 程序

本章我们通过一个最简单的示例来体验 LangGraph 的完整开发流程。我们将构建一个双节点线性工作流：接收用户输入，将其转换为大写，然后输出结果。

完整示例

先看完整的代码，然后逐行解析：

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

# 1. 定义状态结构
class MyState(TypedDict):
    input_text: str
    processed_text: str

# 2. 定义节点函数
def upper_case_node(state: MyState):
    """将输入文本转换为大写"""
    original = state["input_text"]
    result = original.upper()
    return {"processed_text": result}

def output_node(state: MyState):
    """输出最终结果"""
    print(f"处理结果: {state['processed_text']}")
    return {}

# 3. 构建图
builder = StateGraph(MyState)

# 添加节点
builder.add_node("to_upper", upper_case_node)
builder.add_node("output", output_node)

# 设置入口点和边
builder.add_edge(START, "to_upper")
builder.add_edge("to_upper", "output")
builder.add_edge("output", END)

# 4. 编译
app = builder.compile()

# 5. 执行
result = app.invoke({
    "input_text": "hello, langgraph!",
    "processed_text": ""
})

print(f"最终状态: {result}")

逐行解析

1. 定义状态

class MyState(TypedDict):
    input_text: str
    processed_text: str

状态定义了工作流中需要共享的数据结构。这里我们定义了两个字段：

input_text：输入文本
processed_text：处理后的文本

2. 定义节点函数

每个节点是一个普通 Python 函数，接收当前状态，返回状态更新：

def upper_case_node(state: MyState):
    original = state["input_text"]  # 读取状态
    result = original.upper()       # 执行操作
    return {"processed_text": result}  # 返回更新

返回值是一个字典，键名对应 State 中的字段名。LangGraph 会自动将返回的字典合并到当前状态中。

3. 构建图

builder = StateGraph(MyState)

创建一个 StateGraph 实例，泛型参数为状态类型。

builder.add_node("to_upper", upper_case_node)
builder.add_node("output", output_node)

调用 add_node 注册节点。第一个参数是节点名称（字符串），第二个参数是节点函数。

builder.add_edge(START, "to_upper")
builder.add_edge("to_upper", "output")
builder.add_edge("output", END)

用 add_edge 连接节点，形成执行路径。START 和 END 是内置特殊节点。

4. 编译

app = builder.compile()

编译将图结构转换为可执行的应用。编译过程中会验证图的正确性（检查所有节点是否可达、是否有死循环等）。

5. 执行

result = app.invoke({
    "input_text": "hello, langgraph!",
    "processed_text": ""
})

调用 invoke 方法，传入初始状态字典。执行流程：

START -> to_upper -> output -> END

最终 result 包含了所有节点执行完毕后最终状态的全部字段。

执行结果

运行上述代码将输出：

处理结果: HELLO, LANGGRAPH!
最终状态: {'input_text': 'hello, langgraph!', 'processed_text': 'HELLO, LANGGRAPH!'}

流式执行

如果需要观察每个节点的执行过程，可以使用 stream 方法：

# 流式输出每个节点的状态
for event in app.stream({
    "input_text": "hello, langgraph!",
    "processed_text": ""
}):
    print(event)

输出：

{'to_upper': {'processed_text': 'HELLO, LANGGRAPH!'}}
{'output': {}}

每个事件包含节点名称和该节点返回的状态更新。

执行流程可视化

# 打印 ASCII 图
print(app.get_graph().draw_ascii())

输出：

       +----------+       +--------+
       | to_upper | ----> | output |
       +----------+       +--------+

练习：扩展第一个程序

尝试以下修改来加深理解：

添加小写转换节点：在 to_upper 之后添加一个 to_lower 节点
增加计数器：在 State 中添加一个 counter 字段，每次节点执行时递增
条件分支：根据 input_text 的长度选择走大写还是小写路径

这个简单的示例展示了 LangGraph 开发的核心流程：定义状态 -> 实现节点 -> 构建图 -> 编译 -> 执行。所有复杂的 LangGraph 应用都遵循这个基本模式。