DispatchAtlas教程
这些教程是面向-学习的端-到-端走查：每一步都是一个可运行的脚本，所展示的每个输出都是来自捆绑
烟雾数据的真实输出，且每次运行都是确定性的——相同的种子在你的机器上产生相同的数字。先完成
安装与快速开始，
然后用 uv run python <file>.py 从仓库检出运行每个脚本。
🧭 选择一个教程
| 教程 | 你将学到 | 触及的包 |
|---|---|---|
| 建模并求解第一个派遣问题 | 选取一个基准实例，在其上运行两个求解器，比较调度，并阅读求解器元数据。 | dispatchatlas.core、dispatchatlas.bench、dispatchatlas.solve |
| 运行一个小型活动并阅读其结果 | 配置一个带检查点的活动，将其执行到一个工作区，并加载结果以供分析。 | dispatchatlas.lab、dispatchatlas.analytica |
更深的参考路径在教程结束处继续：领域契约用于调度模型， 基准模型用于目录生成，求解器系统用于完整 注册表，活动引擎用于编排细节，以及 分析导出用于证据包。更大的活动需要 就绪中描述的统计与发布关卡。
🛠️ Tutorial 1：建模并求解第一个派遣问题
快速开始用一个求解器求解了一个问题。本教程更深一层：你选取一个特定的基准实例，在其上运行两个 派遣基线，比较它们构建的调度，并阅读解释每个求解器的元数据。
1️⃣ 看烟雾目录提供什么
捆绑的烟雾目录从一个单一根种子为每个调度族物化两个小型确定性实例。将此保存为
list_problems.py 并用 uv run python list_problems.py 运行：
from dispatchatlas.bench import smoke_benchmark_provider
provider = smoke_benchmark_provider(root_seed=20260527)
for problem_id in provider.list_problem_ids():
print(problem_id.value)输出：
smoke-cloud-edge-0
smoke-cloud-edge-1
smoke-workflow-0
smoke-workflow-1
smoke-machine-scheduling-unrelated-0
smoke-machine-scheduling-unrelated-1
smoke-job-shop-0
smoke-job-shop-1
smoke-flexible-job-shop-0
smoke-flexible-job-shop-1
smoke-permutation-flow-shop-0
smoke-permutation-flow-shop-1
smoke-setup-flow-shop-0
smoke-setup-flow-shop-1
smoke-rcpsp-renewable-0
smoke-rcpsp-renewable-1
smoke-open-shop-0
smoke-open-shop-1
smoke-hybrid-flow-shop-0
smoke-hybrid-flow-shop-1
smoke-distributed-permutation-flow-shop-0
smoke-distributed-permutation-flow-shop-1
smoke-no-wait-flow-shop-0
smoke-no-wait-flow-shop-1
smoke-blocking-flow-shop-0
smoke-blocking-flow-shop-1
smoke-distributed-assembly-flow-shop-0
smoke-distributed-assembly-flow-shop-1
smoke-multi-objective-pfsp-0
smoke-multi-objective-pfsp-1
smoke-rcpsp-max-0
smoke-rcpsp-max-1
smoke-rcpsp-multi-mode-0
smoke-rcpsp-multi-mode-1
smoke-multi-project-rcpsp-0
smoke-multi-project-rcpsp-1
smoke-unrelated-parallel-setup-0
smoke-unrelated-parallel-setup-1
smoke-reentrant-fab-0
smoke-reentrant-fab-1
smoke-distributed-flexible-job-shop-0
smoke-distributed-flexible-job-shop-1
smoke-facility-assignment-0
smoke-facility-assignment-1每个 id 命名其调度族和一个从-零开始的实例索引。教程其余部分使用
smoke-job-shop-0。
2️⃣ 在一个实例上运行两个求解器
provider.get_problem 接收一个 ProblemId 并返回一个 ValidatedProblem——问题规约加上其
验证戳记与报告，因此求解器从不接收一个未-验证的实例。下面两个求解器是确定性派遣基线：
earliest-start 按拓扑输入顺序调度任务，而 shortest-processing-time 优先处理较短任务。将
此保存为 compare_solvers.py：
from dispatchatlas.bench import smoke_benchmark_provider
from dispatchatlas.core import ProblemId, TerminationPolicy, derive_seed
from dispatchatlas.solve import default_solver_registry
provider = smoke_benchmark_provider(root_seed=20260527)
validated = provider.get_problem(ProblemId("smoke-job-shop-0"))
spec = validated.spec
print(f"problem: {spec.id.value}")
print(f"tasks: {len(spec.tasks)} resources: {len(spec.resources)}")
registry = default_solver_registry()
for solver_id in ("earliest-start", "shortest-processing-time"):
metadata = registry.get_metadata(solver_id)
solver = registry.create(solver_id)
run = solver.solve(
problem=validated,
stop=TerminationPolicy(max_iterations=metadata.default_stop.max_iterations),
seed=derive_seed(20260610, "docs.tutorial.compare", 0),
)
makespan = run.result.objective_values[0]
print(
f"{solver_id}: feasible={run.result.feasible} "
f"{makespan.objective_name}={makespan.value:.1f}"
)用 uv run python compare_solvers.py 运行：
problem: smoke-job-shop-0
tasks: 9 resources: 3
earliest-start: feasible=True makespan=353.0
shortest-processing-time: feasible=True makespan=433.0两个调度都可行，且在此实例上更朴素的规则胜出：earliest-start 在 353.0 完成，而优先安排 较短任务的 shortest-processing-time 到 433.0 才完成。这值得停下来想一想，因为 shortest-processing-time 在一般情形下是个好规则——在作业车间中，它可能推迟一个后续任务 正在等待的长任务，于是整个调度都跟着一起等。一个规则的名声并不能告诉你它在你的实例上会 做什么。
这正是形态而非数字才是要点的原因。一个实例朝任何方向都证明不了什么——那正是 活动与统计方法的用途 ——但这里的比较（相同的已验证问题、相同的停止准则、相同的派生种子）正是更大证据被构建的方式， 而单次运行恰恰不是。
3️⃣ 阅读求解器元数据
每个已注册求解器携带元数据作为其公共契约：支持的目标、能力标签、随机性、默认停止准则，以及
一个规范引用。将此保存为 inspect_metadata.py：
from dispatchatlas.solve import default_solver_registry
registry = default_solver_registry()
metadata = registry.get_metadata("shortest-processing-time")
print(f"solver: {metadata.solver_id}")
print(f"stochasticity: {metadata.stochasticity}")
print(f"citation: {metadata.citation.reference}")
print("capabilities:", ", ".join(c.value for c in metadata.capabilities))用 uv run python inspect_metadata.py 运行：
solver: shortest-processing-time
stochasticity: deterministic
citation: Smith, W. E. (1956). Various optimizers for single-stage production. Naval Research Logistics Quarterly, 3(1-2), 59-66.
capabilities: single-objective, capacity-aware, precedence-aware, constructive, dispatching, deterministic这些元数据正是 registry.select 据以过滤、以及求解器推荐器
据以解释的。完整注册表——基线、元启发式、精确适配器，以及 NDSO 族——编目于
算法与求解器系统。
🧪 Tutorial 2：运行一个小型活动并阅读其结果
一个活动是一组已配置的运行——基准实例与求解器及目标交叉——以显式种子、预算、检查点，以及一个
磁盘上的结果仓库执行。本教程运行与 experiments/configs/smoke-pilot.json 处被跟踪的配方相同的
形态：两个烟雾实例、两个确定性基线、一个目标。
1️⃣ 配置并运行活动
将此保存为 first_campaign.py。它声明活动，将其验证为一个确定性计划，以一次干运行估计其成本，
然后将其执行到一个本地 my-campaigns/ 工作区：
from pathlib import Path
from dispatchatlas.core import TerminationPolicy
from dispatchatlas.lab import (
CampaignConfig,
CampaignKind,
CampaignStage,
ExecutionMode,
OutputPolicy,
ResourceBudget,
default_campaign_runner,
)
workspace = Path("my-campaigns")
config = CampaignConfig(
campaign_id="first-campaign",
benchmark_ids=("smoke-job-shop-0", "smoke-workflow-0"),
solver_ids=("earliest-start", "shortest-processing-time"),
objectives=("makespan",),
root_seed=20260610,
seed_namespace="docs.tutorial.first-campaign",
stop=TerminationPolicy(max_iterations=5),
output=OutputPolicy(root_dir=str(workspace)),
stage=CampaignStage.SMOKE,
kind=CampaignKind.PILOT,
resources=ResourceBudget(
max_workers=2,
max_concurrent_runs=2,
estimated_seconds_per_run=0.5,
),
execution_mode=ExecutionMode.SEQUENTIAL,
)
runner = default_campaign_runner(workspace)
plan = runner.validate(config)
budget = runner.dry_run(plan)
print(f"planned runs: {len(plan.runs)}")
print(f"estimated wall time: {budget.estimated_wall_time_seconds:.1f}s")
index = runner.run(plan)
print(f"completed runs: {index.run_count}")
print(f"failed attempts: {len(index.failures)}")用 uv run python first_campaign.py 运行：
planned runs: 4
estimated wall time: 2.0s
completed runs: 4
failed attempts: 0四次运行恰是交叉积：2 个基准实例 × 2 个求解器 × 1 个目标，每格一次运行，因为两个求解器都是
确定性的。每次运行的种子从 root_seed 与运行的位置派生，因此重新运行脚本重现相同的记录；
检查点让一个被中断的活动恢复而不重复已完成的工作。
2️⃣ 检查磁盘上落地了什么
runner 写出了一个与被跟踪的实验工作区布局相同的工作区：
my-campaigns/
.checkpoints/first-campaign.json
ENVIRONMENT.md
logs/first-campaign_<timestamp>.log
results/first-campaign/
plan.json
environment.json
earliest-start/smoke-job-shop-0/run_0.json
earliest-start/smoke-workflow-0/run_0.json
shortest-processing-time/smoke-job-shop-0/run_0.json
shortest-processing-time/smoke-workflow-0/run_0.json每次运行一条 JSON 记录，按活动、求解器、基准与重复索引寻址——路径本身即索引。每条记录在其
确定性载荷上做内容-哈希，那正是 replay 模式所据以验证的。
3️⃣ 加载结果以供分析
dispatchatlas.analytica 读取已完成的活动目录而不导入活动运行时。将此保存为
read_results.py：
from pathlib import Path
from dispatchatlas.analytica import load_result_dataset, summarize_dataset
dataset = load_result_dataset(Path("my-campaigns"), "first-campaign")
print(f"campaign: {dataset.campaign_id}")
print(f"completed runs: {len(dataset.completed)}")
summary = summarize_dataset(dataset)
for solver in summary.solver_summaries:
print(
f"{solver.solver_id}: runs={solver.count} "
f"feasible={solver.feasible_count} "
f"mean {solver.objective_name}={solver.mean:.1f}"
)用 uv run python read_results.py 运行：
campaign: first-campaign
completed runs: 4
earliest-start: runs=2 feasible=2 mean makespan=182.1
shortest-processing-time: runs=2 feasible=2 mean makespan=222.1summarize_dataset 按求解器与目标计算描述性统计。每个求解器仅两次运行时，推断方法（显著性
检验、置信区间）路由到限制面而非产生功效-不足的结果——每个随机求解器-实例格 30 次独立运行的
统计-功效下限描述于分析导出。
4️⃣ 导出一个证据包（可选）
同一活动目录馈入导出命令，它写出一个由表格、图形与补充材料构成的披露-过滤证据包：
uv run dispatchatlas export `
--campaign-dir .\my-campaigns\results\first-campaign `
--target-dir .\exports\first-campaign `
--authorized-output-root .\exports `
--tier core该包落地于 exports/first-campaign/evidence-bundles/first-campaign-core/。
证据包页面解释四个层级与包内容。