latchbio-integration

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 サポート（K80、V100、A100）
• タイムアウトとストレージの設定
• コスト最適化の戦略

4. 検証済みワークフロー

• 本番環境対応の事前構築済みパイプライン
• Bulk RNA-seq、DESeq2、パスウェイ解析
• タンパク質構造予測のための AlphaFold と ColabFold
• シングルセルツール（ArchR、scVelo、emptyDropsR）
• CRISPR 解析、系統解析など

クイックスタート

インストールとセットアップ

# Install Latch SDK
uv pip install latch

# Login to Latch
latch login

# Initialize a new workflow
latch init my-workflow

# Register workflow to platform
latch register my-workflow

前提条件:

• Docker がインストールされ、実行中であること
• Latch アカウントの認証情報
• Python 3.8 以上

基本的なワークフローの例

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 モジュールを使用する場合

詳細なドキュメント

このスキルには、機能ごとに整理された包括的なリファレンスドキュメントが含まれています。

references/workflow-creation.md

次の場合に参照してください:

• ワークフローの作成と登録
• タスクの定義とデコレーター
• Python、Nextflow、Snakemake のサポート
• 起動計画と条件付きセクション
• ワークフローの実行（CLI およびプログラムから）
• マルチステップおよび並列パイプライン
• 登録時の問題のトラブルシューティング

主なトピック:

• latch init と latch register コマンド
• @workflow と @task デコレーター
• LatchFile と LatchDir の基礎
• 型アノテーションとドキュメント文字列
• 事前設定済みパラメーターを持つ起動計画
• 条件付き UI セクション

references/data-management.md

次の場合に参照してください:

• LatchFile と LatchDir によるクラウドストレージ
• Registry システム（Projects、Tables、Records）
• リンクされたレコードと関連付け
• 列挙型と型付き列
• 一括操作とトランザクション
• ワークフローとの統合
• アカウントとワークスペースの管理

主なトピック:

• latch:/// パス形式
• ファイル転送と glob パターン
• Registry テーブルの作成と検索
• 列の型（string、number、file、link、enum）
• レコードの CRUD 操作
• ワークフローと Registry の統合

references/resource-configuration.md

以下の場合にこの文書を参照してください:

• タスクリソースデコレーター
• CPU、メモリ、GPU のカスタム構成
• GPU の種類（K80、V100、A100）
• タイムアウトとストレージの設定
• リソース最適化の戦略
• コスト効率の高いワークフロー設計
• モニタリングとデバッグ

主なトピック:

• @small_task、@large_task、@small_gpu_task、@large_gpu_task
• 正確な仕様を指定した @custom_task
• マルチ GPU 構成
• ワークロードの種類別のリソース選択
• プラットフォームの上限とクォータ

references/verified-workflows.md

以下の場合にこの文書を参照してください:

• 構築済みの本番用ワークフロー
• バルク RNA-seq と DESeq2
• AlphaFold と ColabFold
• シングルセル解析（ArchR、scVelo）
• CRISPR 編集解析
• パスウェイエンリッチメント
• カスタムワークフローとの統合

主なトピック:

• 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)

レジストリ連携ワークフロー

from latch import workflow, small_task
from latch.registry.table import Table
from latch.registry.record import Record
from latch.types import LatchFile

@small_task
def process_and_track(sample_id: str, table_id: str) -> str:
    """Process sample and update Registry"""
    # Get sample from registry
    table = Table.get(table_id=table_id)
    records = Record.list(table_id=table_id, filter={"sample_id": sample_id})
    sample = records[0]

    # Process
    input_file = sample.values["fastq_file"]
    output = process(input_file)

    # Update registry
    sample.update(values={"status": "completed", "result": output})
    return "Success"

@workflow
def registry_workflow(sample_id: str, table_id: str):
    """Workflow integrated with Registry"""
    return process_and_track(sample_id=sample_id, table_id=table_id)

ベストプラクティス

ワークフロー設計

1. すべてのパラメーターに型アノテーションを使用する
2. 明確な docstring を記述する（UI に表示されます）
3. まずは標準のタスクデコレーターから始め、必要に応じて規模を拡大する
4. 複雑なワークフローはモジュール化されたタスクに分割する
5. 適切なエラーハンドリングを実装する

データ管理

6. 一貫性のあるフォルダー構造を使用する
7. 一括入力の前にレジストリのスキーマを定義する
8. リレーションシップにはリンクされたレコードを使用する
9. トレーサビリティのため、メタデータをレジストリに保存する

リソース構成

10. リソースは適切なサイズにする（過剰な割り当てをしない）
11. アルゴリズムが対応している場合にのみ GPU を使用する
12. 実行メトリクスを監視し、最適化する
13. 可能な場合は並列実行を前提に設計する

開発ワークフロー

14. レジストリに登録する前に、Docker でローカルテストを行う
15. ワークフローのコードにはバージョン管理を使用する
16. リソース要件を文書化する
17. プロファイリングを行い、実際に必要なリソースを把握する

トラブルシューティング

よくある問題

登録の失敗:

• Docker が実行中であることを確認する
• latch login で認証を確認する
• Dockerfile にすべての依存関係が含まれていることを確認する
• 詳細なログを確認するには --verbose フラグを使用する

リソースの問題:

• メモリ不足: タスクデコレーターでメモリを増やす
• タイムアウト: タイムアウトパラメーターを増やす
• ストレージの問題: ephemeral storage_gib を増やす

データアクセス:

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

型エラー:

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

追加リソース

• 公式ドキュメント: https://docs.latch.bio
• GitHub リポジトリ: https://github.com/latchbio/latch
• Slack コミュニティ: Latch SDK ワークスペースに参加する
• API リファレンス: https://docs.latch.bio/api/latch.html
• ブログ: https://blog.latch.bio

サポート

問題や質問がある場合は、次の手順を実行してください:

1. 上記のドキュメントのリンクを確認する
2. GitHub の Issue を検索する
3. Slack コミュニティで質問する
4. [email protected] に問い合わせる