Files
ai-agent-dev/15_图结构.ipynb
2026-07-08 10:09:42 +08:00

657 lines
21 KiB
Plaintext
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

{
"cells": [
{
"cell_type": "markdown",
"id": "1801d2fc",
"metadata": {},
"source": [
"# 15 图结构\n",
"\n",
"## 学习目标\n",
"1. 理解为什么要用图结构来组织复杂流程\n",
"2. 掌握 LangGraph 中节点、边、状态的配合方式\n",
"3. 学会构建顺序执行、条件分支和循环三种常见图结构\n",
"4. 能够看懂图结构的执行路径,并根据需求修改流程\n",
"5. 为后续多智能体、工作流和复杂 Agent 打下基础"
]
},
{
"cell_type": "markdown",
"id": "1f4e8fef",
"metadata": {},
"source": [
"## 1. 为什么需要图结构\n",
"\n",
"在前面的课程中我们已经学过链Chain\n",
"\n",
"```\n",
"输入 -> Prompt -> 模型 -> 输出\n",
"```\n",
"\n",
"这种结构很适合**线性流程**,也就是每一步都按固定顺序往下走。\n",
"\n",
"但是在真实应用中,流程往往没有这么简单。例如:\n",
"\n",
"- 如果用户的问题不完整,要先追问再继续\n",
"- 如果模型判断需要调用工具,要进入工具节点\n",
"- 如果工具返回的信息还不够,要再循环一次\n",
"- 如果满足某个条件,要走 A 路线,否则走 B 路线\n",
"\n",
"这时,单纯的链就不够用了。我们需要一种更灵活的结构,让流程能够:\n",
"\n",
"- **分支**:根据条件走不同路线\n",
"- **回路**:重复执行直到满足条件\n",
"- **汇合**:多条路径最后回到同一个结果\n",
"\n",
"这就是**图结构Graph**的价值。"
]
},
{
"cell_type": "markdown",
"id": "761e8840",
"metadata": {},
"source": [
"## 2. 图结构的核心组成\n",
"\n",
"在 LangGraph 中,一个图通常由三部分组成:\n",
"\n",
"### 2.1 State状态\n",
"状态可以理解为‘流程执行时共享的一份数据’。\n",
"\n",
"例如一个状态可能长这样:\n",
"\n",
"```python\n",
"{'value': 3, 'result': '通过', 'step': 2}\n",
"```\n",
"\n",
"图中的每个节点都可以读取这份状态,也可以返回对状态的更新。\n",
"\n",
"### 2.2 Node节点\n",
"节点就是图中的一个处理步骤。\n",
"\n",
"你可以把节点理解成‘流水线上的一个工位’。每个工位做一件事,例如:\n",
"\n",
"- 给数字加 1\n",
"- 判断成绩是否及格\n",
"- 调用大模型生成回复\n",
"- 调用搜索工具获取资料\n",
"\n",
"### 2.3 Edge\n",
"边表示‘从哪个节点走到哪个节点’。\n",
"\n",
"边决定了执行顺序。常见有两种:\n",
"\n",
"- **普通边**:固定走向下一个节点\n",
"- **条件边**:根据状态判断下一步走哪条路\n",
"\n",
"可以把图结构想象成一张地铁路线图:\n",
"\n",
"- 状态 = 你当前身上的信息\n",
"- 节点 = 你经过的站点\n",
"- 边 = 站点之间的路线"
]
},
{
"cell_type": "markdown",
"id": "bf0b4be0",
"metadata": {},
"source": [
"## 3. 第一个图:顺序执行\n",
"\n",
"先从最简单的情况开始:\n",
"\n",
"```\n",
"START -> 加一 -> 乘二 -> END\n",
"```\n",
"\n",
"也就是说:\n",
"\n",
"1. 初始值是 `value=3`\n",
"2. 第一个节点先把它加 1变成 4\n",
"3. 第二个节点再把它乘 2变成 8\n",
"4. 最后输出结果"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "527c7a13",
"metadata": {},
"outputs": [],
"source": [
"from typing_extensions import TypedDict\n",
"from langgraph.graph import StateGraph, START, END\n",
"\n",
"# 1. 定义状态结构\n",
"class NumberState(TypedDict):\n",
" value: int\n",
"\n",
"# 2. 定义节点函数\n",
"def add_one(state: NumberState):\n",
" print(f'进入 add_one 前value = {state[\"value\"]}')\n",
" return {'value': state['value'] + 1}\n",
"\n",
"def multiply_two(state: NumberState):\n",
" print(f'进入 multiply_two 前value = {state[\"value\"]}')\n",
" return {'value': state['value'] * 2}\n",
"\n",
"# 3. 创建图\n",
"builder = StateGraph(NumberState)\n",
"\n",
"# 4. 添加节点\n",
"builder.add_node('add_one', add_one)\n",
"builder.add_node('multiply_two', multiply_two)\n",
"\n",
"# 5. 添加边\n",
"builder.add_edge(START, 'add_one')\n",
"builder.add_edge('add_one', 'multiply_two')\n",
"builder.add_edge('multiply_two', END)\n",
"\n",
"# 6. 编译图\n",
"graph = builder.compile()\n",
"\n",
"# 7. 执行图\n",
"result = graph.invoke({'value': 3})\n",
"print('最终结果:', result)"
]
},
{
"cell_type": "markdown",
"id": "62406ccb",
"metadata": {},
"source": [
"### 代码解释\n",
"\n",
"这段代码虽然不长,但非常关键,值得逐行理解。\n",
"\n",
"#### 第 1 步:定义状态\n",
"`NumberState` 规定了图里共享的数据结构,这里只有一个字段:`value`。\n",
"\n",
"也就是说,整张图传来传去的核心数据就是一个整数。\n",
"\n",
"#### 第 2 步:定义节点函数\n",
"`add_one(state)` 和 `multiply_two(state)` 都接收当前状态。\n",
"\n",
"例如当状态是:\n",
"\n",
"```python\n",
"{'value': 3}\n",
"```\n",
"\n",
"那么 `add_one` 返回:\n",
"\n",
"```python\n",
"{'value': 4}\n",
"```\n",
"\n",
"节点函数**不需要返回完整状态**只要返回要更新的字段即可。LangGraph 会把更新结果合并回状态中。\n",
"\n",
"#### 第 3 步:创建图\n",
"`StateGraph(NumberState)` 表示:我们要创建一张图,并且这张图中的状态结构由 `NumberState` 定义。\n",
"\n",
"#### 第 4 步:添加节点\n",
"`builder.add_node('add_one', add_one)` 的意思是:\n",
"\n",
"- 节点名字叫 `add_one`\n",
"- 这个节点真正执行的函数是 Python 函数 `add_one`\n",
"\n",
"节点名字更像是图里的‘标签’,函数才是‘真正干活的人’。\n",
"\n",
"#### 第 5 步:添加边\n",
"这三行定义了执行路径:\n",
"\n",
"- 从 `START` 进入 `add_one`\n",
"- 从 `add_one` 进入 `multiply_two`\n",
"- 从 `multiply_two` 进入 `END`\n",
"\n",
"因此执行顺序是固定的,没有分支,也没有循环。\n",
"\n",
"#### 第 6 步:编译图\n",
"`compile()` 会把‘图的定义’变成一个真正可运行的对象。\n",
"\n",
"你可以把它理解为:前面是在画流程图,这一步是把流程图变成程序。\n",
"\n",
"#### 第 7 步:运行图\n",
"`graph.invoke({'value': 3})` 表示用初始状态 `{'value': 3}` 启动整张图。\n",
"\n",
"执行过程是:\n",
"\n",
"1. `value=3` 进入 `add_one`\n",
"2. 更新为 `value=4`\n",
"3. `value=4` 进入 `multiply_two`\n",
"4. 更新为 `value=8`\n",
"5. 图结束,输出 `{'value': 8}`"
]
},
{
"cell_type": "markdown",
"id": "e86fbc2b",
"metadata": {},
"source": [
"## 4. 查看图结构\n",
"\n",
"除了运行图,我们还可以把图的结构打印出来,帮助自己检查流程是否连对了。"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "029c26ae",
"metadata": {},
"outputs": [],
"source": [
"print('节点:')\n",
"print(graph.get_graph().nodes)\n",
"\n",
"print('\\n边')\n",
"print(graph.get_graph().edges)\n",
"\n",
"print('\\nMermaid 图:')\n",
"print(graph.get_graph().draw_mermaid())"
]
},
{
"cell_type": "markdown",
"id": "996a24d3",
"metadata": {},
"source": [
"### 代码解释\n",
"\n",
"这一段代码的重点不在‘计算’,而在‘观察图’。\n",
"\n",
"#### `graph.get_graph().nodes`\n",
"它会返回当前图中有哪些节点。你可以把它理解为‘流程图里有哪些方框’。\n",
"\n",
"#### `graph.get_graph().edges`\n",
"它会返回节点之间的连接关系,也就是‘箭头从哪里指向哪里’。\n",
"\n",
"#### `draw_mermaid()`\n",
"它会生成 Mermaid 格式的流程图文本。\n",
"\n",
"Mermaid 是一种用文本描述流程图的语法,例如:\n",
"\n",
"```\n",
"START --> add_one --> multiply_two --> END\n",
"```\n",
"\n",
"你可以把输出复制到支持 Mermaid 的工具中查看图形化效果。\n",
"\n",
"这在图结构变复杂之后尤其有用,因为:\n",
"\n",
"- 可以快速确认自己有没有连错边\n",
"- 可以帮助团队成员理解流程\n",
"- 可以作为调试流程的一部分"
]
},
{
"cell_type": "markdown",
"id": "9117ea6b",
"metadata": {},
"source": [
"## 5. 条件分支图\n",
"\n",
"前面的图是固定顺序执行。现在我们来做一个更像真实业务的例子:**根据分数决定是否通过**。\n",
"\n",
"流程如下:\n",
"\n",
"```\n",
"START -> 判断分数 -> 合格 / 不合格 -> END\n",
"```\n",
"\n",
"如果分数大于等于 60就走通过节点否则走补考节点。\n",
"\n",
"这就是典型的**条件分支**。"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "ed38a5d4",
"metadata": {},
"outputs": [],
"source": [
"from typing_extensions import TypedDict\n",
"from langgraph.graph import StateGraph, START, END\n",
"\n",
"class ScoreState(TypedDict):\n",
" score: int\n",
" result: str\n",
"\n",
"def check_score(state: ScoreState):\n",
" print(f'当前分数:{state[\"score\"]}')\n",
" return {}\n",
"\n",
"def pass_node(state: ScoreState):\n",
" return {'result': '成绩合格,顺利通过'}\n",
"\n",
"def fail_node(state: ScoreState):\n",
" return {'result': '成绩不合格,需要补考'}\n",
"\n",
"def route_by_score(state: ScoreState):\n",
" if state['score'] >= 60:\n",
" return 'pass_node'\n",
" return 'fail_node'\n",
"\n",
"builder = StateGraph(ScoreState)\n",
"builder.add_node('check_score', check_score)\n",
"builder.add_node('pass_node', pass_node)\n",
"builder.add_node('fail_node', fail_node)\n",
"\n",
"builder.add_edge(START, 'check_score')\n",
"builder.add_conditional_edges(\n",
" 'check_score',\n",
" route_by_score,\n",
" {\n",
" 'pass_node': 'pass_node',\n",
" 'fail_node': 'fail_node'\n",
" }\n",
")\n",
"builder.add_edge('pass_node', END)\n",
"builder.add_edge('fail_node', END)\n",
"\n",
"score_graph = builder.compile()\n",
"\n",
"print('分数 85 的结果:')\n",
"print(score_graph.invoke({'score': 85, 'result': ''}))\n",
"\n",
"print('\\n分数 45 的结果:')\n",
"print(score_graph.invoke({'score': 45, 'result': ''}))"
]
},
{
"cell_type": "markdown",
"id": "a8a9080b",
"metadata": {},
"source": [
"### 代码解释\n",
"\n",
"这个例子第一次出现了**条件边**,所以需要重点掌握。\n",
"\n",
"#### 状态结构\n",
"这里的状态包含两个字段:\n",
"\n",
"- `score`:输入分数\n",
"- `result`:最终判断结果\n",
"\n",
"也就是说,这张图的目标是:根据 `score` 计算出 `result`。\n",
"\n",
"#### `check_score` 节点\n",
"这个节点本身没有修改状态,只是把当前分数打印出来。\n",
"\n",
"它返回 `{}`,表示‘这个节点不更新任何字段’。\n",
"\n",
"这说明一个节点不一定非要做数据修改,它也可以只承担‘中转’或‘判断前准备’的作用。\n",
"\n",
"#### `pass_node` 和 `fail_node`\n",
"这两个节点分别代表两条不同路径的处理结果:\n",
"\n",
"- 如果通过,更新 `result='成绩合格,顺利通过'`\n",
"- 如果不通过,更新 `result='成绩不合格,需要补考'`\n",
"\n",
"这两个节点就像流程图里分叉后的两条支路。\n",
"\n",
"#### `route_by_score`:路由函数\n",
"这是本例最核心的部分。\n",
"\n",
"它不是真正的业务节点,而是一个‘路线选择器’。它根据当前状态判断:\n",
"\n",
"- `score >= 60` 返回 `'pass_node'`\n",
"- 否则返回 `'fail_node'`\n",
"\n",
"返回值必须和后面条件边映射表中的 key 对应上。\n",
"\n",
"#### `add_conditional_edges(...)`\n",
"这段代码的意思是:\n",
"\n",
"从 `check_score` 节点出来后,不是固定走下一步,而是调用 `route_by_score(state)`。\n",
"\n",
"如果返回 `'pass_node'`,就走到 `pass_node`;如果返回 `'fail_node'`,就走到 `fail_node`。\n",
"\n",
"你可以把它理解成:\n",
"\n",
"- 普通边 = 固定导航\n",
"- 条件边 = 智能导航\n",
"\n",
"#### 最终执行效果\n",
"当输入 `score=85` 时,流程是:\n",
"\n",
"```\n",
"START -> check_score -> pass_node -> END\n",
"```\n",
"\n",
"当输入 `score=45` 时,流程是:\n",
"\n",
"```\n",
"START -> check_score -> fail_node -> END\n",
"```\n",
"\n",
"这就是图结构中‘根据条件走不同路线’的最基本模式。"
]
},
{
"cell_type": "markdown",
"id": "3cf69b96",
"metadata": {},
"source": [
"## 6. 循环图\n",
"\n",
"除了分支,图结构还有一个非常重要的能力:**循环**。\n",
"\n",
"在很多智能体任务中,我们都需要重复执行某些步骤,比如:\n",
"\n",
"- 检索不到足够信息就继续搜索\n",
"- 模型输出格式不对就重新生成\n",
"- 工具调用结果不完整就继续补充\n",
"\n",
"下面我们用一个简单例子说明循环:\n",
"\n",
"让 `count` 从 0 开始,每次加 1直到加到 3 为止。\n",
"\n",
"流程如下:\n",
"\n",
"```\n",
"START -> increment -> 判断是否结束\n",
" ↘ 未结束,回到 increment\n",
" ↘ 已结束,进入 END\n",
"```"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "d33d64ec",
"metadata": {},
"outputs": [],
"source": [
"from typing_extensions import TypedDict\n",
"from langgraph.graph import StateGraph, START, END\n",
"\n",
"class LoopState(TypedDict):\n",
" count: int\n",
"\n",
"def increment(state: LoopState):\n",
" new_count = state['count'] + 1\n",
" print(f'当前 count 从 {state[\"count\"]} 变成 {new_count}')\n",
" return {'count': new_count}\n",
"\n",
"def should_continue(state: LoopState):\n",
" if state['count'] < 3:\n",
" return 'continue'\n",
" return 'stop'\n",
"\n",
"builder = StateGraph(LoopState)\n",
"builder.add_node('increment', increment)\n",
"\n",
"builder.add_edge(START, 'increment')\n",
"builder.add_conditional_edges(\n",
" 'increment',\n",
" should_continue,\n",
" {\n",
" 'continue': 'increment',\n",
" 'stop': END\n",
" }\n",
")\n",
"\n",
"loop_graph = builder.compile()\n",
"result = loop_graph.invoke({'count': 0})\n",
"print('最终状态:', result)"
]
},
{
"cell_type": "markdown",
"id": "26274c73",
"metadata": {},
"source": [
"### 代码解释\n",
"\n",
"这个例子看起来简单,但它已经具备了很多 Agent 工作流的核心思想。\n",
"\n",
"#### `increment` 节点\n",
"它的职责非常明确:每次把 `count` 加 1。\n",
"\n",
"例如:\n",
"\n",
"- 输入 `count=0`,输出 `count=1`\n",
"- 输入 `count=1`,输出 `count=2`\n",
"- 输入 `count=2`,输出 `count=3`\n",
"\n",
"#### `should_continue` 路由函数\n",
"它负责判断是继续循环,还是结束流程。\n",
"\n",
"规则很简单:\n",
"\n",
"- 如果 `count < 3`,返回 `'continue'`\n",
"- 否则返回 `'stop'`\n",
"\n",
"#### 条件边的映射\n",
"这里最关键的一点是:\n",
"\n",
"- `'continue'` 映射到 `'increment'`\n",
"- `'stop'` 映射到 `END`\n",
"\n",
"这就意味着:\n",
"\n",
"- 只要还没达到条件,就重新回到同一个节点\n",
"- 达到条件后,就退出图\n",
"\n",
"这就是循环结构的本质。\n",
"\n",
"#### 执行过程\n",
"如果初始状态是 `{'count': 0}`,执行路径如下:\n",
"\n",
"1. 进入 `increment`,变成 `count=1`\n",
"2. 判断 `1 < 3`,继续\n",
"3. 再次进入 `increment`,变成 `count=2`\n",
"4. 判断 `2 < 3`,继续\n",
"5. 再次进入 `increment`,变成 `count=3`\n",
"6. 判断 `3 < 3` 不成立,停止\n",
"7. 进入 `END`\n",
"\n",
"也就是说,图结构中的循环并不是 `for` 或 `while` 语法写出来的,而是通过‘节点 + 条件边回跳’实现的。\n",
"\n",
"这也是 LangGraph 很强大的地方:它把‘流程控制’变成了图本身的一部分。"
]
},
{
"cell_type": "markdown",
"id": "8c2c4ae1",
"metadata": {},
"source": [
"## 7. 图结构和普通 Python 流程控制有什么区别\n",
"\n",
"你可能会问:\n",
"\n",
"‘这些分支和循环,我直接用 `if`、`while` 不也能写吗?’\n",
"\n",
"当然可以。但图结构有几个明显优势:\n",
"\n",
"| 对比点 | 普通 Python 代码 | 图结构 |\n",
"| --- | --- | --- |\n",
"| 流程表达方式 | 写在代码逻辑里 | 显式画成节点和边 |\n",
"| 可视化能力 | 弱 | 强 |\n",
"| 分支/回路可读性 | 复杂时容易乱 | 更直观 |\n",
"| 与 Agent / 工作流结合 | 需要手动组织 | 天然适合 |\n",
"| 状态传递 | 自己管理变量 | 统一用 State 管理 |\n",
"\n",
"最核心的一点是:\n",
"\n",
"**图结构不是为了替代 Python而是为了让复杂流程更容易设计、理解、调试和扩展。**"
]
},
{
"cell_type": "markdown",
"id": "f82467cf",
"metadata": {},
"source": [
"## 8. 一个更贴近 Agent 的理解方式\n",
"\n",
"如果把图结构放到 AI Agent 场景中,可以这样理解:\n",
"\n",
"- 一个节点可以是‘调用大模型’\n",
"- 一个节点可以是‘调用工具’\n",
"- 一个节点可以是‘检查结果是否满足要求’\n",
"- 条件边可以决定‘是继续调用工具,还是给用户最终答案’\n",
"\n",
"也就是说,后面你学到的复杂 Agent本质上大多都可以拆成\n",
"\n",
"```\n",
"理解问题 -> 决策 -> 调工具 -> 检查结果 -> 继续/结束\n",
"```\n",
"\n",
"而图结构正是把这条路线显式表达出来的最好方式。"
]
},
{
"cell_type": "markdown",
"id": "f8ced671",
"metadata": {},
"source": [
"## 9. 本节小结\n",
"\n",
"本节课你需要记住三个最重要的结论:\n",
"\n",
"1. **State 是共享数据**:节点之间通过状态传递信息\n",
"2. **Node 是处理步骤**:每个节点负责完成一个明确任务\n",
"3. **Edge 决定执行路径**:普通边用于顺序执行,条件边用于分支和循环\n",
"\n",
"掌握了这三点,你就已经理解了图结构的核心。后面无论是多轮工作流、工具调用循环,还是多智能体协作,本质上都只是图结构的进一步扩展。"
]
},
{
"cell_type": "markdown",
"id": "3ec471d7",
"metadata": {},
"source": [
"## 10. 本节练习\n",
"\n",
"1. 修改顺序执行示例,在 `multiply_two` 后面再加一个节点 `minus_three`,让最终结果再减 3\n",
"2. 修改条件分支示例,把及格线从 60 改成 80观察结果变化\n",
"3. 修改循环示例,把终止条件从 `count == 3` 改成 `count == 5`\n",
"4. 尝试自己画出三个示例的流程图\n",
"5. 思考:如果要做一个‘查询资料直到找到答案为止’的 Agent它更像本节中的分支图还是循环图为什么"
]
}
],
"metadata": {
"kernelspec": {
"display_name": ".venv",
"language": "python",
"name": "python3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.14.4"
}
},
"nbformat": 4,
"nbformat_minor": 5
}