之前在业务迭代中做了一个调研类 Agent需求很简单用户输入一个行业关键词Agent 自动生成一份市场调研报告。第一次联调时它在 20 个测试用例里失败了 15 个要么漏掉了关键维度要么分析到一半就停止输出更常见的是把三个步骤混在一起执行最后产出一篇什么都像、什么都不像的“缝合文”。后来我意识到问题不在模型而在任务描述方式。把一个大而模糊的目标直接丢给大模型等同于让实习生第一次接触项目就独立负责完整交付。真正稳定的 Agent几乎都不是靠“模型聪明”而是靠“任务拆得足够细”。本文围绕大模型任务拆解展开先分析 Agent 翻车的本质原因再讲拆解方法、执行模式、完整代码案例最后给出可落地的工程建议。适合正在做 Agent 开发、或者被“提示词越写越长但效果越来越差”折磨的开发者。1. 背景与核心概念Agent 为什么总在翻车1.1 一个典型的翻车现场假设你给 Agent 下达了这样一个指令帮我写一份新能源汽车市场调研报告。如果不做任何拆解大模型通常会这样做直接调用联网搜索工具抓取几篇新闻然后拼凑出一篇报告。表面上看没问题但仔细检查会发现报告缺少市场规模数据只有趋势描述。没有区分国内市场和海外市场。引用了过时的数据没有标注时间。最后没有给出结论只是罗列信息。这不是模型能力不够而是目标本身太模糊。“写一份市场调研报告”这句话里面隐藏着大量隐含步骤明确报告范围、确定时间周期、收集市场规模数据、分析竞争格局、梳理上下游产业链、提炼结论。任何一个步骤缺失最终产出都会“翻车”。1.2 翻车的本质是什么从技术角度看大模型是概率模型它根据输入的 token 预测下一个 token。当指令过长、约束过多、隐含步骤过多时模型很难在生成过程中始终保持全局一致性。可以从两个维度理解上下文窗口压力一个复杂任务如果全部塞进一个 prompt会导致关键指令被淹没。模型生成到后半段时很可能已经忘记了开头的要求。错误累积效应如果任务需要多步执行而每一步都存在一定的错误率那么整体成功率会指数级下降。假设单步成功率为 90%五步之后整体成功率只有 59%。所以不稳定不是模型的“锅”而是任务结构设计的问题。我们需要把一件事拆成多件小事让模型每次只聚焦于一个小目标这才是提高稳定性的关键。1.3 为什么任务拆解能解决这个问题任务拆解不会让模型变聪明但它能让模型更专注、更可控、更容易被验证。拆解之后每一个子任务拥有独立的提示词、独立的输入输出、独立的校验逻辑。一旦某个子任务失败我们可以精准定位并重试而不必让整个任务重新开始。这也是 Agent 工程化落地的基本功。2. 什么是大模型任务拆解2.1 任务拆解的定义任务拆解Task Decomposition是指将一个复杂目标拆分为多个相互独立或存在依赖关系的子任务每个子任务有明确的输入、输出和验收标准。举个例子目标生成一份新能源汽车市场调研报告 拆解后 1. 解析行业关键词确定报告范围和维度 2. 搜索市场规模数据 3. 收集主要玩家与竞争格局 4. 汇总产业链上下游信息 5. 分析数据并提炼核心结论 6. 按照模板生成 Markdown 报告这样拆解之后每个子任务都像一个独立的函数有输入、有输出、有校验规则。Agent 的执行过程也从“一步生成大段内容”变成了“逐步执行小任务”。2.2 拆解后为什么更稳定稳定性提升来源于三方面第一上下文更聚焦。单个子任务的 prompt 更短模型不会被无关信息干扰生成质量明显提升。第二可校验。子任务的输出可以被程序检查比如“是否包含市场规模数据”“是否超过 200 字”不满足条件可以直接重试而不是等到最终结果才发现问题。第三可重试。传统的一体化生成一旦中间出错就要全部重来。拆解之后只有失败的那个子任务需要重跑整体成本和耗时都会下降。2.3 拆解的三种粒度任务拆解不是越细越好粒度过细会导致调用次数暴涨粒度过粗又起不到稳定效果。理解三种粒度有助于拿捏分寸粒度说明示例适合场景元任务最上层的目标描述生成市场调研报告用户输入的原始需求复合任务需要多个步骤完成的子目标收集并整理市场规模数据需要联网、解析、清洗的中间环节原子任务不可再拆的单一操作调用搜索 API 并返回前 10 条结果单个工具调用、单次模型生成在实际开发中我习惯把任务拆到“原子任务”这一层但前提是每个原子任务都能清晰地定义输入输出。如果某个原子任务本身还需要大模型“想半天”说明拆得还不够细。3. 拆解前的准备明确 Agent 的能力边界3.1 先盘点工具和能力在开始拆解任务之前必须清楚 Agent 手上有什么工具。工具决定了子任务的下限没有检索工具就不应该设计“搜索资料”这一步没有代码解释器就不应该让模型执行复杂计算。常见的工具能力包括联网搜索获取实时信息。数据库查询读取结构化业务数据。文件读写保存中间结果、生成最终文档。代码执行运行 Python 或其他语言代码。通知服务通过邮件、IM 等推送结果。拆解任务时每个子任务最好都能对应到某一个具体工具或一段确定性的代码逻辑。如果某个子任务没有任何工具支撑那么它就只是一个“让模型瞎猜”的步骤应尽量避免。3.2 判断标准哪些步骤交给大模型哪些交给代码把步骤交给模型还是代码是一个关键决策。简单来说判断标准是这个步骤是否需要语义理解。需要理解模糊指令、做总结、写文案、判断意图的交给模型需要精确计算、固定格式输出、循环遍历、条件判断的交给代码。操作类型建议执行者原因提取用户意图大模型需要语义理解调用 API 获取数据代码确定性操作清洗数据并去重代码规则明确总结数据趋势大模型需要归纳分析按照模板拼装文档代码格式固定校验输出格式代码可通过正则或 schema 检查这里有个常见的误区很多人为了让 Agent 显得“智能”把所有步骤都交给模型。实际上对于确定性逻辑代码比模型可靠一千倍。一个稳定的 Agent应该是“代码负责流程模型负责理解”。3.3 别把所有判断都交给模型曾经调过这样一个 Agent它需要根据用户输入判断走哪个业务分支。起初只用 prompt 控制结果模型经常选错分支。后来把分支判断逻辑改成了代码用关键词匹配和规则引擎准确率立刻提升到接近 100%。这不是说模型不好而是说模型不擅长“严格的条件判断”。在拆解任务时尽量把分支判断、参数校验、格式检查这类工作交给代码完成。大模型只做它最擅长的理解、总结、生成、规划。4. 拆解执行的三种模式任务拆解不只是把任务“写出来”还要考虑“怎么执行”。根据业务场景的不同有三种常见的执行模式。4.1 固定流程模式Workflow固定流程模式是最容易落地的一种。提前定义好子任务的执行顺序和依赖关系Agent 严格按照流程执行。任务关系是一个有向无环图DAG不会有意外分支。任务A - 任务B - 任务C - 任务D这种方式适合业务流程相对固定的场景比如“读取用户上传文件 - 解析内容 - 调用模型总结 - 生成导出文件”。它的优势是稳定性极高、容易排查问题劣势是不够灵活无法应对开放式需求。4.2 动态规划模式Plan-and-Execute动态规划模式让模型先针对目标任务生成一份执行计划然后 Agent 按照计划逐步执行。每一步完成后可能还会根据结果动态调整后续计划。这种模式适合开放式场景比如“帮我策划一场线下活动”模型先拆解出场地、预算、流程、物料、宣传等模块再逐个执行。与固定流程相比动态规划模式更灵活但风险也更大因为模型生成的计划可能不合理。因此需要在执行阶段增加校验和纠偏机制。4.3 反思修正模式ReAct / Self-Refine反思修正模式是在动态规划基础上增加了“反思”环节。模型执行完一个子任务后会先观察输出结果判断是否满足预期如果不满足就修正计划并重新执行。一个典型的循环是思考Thought - 执行Action - 观察Observation - 再思考Thought - ...这种模式非常适合需要多步探索的任务比如“根据用户问题定位代码 Bug”。它的优点是自愈能力强缺点是 token 消耗和耗时都会增加。实际项目中可以给反思设定上限避免无限循环。5. 完整实战案例一个调研报告的稳定交付这一节我们来实现一个完整案例。目标是从“用户输入行业关键词”到“生成调研报告”整个过程通过任务拆解来实现稳定交付。5.1 需求描述与初步拆解用户输入新能源汽车我们先将目标拆解为以下子任务任务ID任务名称输入输出校验规则T1解析需求用户原始输入行业关键词、调研维度列表关键词非空T2收集市场规模数据行业关键词市场规模数据列表数据条目 0T3收集竞争格局数据行业关键词主要企业列表企业数量 0T4生成调研结论市场规模 竞争数据核心结论文本文本长度 50T5生成 Markdown 报告全部数据 结论Markdown 报告报告包含标题、正文、结论5.2 项目结构设计market_research_agent/ ├── main.py # 主流程入口 ├── planner.py # 任务规划与提示词 ├── executor.py # 子任务执行引擎 ├── tools.py # 工具函数集 ├── validator.py # 校验规则 └── context.py # 上下文管理器5.3 工具函数设计tools.py中定义 Agent 能够调用的工具。这里使用模拟数据代替真实搜索 API方便读者直接运行验证。# 文件路径market_research_agent/tools.py 工具函数集所有工具只做确定性操作不做语义理解。 def search_market_data(keyword: str) - list: 模拟搜索市场规模数据。 # 真实项目中可以替换为搜索 API 或数据库查询 mock_data [ {year: 2022, market_size: 6000, unit: 亿元, growth_rate: 35.2%}, {year: 2023, market_size: 8200, unit: 亿元, growth_rate: 36.7%}, {year: 2024, market_size: 10500, unit: 亿元, growth_rate: 28.0%}, ] return mock_data def search_competitors(keyword: str) - list: 模拟搜索主要企业。 mocks { 新能源汽车: [比亚迪, 特斯拉, 蔚来, 小鹏, 理想], 人工智能: [OpenAI, Google DeepMind, Anthropic, 百度, 阿里], } return mocks.get(keyword, []) def build_report(core_data: dict) - str: 根据结构化数据生成 Markdown 报告。 title f# {core_data[keyword]} 市场调研报告\n\n sections [] sections.append(## 市场规模\n) for item in core_data[market_data]: row f- {item[year]} 年市场规模{item[market_size]} {item[unit]}同比增长 {item[growth_rate]}\n sections.append(row) sections.append(\n## 竞争格局\n) for name in core_data[competitors]: sections.append(f- {name}\n) sections.append(\n## 核心结论\n) sections.append(core_data[conclusion]) return title .join(sections)这里需要说明一个原则工具函数不依赖大模型它的输入输出都是确定性的。这样即使模型在某一步出错工具本身也不会成为不稳定因素。5.4 上下文管理器context.py负责保存每一步的输出供后续任务读取。# 文件路径market_research_agent/context.py 上下文管理器保存子任务输出支持按 key 读取。 class Context: def __init__(self): self._data {} self._errors [] def set(self, key: str, value): self._data[key] value def get(self, key: str): return self._data.get(key) def record_error(self, task_id: str, error: str): self._errors.append({task_id: task_id, error: error}) def get_errors(self): return self._errors def snapshot(self) - dict: 返回当前的上下文快照供外部查看。 return { data: self._data, errors: self._errors, }5.5 校验规则validator.py中定义每个子任务的校验逻辑。子任务执行完之后校验器决定输出是否合格。# 文件路径market_research_agent/validator.py 校验规则每个子任务对应一个校验函数。 def validate_non_empty(value) - bool: return bool(value) def validate_length(value, min_length: int) - bool: return isinstance(value, str) and len(value) min_length def validate_list(value, min_items: int) - bool: return isinstance(value, list) and len(value) min_items def validate_report(report: str) - bool: return all(part in report for part in [# , ## 市场规模, ## 核心结论])5.6 子任务定义与执行引擎executor.py是核心执行引擎。它负责遍历子任务列表按顺序执行并在失败时自动重试。# 文件路径market_research_agent/executor.py 子任务执行引擎。 from dataclasses import dataclass, field from typing import Callable, Any dataclass class SubTask: task_id: str name: str execute: Callable[[Context], Any] output_key: str validator: Callable[[Any], bool] None max_retries: int 3 class Executor: def __init__(self, context: Context): self.context context self.execution_log [] def run(self, subtasks: list[SubTask]): for subtask in subtasks: self._execute_with_retry(subtask) def _execute_with_retry(self, subtask: SubTask): for attempt in range(subtask.max_retries): try: result subtask.execute(self.context) if subtask.validator and not subtask.validator(result): raise ValueError(f校验未通过{subtask.task_id}) self.context.set(subtask.output_key, result) self.execution_log.append({ task_id: subtask.task_id, status: success, attempt: attempt 1, }) print(f[SUCCESS] {subtask.task_id} - {subtask.name}) return except Exception as e: self.context.record_error(subtask.task_id, str(e)) print(f[RETRY] {subtask.task_id} - 第 {attempt 1} 次失败{e}) # 重试耗尽后抛出异常交给上层处理 raise RuntimeError(f子任务 {subtask.task_id} 执行失败重试次数已用完)这个执行引擎非常简单它体现了任务拆解的核心思想子任务之间通过 Context 传递数据每个子任务只负责一件事校验失败就重试。5.7 主流程把拆解和工具串起来main.py是最终入口。它定义了完整的子任务列表并组装执行引擎。# 文件路径market_research_agent/main.py 主流程组装子任务并执行。 from context import Context from executor import Executor, SubTask from tools import search_market_data, search_competitors, build_report from validator import ( validate_non_empty, validate_list, validate_length, validate_report, ) def task_parse_requirement(ctx: Context): T1解析用户输入提取关键词和调研维度。 # 真实项目中这里可以调用大模型 raw_input ctx.get(user_input) keyword raw_input.strip() dimensions [市场规模, 竞争格局, 产业链, 发展趋势] return {keyword: keyword, dimensions: dimensions} def task_collect_market_data(ctx: Context): T2收集市场规模数据。 keyword ctx.get(requirement)[keyword] return search_market_data(keyword) def task_collect_competitors(ctx: Context): T3收集竞争格局数据。 keyword ctx.get(requirement)[keyword] return search_competitors(keyword) def task_generate_conclusion(ctx: Context): T4生成核心结论。 # 真实项目这里应该调用大模型进行语义总结 market_data ctx.get(market_data) latest market_data[-1] conclusion ( f根据最新数据{latest[year]} 年该行业市场规模达到 f{latest[market_size]} {latest[unit]}同比增长 {latest[growth_rate]}。 行业整体保持较快增长头部企业竞争力持续增强。 ) return conclusion def task_build_report(ctx: Context): T5生成 Markdown 报告。 core_data { keyword: ctx.get(requirement)[keyword], market_data: ctx.get(market_data), competitors: ctx.get(competitors), conclusion: ctx.get(conclusion), } return build_report(core_data) def main(): context Context() context.set(user_input, 新能源汽车) subtasks [ SubTask( task_idT1, name解析需求, executetask_parse_requirement, output_keyrequirement, validatorlambda r: validate_non_empty(r.get(keyword)), ), SubTask( task_idT2, name收集市场规模数据, executetask_collect_market_data, output_keymarket_data, validatorlambda d: validate_list(d, min_items1), ), SubTask( task_idT3, name收集竞争格局数据, executetask_collect_competitors, output_keycompetitors, validatorlambda d: validate_list(d, min_items1), ), SubTask( task_idT4, name生成核心结论, executetask_generate_conclusion, output_keyconclusion, validatorlambda c: validate_length(c, min_length50), ), SubTask( task_idT5, name生成 Markdown 报告, executetask_build_report, output_keyreport, validatorvalidate_report, ), ] executor Executor(context) executor.run(subtasks) print(\n 最终报告 ) print(context.get(report)) if __name__ __main__: main()5.8 运行与验证在终端执行cd market_research_agent python main.py预期输出结果大致如下[SUCCESS] T1 - 解析需求 [SUCCESS] T2 - 收集市场规模数据 [SUCCESS] T3 - 收集竞争格局数据 [SUCCESS] T4 - 生成核心结论 [SUCCESS] T5 - 生成 Markdown 报告 最终报告 # 新能源汽车 市场调研报告 ## 市场规模 - 2022 年市场规模6000 亿元同比增长 35.2% - 2023 年市场规模8200 亿元同比增长 36.7% - 2024 年市场规模10500 亿元同比增长 28.0% ## 竞争格局 - 比亚迪 - 特斯拉 - 蔚来 - 小鹏 - 理想 ## 核心结论 根据最新数据2024 年该行业市场规模达到 10500 亿元同比增长 28.0%。行业整体保持较快增长头部企业竞争力持续增强。这个过程看起来简单但它体现了任务拆解最重要的收益每个步骤都可以验证、都可以重试、都可以单独替换。如果某天市场规模数据接口换了只需要修改 T2 这一个函数如果结论生成效果不好只需要优化 T4 这一个环节。6. 防止拆解失控的工程手段任务拆解本身也会带来新问题拆得太碎、递归太深、某个子任务卡住、错误信息丢失。这里整理几个工程上常用的控制手段。6.1 限制子任务数量和递归深度无论动态规划还是固定流程都要设置上限。比如单个任务最多拆分为 8 个子任务。动态规划最多允许 3 层嵌套。整个任务链最多执行 20 步。超过限制直接停止并在日志中记录原因。这能有效防止 Agent 陷入“拆解 - 执行 - 再拆解”的死循环。6.2 增加单步超时与重试上限大模型 API 调用可能因为网络波动、服务端负载等原因变得很慢。应该为每个子任务设置超时时间并在超时或失败时采用指数退避策略重试。代码层面可以这样处理import time def execute_with_timeout(func, timeout30): start time.time() result func() elapsed time.time() - start if elapsed timeout: raise TimeoutError(f执行超时{elapsed:.2f}s) return result注意这个示例是“思路演示”实际工程中建议使用concurrent.futures或asyncio.wait_for来实现真正的超时控制。6.3 结果校验不能只靠“存在”校验规则要尽可能严谨。不要只校验“结果非空”还要校验格式、数据类型、字段完整性。比如生成报告后可以校验def validate_report_strict(report: str) - bool: required_sections [# , ## 市场规模, ## 竞争格局, ## 核心结论] if not all(section in report for section in required_sections): return False # 检查是否包含数字或关键命名实体 if not any(char.isdigit() for char in report): return False return True校验越严格最终交付质量越有保障。6.4 日志与链路追踪当 Agent 进入生产环境后日志就是排查问题的唯一线索。建议至少记录以下信息每个子任务的开始时间、结束时间、耗时。每个子任务的输入摘要和输出摘要。每次重试的原因和次数。上下文快照的关键字段。如果使用 LangChain 之类的框架它自带一定的日志能力如果是自研引擎建议从一开始就把日志结构设计好后续接入监控会省很多力气。6.5 记忆与上下文管理有些 Agent 在长时间运行时会超出上下文窗口的限制。为了减轻这个问题应该做到子任务之间只传递必要的数据而不是把全部历史塞给模型。旧的非关键信息可以存入外部存储或向量数据库需要时再检索。给上下文设置一个“最大长度”超长时对历史进行摘要压缩。7. 常见问题与排查思路实际开发中以下问题出现频率最高整理成排查清单方便查阅。问题现象常见原因解决思路Agent 执行到一半停止某个子任务抛异常且重试耗尽查看执行日志定位失败子任务检查该任务的输入数据和工具调用子任务之间数据对不上上游输出格式与下游预期不一致定义严格的输出 schema使用 JSON 校验库检查结构模型生成了不存在的工具名称工具列表描述不清晰在规划 prompt 中显式枚举工具并在执行前做工具名校验递归拆解停不下来没有限制拆解深度和数量设置 max_depth 和 max_steps超限直接终止重试无效反复报同样的错误失败原因没有反馈给模型将错误信息追加到上下文让模型“看到”前一次失败再去尝试报告内容过于简短校验规则太宽松增加最小长度、关键字段、关键实体等校验条件调用成本过高子任务拆分过细模型调用过多合并可并行执行的任务减少不必要的重试上下文被历史信息占满每步都传入完整历史精简上下文只保留当前任务所需字段7.1 一个典型报错分析有一次执行引擎报了这样一个错误Agent terminated due to error. The agent execution provider did not respond in time.这是典型的执行超时问题。常见原因是模型服务响应过慢或者工具调用阻塞。排查思路是先确认是模型调用超时还是工具调用超时。查看日志里卡在哪个子任务。如果是模型调用问题考虑降低单次请求的 max_tokens或者换用更快的小模型。如果是工具调用问题检查网络连接和第三方服务状态。最后给所有外部调用都加上超时控制和重试机制。7.2 如何避免问题再次出现最好的方式不是等出问题再排查而是从设计上降低出错概率。把“稳定”作为第一优先级不要为了智能而牺牲确定性。所有外部依赖都有降级方案比如搜索失败时使用本地缓存数据。上线前用测试用例集回归统计不同成功率指标。8. 最佳实践与工程建议8.1 拆解粒度控制任务拆解不要走极端。经验值是一个普通业务任务拆成 5 到 8 个子任务比较合理。太少每个子任务可能仍然复杂太多调用次数和失败概率都会上升。一个判断标准如果某个子任务的 prompt 里还需要再写“请一步步思考”说明它还可以继续拆。8.2 工具设计四原则Agent 的工具设计直接决定了任务拆解的下限遵循以下四个原则单一职责一个工具只做一件事比如search_market_data只负责搜索和返回结果不做清洗。参数明确工具参数越简单越好最好只接受字符串或基本类型。返回结构固定API 返回的 JSON 结构要固定方便下游解析。错误可理解工具出错时返回的错误信息要清晰方便模型或代码判断如何重试。8.3 模型调用与代码逻辑分离在 Executor 中模型只是众多工具中的一种。不要把所有逻辑都硬编码在系统提示词里也不要让模型直接操作中间变量。建议做法用独立的llm_generate()函数封装模型调用。模型输出先解析成 JSON再交给后续逻辑处理。代码负责控制流程模型负责语义生成。8.4 安全与权限边界Agent 在生产环境中必须关注安全边界子任务执行时遵循最小权限原则不做多余的数据读取或写入操作。涉及用户隐私数据时先脱敏再传给模型。禁止 Agent 自主执行高风险的数据库写入或删除操作必须经过审批或人工确认。记录完整的调用日志便于审计和回溯。8.5 上线前做“事故预演”在生产环境跑 Agent 之前建议准备一些容易翻车的测试用例比如用户输入为空。用户输入包含错别字。搜索接口返回空数据。模型输出不符合 JSON 格式。某个子任务连续重试失败。把这些场景提前跑通能避免很多线上事故。这也是 Agent 从“能跑”到“稳定交付”的必经之路。9. 总结与学习路线本文从一个调研 Agent 的翻车案例入手解释了 Agent 不稳定的本质原因并给出了任务拆解的完整方法论理解三种拆解粒度、明确模型与代码的能力边界、熟悉三种执行模式、落地一个可运行的调研报告 Agent、掌握防失控的工程手段。如果想进一步深入学习建议按这个路线走先跑通本文的完整代码修改关键字、扩展新工具体会任务拆解的控制感。学习 ReAct 等经典思路理解“思考-行动-观察”循环的意义。研究 LangChain 或 LlamaIndex 中的 Agent 执行器看看主流框架如何做任务编排和重试。接触更多真实业务尝试把固定流程、动态规划、反思修正三种模式组合运用。关注 Agent 的记忆、上下文压缩、工具安全等进阶主题建立完整的系统化能力。最后给你一个非常实用的建议下次你的 Agent 再翻车时不要急着改 prompt先问一句**“这个任务拆得够细吗”** 很多时候一个稳定交付的 Agent 和一个频繁翻车的 Agent差距不在模型而在任务拆解这一步。