顺序图：第一个可运行的 Workflow 与流式状态观测

本文是「LangGraph 核心与实战」系列专栏的第 3 篇。专栏总览参见：《LangGraph 核心架构全景：用状态图构建可控 Agent 工作流》。

顺序图（Sequential Graph）是 LangGraph 中结构最简单、但工程意义最基础的拓扑结构：节点按照确定的先后次序单向执行，没有分支分流，也没有环形回路。

在学习更复杂的动态路由、自纠循环或多 Agent 并行之前，必须先通过顺序图建立标准的开发习惯：如何设计阶段性数据契约、如何使用 add_edge 装配流水线，以及如何利用 stream 接口实时观测每个工位的状态演进。

业务需求：健康指标综合评估流水线

以一个面向用户的健康体检指标评估系统为例，完整走通顺序图的设计与实现。

业务流水线包含 5 道工序：

  1. 参数校验（validate）：检查用户输入的身高与体重数值是否在生理合理区间内；
  2. 指标计算（calculate）：根据身高体重计算 BMI 指数（ ）；
  3. 分级归类（categorize）：根据世界卫生组织（WHO）与国标标准，将 BMI 映射为健康分级（偏瘦、正常、超重、肥胖）；
  4. 健康建议（advise）：根据健康等级匹配针对性的饮食与运动指导策略；
  5. 格式化报告（format）：汇聚全流程数据，生成可直接交付给前端的结构化报告文本。

流水线拓扑图如下：

状态建模（State Modeling）

状态是整条流水线的数据载体。为了保证类型安全，明确划分“外部必填输入”、“中间计算字段”与“最终产出”：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from typing import TypedDict, Optional, List

class HealthAssessmentState(TypedDict):
    # --- 外部输入 ---
    user_name: str
    height_cm: float
    weight_kg: float
    
    # --- 中间加工字段 ---
    is_valid: bool
    error_message: Optional[str]
    bmi_value: Optional[float]
    health_category: Optional[str]
    advice_points: Optional[List[str]]
    
    # --- 最终交付物 ---
    final_report: Optional[str]

在状态建模中，不要将临时变量都作为顶层必填字段。中间字段设置默认初始为 None，允许流水线在启动时仅接收前 3 个用户输入参数。

工序节点实现（Node Implementation）

每个节点严格遵循单一职责与增量返回原则：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
# 节点 1: 输入校验
def validate_node(state: HealthAssessmentState) -> dict:
    h = state.get("height_cm", 0)
    w = state.get("weight_kg", 0)
    
    if h < 50 or h > 260:
        return {"is_valid": False, "error_message": f"身高数据异常 ({h}cm)，合理范围为 50~260cm"}
    if w < 10 or w > 300:
        return {"is_valid": False, "error_message": f"体重数据异常 ({w}kg)，合理范围为 10~300kg"}
        
    return {"is_valid": True, "error_message": None}

# 节点 2: 指标计算
def calculate_node(state: HealthAssessmentState) -> dict:
    if not state.get("is_valid"):
        return {}  # 前序校验失败，跳过计算
        
    h_m = state["height_cm"] / 100.0
    w_kg = state["weight_kg"]
    bmi = round(w_kg / (h_m * h_m), 1)
    return {"bmi_value": bmi}

# 节点 3: 等级分类
def categorize_node(state: HealthAssessmentState) -> dict:
    if not state.get("is_valid") or state.get("bmi_value") is None:
        return {}
        
    bmi = state["bmi_value"]
    if bmi < 18.5:
        cat = "偏瘦"
    elif 18.5 <= bmi < 24.0:
        cat = "正常"
    elif 24.0 <= bmi < 28.0:
        cat = "超重"
    else:
        cat = "肥胖"
        
    return {"health_category": cat}

# 节点 4: 建议生成
def advise_node(state: HealthAssessmentState) -> dict:
    if not state.get("is_valid"):
        return {}
        
    cat = state.get("health_category")
    advice_map = {
        "偏瘦": ["增加高蛋白与优质碳水摄入", "加入抗阻力力量训练以提升肌肉量"],
        "正常": ["继续保持现有的均衡膳食结构", "每周维持至少 150 分钟中等强度有氧运动"],
        "超重": ["减少精制糖和高脂食品摄入", "增加快走、游泳等低关节冲击的有氧运动"],
        "肥胖": ["建议在临床医生或营养师指导下制定减重方案", "严格控制每日总热量摄入"]
    }
    return {"advice_points": advice_map.get(cat, ["保持健康作息"])}

# 节点 5: 格式化收口
def format_node(state: HealthAssessmentState) -> dict:
    name = state["user_name"]
    
    if not state.get("is_valid"):
        report = f"【体检评估失败】\n用户: {name}\n原因: {state.get('error_message')}"
        return {"final_report": report}
        
    advice_text = "\n".join([f"  - {p}" for p in state["advice_points"]])
    report = (
        f"==============================\n"
        f"   健康指标综合评估报告\n"
        f"==============================\n"
        f"用户姓名: {name}\n"
        f"身体数据: 身高 {state['height_cm']}cm / 体重 {state['weight_kg']}kg\n"
        f"BMI 指数: {state['bmi_value']} (评估等级: {state['health_category']})\n\n"
        f"健康建议:\n{advice_text}\n"
        f"=============================="
    )
    return {"final_report": report}

组装图结构与编译

使用 StateGraph 将 5 个工序用确定性边串接：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
from langgraph.graph import StateGraph, START, END

builder = StateGraph(HealthAssessmentState)

# 1. 注册全部工序节点
builder.add_node("validate", validate_node)
builder.add_node("calculate", calculate_node)
builder.add_node("categorize", categorize_node)
builder.add_node("advise", advise_node)
builder.add_node("format", format_node)

# 2. 串联确定性边 (Sequential Edges)
builder.add_edge(START, "validate")
builder.add_edge("validate", "calculate")
builder.add_edge("calculate", "categorize")
builder.add_edge("categorize", "advise")
builder.add_edge("advise", "format")
builder.add_edge("format", END)

# 3. 编译图
app = builder.compile()

执行机制：批量与流式单步观测

1. 同步完整调用（app.invoke）

1
2
3
4
5
6
7
8
normal_input = {
    "user_name": "张工程师",
    "height_cm": 175.0,
    "weight_kg": 68.0
}

result = app.invoke(normal_input)
print(result["final_report"])

输出：

1
2
3
4
5
6
7
8
9
10
11
==============================
   健康指标综合评估报告
==============================
用户姓名: 张工程师
身体数据: 身高 175.0cm / 体重 68.0kg
BMI 指数: 22.2 (评估等级: 正常)

健康建议:
  - 继续保持现有的均衡膳食结构
  - 每周维持至少 150 分钟中等强度有氧运动
==============================

2. 流式单步追踪（app.stream）

在长程执行或异步网络调用中，通过 stream 能够精确捕捉到每一次状态演进：

1
2
3
4
print("--- 启动流式追踪 ---")
for step_event in app.stream(normal_input):
    for node_name, delta in step_event.items():
        print(f"节点 [{node_name:10}] -> 更新字段: {list(delta.keys())}")

输出：

1
2
3
4
5
6
--- 启动流式追踪 ---
节点 [validate  ] -> 更新字段: ['is_valid', 'error_message']
节点 [calculate ] -> 更新字段: ['bmi_value']
节点 [categorize] -> 更新字段: ['health_category']
节点 [advise   ] -> 更新字段: ['advice_points']
节点 [format   ] -> 更新字段: ['final_report']

每个字典元素即时反映了当前节点所贡献的增量输出。

初始 State 的边界处理原则

在调用 app.invoke() 时，许多开发者会有疑问：“输入参数是否必须包含 HealthAssessmentState 定义的所有键？”

答案是不需要：

  1. 只传必需输入：在调用入口处，仅需传入起点节点依赖的初始键（如 user_name、height_cm、weight_kg）；
  2. 缺省键访问防崩溃：在下游节点读取可能尚未生成的字段时，严禁使用直接下标读取 state["bmi_value"]，应采用防御性读取：state.get("bmi_value")；
  3. 不可预知的异常短路：在顺序图中，即使输入异常（如输入身高 -10cm），由于没有条件分支，图依然会走完所有 5 个节点。本例通过 if not state.get("is_valid"): return {} 实现了业务层面的轻量穿透。在下一篇中，我们将用真正的**条件边（Conditional Edges）**在校验失败时直接跳过中间步骤直达终点。

系列导航与参考