alterlab-latchbio

LatchBio 集成

概述

Latch 是一个 Python 框架，用于将生物信息学工作流构建和部署为无服务器流水线。Latch 基于 Flyte 构建，可使用 @workflow/@task 装饰器创建工作流，使用 LatchFile/LatchDir 管理云端数据，配置资源，并集成 Nextflow/Snakemake 流水线。

核心功能

Latch 平台提供四大类功能：

1. 工作流创建与部署

• 使用 Python 装饰器定义无服务器工作流
• 支持原生 Python、Nextflow 和 Snakemake 流水线
• 使用 Docker 自动容器化
• 自动生成无代码用户界面
• 版本控制和可复现性

2. 数据管理

• 云存储抽象（LatchFile、LatchDir）
• 使用 Registry 进行结构化数据组织（Projects → Tables → Records）
• 使用链接和枚举进行类型安全的数据操作
• 在本地和云端之间自动传输文件
• 使用 Glob 模式匹配选择文件

3. 资源配置

• 预配置的任务装饰器（@small_task、@large_task、@small_gpu_task、@large_gpu_task）
• 自定义资源规格（CPU、内存、GPU、存储）
• GPU 节点：NVIDIA T4（@small_gpu_task、g4dn.2xlarge）、A10G（@large_gpu_task、g5.16xlarge）、V100（@v100_x1/x4/x8_task）、L40S（@g6e_*_task）
• 超时和存储配置
• 成本优化策略

4. 已验证工作流

• 由 Latch 维护的可用于生产环境的预构建流水线
• 可从 latch.verified Python 模块导入：rnaseq、deseq2_wf、mafft、trim_galore、gene_ontology_pathway_analysis
• 平台 UI 中还提供许多其他流水线（例如 AlphaFold、单细胞、CRISPR），作为 Verified Workflows 使用；依赖 Python 导入前，请在 latch.verified 中确认确切的导入名称（参见 references/verified-workflows.md）

快速入门

安装和设置

# Install Latch SDK (uv-first; pip install latch also works)
uv pip install latch

# Login to Latch
latch login

# Initialize a new workflow (scaffolds wf/, Dockerfile, version)
latch init my-workflow

# Register workflow to platform (builds container, generates the UI)
latch register my-workflow

先决条件：

• 已安装并运行 Docker（注册过程会在本地构建容器）
• Latch 账户凭据
• Python 3.9+（Latch SDK requires-python >=3.9）

基本工作流示例

from latch import workflow, small_task
from latch.types import LatchFile

@small_task
def process_file(input_file: LatchFile) -> LatchFile:
    """Process a single file"""
    # Processing logic
    return output_file

@workflow
def my_workflow(input_file: LatchFile) -> LatchFile:
    """
    My bioinformatics workflow

    Args:
        input_file: Input data file
    """
    return process_file(input_file=input_file)

何时使用此技能

遇到以下任一场景时，应使用此技能：

工作流开发：

• “为 RNA-seq 分析创建一个 Latch 工作流”
• “将我的流水线部署到 Latch”
• “将我的 Nextflow 流水线转换为 Latch”
• “为我的工作流添加 GPU 支持”
• 使用 @workflow、@task 装饰器

数据管理：

• “在 Latch Registry 中整理我的测序数据”
• “如何使用 LatchFile 和 LatchDir？”
• “在 Latch 中设置样本追踪”
• 使用 latch:/// 路径

资源配置：

• “在 Latch 上为 AlphaFold 配置 GPU”
• “我的任务内存不足”
• “如何优化工作流成本？”
• 使用任务装饰器

已验证工作流：

• “在 Latch 上运行 AlphaFold”
• “使用 DESeq2 进行差异表达分析”
• “可用的预构建工作流”
• 使用 latch.verified 模块

不适用场景

场景	改用
DNAnexus 应用/小程序、dxapp.json、dxpy 或 dx CLI	alterlab-dnanexus
直接运行 nf-core/sarek 变异检测（不使用 Latch）	alterlab-nf-core-sarek
不涉及 Latch 工作流的通用无服务器 Python/GPU 作业	alterlab-modal
SLURM/HPC 或云端批处理作业的提交和轮询	alterlab-remote-compute
LaminDB 中的数据集注册表、本体验证和溯源	alterlab-lamindb

详细文档

根据任务加载对应的参考文档：

• references/workflow-creation.md — latch init/latch register、@workflow/@task 装饰器、类型注解和文档字符串（这些内容会显示在 UI 中）、启动计划、条件部分、CLI/程序化执行、并行 map_task、注册故障排查。
• references/data-management.md — LatchFile/LatchDir 和 latch:/// 路径、glob 模式、Registry API（Project、Table、Record）、列类型、table.update() 事务，以及工作流与 Registry 的集成。
• references/resource-configuration.md — 标准任务装饰器、@custom_task(cpu, memory, storage_gib, timeout)、GPU 任务装饰器（small_gpu_task/large_gpu_task、v100_x{1,4,8}_task、g6e_*_task），以及资源配置和成本优化。
• references/verified-workflows.md — 可从 latch.verified 导入的流水线，以及如何将它们与自定义任务结合使用。

常见工作流模式

完整的 RNA-seq 流水线

from latch import workflow, small_task, large_task
from latch.types import LatchFile, LatchDir

@small_task
def quality_control(fastq: LatchFile) -> LatchFile:
    """Run FastQC"""
    return qc_output

@large_task
def alignment(fastq: LatchFile, genome: str) -> LatchFile:
    """STAR alignment"""
    return bam_output

@small_task
def quantification(bam: LatchFile) -> LatchFile:
    """featureCounts"""
    return counts

@workflow
def rnaseq_pipeline(
    input_fastq: LatchFile,
    genome: str,
    output_dir: LatchDir
) -> LatchFile:
    """RNA-seq analysis pipeline"""
    qc = quality_control(fastq=input_fastq)
    aligned = alignment(fastq=qc, genome=genome)
    return quantification(bam=aligned)

GPU 加速工作流

from latch import workflow, small_task, large_gpu_task
from latch.types import LatchFile

@small_task
def preprocess(input_file: LatchFile) -> LatchFile:
    """Prepare data"""
    return processed

@large_gpu_task
def gpu_computation(data: LatchFile) -> LatchFile:
    """GPU-accelerated analysis"""
    return results

@workflow
def gpu_pipeline(input_file: LatchFile) -> LatchFile:
    """Pipeline with GPU tasks"""
    preprocessed = preprocess(input_file=input_file)
    return gpu_computation(data=preprocessed)

集成 Registry 的工作流

from latch import workflow, small_task
from latch.registry.table import Table

@small_task
def process_and_track(sample_name: str, table_id: str) -> str:
    """Process a sample tracked in a Registry table and write status/result back."""
    table = Table(table_id)  # construct by id; no Table.get classmethod

    # Find the record by name. list_records() yields pages of {record_id: Record}.
    sample = None
    for page in table.list_records():
        for record in page.values():
            if record.get_name() == sample_name:
                sample = record
                break
        if sample is not None:
            break
    if sample is None:
        raise ValueError(f"No record named {sample_name}")

    # Read column values
    values = sample.get_values()
    input_file = values["fastq_file"]  # a LatchFile for a 'file'-typed column
    # ... processing logic produces an output LatchFile ...

    # Write status/result back via the table.update() transaction (upsert by name)
    with table.update() as updater:
        updater.upsert_record(sample_name, status="completed", result=input_file)
    return "Success"

@workflow
def registry_workflow(sample_name: str, table_id: str) -> str:
    """Workflow integrated with the Latch Registry."""
    return process_and_track(sample_name=sample_name, table_id=table_id)

最佳实践

工作流设计

1. 为所有参数使用类型注解
2. 编写清晰的文档字符串（会显示在 UI 中）
3. 从标准任务装饰器开始，需要时再扩展
4. 将复杂工作流拆分为模块化任务
5. 实现适当的错误处理

数据管理

6. 使用一致的文件夹结构
7. 批量录入数据前定义 Registry 架构
8. 使用关联记录表示关系
9. 在 Registry 中存储元数据，以便追踪

资源配置

10. 合理配置资源（不要过度分配）
11. 仅在算法支持时使用 GPU
12. 监控执行指标并进行优化
13. 尽可能针对并行执行进行设计

开发工作流

14. 注册前使用 Docker 在本地测试
15. 对工作流代码使用版本控制
16. 记录资源需求
17. 对工作流进行性能分析，以确定实际需求

故障排查

常见问题

注册失败：

• 确保 Docker 正在运行
• 使用 latch login 检查身份验证
• 确认 Dockerfile 中包含所有依赖项
• 使用 --verbose 标志查看详细日志

资源问题：

• 内存不足：增加任务装饰器中的内存配置
• 超时：增加 timeout 参数
• 存储问题：增加临时存储 storage_gib

数据访问：

• 使用正确的 latch:/// 路径格式
• 确认文件存在于工作区中
• 检查共享工作区的权限

类型错误：

• 为所有参数添加类型注解
• 文件/目录参数使用 LatchFile/LatchDir
• 确保工作流返回类型与实际返回值一致

其他资源

• 官方文档：https://docs.latch.bio
• GitHub 仓库（在此确认当前 API/装饰器名称）：https://github.com/latchbio/latch
• 支持：[email protected]

AlterLab Academic Skills 套件的一部分。