Files
ai-agent-dev/18_条件边.ipynb
2026-07-08 10:09:42 +08:00

595 lines
19 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",
"metadata": {},
"source": [
"# 18 条件边\n",
"\n",
"## 学习目标\n",
"1. 理解条件边Conditional Edges的作用和使用场景\n",
"2. 掌握 `add_conditional_edges` 实现分支流程\n",
"3. 能够根据状态值动态决定图的执行路径\n",
"4. 理解路由函数在图中的作用\n",
"5. 能够读懂并修改一个带分支的图流程"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## 1. 为什么需要条件边\n",
"\n",
"前一节我们学习了普通边。普通边的特点是:**下一步固定不变**。\n",
"\n",
"例如:\n",
"\n",
"```\n",
"START -> node_a -> node_b -> END\n",
"```\n",
"\n",
"这条流程里,不管输入是什么,`node_a` 做完以后都会进入 `node_b`。\n",
"\n",
"但真实业务中,流程往往不是固定直走的,而是会根据情况分支。比如:\n",
"\n",
"- 分数及格,就进入‘通过’节点\n",
"- 分数不及格,就进入‘补考’节点\n",
"- 用户问题完整,就直接回答\n",
"- 用户问题不完整,就先追问\n",
"- 检索结果足够,就结束\n",
"- 检索结果不够,就继续查\n",
"\n",
"这类‘根据当前状态决定下一步去哪’的场景,就要用到**条件边**。"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## 2. 条件边和普通边的区别\n",
"\n",
"可以把两种边对比着理解:\n",
"\n",
"| 类型 | 决定方式 | 典型场景 |\n",
"| --- | --- | --- |\n",
"| 普通边 `add_edge` | 固定写死 | 顺序执行流程 |\n",
"| 条件边 `add_conditional_edges` | 根据状态动态判断 | 分支、回路、路由控制 |\n",
"\n",
"最核心的区别只有一句话:\n",
"\n",
"- 普通边:下一步是提前写死的\n",
"- 条件边:下一步是运行时判断出来的\n",
"\n",
"也就是说,条件边让图开始‘有判断力’。"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## 3. 条件边的基本思路\n",
"\n",
"使用条件边时,通常有三样东西要配合起来:\n",
"\n",
"1. **一个普通节点**:先处理当前状态\n",
"2. **一个路由函数**:根据状态判断下一步去哪\n",
"3. **一个分支映射表**:告诉图不同返回值分别对应哪个节点\n",
"\n",
"写法通常像这样:\n",
"\n",
"```python\n",
"builder.add_conditional_edges(\n",
" '某个节点',\n",
" route_function,\n",
" {\n",
" '路径A': 'node_a',\n",
" '路径B': 'node_b'\n",
" }\n",
")\n",
"```\n",
"\n",
"它的意思是:\n",
"\n",
"- 先执行‘某个节点’\n",
"- 执行完以后,不直接写死下一步\n",
"- 而是调用 `route_function(state)` 看返回什么\n",
"- 如果返回 `'路径A'`,就走到 `node_a`\n",
"- 如果返回 `'路径B'`,就走到 `node_b`"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## 4. 第一个例子:根据分数决定是否通过\n",
"\n",
"先从最直观的例子开始。\n",
"\n",
"流程目标:\n",
"\n",
"- 如果分数大于等于 60进入通过节点\n",
"- 如果分数小于 60进入补考节点\n",
"\n",
"流程图如下:\n",
"\n",
"```\n",
"START -> check_score -> pass_node / fail_node -> END\n",
"```"
]
},
{
"cell_type": "code",
"execution_count": null,
"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'\n",
" return 'fail'\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': 'pass_node',\n",
" 'fail': 'fail_node'\n",
" }\n",
")\n",
"builder.add_edge('pass_node', END)\n",
"builder.add_edge('fail_node', END)\n",
"\n",
"graph = builder.compile()\n",
"\n",
"print('分数 85')\n",
"print(graph.invoke({'score': 85, 'result': ''}))\n",
"\n",
"print('\\n分数 45')\n",
"print(graph.invoke({'score': 45, 'result': ''}))"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"### 代码解释\n",
"\n",
"这个例子非常典型,几乎把条件边最核心的要点都展示出来了。\n",
"\n",
"#### 状态结构\n",
"状态里有两个字段:\n",
"\n",
"- `score`:输入分数\n",
"- `result`:最终判断结果\n",
"\n",
"#### `check_score` 节点\n",
"这个节点本身没有修改状态,只是打印当前分数。\n",
"\n",
"它返回 `{}`,表示‘这个节点不更新任何字段’。\n",
"\n",
"这说明一个节点不一定非要负责生成结果,它也可以只是一个‘分支前的准备步骤’。\n",
"\n",
"#### `pass_node` 和 `fail_node`\n",
"这两个节点分别代表分支后的两条路径:\n",
"\n",
"- 通过路径:写入‘成绩合格,顺利通过’\n",
"- 失败路径:写入‘成绩不合格,需要补考’\n",
"\n",
"#### `route_by_score(state)`\n",
"这是本例的路由函数。它的任务不是修改状态,而是‘判断下一步去哪’。\n",
"\n",
"判断逻辑很简单:\n",
"\n",
"- 如果 `score >= 60`,返回 `'pass'`\n",
"- 否则返回 `'fail'`\n",
"\n",
"注意:这里返回的不是节点函数本身,而是一个**分支标记**。\n",
"\n",
"#### `add_conditional_edges(...)`\n",
"这一段可以拆成三层意思:\n",
"\n",
"1. 从 `check_score` 节点出来以后,不直接写死下一步\n",
"2. 调用 `route_by_score(state)` 得到一个返回值\n",
"3. 根据映射表决定真正跳去哪个节点\n",
"\n",
"也就是说:\n",
"\n",
"- 返回 `'pass'` -> 去 `pass_node`\n",
"- 返回 `'fail'` -> 去 `fail_node`\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",
"metadata": {},
"source": [
"## 5. 第二个例子:问题完整就直接回答,不完整就先追问\n",
"\n",
"下面我们做一个更贴近 AI 助手的例子。\n",
"\n",
"假设用户提问时,有的问题已经很清楚,有的问题信息还不够。\n",
"\n",
"流程目标:\n",
"\n",
"- 如果问题完整,直接进入回答节点\n",
"- 如果问题不完整,先进入追问节点\n",
"\n",
"这个例子可以帮助你理解:条件边不仅能处理分数、数字,也能处理业务判断。"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"from typing_extensions import TypedDict\n",
"from langgraph.graph import StateGraph, START, END\n",
"\n",
"class QuestionState(TypedDict):\n",
" user_question: str\n",
" is_complete: bool\n",
" response: str\n",
"\n",
"def analyze_question(state: QuestionState):\n",
" print(f'收到问题:{state[\"user_question\"]}')\n",
" return {}\n",
"\n",
"def ask_more(state: QuestionState):\n",
" return {'response': '你的问题还不够完整,请补充更多背景信息。'}\n",
"\n",
"def answer_directly(state: QuestionState):\n",
" return {'response': '问题信息已经足够,我现在直接为你回答。'}\n",
"\n",
"def route_question(state: QuestionState):\n",
" if state['is_complete']:\n",
" return 'answer'\n",
" return 'ask_more'\n",
"\n",
"builder = StateGraph(QuestionState)\n",
"builder.add_node('analyze_question', analyze_question)\n",
"builder.add_node('ask_more', ask_more)\n",
"builder.add_node('answer_directly', answer_directly)\n",
"\n",
"builder.add_edge(START, 'analyze_question')\n",
"builder.add_conditional_edges(\n",
" 'analyze_question',\n",
" route_question,\n",
" {\n",
" 'answer': 'answer_directly',\n",
" 'ask_more': 'ask_more'\n",
" }\n",
")\n",
"builder.add_edge('ask_more', END)\n",
"builder.add_edge('answer_directly', END)\n",
"\n",
"graph = builder.compile()\n",
"\n",
"print('完整问题:')\n",
"print(graph.invoke({\n",
" 'user_question': '请解释什么是条件边,并举一个例子',\n",
" 'is_complete': True,\n",
" 'response': ''\n",
"}))\n",
"\n",
"print('\\n不完整问题')\n",
"print(graph.invoke({\n",
" 'user_question': '帮我看看这个',\n",
" 'is_complete': False,\n",
" 'response': ''\n",
"}))"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"### 代码解释\n",
"\n",
"这个例子和前面的‘分数判断’相比,更接近真实 AI 产品中的流程控制。\n",
"\n",
"#### 状态字段的意义\n",
"这里的状态有三个字段:\n",
"\n",
"- `user_question`:用户输入的问题\n",
"- `is_complete`:问题是否完整\n",
"- `response`:最终回复\n",
"\n",
"#### `analyze_question` 节点\n",
"这个节点先接收问题并打印出来,相当于‘进入分析阶段’。\n",
"\n",
"它没有直接修改状态,而是把路由判断交给后面的路由函数。\n",
"\n",
"#### `route_question(state)`\n",
"这个函数根据 `is_complete` 的真假做判断:\n",
"\n",
"- `True` -> 返回 `'answer'`\n",
"- `False` -> 返回 `'ask_more'`\n",
"\n",
"这意味着流程下一步是动态的,而不是提前写死的。\n",
"\n",
"#### 两条分支的业务含义\n",
"- `answer_directly`:说明信息足够,可以直接处理\n",
"- `ask_more`:说明信息不足,需要先追问\n",
"\n",
"#### 为什么这个例子很重要\n",
"它让你看到:条件边并不只是‘程序里的 if 判断’,更像是业务流程里的‘路由器’。\n",
"\n",
"很多 Agent 系统都会有类似逻辑:\n",
"\n",
"- 信息够不够?\n",
"- 要不要调用工具?\n",
"- 要不要继续搜索?\n",
"- 能不能直接输出答案?\n",
"\n",
"而这些判断,往往都可以通过条件边表达出来。"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## 6. 第三个例子:条件边也可以用于继续或结束\n",
"\n",
"条件边不仅可以做左右分支,还经常用来做‘继续/停止’的判断。\n",
"\n",
"这类场景非常常见,例如:\n",
"\n",
"- 检索结果不够,就继续检索\n",
"- 输出格式不对,就继续生成\n",
"- 数量还没到要求,就继续累加\n",
"\n",
"下面用一个简单例子说明:如果 `count < 3`,就继续执行;否则结束。"
]
},
{
"cell_type": "code",
"execution_count": null,
"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 add_one(state: LoopState):\n",
" new_count = state['count'] + 1\n",
" print(f'当前 count 从 {state[\"count\"]} 变成 {new_count}')\n",
" return {'count': new_count}\n",
"\n",
"def route_loop(state: LoopState):\n",
" if state['count'] < 3:\n",
" return 'continue'\n",
" return 'stop'\n",
"\n",
"builder = StateGraph(LoopState)\n",
"builder.add_node('add_one', add_one)\n",
"\n",
"builder.add_edge(START, 'add_one')\n",
"builder.add_conditional_edges(\n",
" 'add_one',\n",
" route_loop,\n",
" {\n",
" 'continue': 'add_one',\n",
" 'stop': END\n",
" }\n",
")\n",
"\n",
"graph = builder.compile()\n",
"result = graph.invoke({'count': 0})\n",
"print('最终结果:', result)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"### 代码解释\n",
"\n",
"这个例子非常重要,因为它说明:**条件边不仅能分叉,也能回跳。**\n",
"\n",
"#### `add_one` 节点\n",
"这个节点每次把 `count` 加 1。\n",
"\n",
"也就是说,它是一个不断重复执行的工作步骤。\n",
"\n",
"#### `route_loop(state)`\n",
"这个路由函数负责判断接下来是继续,还是停止:\n",
"\n",
"- 如果 `count < 3`,返回 `'continue'`\n",
"- 否则返回 `'stop'`\n",
"\n",
"#### 条件边映射\n",
"这里映射关系很关键:\n",
"\n",
"- `'continue'` -> `'add_one'`\n",
"- `'stop'` -> `END`\n",
"\n",
"也就是说:\n",
"\n",
"- 还没满足条件,就重新回到原节点\n",
"- 满足条件后,就离开流程\n",
"\n",
"#### 执行过程\n",
"如果初始值是 `count=0`,路径会是:\n",
"\n",
"1. `0 -> 1`,继续\n",
"2. `1 -> 2`,继续\n",
"3. `2 -> 3`,停止\n",
"4. 进入 `END`\n",
"\n",
"从这个例子你可以看到,条件边不只是“二选一分支”,它还是循环流程的基础。\n",
"\n",
"后面很多 Agent 的‘反复尝试直到成功’流程,本质上都是这种写法。"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## 7. 路由函数应该怎么理解\n",
"\n",
"很多初学者最容易混淆的地方,就是把‘节点函数’和‘路由函数’混在一起。\n",
"\n",
"你可以这样区分:\n",
"\n",
"| 类型 | 主要职责 | 是否负责业务处理 |\n",
"| --- | --- | --- |\n",
"| 节点函数 | 读写状态、执行处理逻辑 | 是 |\n",
"| 路由函数 | 判断下一步走哪条路 | 通常不是重点 |\n",
"\n",
"换句话说:\n",
"\n",
"- 节点函数更像‘工人’\n",
"- 路由函数更像‘调度员’\n",
"\n",
"工人负责真正干活,调度员负责决定下一站去哪。\n",
"\n",
"理解这一点以后,条件边的结构就会非常清晰。"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## 8. 使用条件边时的常见错误\n",
"\n",
"### 8.1 路由函数返回值和映射表对不上\n",
"例如路由函数返回 `'pass'`,但映射表里只有 `'ok'` 和 `'fail'`。\n",
"\n",
"这样图就不知道该往哪里走。\n",
"\n",
"### 8.2 把节点名字和路由标记混为一谈\n",
"路由函数返回的可以是一个标记,例如 `'continue'`、`'stop'`、`'pass'`。\n",
"\n",
"这些标记不一定要和节点名完全相同,但必须和映射表中的 key 对应。\n",
"\n",
"### 8.3 以为条件边只能做两条分支\n",
"其实不止两条。\n",
"\n",
"你完全可以写三条、四条甚至更多路线,只要业务场景需要。\n",
"\n",
"### 8.4 忘记给分支节点连到 `END` 或后续节点\n",
"条件边只负责从当前节点分出去,不代表后面的路径已经自动补全。\n",
"\n",
"分支出去以后,该怎么结束、怎么汇合,仍然要继续写边。"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## 9. 条件边在 Agent 场景中的意义\n",
"\n",
"到了这里,你可以把条件边理解成 LangGraph 里的‘流程决策器’。\n",
"\n",
"在 Agent 场景中,它特别常见,因为 Agent 天生就不是固定直线流程。\n",
"\n",
"例如:\n",
"\n",
"- 如果用户问题明确,直接回答\n",
"- 如果用户问题不明确,先追问\n",
"- 如果需要外部信息,调用工具\n",
"- 如果工具结果不够,再查一次\n",
"- 如果结果足够,就进入最终回答\n",
"\n",
"你会发现,这些逻辑本质上都属于:\n",
"\n",
"**根据当前状态,动态决定下一步去哪。**\n",
"\n",
"这正是条件边最核心的价值。"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## 10. 本节小结\n",
"\n",
"本节最重要的内容有五点:\n",
"\n",
"1. **条件边用于动态决定流程路径**\n",
"2. **`add_conditional_edges` 是实现分支和回路的关键方法**\n",
"3. **路由函数负责返回分支标记,映射表负责把标记映射到节点**\n",
"4. **条件边不仅能做左右分支,也能做继续/停止这样的循环控制**\n",
"5. **很多 Agent 工作流,本质上都是条件边在控制路径**\n",
"\n",
"理解了条件边,你就真正进入了‘动态流程控制’这一层。后面的复杂图,大多都离不开它。"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## 11. 本节练习\n",
"\n",
"1. 修改第一个示例,把及格线从 60 改成 80观察分支结果变化\n",
"2. 修改第二个示例,再增加一种情况:如果问题为空,就返回一个专门的提示节点\n",
"3. 修改第三个示例,把停止条件从 `count < 3` 改成 `count < 5`\n",
"4. 思考:如果一个流程有 3 条不同分支,路由函数和映射表应该怎么设计?\n",
"5. 思考:条件边和普通边最大的本质区别是什么?"
]
}
],
"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
}