前段时间在搭建一个面向 LLM Agent 的自动化运行环境时我被一个问题卡了很久harness 到底能不能同时做到“确定性调度”和“任务编排”网上关于 harness 的资料大多停留在“怎么安装”“怎么启动工作区”很少有人把调度器与任务编排器之间的边界讲清楚更少有人讨论如何在真实代码里把这两个能力组合起来。今天这篇文章就围绕这个问题做一个完整拆解包含概念梳理、架构设计、可运行的 Python 示例以及工程落地时容易踩的坑。无论你是刚接触 harness 的开发者还是正在设计内部任务编排系统这篇文章应该都能提供一些可以直接复用的思路。1. 背景与核心概念1.1 什么是 Harness在软件工程领域harness 这个词并没有严格的统一翻译它通常指“为了控制、观测、驱动某个系统运行而包裹在系统外部的一层可编程框架”。你可以把它理解成“运行控制壳”它不只负责启动一个模型、一个脚本或一组任务还负责决定这些任务以什么顺序运行、在什么条件下重试、把哪些输入传给哪些阶段以及最终把结果汇总成什么结构。最近经常听到的 DeepSeek Harness、Codex Harness、Cursor Harness 等名词本质上都是把某个底层模型或工具包进一个可编程的运行时环境中。harness 的职责边界通常包括定义任务的输入输出协议维护任务间的依赖关系控制任务执行顺序与并发策略记录运行轨迹与状态快照在失败时执行重试或回退策略。所以讨论“harness 能否拥有一个真正确定性的调度器和任务编排器”其实是在问我们能否在 harness 内部构造一套既稳定可复现、又具备完整编排能力的基础设施。1.2 什么是 Deterministic SchedulerDeterministic Scheduler即确定性调度器指的是在给定相同输入和相同任务集合的情况下调度器每次产生的任务执行计划都完全一致。这里的关键并不只是“最终结果一致”而是“过程一致”任务顺序、中间状态、重试次数、日志输出都应当可复现。确定性调度器通常需要满足以下约束调度决策不依赖系统当前时间调度决策不依赖随机数调度决策不依赖并发执行时的竞态条件调度决策只依赖任务定义和输入数据的规范化表示。举个例子同样一组任务 A、B、C如果 A 不依赖 BB 不依赖 C普通调度器可能这次先执行 B下次先执行 A但确定性调度器必须通过某种稳定规则比如任务 ID 字典序选定唯一顺序。1.3 什么是 Task OrchestratorTask Orchestrator即任务编排器负责管理一个任务从开始到结束的完整生命周期。编排器不仅要调度任务顺序还要处理上下文传递前一个任务的输出如何变成后一个任务的输入状态维护每个任务处于 pending、running、succeeded、failed 中的哪个状态执行策略是否重试、是否超时、是否跳过数据汇总最终输出的结构由哪些任务的输出拼接而成。编排器更像一个“工作流引擎”而调度器更关注“顺序与资源分配”。两者确实可以分离也可以集成在同一个 harness 中。1.4 为什么二者可以同时存在很多人会把“确定性”和“编排”看成对立关系理由是真实业务中总有一些任务需要动态决策如果上一个任务失败了下一步要执行另一个分支这听起来就像“不固定”。但这里有一个关键区分动态决策可以使用确定性的决策规则。比如“失败后重试 3 次”“失败后走 fallback 分支”这些规则是确定的。harness 完全可以用固定规则驱动动态流程从而做到既能编排又有确定性。2. 从“能跑”到“可复现”harness 的设计目标2.1 普通任务执行器和 harness 的区别普通任务执行器通常就是“按顺序调用函数”谈不上调度。harness 的复杂度在于它需要面对多个任务、多种输入、多种失败可能性还要保证运行过程可以被审计和回放。一个合格的 harness 应该具备四个能力可编程任务不是写死在 if-else 里的而是可以被声明、注册、组合可观测每一步执行都应该有记录能回答“这个结果是怎么算出来的”可复现相同输入在相同 harness 版本下产生相同运行轨迹可恢复单点失败不影响整体且失败策略明确。2.2 为什么需要确定性在 AI Agent 和 LLM 应用场景下确定性尤其重要。模型本身带有随机性如果外层 harness 的调度也是随机的问题排查会变得极其困难。今天调用同一个 Prompt 得到的结果和昨天不一样可能是模型温度参数导致的也可能是任务执行顺序变化导致的。如果 harness 具备确定性调度至少可以把“业务层波动”和“基础设施层波动”分开。对于 CI/CD、数据处理、自动化测试等场景确定性调度同样关键。一次流水线运行完成后如果无法按原样重放问题定位就会变成“猜谜”。2.3 确定性的边界需要注意确定性调度并不是“物理世界的绝对确定”而是“策略层面的可复现”。外部 API 返回什么、数据库里是否存在某个字段这些因素仍然会影响任务结果。但 harness 能保证的是在相同外部条件下任务的执行路径和顺序是稳定的。理解这个边界很重要否则会陷入一个误区认为只要做了确定性调度所有结果都能复现。实际项目中我们通常会通过日志、状态快照、输入哈希等方式记录外部依赖的实际值用来辅助复现。3. 架构设计把调度与执行分离3.1 核心组件划分在实现一个带确定性调度与任务编排能力的 harness 时我最推荐的方式是拆成四个独立的组件TaskGraph任务图定义任务节点和依赖关系DeterministicScheduler确定性调度器根据任务图和输入生成稳定执行计划TaskOrchestrator任务编排器负责执行计划的具体推进、状态维护和上下文传递Harness对外统一入口把上述组件组合起来。下面用一张简化的职责表来表示组件核心职责是否产生决策决策是否确定性TaskGraph声明任务与依赖否-DeterministicScheduler生成执行计划是是TaskOrchestrator推进状态、执行任务部分失败分支等通过规则保证Harness组合装配否-3.2 为什么要用 DAG任务编排的基础是任务依赖关系。最稳定、最容易推理的依赖模型是有向无环图DAG。每个任务是一个节点节点之间的箭头表示依赖关系。只有所有前置依赖都成功完成后后置任务才会启动。DAG 有三个优点无循环依赖天然避免死锁可以稳定地做拓扑排序便于可视化、审计和测试。3.3 确定性策略怎么落到架构里我把确定性策略拆成三条具体规则节点 ID 必须可计算每个任务节点使用“输入哈希 任务名”生成稳定 ID同层顺序必须稳定拓扑排序后同一批次可执行的任务按 ID 字典序排序执行决策不带随机副作用调度器内部不调用 random、time、uuid 等函数。这三条规则能保证只要输入 payload 和任务定义不变调度器生成的执行计划就永远不变。4. 环境准备与项目结构本文示例使用 Python 标准库实现不需要额外依赖凡是 Python 3.9 以上的环境都可以直接运行。仍然建议先准备好一个虚拟环境mkdir deterministic-harness cd deterministic-harness python3 -m venv venv source venv/bin/activate示例项目的文件结构如下deterministic-harness/ ├── task_graph.py ├── scheduler.py ├── orchestrator.py ├── harness.py └── demo.py每个文件的职责很清晰下面逐一实现。5. 完整代码实现5.1 任务图定义先定义任务节点。每个任务节点包含四个关键字段任务名、执行函数、依赖列表、最大重试次数。# task_graph.py from __future__ import annotations import hashlib import json from dataclasses import dataclass, field from typing import Any, Callable, Dict, List, Optional dataclass class TaskNode: name: str fn: Callable[[Dict[str, Any]], Dict[str, Any]] depends_on: List[str] field(default_factorylist) max_retries: int 0 timeout_seconds: float 30.0 class TaskGraph: def __init__(self) - None: self._tasks: Dict[str, TaskNode] {} def add_task(self, task: TaskNode) - None: for dep in task.depends_on: if dep not in self._tasks: raise ValueError(f依赖任务 {dep} 尚未注册) self._tasks[task.name] task def get_task(self, name: str) - TaskNode: return self._tasks[name] def all_tasks(self) - List[TaskNode]: return list(self._tasks.values()) def dependencies_of(self, name: str) - List[str]: return self._tasks[name].depends_on def compute_task_id(self, payload: Dict[str, Any], task_name: str) - str: normalized json.dumps(payload, sort_keysTrue, ensure_asciiFalse) raw f{task_name}:{normalized} return hashlib.sha256(raw.encode(utf-8)).hexdigest()[:16]TaskGraph 在注册任务时会检查依赖是否已经存在这样可以尽早发现配置错误。compute_task_id 方法体现了确定性原则它不依赖时间、不依赖随机数只依赖任务名和规范化后的输入 payload。5.2 确定性调度器调度器负责把任务图转换成稳定的执行计划。这里采用经典拓扑排序思路并用队列存储当前可执行任务。为了保证确定性每次从队列取出任务时都按任务 ID 排序。# scheduler.py from typing import Any, Dict, List, Set from task_graph import TaskGraph class DeterministicScheduler: def __init__(self, graph: TaskGraph) - None: self.graph graph def build_plan(self, payload: Dict[str, Any]) - List[str]: tasks self.graph.all_tasks() in_degree: Dict[str, int] {} for task in tasks: in_degree[task.name] len(task.depends_on) ready: List[str] [] for task in tasks: if in_degree[task.name] 0: ready.append(task.name) plan: List[str] [] executed: Set[str] set() while ready: # 关键点按确定性 task_id 排序 ready.sort(keylambda name: self.graph.compute_task_id(payload, name)) current ready.pop(0) plan.append(current) executed.add(current) for task in tasks: if task.name in executed: continue if current in task.depends_on: in_degree[task.name] - 1 if in_degree[task.name] 0: ready.append(task.name) if len(plan) ! len(tasks): raise RuntimeError(任务图中存在循环依赖无法生成执行计划) return plan这个调度器的核心逻辑是先统计每个任务的入度入度为 0 的任务进入 ready 队列每次从 ready 队列中按 task_id 排序后取出一个任务执行后更新依赖它的任务入度重复直到所有任务都进入 plan。由于排序 key 使用 compute_task_id而 compute_task_id 只依赖 payload 和任务名所以相同 payload 生成的计划一定一样。5.3 任务编排器编排器负责真正执行任务并管理上下文和状态。# orchestrator.py import time import traceback from typing import Any, Dict, List from scheduler import DeterministicScheduler from task_graph import TaskGraph class TaskOrchestrator: def __init__(self, graph: TaskGraph, scheduler: DeterministicScheduler) - None: self.graph graph self.scheduler scheduler def run(self, payload: Dict[str, Any]) - Dict[str, Any]: plan: List[str] self.scheduler.build_plan(payload) context: Dict[str, Any] { payload: payload, results: {}, trace: [], } for task_name in plan: task self.graph.get_task(task_name) task_id self.graph.compute_task_id(payload, task_name) # 准备输入把前置任务的结果合并进当前上下文 inputs {payload: payload, context: context} attempt 0 last_error: Exception | None None while attempt task.max_retries: try: self._mark(context, task_name, running, task_id) output task.fn(inputs) context[results][task_name] output self._mark(context, task_name, succeeded, task_id) break except Exception as exc: # noqa: BLE001 attempt 1 last_error exc self._mark(context, task_name, failed, task_id) if attempt task.max_retries: time.sleep(0.01) # 固定重试间隔不引入随机等待 else: raise RuntimeError( f任务 {task_name} 失败重试 {task.max_retries} 次后终止 ) from exc return context staticmethod def _mark(context: Dict[str, Any], task_name: str, status: str, task_id: str) - None: context[trace].append( { task: task_name, task_id: task_id, status: status, } )编排器中的几个设计要点context 同时保存 payload、每个任务的结果和完整 trace方便审计重试策略也保持确定性固定次数、固定间隔每次执行前都重新计算 task_id确保 trace 中的 ID 可复现。5.4 Harness 统一入口Harness 只是把 TaskGraph、DeterministicScheduler、TaskOrchestrator 组合起来暴露一个 run 方法。# harness.py from typing import Any, Dict from orchestrator import TaskOrchestrator from scheduler import DeterministicScheduler from task_graph import TaskGraph class Harness: def __init__(self) - None: self.graph TaskGraph() self.scheduler DeterministicScheduler(self.graph) self.orchestrator TaskOrchestrator(self.graph, self.scheduler) def register(self, task) - None: self.graph.add_task(task) def run(self, payload: Dict[str, Any]) - Dict[str, Any]: return self.orchestrator.run(payload)5.5 编写示例任务为了验证调度器与编排器的效果我们设计四个模拟任务extract_user_info从输入 payload 中提取用户信息check_quota检查用户配额依赖 extract_user_infoenrich_user_data补充用户数据依赖 extract_user_infocompose_response聚合前序结果生成最终输出依赖 check_quota 和 enrich_user_data。# demo.py from harness import Harness from task_graph import TaskNode def extract_user_info(inputs): payload inputs[payload] user_id payload.get(user_id, unknown) return {user_id: user_id, source: request} def check_quota(inputs): results inputs[context][results] user_id results[extract_user_info][user_id] # 模拟固定规则检查配额 return {quota: 100, user_id: user_id} def enrich_user_data(inputs): results inputs[context][results] user_id results[extract_user_info][user_id] return {level: gold if user_id.startswith(u) else normal, user_id: user_id} def compose_response(inputs): results inputs[context][results] quota_info results[check_quota] user_info results[enrich_user_data] return { final: { user_id: user_info[user_id], level: user_info[level], quota: quota_info[quota], } } def build_harness() - Harness: harness Harness() harness.register( TaskNode( nameextract_user_info, fnextract_user_info, depends_on[], ) ) harness.register( TaskNode( namecheck_quota, fncheck_quota, depends_on[extract_user_info], ) ) harness.register( TaskNode( nameenrich_user_data, fnenrich_user_data, depends_on[extract_user_info], ) ) harness.register( TaskNode( namecompose_response, fncompose_response, depends_on[check_quota, enrich_user_data], ) ) return harness if __name__ __main__: h build_harness() payload_1 {user_id: u001} result_1 h.run(payload_1) print( 执行计划 ) print(h.scheduler.build_plan(payload_1)) print( 运行轨迹 ) for line in result_1[trace]: print(line) print( 最终结果 ) print(result_1[results][compose_response])运行方式python demo.py预期输出类似 执行计划 [extract_user_info, check_quota, enrich_user_data, compose_response] 运行轨迹 {task: extract_user_info, task_id: f3b0c5d8e1a24b10, status: running} {task: extract_user_info, task_id: f3b0c5d8e1a24b10, status: succeeded} {task: check_quota, task_id: 6a2b0c7f5d9e3a11, status: running} {task: check_quota, task_id: 6a2b0c7f5d9e3a11, status: succeeded} {task: enrich_user_data, task_id: d4c1e9d3b7a2f820, status: running} {task: enrich_user_data, task_id: d4c1e9d3b7a2f820, status: succeeded} {task: compose_response, task_id: a1f2e0c9d6b3a415, status: running} {task: compose_response, task_id: a1f2e0c9d6b3a415, status: succeeded} 最终结果 {final: {user_id: u001, level: gold, quota: 100}}注意示例中的 task_id 只是用于说明格式实际值会因为环境差异而不同。关键在于同一份代码、同一个 payload多次运行时 task_id 保持一致。6. 运行结果验证确定性的实证为了验证调度器确实具备确定性可以在 demo.py 中加入一个简单的重复实验连续运行两次比较执行计划和 trace。# verify_determinism.py from demo import build_harness def main(): h build_harness() payload {user_id: u123} plan_1 h.scheduler.build_plan(payload) result_1 h.run(payload) plan_2 h.scheduler.build_plan(payload) result_2 h.run(payload) print(计划一致:, plan_1 plan_2) trace_1 [(x[task], x[task_id], x[status]) for x in result_1[trace]] trace_2 [(x[task], x[task_id], x[status]) for x in result_2[trace]] print(轨迹一致:, trace_1 trace_2) print(结果一致:, result_1[results] result_2[results]) if __name__ __main__: main()运行后三条输出全部为 True说明在当前 harness 实现中调度和编排具备确定性。7. 常见问题与排查思路实际工程项目里把“确定性调度 任务编排”落地时会遇到不少问题。下面整理几个典型场景。问题现象常见原因解决思路相同输入两次执行计划不一样调度时依赖了 Python 字典的插入顺序或使用了 set/list 无序去重对 ready 队列做稳定排序排序 key 使用任务名或 task_id 的字典序任务执行顺序正确但 trace 不可复现任务 ID 用自增数字或时间戳生成任务 ID 改为输入哈希 任务名保证可复现编排器并发执行后结果不稳定多个无依赖任务并发写同一个共享变量默认单线程按计划执行必须并发时使用线程池 确定性归并策略重试后上下文被污染异常处理时把部分结果写入了 context任务执行前保存上下文快照失败后恢复快照再重试外部 API 超时导致整个流水线失败没有给任务设置超时或降级策略在 TaskNode 中增加 timeout_seconds超时执行 fallback 分支任务结果包含时间戳导致输出不一致任务函数内部调用 time.time() 或 datetime.now()显式注入运行时间或把时间类信息放入额外字段不参与主结果日志顺序和实际执行顺序不一致使用了异步日志或缓冲日志同步写日志并在 trace 中记录 task_id 和状态如果你在处理真实 harness 项目时遇到类似问题建议按照“先复现再定位再修复”的顺序排查。第一步永远是构造一个最小可复现示例然后把调度计划和执行 trace 打印出来对照。8. 工程化最佳实践与设计建议8.1 尽量让任务函数保持纯函数风格在设计 harness 任务时最推荐的做法是让每个任务函数只依赖 inputs 参数而不要直接读取全局变量、环境变量或文件。这样做的好处是任务可测试、可缓存、可重放。如果必须读取外部配置请在 harness 初始化阶段注入而不是在任务里临时读。8.2 任务 ID 的生成规范任务 ID 是确定性的基础。建议的生成规则是task_id SHA256(task_name normalized_input_payload)normalized_input_payload 要求所有对象按键名排序保证字典顺序稳定。这个任务 ID 可以用于日志、数据库主键、分布式锁的 key能极大提升排错效率。8.3 把状态快照纳入编排协议确定性不等于“永远不出错”而是“出错也能定位”。建议在 TaskOrchestrator 中引入状态快照机制每次任务执行前保存当前 context 的深拷贝任务失败时可以从快照恢复。本文示例只是最简实现生产环境可以把快照序列化到本地文件或对象存储。8.4 区分“调度计划”和“执行过程”调度计划是静态的、可打印的执行过程是动态的、有副作用的。工程上要尽量把两者拆开。这样你可以先审查计划再执行任务甚至做“计划对比”比较两个版本的任务图在相同输入下是否会产生相同计划。8.5 并发与确定性之间的取舍当任务数量很大时串行执行可能太慢。如果想引入并行我的建议是采用“确定性分片”方案把无依赖的任务按固定策略分成多个批次每个批次内部串行批次之间并行。每个批次的任务集合由调度器预先算出而不是执行中动态决定。这样可以保持计划的可复现性同时获得一定的并发收益。8.6 日志和审计日志一定要结构化成 JSON至少包含以下字段task_nametask_idstatusstart_time / end_timeinput 摘要output 摘要error_message如果有。有了结构化日志再配合 trace就能回答“这次结果是怎么来的”这个核心问题。8.7 安全边界如果 harness 要面向生产环境建议注意harness 本身不直接暴露可执行任意命令的接口除非有完善的鉴权任务图配置必须经过校验防止循环依赖或非法依赖输入 payload 大小和任务执行时长要做上限控制涉及外部资源数据库、文件、网络时遵循最小权限原则。8.8 可测试性设计确定性的另一个好处是方便测试。你可以为每个任务准备一份固定输入断言输出完全相等然后为整个 harness 准备一份端到端 golden 文件包含预期的执行计划和结果。只要代码改动后跑一遍对比就能快速发现行为变化。9. 总结与下一步学习回到最初的问题harness 能拥有一个真正的确定性调度器和任务编排器吗答案是可以而且二者并不冲突。关键在三点任务图负责描述依赖关系确定性调度器负责生成稳定计划任务编排器负责按计划推进状态同时用固定规则处理重试和失败分支。调度与执行分离决策与副作用分离是这套设计能够成立的根本原因。如果你想进一步深入可以尝试以下方向在 TaskGraph 中加入条件分支节点让编排器根据前序结果选择后续路径同时保持规则确定性为任务增加超时控制和优雅终止机制引入持久化队列让 harness 支持断点恢复研究如何把任务图配置化用 JSON/YAML 描述任务依赖而不是写死在代码里对比不同语言生态下的工作流引擎实现比如 Java 领域的 Spring Batch、Python 领域的 Prefect 等思考它们如何平衡确定性与动态性。最后再给一个实用建议如果你自己动手写 harness先不要急着做并发和动态分支。先把“稳定计划 完整 trace 可重放执行”做扎实这三点到位后再考虑性能优化。一个可复现的慢系统远比一个无法复现的快系统更容易演进和维护。希望这篇笔记对你有帮助也欢迎在评论区分享你在 harness 调度与编排方面的实践经验。
Harness确定性调度与任务编排：架构设计与Python实践
前段时间在搭建一个面向 LLM Agent 的自动化运行环境时我被一个问题卡了很久harness 到底能不能同时做到“确定性调度”和“任务编排”网上关于 harness 的资料大多停留在“怎么安装”“怎么启动工作区”很少有人把调度器与任务编排器之间的边界讲清楚更少有人讨论如何在真实代码里把这两个能力组合起来。今天这篇文章就围绕这个问题做一个完整拆解包含概念梳理、架构设计、可运行的 Python 示例以及工程落地时容易踩的坑。无论你是刚接触 harness 的开发者还是正在设计内部任务编排系统这篇文章应该都能提供一些可以直接复用的思路。1. 背景与核心概念1.1 什么是 Harness在软件工程领域harness 这个词并没有严格的统一翻译它通常指“为了控制、观测、驱动某个系统运行而包裹在系统外部的一层可编程框架”。你可以把它理解成“运行控制壳”它不只负责启动一个模型、一个脚本或一组任务还负责决定这些任务以什么顺序运行、在什么条件下重试、把哪些输入传给哪些阶段以及最终把结果汇总成什么结构。最近经常听到的 DeepSeek Harness、Codex Harness、Cursor Harness 等名词本质上都是把某个底层模型或工具包进一个可编程的运行时环境中。harness 的职责边界通常包括定义任务的输入输出协议维护任务间的依赖关系控制任务执行顺序与并发策略记录运行轨迹与状态快照在失败时执行重试或回退策略。所以讨论“harness 能否拥有一个真正确定性的调度器和任务编排器”其实是在问我们能否在 harness 内部构造一套既稳定可复现、又具备完整编排能力的基础设施。1.2 什么是 Deterministic SchedulerDeterministic Scheduler即确定性调度器指的是在给定相同输入和相同任务集合的情况下调度器每次产生的任务执行计划都完全一致。这里的关键并不只是“最终结果一致”而是“过程一致”任务顺序、中间状态、重试次数、日志输出都应当可复现。确定性调度器通常需要满足以下约束调度决策不依赖系统当前时间调度决策不依赖随机数调度决策不依赖并发执行时的竞态条件调度决策只依赖任务定义和输入数据的规范化表示。举个例子同样一组任务 A、B、C如果 A 不依赖 BB 不依赖 C普通调度器可能这次先执行 B下次先执行 A但确定性调度器必须通过某种稳定规则比如任务 ID 字典序选定唯一顺序。1.3 什么是 Task OrchestratorTask Orchestrator即任务编排器负责管理一个任务从开始到结束的完整生命周期。编排器不仅要调度任务顺序还要处理上下文传递前一个任务的输出如何变成后一个任务的输入状态维护每个任务处于 pending、running、succeeded、failed 中的哪个状态执行策略是否重试、是否超时、是否跳过数据汇总最终输出的结构由哪些任务的输出拼接而成。编排器更像一个“工作流引擎”而调度器更关注“顺序与资源分配”。两者确实可以分离也可以集成在同一个 harness 中。1.4 为什么二者可以同时存在很多人会把“确定性”和“编排”看成对立关系理由是真实业务中总有一些任务需要动态决策如果上一个任务失败了下一步要执行另一个分支这听起来就像“不固定”。但这里有一个关键区分动态决策可以使用确定性的决策规则。比如“失败后重试 3 次”“失败后走 fallback 分支”这些规则是确定的。harness 完全可以用固定规则驱动动态流程从而做到既能编排又有确定性。2. 从“能跑”到“可复现”harness 的设计目标2.1 普通任务执行器和 harness 的区别普通任务执行器通常就是“按顺序调用函数”谈不上调度。harness 的复杂度在于它需要面对多个任务、多种输入、多种失败可能性还要保证运行过程可以被审计和回放。一个合格的 harness 应该具备四个能力可编程任务不是写死在 if-else 里的而是可以被声明、注册、组合可观测每一步执行都应该有记录能回答“这个结果是怎么算出来的”可复现相同输入在相同 harness 版本下产生相同运行轨迹可恢复单点失败不影响整体且失败策略明确。2.2 为什么需要确定性在 AI Agent 和 LLM 应用场景下确定性尤其重要。模型本身带有随机性如果外层 harness 的调度也是随机的问题排查会变得极其困难。今天调用同一个 Prompt 得到的结果和昨天不一样可能是模型温度参数导致的也可能是任务执行顺序变化导致的。如果 harness 具备确定性调度至少可以把“业务层波动”和“基础设施层波动”分开。对于 CI/CD、数据处理、自动化测试等场景确定性调度同样关键。一次流水线运行完成后如果无法按原样重放问题定位就会变成“猜谜”。2.3 确定性的边界需要注意确定性调度并不是“物理世界的绝对确定”而是“策略层面的可复现”。外部 API 返回什么、数据库里是否存在某个字段这些因素仍然会影响任务结果。但 harness 能保证的是在相同外部条件下任务的执行路径和顺序是稳定的。理解这个边界很重要否则会陷入一个误区认为只要做了确定性调度所有结果都能复现。实际项目中我们通常会通过日志、状态快照、输入哈希等方式记录外部依赖的实际值用来辅助复现。3. 架构设计把调度与执行分离3.1 核心组件划分在实现一个带确定性调度与任务编排能力的 harness 时我最推荐的方式是拆成四个独立的组件TaskGraph任务图定义任务节点和依赖关系DeterministicScheduler确定性调度器根据任务图和输入生成稳定执行计划TaskOrchestrator任务编排器负责执行计划的具体推进、状态维护和上下文传递Harness对外统一入口把上述组件组合起来。下面用一张简化的职责表来表示组件核心职责是否产生决策决策是否确定性TaskGraph声明任务与依赖否-DeterministicScheduler生成执行计划是是TaskOrchestrator推进状态、执行任务部分失败分支等通过规则保证Harness组合装配否-3.2 为什么要用 DAG任务编排的基础是任务依赖关系。最稳定、最容易推理的依赖模型是有向无环图DAG。每个任务是一个节点节点之间的箭头表示依赖关系。只有所有前置依赖都成功完成后后置任务才会启动。DAG 有三个优点无循环依赖天然避免死锁可以稳定地做拓扑排序便于可视化、审计和测试。3.3 确定性策略怎么落到架构里我把确定性策略拆成三条具体规则节点 ID 必须可计算每个任务节点使用“输入哈希 任务名”生成稳定 ID同层顺序必须稳定拓扑排序后同一批次可执行的任务按 ID 字典序排序执行决策不带随机副作用调度器内部不调用 random、time、uuid 等函数。这三条规则能保证只要输入 payload 和任务定义不变调度器生成的执行计划就永远不变。4. 环境准备与项目结构本文示例使用 Python 标准库实现不需要额外依赖凡是 Python 3.9 以上的环境都可以直接运行。仍然建议先准备好一个虚拟环境mkdir deterministic-harness cd deterministic-harness python3 -m venv venv source venv/bin/activate示例项目的文件结构如下deterministic-harness/ ├── task_graph.py ├── scheduler.py ├── orchestrator.py ├── harness.py └── demo.py每个文件的职责很清晰下面逐一实现。5. 完整代码实现5.1 任务图定义先定义任务节点。每个任务节点包含四个关键字段任务名、执行函数、依赖列表、最大重试次数。# task_graph.py from __future__ import annotations import hashlib import json from dataclasses import dataclass, field from typing import Any, Callable, Dict, List, Optional dataclass class TaskNode: name: str fn: Callable[[Dict[str, Any]], Dict[str, Any]] depends_on: List[str] field(default_factorylist) max_retries: int 0 timeout_seconds: float 30.0 class TaskGraph: def __init__(self) - None: self._tasks: Dict[str, TaskNode] {} def add_task(self, task: TaskNode) - None: for dep in task.depends_on: if dep not in self._tasks: raise ValueError(f依赖任务 {dep} 尚未注册) self._tasks[task.name] task def get_task(self, name: str) - TaskNode: return self._tasks[name] def all_tasks(self) - List[TaskNode]: return list(self._tasks.values()) def dependencies_of(self, name: str) - List[str]: return self._tasks[name].depends_on def compute_task_id(self, payload: Dict[str, Any], task_name: str) - str: normalized json.dumps(payload, sort_keysTrue, ensure_asciiFalse) raw f{task_name}:{normalized} return hashlib.sha256(raw.encode(utf-8)).hexdigest()[:16]TaskGraph 在注册任务时会检查依赖是否已经存在这样可以尽早发现配置错误。compute_task_id 方法体现了确定性原则它不依赖时间、不依赖随机数只依赖任务名和规范化后的输入 payload。5.2 确定性调度器调度器负责把任务图转换成稳定的执行计划。这里采用经典拓扑排序思路并用队列存储当前可执行任务。为了保证确定性每次从队列取出任务时都按任务 ID 排序。# scheduler.py from typing import Any, Dict, List, Set from task_graph import TaskGraph class DeterministicScheduler: def __init__(self, graph: TaskGraph) - None: self.graph graph def build_plan(self, payload: Dict[str, Any]) - List[str]: tasks self.graph.all_tasks() in_degree: Dict[str, int] {} for task in tasks: in_degree[task.name] len(task.depends_on) ready: List[str] [] for task in tasks: if in_degree[task.name] 0: ready.append(task.name) plan: List[str] [] executed: Set[str] set() while ready: # 关键点按确定性 task_id 排序 ready.sort(keylambda name: self.graph.compute_task_id(payload, name)) current ready.pop(0) plan.append(current) executed.add(current) for task in tasks: if task.name in executed: continue if current in task.depends_on: in_degree[task.name] - 1 if in_degree[task.name] 0: ready.append(task.name) if len(plan) ! len(tasks): raise RuntimeError(任务图中存在循环依赖无法生成执行计划) return plan这个调度器的核心逻辑是先统计每个任务的入度入度为 0 的任务进入 ready 队列每次从 ready 队列中按 task_id 排序后取出一个任务执行后更新依赖它的任务入度重复直到所有任务都进入 plan。由于排序 key 使用 compute_task_id而 compute_task_id 只依赖 payload 和任务名所以相同 payload 生成的计划一定一样。5.3 任务编排器编排器负责真正执行任务并管理上下文和状态。# orchestrator.py import time import traceback from typing import Any, Dict, List from scheduler import DeterministicScheduler from task_graph import TaskGraph class TaskOrchestrator: def __init__(self, graph: TaskGraph, scheduler: DeterministicScheduler) - None: self.graph graph self.scheduler scheduler def run(self, payload: Dict[str, Any]) - Dict[str, Any]: plan: List[str] self.scheduler.build_plan(payload) context: Dict[str, Any] { payload: payload, results: {}, trace: [], } for task_name in plan: task self.graph.get_task(task_name) task_id self.graph.compute_task_id(payload, task_name) # 准备输入把前置任务的结果合并进当前上下文 inputs {payload: payload, context: context} attempt 0 last_error: Exception | None None while attempt task.max_retries: try: self._mark(context, task_name, running, task_id) output task.fn(inputs) context[results][task_name] output self._mark(context, task_name, succeeded, task_id) break except Exception as exc: # noqa: BLE001 attempt 1 last_error exc self._mark(context, task_name, failed, task_id) if attempt task.max_retries: time.sleep(0.01) # 固定重试间隔不引入随机等待 else: raise RuntimeError( f任务 {task_name} 失败重试 {task.max_retries} 次后终止 ) from exc return context staticmethod def _mark(context: Dict[str, Any], task_name: str, status: str, task_id: str) - None: context[trace].append( { task: task_name, task_id: task_id, status: status, } )编排器中的几个设计要点context 同时保存 payload、每个任务的结果和完整 trace方便审计重试策略也保持确定性固定次数、固定间隔每次执行前都重新计算 task_id确保 trace 中的 ID 可复现。5.4 Harness 统一入口Harness 只是把 TaskGraph、DeterministicScheduler、TaskOrchestrator 组合起来暴露一个 run 方法。# harness.py from typing import Any, Dict from orchestrator import TaskOrchestrator from scheduler import DeterministicScheduler from task_graph import TaskGraph class Harness: def __init__(self) - None: self.graph TaskGraph() self.scheduler DeterministicScheduler(self.graph) self.orchestrator TaskOrchestrator(self.graph, self.scheduler) def register(self, task) - None: self.graph.add_task(task) def run(self, payload: Dict[str, Any]) - Dict[str, Any]: return self.orchestrator.run(payload)5.5 编写示例任务为了验证调度器与编排器的效果我们设计四个模拟任务extract_user_info从输入 payload 中提取用户信息check_quota检查用户配额依赖 extract_user_infoenrich_user_data补充用户数据依赖 extract_user_infocompose_response聚合前序结果生成最终输出依赖 check_quota 和 enrich_user_data。# demo.py from harness import Harness from task_graph import TaskNode def extract_user_info(inputs): payload inputs[payload] user_id payload.get(user_id, unknown) return {user_id: user_id, source: request} def check_quota(inputs): results inputs[context][results] user_id results[extract_user_info][user_id] # 模拟固定规则检查配额 return {quota: 100, user_id: user_id} def enrich_user_data(inputs): results inputs[context][results] user_id results[extract_user_info][user_id] return {level: gold if user_id.startswith(u) else normal, user_id: user_id} def compose_response(inputs): results inputs[context][results] quota_info results[check_quota] user_info results[enrich_user_data] return { final: { user_id: user_info[user_id], level: user_info[level], quota: quota_info[quota], } } def build_harness() - Harness: harness Harness() harness.register( TaskNode( nameextract_user_info, fnextract_user_info, depends_on[], ) ) harness.register( TaskNode( namecheck_quota, fncheck_quota, depends_on[extract_user_info], ) ) harness.register( TaskNode( nameenrich_user_data, fnenrich_user_data, depends_on[extract_user_info], ) ) harness.register( TaskNode( namecompose_response, fncompose_response, depends_on[check_quota, enrich_user_data], ) ) return harness if __name__ __main__: h build_harness() payload_1 {user_id: u001} result_1 h.run(payload_1) print( 执行计划 ) print(h.scheduler.build_plan(payload_1)) print( 运行轨迹 ) for line in result_1[trace]: print(line) print( 最终结果 ) print(result_1[results][compose_response])运行方式python demo.py预期输出类似 执行计划 [extract_user_info, check_quota, enrich_user_data, compose_response] 运行轨迹 {task: extract_user_info, task_id: f3b0c5d8e1a24b10, status: running} {task: extract_user_info, task_id: f3b0c5d8e1a24b10, status: succeeded} {task: check_quota, task_id: 6a2b0c7f5d9e3a11, status: running} {task: check_quota, task_id: 6a2b0c7f5d9e3a11, status: succeeded} {task: enrich_user_data, task_id: d4c1e9d3b7a2f820, status: running} {task: enrich_user_data, task_id: d4c1e9d3b7a2f820, status: succeeded} {task: compose_response, task_id: a1f2e0c9d6b3a415, status: running} {task: compose_response, task_id: a1f2e0c9d6b3a415, status: succeeded} 最终结果 {final: {user_id: u001, level: gold, quota: 100}}注意示例中的 task_id 只是用于说明格式实际值会因为环境差异而不同。关键在于同一份代码、同一个 payload多次运行时 task_id 保持一致。6. 运行结果验证确定性的实证为了验证调度器确实具备确定性可以在 demo.py 中加入一个简单的重复实验连续运行两次比较执行计划和 trace。# verify_determinism.py from demo import build_harness def main(): h build_harness() payload {user_id: u123} plan_1 h.scheduler.build_plan(payload) result_1 h.run(payload) plan_2 h.scheduler.build_plan(payload) result_2 h.run(payload) print(计划一致:, plan_1 plan_2) trace_1 [(x[task], x[task_id], x[status]) for x in result_1[trace]] trace_2 [(x[task], x[task_id], x[status]) for x in result_2[trace]] print(轨迹一致:, trace_1 trace_2) print(结果一致:, result_1[results] result_2[results]) if __name__ __main__: main()运行后三条输出全部为 True说明在当前 harness 实现中调度和编排具备确定性。7. 常见问题与排查思路实际工程项目里把“确定性调度 任务编排”落地时会遇到不少问题。下面整理几个典型场景。问题现象常见原因解决思路相同输入两次执行计划不一样调度时依赖了 Python 字典的插入顺序或使用了 set/list 无序去重对 ready 队列做稳定排序排序 key 使用任务名或 task_id 的字典序任务执行顺序正确但 trace 不可复现任务 ID 用自增数字或时间戳生成任务 ID 改为输入哈希 任务名保证可复现编排器并发执行后结果不稳定多个无依赖任务并发写同一个共享变量默认单线程按计划执行必须并发时使用线程池 确定性归并策略重试后上下文被污染异常处理时把部分结果写入了 context任务执行前保存上下文快照失败后恢复快照再重试外部 API 超时导致整个流水线失败没有给任务设置超时或降级策略在 TaskNode 中增加 timeout_seconds超时执行 fallback 分支任务结果包含时间戳导致输出不一致任务函数内部调用 time.time() 或 datetime.now()显式注入运行时间或把时间类信息放入额外字段不参与主结果日志顺序和实际执行顺序不一致使用了异步日志或缓冲日志同步写日志并在 trace 中记录 task_id 和状态如果你在处理真实 harness 项目时遇到类似问题建议按照“先复现再定位再修复”的顺序排查。第一步永远是构造一个最小可复现示例然后把调度计划和执行 trace 打印出来对照。8. 工程化最佳实践与设计建议8.1 尽量让任务函数保持纯函数风格在设计 harness 任务时最推荐的做法是让每个任务函数只依赖 inputs 参数而不要直接读取全局变量、环境变量或文件。这样做的好处是任务可测试、可缓存、可重放。如果必须读取外部配置请在 harness 初始化阶段注入而不是在任务里临时读。8.2 任务 ID 的生成规范任务 ID 是确定性的基础。建议的生成规则是task_id SHA256(task_name normalized_input_payload)normalized_input_payload 要求所有对象按键名排序保证字典顺序稳定。这个任务 ID 可以用于日志、数据库主键、分布式锁的 key能极大提升排错效率。8.3 把状态快照纳入编排协议确定性不等于“永远不出错”而是“出错也能定位”。建议在 TaskOrchestrator 中引入状态快照机制每次任务执行前保存当前 context 的深拷贝任务失败时可以从快照恢复。本文示例只是最简实现生产环境可以把快照序列化到本地文件或对象存储。8.4 区分“调度计划”和“执行过程”调度计划是静态的、可打印的执行过程是动态的、有副作用的。工程上要尽量把两者拆开。这样你可以先审查计划再执行任务甚至做“计划对比”比较两个版本的任务图在相同输入下是否会产生相同计划。8.5 并发与确定性之间的取舍当任务数量很大时串行执行可能太慢。如果想引入并行我的建议是采用“确定性分片”方案把无依赖的任务按固定策略分成多个批次每个批次内部串行批次之间并行。每个批次的任务集合由调度器预先算出而不是执行中动态决定。这样可以保持计划的可复现性同时获得一定的并发收益。8.6 日志和审计日志一定要结构化成 JSON至少包含以下字段task_nametask_idstatusstart_time / end_timeinput 摘要output 摘要error_message如果有。有了结构化日志再配合 trace就能回答“这次结果是怎么来的”这个核心问题。8.7 安全边界如果 harness 要面向生产环境建议注意harness 本身不直接暴露可执行任意命令的接口除非有完善的鉴权任务图配置必须经过校验防止循环依赖或非法依赖输入 payload 大小和任务执行时长要做上限控制涉及外部资源数据库、文件、网络时遵循最小权限原则。8.8 可测试性设计确定性的另一个好处是方便测试。你可以为每个任务准备一份固定输入断言输出完全相等然后为整个 harness 准备一份端到端 golden 文件包含预期的执行计划和结果。只要代码改动后跑一遍对比就能快速发现行为变化。9. 总结与下一步学习回到最初的问题harness 能拥有一个真正的确定性调度器和任务编排器吗答案是可以而且二者并不冲突。关键在三点任务图负责描述依赖关系确定性调度器负责生成稳定计划任务编排器负责按计划推进状态同时用固定规则处理重试和失败分支。调度与执行分离决策与副作用分离是这套设计能够成立的根本原因。如果你想进一步深入可以尝试以下方向在 TaskGraph 中加入条件分支节点让编排器根据前序结果选择后续路径同时保持规则确定性为任务增加超时控制和优雅终止机制引入持久化队列让 harness 支持断点恢复研究如何把任务图配置化用 JSON/YAML 描述任务依赖而不是写死在代码里对比不同语言生态下的工作流引擎实现比如 Java 领域的 Spring Batch、Python 领域的 Prefect 等思考它们如何平衡确定性与动态性。最后再给一个实用建议如果你自己动手写 harness先不要急着做并发和动态分支。先把“稳定计划 完整 trace 可重放执行”做扎实这三点到位后再考虑性能优化。一个可复现的慢系统远比一个无法复现的快系统更容易演进和维护。希望这篇笔记对你有帮助也欢迎在评论区分享你在 harness 调度与编排方面的实践经验。
