alterlab-latchbio

LatchBio の統合

概要

Latch は、バイオインフォマティクスのワークフローをサーバーレスパイプラインとして構築、デプロイするための Python フレームワークです。Flyte をベースとしており、@workflow/@task デコレーターを使ってワークフローを作成し、LatchFile/LatchDir でクラウドデータを管理し、リソースを設定して、Nextflow/Snakemake のパイプラインを統合できます。

主な機能

Latch プラットフォームは、次の 4 つの主な機能を提供します。

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
• さらに多くのパイプライン（例：AlphaFold、シングルセル、CRISPR）が、プラットフォーム UI で検証済みワークフローとして利用できます。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 デコレーター、型アノテーションと docstring（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. 明確な docstring を記述します（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 フラグを使用します

リソースの問題：

• メモリ不足：タスクデコレーターでメモリを増やします
• タイムアウト：タイムアウトパラメーターを増やします
• ストレージの問題：エフェメラルストレージの storage_gib を増やします

データアクセス：

• 正しい latch:/// パス形式を使用します
• ワークスペースにファイルが存在することを確認します
• 共有ワークスペースの権限を確認します

型エラー：

• すべてのパラメーターに型アノテーションを追加します
• ファイル／ディレクトリーのパラメーターには LatchFile／LatchDir を使用します
• ワークフローの戻り値の型が実際の戻り値と一致することを確認します

追加リソース

• 公式ドキュメント：https://docs.latch.bio
• GitHub リポジトリー（現在の API／デコレーター名はこちらで確認してください）：https://github.com/latchbio/latch
• サポート：[email protected]

AlterLab Academic Skills スイートの一部です。