{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 22_任务调度型智能体\n",
    "\n",
    "## 学习目标\n",
    "1. 理解任务调度型智能体的设计思路\n",
    "2. 掌握使用 LangGraph 实现任务分解、执行与结果汇总\n",
    "3. 能够构建简单的自动化任务处理智能体\n",
    "\n",
    "本节课会用一个通俗例子来理解任务调度：**把一个大任务拆成多个小任务，再安排合适的节点依次完成，最后汇总结果**。\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 1. 什么是任务调度型智能体\n",
    "\n",
    "任务调度型智能体可以理解为一个“项目经理型”智能体。用户只提出一个目标，它不会马上盲目执行，而是先思考：\n",
    "\n",
    "1. 这个目标可以拆成哪些步骤？\n",
    "2. 哪些步骤需要先做，哪些步骤可以后做？\n",
    "3. 每一步应该交给哪个工具、函数或智能体执行？\n",
    "4. 每一步完成后，如何把结果合并成最终答案？\n",
    "\n",
    "例如用户说：\n",
    "\n",
    "> 帮我分析一个产品的用户反馈，并给出改进建议。\n",
    "\n",
    "任务调度型智能体可能会拆成：\n",
    "\n",
    "1. 收集用户反馈\n",
    "2. 对反馈进行分类\n",
    "3. 提取高频问题\n",
    "4. 生成改进建议\n",
    "5. 输出结构化报告\n",
    "\n",
    "这类智能体的核心不是“单次回答”，而是“组织多个步骤完成复杂任务”。\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 2. 任务调度的基本流程\n",
    "\n",
    "一个典型任务调度型智能体通常包含 4 个环节：\n",
    "\n",
    "| 环节 | 作用 | 通俗理解 |\n",
    "| --- | --- | --- |\n",
    "| 任务输入 | 接收用户目标 | 用户告诉智能体要做什么 |\n",
    "| 任务分解 | 把大任务拆成小任务 | 项目经理制定待办清单 |\n",
    "| 任务执行 | 按顺序执行每个小任务 | 不同同事完成各自工作 |\n",
    "| 结果汇总 | 整合所有执行结果 | 写成最终报告交给用户 |\n",
    "\n",
    "在 LangGraph 中，我们可以把每个环节看成一个 **节点**，节点之间通过 **边** 连接起来，形成一个工作流。\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 3. 先不用 LangGraph：用普通 Python 理解任务调度\n",
    "\n",
    "在学习 LangGraph 前，先用普通 Python 模拟一个最小任务调度流程。这个例子不依赖任何外部库，更容易理解“任务分解、执行、汇总”的本质。\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def split_task(user_goal):\n",
    "    \"\"\"把用户的大目标拆成多个小任务。\"\"\"\n",
    "    tasks = [\n",
    "        f\"理解用户目标：{user_goal}\",\n",
    "        \"收集与目标相关的信息\",\n",
    "        \"分析关键信息\",\n",
    "        \"生成最终建议\",\n",
    "    ]\n",
    "    return tasks\n",
    "\n",
    "\n",
    "def execute_task(task):\n",
    "    \"\"\"模拟执行一个小任务。\"\"\"\n",
    "    return f\"已完成：{task}\"\n",
    "\n",
    "\n",
    "def summarize_results(results):\n",
    "    \"\"\"把多个小任务的执行结果汇总成最终输出。\"\"\"\n",
    "    report = \"任务执行报告：\\n\"\n",
    "    for index, result in enumerate(results, start=1):\n",
    "        report += f\"{index}. {result}\\n\"\n",
    "    return report\n",
    "\n",
    "\n",
    "user_goal = \"为一家咖啡店设计会员运营方案\"\n",
    "tasks = split_task(user_goal)\n",
    "results = [execute_task(task) for task in tasks]\n",
    "final_report = summarize_results(results)\n",
    "\n",
    "print(final_report)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 代码解释\n",
    "\n",
    "这段代码展示了任务调度的最小模型：\n",
    "\n",
    "1. `split_task(user_goal)` 负责拆解任务。它接收用户的大目标，返回一个任务列表。\n",
    "2. `execute_task(task)` 负责执行单个小任务。这里为了演示，只返回“已完成”的文字。真实项目中，这里可以调用搜索工具、数据库、LLM 或其他业务函数。\n",
    "3. `summarize_results(results)` 负责汇总结果。它把每个小任务的输出整理成最终报告。\n",
    "4. `results = [execute_task(task) for task in tasks]` 是任务调度的执行阶段，会逐个执行任务列表里的每一项。\n",
    "\n",
    "这个例子虽然简单，但已经包含任务调度型智能体最重要的 3 个动作：**拆、做、汇总**。\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 4. LangGraph 的核心概念\n",
    "\n",
    "LangGraph 是一个用“图”组织智能体流程的框架。\n",
    "\n",
    "可以把它想象成一张流程图：\n",
    "\n",
    "```text\n",
    "开始 -> 任务分解 -> 任务执行 -> 结果汇总 -> 结束\n",
    "```\n",
    "\n",
    "在 LangGraph 中常见概念如下：\n",
    "\n",
    "| 概念 | 含义 | 通俗理解 |\n",
    "| --- | --- | --- |\n",
    "| State | 工作流中流动的数据 | 一个不断更新的任务档案袋 |\n",
    "| Node | 处理数据的函数 | 流程图中的一个步骤 |\n",
    "| Edge | 节点之间的连接 | 告诉程序下一步去哪里 |\n",
    "| Graph | 完整工作流 | 整张流程图 |\n",
    "\n",
    "接下来我们用 LangGraph 实现同样的“拆、做、汇总”流程。\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 5. 安装依赖\n",
    "\n",
    "如果当前环境没有安装 LangGraph，可以先运行下面的命令。已经安装过的环境可以跳过。\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 如果没有安装 LangGraph，取消下一行注释后运行\n",
    "# %pip install langgraph\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 代码解释\n",
    "\n",
    "`%pip install langgraph` 是 Jupyter Notebook 中安装 Python 包的写法。\n",
    "\n",
    "这里默认把安装命令注释掉，是为了避免重复安装。如果运行后提示找不到 `langgraph`，再去掉前面的 `#` 执行即可。\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 6. 定义工作流状态 State\n",
    "\n",
    "State 是 LangGraph 中非常重要的概念。它表示工作流运行过程中一直携带和更新的数据。\n",
    "\n",
    "本例中，我们希望 State 保存 4 类信息：\n",
    "\n",
    "1. `goal`：用户输入的大目标\n",
    "2. `tasks`：拆分后的小任务列表\n",
    "3. `results`：每个小任务的执行结果\n",
    "4. `final_answer`：最终汇总答案\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from typing import List, TypedDict\n",
    "\n",
    "\n",
    "class TaskState(TypedDict):\n",
    "    goal: str\n",
    "    tasks: List[str]\n",
    "    results: List[str]\n",
    "    final_answer: str\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 代码解释\n",
    "\n",
    "这段代码定义了工作流中的数据结构：\n",
    "\n",
    "1. `TypedDict` 用来描述字典里应该有哪些字段。它不会改变程序运行方式，但能让代码更清晰。\n",
    "2. `TaskState` 表示整个任务调度过程中共享的数据。每个节点都会读取它，也可以返回新的字段值更新它。\n",
    "3. `goal: str` 表示用户目标是字符串。\n",
    "4. `tasks: List[str]` 表示任务列表，里面每一项都是字符串。\n",
    "5. `results: List[str]` 表示执行结果列表。\n",
    "6. `final_answer: str` 表示最终回答。\n",
    "\n",
    "可以把 `TaskState` 理解为一张任务表，工作流每走一步，就在这张表上补充一些内容。\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 7. 编写节点函数\n",
    "\n",
    "节点函数就是工作流中的一个步骤。每个节点接收当前 `state`，处理后返回要更新的字段。\n",
    "\n",
    "下面定义 3 个节点：\n",
    "\n",
    "1. `planner_node`：任务分解节点\n",
    "2. `worker_node`：任务执行节点\n",
    "3. `summary_node`：结果汇总节点\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def planner_node(state: TaskState):\n",
    "    goal = state[\"goal\"]\n",
    "    tasks = [\n",
    "        f\"明确目标：{goal}\",\n",
    "        \"列出目标用户和使用场景\",\n",
    "        \"设计可执行的行动步骤\",\n",
    "        \"整理成清晰的建议清单\",\n",
    "    ]\n",
    "    return {\"tasks\": tasks}\n",
    "\n",
    "\n",
    "def worker_node(state: TaskState):\n",
    "    results = []\n",
    "    for task in state[\"tasks\"]:\n",
    "        result = f\"完成任务：{task}\"\n",
    "        results.append(result)\n",
    "    return {\"results\": results}\n",
    "\n",
    "\n",
    "def summary_node(state: TaskState):\n",
    "    lines = [\"最终任务调度结果：\"]\n",
    "    for index, result in enumerate(state[\"results\"], start=1):\n",
    "        lines.append(f\"{index}. {result}\")\n",
    "    final_answer = \"\\n\".join(lines)\n",
    "    return {\"final_answer\": final_answer}\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 代码解释\n",
    "\n",
    "这段代码定义了 3 个节点函数：\n",
    "\n",
    "1. `planner_node(state)`：从 `state[\"goal\"]` 读取用户目标，然后生成 `tasks` 任务列表。返回 `{\"tasks\": tasks}` 表示把任务列表写回 State。\n",
    "2. `worker_node(state)`：读取 `state[\"tasks\"]`，循环执行每个任务。这里用字符串模拟执行结果，真实项目中可以替换为调用工具、调用模型、查询数据库等操作。\n",
    "3. `summary_node(state)`：读取 `state[\"results\"]`，把多个结果拼接成一段最终答案。\n",
    "\n",
    "需要注意：节点函数通常不需要返回完整 State，只返回自己负责更新的字段即可。LangGraph 会把这些字段合并回当前 State。\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 8. 用 LangGraph 连接节点\n",
    "\n",
    "现在有了节点函数，下一步就是把这些节点连成一张图。\n",
    "\n",
    "流程如下：\n",
    "\n",
    "```text\n",
    "START -> planner -> worker -> summary -> END\n",
    "```\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from langgraph.graph import END, START, StateGraph\n",
    "\n",
    "\n",
    "workflow = StateGraph(TaskState)\n",
    "\n",
    "workflow.add_node(\"planner\", planner_node)\n",
    "workflow.add_node(\"worker\", worker_node)\n",
    "workflow.add_node(\"summary\", summary_node)\n",
    "\n",
    "workflow.add_edge(START, \"planner\")\n",
    "workflow.add_edge(\"planner\", \"worker\")\n",
    "workflow.add_edge(\"worker\", \"summary\")\n",
    "workflow.add_edge(\"summary\", END)\n",
    "\n",
    "app = workflow.compile()\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 代码解释\n",
    "\n",
    "这段代码把节点组织成可运行的工作流：\n",
    "\n",
    "1. `StateGraph(TaskState)` 创建一张图，并说明这张图中的数据结构是 `TaskState`。\n",
    "2. `add_node(\"planner\", planner_node)` 把 `planner_node` 注册成名为 `planner` 的节点。\n",
    "3. `add_edge(START, \"planner\")` 表示工作流从 `planner` 节点开始。\n",
    "4. `add_edge(\"planner\", \"worker\")` 表示任务分解完成后，进入任务执行节点。\n",
    "5. `add_edge(\"worker\", \"summary\")` 表示任务执行完成后，进入结果汇总节点。\n",
    "6. `add_edge(\"summary\", END)` 表示汇总完成后，工作流结束。\n",
    "7. `workflow.compile()` 会把流程图编译成一个可以调用的应用对象 `app`。\n",
    "\n",
    "这一步相当于把流程图画好，并告诉程序每一步应该怎么走。\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 9. 运行任务调度智能体\n",
    "\n",
    "工作流编译完成后，就可以传入用户目标并运行了。\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "initial_state = {\n",
    "    \"goal\": \"为一家咖啡店设计会员运营方案\",\n",
    "    \"tasks\": [],\n",
    "    \"results\": [],\n",
    "    \"final_answer\": \"\",\n",
    "}\n",
    "\n",
    "final_state = app.invoke(initial_state)\n",
    "\n",
    "print(final_state[\"final_answer\"])\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 代码解释\n",
    "\n",
    "这段代码负责真正运行工作流：\n",
    "\n",
    "1. `initial_state` 是初始状态，里面放入用户目标 `goal`，其余字段先给空值。\n",
    "2. `app.invoke(initial_state)` 会启动 LangGraph 工作流。数据会依次经过 `planner`、`worker`、`summary` 三个节点。\n",
    "3. 运行结束后返回 `final_state`，它包含整个流程运行后的完整状态。\n",
    "4. `final_state[\"final_answer\"]` 取出最终汇总结果并打印。\n",
    "\n",
    "从这个例子可以看到，LangGraph 的优势是把复杂流程拆成清晰节点，每个节点只负责一件事。\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 10. 加入条件判断：根据任务复杂度选择路径\n",
    "\n",
    "真实任务中，不是所有问题都需要复杂调度。\n",
    "\n",
    "例如：\n",
    "\n",
    "- 简单问题：直接回答即可\n",
    "- 复杂问题：先拆分，再执行，再汇总\n",
    "\n",
    "LangGraph 支持条件边，可以根据 State 的内容决定下一步走哪条路径。\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from typing import Literal\n",
    "\n",
    "\n",
    "def judge_complexity(state: TaskState) -> Literal[\"simple\", \"complex\"]:\n",
    "    goal = state[\"goal\"]\n",
    "    if len(goal) <= 12:\n",
    "        return \"simple\"\n",
    "    return \"complex\"\n",
    "\n",
    "\n",
    "def direct_answer_node(state: TaskState):\n",
    "    answer = f\"这是一个简单任务，可以直接处理：{state['goal']}\"\n",
    "    return {\"final_answer\": answer}\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 代码解释\n",
    "\n",
    "这段代码新增了两个函数：\n",
    "\n",
    "1. `judge_complexity(state)` 用来判断任务复杂度。这里为了教学简单，用目标文字长度作为判断标准：短任务走 `simple`，长任务走 `complex`。真实项目中可以让大模型判断复杂度。\n",
    "2. `Literal[\"simple\", \"complex\"]` 表示这个函数只会返回两个固定值之一，方便我们理解后续条件分支。\n",
    "3. `direct_answer_node(state)` 是简单任务的处理节点，不做任务分解，直接生成最终答案。\n",
    "\n",
    "这个例子说明：任务调度不一定永远走同一条流程，可以根据任务情况动态选择路径。\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 11. 构建带条件分支的工作流\n",
    "\n",
    "下面把简单路径和复杂路径组合到同一张图里。\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "branch_workflow = StateGraph(TaskState)\n",
    "\n",
    "branch_workflow.add_node(\"direct_answer\", direct_answer_node)\n",
    "branch_workflow.add_node(\"planner\", planner_node)\n",
    "branch_workflow.add_node(\"worker\", worker_node)\n",
    "branch_workflow.add_node(\"summary\", summary_node)\n",
    "\n",
    "branch_workflow.add_conditional_edges(\n",
    "    START,\n",
    "    judge_complexity,\n",
    "    {\n",
    "        \"simple\": \"direct_answer\",\n",
    "        \"complex\": \"planner\",\n",
    "    },\n",
    ")\n",
    "\n",
    "branch_workflow.add_edge(\"direct_answer\", END)\n",
    "branch_workflow.add_edge(\"planner\", \"worker\")\n",
    "branch_workflow.add_edge(\"worker\", \"summary\")\n",
    "branch_workflow.add_edge(\"summary\", END)\n",
    "\n",
    "branch_app = branch_workflow.compile()\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 代码解释\n",
    "\n",
    "这段代码构建了一个带分支的任务调度工作流：\n",
    "\n",
    "1. `add_conditional_edges` 表示添加条件边。它会先调用 `judge_complexity`，再根据返回值选择下一步。\n",
    "2. 如果返回 `simple`，流程进入 `direct_answer` 节点，然后直接结束。\n",
    "3. 如果返回 `complex`，流程进入 `planner` 节点，再继续执行 `worker` 和 `summary`。\n",
    "4. 这张图有两条路径：一条适合简单任务，一条适合复杂任务。\n",
    "\n",
    "这就是任务调度型智能体的一个关键能力：**不是固定执行所有步骤，而是根据任务情况选择合适流程**。\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 12. 测试条件分支效果\n",
    "\n",
    "分别输入一个简单任务和一个复杂任务，观察工作流走向的差异。\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "simple_state = {\n",
    "    \"goal\": \"写标题\",\n",
    "    \"tasks\": [],\n",
    "    \"results\": [],\n",
    "    \"final_answer\": \"\",\n",
    "}\n",
    "\n",
    "complex_state = {\n",
    "    \"goal\": \"为一家咖啡店设计会员运营方案，并给出执行步骤\",\n",
    "    \"tasks\": [],\n",
    "    \"results\": [],\n",
    "    \"final_answer\": \"\",\n",
    "}\n",
    "\n",
    "print(\"简单任务结果：\")\n",
    "print(branch_app.invoke(simple_state)[\"final_answer\"])\n",
    "\n",
    "print(\"\\n复杂任务结果：\")\n",
    "print(branch_app.invoke(complex_state)[\"final_answer\"])\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 代码解释\n",
    "\n",
    "这段代码测试了条件分支：\n",
    "\n",
    "1. `simple_state` 的目标是“写标题”，文字较短，所以 `judge_complexity` 返回 `simple`，流程会直接进入 `direct_answer`。\n",
    "2. `complex_state` 的目标更长，所以返回 `complex`，流程会进入 `planner -> worker -> summary`。\n",
    "3. 两次都调用 `branch_app.invoke(...)`，但由于输入不同，工作流自动选择了不同执行路径。\n",
    "\n",
    "这说明任务调度型智能体可以做到“简单任务快速处理，复杂任务分步处理”。\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 13. 小结\n",
    "\n",
    "本节课学习了任务调度型智能体的基本设计方法：\n",
    "\n",
    "1. 任务调度型智能体适合处理多步骤、复杂目标。\n",
    "2. 它的核心流程是：任务输入、任务分解、任务执行、结果汇总。\n",
    "3. LangGraph 可以把每个步骤定义成节点，并用边连接成工作流。\n",
    "4. State 是工作流中流动的数据，节点通过读取和更新 State 协同完成任务。\n",
    "5. 条件边可以让智能体根据任务情况选择不同执行路径。\n",
    "\n",
    "掌握这些内容后，就可以继续扩展：把 `worker_node` 替换成真实工具调用、搜索引擎调用、数据库查询或大模型调用，从而构建更实用的自动化任务处理智能体。\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 14. 练习题\n",
    "\n",
    "1. 修改 `planner_node`，让它针对“写一篇公众号文章”生成任务列表。\n",
    "2. 修改 `worker_node`，让不同任务返回不同格式的结果。\n",
    "3. 修改 `judge_complexity`，不要用文字长度判断，而是根据是否包含“方案”“分析”“计划”等关键词判断复杂度。\n",
    "4. 尝试新增一个 `review_node`，在最终输出前检查结果是否完整。\n"
   ]
  }
 ],
 "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": 4
}