任务规划器摘要在大模型应用落地的过程中一个核心痛点始终困扰着开发者用户往往只给出一句模糊的自然语言需求例如帮我分析一下上个月的销售数据并生成一份报告而系统却需要把它转化为一系列可执行、可调度、可验证的具体步骤。这个从一句话到任务链的转换过程正是**任务规划器Task Planner**要解决的核心问题。本文将以 Harness 框架为背景深入剖析任务规划器的设计思路与实现细节。我们将从问题定义出发逐步讲解需求解析、任务分解、依赖建模、执行调度以及失败重试等关键环节并给出完整可运行的代码示例。读完本文你将能够理解如何构建一个能够把模糊需求拆解为可执行多步任务链的规划系统。引言为什么一句话到任务链这么难先来看一个真实的场景。用户在对话框中输入“帮我整理一下本周的客户反馈找出投诉最多的三个问题并给每个问题写一份改进建议。”对人类来说这句话虽然信息密度高但理解起来并不困难。可对机器而言这里隐藏着大量需要推断的信息数据来源客户反馈存在哪里数据库、工单系统还是聊天记录时间范围本周具体是哪几天是否包含周末处理逻辑投诉最多如何定义按数量、按严重程度还是按情绪强度输出形式改进建议要写成什么格式多长给谁看执行顺序哪些步骤必须串行哪些可以并行传统的工作流引擎如 Airflow、Argo要求开发者预先把 DAG有向无环图画好每个节点都是明确的。但大模型应用面对的是动态、开放、模糊的输入静态 DAG 根本无法覆盖。于是任务规划器应运而生——它要做的就是在运行时把一句话编译成一张可执行的 DAG。Harness 的设计哲学是规划Planning与执行Execution分离。规划器只负责产出结构化的任务链执行器只负责按图索骥地跑。这种分离让系统既能借助 LLM 的推理能力处理模糊性又能保持工程上的可控与可观测。正文一、任务规划器的整体架构在 Harness 中任务规划器位于用户输入和任务执行引擎之间扮演着编译器的角色。它的输入是一段自然语言输出是一个结构化的TaskGraph。整体流程可以概括为四个阶段用户输入 │ ▼ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ 意图解析 │ → │ 任务分解 │ → │ 依赖建模 │ → │ 校验优化 │ │ Intent │ │ Decompose │ │ Dependency │ │ Validate │ └─────────────┘ └─────────────┘ └─────────────┘ └─────────────┘ │ ▼ TaskGraph可执行任务链意图解析识别用户真正想要达成的目标Goal而不是字面动作。任务分解把 Goal 拆成若干原子任务Atomic Task每个任务对应一个可调用的工具Tool。依赖建模分析任务之间的数据依赖与顺序依赖构建 DAG。校验优化检查环、冗余、缺失参数并做并行化优化。下面我们先定义核心数据结构这是理解后续所有代码的基础。fromdataclassesimportdataclass,fieldfromtypingimportAny,Callable,Dict,List,OptionalfromenumimportEnumclassTaskStatus(Enum):PENDINGpendingRUNNINGrunningSUCCESSsuccessFAILEDfailedSKIPPEDskippeddataclassclassTask:原子任务规划器产出的最小执行单元task_id:strname:strtool:str# 要调用的工具名params:Dict[str,Any]field(default_factorydict)depends_on:List[str]field(default_factorylist)# 依赖的 task_idstatus:TaskStatusTaskStatus.PENDING result:AnyNoneretry:int0max_retry:int2defis_ready(self,completed:set)-bool:所有依赖都完成时本任务才可执行returnall(depincompletedfordepinself.depends_on)dataclassclassTaskGraph:任务链由多个 Task 组成的 DAGgoal:strtasks:List[Task]field(default_factorylist)defget_task(self,task_id:str)-Optional[Task]:fortinself.tasks:ift.task_idtask_id:returntreturnNonedeftopological_sort(self)-List[Task]:拓扑排序保证依赖顺序indegree{t.task_id:len(t.depends_on)fortinself.tasks}queue[tfortinself.tasksifindegree[t.task_id]0]ordered:List[Task][]whilequeue:nodequeue.pop(0)ordered.append(node)fortinself.tasks:ifnode.task_idint.depends_on:indegree[t.task_id]-1ifindegree[t.task_id]0:queue.append(t)iflen(ordered)!len(self.tasks):raiseValueError(任务图中存在环无法拓扑排序)returnordered这段代码定义了任务规划器的词汇表。Task是原子任务TaskGraph是任务链。注意depends_on字段——它是构建 DAG 的关键也是后续并行调度的依据。二、意图解析从自然语言到结构化目标意图解析是规划器的第一道关口。它的任务不是马上拆步骤而是先搞清楚用户到底想要什么。在 Harness 中我们用一个 LLM Prompt 来完成这一步并强制模型输出 JSON。importjsonfromtypingimportDict,Any INTENT_PROMPT你是一个任务规划助手。请分析用户需求输出结构化意图。 用户需求{user_input} 请严格输出如下 JSON 格式不要输出任何多余文字 {{ goal: 一句话概括用户的核心目标, domain: 所属领域如 data_analysis / customer_service / report, entities: {{关键实体: 值}}, constraints: [约束条件1, 约束条件2], expected_output: 期望的最终产出形式 }} defparse_intent(user_input:str,llm_client)-Dict[str,Any]:调用 LLM 解析用户意图返回结构化字典promptINTENT_PROMPT.format(user_inputuser_input)rawllm_client.chat(prompt)# 容错剥离可能存在的 markdown 代码块标记rawraw.strip().removeprefix(json).removeprefix().removesuffix().strip()try:intentjson.loads(raw)exceptjson.JSONDecodeErrorase:raiseValueError(f意图解析失败模型输出非合法 JSON:{raw})frome# 字段校验保证下游可用forkeyin(goal,domain,entities,constraints,expected_output):ifkeynotinintent:intent[key][]ifkeyin(constraints,)elsereturnintent以开头的客户反馈需求为例解析结果大致是{goal:整理本周客户反馈找出投诉最多的三个问题并给出改进建议,domain:customer_service,entities:{time_range:本周,top_n:3},constraints:[聚焦投诉类反馈,每个问题配一份改进建议],expected_output:包含三个问题及对应建议的报告}这一步的价值在于把模糊的自然语言转化为机器可推理的结构化字段。后续的任务分解会直接消费这些字段而不是重新去猜用户的意思。这也是 Harness 相比纯 Prompt 链式调用的关键优势——意图被显式建模可缓存、可校验、可回溯。三、任务分解把目标拆成原子任务有了结构化意图接下来就是核心环节任务分解。Harness 采用工具目录 LLM 选择的策略。系统预先注册一批可用工具Tool每个工具声明自己的功能和参数 schemaLLM 的职责是从工具目录中挑选合适的工具并组装成任务链。# 工具注册表描述每个工具的能力与参数TOOL_REGISTRY{fetch_feedback:{description:从数据源拉取指定时间范围内的客户反馈,params:{start_date:str,end_date:str,source:str},},classify_feedback:{description:对反馈进行情感与类别分类识别投诉类反馈,params:{feedback_list:list},},aggregate_topics:{description:按主题聚合统计出现频次返回 Top-N 问题,params:{classified_list:list,top_n:int},},generate_suggestion:{description:针对单个问题生成改进建议,params:{issue:str,context:str},},compose_report:{description:把多个问题的建议整合成一份完整报告,params:{sections:list,title:str},},}工具目录是规划器的知识边界。LLM 只能从已注册的工具中选择这既避免了模型幻觉出不存在的能力也让规划结果天然可执行。接下来是分解 Prompt 与解析逻辑DECOMPOSE_PROMPT你是一个任务规划器。根据用户意图和可用工具拆解出可执行的任务链。 用户意图 {intent} 可用工具只能从中选择 {tools} 要求 1. 每个任务必须使用上面列出的工具不能自创工具。 2. 用 task_id 标识任务如 t1、t2。 3. 用 depends_on 声明依赖关系前置任务的 task_id 列表。 4. 没有依赖的任务 depends_on 为空数组。 5. 参数中如果需要引用前置任务的输出用 {{{{t1.result}}}} 占位。 请严格输出 JSON 数组格式如下 [ {{task_id: t1, name: 任务名, tool: 工具名, params: {{}}, depends_on: []}} ] defdecompose(intent:Dict[str,Any],llm_client)-List[Dict]:把意图分解为任务列表tools_descjson.dumps(TOOL_REGISTRY,ensure_asciiFalse,indent2)promptDECOMPOSE_PROMPT.format(intentjson.dumps(intent,ensure_asciiFalse),toolstools_desc,)rawllm_client.chat(prompt)rawraw.strip().removeprefix(json).removeprefix().removesuffix().strip()tasksjson.loads(raw)# 校验工具合法性fortintasks:ift[tool]notinTOOL_REGISTRY:raiseValueError(f任务{t[task_id]}使用了未注册工具:{t[tool]})returntasks对客户反馈场景模型可能产出如下任务链[{task_id:t1,name:拉取本周反馈,tool:fetch_feedback,params:{start_date:2024-06-03,end_date:2024-06-09,source:ticket_system},depends_on:[]},{task_id:t2,name:分类反馈,tool:classify_feedback,params:{feedback_list:{{t1.result}}},depends_on:[t1]},{task_id:t3,name:聚合Top3问题,tool:aggregate_topics,params:{classified_list:{{t2.result}},top_n:3},depends_on:[t2]},{task_id:t4,name:生成改进建议,tool:generate_suggestion,params:{issue:{{t3.result}},context:客户反馈},depends_on:[t3]},{task_id:t5,name:整合报告,tool:compose_report,params:{sections:{{t4.result}},title:本周客户反馈改进报告},depends_on:[t4]}]可以看到任务链清晰、依赖明确且每一步都指向了具体工具。{{t1.result}}这种占位符表示数据在任务间的流动执行器会在运行时替换成真实值。四、依赖建模与图校验任务列表还只是线性的候选集真正的任务链需要构建成 DAG 并做校验。校验主要解决三类问题环检测任务 A 依赖 BB 又依赖 A会导致死锁。悬空依赖depends_on引用了不存在的 task_id。参数完整性工具所需参数是否齐备。defbuild_graph(goal:str,task_dicts:List[Dict])-TaskGraph:把 LLM 输出的任务列表构建为 TaskGraph并做校验tasks[Task(task_idd[task_id],named[name],toold[tool],paramsd.get(params,{}),depends_ond.get(depends_on,[]),)fordintask_dicts]graphTaskGraph(goalgoal,taskstasks)validate_graph(graph)returngraphdefvalidate_graph(graph:TaskGraph)-None:图校验悬空依赖、环、参数完整性ids{t.task_idfortingraph.tasks}# 1. 悬空依赖检测fortingraph.tasks:fordepint.depends_on:ifdepnotinids:raiseValueError(f任务{t.task_id}依赖了不存在的任务{dep})# 2. 环检测借助拓扑排序graph.topological_sort()# 3. 参数完整性校验fortingraph.tasks:schemaTOOL_REGISTRY[t.tool][params]forparam_nameinschema:ifparam_namenotint.params:raiseValueError(f任务{t.task_id}缺少工具{t.tool}的必需参数:{param_name})校验通过后topological_sort()会返回一个合法的执行序列。但注意拓扑排序给出的是一种合法顺序而不是唯一最优顺序。真正执行时我们更关心的是哪些任务可以并行。defget_parallel_batches(graph:TaskGraph)-List[List[Task]]:把任务图划分为若干批次同批次任务可并行执行remaining{t.task_id:set(t.depends_on)fortingraph.tasks}completed:setset()batches:List[List[Task]][]whileremaining:# 找出所有依赖已满足的任务ready_ids[tidfortid,depsinremaining.items()ifdepscompleted]ifnotready_ids:raiseValueError(存在无法满足的依赖可能是环或逻辑错误)batch[graph.get_task(tid)fortidinready_ids]batches.append(batch)fortidinready_ids:remaining.pop(tid)completed.add(tid)returnbatches对上面的客户反馈任务链批次划分结果是批次 1[t1]批次 2[t2]批次 3[t3]批次 4[t4]批次 5[t5]看起来是纯串行。但如果我们把需求改成给每个问题分别生成建议那么t4就可以被拆成t4a/t4b/t4c三个任务它们同属批次 4可以并行执行——这正是任务规划器在性能优化上的价值所在。五、执行调度与失败重试规划出来的任务链最终要落到执行器上。Harness 的执行器遵循批次驱动 依赖等待的模型每一批次内的任务并发执行全部完成后进入下一批次。importconcurrent.futuresimportrefromtypingimportCallableclassExecutor:def__init__(self,tools:Dict[str,Callable]):self.toolstoolsdef_resolve_params(self,params:Dict,context:Dict)-Dict:把 {{t1.result}} 占位符替换为真实结果resolved{}fork,vinparams.items():ifisinstance(v,str):# 匹配 {{task_id.result}}defrepl(m):tidm.group(1)returnstr(context.get(tid,{}).get(result))resolved[k]re.sub(r\{\{(\w)\.result\}\},repl,v)else:resolved[k]vreturnresolveddef_run_task(self,task:Task,context:Dict)-Any:执行单个任务带重试tool_fnself.tools[task.tool]paramsself._resolve_params(task.params,context)last_errNoneforattemptinrange