Agent 的任务分解策略Plan-and-Solve 比 Chain-of-Thought 更适合复杂任务一、深度引言与场景痛点如果你给 Agent 布置一个任务比如帮我分析这家公司过去三年的财报趋势和同行业对比后输出风险报告Chain-of-Thought 的方式是做一步看一步先查财报发现数据不全再补查发现需要行业数据又去查。等到推理链超过十几步Agent 自己都忘了最初想干啥。这不是模型能力不够而是推理方式不对。复杂任务的特点是步骤之间有依赖、中间信息需要聚合、路径可能分叉。Chain-of-Thought 擅长线性推理但面对需要分治、并行、回溯的任务它就像在迷宫里边走边猜——迟早出错。Plan-and-Solve 的思路就简单了先花 10% 的时间做计划把复杂任务拆成可独立验证的子任务再逐块执行。这相当于先画地图再走路比边走边问路靠谱得多。二、底层机制与原理深度剖析flowchart TD A[用户输入复杂任务] -- B[任务分析阶段] B -- C{复杂度判断} C --|简单任务| D[直接推理执行] C --|复杂任务| E[生成执行计划] E -- F[依赖关系排序] F -- G[子任务并行分配] G -- H1[子任务1: 数据采集] G -- H2[子任务2: 预处理] G -- H3[子任务3: 分析计算] H1 -- I[中间结果验证] H2 -- I H3 -- I I -- J{所有子任务通过?} J --|否| K[标记失败任务, 重新规划] K -- E J --|是| L[结果聚合与推理] L -- M[输出最终答案]图中核心的分叉在于复杂度判断这一步。不是所有任务都需要 Plan-and-Solve对于今天天气怎么样这种单步任务直接推理更快。Plan-and-Solve 的威力体现在依赖关系排序让子任务按正确顺序执行中间验证保证每一步结果可信失败重规划让系统有自愈能力。和 Chain-of-Thought 的关键区别是CoT 在推理过程中顺便分解Plan-and-Solve 是先显式生成计划再由执行器逐块执行。显式计划的好处是可审计、可打断、可重试、可并行。想象一下你做项目时是先写个甘特图再干活还是干到哪算哪前者显然更可控。三、生产级代码实现from __future__ import annotations import asyncio import json from dataclasses import dataclass, field from enum import Enum from typing import Any, Callable, Awaitable class TaskStatus(Enum): PENDING pending RUNNING running SUCCESS success FAILED failed SKIPPED skipped dataclass class SubTask: id: str description: str dependencies: list[str] field(default_factorylist) status: TaskStatus TaskStatus.PENDING result: Any None error: str dataclass class Plan: task_id: str subtasks: list[SubTask] raw_plan: str class PlanGenerator: 计划生成器 —— 调用 LLM 将复杂任务拆解为子任务 DAG async def generate(self, task: str, llm_call: Callable[[str], Awaitable[str]]) - Plan: prompt ( 你是一个任务分解专家。请将以下复杂任务拆解为可独立执行的子任务 输出 JSON 格式每个子任务包含 id、描述和依赖的 id 列表\n f任务{task} ) raw await llm_call(prompt) try: parsed json.loads(raw) except json.JSONDecodeError: # 降级将整个任务作为单个子任务 parsed [{id: task_0, description: task, dependencies: []}] subtasks [ SubTask( iditem[id], descriptionitem[description], dependenciesitem.get(dependencies, []), ) for item in parsed ] return Plan(task_idtask[:20], subtaskssubtasks, raw_planraw) class PlanExecutor: 计划执行器 —— 按拓扑排序逐个执行子任务支持并行 def __init__(self, execute_fn: Callable[[SubTask], Awaitable[Any]]): self._execute execute_fn self._completed: dict[str, SubTask] {} async def execute(self, plan: Plan, max_retries: int 2) - list[SubTask]: remaining {st.id: st for st in plan.subtasks} results: list[SubTask] [] while remaining: ready [ st for st in remaining.values() if all(dep in self._completed for dep in st.dependencies) ] if not ready: # 存在循环依赖或死锁 failed [st for st in remaining.values() if st.status TaskStatus.FAILED] if failed: raise RuntimeError(f无法继续执行以下任务失败: {[f.id for f in failed]}) raise RuntimeError(检测到计划中存在循环依赖) # 并行执行所有就绪的子任务 tasks [self._run_with_retry(st, max_retries) for st in ready] completed await asyncio.gather(*tasks, return_exceptionsTrue) for st, result in zip(ready, completed): if isinstance(result, Exception): st.status TaskStatus.FAILED st.error str(result) else: st.status TaskStatus.SUCCESS st.result result self._completed[st.id] st results.append(st) del remaining[st.id] return results async def _run_with_retry(self, subtask: SubTask, max_retries: int) - Any: last_error None for attempt in range(max_retries 1): try: subtask.status TaskStatus.RUNNING return await self._execute(subtask) except Exception as e: last_error e if attempt max_retries: await asyncio.sleep(0.5 * (2 ** attempt)) raise last_error or RuntimeError(未知执行错误) async def plan_and_solve( task: str, llm_call: Callable[[str], Awaitable[str]], execute_fn: Callable[[SubTask], Awaitable[Any]], ) - str: generator PlanGenerator() plan await generator.generate(task, llm_call) executor PlanExecutor(execute_fn) try: results await executor.execute(plan) except RuntimeError as e: return f任务执行失败: {e} success_count sum(1 for r in results if r.status TaskStatus.SUCCESS) return f计划完成: {success_count}/{len(plan.subtasks)} 个子任务成功执行这段代码看着长但核心就两件事生成计划、按拓扑序执行计划。PlanGenerator负责调用 LLM 拆解任务PlanExecutor负责把子任务 DAG 安全地跑完。并行执行能力来自asyncio.gather天然支持多个无依赖子任务同时进行。退避重试用了指数延时避免瞬时故障时反复冲撞。最关键的是任务失败不会让整个流程崩溃只标记状态方便后续重新规划。四、边界分析与架构权衡Plan-and-Solve 最大的代价是多了一次 LLM 调用。对于简单任务这个开销是浪费的。所以复杂度判断器必须在简单任务上直接走快速通道。判断器可以是规则比如输入长度、关键词匹配也可以是一个轻量分类模型。一个实用的经验是如果任务描述超过 50 字大概率需要 Plan-and-Solve。另一个边界是计划的质量。LLM 生成的计划可能遗漏子任务、错误判断依赖关系、甚至把不可分解的任务强行拆碎。所以计划不应盲信执行器要做依赖关系环检测和最终结果交叉验证。如果某个子任务依赖了不在计划中的 ID执行器应立即报告而非默默跳过。并行度控制也需要边界意识。理论上所有无依赖子任务都能并行但如果同时跑 20 个都去调外部 API可能触发限流。实际落地要加上并发上限比如asyncio.Semaphore控制在 3~5 个并发。还有一个容易忽略的问题子任务的执行顺序虽然由 DAG 决定但结果聚合的 LLM 调用需要按自然语义组织。建议在执行结束后把成功任务的摘要按逻辑顺序重新排版再喂给汇总模型。否则汇总模型看到的是一团乱码。五、总结Plan-and-Solve 本质上是把推理从线性链变成了有向无环图。复杂任务能显著受益于这种结构执行可并行、失败可重试、进度可审计。但成本也比 Chain-of-Thought 高了约 20~30% 的 Token 消耗所以不适合今天天气如何这种轻量任务。落地时重点做好三件事用一个轻量判断器把简单任务快速分流给执行器加上并发上限和指数退避重试子任务失败后要有优雅降级策略而不是让整个 Agent 崩溃。计划不完美没关系关键是系统能识别计划不完美并自我修复。
Agent 的任务分解策略Plan-and-Solve 比 Chain-of-Thought 更适合复杂任务一、深度引言与场景痛点如果你给 Agent 布置一个任务比如帮我分析这家公司过去三年的财报趋势和同行业对比后输出风险报告Chain-of-Thought 的方式是做一步看一步先查财报发现数据不全再补查发现需要行业数据又去查。等到推理链超过十几步Agent 自己都忘了最初想干啥。这不是模型能力不够而是推理方式不对。复杂任务的特点是步骤之间有依赖、中间信息需要聚合、路径可能分叉。Chain-of-Thought 擅长线性推理但面对需要分治、并行、回溯的任务它就像在迷宫里边走边猜——迟早出错。Plan-and-Solve 的思路就简单了先花 10% 的时间做计划把复杂任务拆成可独立验证的子任务再逐块执行。这相当于先画地图再走路比边走边问路靠谱得多。二、底层机制与原理深度剖析flowchart TD A[用户输入复杂任务] -- B[任务分析阶段] B -- C{复杂度判断} C --|简单任务| D[直接推理执行] C --|复杂任务| E[生成执行计划] E -- F[依赖关系排序] F -- G[子任务并行分配] G -- H1[子任务1: 数据采集] G -- H2[子任务2: 预处理] G -- H3[子任务3: 分析计算] H1 -- I[中间结果验证] H2 -- I H3 -- I I -- J{所有子任务通过?} J --|否| K[标记失败任务, 重新规划] K -- E J --|是| L[结果聚合与推理] L -- M[输出最终答案]图中核心的分叉在于复杂度判断这一步。不是所有任务都需要 Plan-and-Solve对于今天天气怎么样这种单步任务直接推理更快。Plan-and-Solve 的威力体现在依赖关系排序让子任务按正确顺序执行中间验证保证每一步结果可信失败重规划让系统有自愈能力。和 Chain-of-Thought 的关键区别是CoT 在推理过程中顺便分解Plan-and-Solve 是先显式生成计划再由执行器逐块执行。显式计划的好处是可审计、可打断、可重试、可并行。想象一下你做项目时是先写个甘特图再干活还是干到哪算哪前者显然更可控。三、生产级代码实现from __future__ import annotations import asyncio import json from dataclasses import dataclass, field from enum import Enum from typing import Any, Callable, Awaitable class TaskStatus(Enum): PENDING pending RUNNING running SUCCESS success FAILED failed SKIPPED skipped dataclass class SubTask: id: str description: str dependencies: list[str] field(default_factorylist) status: TaskStatus TaskStatus.PENDING result: Any None error: str dataclass class Plan: task_id: str subtasks: list[SubTask] raw_plan: str class PlanGenerator: 计划生成器 —— 调用 LLM 将复杂任务拆解为子任务 DAG async def generate(self, task: str, llm_call: Callable[[str], Awaitable[str]]) - Plan: prompt ( 你是一个任务分解专家。请将以下复杂任务拆解为可独立执行的子任务 输出 JSON 格式每个子任务包含 id、描述和依赖的 id 列表\n f任务{task} ) raw await llm_call(prompt) try: parsed json.loads(raw) except json.JSONDecodeError: # 降级将整个任务作为单个子任务 parsed [{id: task_0, description: task, dependencies: []}] subtasks [ SubTask( iditem[id], descriptionitem[description], dependenciesitem.get(dependencies, []), ) for item in parsed ] return Plan(task_idtask[:20], subtaskssubtasks, raw_planraw) class PlanExecutor: 计划执行器 —— 按拓扑排序逐个执行子任务支持并行 def __init__(self, execute_fn: Callable[[SubTask], Awaitable[Any]]): self._execute execute_fn self._completed: dict[str, SubTask] {} async def execute(self, plan: Plan, max_retries: int 2) - list[SubTask]: remaining {st.id: st for st in plan.subtasks} results: list[SubTask] [] while remaining: ready [ st for st in remaining.values() if all(dep in self._completed for dep in st.dependencies) ] if not ready: # 存在循环依赖或死锁 failed [st for st in remaining.values() if st.status TaskStatus.FAILED] if failed: raise RuntimeError(f无法继续执行以下任务失败: {[f.id for f in failed]}) raise RuntimeError(检测到计划中存在循环依赖) # 并行执行所有就绪的子任务 tasks [self._run_with_retry(st, max_retries) for st in ready] completed await asyncio.gather(*tasks, return_exceptionsTrue) for st, result in zip(ready, completed): if isinstance(result, Exception): st.status TaskStatus.FAILED st.error str(result) else: st.status TaskStatus.SUCCESS st.result result self._completed[st.id] st results.append(st) del remaining[st.id] return results async def _run_with_retry(self, subtask: SubTask, max_retries: int) - Any: last_error None for attempt in range(max_retries 1): try: subtask.status TaskStatus.RUNNING return await self._execute(subtask) except Exception as e: last_error e if attempt max_retries: await asyncio.sleep(0.5 * (2 ** attempt)) raise last_error or RuntimeError(未知执行错误) async def plan_and_solve( task: str, llm_call: Callable[[str], Awaitable[str]], execute_fn: Callable[[SubTask], Awaitable[Any]], ) - str: generator PlanGenerator() plan await generator.generate(task, llm_call) executor PlanExecutor(execute_fn) try: results await executor.execute(plan) except RuntimeError as e: return f任务执行失败: {e} success_count sum(1 for r in results if r.status TaskStatus.SUCCESS) return f计划完成: {success_count}/{len(plan.subtasks)} 个子任务成功执行这段代码看着长但核心就两件事生成计划、按拓扑序执行计划。PlanGenerator负责调用 LLM 拆解任务PlanExecutor负责把子任务 DAG 安全地跑完。并行执行能力来自asyncio.gather天然支持多个无依赖子任务同时进行。退避重试用了指数延时避免瞬时故障时反复冲撞。最关键的是任务失败不会让整个流程崩溃只标记状态方便后续重新规划。四、边界分析与架构权衡Plan-and-Solve 最大的代价是多了一次 LLM 调用。对于简单任务这个开销是浪费的。所以复杂度判断器必须在简单任务上直接走快速通道。判断器可以是规则比如输入长度、关键词匹配也可以是一个轻量分类模型。一个实用的经验是如果任务描述超过 50 字大概率需要 Plan-and-Solve。另一个边界是计划的质量。LLM 生成的计划可能遗漏子任务、错误判断依赖关系、甚至把不可分解的任务强行拆碎。所以计划不应盲信执行器要做依赖关系环检测和最终结果交叉验证。如果某个子任务依赖了不在计划中的 ID执行器应立即报告而非默默跳过。并行度控制也需要边界意识。理论上所有无依赖子任务都能并行但如果同时跑 20 个都去调外部 API可能触发限流。实际落地要加上并发上限比如asyncio.Semaphore控制在 3~5 个并发。还有一个容易忽略的问题子任务的执行顺序虽然由 DAG 决定但结果聚合的 LLM 调用需要按自然语义组织。建议在执行结束后把成功任务的摘要按逻辑顺序重新排版再喂给汇总模型。否则汇总模型看到的是一团乱码。五、总结Plan-and-Solve 本质上是把推理从线性链变成了有向无环图。复杂任务能显著受益于这种结构执行可并行、失败可重试、进度可审计。但成本也比 Chain-of-Thought 高了约 20~30% 的 Token 消耗所以不适合今天天气如何这种轻量任务。落地时重点做好三件事用一个轻量判断器把简单任务快速分流给执行器加上并发上限和指数退避重试子任务失败后要有优雅降级策略而不是让整个 Agent 崩溃。计划不完美没关系关键是系统能识别计划不完美并自我修复。
：原理、代码与优化实战)
