AI任务分解与执行框架：从原理到实战构建智能工作流引擎

1. 项目概述与核心价值最近在折腾AI应用开发的朋友估计都绕不开一个核心痛点如何让一个AI模型比如ChatGPT真正理解并执行复杂的、多步骤的任务我们常常遇到的情况是你给AI一个指令它可能只完成第一步或者给出的结果过于笼统无法直接驱动后续的自动化流程。这正是我最初接触到tmhglnd/chatgpt-n4m这个项目时眼前一亮的根本原因。这个项目本质上是一个为ChatGPT等大语言模型设计的“任务分解与执行框架”。它的名字“N4M”可以理解为“N for M”即一个模型N为多种任务M服务但其核心在于“分解”与“编排”。简单来说它试图解决的是“大模型指令执行粒度粗”的问题。想象一下你是一位项目经理你告诉AI助手“帮我规划一个线上营销活动。”一个未经训练的模型可能会给你一份冗长的、包含各种可能性的清单。但有了chatgpt-n4m它会将这个模糊的指令自动拆解成一系列原子化的、可执行的具体任务比如1. 确定目标受众画像2. 分析竞品近期活动3. 制定预算分配表4. 设计核心宣传文案5. 排定社交媒体发布日程。更重要的是它不仅能拆解还能在一定程度上“调度”这些子任务的执行顺序和依赖关系甚至可以将某些子任务分配给更专业的工具或API去完成。这个项目的价值对于开发者、自动化工程师以及任何希望将AI能力深度集成到工作流中的人来说是巨大的。它不是一个简单的聊天接口封装而是一个迈向“AI智能体”或“AI工作流引擎”的实践性框架。通过它你可以构建出能够理解复杂需求、自主规划步骤、并调用外部资源完成目标的AI应用。接下来我将深入拆解这个项目的设计思路、核心实现并分享在复现和扩展过程中的一系列实战经验与避坑指南。2. 架构设计与核心思路拆解要理解chatgpt-n4m我们不能只看代码首先要理解其背后的设计哲学。它的核心思路源于对人类解决复杂问题过程的模仿规划、分解、执行、验证。2.1 核心组件与数据流项目的架构通常围绕几个核心组件展开虽然具体实现可能因版本而异但其思想是相通的。1. 任务解析器这是整个系统的“大脑”。它接收用户的自然语言指令并利用大语言模型的理解能力将其解析成一个结构化的任务描述。这个描述不仅包括任务目标更重要的是初步识别出其中可能隐含的子任务和这些任务之间的逻辑关系顺序、并行、条件依赖。例如指令“下载某网站文章并总结”会被解析出“网络请求”、“内容解析”、“文本摘要”等子任务模块。2. 任务分解器解析器给出了初步方向分解器则负责细化。它再次调用大语言模型根据已有的任务描述、可用的工具库以及上下文将任务递归地分解为一个个原子任务。原子任务的定义是关键它意味着这个任务可以被一个单一的、明确的函数或工具调用所完成。分解过程会生成一个任务树或任务图。3. 工具/技能库这是系统的“手”和“脚”。一个框架的强大与否很大程度上取决于它能调用多少外部能力。chatgpt-n4m通常会设计一个统一的工具注册和调用接口。工具可以非常多样从简单的计算器、时间函数到复杂的网络爬虫、数据库查询、调用其他AI服务如图像生成、操作本地文件等。每个工具都有清晰的输入、输出描述这些描述会被用于任务分解和模型调用时的提示词构建。4. 任务调度与执行引擎这是系统的“脊柱”。它负责管理任务图的状态根据依赖关系决定下一个要执行哪个原子任务调用相应的工具并处理执行结果。它需要处理成功、失败、重试、超时等情况并将结果传递给后续任务或作为最终结果返回。这里涉及到状态管理、错误处理、循环控制等经典的程序设计问题。5. 上下文管理器大语言模型是健忘的需要持续的上下文提示。上下文管理器负责维护整个任务执行过程中的对话历史和中间结果。它将先前的任务结果、工具执行输出等以结构化的方式整合到后续与模型交互的提示词中确保模型始终“知道”已经做了什么接下来该做什么。数据流大致如下用户指令 - 任务解析器 - 初始任务描述 - 任务分解器 - 任务图 - 调度引擎 - 循环选择就绪任务 - 调用工具 - 更新结果和上下文 - 判断任务图完成状态- 最终结果输出。2.2 为什么选择这样的架构这种架构的优势在于其松耦合和可扩展性。模型无关性核心的解析和分解能力虽然依赖大语言模型但可以通过抽象接口连接不同的模型提供商如OpenAI API、Azure OpenAI、或本地部署的模型。这避免了被单一供应商锁定。工具生态驱动系统的能力边界不取决于模型本身而取决于你为其注册的工具。你可以从一个简单的工具集开始随着需求增长不断丰富工具库系统的能力就会像搭积木一样增强。透明与可调试任务图提供了整个执行过程的“地图”。当复杂任务失败时你可以清晰地看到是在哪个原子任务、调用哪个工具时出的问题而不是面对一个笼统的“模型回答错误”。这对于调试和优化至关重要。注意在实现时需要平衡“模型的规划能力”和“系统的确定性”。过度依赖模型进行动态规划可能导致执行路径不稳定而过于僵化的静态任务模板又失去了灵活性。成熟的框架通常采用“混合规划”策略即预定义一些常见任务模式同时允许模型在框架内进行有限度的创造性分解。3. 核心模块实现与关键技术点理解了架构我们来看看具体实现时有哪些关键技术和细节需要把握。这里我会结合常见实现方式进行分析。3.1 任务描述的标准化提示词工程的艺术如何让大语言模型理解我们需要它进行“任务分解”这完全依赖于我们设计的提示词。一个优秀的任务解析/分解提示词通常包含以下几个部分角色定义明确告诉模型它现在是一个“任务规划专家”或“工作流分解引擎”。能力描述列出系统当前可用的所有工具及其功能、输入输出格式。这是模型进行可行规划的基础。例如“你可以使用以下工具web_search(query)返回搜索结果的摘要python_executor(code)执行一段Python代码并返回结果...”约束条件设定分解规则。例如“请将任务分解为不可再分的原子步骤”、“每个原子步骤必须对应一个可用的工具调用”、“如果任务涉及获取信息请先使用搜索工具”。输出格式规范强制要求模型以指定的结构化格式如JSON、YAML输出任务列表。这是实现程序自动化处理的关键。例如“请以以下JSON格式输出{tasks: [{id: 1, description: ..., tool: ..., input: {...}}, ...]}”示例提供1-2个从自然语言指令到结构化任务列表的完整示例。Few-shot learning能极大提升模型输出的准确性和格式符合度。在实际编码中这部分通常体现为一个或多个模板字符串在执行时动态填入工具列表、用户指令等信息然后发送给大语言模型API。# 一个简化的提示词构建示例非原项目代码仅为说明思路 def build_decomposition_prompt(user_input, available_tools): tools_description \n.join([f- {t.name}: {t.description} (输入: {t.input_schema}, 输出: {t.output_schema}) for t in available_tools]) prompt_template 你是一个高级任务规划AI。你的目标是将用户的复杂指令分解为一系列可执行的原子步骤。 当前可用的工具如下 {tools_desc} 请遵循以下规则 1. 分解后的每个原子任务必须能由上述一个工具直接完成。 2. 输出一个JSON数组每个元素代表一个任务。 3. 每个任务对象包含字段id顺序号, description任务描述, tool工具名, input输入参数字典。 示例 用户指令“查一下北京今天的天气然后告诉我是否适合户外跑步。” 输出 [ {{id: 1, description: 获取北京当前天气信息, tool: web_search, input: {{query: 北京 今天 天气}}}}, {{id: 2, description: 根据天气信息判断是否适合跑步, tool: llm_analyze, input: {{context: [上一步的结果], question: 根据天气是否适合户外跑步请简要说明理由。}}}} ] 现在请分解以下指令 用户指令{user_input} return prompt_template.format(tools_desctools_description, user_inputuser_input)3.2 工具抽象与动态调用工具库的设计是另一个核心。我们需要一个统一的接口来定义和调用工具。1. 工具定义通常使用一个基类或装饰器。每个工具需要明确其名称、描述、参数模式JSON Schema是常用选择和执行函数。# 工具基类示例 class Tool: def __init__(self, name, description, func, input_schema): self.name name self.description description self.func func # 实际执行的函数 self.input_schema input_schema # 定义输入格式 def execute(self, **kwargs): # 这里可以加入参数验证、错误处理、日志记录等 try: validate_input(kwargs, self.input_schema) # 验证输入是否符合schema result self.func(**kwargs) return {status: success, data: result} except Exception as e: return {status: error, message: str(e)} # 工具注册表 class ToolRegistry: def __init__(self): self._tools {} def register(self, tool): self._tools[tool.name] tool def get_tool(self, name): return self._tools.get(name) def list_tools(self): return list(self._tools.values())2. 动态加载与安全工具可能来自不同模块。框架应支持动态发现和注册。更重要的是安全。允许模型决定执行任意Python代码或系统命令是极其危险的。在生产环境中必须对工具进行沙箱隔离或严格的白名单控制。例如python_executor工具应该在一个受限的、无网络和文件写入权限的沙箱环境中运行或者只允许执行预审核过的代码片段。3.3 任务图的执行与状态管理执行引擎需要维护一个任务图。最简单的形式是一个有依赖关系的任务列表DAG有向无环图。每个任务节点包含其状态pending,ready,running,success,failed、结果、以及它依赖的前置任务ID列表。调度器的核心逻辑是一个循环扫描所有任务找出所有状态为pending且其所有依赖任务状态均为success的任务将其状态置为ready。从ready任务中选取一个可以是按ID顺序或基于优先级执行。调用对应工具根据工具返回结果更新任务状态为success或failed。将工具执行结果存入该任务节点并可能将其作为后续任务的输入参数。重复1-4直到所有任务完成或出现无法继续的失败。这里的关键是错误处理与重试。一个工具调用可能因为网络超时、API限制等临时原因失败。引擎应具备重试机制例如最多重试3次且重试间隔逐渐增加。对于某些非关键任务失败可能还需要设计“降级”策略或让模型决定如何调整计划。4. 实战复现从零构建一个简化版N4M引擎理论说了这么多我们来动手实现一个极度简化但核心逻辑完整的版本。这个示例将帮助你理解整个数据流转过程。4.1 环境准备与基础依赖我们使用Python主要依赖openai库或其他兼容大模型API的库和pydantic用于数据验证。# 创建虚拟环境并安装依赖假设使用OpenAI API python -m venv venv source venv/bin/activate # Linux/Mac # venv\Scripts\activate # Windows pip install openai pydantic requests4.2 实现核心类我们创建几个核心文件tool.py,task.py,planner.py,engine.py。1. 工具定义 (tool.py)from pydantic import BaseModel, Field from typing import Any, Callable, Dict import requests import json class ToolInputSchema(BaseModel): 工具输入参数的模型定义使用Pydantic便于验证。 # 这是一个基类具体工具的输入模式需要继承它并定义字段。 pass class Tool: def __init__(self, name: str, description: str, func: Callable, input_schema: type[BaseModel]): self.name name self.description description self.func func self.input_schema input_schema def execute(self, input_data: Dict[str, Any]) - Dict[str, Any]: try: # 1. 验证输入 validated_input self.input_schema(**input_data) # 2. 执行函数 result self.func(**validated_input.dict()) return {status: success, output: result} except Exception as e: return {status: error, message: fTool execution failed: {str(e)}} # 定义几个示例工具 class SearchInput(ToolInputSchema): query: str Field(description搜索查询词) def web_search(query: str) - str: 模拟一个网络搜索实际项目中可能调用SerperAPI、Google Search API等。 # 这里简化为返回固定文本 return f关于{query}的模拟搜索结果这是一个示例结果。 class CalculatorInput(ToolInputSchema): expression: str Field(description数学表达式如 1 2 * 3) def calculator(expression: str) - str: 一个简单的计算器注意实际使用中eval非常危险这里仅为演示生产环境需用安全计算库如asteval try: # 警告在生产环境中绝对不要使用eval执行未经验证的字符串。 # 此处仅为演示请替换为安全的表达式求值库。 result eval(expression, {__builtins__: {}}, {}) return str(result) except Exception as e: return f计算错误: {e} # 创建工具实例并注册 tool_registry {} search_tool Tool(nameweb_search, description执行网络搜索, funcweb_search, input_schemaSearchInput) calc_tool Tool(namecalculator, description执行数学计算, funccalculator, input_schemaCalculatorInput) tool_registry[search_tool.name] search_tool tool_registry[calc_tool.name] calc_tool2. 任务与规划 (task.py和planner.py)# task.py from pydantic import BaseModel from typing import List, Optional, Any, Dict class AtomicTask(BaseModel): id: int description: str tool_name: str input: Dict[str, Any] status: str pending # pending, ready, running, success, failed result: Optional[Any] None depends_on: List[int] [] # 依赖的任务ID列表 # planner.py import openai import json from typing import List from .tool import tool_registry # 设置你的OpenAI API密钥请从环境变量读取不要硬编码 # openai.api_key sk-... class Planner: def __init__(self, modelgpt-3.5-turbo): self.model model def decompose(self, user_instruction: str) - List[AtomicTask]: # 1. 构建提示词 tools_info [] for tool in tool_registry.values(): # 将Pydantic schema转换为JSON Schema字符串描述便于模型理解 schema_dict tool.input_schema.schema() tools_info.append(f- {tool.name}: {tool.description}. 输入格式示例: {json.dumps(schema_dict, ensure_asciiFalse)}) tools_desc \n.join(tools_info) prompt f 你是一个任务分解AI。请将用户的指令分解为一系列可顺序执行的原子任务。 每个原子任务必须能由下面列出的一个工具直接完成。 可用工具 {tools_desc} 输出要求返回一个JSON数组每个元素是一个任务对象。 任务对象字段 - id: 整数从1开始的顺序号。 - description: 字符串对该步骤的清晰描述。 - tool: 字符串必须匹配上述工具名之一。 - input: 对象该工具所需的输入参数。 - depends_on: 整数数组此任务所依赖的前置任务ID列表。如果没有依赖则为空数组。 示例指令“先搜索中国的首都是什么然后计算这个城市名字的长度乘以2。” 示例输出 json [ {{id: 1, description: 搜索中国的首都, tool: web_search, input: {{query: 中国 首都}}, depends_on: []}}, {{id: 2, description: 计算首都城市名字的长度, tool: calculator, input: {{expression: len(北京)}}, depends_on: [1]}}, {{id: 3, description: 将上一步的结果乘以2, tool: calculator, input: {{expression: 4 * 2}}, depends_on: [2]}} ]注意input中的参数值可以是具体值也可以是引用前置任务结果的占位符如${{task_1.output}}。在本次简化版中我们先使用具体值。现在请分解以下指令 指令{user_instruction} # 2. 调用大模型 response openai.ChatCompletion.create( modelself.model, messages[{role: user, content: prompt}], temperature0.1, # 低温度保证输出格式稳定 ) content response.choices[0].message.content # 3. 解析模型输出这里简单处理实际需要更健壮的解析可能包含代码块标记 # 尝试提取JSON部分 import re json_match re.search(rjson\n([\s\S]*?)\n, content) if json_match: json_str json_match.group(1) else: # 如果没有代码块假设整个内容是JSON json_str content.strip() try: task_dicts json.loads(json_str) except json.JSONDecodeError as e: print(f解析模型输出为JSON失败: {e}) print(f模型原始输出:\n{content}) return [] # 4. 转换为AtomicTask对象列表 tasks [] for td in task_dicts: task AtomicTask( idtd[id], descriptiontd[description], tool_nametd[tool], inputtd[input], depends_ontd.get(depends_on, []) ) tasks.append(task) return tasks**3. 执行引擎 (engine.py)** python from typing import List from .task import AtomicTask from .tool import tool_registry class ExecutionEngine: def __init__(self): self.tasks: List[AtomicTask] [] def load_tasks(self, tasks: List[AtomicTask]): self.tasks tasks def run(self) - dict: 执行所有任务返回最终结果 final_output None while True: # 找出所有可执行的任务状态为pending且所有依赖已完成 ready_tasks [] for task in self.tasks: if task.status pending: # 检查依赖 deps_met all( any(t.id dep_id and t.status success for t in self.tasks) for dep_id in task.depends_on ) if deps_met or not task.depends_on: task.status ready ready_tasks.append(task) if not ready_tasks: # 没有就绪任务检查是否全部完成或死锁 all_done all(t.status in [success, failed] for t in self.tasks) if all_done: # 找出最后一个任务的结果作为最终输出简化逻辑 successful_tasks [t for t in self.tasks if t.status success] if successful_tasks: final_output successful_tasks[-1].result break else: # 存在死锁有任务未完成但无就绪任务 print(错误任务图存在死锁或循环依赖) break # 执行一个就绪任务简化按顺序执行第一个 current_task ready_tasks[0] current_task.status running print(f执行任务 {current_task.id}: {current_task.description}) # 获取工具并执行 tool tool_registry.get(current_task.tool_name) if not tool: current_task.status failed current_task.result f工具 {current_task.tool_name} 未找到 print(f 失败{current_task.result}) continue # 在实际项目中这里需要处理输入参数中的动态引用如 ${{task_1.output}} # 本例简化直接使用预设的input字典 execution_result tool.execute(current_task.input) if execution_result[status] success: current_task.status success current_task.result execution_result.get(output) print(f 成功结果 - {current_task.result}) else: current_task.status failed current_task.result execution_result.get(message) print(f 失败{current_task.result}) # 打印所有任务状态 print(\n 任务执行总结 ) for task in self.tasks: print(f任务{task.id} [{task.status}]: {task.description}) return {final_output: final_output, all_tasks: self.tasks}4.3 运行一个端到端示例创建一个主文件main.py来串联整个流程from planner import Planner from engine import ExecutionEngine import asyncio async def main(): user_instruction 先搜索Python的最新版本号是多少然后计算这个版本号主版本数字的平方。 print(f用户指令: {user_instruction}) print(*50) # 1. 规划分解任务 print(步骤1: 任务规划与分解...) planner Planner() tasks planner.decompose(user_instruction) if not tasks: print(任务分解失败。) return print(f分解出 {len(tasks)} 个原子任务:) for t in tasks: print(f [{t.id}] {t.description} (工具: {t.tool_name}, 依赖: {t.depends_on})) print(\n *50) print(步骤2: 执行任务...) # 2. 执行 engine ExecutionEngine() engine.load_tasks(tasks) result engine.run() print(\n *50) print(最终结果:) print(result[final_output]) if __name__ __main__: # 注意由于使用了同步的openai调用这里直接运行。实际项目中建议使用异步。 import asyncio asyncio.run(main())运行这个程序你会看到控制台输出任务分解和执行的过程。由于我们的web_search工具是模拟的它会返回固定文本。在实际应用中你需要替换为真实的搜索API并让模型学会从搜索结果中提取“版本号”这样的具体信息这可能需要更复杂的提示词设计或多轮交互。5. 进阶优化与生产级考量我们构建的简化版演示了核心流程但要达到tmhglnd/chatgpt-n4m或类似生产级框架的水平还需要考虑大量优化。5.1 动态参数与上下文传递在之前的示例中任务的input是静态的。但在真实场景中任务B的输入往往依赖于任务A的输出。我们需要一种机制让模型在分解任务时能够用占位符如{{task_1.output}}来引用前置任务的结果并且执行引擎能在运行时动态替换这些占位符为实际值。这需要在Planner的提示词中明确教导模型使用这种语法并在ExecutionEngine执行任务前解析input字典递归地查找并替换所有占位符。这涉及到模板渲染和上下文查找。5.2 循环与条件分支复杂的任务可能包含循环例如“重复搜索直到找到满意答案”或条件分支例如“如果天气为晴则推荐户外活动否则推荐室内活动”。这需要扩展任务图的表达能力使其不仅能表示线性依赖还能表示循环和条件节点。这通常通过引入特殊的“控制工具”来实现例如while_loop(condition, task_subgraph): 当条件满足时重复执行子任务图。if_else(condition, task_if, task_else): 根据条件选择执行路径。模型在分解时需要生成包含这些控制节点的任务图。执行引擎则需要解释这些节点动态地展开或选择执行路径。5.3 模型的自我反思与纠错大模型并非完美它可能分解出不可行或低效的任务计划。一个成熟的框架会引入“反思”环节。即在任务执行失败或结果不理想时让模型回顾之前的计划、执行历史和错误信息重新规划或调整后续步骤。这可以通过在关键节点如任务失败、或所有任务完成后进行结果评估插入一个调用模型的“反思”任务来实现。5.4 持久化与状态恢复长时间运行或复杂的任务流可能需要中断后继续执行。这就需要将整个任务图的状态包括每个任务的状态、输入输出结果持久化到数据库或文件中。当系统重启或从故障中恢复时可以从上次保存的状态点继续执行而不是从头开始。5.5 监控、日志与可观测性在生产环境中清晰的日志和监控至关重要。你需要记录每个任务的开始、结束时间、状态。每次模型调用的请求和响应注意脱敏。每个工具调用的输入、输出和错误信息。整个工作流的执行图谱和实时状态。这有助于调试问题、分析性能瓶颈是模型调用慢还是某个工具慢和计算成本。6. 常见问题、踩坑记录与排查技巧在开发和测试这类系统的过程中我遇到了不少典型问题这里分享一些排查思路。问题1模型不按指定格式输出JSON。现象模型返回了自然语言描述而不是解析器期待的JSON数组。原因提示词不够明确或模型温度参数过高。解决强化格式指令在提示词中非常明确地指定格式使用“你必须输出JSON且只输出JSON”、“输出必须是一个有效的JSON数组不要有任何其他解释文字”等强约束语句。提供更清晰的示例示例的输入输出要典型且格式完美。降低温度将temperature参数设为0.1或0减少随机性。使用函数调用如果使用的模型支持如gpt-3.5-turbo-1106及以上版本使用OpenAI的“函数调用”功能让模型以调用预定义函数的方式输出结构化数据这比在提示词中约束格式要可靠得多。后处理与重试在代码中尝试从返回文本中提取JSON部分用正则匹配json ...如果解析失败可以尝试让模型修正发送错误信息并要求重试。问题2工具执行失败导致整个流程中断。现象某个原子任务调用工具时出错如API超时、参数错误任务状态变为failed后续依赖任务无法执行。原因工具本身不可靠或输入参数不符合预期。解决完善的错误处理在每个工具的执行函数内部进行try-catch返回统一的错误结构而不是抛出异常导致引擎崩溃。重试机制对于网络超时等临时错误在执行引擎层面实现指数退避的重试逻辑。输入验证与清洗在工具执行前严格验证输入参数是否符合schema。对于模型生成的参数可能需要进行额外的清洗如去除多余空格、引号。备选方案/降级设计任务流时考虑关键路径。对于非关键任务失败是否可以跳过或者是否有备用工具这可能需要更复杂的任务图逻辑。问题3模型分解的任务顺序不合理或工具选择错误。现象模型规划的步骤逻辑混乱比如在获取必要信息之前就尝试进行计算。原因模型对任务领域知识或工具能力理解不足。解决优化工具描述确保工具的描述清晰、无歧义并说明其前置条件。例如“get_user_profile工具需要在用户已登录后才能调用”。在提示词中加入领域知识在分解提示词的开头加入关于该领域任务的一般性步骤指导。例如“在规划数据分析任务时通常步骤是1. 数据获取2. 数据清洗3. 数据分析4. 结果可视化。”迭代式规划不要试图一次生成完美的完整计划。可以采用“逐步展开”的方式先让模型生成一个高层计划然后对每个高层步骤再递归地进行细化分解。这样更容易控制。人工审核或干预对于非常重要的流程可以引入人工审核步骤在模型生成计划后由人确认或调整后再执行。问题4任务执行陷入无限循环或死锁。现象执行引擎卡住找不到可以执行的ready任务但还有任务处于pending状态。原因任务图中存在循环依赖A依赖BB又依赖A或者模型错误地设置了依赖关系。解决依赖环检测在执行前或加载任务图时运行一次简单的环检测算法如拓扑排序。如果发现环则报错并终止。可视化任务图将生成的任务图可视化出来可以使用Graphviz等库人工检查依赖关系是否合理。这对于调试复杂的任务流非常有用。设置超时和最大步数在执行引擎中设置一个全局计数器或超时时间防止因意外循环导致资源耗尽。问题5处理长上下文和令牌限制。现象当任务步骤很多中间结果也很丰富时组合起来的上下文可能超过模型的令牌限制。原因大模型有上下文窗口限制如4K、8K、16K、128K令牌。解决摘要与压缩不要将每个任务的原始输出都塞进后续任务的提示词。对于较长的文本结果如一篇搜索得到的文章让模型或一个专用工具先对其进行摘要只传递摘要。选择性上下文只将最相关的历史信息放入上下文。可以设计一个“上下文选择器”根据当前任务从历史中筛选出最关键的几个结果。分阶段规划与执行将大任务拆分成相对独立的子阶段。每个阶段规划、执行、总结然后将总结作为下一个阶段的输入。这样每个阶段的上下文都是可控的。使用支持长上下文的模型如果成本允许优先选择上下文窗口更大的模型。构建一个强大的AI任务分解与执行框架是一个持续迭代的过程。从tmhglnd/chatgpt-n4m这类项目中我们学到的最重要的不是具体的代码而是这种“让模型思考让系统执行”的架构范式。它有效地将大语言模型的规划、推理能力与确定性程序的可靠执行能力结合起来为构建真正智能、自动化的AI应用打开了大门。