最近在整理 Harness 类工具社区里讨论比较多的 DeepSeek Harness、Codex Harness 都属于这个方向时收到一个很有意思的问题一个 Harness 内部能不能设计出真正确定性的调度器和任务编排器这个问题看起来偏基础实际做起来并不简单。真正常见的落地情况是任务一多执行顺序开始飘并发一开结果对不上LLM 调用一进来日志混乱到没法复现。很多开发者的第一反应是“加个 seed、设 temperature0”但真正要解决的是调度层和编排层的设计问题。这篇文章会从概念拆解开始讲清楚确定性调度器、任务编排器和 Harness 之间的关系再给出一套可复现的最小实现最后聊一聊在 DeepSeek Harness / Codex Harness 这类工具落地时的工程建议。适合正在设计 Agent 工具链、工作流框架或者想深入理解 Harness 架构的开发者。1. 背景什么是 Harness为什么调度与编排是关键问题1.1 Harness 是什么Harness 的本义是“线束、防护装置”在 AI Agent 领域它通常指包裹在模型之外的一整套执行框架接收输入、组织上下文、调用工具、管理对话历史、调度多个子任务最终把模型能力安全地暴露给上层应用。你可以把 Harness 理解成一个“Agent 的驾驶舱”。模型只是发动机Harness 负责挂挡、转向、踩刹车。这也是为什么社区里会出现deepseek harness、codex harness这类工具——它们做的事情本质上是把模型能力包装成工程化的执行体系。一个完整的 Harness 通常包含上下文管理模块工具调用模块Tool Calling对话归档与工作区管理任务调度与编排模块插件扩展机制其中调度与编排模块决定了多个任务什么时候执行、按什么顺序执行、失败后怎么办是整个系统能否稳定工作的核心。1.2 为什么调度与编排是 Harness 的核心问题在早期 Agent 应用中我们通常只有一个模型、一次请求、一次响应调度问题并不突出。但随着 Harness 类工具逐渐走向本地部署和桌面化任务开始变多一个任务需要先收集资料再预处理再调用大模型最后归档结果多个任务可能存在依赖关系比如 B 必须等 A 完成后才能开始部分任务可能并发执行以提高整体吞吐某些 LLM 调用失败后需要重试但重试不能导致状态错乱。如果调度器不稳定任务可能“随机”失败如果编排器没有清晰的状态机依赖关系和重试逻辑就很难写。所以理解 Harness 的调度与编排设计是做好 Agent 工程化的基础。1.3 确定性问题从哪里冒出来的“确定性”这个词在不同的书里有不同含义但在 Harness 语境下它通常覆盖两个层面同一次任务每次运行结果一致同样的输入、同样的代码版本、同样的模型配置执行顺序和输出结果可复现。失败重试后状态可恢复即使某一步崩溃也能从历史状态继续而不是从头开始或者产生脏数据。真正难以做到的往往是第一点。LLM 自带随机性外部 API 返回有波动并发环境下日志顺序和任务完成顺序不可控这些都让“确定性”看起来像一个遥不可及的目标。2. 核心概念调度器、编排器与确定性的边界2.1 调度器Scheduler解决的是“时机”问题调度器回答的是“什么时候该执行哪个任务”。它负责从待执行队列里挑选任务决定执行顺序并管理并发度。在 Harness 中调度器通常需要处理优先级排序依赖就绪判断超时控制重试策略并发控制一个最简单的调度器可以是一个for循环按顺序执行所有任务。但在复杂场景下调度器需要感知依赖图否则就会出现“任务 B 在任务 A 之前执行”的尴尬情况。2.2 编排器Orchestrator解决的是“协作”问题编排器更关注任务之间的依赖关系、状态流转和全局一致性。比如A 完成后触发 B 和 CB 和 C 都完成后触发 D如果 B 失败则进入回滚流程每一步的状态都要记录下来支持断点续跑。一个健壮的任务编排器本质上是一个状态机管理器。每个任务都有PENDING - READY - RUNNING - SUCCEEDED / FAILED这样的状态流转编排器根据状态决定下一步动作。2.3 确定性的边界调度层 vs 执行层我们需要分清一个界限调度器可以做到确定性但执行结果不一定能完全确定。调度层确定性任务入队顺序稳定、优先级判断稳定、状态流转稳定不会因为并发产生乱序。执行层确定性LLM 输出稳定、外部 API 返回稳定。这个层次很难 100% 保证。换句话说Harness 可以拥有“真正确定性的调度器”但执行层只能做到“尽力确定性”。设计良好的 Harness应该把这两层分开调度层严格确定性执行层通过快照、缓存、固定参数等手段降低随机性。3. 为什么真正的确定性这么难3.1 LLM 输出的天然随机性即使设置temperature0大模型输出也可能因为采样实现、浮点计算、并发批次等因素出现细微差异。这意味着同一个 prompt 多次调用结果可能不完全一致。如果 Harness 调度器把“比较任务输出”作为下一步决策的依据那么随机性就会被放大。比如“判断前一任务输出是否包含某个关键词再决定走哪个分支”一旦输出抖动整个流程就会不稳定。3.2 外部依赖的不稳定性Harness 经常需要调用外部工具联网搜索、数据库查询、本地命令、第三方 API。外部服务的响应时间、返回内容都可能变化这种变化很难从 Harness 内部完全屏蔽。如果任务编排器把外部 API 的响应时间当作隐式排序依据那么日志顺序和任务完成顺序就无法复现。3.3 并发与竞态条件多个任务并发执行时如果它们共享同一个全局变量、同一个文件、同一个数据库记录就会出现竞态条件。结果是谁能先拿到锁、谁能先写入数据取决于操作系统线程调度而不是业务逻辑本身。这类问题在本地调试时很难发现一旦进入桌面版或服务端部署并发度上升问题就会频繁出现。3.4 状态管理混乱Harness 运行过程中任务会产生大量中间状态临时文件、缓存、内存对象、数据库记录。如果状态没有被统一管理重试时可能读到脏数据也可能覆盖已完成任务的结果。状态管理混乱也是导致“重试后结果不一致”的主要元凶之一。4. 实现确定性调度的架构思路回到标题的问题Can a harness have a true deterministic scheduler and task orchestrator?答案是可以但需要在架构上做出取舍。下面是我在实践中验证过的几个思路。4.1 把调度器设计成纯函数调度器不应直接操作“任务执行”而应该只负责计算“执行顺序”。输入是任务依赖图输出是一个有序的任务 ID 列表。这个计算过程不访问网络、不读取随机数、不依赖系统时间因此天然确定。这样设计的好处是调度逻辑可以被单测直接覆盖同一个依赖图在任何机器上得到相同顺序方便把调度结果持久化用于复盘和断点恢复。4.2 用稳定排序消除无序Python 的set遍历顺序不稳定字典在旧版本中的遍历顺序也不稳定。只要有集合参与迭代就可能引入不确定性。所以调度器内部必须统一使用“可排序的 key”例如任务 ID、优先级、创建时间戳。稳定排序的规则可以是先按优先级排序同优先级按任务 ID 字典序排序同任务 ID 的依赖关系按定义顺序执行。这样即使并发环境也能保证调度顺序可复现。4.3 把执行状态显式建模编排器应该使用显式状态机而不是靠“任务是否已经跑过”来推断状态。推荐定义统一的状态枚举并把每次状态变更写入事件日志。class TaskState(Enum): PENDING 0 READY 1 RUNNING 2 SUCCEEDED 3 FAILED 4 SKIPPED 5状态变更时只允许从合法的上游状态迁移到下游状态。这样即使并发环境也可以通过加锁或队列保证状态流转的顺序。4.4 隔离随机性Harness 中哪些随机源会影响调度确定性计时器任务超时、延迟重试随机数random、numpy.random并发调度线程调度外部 API网络波动为了让调度层面确定可以把这些随机源隔离到执行层。调度器内部不调用time.sleep、不使用系统时钟作为唯一排序依据。如果必须等待也要用虚拟时间或固定步长。4.5 记录快照不只是日志为了实现可复现Harness 应该保存“输入快照”和“输出快照”。输入快照记录每个任务执行前收到的参数、环境变量、依赖任务结果。输出快照记录每个任务执行后的结果、耗时、异常信息。有了快照即使执行层存在随机性我们也可以回放执行过程定位是哪一步产生了偏差。5. 实战用 Python 构建一个最小确定性调度与编排系统下面我们实现一个最小但完整可运行的确定性调度器与任务编排器。核心是依赖解析、稳定排序、状态机流转、结果快照。5.1 项目结构deterministic-harness/ ├── task.py ├── scheduler.py ├── orchestrator.py ├── demo.py └── README.md5.2 定义任务模型文件task.pyfrom __future__ import annotations from dataclasses import dataclass, field from enum import Enum from typing import Any, Callable, Dict, List class TaskState(Enum): PENDING PENDING READY READY RUNNING RUNNING SUCCEEDED SUCCEEDED FAILED FAILED SKIPPED SKIPPED dataclass class Task: task_id: str fn: Callable[[], Any] deps: List[str] field(default_factorylist) priority: int 0 state: TaskState TaskState.PENDING result: Any None error: str def to_snapshot(self) - Dict[str, Any]: 输出该任务的可持久化快照。 return { task_id: self.task_id, deps: self.deps, priority: self.priority, state: self.state.value, result: self.result, error: self.error, }这里的关键点是把state定义为枚举而不是用布尔值表示“是否完成”。这样后续编排器可以精准判断依赖是否就绪。5.3 实现确定性调度器文件scheduler.pyfrom __future__ import annotations from typing import Dict, List from task import Task class DeterministicScheduler: 确定性调度器 1. 拓扑排序解决依赖关系 2. 使用稳定排序保证同一依赖图多次执行顺序一致。 def schedule(self, tasks: Dict[str, Task]) - List[str]: visited: set set() order: List[str] [] def visit(task_id: str) - None: if task_id in visited: return visited.add(task_id) task tasks[task_id] # 先访问依赖保证父任务先执行 for dep in task.deps: if dep in tasks: visit(dep) else: raise ValueError(ftask {task_id} 依赖不存在的任务 {dep}) order.append(task_id) # 按任务 ID 和优先级稳定遍历避免 set 顺序不稳定 for task_id in sorted(tasks.keys()): visit(task_id) return order调度器没有执行任何副作用它只是返回一个任务执行顺序。这样调度决策本身是纯函数天然具备确定性。5.4 实现任务编排器文件orchestrator.pyfrom __future__ import annotations from typing import Dict, List from scheduler import DeterministicScheduler from task import Task, TaskState class TaskOrchestrator: def __init__(self, tasks: Dict[str, Task]): self.tasks tasks self.scheduler DeterministicScheduler() self._execution_log: List[str] [] property def execution_log(self) - List[str]: return self._execution_log def run(self) - Dict[str, object]: order self.scheduler.schedule(self.tasks) results: Dict[str, object] {} for task_id in order: task self.tasks[task_id] # 检查依赖是否全部成功 if not self._deps_succeeded(task): task.state TaskState.SKIPPED self._execution_log.append(f{task_id}:SKIPPED) continue task.state TaskState.RUNNING self._execution_log.append(f{task_id}:RUNNING) try: task.result task.fn() task.state TaskState.SUCCEEDED self._execution_log.append(f{task_id}:SUCCEEDED) except Exception as exc: task.state TaskState.FAILED task.error str(exc) self._execution_log.append(f{task_id}:FAILED) # 实际项目中可以选择抛错或继续执行后续可独立运行的任务 raise results[task_id] task.result return results def _deps_succeeded(self, task: Task) - bool: for dep in task.deps: dep_task self.tasks.get(dep) if dep_task is None: return False if dep_task.state ! TaskState.SUCCEEDED: return False return True这里有两个容易踩的坑不能只判断“依赖有没有被访问过”要判断“依赖是否成功”。任务出现异常时要么中断整个流程要么进入降级分支不要静默吞掉异常。5.5 编写演示任务流文件demo.pyfrom orchestrator import TaskOrchestrator from task import Task def build_tasks(): # 模拟一个 Harness 工作流 # A 收集资料 - B 预处理 - C 并发完成(此处串行演示) - D 归档 tasks {} def task_a(): print([A] 收集资料完成) return {docs: [doc1, doc2]} def task_b(): print([B] 预处理完成) return {clean: True} def task_c(): print([C] 调用 LLM 完成) return {answer: hello harness} def task_d(): print([D] 归档对话完成) return {archived: True} tasks[A] Task(task_idA, fntask_a, priority1) tasks[B] Task(task_idB, fntask_b, deps[A], priority1) tasks[C] Task(task_idC, fntask_c, deps[B], priority1) tasks[D] Task(task_idD, fntask_d, deps[C], priority1) return tasks def main(): tasks build_tasks() orchestrator TaskOrchestrator(tasks) results orchestrator.run() print(\n 执行顺序日志 ) for log in orchestrator.execution_log: print(log) print(\n 执行结果 ) for task_id, result in results.items(): print(f{task_id}: {result}) if __name__ __main__: main()运行cd deterministic-harness python demo.py预期输出[A] 收集资料完成 [B] 预处理完成 [C] 调用 LLM 完成 [D] 归档对话完成 执行顺序日志 A:RUNNING A:SUCCEEDED B:RUNNING B:SUCCEEDED C:RUNNING C:SUCCEEDED D:RUNNING D:SUCCEEDED 执行结果 A: {docs: [doc1, doc2]} B: {clean: True} C: {answer: hello harness} D: {archived: True}多次运行上面的脚本执行顺序始终是A - B - C - D这就是调度层面确定性的直接体现。5.6 扩展加入并发与确定性如果你想在真实 Harness 中引入并发但又不破坏确定性可以这样改并发任务之间不共享可变状态每个任务只能消费上游任务的返回值并且把任务执行结果统一写入中心化结果表。# 伪代码展示思路 from concurrent.futures import ThreadPoolExecutor def run_concurrent(self, max_workers: int 4) - Dict[str, object]: order self.scheduler.schedule(self.tasks) results: Dict[str, object] {} with ThreadPoolExecutor(max_workersmax_workers) as pool: future_map {} for task_id in order: task self.tasks[task_id] # 提交前先判断依赖是否已经完成 if not self._deps_succeeded(task): task.state TaskState.SKIPPED continue future pool.submit(task.fn) future_map[future] task_id for future in futures.as_completed(future_map): task_id future_map[future] try: results[task_id] future.result() except Exception as exc: results[task_id] fERROR: {exc} return results但要注意并发执行结果的完成顺序不可控因此最终结果的组装顺序应该按order重新排序而不是按任务完成顺序。否则日志和结果列表都可能出现不确定性。6. Harness 场景下的落地配置从任务编排到工作区上面的 Python 示例是一个非常小的框架级实现。在实际的 DeepSeek Harness、Codex Harness 类项目中我们通常不会手写调度器而是利用工具自带的配置能力。6.1 工作区与任务配置很多 Harness 工具会提供一个“工作区Workspace”概念用来存放输入、中间产物、归档对话。一个基于 YAML 的示例任务配置可能长这样# workspace/tasks.yaml version: 1 deterministic: true seed: 42 tasks: collect: type: plugin plugin: collector params: source: local preprocess: type: plugin plugin: preprocessor deps: - collect ask_model: type: llm deps: - preprocess params: model: deepseek-chat temperature: 0 max_tokens: 2048 archive: type: plugin plugin: archiver deps: - ask_model这种配置的意义在于deterministic: true表示调度器开启确定性模式deps显式声明依赖避免隐式顺序temperature: 0是执行层的“降低随机性”手段。实际运行时Harness 会先解析依赖图再让调度器生成执行顺序最后逐个子任务执行并写入工作区。6.2 命令示例以社区讨论较多的 Harness 类工具为例本地启动命令通常类似# 安装依赖 pnpm install # 启动 Web 界面 pnpm dsh web如果你遇到类似命令建议先查看该项目的package.json或官方 README因为不同版本命令可能不同。重点是理解这类命令启动的完整系统内部应该有一个全局的调度器负责管理所有后台任务。6.3 插件隔离Harness 通常会提供插件机制让外部工具以插件形式接入。为了保护调度确定性插件应该遵循以下约定插件之间不能直接共享全局变量插件只能通过参数传入数据通过返回值输出数据插件内部不允许修改工作区的核心状态插件执行失败时必须返回结构化错误而不是抛出不透明异常。这样无论是内置任务还是第三方插件调度器都能以统一方式编排。7. 常见问题与排查思路在 Harness 类项目中使用确定性调度与任务编排时容易踩到下面这些坑。问题现象常见原因解决思路任务执行顺序每次不一致并发执行后按完成顺序组装结果按调度顺序重新排序不要使用完成顺序重试后结果不同共享状态被修改重试读到脏数据每个任务使用独立快照避免共享可变对象LLM 输出不稳定temperature 过高或未固定随机种子设置 temperature0固定 seed保留输入快照pnpm dsh web卡住依赖未装完或 pnpm 版本不一致先执行pnpm install清缓存重试锁定 pnpm 版本Windows 下启动失败Node 版本过旧、脚本权限受限使用 Node 长期支持版本检查 PowerShell 执行策略任务依赖判断不准只判断依赖“被访问过”没判断是否成功状态机必须显式区分 SUCCEEDED 和 FAILED归档对话找不到工作区路径或数据库位置未明确查看 Harness 配置中的工作区路径检查日志调度结果不好复现在调度器内部使用了系统时间或随机数将随机性和时间隔离到执行层调度层只处理排序排查确定性问题时我建议按这个顺序先固定代码版本和依赖版本记录输入快照和依赖图串行执行一次验证是否有确定性再开并发观察结果差异对比快照定位开始偏离的任务。8. 最佳实践与工程建议8.1 调度器与执行器分离这是最重要的一条。调度器只回答“做什么、按什么顺序”执行器只回答“怎么做”。千万不要在调度器里写业务逻辑否则一旦业务变化调度逻辑就会被污染确定性也就无从谈起。8.2 幂等设计每个任务必须支持重复执行且结果一致。具体做法是任务执行前先检查是否已有结果快照如果有且校验通过直接返回历史结果否则再执行。def run_with_idempotency(task: Task, snapshot_store: Dict[str, object]) - object: if task.task_id in snapshot_store: return snapshot_store[task.task_id] result task.fn() snapshot_store[task.task_id] result return result8.3 结构化日志Harness 的日志不应该全是字符串。推荐使用结构化字段task_idstatetimestampinput_digestoutput_digestduration_ms这样即使并发执行日志也能按任务聚合排查问题效率会高很多。8.4 安全与最小权限Harness 经常需要调用本地命令、读写文件、访问数据库。在做调度编排时必须坚持最小权限原则每个任务只授予完成自身工作所需的最小权限插件代码默认在沙箱中执行涉及删除、写入生产环境的操作要走审批流程所有命令执行必须记录日志方便审计。8.5 版本锁定依赖版本、模型版本、插件版本都要锁定。LLM 模型更新通常会导致输出变化如果 Harness 没有版本锁定调度器再确定最终结果也可能会变。{ models: { deepseek-chat: { version: 2025-01-01, temperature: 0 } }, plugins: { collector: 1.2.3, archiver: 0.9.1 } }8.6 测试策略确定性调度器非常适合做单元测试。你不需要真实执行任务只需要验证调度顺序是否符合预期。def test_scheduler_order(): tasks build_tasks() scheduler DeterministicScheduler() order scheduler.schedule(tasks) assert order [A, B, C, D]对编排器可以用 Mock 任务模拟成功、失败、依赖缺失等情况验证状态流转是否符合预期。9. 下一步可以继续深入的方向这篇文章用最小实现演示了确定性调度器与任务编排器的核心思路。接下来你可以继续往下探索把调度顺序持久化到 SQLite实现断点续跑在编排器中加入虚拟时间模拟超时和重试把任务执行结果写入工作区结合 Harness 的对话归档能力做回放研究 DAG有向无环图调度细节比如并行分支、失败传播策略给调度器加一层反压机制避免任务堆积。回到最初的问题Harness 能不能拥有真正确定性的调度器和任务编排器我的答案是调度和编排层面的确定性是完全可以做到的前提是把随机源隔离在执行层把调度器设计成纯函数并用状态机和快照管理整个任务生命周期。至于 LLM 输出等执行层的不确定性更多是靠参数固定、缓存和快照来“压低概率”而不是追求绝对一致。如果你正在设计自己的 Harness 或工作流框架建议先把调度器的执行顺序和任务状态机做好再考虑并发和插件化。地基稳了上层再复杂都不怕。如果这篇文章对你有帮助建议收藏备用动手跑一遍最小例子你会对确定性调度有更直观的感觉。