1. 引言：为什么你的 Agent 总是「卡住」

很多团队在落地 AI Agent 时都会遇到同一个尴尬场景：一个看似「全能」的 Agent，接到复杂任务后不是超时，就是在中间步骤迷失方向，最后返回一堆半成品甚至胡言乱语。

问题的根源往往不是模型不够聪明，而是我们把「一口」的任务塞给了它：让它理解需求、规划步骤、调用工具、校验结果、处理异常，甚至还要自我反思——所有这些责任都压在一个没有边界的上下文和模糊的执行流程里。

这篇文章不聊宏大的 Agent 架构哲学，只聚焦一个具体且关键的工程问题：如何把复杂任务拆分成可执行、可观测、可恢复的子任务。我们会从拆分原则讲起，落到可运行的工程实践，并给出一个可复用的任务拆分框架。

2. 任务拆分的本质：把「认知负担」变成「执行链路」

Agent 的每一次推理都消耗有限的上下文窗口和注意力。当一个任务包含多个目标、多种工具、多处条件分支时，单靠一次长上下文推理，模型很容易出现：

• 目标漂移：做着做着忘了原始需求；
• 步骤遗漏：跳过了关键的校验或回滚；
• 工具误用：在错误的时候调用错误的工具；
• 难以调试：失败时不知道是哪一步出了问题。

任务拆分，本质上是把「一个大而模糊的认知任务」转化为「一串小而明确的执行步骤」。每一个子任务都应该是：

1. 目标单一：一句话能说清要做什么；
2. 有明确输入输出：数据从哪来、产出是什么；
3. 可独立验证：完成与否有清晰的判定标准；
4. 失败可恢复：出错时能重试或降级，而不是整体崩溃。

满足这四个条件的子任务，才能被可靠地编排、观测和优化。

3. 拆分原则：先问四个问题

在动手写任何编排代码之前，先对原始任务问四个问题。

3.1 这个任务有几个「稳定的中间态」？

把任务想象成一条流水线。凡是可以「存盘」的节点，都是天然的拆分点。例如「写一份行业调研报告」可以拆成：检索资料 → 提取要点 → 生成大纲 → 分节撰写 → 汇总校对。每一步的产出都是可以独立保存和检查的中间态。

3.2 哪些步骤依赖外部工具？

凡是需要调用搜索、数据库、代码执行器、第三方 API 的步骤，都应该独立成子任务。外部调用是失败率最高的环节，独立出来才能单独重试、设置超时和降级策略。

3.3 哪些步骤需要「人的确认」？

如果任务中存在高风险决策（如发送邮件、执行数据库变更、对外发布内容），应该在这些节点前设置确认点。拆分成子任务后，可以在步骤之间插入人工审批，而不是让 Agent 一口气执行到底。

3.4 哪些步骤失败后必须回滚？

有副作用的操作（写文件、改配置、调支付接口）需要和纯计算操作分开。纯计算失败可以重试，有副作用的操作失败则可能需要补偿逻辑。混在一起只会让错误处理变成一锅粥。

4. 三种实用的拆分模式

根据任务形态，有三种最常用的拆分模式。

4.1 管道模式（Pipeline）

适用场景：步骤之间有明确的先后依赖，后一步必须等前一步完成。

输入 → 步骤A → 步骤B → 步骤C → 输出

典型例子：代码生成任务可以拆成「理解需求 → 生成代码 → 运行测试 → 修复报错」。每一步的产出是下一步的输入，链路清晰。

4.2 计划-执行模式（Plan-Execute）

适用场景：任务一开始无法完全确定后续步骤，需要先规划再逐步执行。

用户需求 → 规划器生成步骤清单 → 逐步执行 → 根据执行结果动态调整计划

这是目前工程实践中最常用的一种。规划器先输出一个任务列表，执行器逐个完成，每完成一步都可以根据反馈更新剩余计划。

4.3 分而治之模式（Divide-and-Conquer）

适用场景：任务可以被切分成多个相互独立的子任务，适合并行处理。

大任务 → 切分成 N 个独立子任务 → 并行/串行执行 → 合并结果

典型例子：同时分析多个数据源、同时审查多个文件。子任务之间没有依赖，可以并发执行，最后用一个汇总步骤统一结果。

5. 工程实践：一个可运行的任务拆分框架

下面用一个精简的 Python 实现，展示如何在工程中落地「计划-执行」模式。这个框架不依赖任何特定的大模型 SDK，核心是把任务规划、执行、验证三个环节解耦。

5.1 定义子任务的数据结构

from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Callable, Optional


class TaskStatus(Enum):
    PENDING = "pending"
    RUNNING = "running"
    SUCCESS = "success"
    FAILED = "failed"
    SKIPPED = "skipped"


@dataclass
class SubTask:
    """一个可执行、可观测、可恢复的子任务。"""

    id: str
    description: str
    # 执行函数：接收上下文，返回结果
    execute: Callable[[dict[str, Any]], Any]
    # 验证函数：判断执行结果是否满足要求
    validate: Optional[Callable[[Any], bool]] = None
    # 失败后最多重试次数
    max_retries: int = 1
    # 该任务是否需要人工确认后才执行
    requires_approval: bool = False
    # 前置任务 id 列表：都成功后才执行本任务
    depends_on: list[str] = field(default_factory=list)
    status: TaskStatus = TaskStatus.PENDING
    result: Any = None
    error: Optional[str] = None

5.2 实现任务编排器

编排器负责按依赖关系调度任务、处理重试与验证，并在需要时暂停等待人工确认。

class TaskOrchestrator:
    """按依赖关系执行子任务的编排器。"""

    def __init__(
        self,
        tasks: list[SubTask],
        approve: Callable[[SubTask], bool] | None = None,
    ):
        self.tasks = {t.id: t for t in tasks}
        self.approve = approve or (lambda _: True)
        self.context: dict[str, Any] = {}

    def run(self) -> dict[str, Any]:
        pending = list(self.tasks.values())

        while pending:
            ready = [
                t for t in pending
                if all(self.tasks[d].status == TaskStatus.SUCCESS for d in t.depends_on)
                and t.status == TaskStatus.PENDING
            ]
            if not ready:
                # 没有可执行任务：要么全部完成，要么存在死锁
                unfinished = [t for t in pending if t.status == TaskStatus.PENDING]
                if unfinished:
                    raise RuntimeError(
                        f"存在无法调度的任务（疑似循环依赖）："
                        f"{[t.id for t in unfinished]}"
                    )
                break

            for task in ready:
                if task.requires_approval and not self.approve(task):
                    task.status = TaskStatus.SKIPPED
                    continue

                self._execute_with_retry(task)

            # 移除已完成任务，进入下一轮
            pending = [
                t for t in pending
                if t.status in (TaskStatus.PENDING, TaskStatus.RUNNING)
            ]

        return self.context

    def _execute_with_retry(self, task: SubTask) -> None:
        task.status = TaskStatus.RUNNING
        for attempt in range(task.max_retries + 1):
            try:
                task.result = task.execute(self.context)
                self.context[task.id] = task.result

                if task.validate and not task.validate(task.result):
                    raise ValueError(f"验证未通过：{task.id}")

                task.status = TaskStatus.SUCCESS
                task.error = None
                print(f"[OK]   {task.id}: {task.description}")
                return
            except Exception as exc:  # noqa: BLE001
                task.error = str(exc)
                if attempt < task.max_retries:
                    print(f"[RETRY] {task.id} 第 {attempt + 1} 次重试：{exc}")
                else:
                    task.status = TaskStatus.FAILED
                    print(f"[FAIL] {task.id}: {exc}")
                    raise

5.3 组装一个真实案例

假设我们要构建一个「分析用户反馈并生成周报」的 Agent。原始需求如果直接丢给模型，很容易遗漏环节。拆成子任务后，每一步都清晰可控。

def fetch_feedback(ctx: dict[str, Any]) -> list[str]:
    """模拟从数据库/API 拉取本周用户反馈。"""
    return [
        "登录太慢了，每次要等十秒",
        "导出报表的按钮找不到",
        "新版本的暗色模式很舒服",
        "移动端列表页会闪退",
    ]


def summarize_feedback(ctx: dict[str, Any]) -> dict[str, int]:
    """模拟对反馈做分类统计（实际场景可调用大模型）。"""
    feedback = ctx["fetch"]
    stats = {"功能问题": 0, "体验好评": 0, "性能问题": 0, "崩溃问题": 0}
    keywords = {
        "功能问题": ["找不到", "没有"],
        "体验好评": ["很舒服", "好用"],
        "性能问题": ["慢", "卡"],
        "崩溃问题": ["闪退", "崩溃"],
    }
    for item in feedback:
        for category, words in keywords.items():
            if any(w in item for w in words):
                stats[category] += 1
    return stats


def draft_report(ctx: dict[str, Any]) -> str:
    """根据统计结果生成周报草稿。"""
    stats = ctx["summarize"]
    total = sum(stats.values())
    return (
        "本周用户反馈周报：\n"
        f"共收集反馈 {total} 条，其中 \n"
        + "\n".join(f"- {k}：{v} 条" for k, v in stats.items())
    )


def validate_report(text: str) -> bool:
    """校验报告草稿是否符合格式要求。"""
    return bool(text) and "本周用户反馈周报" in text


if __name__ == "__main__":
    tasks = [
        SubTask(
            id="fetch",
            description="拉取本周用户反馈",
            execute=fetch_feedback,
            validate=lambda r: isinstance(r, list) and len(r) > 0,
        ),
        SubTask(
            id="summarize",
            description="对反馈进行分类统计",
            execute=summarize_feedback,
            depends_on=["fetch"],
            validate=lambda r: isinstance(r, dict) and sum(r.values()) > 0,
        ),
        SubTask(
            id="draft",
            description="生成周报草稿",
            execute=draft_report,
            depends_on=["summarize"],
            validate=validate_report,
            max_retries=2,
        ),
    ]

    orchestrator = TaskOrchestrator(tasks)
    result = orchestrator.run()
    print("\n" + result["draft"])

运行结果会清晰展示每个子任务的执行状态，任何一步失败都能定位到具体环节，而不是面对一个「整体失败」的黑盒。

5.4 把大模型接入子任务

实际场景中，子任务的 execute 函数内部会调用大模型。关键点是：给每个子任务的提示词只包含「这一步」所需的上下文，而不是把整个任务历史和全部资料都塞进去。

def llm_extract_insights(ctx: dict[str, Any]) -> str:
    """只接收上一步的统计结果，生成洞察。"""
    stats = ctx["summarize"]
    prompt = (
        "根据以下分类统计，用一句话概括本周最值得关注的问题：\n"
        f"{stats}\n"
        "只输出一句话结论。"
    )
    # 实际调用：
    # return call_llm(prompt)
    return "用户集中反馈移动端崩溃问题，建议优先排查。"

把上下文收窄到子任务所需的最小集合，能显著降低模型的推理负担，减少无关信息干扰，这也是任务拆分在工程上最直接的收益之一。

6. 常见的拆分误区

6.1 拆得太碎

每个子任务都应该是一个有意义的「工作单元」。如果把「分析需求」拆成「读第一个字、读第二个字」，只会增加编排开销，反而降低效率。拆分的粒度以「一个子任务恰好能被一次可靠执行完成」为宜。

6.2 忽略依赖关系

任务之间除了显式的前后依赖，还有隐式的数据依赖。例如步骤 C 需要步骤 A 和步骤 B 的产出合并后才能执行。在 depends_on 中遗漏任何一环，都会导致执行结果错误甚至死锁。

6.3 把验证逻辑混在执行逻辑里

执行和验证应该是分离的。执行负责「做完」，验证负责「做对」。很多 Agent 失败的原因，就是把「做对」的责任又推回给了模型的一次生成，缺少独立的校验环节。

6.4 没有为失败设计

好的拆分不仅是「成功路径」的拆分，更是「失败路径」的拆分。每个子任务都应该回答一个问题：如果这一步失败了，是重试、跳过、降级，还是终止整个任务并通知用户？ 提前定义好失败策略，才能让 Agent 在面对真实世界的异常时不至于全线崩溃。

7. 落地检查清单

在把任务拆分方案上线前，可以用下面这份清单快速自查：

• 每个子任务的目标是否能用一句话描述清楚？
• 每个子任务的输入输出是否明确定义？
• 依赖关系是否完整，是否存在循环依赖？
• 外部工具调用是否独立成任务并配置了重试与超时？
• 有副作用的操作是否设置了人工确认或回滚机制？
• 每个子任务是否有独立的验证逻辑？
• 失败时的处理策略是否明确（重试/跳过/降级/终止）？
• 是否记录了每个子任务的执行状态和耗时，便于观测调优？

8. 总结

「别让你的 Agent 一口吃成胖子」，本质上是在提醒我们：Agent 的能力边界，往往由任务拆分的质量决定。模型再强，也架不住一个没有结构、没有边界、没有失败策略的混沌任务。

好的任务拆分，是在为 Agent 搭建一条清晰的执行链路：

1. 从「稳定中间态」和「外部依赖」中找拆分点；
2. 用管道、计划-执行、分而治之三种模式组织子任务；
3. 把执行、验证、重试、人工确认解耦到独立的环节；
4. 为每一步定义清晰的成功标准和失败策略。

当你的 Agent 不再「一口吞下」整个任务，而是按部就班地完成一个个小目标时，它的可靠性、可观测性和可维护性，都会上一个台阶。而这一切的起点，只是停下来问一句：这一步，真的不能再拆了吗？