# Genomic variant effect prediction > [!NOTE] > Code available [here](https://github.com/unionai/unionai-examples/tree/main/v2/tutorials/genomic_variant_effect). This tutorial demonstrates zero-shot variant effect prediction (VEP) with HuggingFace [Carbon](https://huggingface.co/HuggingFaceBio/Carbon-3B). The pipeline loads clinically relevant variants across genes such as BRCA2, TP53, CFTR, KRAS, and HBB, scores each mutation with a log-likelihood ratio, and produces rich HTML reports with DNA tracks, lollipop plots, confusion matrices, and ranked pathogenicity tables. Flyte provides: - **GPU-backed inference** for Carbon scoring with live progress reports. - **CPU analysis tasks** for visualization and accuracy metrics without holding a GPU. - **End-to-end orchestration** from variant loading through summary reporting. ## Define the task environments ``` # /// script # requires-python = ">=3.12" # dependencies = [ # "flyte>=2.4.0", # "torch>=2.9.0", # "transformers>=4.49.0", # "accelerate>=0.34.0", # "numpy", # ] # main = "pipeline" # params = "" # /// import json import logging import math import os import tempfile import flyte import flyte.io import flyte.report # {{docs-fragment env}} main_img = flyte.Image.from_uv_script(__file__, name="genomic-variant-effect", pre=True) gpu_env = flyte.TaskEnvironment( name="genomic-variant-effect-gpu", image=main_img, resources=flyte.Resources(cpu=4, memory="24Gi", gpu=1), ) cpu_env = flyte.TaskEnvironment( name="genomic-variant-effect-cpu", image=main_img, resources=flyte.Resources(cpu=2, memory="6Gi"), depends_on=[gpu_env], ) # {{/docs-fragment env}} logging.basicConfig(level=logging.WARNING, format="%(message)s", force=True) log = logging.getLogger(__name__) log.setLevel(logging.INFO) # ------------------------------------------------------------------ # Default gene variants — clinically relevant mutations # ------------------------------------------------------------------ # Each entry: gene name -> { "sequence": reference DNA, "variants": [{ "pos": 0-indexed, "ref": base, "alt": base, "name": "...", "known_effect": "..." }] } # Sequences are short windows (~120-200bp) around the variant site for tractable inference. DEFAULT_GENE_VARIANTS = { "BRCA2 (Breast Cancer)": { "description": "Tumor suppressor critical for DNA repair via homologous recombination. Mutations dramatically increase breast and ovarian cancer risk.", "sequence": "ATGGCCTCGAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAG", "variants": [ {"pos": 12, "ref": "A", "alt": "T", "name": "c.37A>T", "known_effect": "pathogenic", "clinical": "Nonsense mutation — truncates protein early"}, {"pos": 18, "ref": "G", "alt": "A", "name": "c.55G>A", "known_effect": "benign", "clinical": "Synonymous — no amino acid change"}, {"pos": 30, "ref": "C", "alt": "T", "name": "c.91C>T", "known_effect": "pathogenic", "clinical": "Missense in DNA-binding domain"}, {"pos": 45, "ref": "G", "alt": "C", "name": "c.136G>C", "known_effect": "uncertain", "clinical": "Variant of uncertain significance (VUS)"}, ], }, "TP53 (Tumor Suppressor)": { "description": "Guardian of the genome. Activates DNA repair, cell cycle arrest, and apoptosis. Mutated in >50% of human cancers.", "sequence": "ATGGAGGAGCCGCAGTCAGATCCTAGCGTGAGTTTGCACCCTTCAGAGACAGAAACCACTGGATTGGAGACTACTTCCTGAAACAACGTTCTGTCCCCCTTGCCGTCCCAAGCAATGGATGAT", "variants": [ {"pos": 15, "ref": "C", "alt": "T", "name": "R175H", "known_effect": "pathogenic", "clinical": "Hotspot — gain-of-function, dominant negative. Most common TP53 mutation in cancer"}, {"pos": 36, "ref": "T", "alt": "C", "name": "P72R", "known_effect": "benign", "clinical": "Common polymorphism — subtle effect on apoptosis efficiency"}, {"pos": 54, "ref": "C", "alt": "A", "name": "G245S", "known_effect": "pathogenic", "clinical": "Contact mutant — disrupts DNA binding"}, {"pos": 72, "ref": "T", "alt": "G", "name": "R248W", "known_effect": "pathogenic", "clinical": "Structural mutant — destabilizes DNA-binding loop"}, {"pos": 90, "ref": "C", "alt": "T", "name": "R273H", "known_effect": "pathogenic", "clinical": "Contact mutant — directly contacts DNA bases"}, ], }, "CFTR (Cystic Fibrosis)": { "description": "Chloride channel protein. Mutations cause cystic fibrosis — the most common lethal genetic disease in people of European descent.", "sequence": "ATGCAGAGGTCGCCTCTGGAAAAGGCCAGCGTTGTCTCCAAACTTTTTTTCAGCTGGACCAGACCAATTTTGAGGAAAGGATACAGACAGCGCCTGGAATTGTCAGACATATACCAAATCCCTTC", "variants": [ {"pos": 9, "ref": "G", "alt": "A", "name": "G85E", "known_effect": "pathogenic", "clinical": "Disrupts chloride channel processing"}, {"pos": 24, "ref": "C", "alt": "T", "name": "R117H", "known_effect": "pathogenic", "clinical": "Reduces channel conductance — milder CF phenotype"}, {"pos": 48, "ref": "T", "alt": "C", "name": "I148T", "known_effect": "benign", "clinical": "Previously misclassified — now known benign polymorphism"}, {"pos": 66, "ref": "A", "alt": "G", "name": "R334W", "known_effect": "pathogenic", "clinical": "Gating mutation — channel opens less frequently"}, ], }, "KRAS (Oncogene)": { "description": "GTPase signal switch. KRAS mutations are the most common oncogenic driver — found in ~25% of all human cancers, especially pancreatic, colorectal, and lung.", "sequence": "ATGACTGAATATAAACTTGTGGTAGTTGGAGCTGGTGGCGTAGGCAAGAGTGCCTTGACGATACAGCTAATTCAGAATCATTTTGTGGACGAATATGATCCAACAATAGAGGATTCCTACAGGAA", "variants": [ {"pos": 34, "ref": "G", "alt": "T", "name": "G12V", "known_effect": "pathogenic", "clinical": "Locks KRAS in active state — constitutive proliferation signal"}, {"pos": 35, "ref": "G", "alt": "A", "name": "G12D", "known_effect": "pathogenic", "clinical": "Most common KRAS mutation in pancreatic cancer"}, {"pos": 37, "ref": "G", "alt": "T", "name": "G13D", "known_effect": "pathogenic", "clinical": "Constitutively active — common in colorectal cancer"}, {"pos": 60, "ref": "C", "alt": "A", "name": "Q61K", "known_effect": "pathogenic", "clinical": "Impairs GTP hydrolysis — locked ON state"}, ], }, "HBB (Sickle Cell)": { "description": "Beta-globin subunit of hemoglobin. The sickle cell mutation (E6V) is the most well-known single-base disease variant in humans.", "sequence": "ATGGTGCATCTGACTCCTGAGGAGAAGTCTGCCGTTACTGCCCTGTGGGGCAAGGTGAACGTGGATGAAGTTGGTGGTGAGGCCCTGGGCAGGCTGCTGGTGGTCTACCCTTGGACCCAGAGG", "variants": [ {"pos": 17, "ref": "A", "alt": "T", "name": "E6V (HbS)", "known_effect": "pathogenic", "clinical": "THE sickle cell mutation — causes hemoglobin polymerization under low O2"}, {"pos": 19, "ref": "G", "alt": "A", "name": "E6K (HbC)", "known_effect": "pathogenic", "clinical": "Hemoglobin C disease — milder than sickle cell but causes crystal formation"}, {"pos": 36, "ref": "G", "alt": "A", "name": "E26K", "known_effect": "benign", "clinical": "Hemoglobin E — most common Hb variant worldwide, mild effect"}, {"pos": 78, "ref": "C", "alt": "T", "name": "Q39X", "known_effect": "pathogenic", "clinical": "Nonsense — causes beta-thalassemia (no functional beta-globin)"}, ], }, } # DNA base colors (classic genomics color scheme) BASE_COLORS = {"A": "#2ecc71", "T": "#e74c3c", "G": "#f39c12", "C": "#3498db"} BASE_COMPLEMENT = {"A": "T", "T": "A", "G": "C", "C": "G"} # Pathogenicity color scheme EFFECT_COLORS = { "pathogenic": "#dc2626", "benign": "#059669", "uncertain": "#f59e0b", } EFFECT_BADGES = { "pathogenic": "badge-danger", "benign": "badge-success", "uncertain": "badge-warning", } # ------------------------------------------------------------------ # Report styling — genomics-themed deep blues and teals # ------------------------------------------------------------------ REPORT_CSS = """ """ def _wrap_report(html: str) -> str: return f'{REPORT_CSS}
{html}
' # ------------------------------------------------------------------ # SVG chart helpers # ------------------------------------------------------------------ def _make_bar_chart( labels: list[str], series: dict[str, list[float]], title: str = "", colors: list[str] | None = None, width: int = 700, height: int = 300, value_format: str = ".2f", ) -> str: """Generate an SVG grouped bar chart.""" if not labels: return "" default_colors = ["#2563eb", "#1e3a5f", "#3b82f6", "#60a5fa", "#93c5fd"] colors = colors or default_colors ml, mr, mt, mb = 70, 20, 40, 80 cw = width - ml - mr ch = height - mt - mb all_vals = [v for vals in series.values() for v in vals] y_max = max(abs(v) for v in all_vals) if all_vals else 1 y_min = min(all_vals) if all_vals else 0 # For VEP scores (negative = more damaging), we need to handle negative values if y_min >= 0: y_min_plot = 0 y_max_plot = y_max * 1.15 or 1 else: y_max_plot = max(y_max * 1.15, 0.1) y_min_plot = y_min * 1.15 y_range = y_max_plot - y_min_plot or 1 n_groups = len(labels) n_series = len(series) group_width = cw / n_groups bar_width = group_width * 0.7 / max(n_series, 1) gap = group_width * 0.15 def sy(v): return mt + ch - (v - y_min_plot) / y_range * ch svg = [ f'', f'', ] # Grid lines for i in range(6): y_tick = y_min_plot + y_range * i / 5 py = sy(y_tick) svg.append( f'' ) svg.append( f'{y_tick:{value_format}}' ) # Zero line if y_min_plot < 0 < y_max_plot: zy = sy(0) svg.append( f'' ) # Bars for gi, label in enumerate(labels): gx = ml + gi * group_width + gap for si, (name, vals) in enumerate(series.items()): color = colors[si % len(colors)] bx = gx + si * bar_width val = vals[gi] if val >= 0: by = sy(val) bh = sy(0) - by if y_min_plot < 0 else mt + ch - by else: by = sy(0) if y_min_plot < 0 else mt + ch bh = sy(val) - by svg.append( f'' ) text_y = by - 4 if val >= 0 else by + bh + 12 svg.append( f'' f'{val:{value_format}}' ) # Rotated group label lx = gx + n_series * bar_width / 2 svg.append( f'{label}' ) # Title if title: svg.append( f'{title}' ) # Legend if n_series > 1: lx = ml + cw - len(series) * 110 for si, name in enumerate(series): color = colors[si % len(colors)] svg.append( f'' ) svg.append( f'{name}' ) svg.append("") return "\n".join(svg) def _make_heatmap( matrix: list[list[float]], row_labels: list[str], col_labels: list[str], title: str = "", width: int = 700, height: int = 500, value_format: str = ".2f", diverging: bool = False, ) -> str: """Generate an SVG heatmap. If diverging=True, uses red-white-blue scale centered at 0.""" n_rows = len(matrix) n_cols = len(matrix[0]) if matrix else 0 if not n_rows or not n_cols: return "" show_values = n_rows <= 10 and n_cols <= 12 flat = [v for row in matrix for v in row] v_min = min(flat) v_max = max(flat) if diverging: abs_max = max(abs(v_min), abs(v_max)) or 1 def get_color(v): t = v / abs_max # -1 to 1 if t < 0: # White to red (negative = damaging) r = 255 g = int(255 * (1 + t)) b = int(255 * (1 + t)) else: # White to blue (positive = benign) r = int(255 * (1 - t)) g = int(255 * (1 - t)) b = 255 return f"rgb({r},{g},{b})" else: v_range = v_max - v_min or 1 def get_color(v): t = (v - v_min) / v_range r = int(255 - t * (255 - 30)) g = int(255 - t * (255 - 58)) b = int(255 - t * (255 - 95)) return f"rgb({r},{g},{b})" # Layout ml = max(140, max(len(l) for l in row_labels) * 7 + 20) if row_labels else 140 mr = 20 mt = 80 if col_labels else 40 mb = 30 cw = width - ml - mr ch = height - mt - mb cell_w = cw / n_cols cell_h = ch / n_rows svg = [ f'', f'', ] if title: svg.append( f'{title}' ) # Column labels (rotated) for j, label in enumerate(col_labels): cx = ml + j * cell_w + cell_w / 2 svg.append( f'{label}' ) # Row labels + cells for i, row_label in enumerate(row_labels): ry = mt + i * cell_h + cell_h / 2 svg.append( f'{row_label}' ) for j in range(n_cols): val = matrix[i][j] color = get_color(val) cx = ml + j * cell_w cy = mt + i * cell_h svg.append( f'' ) if show_values: if diverging: t = abs(val) / (max(abs(v_min), abs(v_max)) or 1) else: t = (val - v_min) / (v_max - v_min or 1) text_color = "#fff" if t > 0.55 else "#1a1a2e" font_size = min(10, int(cell_w / 4), int(cell_h / 2.5)) font_size = max(7, font_size) svg.append( f'{val:{value_format}}' ) svg.append("") return "\n".join(svg) def _make_dna_track( sequence: str, variants: list[dict], gene_name: str = "", width: int = 900, ) -> str: """Render a color-coded DNA sequence track with variant positions highlighted.""" chars_per_line = 60 char_w = 11 line_h = 22 label_w = 50 n_lines = (len(sequence) + chars_per_line - 1) // chars_per_line # Extra space for variant annotations variant_positions = {v["pos"] for v in variants} svg_h = n_lines * line_h + 60 svg = [ f'', f'', ] if gene_name: svg.append( f'{gene_name}' ) y_offset = 28 for line_idx in range(n_lines): start = line_idx * chars_per_line end = min(start + chars_per_line, len(sequence)) chunk = sequence[start:end] y = y_offset + line_idx * line_h # Position label svg.append( f'{start + 1}' ) for ci, base in enumerate(chunk): abs_pos = start + ci x = label_w + ci * char_w color = BASE_COLORS.get(base, "#6b7280") is_variant = abs_pos in variant_positions if is_variant: # Red highlight for variant positions svg.append( f'' ) # Small triangle marker above svg.append( f'' ) else: # Subtle background svg.append( f'' ) svg.append( f'{base}' ) # Spacer every 10 bases if (abs_pos + 1) % 10 == 0 and ci < len(chunk) - 1: svg.append( f'' ) # Legend ly = svg_h - 12 lx = label_w for base, color in BASE_COLORS.items(): svg.append(f'') svg.append(f'{base}') lx += 30 lx += 10 svg.append(f'') svg.append(f'Variant site') svg.append("") return "\n".join(svg) def _make_lollipop_plot( variants: list[dict], seq_length: int, title: str = "", width: int = 700, height: int = 200, ) -> str: """Generate a lollipop plot showing variant positions along a gene with effect scores.""" if not variants: return "" ml, mr, mt, mb = 50, 30, 40, 40 cw = width - ml - mr ch = height - mt - mb svg = [ f'', f'', ] if title: svg.append( f'{title}' ) # Gene track (horizontal bar) track_y = mt + ch * 0.7 svg.append( f'' ) # Position labels at ends svg.append( f'1' ) svg.append( f'{seq_length}' ) # Lollipops scores = [v.get("score", 0) for v in variants] max_score = max(abs(s) for s in scores) if scores else 1 for v in variants: pos = v["pos"] score = v.get("score", 0) effect = v.get("known_effect", "uncertain") color = EFFECT_COLORS.get(effect, "#6b7280") x = ml + (pos / max(seq_length - 1, 1)) * cw stem_h = max(20, abs(score) / max_score * (ch * 0.6)) circle_y = track_y - stem_h - 6 # Stem svg.append( f'' ) # Circle svg.append( f'' ) # Label svg.append( f'{v.get("name", "")}' ) # Legend lx = ml for effect, color in EFFECT_COLORS.items(): svg.append(f'') svg.append( f'' f'{effect.title()}' ) lx += len(effect) * 7 + 24 svg.append("") return "\n".join(svg) def _make_score_comparison_chart( gene_results: dict, width: int = 800, height: int = 350, ) -> str: """Generate a dot plot comparing VEP scores across all genes, colored by known effect.""" all_variants = [] for gene_name, gene_data in gene_results.items(): for v in gene_data["variants"]: all_variants.append({ "gene": gene_name.split("(")[0].strip(), "name": v["name"], "score": v.get("score", 0), "known_effect": v["known_effect"], }) if not all_variants: return "" ml, mr, mt, mb = 120, 30, 40, 30 cw = width - ml - mr ch = height - mt - mb scores = [v["score"] for v in all_variants] s_min, s_max = min(scores), max(scores) s_pad = (s_max - s_min) * 0.1 or 0.5 s_min -= s_pad s_max += s_pad s_range = s_max - s_min or 1 def sx(s): return ml + (s - s_min) / s_range * cw svg = [ f'', f'', f'' f'Variant Effect Scores — All Genes', ] # Grid for i in range(6): gx = s_min + s_range * i / 5 px = sx(gx) svg.append( f'' ) svg.append( f'{gx:.2f}' ) # Zero line if s_min < 0 < s_max: zx = sx(0) svg.append( f'' ) # Rows per variant row_h = ch / max(len(all_variants), 1) for i, v in enumerate(all_variants): y = mt + i * row_h + row_h / 2 color = EFFECT_COLORS.get(v["known_effect"], "#6b7280") px = sx(v["score"]) # Label label = f'{v["gene"]} {v["name"]}' svg.append( f'{label}' ) # Connector line svg.append( f'' ) # Dot svg.append( f'' ) svg.append("") return "\n".join(svg) # ------------------------------------------------------------------ # Task 1: Load and validate gene variants # ------------------------------------------------------------------ @cpu_env.task(cache="auto") async def load_variants( variants_json: str = "", ) -> flyte.io.Dir: """Load gene variant definitions, validate sequences, and save to a temp directory.""" if variants_json: genes = json.loads(variants_json) else: genes = DEFAULT_GENE_VARIANTS # Validate valid_bases = set("ATGC") for gene_name, gene_data in genes.items(): seq = gene_data["sequence"].upper() invalid = set(seq) - valid_bases if invalid: log.warning(f"{gene_name}: invalid bases {invalid} — removing them") seq = "".join(b for b in seq if b in valid_bases) gene_data["sequence"] = seq for v in gene_data["variants"]: pos = v["pos"] if pos < 0 or pos >= len(seq): log.warning(f"{gene_name} variant {v['name']}: position {pos} out of range [0, {len(seq)})") elif seq[pos] != v["ref"]: log.warning(f"{gene_name} variant {v['name']}: expected ref={v['ref']} at pos {pos}, found {seq[pos]}") total_variants = sum(len(g["variants"]) for g in genes.values()) log.info(f"Loaded {len(genes)} genes with {total_variants} variants") out_dir = tempfile.mkdtemp(prefix="genomic_vep_") with open(os.path.join(out_dir, "genes.json"), "w") as f: json.dump(genes, f) return await flyte.io.Dir.from_local(out_dir) # ------------------------------------------------------------------ # Task 2: Run Carbon model for variant effect scoring # ------------------------------------------------------------------ @gpu_env.task(report=True) async def score_variants( variants_dir: flyte.io.Dir, model_name: str = "HuggingFaceBio/Carbon-3B", ) -> str: """Score each variant using Carbon's log-likelihood ratio. For each variant, we compute: score = log P(alt_sequence) - log P(ref_sequence) A negative score means the model considers the variant less likely than the reference — suggestive of a damaging/pathogenic effect. A score near zero means the model sees little difference (likely benign). """ import torch from transformers import AutoModelForCausalLM, AutoTokenizer log.info(f"Loading Carbon model: {model_name}") # Load model device = "cuda" if torch.cuda.is_available() else "cpu" if device == "cpu": log.warning("Running on CPU — inference will be slow. GPU recommended for production.") tokenizer = AutoTokenizer.from_pretrained(model_name, trust_remote_code=True) model = AutoModelForCausalLM.from_pretrained( model_name, trust_remote_code=True, dtype=torch.bfloat16 if device == "cuda" else torch.float32, ).to(device) model.eval() # Load variants variants_path = await variants_dir.download() with open(os.path.join(variants_path, "genes.json")) as f: genes = json.load(f) results = {} total_variants = sum(len(g["variants"]) for g in genes.values()) scored = 0 progress_html = """

Carbon Variant Effect Scoring

Model: {model}
Device: {device}
Progress: {scored}/{total} variants scored
""" for gene_name, gene_data in genes.items(): ref_seq = gene_data["sequence"] gene_results = { "description": gene_data.get("description", ""), "sequence": ref_seq, "variants": [], } # Score reference sequence ref_prompt = f"{ref_seq}" ref_inputs = tokenizer(ref_prompt, return_tensors="pt", add_special_tokens=False).to(device) with torch.no_grad(): ref_output = model(**ref_inputs, labels=ref_inputs["input_ids"]) ref_loss = ref_output.loss.item() ref_ll = -ref_loss * ref_inputs["input_ids"].shape[1] for variant in gene_data["variants"]: scored += 1 await flyte.report.replace.aio( _wrap_report(progress_html.format( model=model_name, device=device, scored=scored, total=total_variants )), do_flush=True, ) # Create mutant sequence pos = variant["pos"] alt_seq = ref_seq[:pos] + variant["alt"] + ref_seq[pos + 1:] # Score mutant alt_prompt = f"{alt_seq}" alt_inputs = tokenizer(alt_prompt, return_tensors="pt", add_special_tokens=False).to(device) with torch.no_grad(): alt_output = model(**alt_inputs, labels=alt_inputs["input_ids"]) alt_loss = alt_output.loss.item() alt_ll = -alt_loss * alt_inputs["input_ids"].shape[1] # VEP score: positive = model prefers alt (likely benign), negative = model prefers ref (likely pathogenic) vep_score = alt_ll - ref_ll gene_results["variants"].append({ **variant, "score": round(vep_score, 4), "ref_ll": round(ref_ll, 4), "alt_ll": round(alt_ll, 4), }) log.info( f" {gene_name} | {variant['name']}: score={vep_score:.4f} " f"(known: {variant['known_effect']})" ) results[gene_name] = gene_results # Generate scoring report html_parts = [ "

Carbon Variant Effect Scoring

", '
', f'
{len(genes)}
Genes
', f'
{total_variants}
Variants Scored
', f'
{model_name.split("/")[-1]}
Model
', f'
{device.upper()}
Device
', "
", ] # Per-gene tables for gene_name, gene_data in results.items(): html_parts.append(f'

{gene_name}

') html_parts.append(f'
{gene_data["description"]}
') html_parts.append("") for v in gene_data["variants"]: badge = EFFECT_BADGES.get(v["known_effect"], "badge-info") direction = "damaging" if v["score"] < -0.1 else "neutral" if abs(v["score"]) <= 0.1 else "tolerated" html_parts.append( f'' f'' f'' f'' f'' f'' f'' f'' ) html_parts.append("
VariantRefAltVEP ScoreKnown EffectClinical
{v["name"]}{v["ref"]}{v["alt"]}{v["score"]:.4f} ({direction}){v["known_effect"]}{v.get("clinical", "")}
") await flyte.report.replace.aio(_wrap_report("\n".join(html_parts)), do_flush=True) return json.dumps(results) # ------------------------------------------------------------------ # Task 3: Analyze and visualize variant effects # ------------------------------------------------------------------ @cpu_env.task(report=True) async def analyze_effects( scores_json: str, variants_dir: flyte.io.Dir, ) -> str: """Analyze VEP scores: classification accuracy, gene-level summaries, and rich visualizations.""" results = json.loads(scores_json) variants_path = await variants_dir.download() with open(os.path.join(variants_path, "genes.json")) as f: genes = json.load(f) html_parts = ["

Variant Effect Analysis

"] # ------------------------------------------------------------------ # Overall accuracy: does the model's score direction match known labels? # ------------------------------------------------------------------ all_variants = [] correct = 0 total_known = 0 true_pos = 0 false_pos = 0 true_neg = 0 false_neg = 0 for gene_name, gene_data in results.items(): for v in gene_data["variants"]: all_variants.append({**v, "gene": gene_name}) if v["known_effect"] in ("pathogenic", "benign"): total_known += 1 predicted_pathogenic = v["score"] < -0.05 actual_pathogenic = v["known_effect"] == "pathogenic" if predicted_pathogenic == actual_pathogenic: correct += 1 if predicted_pathogenic and actual_pathogenic: true_pos += 1 elif predicted_pathogenic and not actual_pathogenic: false_pos += 1 elif not predicted_pathogenic and actual_pathogenic: false_neg += 1 else: true_neg += 1 accuracy = correct / total_known if total_known else 0 precision = true_pos / (true_pos + false_pos) if (true_pos + false_pos) else 0 recall = true_pos / (true_pos + false_neg) if (true_pos + false_neg) else 0 html_parts.append('
') html_parts.append(f'
{accuracy:.0%}
Direction Accuracy
') html_parts.append(f'
{precision:.0%}
Precision (Pathogenic)
') html_parts.append(f'
{recall:.0%}
Recall (Pathogenic)
') html_parts.append(f'
{len(all_variants)}
Total Variants
') html_parts.append("
") html_parts.append( '
' "How to read VEP scores: Negative scores mean Carbon considers the variant " "less likely than the reference sequence — suggestive of a damaging effect. Scores near " "zero indicate the model sees little difference (likely benign). The magnitude indicates " "confidence." "
" ) # ------------------------------------------------------------------ # Cross-gene score comparison dot plot # ------------------------------------------------------------------ html_parts.append('
') html_parts.append(_make_score_comparison_chart(results)) html_parts.append("
") # ------------------------------------------------------------------ # Per-gene visualizations # ------------------------------------------------------------------ for gene_name, gene_data in results.items(): short_name = gene_name.split("(")[0].strip() html_parts.append(f'

{gene_name}

') html_parts.append(f'
{gene_data["description"]}
') # DNA track with variant positions highlighted html_parts.append('
') html_parts.append(_make_dna_track( gene_data["sequence"], gene_data["variants"], gene_name=f"{short_name} Reference Sequence", )) html_parts.append("
") # Lollipop plot html_parts.append('
') html_parts.append(_make_lollipop_plot( gene_data["variants"], len(gene_data["sequence"]), title=f"{short_name} — Variant Positions & Effect Scores", )) html_parts.append("
") # Score bar chart for this gene variant_names = [v["name"] for v in gene_data["variants"]] variant_scores = [v["score"] for v in gene_data["variants"]] html_parts.append('
') html_parts.append(_make_bar_chart( variant_names, {"VEP Score": variant_scores}, title=f"{short_name} — Log-Likelihood Ratio Scores", colors=[EFFECT_COLORS.get(v["known_effect"], "#6b7280") for v in gene_data["variants"]], value_format=".3f", )) html_parts.append("
") # Variant detail cards for v in gene_data["variants"]: badge = EFFECT_BADGES.get(v["known_effect"], "badge-info") score_color = "#dc2626" if v["score"] < -0.1 else "#059669" if v["score"] > 0.05 else "#f59e0b" html_parts.append( f'
' f'
' f'{v["name"]}' f'{v["known_effect"]}' f'
' f'
' f'
Ref base: ' f'{v["ref"]}
' f'
Alt base: ' f'{v["alt"]}
' f'
VEP Score: ' f'{v["score"]:.4f}
' f'
' f'
{v.get("clinical", "")}
' f'
' ) # ------------------------------------------------------------------ # Confusion matrix as heatmap # ------------------------------------------------------------------ html_parts.append("

Classification Performance

") html_parts.append( '
' "Using a simple threshold (score < -0.05 = predicted pathogenic). " "This is zero-shot — no training on these specific variants." "
" ) conf_matrix = [[true_pos, false_neg], [false_pos, true_neg]] html_parts.append('
') html_parts.append(_make_heatmap( conf_matrix, ["Actual Pathogenic", "Actual Benign"], ["Predicted Pathogenic", "Predicted Benign"], title="Confusion Matrix (Known Variants Only)", value_format=".0f", width=400, height=300, )) html_parts.append("
") # ------------------------------------------------------------------ # Score distribution by known effect # ------------------------------------------------------------------ html_parts.append("

Score Distribution by Known Effect

") pathogenic_scores = [v["score"] for v in all_variants if v["known_effect"] == "pathogenic"] benign_scores = [v["score"] for v in all_variants if v["known_effect"] == "benign"] uncertain_scores = [v["score"] for v in all_variants if v["known_effect"] == "uncertain"] stats_html = '
' for label, scores, color in [ ("Pathogenic", pathogenic_scores, "#dc2626"), ("Benign", benign_scores, "#059669"), ("Uncertain", uncertain_scores, "#f59e0b"), ]: if scores: mean_s = sum(scores) / len(scores) min_s = min(scores) max_s = max(scores) stats_html += ( f'
' f'{label} (n={len(scores)})
' f'Mean: {mean_s:.4f}
' f'Range: [{min_s:.4f}, {max_s:.4f}]' f'
' ) stats_html += "
" html_parts.append(stats_html) # Summary analysis = { "total_variants": len(all_variants), "total_known": total_known, "accuracy": round(accuracy, 4), "precision": round(precision, 4), "recall": round(recall, 4), "true_pos": true_pos, "false_pos": false_pos, "true_neg": true_neg, "false_neg": false_neg, "pathogenic_mean_score": round(sum(pathogenic_scores) / len(pathogenic_scores), 4) if pathogenic_scores else None, "benign_mean_score": round(sum(benign_scores) / len(benign_scores), 4) if benign_scores else None, } await flyte.report.replace.aio(_wrap_report("\n".join(html_parts)), do_flush=True) return json.dumps(analysis) # ------------------------------------------------------------------ # Task 4: Generate comprehensive summary report # ------------------------------------------------------------------ @cpu_env.task(report=True) async def generate_summary( scores_json: str, analysis_json: str, ) -> str: """Generate the final summary report combining all results.""" results = json.loads(scores_json) analysis = json.loads(analysis_json) html_parts = [ "

Genomic Variant Effect Prediction — Summary

", '
' "This pipeline uses HuggingFace Carbon, an autoregressive genomic foundation model " "trained on 1 trillion tokens of DNA sequence, to perform zero-shot variant effect " "prediction. No fine-tuning or labeled training data was used — the model scores " "variants purely based on its learned understanding of DNA sequence grammar." "
", ] # Key metrics html_parts.append('
') html_parts.append(f'
{len(results)}
Genes Analyzed
') html_parts.append(f'
{analysis["total_variants"]}
Variants Scored
') html_parts.append(f'
{analysis["accuracy"]:.0%}
Direction Accuracy
') html_parts.append(f'
{analysis["precision"]:.0%}
Precision
') html_parts.append(f'
{analysis["recall"]:.0%}
Recall
') html_parts.append("
") # Gene summary table html_parts.append("

Per-Gene Summary

") html_parts.append( "" "" ) for gene_name, gene_data in results.items(): variants = gene_data["variants"] scores = [v["score"] for v in variants] mean_score = sum(scores) / len(scores) if scores else 0 n_path = sum(1 for v in variants if v["known_effect"] == "pathogenic") n_benign = sum(1 for v in variants if v["known_effect"] == "benign") n_unc = sum(1 for v in variants if v["known_effect"] == "uncertain") short = gene_name.split("(")[0].strip() html_parts.append( f"" f"" f'' f'' f'' ) html_parts.append("
GeneVariantsMean ScorePathogenicBenignUncertain
{short}{len(variants)}{mean_score:.4f}{n_path}{n_benign}{n_unc}
") # Cross-gene heatmap: gene x metric gene_names = [g.split("(")[0].strip() for g in results.keys()] metrics = ["Mean Score", "Min Score", "Max Score", "# Variants"] matrix = [] for gene_data in results.values(): scores = [v["score"] for v in gene_data["variants"]] matrix.append([ sum(scores) / len(scores) if scores else 0, min(scores) if scores else 0, max(scores) if scores else 0, len(scores), ]) html_parts.append("

Gene-Level Metrics

") html_parts.append('
') html_parts.append(_make_heatmap( matrix, gene_names, metrics, title="Gene-Level VEP Score Summary", value_format=".2f", width=600, height=350, )) html_parts.append("
") # All variants ranked by score (most damaging first) html_parts.append("

All Variants Ranked by Impact

") all_vars_sorted = [] for gene_name, gene_data in results.items(): for v in gene_data["variants"]: all_vars_sorted.append({**v, "gene": gene_name.split("(")[0].strip()}) all_vars_sorted.sort(key=lambda x: x["score"]) html_parts.append( "" "" ) for i, v in enumerate(all_vars_sorted): badge = EFFECT_BADGES.get(v["known_effect"], "badge-info") score_color = "#dc2626" if v["score"] < -0.1 else "#059669" if v["score"] > 0.05 else "#f59e0b" html_parts.append( f'' f'' f'' f'' ) html_parts.append("
#GeneVariantScoreKnownClinical Significance
{i + 1}{v["gene"]}{v["name"]}{v["score"]:.4f}{v["known_effect"]}{v.get("clinical", "")}
") # Method note html_parts.append( '
' "Method: Zero-shot variant effect prediction using log-likelihood ratio scoring. " "For each variant, we compute score = log P(mutant sequence | Carbon) - log P(reference sequence | Carbon). " "Negative scores indicate the model considers the mutant less probable than the reference, " "which correlates with pathogenicity. This approach requires no fine-tuning and generalizes " "across genes and variant types.

" "Limitations: These are short sequence windows — real clinical VEP would use longer " "genomic context (Carbon supports up to 786kbp). The threshold for pathogenicity classification " "(-0.05) is a simple heuristic; clinical use requires calibrated thresholds per gene." "
" ) await flyte.report.replace.aio(_wrap_report("\n".join(html_parts)), do_flush=True) return json.dumps({"status": "complete", "analysis": analysis}) # ------------------------------------------------------------------ # Pipeline orchestrator # ------------------------------------------------------------------ # {{docs-fragment pipeline}} @cpu_env.task(report=True) async def pipeline( variants_json: str = "", model_name: str = "HuggingFaceBio/Carbon-3B", ) -> tuple[str, str]: """ End-to-end genomic variant effect prediction pipeline. Returns (scores JSON, analysis JSON). 1. Load and validate gene variants 2. Score variants with Carbon (log-likelihood ratio) 3. Analyze effects — accuracy, visualizations, classification 4. Generate comprehensive summary report """ log.info("Starting genomic variant effect prediction pipeline...") def _pipeline_progress(step: int, label: str) -> str: steps = [ "Load Variants", "Carbon Scoring", "Analyze Effects", "Generate Summary", ] dots = "" for i, s in enumerate(steps): if i + 1 < step: icon = '✓' elif i + 1 == step: icon = '●' else: icon = '○' dots += f"{icon} {s}" return f"""

Genomic Variant Effect Prediction

{dots}

{label}

""" # Stage 1: Load variants await flyte.report.replace.aio( _wrap_report(_pipeline_progress(1, "Loading and validating gene variants...")), do_flush=True, ) var_dir = await load_variants(variants_json=variants_json) # Stage 2: Score with Carbon await flyte.report.replace.aio( _wrap_report(_pipeline_progress(2, "Running Carbon model for variant effect scoring...")), do_flush=True, ) scores_json = await score_variants(variants_dir=var_dir, model_name=model_name) # Stage 3: Analyze effects await flyte.report.replace.aio( _wrap_report(_pipeline_progress(3, "Analyzing variant effects and generating visualizations...")), do_flush=True, ) analysis_json = await analyze_effects(scores_json=scores_json, variants_dir=var_dir) # Stage 4: Summary await flyte.report.replace.aio( _wrap_report(_pipeline_progress(4, "Generating comprehensive summary report...")), do_flush=True, ) summary_json = await generate_summary(scores_json=scores_json, analysis_json=analysis_json) # Final pipeline report analysis = json.loads(analysis_json) results = json.loads(scores_json) final_html = f"""

Pipeline Complete

{len(results)}
Genes Analyzed
{analysis['total_variants']}
Variants Scored
{analysis['accuracy']:.0%}
Direction Accuracy
{analysis['precision']:.0%}
Precision
{analysis['recall']:.0%}
Recall
Model: HuggingFace Carbon | Method: Zero-shot log-likelihood ratio scoring | Genes: {', '.join(g.split('(')[0].strip() for g in results.keys())}
All 4 pipeline stages completed successfully. View individual task reports for detailed visualizations including DNA sequence tracks, variant lollipop plots, VEP score charts, confusion matrices, and ranked variant tables.
""" await flyte.report.replace.aio(_wrap_report(final_html), do_flush=True) log.info("Pipeline complete.") return scores_json, analysis_json # {{/docs-fragment pipeline}} if __name__ == "__main__": flyte.init_from_config() run = flyte.run(pipeline) print(run.url) run.wait() ``` *Source: https://github.com/unionai/unionai-examples/blob/main/v2/tutorials/genomic_variant_effect/genomic_variant_effect.py* ``` # /// script # requires-python = ">=3.12" # dependencies = [ # "flyte>=2.4.0", # "torch>=2.9.0", # "transformers>=4.49.0", # "accelerate>=0.34.0", # "numpy", # ] # /// ``` ## Orchestrate the pipeline The `pipeline` task loads variants, scores them with Carbon, analyzes classification accuracy against known labels, and generates a summary report. ``` # /// script # requires-python = ">=3.12" # dependencies = [ # "flyte>=2.4.0", # "torch>=2.9.0", # "transformers>=4.49.0", # "accelerate>=0.34.0", # "numpy", # ] # main = "pipeline" # params = "" # /// import json import logging import math import os import tempfile import flyte import flyte.io import flyte.report # {{docs-fragment env}} main_img = flyte.Image.from_uv_script(__file__, name="genomic-variant-effect", pre=True) gpu_env = flyte.TaskEnvironment( name="genomic-variant-effect-gpu", image=main_img, resources=flyte.Resources(cpu=4, memory="24Gi", gpu=1), ) cpu_env = flyte.TaskEnvironment( name="genomic-variant-effect-cpu", image=main_img, resources=flyte.Resources(cpu=2, memory="6Gi"), depends_on=[gpu_env], ) # {{/docs-fragment env}} logging.basicConfig(level=logging.WARNING, format="%(message)s", force=True) log = logging.getLogger(__name__) log.setLevel(logging.INFO) # ------------------------------------------------------------------ # Default gene variants — clinically relevant mutations # ------------------------------------------------------------------ # Each entry: gene name -> { "sequence": reference DNA, "variants": [{ "pos": 0-indexed, "ref": base, "alt": base, "name": "...", "known_effect": "..." }] } # Sequences are short windows (~120-200bp) around the variant site for tractable inference. DEFAULT_GENE_VARIANTS = { "BRCA2 (Breast Cancer)": { "description": "Tumor suppressor critical for DNA repair via homologous recombination. Mutations dramatically increase breast and ovarian cancer risk.", "sequence": "ATGGCCTCGAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAGCAG", "variants": [ {"pos": 12, "ref": "A", "alt": "T", "name": "c.37A>T", "known_effect": "pathogenic", "clinical": "Nonsense mutation — truncates protein early"}, {"pos": 18, "ref": "G", "alt": "A", "name": "c.55G>A", "known_effect": "benign", "clinical": "Synonymous — no amino acid change"}, {"pos": 30, "ref": "C", "alt": "T", "name": "c.91C>T", "known_effect": "pathogenic", "clinical": "Missense in DNA-binding domain"}, {"pos": 45, "ref": "G", "alt": "C", "name": "c.136G>C", "known_effect": "uncertain", "clinical": "Variant of uncertain significance (VUS)"}, ], }, "TP53 (Tumor Suppressor)": { "description": "Guardian of the genome. Activates DNA repair, cell cycle arrest, and apoptosis. Mutated in >50% of human cancers.", "sequence": "ATGGAGGAGCCGCAGTCAGATCCTAGCGTGAGTTTGCACCCTTCAGAGACAGAAACCACTGGATTGGAGACTACTTCCTGAAACAACGTTCTGTCCCCCTTGCCGTCCCAAGCAATGGATGAT", "variants": [ {"pos": 15, "ref": "C", "alt": "T", "name": "R175H", "known_effect": "pathogenic", "clinical": "Hotspot — gain-of-function, dominant negative. Most common TP53 mutation in cancer"}, {"pos": 36, "ref": "T", "alt": "C", "name": "P72R", "known_effect": "benign", "clinical": "Common polymorphism — subtle effect on apoptosis efficiency"}, {"pos": 54, "ref": "C", "alt": "A", "name": "G245S", "known_effect": "pathogenic", "clinical": "Contact mutant — disrupts DNA binding"}, {"pos": 72, "ref": "T", "alt": "G", "name": "R248W", "known_effect": "pathogenic", "clinical": "Structural mutant — destabilizes DNA-binding loop"}, {"pos": 90, "ref": "C", "alt": "T", "name": "R273H", "known_effect": "pathogenic", "clinical": "Contact mutant — directly contacts DNA bases"}, ], }, "CFTR (Cystic Fibrosis)": { "description": "Chloride channel protein. Mutations cause cystic fibrosis — the most common lethal genetic disease in people of European descent.", "sequence": "ATGCAGAGGTCGCCTCTGGAAAAGGCCAGCGTTGTCTCCAAACTTTTTTTCAGCTGGACCAGACCAATTTTGAGGAAAGGATACAGACAGCGCCTGGAATTGTCAGACATATACCAAATCCCTTC", "variants": [ {"pos": 9, "ref": "G", "alt": "A", "name": "G85E", "known_effect": "pathogenic", "clinical": "Disrupts chloride channel processing"}, {"pos": 24, "ref": "C", "alt": "T", "name": "R117H", "known_effect": "pathogenic", "clinical": "Reduces channel conductance — milder CF phenotype"}, {"pos": 48, "ref": "T", "alt": "C", "name": "I148T", "known_effect": "benign", "clinical": "Previously misclassified — now known benign polymorphism"}, {"pos": 66, "ref": "A", "alt": "G", "name": "R334W", "known_effect": "pathogenic", "clinical": "Gating mutation — channel opens less frequently"}, ], }, "KRAS (Oncogene)": { "description": "GTPase signal switch. KRAS mutations are the most common oncogenic driver — found in ~25% of all human cancers, especially pancreatic, colorectal, and lung.", "sequence": "ATGACTGAATATAAACTTGTGGTAGTTGGAGCTGGTGGCGTAGGCAAGAGTGCCTTGACGATACAGCTAATTCAGAATCATTTTGTGGACGAATATGATCCAACAATAGAGGATTCCTACAGGAA", "variants": [ {"pos": 34, "ref": "G", "alt": "T", "name": "G12V", "known_effect": "pathogenic", "clinical": "Locks KRAS in active state — constitutive proliferation signal"}, {"pos": 35, "ref": "G", "alt": "A", "name": "G12D", "known_effect": "pathogenic", "clinical": "Most common KRAS mutation in pancreatic cancer"}, {"pos": 37, "ref": "G", "alt": "T", "name": "G13D", "known_effect": "pathogenic", "clinical": "Constitutively active — common in colorectal cancer"}, {"pos": 60, "ref": "C", "alt": "A", "name": "Q61K", "known_effect": "pathogenic", "clinical": "Impairs GTP hydrolysis — locked ON state"}, ], }, "HBB (Sickle Cell)": { "description": "Beta-globin subunit of hemoglobin. The sickle cell mutation (E6V) is the most well-known single-base disease variant in humans.", "sequence": "ATGGTGCATCTGACTCCTGAGGAGAAGTCTGCCGTTACTGCCCTGTGGGGCAAGGTGAACGTGGATGAAGTTGGTGGTGAGGCCCTGGGCAGGCTGCTGGTGGTCTACCCTTGGACCCAGAGG", "variants": [ {"pos": 17, "ref": "A", "alt": "T", "name": "E6V (HbS)", "known_effect": "pathogenic", "clinical": "THE sickle cell mutation — causes hemoglobin polymerization under low O2"}, {"pos": 19, "ref": "G", "alt": "A", "name": "E6K (HbC)", "known_effect": "pathogenic", "clinical": "Hemoglobin C disease — milder than sickle cell but causes crystal formation"}, {"pos": 36, "ref": "G", "alt": "A", "name": "E26K", "known_effect": "benign", "clinical": "Hemoglobin E — most common Hb variant worldwide, mild effect"}, {"pos": 78, "ref": "C", "alt": "T", "name": "Q39X", "known_effect": "pathogenic", "clinical": "Nonsense — causes beta-thalassemia (no functional beta-globin)"}, ], }, } # DNA base colors (classic genomics color scheme) BASE_COLORS = {"A": "#2ecc71", "T": "#e74c3c", "G": "#f39c12", "C": "#3498db"} BASE_COMPLEMENT = {"A": "T", "T": "A", "G": "C", "C": "G"} # Pathogenicity color scheme EFFECT_COLORS = { "pathogenic": "#dc2626", "benign": "#059669", "uncertain": "#f59e0b", } EFFECT_BADGES = { "pathogenic": "badge-danger", "benign": "badge-success", "uncertain": "badge-warning", } # ------------------------------------------------------------------ # Report styling — genomics-themed deep blues and teals # ------------------------------------------------------------------ REPORT_CSS = """ """ def _wrap_report(html: str) -> str: return f'{REPORT_CSS}
{html}
' # ------------------------------------------------------------------ # SVG chart helpers # ------------------------------------------------------------------ def _make_bar_chart( labels: list[str], series: dict[str, list[float]], title: str = "", colors: list[str] | None = None, width: int = 700, height: int = 300, value_format: str = ".2f", ) -> str: """Generate an SVG grouped bar chart.""" if not labels: return "" default_colors = ["#2563eb", "#1e3a5f", "#3b82f6", "#60a5fa", "#93c5fd"] colors = colors or default_colors ml, mr, mt, mb = 70, 20, 40, 80 cw = width - ml - mr ch = height - mt - mb all_vals = [v for vals in series.values() for v in vals] y_max = max(abs(v) for v in all_vals) if all_vals else 1 y_min = min(all_vals) if all_vals else 0 # For VEP scores (negative = more damaging), we need to handle negative values if y_min >= 0: y_min_plot = 0 y_max_plot = y_max * 1.15 or 1 else: y_max_plot = max(y_max * 1.15, 0.1) y_min_plot = y_min * 1.15 y_range = y_max_plot - y_min_plot or 1 n_groups = len(labels) n_series = len(series) group_width = cw / n_groups bar_width = group_width * 0.7 / max(n_series, 1) gap = group_width * 0.15 def sy(v): return mt + ch - (v - y_min_plot) / y_range * ch svg = [ f'', f'', ] # Grid lines for i in range(6): y_tick = y_min_plot + y_range * i / 5 py = sy(y_tick) svg.append( f'' ) svg.append( f'{y_tick:{value_format}}' ) # Zero line if y_min_plot < 0 < y_max_plot: zy = sy(0) svg.append( f'' ) # Bars for gi, label in enumerate(labels): gx = ml + gi * group_width + gap for si, (name, vals) in enumerate(series.items()): color = colors[si % len(colors)] bx = gx + si * bar_width val = vals[gi] if val >= 0: by = sy(val) bh = sy(0) - by if y_min_plot < 0 else mt + ch - by else: by = sy(0) if y_min_plot < 0 else mt + ch bh = sy(val) - by svg.append( f'' ) text_y = by - 4 if val >= 0 else by + bh + 12 svg.append( f'' f'{val:{value_format}}' ) # Rotated group label lx = gx + n_series * bar_width / 2 svg.append( f'{label}' ) # Title if title: svg.append( f'{title}' ) # Legend if n_series > 1: lx = ml + cw - len(series) * 110 for si, name in enumerate(series): color = colors[si % len(colors)] svg.append( f'' ) svg.append( f'{name}' ) svg.append("") return "\n".join(svg) def _make_heatmap( matrix: list[list[float]], row_labels: list[str], col_labels: list[str], title: str = "", width: int = 700, height: int = 500, value_format: str = ".2f", diverging: bool = False, ) -> str: """Generate an SVG heatmap. If diverging=True, uses red-white-blue scale centered at 0.""" n_rows = len(matrix) n_cols = len(matrix[0]) if matrix else 0 if not n_rows or not n_cols: return "" show_values = n_rows <= 10 and n_cols <= 12 flat = [v for row in matrix for v in row] v_min = min(flat) v_max = max(flat) if diverging: abs_max = max(abs(v_min), abs(v_max)) or 1 def get_color(v): t = v / abs_max # -1 to 1 if t < 0: # White to red (negative = damaging) r = 255 g = int(255 * (1 + t)) b = int(255 * (1 + t)) else: # White to blue (positive = benign) r = int(255 * (1 - t)) g = int(255 * (1 - t)) b = 255 return f"rgb({r},{g},{b})" else: v_range = v_max - v_min or 1 def get_color(v): t = (v - v_min) / v_range r = int(255 - t * (255 - 30)) g = int(255 - t * (255 - 58)) b = int(255 - t * (255 - 95)) return f"rgb({r},{g},{b})" # Layout ml = max(140, max(len(l) for l in row_labels) * 7 + 20) if row_labels else 140 mr = 20 mt = 80 if col_labels else 40 mb = 30 cw = width - ml - mr ch = height - mt - mb cell_w = cw / n_cols cell_h = ch / n_rows svg = [ f'', f'', ] if title: svg.append( f'{title}' ) # Column labels (rotated) for j, label in enumerate(col_labels): cx = ml + j * cell_w + cell_w / 2 svg.append( f'{label}' ) # Row labels + cells for i, row_label in enumerate(row_labels): ry = mt + i * cell_h + cell_h / 2 svg.append( f'{row_label}' ) for j in range(n_cols): val = matrix[i][j] color = get_color(val) cx = ml + j * cell_w cy = mt + i * cell_h svg.append( f'' ) if show_values: if diverging: t = abs(val) / (max(abs(v_min), abs(v_max)) or 1) else: t = (val - v_min) / (v_max - v_min or 1) text_color = "#fff" if t > 0.55 else "#1a1a2e" font_size = min(10, int(cell_w / 4), int(cell_h / 2.5)) font_size = max(7, font_size) svg.append( f'{val:{value_format}}' ) svg.append("") return "\n".join(svg) def _make_dna_track( sequence: str, variants: list[dict], gene_name: str = "", width: int = 900, ) -> str: """Render a color-coded DNA sequence track with variant positions highlighted.""" chars_per_line = 60 char_w = 11 line_h = 22 label_w = 50 n_lines = (len(sequence) + chars_per_line - 1) // chars_per_line # Extra space for variant annotations variant_positions = {v["pos"] for v in variants} svg_h = n_lines * line_h + 60 svg = [ f'', f'', ] if gene_name: svg.append( f'{gene_name}' ) y_offset = 28 for line_idx in range(n_lines): start = line_idx * chars_per_line end = min(start + chars_per_line, len(sequence)) chunk = sequence[start:end] y = y_offset + line_idx * line_h # Position label svg.append( f'{start + 1}' ) for ci, base in enumerate(chunk): abs_pos = start + ci x = label_w + ci * char_w color = BASE_COLORS.get(base, "#6b7280") is_variant = abs_pos in variant_positions if is_variant: # Red highlight for variant positions svg.append( f'' ) # Small triangle marker above svg.append( f'' ) else: # Subtle background svg.append( f'' ) svg.append( f'{base}' ) # Spacer every 10 bases if (abs_pos + 1) % 10 == 0 and ci < len(chunk) - 1: svg.append( f'' ) # Legend ly = svg_h - 12 lx = label_w for base, color in BASE_COLORS.items(): svg.append(f'') svg.append(f'{base}') lx += 30 lx += 10 svg.append(f'') svg.append(f'Variant site') svg.append("") return "\n".join(svg) def _make_lollipop_plot( variants: list[dict], seq_length: int, title: str = "", width: int = 700, height: int = 200, ) -> str: """Generate a lollipop plot showing variant positions along a gene with effect scores.""" if not variants: return "" ml, mr, mt, mb = 50, 30, 40, 40 cw = width - ml - mr ch = height - mt - mb svg = [ f'', f'', ] if title: svg.append( f'{title}' ) # Gene track (horizontal bar) track_y = mt + ch * 0.7 svg.append( f'' ) # Position labels at ends svg.append( f'1' ) svg.append( f'{seq_length}' ) # Lollipops scores = [v.get("score", 0) for v in variants] max_score = max(abs(s) for s in scores) if scores else 1 for v in variants: pos = v["pos"] score = v.get("score", 0) effect = v.get("known_effect", "uncertain") color = EFFECT_COLORS.get(effect, "#6b7280") x = ml + (pos / max(seq_length - 1, 1)) * cw stem_h = max(20, abs(score) / max_score * (ch * 0.6)) circle_y = track_y - stem_h - 6 # Stem svg.append( f'' ) # Circle svg.append( f'' ) # Label svg.append( f'{v.get("name", "")}' ) # Legend lx = ml for effect, color in EFFECT_COLORS.items(): svg.append(f'') svg.append( f'' f'{effect.title()}' ) lx += len(effect) * 7 + 24 svg.append("") return "\n".join(svg) def _make_score_comparison_chart( gene_results: dict, width: int = 800, height: int = 350, ) -> str: """Generate a dot plot comparing VEP scores across all genes, colored by known effect.""" all_variants = [] for gene_name, gene_data in gene_results.items(): for v in gene_data["variants"]: all_variants.append({ "gene": gene_name.split("(")[0].strip(), "name": v["name"], "score": v.get("score", 0), "known_effect": v["known_effect"], }) if not all_variants: return "" ml, mr, mt, mb = 120, 30, 40, 30 cw = width - ml - mr ch = height - mt - mb scores = [v["score"] for v in all_variants] s_min, s_max = min(scores), max(scores) s_pad = (s_max - s_min) * 0.1 or 0.5 s_min -= s_pad s_max += s_pad s_range = s_max - s_min or 1 def sx(s): return ml + (s - s_min) / s_range * cw svg = [ f'', f'', f'' f'Variant Effect Scores — All Genes', ] # Grid for i in range(6): gx = s_min + s_range * i / 5 px = sx(gx) svg.append( f'' ) svg.append( f'{gx:.2f}' ) # Zero line if s_min < 0 < s_max: zx = sx(0) svg.append( f'' ) # Rows per variant row_h = ch / max(len(all_variants), 1) for i, v in enumerate(all_variants): y = mt + i * row_h + row_h / 2 color = EFFECT_COLORS.get(v["known_effect"], "#6b7280") px = sx(v["score"]) # Label label = f'{v["gene"]} {v["name"]}' svg.append( f'{label}' ) # Connector line svg.append( f'' ) # Dot svg.append( f'' ) svg.append("") return "\n".join(svg) # ------------------------------------------------------------------ # Task 1: Load and validate gene variants # ------------------------------------------------------------------ @cpu_env.task(cache="auto") async def load_variants( variants_json: str = "", ) -> flyte.io.Dir: """Load gene variant definitions, validate sequences, and save to a temp directory.""" if variants_json: genes = json.loads(variants_json) else: genes = DEFAULT_GENE_VARIANTS # Validate valid_bases = set("ATGC") for gene_name, gene_data in genes.items(): seq = gene_data["sequence"].upper() invalid = set(seq) - valid_bases if invalid: log.warning(f"{gene_name}: invalid bases {invalid} — removing them") seq = "".join(b for b in seq if b in valid_bases) gene_data["sequence"] = seq for v in gene_data["variants"]: pos = v["pos"] if pos < 0 or pos >= len(seq): log.warning(f"{gene_name} variant {v['name']}: position {pos} out of range [0, {len(seq)})") elif seq[pos] != v["ref"]: log.warning(f"{gene_name} variant {v['name']}: expected ref={v['ref']} at pos {pos}, found {seq[pos]}") total_variants = sum(len(g["variants"]) for g in genes.values()) log.info(f"Loaded {len(genes)} genes with {total_variants} variants") out_dir = tempfile.mkdtemp(prefix="genomic_vep_") with open(os.path.join(out_dir, "genes.json"), "w") as f: json.dump(genes, f) return await flyte.io.Dir.from_local(out_dir) # ------------------------------------------------------------------ # Task 2: Run Carbon model for variant effect scoring # ------------------------------------------------------------------ @gpu_env.task(report=True) async def score_variants( variants_dir: flyte.io.Dir, model_name: str = "HuggingFaceBio/Carbon-3B", ) -> str: """Score each variant using Carbon's log-likelihood ratio. For each variant, we compute: score = log P(alt_sequence) - log P(ref_sequence) A negative score means the model considers the variant less likely than the reference — suggestive of a damaging/pathogenic effect. A score near zero means the model sees little difference (likely benign). """ import torch from transformers import AutoModelForCausalLM, AutoTokenizer log.info(f"Loading Carbon model: {model_name}") # Load model device = "cuda" if torch.cuda.is_available() else "cpu" if device == "cpu": log.warning("Running on CPU — inference will be slow. GPU recommended for production.") tokenizer = AutoTokenizer.from_pretrained(model_name, trust_remote_code=True) model = AutoModelForCausalLM.from_pretrained( model_name, trust_remote_code=True, dtype=torch.bfloat16 if device == "cuda" else torch.float32, ).to(device) model.eval() # Load variants variants_path = await variants_dir.download() with open(os.path.join(variants_path, "genes.json")) as f: genes = json.load(f) results = {} total_variants = sum(len(g["variants"]) for g in genes.values()) scored = 0 progress_html = """

Carbon Variant Effect Scoring

Model: {model}
Device: {device}
Progress: {scored}/{total} variants scored
""" for gene_name, gene_data in genes.items(): ref_seq = gene_data["sequence"] gene_results = { "description": gene_data.get("description", ""), "sequence": ref_seq, "variants": [], } # Score reference sequence ref_prompt = f"{ref_seq}" ref_inputs = tokenizer(ref_prompt, return_tensors="pt", add_special_tokens=False).to(device) with torch.no_grad(): ref_output = model(**ref_inputs, labels=ref_inputs["input_ids"]) ref_loss = ref_output.loss.item() ref_ll = -ref_loss * ref_inputs["input_ids"].shape[1] for variant in gene_data["variants"]: scored += 1 await flyte.report.replace.aio( _wrap_report(progress_html.format( model=model_name, device=device, scored=scored, total=total_variants )), do_flush=True, ) # Create mutant sequence pos = variant["pos"] alt_seq = ref_seq[:pos] + variant["alt"] + ref_seq[pos + 1:] # Score mutant alt_prompt = f"{alt_seq}" alt_inputs = tokenizer(alt_prompt, return_tensors="pt", add_special_tokens=False).to(device) with torch.no_grad(): alt_output = model(**alt_inputs, labels=alt_inputs["input_ids"]) alt_loss = alt_output.loss.item() alt_ll = -alt_loss * alt_inputs["input_ids"].shape[1] # VEP score: positive = model prefers alt (likely benign), negative = model prefers ref (likely pathogenic) vep_score = alt_ll - ref_ll gene_results["variants"].append({ **variant, "score": round(vep_score, 4), "ref_ll": round(ref_ll, 4), "alt_ll": round(alt_ll, 4), }) log.info( f" {gene_name} | {variant['name']}: score={vep_score:.4f} " f"(known: {variant['known_effect']})" ) results[gene_name] = gene_results # Generate scoring report html_parts = [ "

Carbon Variant Effect Scoring

", '
', f'
{len(genes)}
Genes
', f'
{total_variants}
Variants Scored
', f'
{model_name.split("/")[-1]}
Model
', f'
{device.upper()}
Device
', "
", ] # Per-gene tables for gene_name, gene_data in results.items(): html_parts.append(f'

{gene_name}

') html_parts.append(f'
{gene_data["description"]}
') html_parts.append("") for v in gene_data["variants"]: badge = EFFECT_BADGES.get(v["known_effect"], "badge-info") direction = "damaging" if v["score"] < -0.1 else "neutral" if abs(v["score"]) <= 0.1 else "tolerated" html_parts.append( f'' f'' f'' f'' f'' f'' f'' f'' ) html_parts.append("
VariantRefAltVEP ScoreKnown EffectClinical
{v["name"]}{v["ref"]}{v["alt"]}{v["score"]:.4f} ({direction}){v["known_effect"]}{v.get("clinical", "")}
") await flyte.report.replace.aio(_wrap_report("\n".join(html_parts)), do_flush=True) return json.dumps(results) # ------------------------------------------------------------------ # Task 3: Analyze and visualize variant effects # ------------------------------------------------------------------ @cpu_env.task(report=True) async def analyze_effects( scores_json: str, variants_dir: flyte.io.Dir, ) -> str: """Analyze VEP scores: classification accuracy, gene-level summaries, and rich visualizations.""" results = json.loads(scores_json) variants_path = await variants_dir.download() with open(os.path.join(variants_path, "genes.json")) as f: genes = json.load(f) html_parts = ["

Variant Effect Analysis

"] # ------------------------------------------------------------------ # Overall accuracy: does the model's score direction match known labels? # ------------------------------------------------------------------ all_variants = [] correct = 0 total_known = 0 true_pos = 0 false_pos = 0 true_neg = 0 false_neg = 0 for gene_name, gene_data in results.items(): for v in gene_data["variants"]: all_variants.append({**v, "gene": gene_name}) if v["known_effect"] in ("pathogenic", "benign"): total_known += 1 predicted_pathogenic = v["score"] < -0.05 actual_pathogenic = v["known_effect"] == "pathogenic" if predicted_pathogenic == actual_pathogenic: correct += 1 if predicted_pathogenic and actual_pathogenic: true_pos += 1 elif predicted_pathogenic and not actual_pathogenic: false_pos += 1 elif not predicted_pathogenic and actual_pathogenic: false_neg += 1 else: true_neg += 1 accuracy = correct / total_known if total_known else 0 precision = true_pos / (true_pos + false_pos) if (true_pos + false_pos) else 0 recall = true_pos / (true_pos + false_neg) if (true_pos + false_neg) else 0 html_parts.append('
') html_parts.append(f'
{accuracy:.0%}
Direction Accuracy
') html_parts.append(f'
{precision:.0%}
Precision (Pathogenic)
') html_parts.append(f'
{recall:.0%}
Recall (Pathogenic)
') html_parts.append(f'
{len(all_variants)}
Total Variants
') html_parts.append("
") html_parts.append( '
' "How to read VEP scores: Negative scores mean Carbon considers the variant " "less likely than the reference sequence — suggestive of a damaging effect. Scores near " "zero indicate the model sees little difference (likely benign). The magnitude indicates " "confidence." "
" ) # ------------------------------------------------------------------ # Cross-gene score comparison dot plot # ------------------------------------------------------------------ html_parts.append('
') html_parts.append(_make_score_comparison_chart(results)) html_parts.append("
") # ------------------------------------------------------------------ # Per-gene visualizations # ------------------------------------------------------------------ for gene_name, gene_data in results.items(): short_name = gene_name.split("(")[0].strip() html_parts.append(f'

{gene_name}

') html_parts.append(f'
{gene_data["description"]}
') # DNA track with variant positions highlighted html_parts.append('
') html_parts.append(_make_dna_track( gene_data["sequence"], gene_data["variants"], gene_name=f"{short_name} Reference Sequence", )) html_parts.append("
") # Lollipop plot html_parts.append('
') html_parts.append(_make_lollipop_plot( gene_data["variants"], len(gene_data["sequence"]), title=f"{short_name} — Variant Positions & Effect Scores", )) html_parts.append("
") # Score bar chart for this gene variant_names = [v["name"] for v in gene_data["variants"]] variant_scores = [v["score"] for v in gene_data["variants"]] html_parts.append('
') html_parts.append(_make_bar_chart( variant_names, {"VEP Score": variant_scores}, title=f"{short_name} — Log-Likelihood Ratio Scores", colors=[EFFECT_COLORS.get(v["known_effect"], "#6b7280") for v in gene_data["variants"]], value_format=".3f", )) html_parts.append("
") # Variant detail cards for v in gene_data["variants"]: badge = EFFECT_BADGES.get(v["known_effect"], "badge-info") score_color = "#dc2626" if v["score"] < -0.1 else "#059669" if v["score"] > 0.05 else "#f59e0b" html_parts.append( f'
' f'
' f'{v["name"]}' f'{v["known_effect"]}' f'
' f'
' f'
Ref base: ' f'{v["ref"]}
' f'
Alt base: ' f'{v["alt"]}
' f'
VEP Score: ' f'{v["score"]:.4f}
' f'
' f'
{v.get("clinical", "")}
' f'
' ) # ------------------------------------------------------------------ # Confusion matrix as heatmap # ------------------------------------------------------------------ html_parts.append("

Classification Performance

") html_parts.append( '
' "Using a simple threshold (score < -0.05 = predicted pathogenic). " "This is zero-shot — no training on these specific variants." "
" ) conf_matrix = [[true_pos, false_neg], [false_pos, true_neg]] html_parts.append('
') html_parts.append(_make_heatmap( conf_matrix, ["Actual Pathogenic", "Actual Benign"], ["Predicted Pathogenic", "Predicted Benign"], title="Confusion Matrix (Known Variants Only)", value_format=".0f", width=400, height=300, )) html_parts.append("
") # ------------------------------------------------------------------ # Score distribution by known effect # ------------------------------------------------------------------ html_parts.append("

Score Distribution by Known Effect

") pathogenic_scores = [v["score"] for v in all_variants if v["known_effect"] == "pathogenic"] benign_scores = [v["score"] for v in all_variants if v["known_effect"] == "benign"] uncertain_scores = [v["score"] for v in all_variants if v["known_effect"] == "uncertain"] stats_html = '
' for label, scores, color in [ ("Pathogenic", pathogenic_scores, "#dc2626"), ("Benign", benign_scores, "#059669"), ("Uncertain", uncertain_scores, "#f59e0b"), ]: if scores: mean_s = sum(scores) / len(scores) min_s = min(scores) max_s = max(scores) stats_html += ( f'
' f'{label} (n={len(scores)})
' f'Mean: {mean_s:.4f}
' f'Range: [{min_s:.4f}, {max_s:.4f}]' f'
' ) stats_html += "
" html_parts.append(stats_html) # Summary analysis = { "total_variants": len(all_variants), "total_known": total_known, "accuracy": round(accuracy, 4), "precision": round(precision, 4), "recall": round(recall, 4), "true_pos": true_pos, "false_pos": false_pos, "true_neg": true_neg, "false_neg": false_neg, "pathogenic_mean_score": round(sum(pathogenic_scores) / len(pathogenic_scores), 4) if pathogenic_scores else None, "benign_mean_score": round(sum(benign_scores) / len(benign_scores), 4) if benign_scores else None, } await flyte.report.replace.aio(_wrap_report("\n".join(html_parts)), do_flush=True) return json.dumps(analysis) # ------------------------------------------------------------------ # Task 4: Generate comprehensive summary report # ------------------------------------------------------------------ @cpu_env.task(report=True) async def generate_summary( scores_json: str, analysis_json: str, ) -> str: """Generate the final summary report combining all results.""" results = json.loads(scores_json) analysis = json.loads(analysis_json) html_parts = [ "

Genomic Variant Effect Prediction — Summary

", '
' "This pipeline uses HuggingFace Carbon, an autoregressive genomic foundation model " "trained on 1 trillion tokens of DNA sequence, to perform zero-shot variant effect " "prediction. No fine-tuning or labeled training data was used — the model scores " "variants purely based on its learned understanding of DNA sequence grammar." "
", ] # Key metrics html_parts.append('
') html_parts.append(f'
{len(results)}
Genes Analyzed
') html_parts.append(f'
{analysis["total_variants"]}
Variants Scored
') html_parts.append(f'
{analysis["accuracy"]:.0%}
Direction Accuracy
') html_parts.append(f'
{analysis["precision"]:.0%}
Precision
') html_parts.append(f'
{analysis["recall"]:.0%}
Recall
') html_parts.append("
") # Gene summary table html_parts.append("

Per-Gene Summary

") html_parts.append( "" "" ) for gene_name, gene_data in results.items(): variants = gene_data["variants"] scores = [v["score"] for v in variants] mean_score = sum(scores) / len(scores) if scores else 0 n_path = sum(1 for v in variants if v["known_effect"] == "pathogenic") n_benign = sum(1 for v in variants if v["known_effect"] == "benign") n_unc = sum(1 for v in variants if v["known_effect"] == "uncertain") short = gene_name.split("(")[0].strip() html_parts.append( f"" f"" f'' f'' f'' ) html_parts.append("
GeneVariantsMean ScorePathogenicBenignUncertain
{short}{len(variants)}{mean_score:.4f}{n_path}{n_benign}{n_unc}
") # Cross-gene heatmap: gene x metric gene_names = [g.split("(")[0].strip() for g in results.keys()] metrics = ["Mean Score", "Min Score", "Max Score", "# Variants"] matrix = [] for gene_data in results.values(): scores = [v["score"] for v in gene_data["variants"]] matrix.append([ sum(scores) / len(scores) if scores else 0, min(scores) if scores else 0, max(scores) if scores else 0, len(scores), ]) html_parts.append("

Gene-Level Metrics

") html_parts.append('
') html_parts.append(_make_heatmap( matrix, gene_names, metrics, title="Gene-Level VEP Score Summary", value_format=".2f", width=600, height=350, )) html_parts.append("
") # All variants ranked by score (most damaging first) html_parts.append("

All Variants Ranked by Impact

") all_vars_sorted = [] for gene_name, gene_data in results.items(): for v in gene_data["variants"]: all_vars_sorted.append({**v, "gene": gene_name.split("(")[0].strip()}) all_vars_sorted.sort(key=lambda x: x["score"]) html_parts.append( "" "" ) for i, v in enumerate(all_vars_sorted): badge = EFFECT_BADGES.get(v["known_effect"], "badge-info") score_color = "#dc2626" if v["score"] < -0.1 else "#059669" if v["score"] > 0.05 else "#f59e0b" html_parts.append( f'' f'' f'' f'' ) html_parts.append("
#GeneVariantScoreKnownClinical Significance
{i + 1}{v["gene"]}{v["name"]}{v["score"]:.4f}{v["known_effect"]}{v.get("clinical", "")}
") # Method note html_parts.append( '
' "Method: Zero-shot variant effect prediction using log-likelihood ratio scoring. " "For each variant, we compute score = log P(mutant sequence | Carbon) - log P(reference sequence | Carbon). " "Negative scores indicate the model considers the mutant less probable than the reference, " "which correlates with pathogenicity. This approach requires no fine-tuning and generalizes " "across genes and variant types.

" "Limitations: These are short sequence windows — real clinical VEP would use longer " "genomic context (Carbon supports up to 786kbp). The threshold for pathogenicity classification " "(-0.05) is a simple heuristic; clinical use requires calibrated thresholds per gene." "
" ) await flyte.report.replace.aio(_wrap_report("\n".join(html_parts)), do_flush=True) return json.dumps({"status": "complete", "analysis": analysis}) # ------------------------------------------------------------------ # Pipeline orchestrator # ------------------------------------------------------------------ # {{docs-fragment pipeline}} @cpu_env.task(report=True) async def pipeline( variants_json: str = "", model_name: str = "HuggingFaceBio/Carbon-3B", ) -> tuple[str, str]: """ End-to-end genomic variant effect prediction pipeline. Returns (scores JSON, analysis JSON). 1. Load and validate gene variants 2. Score variants with Carbon (log-likelihood ratio) 3. Analyze effects — accuracy, visualizations, classification 4. Generate comprehensive summary report """ log.info("Starting genomic variant effect prediction pipeline...") def _pipeline_progress(step: int, label: str) -> str: steps = [ "Load Variants", "Carbon Scoring", "Analyze Effects", "Generate Summary", ] dots = "" for i, s in enumerate(steps): if i + 1 < step: icon = '✓' elif i + 1 == step: icon = '●' else: icon = '○' dots += f"{icon} {s}" return f"""

Genomic Variant Effect Prediction

{dots}

{label}

""" # Stage 1: Load variants await flyte.report.replace.aio( _wrap_report(_pipeline_progress(1, "Loading and validating gene variants...")), do_flush=True, ) var_dir = await load_variants(variants_json=variants_json) # Stage 2: Score with Carbon await flyte.report.replace.aio( _wrap_report(_pipeline_progress(2, "Running Carbon model for variant effect scoring...")), do_flush=True, ) scores_json = await score_variants(variants_dir=var_dir, model_name=model_name) # Stage 3: Analyze effects await flyte.report.replace.aio( _wrap_report(_pipeline_progress(3, "Analyzing variant effects and generating visualizations...")), do_flush=True, ) analysis_json = await analyze_effects(scores_json=scores_json, variants_dir=var_dir) # Stage 4: Summary await flyte.report.replace.aio( _wrap_report(_pipeline_progress(4, "Generating comprehensive summary report...")), do_flush=True, ) summary_json = await generate_summary(scores_json=scores_json, analysis_json=analysis_json) # Final pipeline report analysis = json.loads(analysis_json) results = json.loads(scores_json) final_html = f"""

Pipeline Complete

{len(results)}
Genes Analyzed
{analysis['total_variants']}
Variants Scored
{analysis['accuracy']:.0%}
Direction Accuracy
{analysis['precision']:.0%}
Precision
{analysis['recall']:.0%}
Recall
Model: HuggingFace Carbon | Method: Zero-shot log-likelihood ratio scoring | Genes: {', '.join(g.split('(')[0].strip() for g in results.keys())}
All 4 pipeline stages completed successfully. View individual task reports for detailed visualizations including DNA sequence tracks, variant lollipop plots, VEP score charts, confusion matrices, and ranked variant tables.
""" await flyte.report.replace.aio(_wrap_report(final_html), do_flush=True) log.info("Pipeline complete.") return scores_json, analysis_json # {{/docs-fragment pipeline}} if __name__ == "__main__": flyte.init_from_config() run = flyte.run(pipeline) print(run.url) run.wait() ``` *Source: https://github.com/unionai/unionai-examples/blob/main/v2/tutorials/genomic_variant_effect/genomic_variant_effect.py* ## Run the workflow From the [example directory](https://github.com/unionai/unionai-examples/tree/main/v2/tutorials/genomic_variant_effect): ``` cd v2/tutorials/genomic_variant_effect uv run --script genomic_variant_effect.py ``` Use a smaller Carbon model for faster iteration: ``` flyte run genomic_variant_effect.py pipeline --model_name HuggingFaceBio/Carbon-500M ``` Negative VEP scores indicate the model prefers the reference allele over the alternate — a signal correlated with pathogenicity in this zero-shot setup. --- **Source**: https://github.com/unionai/unionai-docs/blob/main/content/tutorials/biotech-healthcare/genomic-variant-effect/_index.md **HTML**: https://www.union.ai/docs/v2/union/tutorials/biotech-healthcare/genomic-variant-effect/