13|LangGraph 构件:设计可控的状态机
理解 State、Node、Edge 的职责和组合方式。
理解 State、Node、Edge 的职责和组合方式。
规划像登山路线:天气稳定可按固定路线,遇到塌方要根据新观察重规划;每个营地都是检查点。

本课交付物
为竞品研究生成任务图,标注依赖、可并行节点、预算与失败回退。
核心讲解
要使用 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、请求用户提供权限或标未知;不能绕过认证。验证下游报告是否保留缺口,而不是因一个节点失败全部崩溃。
通过标准包括任务图无环、依赖正确、预算有限、重规划有原因、失败能局部降级。规划的目的不是显得聪明,而是让下一步和剩余风险都可见。