Agent 任务拆分策略大任务拆小了调度开销不能比计算大一、当 Agent 花在想怎么干上的时间比干还多给 Agent 提了一个需求分析这份 200 页的 PDF 并生成摘要和关键图表。Agent 思考了 30 秒生成了一个 12 步的计划然后用 45 秒执行。但总耗时 75 秒里有 30 秒花在任务规划上。如果拆得更细——分析第 1 页生成段落摘要分析第 2 页……——单页耗时可能降到 2 秒。但 200 页就有 200 次独立调度调度开销加起来可能超过 60 秒。这是一个经典的最优粒度问题任务拆分太粗单步处理压力大、容易出错拆分太细调度开销吃掉效率。找到那个让总耗时最短的分割点就是这篇文章要解决的问题。二、任务拆分作为优化问题调度成本不能忽略把 Agent 的任务执行建模为T_total N * (T_exec(粒度) T_overhead)其中 N 是子任务数量T_exec 是每个子任务的 LLM 推理时间T_overhead 是每次调度的固定开销上下文切换、工具调用、结果汇总。T_exec 随粒度变粗而增长近似线性T_overhead 是一个固定值。总成本曲线是一个 U 型左侧细粒度 → 调度开销吃掉一切右侧粗粒度 → 单步超时或出错。graph TD A[接收复杂任务 Q] -- B{任务特征分析} B -- C[计算任务复杂度 score] C -- D{D 阈值?} D --|是| E[直接执行不拆分] D --|否| F[DAG 分解器] F -- G[生成任务依赖图] G -- H{存在并行分支?} H --|是| I[标记并行组] H --|否| J[串行执行序] I -- K[分配 Worker] J -- K K -- L[运行并监控] L -- M{任何子任务失败?} M --|是| N[重试或降级] M --|否| O[结果合并] N -- L O -- P[输出最终结果]见证奇迹的时刻当拆分粒度从一页一页来调整到每 5 页一批总耗时从 120 秒降到了 48 秒——不是因为 LLM 变快了是因为调度次数从 200 降到了 40。四、边界分析与架构权衡Agent 任务拆分的核心权衡在于三个维度维度粗粒度细粒度单步成功率低太复杂高简单明确调度开销低次数少高次数多并行度低依赖多高独立性强错误恢复成本高重做整个大任务低只重做一个小任务跨任务一致性好上下文连贯差需要额外对齐一个被称为拆分悖论的现象在实践中反复出现你越是详细地规划子任务的输入输出格式Agent 越有可能在某个中间步骤偏离规范导致下游任务接收到格式不兼容的输入。这个问题在超过 10 个子任务时变得尤为明显。解决方案是引入检查点机制在关键分叉节点插入一个验证步骤它不产生业务输出只做一件事——检查上游输出的格式和内容是否满足下游预期。这种检查点虽然增加了总任务数但带来的可靠性提升远超其开销。在内部评测中增加 2 个检查点将复杂任务的端到端成功率从 73% 提升到了 91%。三、动态容量拆分让 Agent 自己决定粒度以下是一个基于 Token 容量感知的任务拆分框架。它的核心思想是不让人类预设拆分粒度而是让 Agent 根据自身上下文窗口的可用容量动态计算。from typing import List, Dict, Optional from dataclasses import dataclass, field from enum import Enum class SplitStrategy(Enum): 拆分策略枚举 FIXED_SIZE fixed_size # 固定大小拆分 SEMANTIC semantic # 语义边界拆分 CAPACITY_AWARE capacity_aware # 容量感知拆分 dataclass class SubTask: 子任务数据结构 id: str description: str estimated_tokens: int dependencies: List[str] field(default_factorylist) # 设计原因每个子任务自带令牌预算估计 # 拆分器根据预算判断是否能装进上下文窗口 input_schema: Optional[Dict] None output_schema: Optional[Dict] None class CapacityAwareSplitter: 容量感知任务拆分器 —— 核心思想上下文窗口 硬约束 def __init__( self, max_context_tokens: int 128000, # 上下文窗口大小 safety_margin: float 0.15, # 安全边界 15% overhead_per_task: int 500, # 每个子任务的提示词开销 ): self.max_tokens max_context_tokens self.safety_margin safety_margin self.overhead_per_task overhead_per_task # 有效容量 窗口大小 × (1 - 冗余) - 系统提示词 self.effective_capacity int(max_context_tokens * (1 - safety_margin)) def estimate_chunk_tokens(self, text: str) - int: 估算文本的 Token 数 设计原因使用字符数/3.5 近似而非精确 Tokenize 因为真实 Tokenize 依赖具体模型而这里只需要粒度估计。 中文场景下平均每 Token 约 1.5 个字符取 2 作为保守估计。 # 保守估计中文 2 字符/Token英文 4 字符/Token return len(text) // 2 def split_by_capacity( self, content: str, instruction: str, strategy: SplitStrategy SplitStrategy.CAPACITY_AWARE, chunk_size: int 2000, ) - List[SubTask]: 根据容量自动拆分任务 设计原因不同 LLM 的上下文窗口不同4K~128K 硬编码拆分粒度会导致跨模型迁移困难。 容量感知拆分让同一套逻辑适配不同模型。 total_tokens self.estimate_chunk_tokens(content) instruction_tokens self.estimate_chunk_tokens(instruction) # 计算每个子任务可用的有效容量 available_per_task self.effective_capacity - instruction_tokens - self.overhead_per_task if strategy SplitStrategy.CAPACITY_AWARE: # 容量感知根据内容总 Token 和有效容量计算最优拆分份数 n_tasks max(1, total_tokens // available_per_task 1) chunk_token_size total_tokens // n_tasks elif strategy SplitStrategy.FIXED_SIZE: # 固定大小用户指定的粒度 n_tasks max(1, total_tokens // self.estimate_chunk_tokens(str(chunk_size))) chunk_token_size total_tokens // n_tasks else: n_tasks 1 chunk_token_size total_tokens # 设计原因限制最大子任务数不超过 20 # 经验表明超过 20 个子任务调度开销开始显著超过计算收益 n_tasks min(n_tasks, 20) subtasks [] for i in range(n_tasks): start_chars i * (len(content) // n_tasks) end_chars (i 1) * (len(content) // n_tasks) if i n_tasks - 1 else len(content) chunk content[start_chars:end_chars] # 子任务依赖顺序依赖后一个依赖前一个的汇总结果 deps [ftask_{i-1}] if i 0 else [] subtask SubTask( idftask_{i}, descriptionf处理第 {i1}/{n_tasks} 部分内容{chunk[:50]}..., estimated_tokensself.estimate_chunk_tokens(chunk), dependenciesdeps, input_schema{chunk_index: i, total_chunks: n_tasks}, output_schema{summary: string, key_points: list}, ) subtasks.append(subtask) return subtasks def should_split(self, task_complexity: float, content_length: int) - bool: 判断是否应该拆分 Args: task_complexity: 0~1 的任务复杂度评分 content_length: 内容长度字符数 设计原因两张判断表 —— 复杂度高了一定拆内容长了一定拆 只有两者都低的时候才不拆 content_tokens self.estimate_chunk_tokens(str(content_length)) # 内容超过有效容量的 70% → 必须拆 if content_tokens self.effective_capacity * 0.7: return True # 复杂度 0.6 → 建议拆分以降低单步难度 if task_complexity 0.6: return True return False五、总结Agent 任务拆分是一个优化问题而不是一个设计模式。核心结论调度开销不是虚拟成本在 20 子任务时可以超过总耗时的 40%容量感知拆分比固定粒度更鲁棒能自适应不同上下文窗口任务依赖图DAG是并行调度的前提没有 DAG 就只能串行子任务数控制在 20 以内是调度开销可控的经验临界点检查点子任务不产出业务结果是提升链路可靠性的低成本方案在最终架构上推荐将是否拆分和如何拆分做两层决策第一层判断任务复杂度是否需要拆第二层根据可用容量计算最优粒度。两层分离的好处是你可以独立调优复杂度评分器和容量估算器。