latchbio-integration by davila7/claude-code-templates
npx skills add https://github.com/davila7/claude-code-templates --skill latchbio-integrationLatch 是一个用于构建和部署生物信息学工作流作为无服务器管道的 Python 框架。它基于 Flyte 构建,您可以使用 @workflow/@task 装饰器创建工作流,使用 LatchFile/LatchDir 管理云数据,配置资源,并集成 Nextflow/Snakemake 管道。
Latch 平台提供四个主要功能领域:
# 安装 Latch SDK
python3 -m uv pip install latch
# 登录 Latch
latch login
# 初始化新工作流
latch init my-workflow
# 将工作流注册到平台
latch register my-workflow
广告位招租
在这里展示您的产品或服务
触达数万 AI 开发者,精准高效
前提条件:
from latch import workflow, small_task
from latch.types import LatchFile
@small_task
def process_file(input_file: LatchFile) -> LatchFile:
"""处理单个文件"""
# 处理逻辑
return output_file
@workflow
def my_workflow(input_file: LatchFile) -> LatchFile:
"""
我的生物信息学工作流
参数:
input_file: 输入数据文件
"""
return process_file(input_file=input_file)
当遇到以下任何场景时,应使用此技能:
工作流开发:
@workflow、@task 装饰器数据管理:
latch:/// 路径资源配置:
已验证工作流:
latch.verified 模块此技能包含按功能组织的全面参考文档:
阅读此文档以了解:
关键主题:
latch init 和 latch register 命令@workflow 和 @task 装饰器阅读此文档以了解:
关键主题:
latch:/// 路径格式阅读此文档以了解:
关键主题:
@small_task、@large_task、@small_gpu_task、@large_gpu_task@custom_task阅读此文档以了解:
关键主题:
latch.verified 模块导入from latch import workflow, small_task, large_task
from latch.types import LatchFile, LatchDir
@small_task
def quality_control(fastq: LatchFile) -> LatchFile:
"""运行 FastQC"""
return qc_output
@large_task
def alignment(fastq: LatchFile, genome: str) -> LatchFile:
"""STAR 比对"""
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 分析管道"""
qc = quality_control(fastq=input_fastq)
aligned = alignment(fastq=qc, genome=genome)
return quantification(bam=aligned)
from latch import workflow, small_task, large_gpu_task
from latch.types import LatchFile
@small_task
def preprocess(input_file: LatchFile) -> LatchFile:
"""准备数据"""
return processed
@large_gpu_task
def gpu_computation(data: LatchFile) -> LatchFile:
"""GPU 加速分析"""
return results
@workflow
def gpu_pipeline(input_file: LatchFile) -> LatchFile:
"""包含 GPU 任务的管道"""
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:
"""处理样本并更新 Registry"""
# 从 registry 获取样本
table = Table.get(table_id=table_id)
records = Record.list(table_id=table_id, filter={"sample_id": sample_id})
sample = records[0]
# 处理
input_file = sample.values["fastq_file"]
output = process(input_file)
# 更新 registry
sample.update(values={"status": "completed", "result": output})
return "Success"
@workflow
def registry_workflow(sample_id: str, table_id: str):
"""与 Registry 集成的工作流"""
return process_and_track(sample_id=sample_id, table_id=table_id)
注册失败:
latch login 检查认证--verbose 标志获取详细日志资源问题:
数据访问:
latch:/// 路径格式类型错误:
如有问题或疑问:
每周安装数
137
仓库
GitHub 星标数
23.5K
首次出现
2026年1月21日
安全审计
安装于
claude-code121
opencode112
cursor106
gemini-cli105
antigravity101
codex96
Latch is a Python framework for building and deploying bioinformatics workflows as serverless pipelines. Built on Flyte, create workflows with @workflow/@task decorators, manage cloud data with LatchFile/LatchDir, configure resources, and integrate Nextflow/Snakemake pipelines.
The Latch platform provides four main areas of functionality:
# Install Latch SDK
python3 -m 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
Prerequisites:
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)
This skill should be used when encountering any of the following scenarios:
Workflow Development:
@workflow, @task decoratorsData Management:
latch:/// pathsResource Configuration:
Verified Workflows:
latch.verified moduleThis skill includes comprehensive reference documentation organized by capability:
Read this for:
Key topics:
latch init and latch register commands@workflow and @task decoratorsRead this for:
Key topics:
latch:/// path formatRead this for:
Key topics:
@small_task, @large_task, @small_gpu_task, @large_gpu_task@custom_task with precise specificationsRead this for:
Key topics:
latch.verified module importsfrom 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)
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)
Registration Failures:
latch login--verbose flag for detailed logsResource Problems:
Data Access:
latch:/// path formatType Errors:
For issues or questions:
Weekly Installs
137
Repository
GitHub Stars
23.5K
First Seen
Jan 21, 2026
Security Audits
Gen Agent Trust HubPassSocketPassSnykPass
Installed on
claude-code121
opencode112
cursor106
gemini-cli105
antigravity101
codex96
Azure 升级评估与自动化工具 - 轻松迁移 Functions 计划、托管层级和 SKU
90,800 周安装