aaihubhub
AI Agent 小白系统课:从大模型到会行动的智能体 · 第 13 课 / 共 20 课 ↗ AI 教程

13|LangGraph 构件:设计可控的状态机

理解 State、Node、Edge 的职责和组合方式。

kimi
LESSON BRIEFING13 / 20

理解 State、Node、Edge 的职责和组合方式。

规划像登山路线:天气稳定可按固定路线,遇到塌方要根据新观察重规划;每个营地都是检查点。

LangGraph 节点、边和状态构件

本课交付物

为竞品研究生成任务图,标注依赖、可并行节点、预算与失败回退。


核心讲解

要使用 LangGraph 构建应用程序,需要理解其核心组件。让我们探索构成 LangGraph 应用程序的基础构建模块。

LangGraph 应用程序从 entrypoint 开始,根据执行情况,流程可能流向不同的函数直到抵达 END。

1. 状态(State)

State 是 LangGraph 中的核心概念,表示流经应用程序的所有信息。

from typing_extensions import TypedDict

class State(TypedDict):
    graph_state: str

状态是 用户自定义 的,因此需要仔细设计字段以包含决策过程所需的所有数据!

💡 提示: 仔细考虑应用程序需要在步骤之间跟踪哪些信息。

2. 节点(Nodes)

Nodes 是 Python 函数。每个节点:

  • 接收状态作为输入
  • 执行某些操作
  • 返回状态更新
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!"}

举例, 节点可以包含:

  • LLM 调用: 生成文本或做出决策
  • 工具调用: 与外部系统交互
  • 条件逻辑: 决定后续步骤
  • 人工干预: 获取用户输入

💡 信息: 像 START 和 END 这样的必要节点已直接包含在 LangGraph 中。

3. 边(Edges)

Edges 连接节点并定义图中的可能路径:


from typing import Literal

def decide_mood(state) -> Literal["node_2", "node_3"]:
    
    # 通常我们会根据状态决定下一个节点
    user_input = state['graph_state'] 
    
    # 这里我们在节点2和节点3之间简单实现 50/50 的概率分配
    if random.random() < 0.5:

        # 50% 时间, 我们返回节点2
        return "node_2"
    
    # 50% 时间, 我们返回节点3
    return "node_3"

边可以是:

  • 直接边: 始终从节点 A 到节点 B
  • 条件边: 根据当前状态选择下一个节点

4. 状态图(StateGraph)

StateGraph 是包含整个 agent 工作流的容器:

from IPython.display import Image, display
from langgraph.graph import StateGraph, START, END

# 构建图表
builder = StateGraph(State)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_node("node_3", node_3)

# 连接逻辑
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)

# 编译
graph = builder.compile()

可以可视化图表:

# 可视化
display(Image(graph.get_graph().draw_mermaid_png()))

最重要的是可以调用:

graph.invoke({"graph_state" : "Hi, this is Lance."})

output :

---Node 3---
{'graph_state': 'Hi, this is Lance. I am sad!'}

下一步?

下一节我们将通过构建第一个图表来实践这些概念。该图表将让 小安 能够处理电子邮件,进行分类,并在邮件真实时起草初步回复。


本站实战工作台

计划不是一次性预言

一次性计划适合稳定环境;动态重规划适合观察会改变路线;任务图适合显式依赖与并行。Agent 计划的基本单元应是可验证下一步,而不是无法检查的“完成研究”。

竞品研究任务图

先澄清产品范围,再并行收集官方定价、功能文档和发布记录;随后去重与时间校验;最后比较并写报告。价格比较依赖前三项,不能提前。

clarify
 ├─ official pricing ─┐
 ├─ feature docs ─────┼→ normalize → cross-check → report
 └─ release notes ────┘

每个节点写输入、工具、预算、成功证据和失败回退。搜索最多三次;官方价格缺失则标未知,不用第三方估算冒充。

何时重规划

只有新观察改变假设、资源或约束时重规划。每轮都重写整份计划会浪费 token 并失去历史。记录 planVersion 与变更理由,已完成节点保持不可变。

并行的代价

并行节点不能共享可变文件或同一浏览器会话。结果通过统一 schema 合并;同一事实冲突时进入 cross-check,而不是后写覆盖先写。

计划与执行分离

计划器只能提出 TaskNode,执行器再次检查 allowedTools、依赖完成状态和预算。不要让一段模型文本同时创造计划并绕过策略执行。每个节点开始和结束都产生事件,失败节点保留现场;人工可以暂停、跳过低价值节点或批准新的预算,但所有变更写入 planVersion。

检查关键路径:若一个慢节点阻塞全部报告,是否能先交付不依赖它的部分?局部结果必须明确标为 partial,不能用完整语气掩盖缺口。

失败注入

让 pricing 节点遭遇登录墙。策略可以换官方销售 PDF、请求用户提供权限或标未知;不能绕过认证。验证下游报告是否保留缺口,而不是因一个节点失败全部崩溃。

通过标准包括任务图无环、依赖正确、预算有限、重规划有原因、失败能局部降级。规划的目的不是显得聪明,而是让下一步和剩余风险都可见。