企业官网建设流程全解析

任务规划器

摘要

在大模型应用落地的过程中，一个核心痛点始终困扰着开发者：用户往往只给出一句模糊的自然语言需求，例如"帮我分析一下上个月的销售数据并生成一份报告"，而系统却需要把它转化为一系列可执行、可调度、可验证的具体步骤。这个从"一句话"到"任务链"的转换过程，正是**任务规划器（Task Planner）**要解决的核心问题。

本文将以 Harness 框架为背景，深入剖析任务规划器的设计思路与实现细节。我们将从问题定义出发，逐步讲解需求解析、任务分解、依赖建模、执行调度以及失败重试等关键环节，并给出完整可运行的代码示例。读完本文，你将能够理解如何构建一个能够把模糊需求拆解为可执行多步任务链的规划系统。

引言：为什么"一句话"到"任务链"这么难？

先来看一个真实的场景。用户在对话框中输入：

“帮我整理一下本周的客户反馈，找出投诉最多的三个问题，并给每个问题写一份改进建议。”

对人类来说，这句话虽然信息密度高，但理解起来并不困难。可对机器而言，这里隐藏着大量需要推断的信息：

1. 数据来源：客户反馈存在哪里？数据库、工单系统还是聊天记录？
2. 时间范围："本周"具体是哪几天？是否包含周末？
3. 处理逻辑："投诉最多"如何定义？按数量、按严重程度还是按情绪强度？
4. 输出形式："改进建议"要写成什么格式？多长？给谁看？
5. 执行顺序：哪些步骤必须串行，哪些可以并行？

传统的工作流引擎（如 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):PENDING="pending"RUNNING="running"SUCCESS="success"FAILED="failed"SKIPPED="skipped"@dataclassclassTask:"""原子任务：规划器产出的最小执行单元"""task_id:strname:strtool:str# 要调用的工具名params:Dict[str,Any]=field(default_factory=dict)depends_on:List[str]=field(default_factory=list)# 依赖的 task_idstatus:TaskStatus=TaskStatus.PENDING result:Any=Noneretry:int=0max_retry:int=2defis_ready(self,completed:set)->bool:"""所有依赖都完成时，本任务才可执行"""returnall(depincompletedfordepinself.depends_on)@dataclassclassTaskGraph:"""任务链：由多个 Task 组成的 DAG"""goal:strtasks:List[Task]=field(default_factory=list)defget_task(self,task_id:str)->Optional[Task]:fortinself.tasks:ift.task_id==task_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:node=queue.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 解析用户意图，返回结构化字典"""prompt=INTENT_PROMPT.format(user_input=user_input)raw=llm_client.chat(prompt)# 容错：剥离可能存在的 markdown 代码块标记raw=raw.strip().removeprefix("```json").removeprefix("```").removesuffix("```").strip()try:intent=json.loads(raw)exceptjson.JSONDecodeErrorase:raiseValueError(f"意图解析失败，模型输出非合法 JSON:{raw}")frome# 字段校验，保证下游可用forkeyin("goal","domain","entities","constraints","expected_output"):ifkeynotinintent:intent[key]=[]ifkeyin("constraints",)else""returnintent

以开头的客户反馈需求为例，解析结果大致是：

{"goal":"整理本周客户反馈，找出投诉最多的三个问题并给出改进建议","domain":"customer_service","entities":{"time_range":"本周","top_n":3},"constraints":["聚焦投诉类反馈","每个问题配一份改进建议"],"expected_output":"包含三个问题及对应建议的报告"}

这一步的价值在于：把模糊的自然语言转化为机器可推理的结构化字段。后续的任务分解会直接消费这些字段，而不是重新去"猜"用户的意思。这也是 Harness 相比纯 Prompt 链式调用的关键优势——意图被显式建模，可缓存、可校验、可回溯。

三、任务分解：把目标拆成原子任务

有了结构化意图，接下来就是核心环节：任务分解。Harness 采用"工具目录 + LLM 选择"的策略。系统预先注册一批可用工具（Tool），每个工具声明自己的功能和参数 schema；LLM 的职责是从工具目录中挑选合适的工具，并组装成任务链。

# 工具注册表：描述每个工具的能力与参数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_desc=json.dumps(TOOL_REGISTRY,ensure_ascii=False,indent=2)prompt=DECOMPOSE_PROMPT.format(intent=json.dumps(intent,ensure_ascii=False),tools=tools_desc,)raw=llm_client.chat(prompt)raw=raw.strip().removeprefix("```json").removeprefix("```").removesuffix("```").strip()tasks=json.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 并做校验。校验主要解决三类问题：

1. 环检测：任务 A 依赖 B，B 又依赖 A，会导致死锁。
2. 悬空依赖：depends_on引用了不存在的 task_id。
3. 参数完整性：工具所需参数是否齐备。

defbuild_graph(goal:str,task_dicts:List[Dict])->TaskGraph:"""把 LLM 输出的任务列表构建为 TaskGraph，并做校验"""tasks=[Task(task_id=d["task_id"],name=d["name"],tool=d["tool"],params=d.get("params",{}),depends_on=d.get("depends_on",[]),)fordintask_dicts]graph=TaskGraph(goal=goal,tasks=tasks)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:schema=TOOL_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:set=set()batches:List[List[Task]]=[]whileremaining:# 找出所有依赖已满足的任务ready_ids=[tidfortid,depsinremaining.items()ifdeps<=completed]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.tools=toolsdef_resolve_params(self,params:Dict,context:Dict)->Dict:"""把 {{t1.result}} 占位符替换为真实结果"""resolved={}fork,vinparams.items():ifisinstance(v,str):# 匹配 {{task_id.result}}defrepl(m):tid=m.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_fn=self.tools[task.tool]params=self._resolve_params(task.params,context)last_err=Noneforattemptinrange