企业官网建设流程全解析

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_factory=list) 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( id=item["id"], description=item["description"], dependencies=item.get("dependencies", []), ) for item in parsed ] return Plan(task_id=task[:20], subtasks=subtasks, raw_plan=raw) 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_exceptions=True) 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 崩溃。计划不完美没关系，关键是系统能识别计划不完美并自我修复。