{ "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 }