tilebox-workflow-authoring

编写 Tilebox 工作流

创建或修改 Python 工作流源代码以及本地／运行时行为。平台 API 请参阅 Tilebox 参考文档；可移植的科学计算与 IO 请参阅地理空间参考文档；端到端组合方案请参阅 recipes。

从最新 API 和合适的技能开始

如果 API 不明确，或记忆中的行为可能已经过时，请检查已安装的版本或搜索最新文档：

tilebox docs search "Task ExecutionContext submit_subtasks"
tilebox docs search "logging tracing context.logger context.tracer"
tilebox docs search "caches job_cache"

将相邻职责委派给对应技能：

• tilebox-workflow-releases：项目初始化、pyproject.toml 脚手架、构建、发布、部署和运行器配置。
• tilebox-workflow-jobs：提交、等待、列出、检查、重试、取消和集群操作。
• tilebox-datasets：数据源／产品建议、准确的数据集 slug、提供方凭据、辅助目录和 CLI 检查。
• tilebox-cli：身份验证、CLI 探索、JSON 输出和文档搜索。
• tilebox-workflow-automations：cron 和存储触发的自动化。

对于新项目，请通过 tilebox-workflow-releases 使用 tilebox workflow init；不要手动搭建项目配置。初始化后，可以在编写工作流时修改生成的代码。

编码前先规划

对于以结果为导向的地球观测任务，请阅读 reference/geospatial/project-planning.md。确认分辨率、时间覆盖范围、观测质量、可比性、所需输入、验证方式和交付内容。使用 tilebox-datasets 选择并检查数据源；不要重复其中的数据目录或凭据说明。

绘制任务图：

1. 确定根阶段、工作阶段和聚合阶段。
2. 选择扇出维度：场景、产品、AOI、时间窗口、空间分块或模型图块。
3. 使用 depends_on 标记真正的屏障；避免不必要的任务链。
4. 选择使用任务字段、context.job_cache，还是持久化对象／Zarr 工件。
5. 为幂等的网络／存储操作选择重试行为。

保持实现精简

• 从满足请求的最简结果开始。只有在保证正确性或后续迭代提出要求时，才增加更具体的筛选、质量策略或处理阶段。
• 如果用户要求编写工作流，请交付可运行的 Tilebox 任务和运行器注册。过程中可以通过本地预览验证处理效果，但不要止步于独立脚本，也不要要求用户逐一提出每个实现阶段。
• 在耗费大量资源进行处理之前，先选出满足请求结果所需的输入；默认不要计算所有候选产品。
• 避免重复查询相同的数据点：缓存所选资产／元数据供后续使用，或划分时间范围并在叶任务中查询。
• 如果工作集能放入内存，则每个资产的完整 AOI 窗口只读取一次；只有在明确进行空间分块时才细分读取，不要为了匹配较小的输出块而细分。
• 将特定运行的选项作为任务输入公开；除非用户需要配置，否则将实现和显示默认值保留在负责这些内容的函数中。
• 仅在有助于提升可读性、可测试性或复用性时提取处理函数；简单处理可以留在 execute 中。只有在操作确实共享时才复用函数，并通过参数处理差异；内联简单的 API 封装，避免构建臆想出来的框架。
• 在编写自定义实现之前，先使用现有库操作来处理网格、数组掩膜、重投影、图像合成和编码。明确处理校准、nodata、对齐和科学验证。
• 验证外部输入和科学前提；让意外错误传递给 Tilebox，不要添加臆想的回退机制或自定义重试层。
• 生成所需的交付物以及图中下游任务会使用的中间产物。除非用户要求，否则不要添加清单、来源／质量附属文件或额外导出。在现有工件元数据或结构化日志中保留有用的来源 ID、参数和质量信息。

按职责划分结构

对于多阶段工作流，runner.py 仅用于注册任务和配置 runner。按职责组织任务代码和处理函数；只有在能形成有效边界时才添加模块，不要为了遵循固定布局或为每个任务单独创建文件而添加模块。对于小型原型，单个文件即可。

测试实质性的处理函数，例如选择、遮罩、变换和编码，不要模拟 ExecutionContext；不要仅为了测试而提取琐碎的辅助函数。通过发布构建、任务发现和代表性作业的实际图来验证编排，而不是用任务类测试提交、进度或日志记录。将所有运行时模块纳入发布包，并验证从构建产物中导入模块。

要求任务级并行

任务级扇出是工作流设计要求，不是可选优化。根据所需输出和工作负载选择扇出维度：场景、时间段、AOI、产品、空间块或模型分块。将该维度上的独立单元表示为单独的 Tilebox 任务，并通过 context.submit_subtask(s) 提交。根据结果推断这种分解方式；绝不要要求用户说出 submit_subtasks、指定任务数量或规定 DAG。例如，需要合并所有帧的 12 场景延时摄影，应采用类似 1 个根任务 + 12 个场景工作任务 + 1 个编码器任务 的图，而不是让单个任务执行 12 次迭代循环。

如果不同输出使用不同源资产，即使属于同一场景或时间段，也要将独立计算的输出拆分为不同任务。共享元数据或共用 QA 遮罩不要求串行执行。如果多个输出可以复用同一次昂贵的读取或计算，则应将它们放在一起。

对于包含若干独立输出栅格的小型工作流，可按输出 TIFF 扇出，例如为每个对比日期分别生成 RGB 和 NBR。只要每个输出、读取数据和处理临时数据在预期 runner 并发下都能满足单个工作任务的内存预算，每个工作任务就可以从选定的 granule 中读取若干波段，并在内存中进行镶嵌。这样可以保持输出级并行，而无需按 granule 创建任务，也无需使用 Zarr 中间结果进行镶嵌。

对于需要分布式镶嵌的较大区域或输入集，应按 granule 扇出，将超大 granule 的工作细分为有界窗口，然后按每个输出块进行聚合。更改任务图时，应保持所请求的合成规则不变。分块归约请参阅 reference/tilebox/geospatial-task-graphs.md。

如果输出需要合并多个工作任务的结果，请使用编排任务提交独立工作任务，并通过 depends_on 提交聚合或发布任务。在有界输出工作任务中使用循环或异步读取是允许的；但它们不能替代独立输出之间的扇出，也不能替代较大规模镶嵌所需的 granule/分块扇出。绝不要在单个任务中用 asyncio.gather、线程或本地多进程来取代这种扇出；额外的 runner 只能并行执行彼此独立的 Tilebox 任务。首次运行规模较小或调度开销较大，都不是将工作流合并为单个任务的理由。应将粒度过细的工作批量分配给多个负载均衡的工作任务。只有在工作确实不可拆分或必须按依赖顺序执行，或者用户明确要求时，才使用单任务工作流。

编码前，记录预期的图结构。运行代表性作业后，将 execution_stats.total_tasks 和 task_summaries 与该结构进行比较。进度单元和子级跟踪跨度都不是任务，不能证明存在扇出。如果图不匹配，应先修正工作流，再报告完成情况。当 runner 容量允许时，使用任务时间戳或跨度来验证独立工作任务确实有并行执行。

核心任务语义请阅读 reference/tilebox/tasks-and-graphs.md，地理空间阶段映射请阅读 reference/tilebox/geospatial-task-graphs.md，边界说明请阅读 reference/tilebox/state-and-artifacts.md。

定义类型化任务

Task 会应用 dataclass 行为；字段会作为输入序列化。

为每个字段选择能表达其含义的最简单、符合惯例的类型。在自定义表示之前，先检查 Python、Tilebox 或处理库是否已提供相应的值类型。不要仅仅为了让值可序列化，就将已有类型重新实现为自定义 dataclass、命名元组，或临时拼凑的元组／列表／字典结构。普通容器仍然适用，前提是它们是自然的 Python 结构，例如用于表示时间范围的 tuple[datetime, datetime]。请阅读 reference/tilebox/tasks-and-graphs.md，了解支持的类型和示例。

from tilebox.workflows import ExecutionContext, Task


class ProcessScene(Task):
    scene_id: str

    @staticmethod
    def identifier() -> tuple[str, str]:
        return "tilebox.com/example/ProcessScene", "v1.0"

    def execute(self, context: ExecutionContext) -> None:
        context.current_task.display = f"ProcessScene({self.scene_id})"
        context.logger.info("Processing scene", scene_id=self.scene_id)

• 原型可使用默认的 v0.0 标识符。稳定标识符会返回 (name, vX.Y)。
• 在迭代期间以及正式发布前，保持初始版本不变。一旦需要考虑兼容性，小版本应向前兼容；输入或行为发生不兼容变更时，应提升主版本。
• 完整序列化输入不得超过 2048 字节。传递紧凑的纯 Python 值或受支持的库值、ID、键和小型配置，而不是数组、数据帧、xarray 数据集、大型几何数据、清单、凭据、客户端、打开的文件或本地路径。请根据 reference/tilebox/state-and-artifacts.md 使用 context.job_cache 或持久化存储。
• 将作业使用的每个任务类都注册到 runner。

直接在任务中等待异步 IO

只要任务使用异步 API（包括 tilebox.storage.aio、异步 HTTP 客户端或有界并发请求），就定义 async def execute(self, context: ExecutionContext) -> None。直接等待这些 API；不要用 asyncio.run(...) 包装它们。工作流执行器会等待异步 execute 方法，同时保持正常的失败、重试、跟踪和完成行为。

import asyncio

from niquests import AsyncSession


class FetchMetadata(Task):
    urls: list[str]

    async def execute(self, context: ExecutionContext) -> None:
        async with AsyncSession() as client:
            responses = await asyncio.gather(
                *(client.get(url) for url in self.urls)
            )
        for response in responses:
            response.raise_for_status()
        context.logger.info("Metadata fetched", count=len(responses))

根据客户端的事件循环行为决定其生命周期：在 execute 内创建 niquests.AsyncSession，而 tilebox.storage.aio.Client 可以在多次任务执行之间重复使用。详情请参阅 reference/tilebox/tasks-and-graphs.md。对于大量请求，应使用信号量或有界批次进行限制。对于选定的 worker 单元内部的多个 IO 请求，适合使用异步并发，例如针对单个场景或有界内存镶嵌图的多个波段或字节范围请求。它不能替代必需的任务级扇出。请在同步 runner 入口点中调用 runner.run_all() 和 runner.run_forever()，不要从异步代码中调用它们。

构建简单的动态图

class ProcessScenes(Task):
    scene_ids: list[str]

    def execute(self, context: ExecutionContext) -> None:
        context.current_task.display = f"ProcessScenes(n={len(self.scene_ids)})"
        workers = context.submit_subtasks(
            [ProcessScene(scene_id) for scene_id in self.scene_ids],
            max_retries=3,
        )
        context.submit_subtask(PublishSummary(), depends_on=workers)

• 对单个子任务使用 submit_subtask，对同质批次使用 submit_subtasks。
• 优先使用共享阶段屏障：先提交所有 A 任务，再使用 depends_on=all_a_handles 提交 B 批次，而不是为每个条目分别创建 B[i] → A[i] 依赖组。对于只使用单个生产者输出的单个下游任务，应依赖该生产者，而不是无关的同级任务。
• 仅对非关键工作使用 optional=True。
• 确保副作用幂等：使用确定性键、可安全覆盖的写入、有效输出检查或原子提交。重试时使用原始任务输入。

进度与可观测性

在执行耗时工作之前，设置简洁的 current_task.display 标签。对于有实际意义的扇出，提交任务应调用 context.progress(name).add(n)；每个成功的 worker 应在完成其负责的工作单元后调用 .done(1)。总数与完成数必须一致。

使用 context.logger 记录结构化字段，并使用 logger.bind 绑定重复出现的上下文。在耗时阶段外层使用 context.tracer.span(...)；需要创建子 span 的辅助函数应接收该 tracer，而不是自行构造全局 OpenTelemetry tracer。仅在需要恢复或提供有用上下文时添加异常处理器；使用 logger.exception 记录异常并重新抛出。在 runner 入口点配置控制台日志记录，不要在任务类中配置。

数据集、资产和存储边界

• 选择数据源后的查询机制：reference/tilebox/datasets-and-datapoints.md。
• 资产解码和元数据：reference/tilebox/assets.md。
• 资产字节访问和有界 COG 窗口：reference/tilebox/storage-access.md。

不要重建提供方路径、重复存储 API 代码片段或嵌入数据源／提供方设置。明确应用比例／偏移、无数据值、掩码、对齐和重投影。将下载操作放在使用该数据的任务中，并向其他任务传递持久 ID／键，而非本地路径。

依赖项和操作

在 pyproject.toml 中声明运行时依赖项，包括类型出现在任务字段中的可选库。Tilebox 会延迟加载这些库的序列化器，但不会安装这些库。运行 uv sync，避免使用可编辑／本地路径依赖，并参阅 reference/tilebox/dependencies-and-packaging.md。仅使用创作功能处理数据源并执行针对性的本地检查。使用 tilebox-workflow-releases 进行初始化／构建／部署，使用 tilebox-workflow-jobs 进行提交／等待／集群操作。

参考资料导航

Tilebox 平台

参考资料	适用场景
reference/tilebox/tasks-and-graphs.md	定义／管理任务版本、选择受支持的输入类型、遵守输入限制、提交依赖项、处理重试、进度和可观测性，以及任务注册和运行器模式。
reference/tilebox/datasets-and-datapoints.md	查询已选定的数据集、检查样本、准确选择一个数据点，或遍历数据点。
reference/tilebox/assets.md	解码 AssetCollection、语义键、栅格元数据、比例／偏移或显式覆盖项。
reference/tilebox/storage-access.md	解析／读取／流式传输／下载／打开 Tilebox 资产、访问策略、匿名访问、窗口或并发。
reference/tilebox/state-and-artifacts.md	选择任务输入、作业缓存、对象／Zarr 工件、本地临时存储和支持安全重试的键。
reference/tilebox/dependencies-and-packaging.md	声明兼容 uv 的工作流依赖项和适合安全发布的程序包。
reference/tilebox/geospatial-task-graphs.md	将场景、窗口、分块、归约和进度映射到 Tilebox 任务。

Tilebox 与地理空间混合配方

参考资料	适用场景
reference/tilebox/recipes/sentinel-2-cog.md	读取公开的 L2A RGB 和 SCL，并将 20 m 分类数据与 10 m RGB 对齐。
reference/tilebox/recipes/sentinel-1-sar.md	构建 SAR 变化、洪水或海事任务图。
reference/tilebox/recipes/time-series-composite.md	合成已对齐的场景或生成延时影像。
reference/tilebox/recipes/geospatial-ml.md	并行分发模型图块并汇总地理空间预测结果。

便携式地理空间

参考资料	适用场景
reference/geospatial/project-planning.md	将预期成果转化为可行的数据、验证和输出要求。
reference/geospatial/raster-fundamentals.md	处理 CRS、变换、形状、dtype、nodata、掩膜、缩放和单位。
reference/geospatial/io/cloud-native-raster.md	在规范资产之外直接读取云原生 GeoTIFF/COG。
reference/geospatial/io/object-storage.md	将 obstore 用于 S3、GCS、Azure、本地或兼容存储。
reference/geospatial/io/zarr.md	设计 Zarr 架构、分块、区域写入和带标签的读取。
reference/geospatial/io/cog-output.md	编写正确的 COG 栅格，并在需要时测试输入或写入器配置。
reference/geospatial/io/formats-and-encoding.md	选择 COG/Zarr/NetCDF/GeoParquet，并保留关键编码语义。
reference/geospatial/processing/grids-and-reprojection.md	选择网格、进行重投影和语义重采样。
reference/geospatial/processing/masking-and-qa.md	处理 nodata、掩膜、QA、类别和形态学。
reference/geospatial/processing/time-series.md	对齐、合成、设置样式并验证时序影像。
reference/geospatial/processing/ml-inference.md	进行分块、加载模型、归一化和生成预测输出。
reference/geospatial/products/sentinel-1-sar.md	了解 SAR 产品选择、几何、单位和注意事项。
reference/geospatial/products/sentinel-2-l2a.md	了解 L2A 波段、分辨率、缩放和 SCL。
reference/geospatial/products/landsat-collection-2.md	了解 Landsat Level-2 波段、缩放和 QA 位。

验证清单

1. 所有已提交的任务均已注册，且标识符/版本一致。
2. 任务字段使用惯用的 Python 类型或现有库的值类型，而不是在自定义数据类或临时容器中重复定义相同结构。
3. 每个序列化任务输入均不超过 2048 字节。
4. 跨任务数据使用恰当的状态/产物边界和确定性键。
5. 可重试的执行具有可重入性，并且与输入兼容。
6. 高扇出任务具有标签、有用的进度信息、结构化日志，以及适当的追踪跨度。
7. 依赖项能够同步，任务模块可在预期环境中导入。
8. 对于脚手架生成的发布项目，编辑生成的任务代码后，tilebox workflow build-release --debug --json 能够成功运行。
9. 非简单工作流使用结构清晰的包模块，并将运行器入口限制为注册和运行器级配置。
10. 有实质内容的底层处理函数在适用时有针对性测试；Tilebox 任务编排通过构建/任务发现和代表性作业验证，而不是通过任务类单元测试验证。
11. 地球观测输出经过有代表性的科学验证、覆盖范围/对齐检查，并包含溯源信息和明确说明的限制。
12. 代表性作业的实际任务摘要与计划中的根任务、扇出和聚合阶段相符；进度计数器或子追踪跨度不视为任务扇出。