fraclab-sdk 0.1.65

SDK for managing snapshots, algorithms, and run execution

Fraclab SDK Reference

版本: 0.1.65 Python: >=3.11

Fraclab SDK 是一个算法开发与执行框架，帮助算法开发者快速构建、测试和部署数据处理算法。

完整文档站源码位于 docs/，站点配置位于 mkdocs.yml。

本地预览：

poetry install --with docs
poetry run mkdocs serve

这里是基于当前仓库源码本地起文档站，不依赖 GitHub Pages 或 PyPI。

静态构建：

poetry run mkdocs build

---

目录

• 1. 安装
• 2. 快速开始：编写你的第一个算法（从 Bundle 到运行）
  • 2.1 开发闭环（先看这个）
  • 2.2 准备 Bundle 路径
  • 2.3 编写算法入口 main.py
  • 2.4 定义 InputSpec
  • 2.5 定义 OutputContract
  • 2.6 创建算法清单
  • 2.7 用 Bundle 编译并导出算法包
  • 2.8 导入并运行算法包
  • 2.9 SelectionModel 与 run_ds 的关系
• 3. Bundle 与 Snapshot（概念与关系）
• 4. 算法开发详解
• 5. CLI 命令行工具
• 6. SDK 内部模块
• 7. 数据模型
• 8. 错误处理
• 9. 安全特性
• 10. Workbench 图形界面
• 11. 完整示例
• 12. 附录 A: Bundle 结构详解

---

1. 安装

PyPI 发布版安装（核心 SDK / CLI）：

pip install fraclab-sdk

PyPI 发布版安装并启用 Workbench UI：

pip install "fraclab-sdk[workbench]"
fraclab-workbench           # CLI entry point
# 或
python -m fraclab_sdk.workbench

如果你当前就在仓库里开发或本地试用，不需要先传 GitHub 或 PyPI。直接按源码安装：

poetry install
poetry run fraclab-sdk --help

如果还要启用 Workbench 和文档：

poetry install --extras workbench --with docs
poetry run fraclab-workbench
poetry run mkdocs serve

Workbench UI 内置英文 / 简体中文双语支持，可在每个页面右上角切换语言；也可以通过环境变量 FRACLAB_WORKBENCH_LANG=en 或 FRACLAB_WORKBENCH_LANG=zh-CN 设定初始语言。

CLI 入口

安装后提供三个命令行入口：

命令	说明
fraclab-sdk	主 CLI 工具（快照管理、算法管理、运行管理、验证等）
fraclab-runner	算法子进程 runner（内部使用，由 RunManager.execute() 自动调用）
fraclab-workbench	启动 Workbench 图形界面

依赖说明

对“算法代码允许使用哪些第三方包”，按下面这份白名单理解即可：

• fraclab-sdk
• numpy>=2.1.1
• pandas>=2.2.3
• scipy>=1.13.1
• matplotlib>=3.9.2
• fastapi>=0.115.0

另外，Python 标准库也可以直接使用。

除上面这些以外，其他第三方包都不应作为算法依赖来写。即使某些包会随着 SDK 或 Workbench 一起安装，也不属于算法可依赖契约。

例如下面这些都不要当成“算法可直接使用的依赖”：

• pydantic
• plotly
• streamlit
• streamlit-plotly-events
• pyarrow
• typer
• rich

补充说明：

• 这里说的是算法代码的依赖约定，包括 main.py 和算法工作区里的 schema 定义代码。
• fraclab-sdk[workbench] 只是额外安装 Workbench UI 侧依赖：streamlit>=1.30、streamlit-plotly-events>=0.0.6、pyarrow>=16.0.0；这不等于算法运行时默认允许依赖它们。
• SDK 包自身的完整安装依赖，以 pyproject.toml 为准；不要把 SDK 内部实现依赖误读成算法白名单。

算法运行时中文字体

SDK 现在按 sandbox 运行时对齐了 matplotlib 的中文字体默认值。通过 RunManager / fraclab-runner 执行算法时，runner 会在加载你的 main.py 之前自动调用：

from fraclab_sdk.runtime import configure_matplotlib_runtime_fonts

默认会把当前环境里可见的以下 family 预置到 matplotlib 的 font.sans-serif / font.serif 前面，并同时设置 axes.unicode_minus = False：

• WenQuanYi Micro Hei
• Noto Sans CJK JP
• Noto Serif CJK JP

对应的系统包是：

• fonts-wqy-microhei
• fonts-noto-cjk

说明：

• SimHei 当前不在 sandbox 容器里安装，也不会被 SDK 作为可用 runtime family 暴露。
• 如果你脱离 runner、直接本地调试算法脚本，需要手动先调用 configure_matplotlib_runtime_fonts()。
• 如果算法里想显式引用这些 family，可以直接使用：

from fraclab_sdk.runtime import RUNTIME_CJK_FONT_FAMILIES

如果想在算法内动态检查当前环境已安装了哪些 CJK 字体 family：

from fraclab_sdk.runtime import get_available_runtime_cjk_font_families

available = get_available_runtime_cjk_font_families()
# 例如 ['WenQuanYi Micro Hei', 'Noto Sans CJK JP']

完整运行时字体导出列表（fraclab_sdk.runtime）：

导出名	类型	说明
RUNTIME_CJK_FONT_FAMILIES	list[str]	SDK 预定义的 CJK font family 列表
RUNTIME_CJK_FONT_PACKAGES	list[str]	对应的系统字体包名
configure_matplotlib_runtime_fonts()	函数	配置 matplotlib CJK 字体（runner 自动调用）
get_available_runtime_cjk_font_families()	函数	检测当前环境已安装的 CJK font family

---

2. 快速开始：编写你的第一个算法（从 Bundle 到运行）

2.1 开发闭环（先看这个）

对初次使用者，先记住这一条主线：

1. 你拿到平台给的 Bundle 目录（含 manifest.json、ds.json、drs.json、data/）
2. 你编写算法源码（main.py + schema/* + manifest.json）
3. 你用 Bundle 编译，生成 dist/*.json（尤其 dist/ds.json 与 dist/drs.json）
4. 你导出算法 zip
5. 你导入 Snapshot（来自 Bundle）和算法 zip
6. 你创建 run、执行 run、查看结果

推荐一条命令完成“编译 + 导出”：

fraclab-sdk algo export ./my-algorithm ./my-algorithm.zip --auto-compile --bundle /path/to/bundle

说明（与代码行为一致）：

• export 本身要求 dist/params.schema.json、dist/output_contract.json、dist/ds.json、dist/drs.json 已存在
• --auto-compile 会在缺少这些文件时自动调用 compile
• compile 阶段如果没有可用的 dist/ds.json / dist/drs.json，就必须提供 --bundle（用于复制 bundle 的 ds.json 与 drs.json）

2.2 准备 Bundle 路径

Bundle 是算法开发和运行的共同输入。你至少会在两个阶段用到它：

• 编译阶段：提供 DS/DRS（生成 dist/ds.json 与 dist/drs.json）
• 运行阶段：导入为 Snapshot，作为 run 的数据来源

最小检查：

fraclab-sdk validate bundle /path/to/bundle

2.3 编写算法入口 main.py

算法开发者主要使用 fraclab_sdk.runtime 模块中的两个核心类：

from fraclab_sdk.runtime import DataClient, OutputClient

• DataClient: 读取输入数据
• OutputClient: 写入输出结果

入口签名与模板

创建 main.py 作为算法入口文件。

入口函数签名约定

算法入口函数必须严格遵循以下签名:

def run(ctx):
    ...

约定	要求	说明
文件名	main.py	必须是 main.py，不能是其他名称
函数名	run	必须是 run，区分大小写
参数	仅 ctx 一个参数	SDK 使用 module.run(ctx) 调用，不支持其他签名
返回值	无要求	返回值会被忽略

警告: 以下写法都会导致运行失败:

# ❌ 错误: 参数名不影响，但参数个数必须是 1
def run(ctx, extra_arg):  # TypeError: run() missing required argument

# ❌ 错误: 函数名错误
def execute(ctx):  # AttributeError: module has no attribute 'run'

# ❌ 错误: 放在其他文件
# helper.py 中定义 run()  # 不会被加载

最小可运行模板

# main.py
def run(ctx):
    """算法入口函数 - 最小模板。

    Args:
        ctx: RunContext，包含:
            - ctx.data_client: DataClient 实例
            - ctx.params: dict[str, Any]，用户参数
            - ctx.output: OutputClient 实例
            - ctx.logger: logging.Logger 实例
            - ctx.run_context: dict，运行上下文
    """
    logger = ctx.logger
    logger.info("算法开始执行")

    # 你的逻辑...

    logger.info("算法执行完成")

完整示例

# main.py
def run(ctx):
    """算法入口函数。

    Args:
        ctx: RunContext，包含:
            - ctx.data_client: DataClient 实例
            - ctx.params: dict，用户参数
            - ctx.output: OutputClient 实例
            - ctx.logger: Logger 实例
            - ctx.run_context: dict，运行上下文
    """
    dc = ctx.data_client
    out = ctx.output
    params = ctx.params
    logger = ctx.logger

    # 获取参数
    threshold = params.get("threshold", 0.5)
    logger.info(f"开始处理，阈值: {threshold}")

    # 读取输入数据
    for dataset_key in dc.get_dataset_keys():
        count = dc.get_item_count(dataset_key)
        logger.info(f"数据集 {dataset_key} 包含 {count} 个项目")

        for i in range(count):
            # 读取 NDJSON 对象（此处 i 是 run_ds 中的连续索引 0..count-1）
            obj = dc.read_object(dataset_key, i)

            # 处理数据...
            result = process(obj, threshold)

            # 写入结果
            out.write_scalar(f"{dataset_key}_result_{i}", result)

    # ⚠️ 注意: 若需要匹配时间窗的 itemKey，不能用 range(count) 遍历。
    # timeWindows 中的 itemKey 是选择器输出的标识，可能是稀疏集合（如 00/04/05），
    # 需要按 itemKey 列表匹配对应的 parquet 目录或 NDJSON 行。
    # 示例: 按 itemKey 匹配时间窗
    tw_key = f"timeWindows_{dataset_key}"
    windows = params.get(tw_key)
    if windows:
        for w in windows:
            item_key = w.get("itemKey")
            t_min, t_max = w["min"], w["max"]
            logger.info(f"  item={item_key} window=[{t_min}, {t_max}]")

    # 写入汇总结果
    out.write_json("summary", {"status": "completed", "threshold": threshold})

    # output/manifest.json 由 SDK 自动生成，算法代码不要手写或修改 manifest。

    logger.info("算法执行完成")

def process(data, threshold):
    """你的数据处理逻辑"""
    return data.get("value", 0) > threshold

2.4 定义输入参数规格 (InputSpec)

创建 schema/inputspec.py 定义算法接受的参数：

# schema/inputspec.py
from .base import CamelModel, Field

class InputParams(CamelModel):
    threshold: float = Field(
        default=0.5,
        ge=0.0,
        le=1.0,
        description="检测阈值"
    )

    debug: bool = Field(
        default=False,
        description="启用调试模式"
    )

# 必须导出 INPUT_SPEC
INPUT_SPEC = InputParams

ctx.params 的类型与访问方式

ctx.params 是 dict[str, Any] 类型，键名来自 params.json (JSON 原始键名)。

# main.py - 算法代码
def run(ctx):
    # ctx.params 是 dict，使用 dict 访问方式
    threshold = ctx.params.get("threshold", 0.5)
    debug = ctx.params.get("debug", False)

    # 嵌套对象同样是 dict
    filters = ctx.params.get("filters", {})
    min_depth = filters.get("minDepth", 0)

全局 JSON 命名规则

算法作者直接编写或直接使用的 JSON，统一使用 camelCase。
Bundle 自带的 ds.json / drs.json 跟随 Bundle 规范，保持平台原始格式，不适用这里的 camelCase 规则。

文件	命名风格	说明
params.schema.json	camelCase	InputSpec 生成的 JSON Schema，字段名必须 camelCase
output_contract.json	camelCase	输出合约，所有键递归检查 camelCase
ds.json	Bundle 原始格式	数据规格，由平台提供，字段如 key、resource
drs.json	Bundle 原始格式	数据需求规格，由平台提供，字段如 key、resource
运行输出 manifest.json	camelCase	运行结果清单，所有键递归检查 camelCase
算法访问 ctx.params	camelCase	dict 访问，键名与 JSON 一致

InputSpec 定义示例：

# schema/inputspec.py
from .base import CamelModel, Field

class MyParams(CamelModel):
    maxItems: int = Field(default=10)
    datasetKey: str = Field(default="wells")

INPUT_SPEC = MyParams

对应的 params.json：

{
  "maxItems": 5,
  "datasetKey": "stages"
}

算法中访问：

def run(ctx):
    # ctx.params 是 dict，键名统一 camelCase
    max_items = ctx.params.get("maxItems", 10)
    dataset_key = ctx.params.get("datasetKey", "wells")

数值参数精度（Workbench）

在 Workbench 的参数输入 UI 中，number 类型字段的显示精度按 InputSpec 生成的 schema step 决定：

• 有 step：小数位数由 step 推导（例如 step=0.01 显示 2 位，step=0.001 显示 3 位）
• 无 step：按整数显示（不显示小数位）

示例：

from .base import CamelModel, Field

class InputParams(CamelModel):
    threshold: float = Field(default=0.5, json_schema_extra={"step": 0.01})  # 2 位小数
    gain: float = Field(default=1.0)  # 未设置 step，Workbench 按整数样式显示

时间窗参数（uiType="time_window"）

Run 页面使用一个统一时间窗选择器（位于参数区底部），在组件内切换 dataset。

关键规则：

• 字段 shape 只允许：Optional[List[TimeWindow]]（TimeWindow = {min,max}）。
  • 若使用 time_window_list() helper：字段类型直接写 helper 返回的模板类型，不要再写 Optional[...]。
  • 例如：stage_windows: WindowsTemplate = Field(default=None, ...)（正确），Optional[WindowsTemplate]（错误）。
• 每个时间窗字段必须配置 bindDatasetKey（顶层或 json_schema_extra）。
• 每个时间窗字段必须显式写 unit="us"；运行时固定按 epoch 微秒（UTC）处理，不接受秒/毫秒，也不做单位换算。
• 匹配依据是当前 run 的 input/ds.json 中已选 dataset keys，不是 params 里的 datasetKey 字段。
• 同一个选择器在切换 dataset 时会套用该 dataset 对应字段的窗口约束（minItems/maxItems）。
• 语义约定（必须统一）：
  • null 表示“未启用该模板”（推荐默认值）
  • [] 不允许作为有效值（建议 minItems >= 1）；若用户删空窗口，应回写为 null
• 若定义 windowSlots 与 windowSlotFallbackNote，图像上方会显示”下一时间窗”备注：
  • 还未选窗：显示第 1 条备注
  • 选完第 1 个窗后：显示第 2 条备注
  • 依次类推；超出后显示 fallback 备注

可复制示例（推荐）：

from __future__ import annotations

from .base import CamelModel, Field, schemaExtra, showWhenCondition, time_window_list

WindowsTemplate_1to3 = time_window_list(
    min_items=1,
    max_items=3,
    title="Windows",
    description="Template windows (1..3).",
)

WindowsTemplate_2 = time_window_list(
    min_items=2,
    max_items=2,
    title="Windows",
    description="Template windows (exactly 2).",
)


class INPUT_SPEC(CamelModel):
    emitDebugArtifacts: bool = Field(
        default=True,
        json_schema_extra=schemaExtra(group="Plots", order=20),
    )

    # 注意：time_window_list() 已经是 Optional[list[TimeWindow]]，不要再包 Optional[...]
    timeWindows_fracRecord_stage_5712: WindowsTemplate_1to3 = Field(
        default=None,
        title="Time Windows (fracRecord_stage_5712)",
        json_schema_extra=schemaExtra(
            group="Plots",
            order=23,
            uiType="time_window",
            unit="us",
            bindDatasetKey="fracRecord_stage_5712",
            showWhen=showWhenCondition("emitDebugArtifacts", op="equals", value=True),
            windowSlots=[
                {"title": "Window 1", "note": "Primary interval."},
                {"title": "Window 2", "note": "Secondary interval."},
                {"title": "Window 3", "note": "Optional baseline interval."},
            ],
            windowSlotFallbackNote="Pick an interval consistent with the algorithm expectation.",
        ),
    )

    timeWindows_fracRecord_stage_9072: WindowsTemplate_2 = Field(
        default=None,
        title="Time Windows (fracRecord_stage_9072)",
        json_schema_extra=schemaExtra(
            group="Plots",
            order=24,
            uiType="time_window",
            unit="us",
            bindDatasetKey="fracRecord_stage_9072",
            showWhen=showWhenCondition("emitDebugArtifacts", op="equals", value=True),
            windowSlots=[
                {"title": "Window 1", "note": "Primary interval."},
                {"title": "Window 2", "note": "Secondary interval."},
            ],
            windowSlotFallbackNote="Pick an interval consistent with the algorithm expectation.",
        ),
    )

注意：

• 可以定义多个时间窗字段，但 Run 页面只渲染一个统一选择器。
• 统一选择器内部切换 dataset，并自动使用对应 bindDatasetKey 字段的约束和备注。
• 每个 bindDatasetKey 只能对应一个时间窗字段（重复绑定会报错）。
• bindDatasetKey 必须与该算法 DRS（dist/drs.json）中的 datasets[*].key 一致；不一致会导致运行期无法绑定到 selector items。
• time_window_list() 返回值已内置 Optional[list[TimeWindow]]，调用方不要写成 Optional[time_window_list(...)]（会产生双重 Optional schema，增加 UI/校验歧义）。

已知 lint warning: time_window_list() 生成的 TimeWindow 子字段（min/max/itemKey）在 JSON Schema 的 $defs 中已有 title，但 Pydantic 生成的 timeWindows_* 顶层字段本身 可能产生额外的 leaf 节点缺少 title，导致 FIELD_MISSING_TITLE warning（每个 timeWindows_* 字段一条）。 此 warning 不影响校验通过（validate 仍然 valid=True），仅表示 UI 显示可能缺少中文标签。 若要消除 warning，可在 Field(...) 中显式传入 title 参数。

TimeWindow 运行时结构

选择器输出到 params.json 的 timeWindows_<datasetKey> 值为 per-item 窗口列表，每条记录包含 itemKey：

{
  "timeWindows_fracRecord_stage_5712": [
    {"itemKey": "00000", "min": 1768521600000000.0, "max": 1768522800000000.0},
    {"itemKey": "00004", "min": 1768530000000000.0, "max": 1768531200000000.0},
    {"itemKey": "00005", "min": 1768535000000000.0, "max": 1768536000000000.0}
  ]
}

关键点：

• itemKey: 标识选中项在 dataset 中的编号（如 "00000"、"00004"）。itemKey 集合可以是稀疏的（不保证 0..N-1 连续），算法必须按选中 itemKey 列表处理，禁止按连续索引 range(count) 遍历。
• min/max: 时间窗边界，统一为 epoch microseconds (UTC)。
• 若该 dataset 未启用时间窗（用户未选），值为 null。
• 空列表 [] 不是合法值（建议 minItems >= 1）。

时间值语义与单位规范

统一规则：时间窗 min/max 始终为 epoch microseconds (UTC)。

来源	说明
Workbench 选择器输出	无论图表如何显示，min/max 一律归一化为 epoch 微秒后写入 params.json
算法内部处理	必须与时间窗同单位（epoch 微秒）；禁止"时间列 ms、窗口 us"混用
Parquet/NDJSON 原始时间列	Workbench 会自动推断单位（ns/us/ms/s）并归一化，但算法读取原始数据时需自行对齐

Workbench 时间列识别优先级（大小写不敏感）：

timestamp > ts > time > datetime > bucket

Workbench 会从上述候选列中选一个作为 x 轴，并按以下规则归一化：

• 数值时间戳：推断单位（ns/us/ms/s），归一到 epoch 微秒
• datetime 列：统一到 UTC 后转 epoch 微秒

算法侧强制要求：

1. 算法读取数据后，必须将时间列归一到 epoch 微秒后再与窗口 min/max 做比较
2. 在算法启动时记录日志：检测到的时间类型、归一化单位、首尾时间戳
3. 对窗口做入参校验：若窗口超出数据范围，应报 warning

常见错误排查：

• 症状：图上窗口看起来正确，但算法输出切片偏移/空结果
• 原因：算法把 datetime 误判为 numeric，或 numeric 未做单位对齐
• 排查：打印 time mode、time_col dtype、window min/max、data time range

迁移提示（旧写法 -> 新写法）：

• List[List[TimeWindow]] -> Optional[list[TimeWindow]]
• FloatRange（单对象）-> Optional[list[TimeWindow]]
• List[TimeWindow]（非 Optional）-> Optional[list[TimeWindow]]

2.5 定义输出合约 (OutputContract)

创建 schema/output_contract.py 声明算法的输出结构。

OUTPUT_CONTRACT 必须是 Pydantic 模型实例（与 INPUT_SPEC 一样）。SDK 的 validate / compile 调用 .model_dump() 将其序列化为 JSON。不要写 plain dict——会导致 AttributeError: 'dict' object has no attribute 'dict'。

# schema/output_contract.py
from fraclab_sdk.models.output_contract import (
    BlobOutputSchema,
    ObjectOutputSchema,
    OutputContract,
    OutputDatasetContract,
    ScalarOutputSchema,
)

OUTPUT_CONTRACT = OutputContract(
    datasets=[
        OutputDatasetContract(
            key="metrics",
            kind="scalar",
            owner="well",
            cardinality="many",
            required=True,
            dimensions=["stage"],
            schema=ScalarOutputSchema(type="scalar", dtype="float"),
            role="primary",
            description="每个井/阶段的评估指标",
        ),
        OutputDatasetContract(
            key="summary",
            kind="object",
            owner="platform",
            cardinality="one",
            required=True,
            schema=ObjectOutputSchema(type="object"),
            role="primary",
            description="汇总结果",
        ),
        OutputDatasetContract(
            key="debug_plots",
            kind="blob",
            owner="well",
            cardinality="many",
            required=False,
            schema=BlobOutputSchema(type="blob", mime="image/png"),
            role="debug",
            description="调试图表 (可选)",
        ),
    ]
)

常见错误:

# ❌ plain dict → validate 时 AttributeError
OUTPUT_CONTRACT = {"datasets": [...]}

# ❌ BlobOutputSchema 缺少 type → Field required
schema=BlobOutputSchema(mime="image/png")

# ✅ 正确写法
schema=BlobOutputSchema(type="blob", mime="image/png")

Schema 类型一览

导入类	type 值 (必填)	可选字段	对应 kind
ScalarOutputSchema	"scalar"	dtype ("float" / "int" / "string" / "bool"), precision	"scalar"
ObjectOutputSchema	"object"	—	"object"
BlobOutputSchema	"blob"	mime (如 "image/png"), ext (如 ".png")	"blob"
FrameOutputSchema	"frame"	index ("time" / "depth" / "none")	"frame"

所有 schema 类的 type 字段均为 Literal 必填，不提供会触发 Pydantic Field required 错误。

OutputDatasetContract 字段规范

字段	必填	类型	可选值	说明
key	是	str	—	数据集唯一键名
kind	是	Literal	"scalar" / "object" / "blob" / "frame"	数据类型
owner	是	Literal	"stage" / "well" / "platform"	所有者级别
cardinality	否	Literal	"one" / "many"	项目数量约束，默认 "many"
required	否	bool	—	是否必须产出，默认 True
dimensions	否	list[str]	—	维度键列表
schema	是	*OutputSchema	—	上方 Schema 类之一
role	否	Literal	"primary" / "supporting" / "debug"	输出角色
description	否	str	—	描述说明

kind 与 OutputClient 方法对应

kind	OutputClient 方法	说明
"scalar"	write_scalar()	标量值 (数字/字符串/布尔)
"object"	write_json()	JSON 对象
"blob"	write_blob() / write_file()	二进制文件
"frame"	(暂不支持)	表格数据

owner 级别说明

owner	含义	OutputClient owner 参数
"platform"	平台级 (全局)	owner={"platformId": "..."}
"well"	井级	owner={"wellId": "..."}
"stage"	阶段级	owner={"stageId": "..."}

cardinality 约束

cardinality	含义	验证规则
"one"	恰好一个项目	required=True 时必须 1 项; required=False 时最多 1 项
"many"	一个或多个	required=True 时至少 1 项; required=False 时 0 项或多项

dimensions 使用

当数据集有维度约束时，写入时必须提供对应的 dims:

# OutputContract 定义: dimensions=["stage", "iteration"]
out.write_scalar(
    "loss",
    0.05,
    dataset_key="training_metrics",
    owner={"wellId": "W001"},
    dims={"stage": 1, "iteration": 100}  # 必须包含所有定义的维度
)

2.6 创建算法清单

创建 manifest.json — 这是打包、导入、发布的唯一标准清单：

{
  "manifestVersion": "1",
  "algorithmId": "my-algorithm",
  "name": "My Algorithm",
  "summary": "算法简短描述 (必填)",
  "notes": "详细说明、使用注意事项等 (可选)",
  "tags": ["analysis", "well-log"],
  "authors": [
    {
      "name": "张三",
      "email": "[e-mail]",
      "organization": "示例公司"
    }
  ],
  "contractVersion": "1.0.0",
  "codeVersion": "1.0.0",
  "files": {
    "paramsSchemaPath": "dist/params.schema.json",
    "outputContractPath": "dist/output_contract.json",
    "dsPath": "dist/ds.json",
    "drsPath": "dist/drs.json"
  },
  "requires": {
    "sdk": "0.1.65",
    "core": "1.0.0"
  },
  "repository": "https://github.com/example/my-algorithm",
  "homepage": "https://example.com/my-algorithm",
  "license": "MIT"
}

字段规范详解

字段	必填	类型	约束	说明
manifestVersion	是	"1"	固定值	清单版本
algorithmId	是	string	1-128 字符	算法唯一标识符 (用于导入/引用)
name	是	string	1-256 字符	算法显示名称
summary	是	string	1-256 字符	简短描述
notes	否	string	-	详细说明
tags	否	string[]	每项 1-256 字符	标签列表
authors	是	Author[]	至少 1 项	作者列表
authors[].name	是	string	1-256 字符	作者姓名
authors[].email	否	string	3-320 字符	邮箱地址
authors[].organization	否	string	1-256 字符	所属组织
contractVersion	是	string	SemVer 格式	输出合约版本 (如 1.0.0)
codeVersion	是	string	-	代码版本 (用作算法版本标识)
files	是	object	-	产物文件路径 (见下表)
requires	否	object	-	兼容性要求
requires.sdk	否	string	SemVer 格式	SDK 最低版本
requires.core	否	string	SemVer 格式	Core 最低版本
repository	否	string	1-2048 字符	代码仓库 URL
homepage	否	string	1-2048 字符	主页 URL
license	否	string	1-256 字符	许可证标识

files 字段详解

files 用于指定编译产物的位置，导入时 SDK 根据此字段定位文件：

字段	说明
files.paramsSchemaPath	参数 JSON Schema 路径（导入阶段必需）
files.dsPath	DS 文件路径（导出阶段会自动补齐）
files.drsPath	DRS 文件路径（导出阶段会自动补齐）
files.outputContractPath	输出合约路径（建议提供；缺失时运行阶段可跳过合约校验）

路径规则:

• 所有路径均为相对于算法包根目录的路径
• SDK 导入最低要求是 files.paramsSchemaPath
• files.dsPath / files.drsPath / files.outputContractPath 可选（建议提供）
• 推荐使用 dist/ 前缀 (如 dist/params.schema.json)

导入阶段：SDK 最小算法包要求

导入算法包 (zip 或目录) 时，SDK 验证以下文件:

文件	必须	说明
main.py	是	算法入口文件，必须包含 run(ctx) 函数
manifest.json	是	算法清单 (含 files.*Path 字段)
dist/params.schema.json	是	参数 JSON Schema (路径由 files.paramsSchemaPath 指定)
dist/ds.json	否	数据规格（通常在导出阶段从 Bundle 注入）
dist/drs.json	否	数据需求规格（通常在导出阶段从 Bundle 注入）
dist/output_contract.json	否	输出合约（建议提供）

重要: 文件实际位置由 manifest.json 的 files.*Path 字段决定。

导入阶段常见失败原因

1. manifest.json not found: 包内缺少 manifest.json，或 zip 解压后目录结构嵌套
2. main.py not found: 入口文件缺失
3. dist/params.schema.json not found: 未执行 fraclab-sdk algo compile
4. contractVersion must be semver-like: contractVersion 格式错误，应为 x.y.z
5. authors must contain at least one author: authors 列表为空

导出阶段要求（发布包）

fraclab-sdk algo export ... 会要求完整 dist 产物：

• dist/params.schema.json
• dist/output_contract.json
• dist/ds.json
• dist/drs.json

推荐使用：

fraclab-sdk algo export ./my-algorithm ./my-algorithm.zip --auto-compile --bundle /path/to/bundle

导出页会从所选 Bundle 注入 ds/drs，并自动补齐 files.dsPath / files.drsPath（若缺失）。

2.6.1 项目结构

完整的算法工作区结构：

my-algorithm/
├── manifest.json           # 算法清单 (导出包必须)
├── main.py                 # 算法入口 (必须包含 run 函数)
├── schema/
│   ├── __init__.py
│   ├── inputspec.py        # INPUT_SPEC 定义
│   └── output_contract.py  # OUTPUT_CONTRACT 定义
├── lib/                    # 可选: 算法依赖模块
│   └── utils.py
└── dist/                   # 编译产物 (自动生成)
    ├── params.schema.json  # 从 INPUT_SPEC 编译
    ├── output_contract.json # 从 OUTPUT_CONTRACT 编译
    ├── ds.json             # 从 Bundle 复制
    └── drs.json            # 从 Bundle 复制

导出后的算法包结构 (zip 内或目录):

my-algorithm.zip/
├── manifest.json           # 必须: 算法清单 (含 files.*Path)
├── main.py                 # 必须: 入口文件
├── dist/                   # 编译产物目录
│   ├── params.schema.json  # 必须: 参数 Schema
│   ├── ds.json             # 必须: 数据规格
│   ├── drs.json            # 必须: 数据需求规格
│   └── output_contract.json # 必须: 输出合约
├── schema/                 # 可选: schema 源码
│   ├── __init__.py
│   ├── inputspec.py
│   └── output_contract.py
└── README.md               # 可选: 说明文件

2.7 用 Bundle 编译并导出算法包

推荐命令（单命令闭环）：

fraclab-sdk algo export ./my-algorithm ./my-algorithm.zip --auto-compile --bundle /path/to/bundle

为什么推荐这一条：

1. 最终可导入运行的产物必须是包含 dist/*.json 的算法包
2. export 只负责打包，不会凭空生成缺失的 dist/*.json
3. --auto-compile --bundle 能在需要时先补齐 dist/*.json，再打包

等价的两步写法（功能上成立，但更容易漏步骤）：

fraclab-sdk algo compile ./my-algorithm --bundle /path/to/bundle
fraclab-sdk algo export ./my-algorithm ./my-algorithm.zip

注意：

• 如果你已经有有效的 dist/ds.json 与 dist/drs.json，compile 可以不传 --bundle
• 但对首次构建，通常都应显式传 --bundle

2.8 导入并运行算法包

# 1) 导入 Bundle -> Snapshot
fraclab-sdk snapshot import /path/to/bundle

# 2) 导入算法 zip
fraclab-sdk algo import ./my-algorithm.zip

# 3) 创建并执行 run
fraclab-sdk run create <snapshot_id> <algorithm_id> <version> --params params.json
fraclab-sdk run exec <run_id> --timeout 300

# 4) 查看结果
fraclab-sdk results list <run_id>

2.9 SelectionModel 与 run_ds 的关系

这是 run 侧最容易混淆的点：

• SelectionModel：你给 SDK 的“选择意图”（基于 snapshot 索引选哪些 item）
• run_ds：SDK 根据 selection 生成的“运行时 DataSpec 子集”（会重建为 0..N-1，并保留 sourceItemIndex 映射）

关系链（代码路径）：

1. selection = SelectionModel.from_snapshot_and_drs(snapshot, algorithm.drs)
2. 你调用 run_mgr.create_run(..., selection=selection, ...)
3. create_run() 内部调用 selection.build_run_ds()
4. Materializer 用 run_ds 物化 runs/<run_id>/input/

结论：

• SDK 调用方通常只需要管理 SelectionModel，不用手动传 run_ds
• run_ds 是运行前的内部物化输入模型；你可在调试/诊断时显式查看它

---

3. Bundle 与 Snapshot（概念与关系）

3.1 Bundle 是什么

Bundle 是平台提供的原始数据包目录，至少包含：

• manifest.json
• ds.json
• drs.json
• data/

Bundle 的两个用途：

1. 给算法编译提供 DS/DRS（拷贝到 dist/ds.json、dist/drs.json）
2. 导入 SDK 后生成 Snapshot，供 run 选择与执行

3.2 Snapshot 是什么

Snapshot 是 Bundle 导入 SDK 后形成的内部快照副本（默认在 ~/.fraclab/snapshots/<snapshot_id>）。

• 导入命令：fraclab-sdk snapshot import /path/to/bundle
• 运行时你用的是 snapshot_id，不是原始 bundle 路径
• run create 阶段会基于 snapshot + selection 生成 run 输入

3.3 Bundle / Snapshot / RunInput 的关系

Bundle (原始数据包)
  -> snapshot import
Snapshot (SDK库内快照)
  + SelectionModel (选中的 snapshot item 索引)
  -> build_run_ds()
Run Input (runs/<run_id>/input: ds.json/drs.json/data/)

其中 run_ds 是 “Run Input 里的 ds.json 对应对象”，不是 Bundle 的 ds.json 原样复制。

3.4 为什么不能随便改 Bundle

Snapshot 导入与校验依赖 manifest.json 中的哈希字段（dsSha256、drsSha256）。 任何对 ds.json/drs.json 的手工修改都会触发校验失败。

fraclab-sdk validate bundle /path/to/bundle

常见错误：

• ds.json hash mismatch: 文件被改动或损坏
• drs.json not found: Bundle 不完整
• manifest.json not found: 非有效 Bundle 目录

详细目录与字段见 11. 附录 A: Bundle 结构详解

---

4. 算法开发详解

DataClient - 读取输入数据

DataClient 提供统一的数据读取接口。

from fraclab_sdk.runtime import DataClient
from pathlib import Path

dc = DataClient(Path("input"))

获取数据集信息

# 获取所有数据集键
keys = dc.get_dataset_keys()  # ["wells", "frames", ...]

# 获取数据集中的项目数量
count = dc.get_item_count("wells")  # 10

# 获取数据集布局类型
layout = dc.get_layout("wells")  # "object_ndjson_lines" 或 "frame_parquet_item_dirs"

读取 NDJSON 数据

用于 layout="object_ndjson_lines" 的数据集：

# 读取单个对象 (按索引)
obj = dc.read_object("wells", 0)  # 返回 dict

# 迭代所有对象
for idx, obj in dc.iterate_objects("wells"):
    print(f"Item {idx}: {obj}")

读取 Parquet 数据

用于 layout="frame_parquet_item_dirs" 的数据集：

# 获取 parquet 文件目录
parquet_dir = dc.get_parquet_dir("frames", 0)

# 获取所有 parquet 文件列表
parquet_files = dc.get_parquet_files("frames", 0)

# 使用 pandas 读取
import pandas as pd
df = pd.read_parquet(parquet_dir)

注意: Parquet item 目录命名为 item-{index:05d}（5 位零填充）。当结合时间窗使用时， timeWindows 中的 itemKey（如 "00000"、"00004"）直接对应目录名中的编号。 若选择器选择了稀疏 item（如 00/04/05 而非 00/01/02），算法应按 itemKey 定位 对应的 item-00000/、item-00004/、item-00005/ 目录，不可假设 item 连续。

OutputClient - 写入输出结果

OutputClient 提供统一的算法输出入口，固定根目录为 run/output/artifacts，并自动防止路径逃逸攻击。
算法代码里始终使用运行时注入的 ctx.output，不要自己实例化 OutputClient，也不要手写或修改 manifest.json。

• ctx.output.dir: 运行时已经准备好的唯一可写输出目录。服务器侧只保证这个固定目录可写，不要在算法里创建子目录。
• write_*(): 写入并显式登记结果，适合需要自定义 dataset_key / item_key / owner / dims / meta 的输出。
• 直接写到 ctx.output.dir 下、但未显式登记的文件，也会在 runner 结束时自动纳入 output/manifest.json，默认归到 artifacts 数据集。

最常用写法

def run(ctx):
    out = ctx.output

    out.write_scalar("score", 0.95)
    out.write_json("summary", {"status": "ok"})

    plot_path = out.dir / "result_plot.png"
    fig.savefig(plot_path, dpi=200)
    # 这个 PNG 会在运行结束后被 SDK 自动收进 output/manifest.json

    report_path = out.dir / "report.csv"
    df.to_csv(report_path, index=False)
    out.register_file(
        "report",
        report_path,
        mime_type="text/csv",
        dataset_key="reports",
    )

写入标量值

# 基本用法
out.write_scalar("score", 0.95)
out.write_scalar("count", 42)
out.write_scalar("name", "result_a")

# 指定数据集和所有者
out.write_scalar(
    "accuracy",
    0.87,
    dataset_key="metrics",
    owner={"wellId": "W001"},
    dims={"stage": 1},
    meta={"unit": "percent"}
)

写入 JSON

# 基本用法
path = out.write_json("metrics", {"accuracy": 0.95, "loss": 0.05})

# 自定义文件名
path = out.write_json("results", data, filename="analysis_results.json")

# 完整参数
path = out.write_json(
    "summary",
    {"status": "ok"},
    filename="summary.json",
    dataset_key="outputs",
    owner={"platformId": "P001"}
)

写入二进制文件

# 写入字节数据
image_bytes = generate_plot()
path = out.write_blob(
    "plot",
    image_bytes,
    "plot.png",
    mime_type="image/png"
)

# 复制现有文件
path = out.write_file(
    "report",
    Path("/tmp/generated_report.pdf"),
    filename="report.pdf",
    mime_type="application/pdf"
)

系统自动生成的 Output Manifest 参考

output/manifest.json 由 SDK 在 run(ctx) 完成后自动生成。
下面这部分只是帮助你理解 ctx.output.write_*() 如何映射到结果清单，不是让你在算法里手写 manifest。

参数映射表

OutputClient 参数	manifest.json 字段	OutputContract 字段	说明
artifact_key	artifact.artifactKey	-	制品唯一标识
dataset_key	datasetKey	datasets[].key	数据集键，默认 "artifacts"
owner	item.owner	datasets[].owner	所有者: {wellId, stageId, platformId}
dims	item.dims	datasets[].dimensions	维度值字典
meta	item.meta	-	元数据 (manifest 专用)
item_key	item.itemKey	-	项目键，默认等于 artifact_key
(写入类型)	artifact.type	datasets[].kind	制品类型

写入操作到 Manifest 的转换

# 算法代码中的写入
out.write_scalar(
    "accuracy",           # artifact_key
    0.95,                 # value
    dataset_key="metrics",
    owner={"wellId": "W001"},
    dims={"stage": 1},
    meta={"unit": "percent"}
)

SDK 自动生成的 output/manifest.json 片段:

{
  "datasets": [
    {
      "datasetKey": "metrics",
      "items": [
        {
          "itemKey": "accuracy",
          "owner": { "wellId": "W001" },
          "dims": { "stage": 1 },
          "meta": { "unit": "percent" },
          "artifact": {
            "artifactKey": "accuracy",
            "type": "scalar",
            "value": 0.95
          }
        }
      ]
    }
  ]
}

类型对应关系

写入方法	manifest artifact.type	OutputContract kind	说明
write_scalar()	"scalar"	"scalar"	标量值 (直接存 value)
write_json()	"json"	"object"	JSON 对象 (存 uri)
write_blob()	"blob"	"blob"	二进制文件 (存 uri + mimeType)
write_file()	"blob"	"blob"	复制文件 (存 uri + mimeType)

OutputContract 定义与 OutputClient 使用示例

OutputContract 定义 (schema/output_contract.py):

from fraclab_sdk.models.output_contract import (
    BlobOutputSchema,
    OutputContract,
    OutputDatasetContract,
    ScalarOutputSchema,
)

OUTPUT_CONTRACT = OutputContract(
    datasets=[
        OutputDatasetContract(
            key="metrics",
            kind="scalar",
            owner="well",
            cardinality="many",
            dimensions=["stage"],
            schema=ScalarOutputSchema(type="scalar", dtype="float"),
        ),
        OutputDatasetContract(
            key="reports",
            kind="blob",
            owner="well",
            cardinality="one",
            schema=BlobOutputSchema(type="blob", mime="application/pdf"),
        ),
    ]
)

对应的算法写入代码:

def run(ctx):
    out = ctx.output

    # 符合 "metrics" 数据集定义
    # owner="well" → 必须提供 wellId
    # dimensions=["stage"] → dims 必须包含 stage 键
    out.write_scalar(
        "accuracy",
        0.95,
        dataset_key="metrics",
        owner={"wellId": "W001"},
        dims={"stage": 1}
    )

    # 符合 "reports" 数据集定义
    # cardinality="one" → 该数据集只能有一个项目
    out.write_file(
        "report",
        Path("/tmp/report.pdf"),
        dataset_key="reports",
        owner={"wellId": "W001"},
        mime_type="application/pdf"
    )

验证输出与合约一致性

# 验证运行输出是否符合合约
fraclab-sdk validate run-manifest output/manifest.json --contract dist/output_contract.json

验证检查项:

• 合约中所有 required=true 的数据集必须存在
• 数据集的 cardinality 约束 (one/many)
• owner 类型匹配 (well/stage/platform)
• dimensions 键集合匹配
• kind 与 artifact.type 兼容

日志记录

使用 ctx.logger 记录日志：

def run(ctx):
    logger = ctx.logger

    logger.debug("调试信息")
    logger.info("常规信息")
    logger.warning("警告信息")
    logger.error("错误信息")

日志会同时输出到：

• 控制台 (INFO 及以上级别)
• output/_logs/algorithm.log 文件 (DEBUG 及以上级别)

---

5. CLI 命令行工具

安装后可使用 fraclab-sdk 命令。

运行闭环黄金路径

以下是从导入到执行的完整流程示例。

1. 导入快照和算法

# 导入数据快照 (Bundle)
$ fraclab-sdk snapshot import /path/to/my-bundle
Imported snapshot: a1b2c3d4

# 导入算法包
$ fraclab-sdk algo import ./my-algorithm.zip
Imported algorithm: my-algorithm:1.0.0

2. 查看已导入资源

# 列出快照
$ fraclab-sdk snapshot list
a1b2c3d4    my-bundle-v1    2024-01-15T10:30:00
e5f6g7h8    test-bundle     2024-01-14T09:00:00

# 列出算法
$ fraclab-sdk algo list
my-algorithm    1.0.0    2024-01-15T11:00:00
other-algo      2.1.0    2024-01-10T08:00:00

ID 格式说明:

• snapshot_id: 8 位十六进制字符串 (如 a1b2c3d4)
• algorithm_id: 算法的 algorithmId 字段值 (如 my-algorithm)
• version: 算法的 codeVersion 字段值，遵循 SemVer (如 1.0.0)

3. 准备参数文件

创建 params.json:

{
  "threshold": 0.8,
  "debug": false,
  "outputFormat": "detailed",
  "filters": {
    "minDepth": 1000,
    "maxDepth": 5000
  }
}

参数文件要求:

• JSON 格式，编码 UTF-8
• 键名对应 InputSpec 中定义的字段
• 未提供的字段使用 InputSpec 中的默认值

4. 创建并执行运行

# 创建运行 (自动选择所有数据项)
$ fraclab-sdk run create a1b2c3d4 my-algorithm 1.0.0 --params params.json
f9e8d7c6

# 执行运行
$ fraclab-sdk run exec f9e8d7c6 --timeout 300
succeeded (exit_code=0)

5. 查看结果

# 列出产出的制品
$ fraclab-sdk results list f9e8d7c6
Status: succeeded
accuracy    scalar
summary     json      file:///Users/.../output/artifacts/summary.json
report      blob      file:///Users/.../output/artifacts/report.pdf

# 查看运行日志
$ fraclab-sdk run tail f9e8d7c6
[INFO] 2024-01-15 12:00:00 - 开始处理，阈值: 0.8
[INFO] 2024-01-15 12:00:01 - 数据集 wells 包含 3 个项目
[INFO] 2024-01-15 12:00:05 - 算法执行完成

# 查看错误日志 (如有)
$ fraclab-sdk run tail f9e8d7c6 --stderr

Workbench 提示：结果页面会展示本次运行的输出目录路径（含 _logs 日志），即使运行失败也能点开路径定位调试。

6. 运行目录结构

执行完成后，~/.fraclab/runs/<run_id>/ 目录结构:

f9e8d7c6/
├── run_meta.json              # 运行元数据
├── input/                     # 输入目录 (物化后的数据)
│   ├── manifest.json          # 输入清单 (含哈希)
│   ├── ds.json                # 运行数据规格 (重新索引)
│   ├── drs.json               # 算法 DRS
│   ├── params.json            # 用户参数
│   ├── run_context.json       # 运行上下文
│   └── data/                  # 数据目录
│       └── wells/
│           └── object.ndjson
└── output/                    # 输出目录
    ├── manifest.json          # 输出清单 ★ 核心结果文件
    ├── artifacts/             # 制品文件目录
    │   ├── summary.json
    │   └── report.pdf
    ├── _logs/                 # 日志目录
    │   ├── stdout.log         # 标准输出
    │   ├── stderr.log         # 标准错误
    │   ├── algorithm.log      # 算法日志 (DEBUG 级别)
    │   └── execute.json       # 执行元数据

7. 输出 manifest.json 完整示例

{
  "schemaVersion": "1.0",
  "run": {
    "runId": "f9e8d7c6",
    "algorithmId": "my-algorithm",
    "contractVersion": "1.0.0",
    "codeVersion": "1.0.0"
  },
  "status": "succeeded",
  "startedAt": "2024-01-15T12:00:00.000Z",
  "completedAt": "2024-01-15T12:00:05.123Z",
  "datasets": [
    {
      "datasetKey": "artifacts",
      "items": [
        {
          "itemKey": "accuracy",
          "artifact": {
            "artifactKey": "accuracy",
            "type": "scalar",
            "value": 0.95
          }
        },
        {
          "itemKey": "summary",
          "artifact": {
            "artifactKey": "summary",
            "type": "json",
            "mimeType": "application/json",
            "uri": "file:///Users/.../output/artifacts/summary.json"
          }
        }
      ]
    },
    {
      "datasetKey": "reports",
      "items": [
        {
          "itemKey": "report",
          "owner": { "wellId": "W001" },
          "artifact": {
            "artifactKey": "report",
            "type": "blob",
            "mimeType": "application/pdf",
            "uri": "file:///Users/.../output/artifacts/report.pdf"
          }
        }
      ]
    }
  ]
}

算法开发命令

编译算法

# 编译算法工作区
fraclab-sdk algo compile ./my-algorithm --bundle /path/to/bundle

# 生成:
# - dist/params.schema.json (从 schema.inputspec:INPUT_SPEC)
# - dist/output_contract.json (从 schema.output_contract:OUTPUT_CONTRACT)
# - dist/ds.json (从 bundle 复制)
# - dist/drs.json (从 bundle 复制)

可选标志：

标志	说明
--bundle, -b	Bundle 路径，用于复制 ds.json / drs.json
--skip-inputspec	跳过 InputSpec 编译（不生成 params.schema.json）
--skip-output-contract	跳过 OutputContract 编译（不生成 output_contract.json）

初始化算法工作区

# 创建本地算法脚手架（默认写入 ~/.fraclab/workspace_algorithms）
fraclab-sdk algo init my-algorithm --code-version 0.1.0 --contract-version 1.0.0

# 常用可选参数
fraclab-sdk algo init my-algorithm \
  --name "My Algorithm" \
  --summary "Algorithm summary" \
  --author-name "Your Name" \
  --author-email "[e-mail]" \
  --tag test --tag smoke

导出算法包

# 导出为 zip 包
fraclab-sdk algo export ./my-algorithm ./my-algorithm.zip

# 自动编译后导出
fraclab-sdk algo export ./my-algorithm ./my-algorithm.zip --auto-compile --bundle /path/to/bundle

导入算法

# 导入算法到 SDK 库
fraclab-sdk algo import ./my-algorithm.zip

# 列出已导入的算法
fraclab-sdk algo list

验证命令

# 验证 InputSpec
fraclab-sdk validate inputspec ./my-algorithm

# 验证 OutputContract
fraclab-sdk validate output-contract ./my-algorithm

# 验证 Bundle 完整性
fraclab-sdk validate bundle /path/to/bundle

# 验证运行输出清单
fraclab-sdk validate run-manifest /path/to/manifest.json --contract /path/to/contract.json

快照管理命令

# 导入数据快照
fraclab-sdk snapshot import /path/to/bundle

# 列出已导入快照
fraclab-sdk snapshot list

运行管理命令

# 创建运行
fraclab-sdk run create <snapshot_id> <algorithm_id> <version> --params params.json

# 执行运行
fraclab-sdk run exec <run_id> --timeout 300

# 查看日志
fraclab-sdk run tail <run_id>
fraclab-sdk run tail <run_id> --stderr

结果查看命令

# 列出运行产出的制品
fraclab-sdk results list <run_id>

调试模式

# 显示完整堆栈跟踪
fraclab-sdk --debug <command>

---

6. SDK 内部模块

以下模块供进阶使用或二次开发。

SDKConfig - 配置管理

from fraclab_sdk import SDKConfig

# 使用默认路径 (~/.fraclab)
config = SDKConfig()

# 显式指定路径
config = SDKConfig(sdk_home="/custom/path")

# 通过环境变量: export FRACLAB_SDK_HOME=/custom/path
config = SDKConfig()  # 自动读取

属性：

属性	类型	说明
sdk_home	Path	SDK 根目录
snapshots_dir	Path	快照存储目录
algorithms_dir	Path	算法存储目录
runs_dir	Path	运行存储目录

SnapshotLibrary - 快照管理

from fraclab_sdk import SnapshotLibrary, SDKConfig

lib = SnapshotLibrary(SDKConfig())

# 导入快照
snapshot_id = lib.import_snapshot("/path/to/bundle")

# 列出快照
snapshots = lib.list_snapshots()

# 获取快照句柄
snapshot = lib.get_snapshot(snapshot_id)

# 删除快照
lib.delete_snapshot(snapshot_id)

SnapshotHandle - 快照访问

snapshot = lib.get_snapshot(snapshot_id)

# 属性
snapshot.directory    # Path: 快照目录
snapshot.manifest     # BundleManifest: 清单
snapshot.dataspec     # DataSpec: 数据规格
snapshot.drs          # DRS: 数据需求规格
snapshot.data_root    # Path: 数据目录

# 方法
datasets = snapshot.get_datasets()                      # 所有数据集
items = snapshot.get_items("wells")                     # 数据集的所有项目
data = snapshot.read_object_line("wells", 0)            # 读取 NDJSON 行
item_dir = snapshot.get_item_dir("frames", 0)           # 获取项目目录
parts = snapshot.read_frame_parts("frames", 0)          # Parquet 分片列表

AlgorithmLibrary - 算法管理

from fraclab_sdk import AlgorithmLibrary, SDKConfig

lib = AlgorithmLibrary(SDKConfig())

# 导入算法
algorithm_id, version = lib.import_algorithm("/path/to/algo.zip")

# 列出算法
algorithms = lib.list_algorithms()

# 获取算法句柄
algorithm = lib.get_algorithm(algorithm_id, version)

# 删除算法
lib.delete_algorithm(algorithm_id, version)

AlgorithmHandle - 算法访问

algorithm = lib.get_algorithm(algorithm_id, version)

# 属性
algorithm.directory      # Path: 算法目录
algorithm.manifest       # AlgorithmManifest: 清单
algorithm.drs            # DRS: 数据需求规格
algorithm.params_schema  # dict: 参数 JSON Schema
algorithm.algorithm_path # Path: main.py 路径

SelectionModel - 数据选择

from fraclab_sdk import SelectionModel

# 从快照和 DRS 创建选择模型
selection = SelectionModel.from_snapshot_and_drs(snapshot, algorithm.drs)

# 获取可选数据集
for ds in selection.get_selectable_datasets():
    print(f"{ds.dataset_key}: 共 {ds.total_items} 项, 基数要求: {ds.cardinality}")

# 设置选择
selection.set_selected("wells", [0, 1, 2])

# 获取当前选择
selected = selection.get_selected("wells")  # [0, 1, 2]

# 验证选择
errors = selection.validate()
if not errors:
    print("选择有效")

# 构建运行数据规格 (重新索引)
run_ds = selection.build_run_ds()

# 获取索引映射 (run_index -> snapshot_index)
mapping = selection.get_selection_mapping("wells")

基数规则：

• "one": 必须恰好选择 1 个
• "many": 必须至少选择 1 个
• "zeroOrMany": 可以选择 0 个或多个

RunManager - 运行管理

from fraclab_sdk import RunManager, SDKConfig

mgr = RunManager(SDKConfig())

# 创建运行
run_id = mgr.create_run(
    snapshot_id=snapshot_id,
    algorithm_id=algorithm_id,
    algorithm_version=version,
    selection=selection,
    params={"threshold": 0.8}
)

# 执行运行
result = mgr.execute(run_id, timeout_s=300)
print(f"状态: {result.status}")
print(f"退出码: {result.exit_code}")

# 查询状态
status = mgr.get_run_status(run_id)

# 列出运行
runs = mgr.list_runs()

# 获取运行目录
run_dir = mgr.get_run_dir(run_id)

# 删除运行
mgr.delete_run(run_id)

ResultReader - 结果读取

from fraclab_sdk import ResultReader

reader = ResultReader(run_dir)

# 检查清单
if reader.has_manifest():
    manifest = reader.read_manifest()
    print(f"状态: {manifest.status}")

# 列出制品
artifacts = reader.list_artifacts()
for art in artifacts:
    print(f"{art.artifactKey}: {art.type}")

# 获取制品
artifact = reader.get_artifact("score")
path = reader.get_artifact_path("metrics")
data = reader.read_artifact_json("metrics")
value = reader.read_artifact_scalar("score")

# 读取日志
stdout = reader.read_stdout()
stderr = reader.read_stderr()
algo_log = reader.read_algorithm_log()

Devkit - 开发工具

from fraclab_sdk.devkit import (
    compile_algorithm,
    export_algorithm_package,
    validate_algorithm_signature,
    validate_inputspec,
    validate_output_contract,
    validate_bundle,
    validate_run_manifest,
)

# 编译
result = compile_algorithm(
    workspace="/path/to/workspace",
    bundle_path="/path/to/bundle",
)

# 导出
result = export_algorithm_package(
    workspace="/path/to/workspace",
    output="/path/to/output.zip",
    auto_compile=True,
    bundle_path="/path/to/bundle",
)

# 验证
result = validate_algorithm_signature("/path/to/workspace")  # 检查 main.py 的 run() 签名
result = validate_inputspec("/path/to/workspace")
result = validate_output_contract("/path/to/workspace")
result = validate_bundle("/path/to/bundle")
result = validate_run_manifest(
    manifest_path="/path/to/manifest.json",
    contract_path="/path/to/contract.json"
)

if result.valid:
    print("验证通过")
else:
    for issue in result.errors:
        print(f"[{issue.code}] {issue.path}: {issue.message}")

---

7. 数据模型

DRS (Data Requirement Specification)

算法对输入数据的需求定义。

from fraclab_sdk.models import DRS

drs = DRS.model_validate_json(json_string)
dataset = drs.get_dataset("wells")
print(dataset.cardinality)  # "one", "many", "zeroOrMany"

DataSpec

数据规格定义，描述快照中的数据集结构。

from fraclab_sdk.models import DataSpec

ds = DataSpec.model_validate_json(json_string)
dataset = ds.get_dataset("wells")
keys = ds.get_dataset_keys()

OutputContract

算法输出合约，声明算法产出的数据结构。

from fraclab_sdk.models import OutputContract

contract = OutputContract.model_validate_json(json_string)
dataset = contract.get_dataset("results")  # OutputDatasetContract | None
print(contract.datasets)                   # list[OutputDatasetContract]

RunOutputManifest

运行输出清单，记录算法执行的结果。

from fraclab_sdk.models import RunOutputManifest

manifest = RunOutputManifest.model_validate_json(json_string)
artifact = manifest.get_artifact("score")
all_artifacts = manifest.list_all_artifacts()

---

8. 错误处理

异常类型

from fraclab_sdk.errors import (
    FraclabError,          # 基类
    SnapshotError,         # 快照相关
    HashMismatchError,     # 哈希不匹配
    PathTraversalError,    # 路径穿越攻击
    AlgorithmError,        # 算法相关
    SelectionError,        # 选择相关
    DatasetKeyError,       # 数据集键不存在
    CardinalityError,      # 基数验证失败
    MaterializeError,      # 物化错误
    RunError,              # 运行相关
    TimeoutError,          # 执行超时
    ResultError,           # 结果读取
    OutputContainmentError,# 输出路径逃逸
)

退出码

from fraclab_sdk.errors import ExitCode

ExitCode.SUCCESS        # 0: 成功
ExitCode.GENERAL_ERROR  # 1: 一般错误
ExitCode.INPUT_ERROR    # 2: 输入/验证错误
ExitCode.RUN_FAILED     # 3: 运行失败
ExitCode.TIMEOUT        # 4: 超时
ExitCode.INTERNAL_ERROR # 5: 内部错误

错误处理示例

from fraclab_sdk.errors import FraclabError, HashMismatchError, CardinalityError

try:
    snapshot_id = lib.import_snapshot(path)
except HashMismatchError as e:
    print(f"文件 {e.file_name} 哈希不匹配")
    print(f"预期: {e.expected}")
    print(f"实际: {e.actual}")
except FraclabError as e:
    print(f"SDK 错误 (退出码={e.exit_code}): {e}")

---

9. 安全特性

SDK 内置多项安全机制：

1. 路径穿越防护: 导入时验证所有路径，拒绝 .. 和绝对路径
2. 哈希验证: 验证 ds.json 和 drs.json 的 SHA256 哈希
3. 输出隔离: 算法只能写入指定的 output 目录
4. 进程隔离: 算法在子进程中执行，有超时控制
5. 原子写入: 使用 tmp + rename 确保文件完整性
6. 符号链接检查: 导出算法包时验证符号链接不指向工作区外部

环境变量汇总

变量	默认值	说明
FRACLAB_SDK_HOME	~/.fraclab	SDK 数据根目录（快照、算法、运行）
FRACLAB_WORKBENCH_LANG	zh-CN	Workbench 初始 UI 语言（en / zh-CN）
FRACLAB_WORKBENCH_UPLOAD_MB	2048	Workbench 最大上传文件大小 (MB)
FRACLAB_WORKBENCH_MESSAGE_MB	2048	Workbench WebSocket 最大消息大小 (MB)

---

10. Workbench 图形界面

Workbench 是基于 Streamlit 的本地图形界面，提供算法开发、数据浏览、运行管理的全流程可视化操作。

启动

# CLI 入口
fraclab-workbench

# 或直接调用模块
python -m fraclab_sdk.workbench

环境变量

变量	默认值	说明
FRACLAB_WORKBENCH_LANG	zh-CN	初始 UI 语言（en 或 zh-CN）
FRACLAB_WORKBENCH_UPLOAD_MB	2048	最大上传文件大小 (MB)
FRACLAB_WORKBENCH_MESSAGE_MB	2048	Streamlit WebSocket 最大消息大小 (MB)

双语支持

Workbench 内置英文 / 简体中文双语支持：

• 默认语言为 简体中文（zh-CN）
• 可在每个页面右上角的语言选择器中切换语言，切换后会自动持久化，并在刷新后保持
• 也可通过环境变量 FRACLAB_WORKBENCH_LANG=en 在首次启动时设为英文
• 内部使用 tx(en, zh) 函数实现翻译，所有 UI 文案均通过该函数双语化

页面说明

页面	中文名	功能
Home	工作台总览	显示 SDK 概览：快照数、算法数、运行数统计
1. Snapshots	快照管理	导入 Bundle/ZIP、列出快照、删除快照
2. Browse	数据浏览	浏览快照内的数据集，预览 NDJSON / Parquet 数据及图表
3. Selection	运行配置	选择快照、算法，配置数据项选择和参数，包含时间窗选择器
4. Run	运行管理	创建和执行运行，查看运行状态和日志
5. Results	运行结果	查看运行产出的制品（标量、JSON、图片），浏览输出目录
6. Algorithm Edit	算法代码编辑	在线编辑 main.py 算法入口代码
7. Schema Edit	输入参数编辑	在线编辑 schema/inputspec.py 和 schema/base.py
8. Output Edit	输出结果定义	在线编辑 schema/output_contract.py
9. Export Algorithm	算法打包导出	编译并导出算法为 ZIP 包，可选绑定 Bundle

数据预览

Browse 页面支持两种数据布局的预览：

• object_ndjson_lines: 以表格形式展示 NDJSON 对象
• frame_parquet_item_dirs: 读取 Parquet 文件并自动绘制时序图表

Parquet 预览会自动识别时间列（优先级：timestamp > bucket > ts > ts_us > time > datetime > date），推断时间单位（ns/us/ms/s），并对大数据集进行降采样以保证显示性能。

---

11. 完整示例

算法开发完整流程

# 1. 编写算法
# main.py
def run(ctx):
    dc = ctx.data_client
    out = ctx.output

    for key in dc.get_dataset_keys():
        for idx, obj in dc.iterate_objects(key):
            result = analyze(obj)
            out.write_scalar(f"result_{key}_{idx}", result)

    out.write_json("summary", {"completed": True})

# 2. 定义规格
# schema/inputspec.py
from .base import CamelModel, Field

class Params(CamelModel):
    threshold: float = Field(default=0.5)

INPUT_SPEC = Params

# 3. 编译并导出
# $ fraclab-sdk algo compile ./my-algo --bundle /path/to/bundle
# $ fraclab-sdk algo export ./my-algo ./my-algo.zip

使用 SDK 执行算法

from fraclab_sdk import (
    SDKConfig,
    SnapshotLibrary,
    AlgorithmLibrary,
    SelectionModel,
    RunManager,
    ResultReader,
)

config = SDKConfig()

# 导入资源
snap_lib = SnapshotLibrary(config)
algo_lib = AlgorithmLibrary(config)
run_mgr = RunManager(config)

snapshot_id = snap_lib.import_snapshot("/path/to/bundle")
algorithm_id, version = algo_lib.import_algorithm("/path/to/algo.zip")

# 创建选择
snapshot = snap_lib.get_snapshot(snapshot_id)
algorithm = algo_lib.get_algorithm(algorithm_id, version)
selection = SelectionModel.from_snapshot_and_drs(snapshot, algorithm.drs)
selection.set_selected("wells", [0, 1, 2])

# 执行
run_id = run_mgr.create_run(
    snapshot_id, algorithm_id, version,
    selection, {"threshold": 0.8}
)
result = run_mgr.execute(run_id)

# 读取结果
reader = ResultReader(run_mgr.get_run_dir(run_id))
for art in reader.list_artifacts():
    print(f"{art.artifactKey}: {art.type}")

---

12. 附录 A: Bundle 结构详解

本节面向平台开发者或需要排查导入问题的用户。普通算法开发者无需了解这些细节。

最小可用 Bundle 目录结构

my-bundle/
├── manifest.json          # 必须: 清单文件 (含哈希校验)
├── ds.json                # 必须: 数据规格 (DataSpec)
├── drs.json               # 必须: 数据需求规格 (DRS)
└── data/                  # 必须: 数据目录
    └── wells/             # 数据集目录 (datasetKey)
        └── object.ndjson  # NDJSON 格式数据文件

manifest.json 完整示例

{
  "bundleVersion": "1.0.0",
  "createdAtUs": 1706000000000000,
  "specFiles": {
    "dsPath": "ds.json",
    "drsPath": "drs.json",
    "dsSha256": "a1b2c3d4e5f6...(64位十六进制)",
    "drsSha256": "f6e5d4c3b2a1...(64位十六进制)"
  },
  "dataRoot": "data",
  "datasets": {
    "wells": {
      "layout": "object_ndjson_lines",
      "count": 3
    }
  }
}

字段说明:

字段	必填	说明
bundleVersion	是	固定值 "1.0.0"
createdAtUs	是	创建时间 (微秒时间戳)
specFiles.dsPath	是	ds.json 相对路径，默认 "ds.json"
specFiles.drsPath	是	drs.json 相对路径，默认 "drs.json"
specFiles.dsSha256	是	ds.json 的 SHA256 哈希
specFiles.drsSha256	是	drs.json 的 SHA256 哈希
dataRoot	是	数据目录相对路径，默认 "data"
datasets	是	数据集清单，key 为数据集标识
datasets.*.layout	是	"object_ndjson_lines" 或 "frame_parquet_item_dirs"
datasets.*.count	是	数据集中的项目数量

ds.json (DataSpec) 完整示例

{
  "schemaVersion": "1.0.0",
  "datasets": [
    {
      "key": "wells",
      "resource": "well",
      "layout": "object_ndjson_lines",
      "items": [
        { "owner": { "wellId": "W001" } },
        { "owner": { "wellId": "W002" } },
        { "owner": { "wellId": "W003" } }
      ]
    }
  ]
}

字段说明:

字段	必填	说明
datasets[].key	是	数据集键名，对应 data/ 下的目录名
datasets[].resource	否	资源类型标识
datasets[].layout	否	数据布局，决定数据存储格式
datasets[].items	否	项目列表，每个元素对应一条数据
datasets[].items[].owner	否	所有者标识，包含 platformId/wellId/stageId

drs.json (DRS) 完整示例

{
  "schemaVersion": "1.0.0",
  "datasets": [
    {
      "key": "wells",
      "resource": "well",
      "cardinality": "many",
      "description": "井数据集"
    }
  ]
}

字段说明:

字段	必填	说明
datasets[].key	是	数据集键名，必须与 ds.json 中的 datasets[].key 匹配
datasets[].resource	否	资源类型标识
datasets[].cardinality	是	基数要求: "one" / "many" / "zeroOrMany"
datasets[].description	否	数据集描述

基数规则:

• "one": 必须恰好选择 1 个项目
• "many": 必须至少选择 1 个项目
• "zeroOrMany": 可以选择 0 个或多个项目

数据目录布局

布局 1: object_ndjson_lines (NDJSON 格式)

data/
└── wells/
    ├── object.ndjson      # 必须: 每行一个 JSON 对象
    └── object.idx.u64     # 可选: 索引文件 (加速随机访问)

object.ndjson 示例 (每行一个 JSON 对象):

{"wellId": "W001", "name": "Well Alpha", "depth": 3000}
{"wellId": "W002", "name": "Well Beta", "depth": 3500}
{"wellId": "W003", "name": "Well Gamma", "depth": 2800}

布局 2: frame_parquet_item_dirs (Parquet 格式)

data/
└── frames/
    └── parquet/
        ├── item-00000/    # 第 0 个项目
        │   └── data.parquet
        ├── item-00001/    # 第 1 个项目
        │   └── data.parquet
        └── item-00002/    # 第 2 个项目
            └── data.parquet

目录命名规则: item-{index:05d} (5 位数字，前导零填充)

生成哈希值

SDK 提供了内置的哈希工具：

from fraclab_sdk.materialize.hash import compute_file_sha256

ds_hash = compute_file_sha256("ds.json")
drs_hash = compute_file_sha256("drs.json")
print(f"dsSha256: {ds_hash}")
print(f"drsSha256: {drs_hash}")

也可以直接使用 Python 标准库：

import hashlib
from pathlib import Path

content = Path("ds.json").read_bytes()
ds_hash = hashlib.sha256(content).hexdigest()