From c1594a56e2d892089ca2bb3d90376f385fb9e2cd Mon Sep 17 00:00:00 2001 From: xipang Date: Wed, 10 Dec 2025 23:39:32 +0100 Subject: [PATCH 1/4] push workflow --- .../723.gpu-content-moderation/config.json | 6 + .../definition.json | 26 +++ .../723.gpu-content-moderation/input.py | 26 +++ .../python/ingest.py | 92 +++++++++ .../python/moderate.py | 182 ++++++++++++++++++ .../python/review.py | 114 +++++++++++ 6 files changed, 446 insertions(+) create mode 100644 benchmarks/600.workflows/723.gpu-content-moderation/config.json create mode 100644 benchmarks/600.workflows/723.gpu-content-moderation/definition.json create mode 100644 benchmarks/600.workflows/723.gpu-content-moderation/input.py create mode 100644 benchmarks/600.workflows/723.gpu-content-moderation/python/ingest.py create mode 100644 benchmarks/600.workflows/723.gpu-content-moderation/python/moderate.py create mode 100644 benchmarks/600.workflows/723.gpu-content-moderation/python/review.py diff --git a/benchmarks/600.workflows/723.gpu-content-moderation/config.json b/benchmarks/600.workflows/723.gpu-content-moderation/config.json new file mode 100644 index 000000000..a11a12730 --- /dev/null +++ b/benchmarks/600.workflows/723.gpu-content-moderation/config.json @@ -0,0 +1,6 @@ +{ + "timeout": 240, + "memory": 4096, + "languages": ["python"], + "modules": [] +} diff --git a/benchmarks/600.workflows/723.gpu-content-moderation/definition.json b/benchmarks/600.workflows/723.gpu-content-moderation/definition.json new file mode 100644 index 000000000..52b7a9605 --- /dev/null +++ b/benchmarks/600.workflows/723.gpu-content-moderation/definition.json @@ -0,0 +1,26 @@ +{ + "root": "ingest", + "states": { + "ingest": { + "type": "task", + "func_name": "ingest", + "next": "moderate-map" + }, + "moderate-map": { + "type": "map", + "root": "moderate", + "array": "posts", + "next": "review", + "states": { + "moderate": { + "type": "task", + "func_name": "moderate" + } + } + }, + "review": { + "type": "task", + "func_name": "review" + } + } +} diff --git a/benchmarks/600.workflows/723.gpu-content-moderation/input.py b/benchmarks/600.workflows/723.gpu-content-moderation/input.py new file mode 100644 index 000000000..79a50a761 --- /dev/null +++ b/benchmarks/600.workflows/723.gpu-content-moderation/input.py @@ -0,0 +1,26 @@ +size_generators = { + "test": (30, 100), # 30 posts, avg 100 tokens per post + "small": (100, 200), + "large": (250, 300), +} + + +def buckets_count(): + # No object storage buckets required for this workflow. + return (0, 0) + + +def generate_input( + data_dir, + size, + benchmarks_bucket, + input_buckets, + output_buckets, + upload_func, + nosql_func, +): + n_posts, avg_tokens = size_generators[size] + return { + "n_posts": n_posts, + "avg_tokens": avg_tokens, + } diff --git a/benchmarks/600.workflows/723.gpu-content-moderation/python/ingest.py b/benchmarks/600.workflows/723.gpu-content-moderation/python/ingest.py new file mode 100644 index 000000000..8dc8cb272 --- /dev/null +++ b/benchmarks/600.workflows/723.gpu-content-moderation/python/ingest.py @@ -0,0 +1,92 @@ +import uuid +import numpy as np + + +POST_CATEGORIES = ["news", "personal", "commercial", "entertainment", "educational", "political", "social"] +LANGUAGES = ["en", "es", "fr", "de", "pt", "it"] + +SAFE_WORDS = [ + "cat", "dog", "food", "travel", "music", "art", "game", "tech", "book", "movie", + "friend", "family", "work", "love", "happy", "great", "amazing", "beautiful", "fun", "enjoy", + "today", "yesterday", "tomorrow", "share", "post", "comment", "like", "follow", "update" +] + +BORDERLINE_WORDS = [ + "debate", "opinion", "protest", "controversial", "argument", "disagree", "conflict", + "political", "election", "policy", "criticism", "oppose", "challenge" +] + +UNSAFE_WORDS = [ + "hate", "violence", "spam", "scam", "fake", "attack", "threat", "abuse", "harass", + "derogatory", "discriminate", "offensive", "vulgar", "explicit" +] + + +def generate_text_content(avg_tokens: int, seed: int, violation_prob: float = 0.2) -> tuple: + rng = np.random.default_rng(seed=seed) + + has_violation = rng.random() < violation_prob + + n_tokens = max(10, int(rng.normal(avg_tokens, avg_tokens * 0.3))) + + if has_violation: + violation_type = rng.choice( + ["hate_speech", "violence", "spam", "misinformation", "harassment"], + p=[0.25, 0.2, 0.3, 0.15, 0.1] + ) + + unsafe_ratio = rng.uniform(0.15, 0.4) + n_unsafe = int(n_tokens * unsafe_ratio) + n_safe = n_tokens - n_unsafe + + words = ( + list(rng.choice(UNSAFE_WORDS, size=n_unsafe, replace=True)) + + list(rng.choice(SAFE_WORDS, size=n_safe, replace=True)) + ) + else: + borderline_ratio = rng.uniform(0, 0.2) + n_borderline = int(n_tokens * borderline_ratio) + n_safe = n_tokens - n_borderline + + words = ( + list(rng.choice(BORDERLINE_WORDS, size=n_borderline, replace=True)) + + list(rng.choice(SAFE_WORDS, size=n_safe, replace=True)) + ) + violation_type = None + + rng.shuffle(words) + text = " ".join(words) + + return text, violation_type, n_tokens + + +def handler(event): + n_posts = int(event["n_posts"]) + avg_tokens = int(event["avg_tokens"]) + batch_id = event.get("batch_id", str(uuid.uuid4())[:8]) + + posts = [] + for idx in range(n_posts): + rng = np.random.default_rng(seed=idx) + + category = rng.choice(POST_CATEGORIES) + language = rng.choice(LANGUAGES) + + text, violation_type, n_tokens = generate_text_content(avg_tokens, seed=idx) + + post = { + "post_id": f"post-{batch_id}-{idx:04d}", + "category": category, + "language": language, + "text": text, + "text_tokens": n_tokens, + "timestamp": 1700000000 + idx * 60, + "user_id": f"user_{rng.integers(1000, 9999)}", + "engagement_score": float(rng.beta(2, 5)), # Simulates likes/shares + "batch_id": batch_id, + "true_violation": violation_type, # Ground truth for evaluation + } + + posts.append(post) + + return {"posts": posts} diff --git a/benchmarks/600.workflows/723.gpu-content-moderation/python/moderate.py b/benchmarks/600.workflows/723.gpu-content-moderation/python/moderate.py new file mode 100644 index 000000000..04c85af93 --- /dev/null +++ b/benchmarks/600.workflows/723.gpu-content-moderation/python/moderate.py @@ -0,0 +1,182 @@ +import numpy as np + + +UNSAFE_KEYWORDS = ["hate", "violence", "spam", "scam", "fake", "attack", "threat", "abuse", "harass"] +BORDERLINE_KEYWORDS = ["controversial", "debate", "protest", "argument"] + + +def tokenize_and_embed(text: str, seed: int) -> np.ndarray: + words = text.lower().split() + n_tokens = len(words) + + rng = np.random.default_rng(seed=seed) + embedding_dim = 768 + + embeddings = rng.normal(0, 0.1, (n_tokens, embedding_dim)).astype(np.float32) + + for i, word in enumerate(words): + if any(unsafe in word for unsafe in UNSAFE_KEYWORDS): + embeddings[i] += rng.normal(2.0, 0.5, embedding_dim).astype(np.float32) + elif any(border in word for border in BORDERLINE_KEYWORDS): + embeddings[i] += rng.normal(0.5, 0.3, embedding_dim).astype(np.float32) + + sentence_embedding = embeddings.mean(axis=0) + + return sentence_embedding + + +def transformer_classification(embedding: np.ndarray, seed: int) -> dict: + rng = np.random.default_rng(seed=seed) + + # Simulate classification head (linear layer + softmax) + embedding_norm = np.linalg.norm(embedding) + + # Higher norm suggests more extreme content (either very safe or very toxic) + base_score = 1.0 / (1.0 + np.exp(-embedding_norm / 10)) + + noise = rng.normal(0, 0.05) + toxicity_score = np.clip(base_score + noise, 0, 1) + + return { + "toxicity_score": float(toxicity_score), + "embedding_norm": float(embedding_norm), + } + + +def classify_violation_type(text: str, toxicity_score: float, seed: int) -> list: + rng = np.random.default_rng(seed=seed) + violations = [] + + words = text.lower() + + if "hate" in words or "discriminate" in words or "derogatory" in words: + confidence = min(0.95, toxicity_score + rng.uniform(0.1, 0.2)) + violations.append({ + "type": "hate_speech", + "confidence": float(confidence), + }) + + if "violence" in words or "attack" in words or "threat" in words: + confidence = min(0.95, toxicity_score + rng.uniform(0.1, 0.2)) + violations.append({ + "type": "violence", + "confidence": float(confidence), + }) + + if "spam" in words or "scam" in words: + confidence = min(0.9, toxicity_score + rng.uniform(0.05, 0.15)) + violations.append({ + "type": "spam", + "confidence": float(confidence), + }) + + if "fake" in words: + confidence = min(0.85, toxicity_score + rng.uniform(0.05, 0.15)) + violations.append({ + "type": "misinformation", + "confidence": float(confidence), + }) + + if "harass" in words or "abuse" in words: + confidence = min(0.9, toxicity_score + rng.uniform(0.1, 0.2)) + violations.append({ + "type": "harassment", + "confidence": float(confidence), + }) + + return violations + + +def sentiment_analysis(text: str, seed: int) -> dict: + rng = np.random.default_rng(seed=seed) + + positive_words = ["happy", "love", "great", "amazing", "beautiful", "enjoy", "fun"] + negative_words = ["hate", "violence", "abuse", "offensive", "threat"] + + words = text.lower().split() + pos_count = sum(1 for w in words if any(p in w for p in positive_words)) + neg_count = sum(1 for w in words if any(n in w for n in negative_words)) + + total = len(words) + if total > 0: + sentiment_score = (pos_count - neg_count) / total + else: + sentiment_score = 0.0 + + # Add noise + sentiment_score += rng.normal(0, 0.1) + sentiment_score = np.clip(sentiment_score, -1, 1) + + return { + "sentiment_score": float(sentiment_score), + "polarity": "positive" if sentiment_score > 0.1 else "negative" if sentiment_score < -0.1 else "neutral", + } + + +def calculate_spam_score(post: dict) -> float: + spam_score = 0.0 + + text = post.get("text", "") + words = text.lower().split() + + unique_words = len(set(words)) + if unique_words < len(words) * 0.3: + spam_score += 0.4 + + if "spam" in text or "scam" in text: + spam_score += 0.5 + + engagement = post.get("engagement_score", 0.5) + if engagement < 0.1: + spam_score += 0.2 + + return min(spam_score, 1.0) + + +def handler(post): + post_id = post["post_id"] + text = post.get("text", "") + category = post.get("category", "unknown") + + seed = hash(post_id) % 100000 + + embedding = tokenize_and_embed(text, seed) + + toxicity_result = transformer_classification(embedding, seed + 1) + toxicity_score = toxicity_result["toxicity_score"] + + violations = classify_violation_type(text, toxicity_score, seed + 2) + + sentiment = sentiment_analysis(text, seed + 3) + + spam_score = calculate_spam_score(post) + + overall_score = ( + 0.5 * toxicity_score + + 0.3 * spam_score + + 0.2 * (1.0 if sentiment["sentiment_score"] < -0.3 else 0.0) + ) + + if overall_score > 0.75: + action = "REMOVE" + elif overall_score > 0.5: + action = "REVIEW" + elif overall_score > 0.3: + action = "FLAG" + else: + action = "APPROVE" + + if spam_score > 0.7: + action = "REMOVE" + + return { + "post_id": post_id, + "category": category, + "toxicity_score": float(toxicity_score), + "spam_score": float(spam_score), + "sentiment": sentiment, + "violations": violations, + "overall_score": float(overall_score), + "action": action, + "true_violation": post.get("true_violation"), + } diff --git a/benchmarks/600.workflows/723.gpu-content-moderation/python/review.py b/benchmarks/600.workflows/723.gpu-content-moderation/python/review.py new file mode 100644 index 000000000..3f3bcf4ce --- /dev/null +++ b/benchmarks/600.workflows/723.gpu-content-moderation/python/review.py @@ -0,0 +1,114 @@ +def calculate_accuracy(posts: list) -> dict: + true_violations = [p for p in posts if p.get("true_violation") is not None] + + detected_violations = [p for p in posts if p.get("action") in ["REMOVE", "REVIEW"]] + + true_positives = len([ + p for p in posts + if p.get("true_violation") is not None and p.get("action") in ["REMOVE", "REVIEW"] + ]) + + false_positives = len([ + p for p in posts + if p.get("true_violation") is None and p.get("action") in ["REMOVE", "REVIEW"] + ]) + + false_negatives = len([ + p for p in posts + if p.get("true_violation") is not None and p.get("action") not in ["REMOVE", "REVIEW"] + ]) + + precision = true_positives / len(detected_violations) if detected_violations else 0.0 + recall = true_positives / len(true_violations) if true_violations else 0.0 + f1 = 2 * (precision * recall) / (precision + recall) if (precision + recall) > 0 else 0.0 + + return { + "precision": round(precision, 4), + "recall": round(recall, 4), + "f1_score": round(f1, 4), + "true_positives": true_positives, + "false_positives": false_positives, + "false_negatives": false_negatives, + "total_actual_violations": len(true_violations), + } + + +def handler(event): + posts = event.get("posts", []) + + action_counts = {} + for post in posts: + action = post.get("action", "UNKNOWN") + action_counts[action] = action_counts.get(action, 0) + 1 + + violation_counts = {} + for post in posts: + for violation in post.get("violations", []): + vtype = violation["type"] + violation_counts[vtype] = violation_counts.get(vtype, 0) + 1 + + category_stats = {} + for post in posts: + category = post.get("category", "unknown") + toxicity = post.get("toxicity_score", 0) + + if category not in category_stats: + category_stats[category] = { + "count": 0, + "avg_toxicity": 0.0, + "removed": 0, + } + + category_stats[category]["count"] += 1 + category_stats[category]["avg_toxicity"] += toxicity + if post.get("action") == "REMOVE": + category_stats[category]["removed"] += 1 + + for category in category_stats: + count = category_stats[category]["count"] + category_stats[category]["avg_toxicity"] = round( + category_stats[category]["avg_toxicity"] / count, 4 + ) + + needs_review = [ + { + "post_id": p["post_id"], + "category": p.get("category", "unknown"), + "toxicity_score": p.get("toxicity_score", 0), + "spam_score": p.get("spam_score", 0), + "violations": [v["type"] for v in p.get("violations", [])], + "sentiment": p.get("sentiment", {}).get("polarity", "neutral"), + } + for p in posts + if p.get("action") == "REVIEW" + ] + + needs_review.sort(key=lambda p: p["toxicity_score"], reverse=True) + + accuracy = calculate_accuracy(posts) + + avg_toxicity = sum(p.get("toxicity_score", 0) for p in posts) / len(posts) + avg_spam = sum(p.get("spam_score", 0) for p in posts) / len(posts) + + removal_rate = action_counts.get("REMOVE", 0) / len(posts) if posts else 0 + quality_level = "POOR" if removal_rate > 0.3 else "MODERATE" if removal_rate > 0.15 else "GOOD" + + sentiment_dist = {"positive": 0, "negative": 0, "neutral": 0} + for post in posts: + polarity = post.get("sentiment", {}).get("polarity", "neutral") + sentiment_dist[polarity] = sentiment_dist.get(polarity, 0) + 1 + + return { + "summary": f"Moderated {len(posts)} text posts", + "total_posts": len(posts), + "action_summary": action_counts, + "violation_summary": violation_counts, + "category_statistics": category_stats, + "sentiment_distribution": sentiment_dist, + "quality_level": quality_level, + "removal_rate": round(removal_rate, 4), + "avg_toxicity_score": round(avg_toxicity, 4), + "avg_spam_score": round(avg_spam, 4), + "posts_needing_review": needs_review[:20], # Top 20 + "accuracy_metrics": accuracy, + } From 38cd0f8ed7c0f37d71001e2924dd9bd2d2dc29f7 Mon Sep 17 00:00:00 2001 From: xipang Date: Wed, 10 Dec 2025 23:43:50 +0100 Subject: [PATCH 2/4] push workflow --- .../695.protein-screen/definition.json | 26 ++++ .../710.gpu-single-infer/config.json | 6 + .../710.gpu-single-infer/definition.json | 19 +++ .../710.gpu-single-infer/input.py | 15 ++ .../python/fetch_image.py | 13 ++ .../710.gpu-single-infer/python/infer.py | 27 ++++ .../python/requirements.txt | 3 + .../710.gpu-single-infer/python/summarize.py | 6 + .../711.gpu-embed-batch/config.json | 6 + .../711.gpu-embed-batch/definition.json | 19 +++ .../711.gpu-embed-batch/input.py | 15 ++ .../711.gpu-embed-batch/python/embed.py | 23 +++ .../711.gpu-embed-batch/python/fetch_batch.py | 11 ++ .../python/requirements.txt | 2 + .../711.gpu-embed-batch/python/summarize.py | 7 + .../717.gpu-traffic-count/config.json | 6 + .../717.gpu-traffic-count/definition.json | 27 ++++ .../717.gpu-traffic-count/input.py | 63 +++++++++ .../717.gpu-traffic-count/python/aggregate.py | 65 +++++++++ .../717.gpu-traffic-count/python/detect.py | 81 +++++++++++ .../python/requirements.txt | 3 + .../717.gpu-traffic-count/python/split.py | 26 ++++ .../718.gpu-document-qna/config.json | 6 + .../718.gpu-document-qna/definition.json | 26 ++++ .../718.gpu-document-qna/input.py | 59 ++++++++ .../718.gpu-document-qna/python/answer.py | 67 +++++++++ .../718.gpu-document-qna/python/embed.py | 75 ++++++++++ .../python/requirements.txt | 4 + .../718.gpu-document-qna/python/split.py | 25 ++++ .../720.gpu-medical-scan/config.json | 6 + .../720.gpu-medical-scan/definition.json | 26 ++++ .../720.gpu-medical-scan/input.py | 28 ++++ .../720.gpu-medical-scan/python/aggregate.py | 55 ++++++++ .../720.gpu-medical-scan/python/preprocess.py | 62 +++++++++ .../720.gpu-medical-scan/python/segment.py | 89 ++++++++++++ .../723.gpu-content-moderation/input.py | 2 +- .../python/ingest.py | 85 ++++++++++-- .../python/moderate.py | 74 ++++++---- .../python/review.py | 37 +++-- sebs/sonataflow/__init__.py | 2 + sebs/sonataflow/translator.py | 131 ++++++++++++++++++ sebs/sonataflow/workflow.py | 47 +++++++ tools/sonataflow_builder.py | 111 +++++++++++++++ 43 files changed, 1431 insertions(+), 55 deletions(-) create mode 100644 benchmarks/600.workflows/695.protein-screen/definition.json create mode 100644 benchmarks/600.workflows/710.gpu-single-infer/config.json create mode 100644 benchmarks/600.workflows/710.gpu-single-infer/definition.json create mode 100644 benchmarks/600.workflows/710.gpu-single-infer/input.py create mode 100644 benchmarks/600.workflows/710.gpu-single-infer/python/fetch_image.py create mode 100644 benchmarks/600.workflows/710.gpu-single-infer/python/infer.py create mode 100644 benchmarks/600.workflows/710.gpu-single-infer/python/requirements.txt create mode 100644 benchmarks/600.workflows/710.gpu-single-infer/python/summarize.py create mode 100644 benchmarks/600.workflows/711.gpu-embed-batch/config.json create mode 100644 benchmarks/600.workflows/711.gpu-embed-batch/definition.json create mode 100644 benchmarks/600.workflows/711.gpu-embed-batch/input.py create mode 100644 benchmarks/600.workflows/711.gpu-embed-batch/python/embed.py create mode 100644 benchmarks/600.workflows/711.gpu-embed-batch/python/fetch_batch.py create mode 100644 benchmarks/600.workflows/711.gpu-embed-batch/python/requirements.txt create mode 100644 benchmarks/600.workflows/711.gpu-embed-batch/python/summarize.py create mode 100644 benchmarks/600.workflows/717.gpu-traffic-count/config.json create mode 100644 benchmarks/600.workflows/717.gpu-traffic-count/definition.json create mode 100644 benchmarks/600.workflows/717.gpu-traffic-count/input.py create mode 100644 benchmarks/600.workflows/717.gpu-traffic-count/python/aggregate.py create mode 100644 benchmarks/600.workflows/717.gpu-traffic-count/python/detect.py create mode 100644 benchmarks/600.workflows/717.gpu-traffic-count/python/requirements.txt create mode 100644 benchmarks/600.workflows/717.gpu-traffic-count/python/split.py create mode 100644 benchmarks/600.workflows/718.gpu-document-qna/config.json create mode 100644 benchmarks/600.workflows/718.gpu-document-qna/definition.json create mode 100644 benchmarks/600.workflows/718.gpu-document-qna/input.py create mode 100644 benchmarks/600.workflows/718.gpu-document-qna/python/answer.py create mode 100644 benchmarks/600.workflows/718.gpu-document-qna/python/embed.py create mode 100644 benchmarks/600.workflows/718.gpu-document-qna/python/requirements.txt create mode 100644 benchmarks/600.workflows/718.gpu-document-qna/python/split.py create mode 100644 benchmarks/600.workflows/720.gpu-medical-scan/config.json create mode 100644 benchmarks/600.workflows/720.gpu-medical-scan/definition.json create mode 100644 benchmarks/600.workflows/720.gpu-medical-scan/input.py create mode 100644 benchmarks/600.workflows/720.gpu-medical-scan/python/aggregate.py create mode 100644 benchmarks/600.workflows/720.gpu-medical-scan/python/preprocess.py create mode 100644 benchmarks/600.workflows/720.gpu-medical-scan/python/segment.py create mode 100644 sebs/sonataflow/__init__.py create mode 100644 sebs/sonataflow/translator.py create mode 100644 sebs/sonataflow/workflow.py create mode 100644 tools/sonataflow_builder.py diff --git a/benchmarks/600.workflows/695.protein-screen/definition.json b/benchmarks/600.workflows/695.protein-screen/definition.json new file mode 100644 index 000000000..6c112465c --- /dev/null +++ b/benchmarks/600.workflows/695.protein-screen/definition.json @@ -0,0 +1,26 @@ +{ + "root": "generate", + "states": { + "generate": { + "type": "task", + "func_name": "generate", + "next": "evaluate-map" + }, + "evaluate-map": { + "type": "map", + "root": "evaluate", + "array": "schedules", + "next": "select", + "states": { + "evaluate": { + "type": "task", + "func_name": "evaluate" + } + } + }, + "select": { + "type": "task", + "func_name": "select" + } + } +} diff --git a/benchmarks/600.workflows/710.gpu-single-infer/config.json b/benchmarks/600.workflows/710.gpu-single-infer/config.json new file mode 100644 index 000000000..e57b5e8d1 --- /dev/null +++ b/benchmarks/600.workflows/710.gpu-single-infer/config.json @@ -0,0 +1,6 @@ +{ + "timeout": 300, + "memory": 1024, + "languages": ["python"], + "modules": [] +} diff --git a/benchmarks/600.workflows/710.gpu-single-infer/definition.json b/benchmarks/600.workflows/710.gpu-single-infer/definition.json new file mode 100644 index 000000000..aec4de72d --- /dev/null +++ b/benchmarks/600.workflows/710.gpu-single-infer/definition.json @@ -0,0 +1,19 @@ +{ + "root": "fetch", + "states": { + "fetch": { + "type": "task", + "func_name": "fetch_image", + "next": "infer" + }, + "infer": { + "type": "task", + "func_name": "infer", + "next": "summarize" + }, + "summarize": { + "type": "task", + "func_name": "summarize" + } + } +} diff --git a/benchmarks/600.workflows/710.gpu-single-infer/input.py b/benchmarks/600.workflows/710.gpu-single-infer/input.py new file mode 100644 index 000000000..811632497 --- /dev/null +++ b/benchmarks/600.workflows/710.gpu-single-infer/input.py @@ -0,0 +1,15 @@ +def buckets_count(): + return (0, 0) + + +def generate_input( + data_dir, + size, + benchmarks_bucket, + input_buckets, + output_buckets, + upload_func, + nosql_func, +): + # Minimal input: optionally pass image_url or image_path via event overrides. + return {} diff --git a/benchmarks/600.workflows/710.gpu-single-infer/python/fetch_image.py b/benchmarks/600.workflows/710.gpu-single-infer/python/fetch_image.py new file mode 100644 index 000000000..eacc25523 --- /dev/null +++ b/benchmarks/600.workflows/710.gpu-single-infer/python/fetch_image.py @@ -0,0 +1,13 @@ +import os +from uuid import uuid4 +from PIL import Image + + +def handler(event): + if "image_path" in event: + return {"image_path": event["image_path"]} + + img = Image.new("RGB", (320, 240), color=(64, 128, 192)) + tmp_path = os.path.join("/tmp", f"{uuid4().hex}.jpg") + img.save(tmp_path, format="JPEG") + return {"image_path": tmp_path} diff --git a/benchmarks/600.workflows/710.gpu-single-infer/python/infer.py b/benchmarks/600.workflows/710.gpu-single-infer/python/infer.py new file mode 100644 index 000000000..0976501d4 --- /dev/null +++ b/benchmarks/600.workflows/710.gpu-single-infer/python/infer.py @@ -0,0 +1,27 @@ +import torch +import torchvision +import torchvision.transforms as T +from PIL import Image + +_model = torchvision.models.resnet18(weights="DEFAULT").cuda().eval() +_transform = T.Compose( + [ + T.Resize(256), + T.CenterCrop(224), + T.ToTensor(), + T.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]), + ] +) + + +def handler(event): + path = event["image_path"] + img = Image.open(path).convert("RGB") + x = _transform(img).unsqueeze(0).cuda() + + with torch.inference_mode(): + logits = _model(x) + top_idx = int(torch.argmax(logits, dim=1).item()) + score = float(torch.nn.functional.softmax(logits, dim=1)[0, top_idx].item()) + + return {"top1_idx": top_idx, "top1_score": score, "image_path": path} diff --git a/benchmarks/600.workflows/710.gpu-single-infer/python/requirements.txt b/benchmarks/600.workflows/710.gpu-single-infer/python/requirements.txt new file mode 100644 index 000000000..4c4968110 --- /dev/null +++ b/benchmarks/600.workflows/710.gpu-single-infer/python/requirements.txt @@ -0,0 +1,3 @@ +torch +torchvision +pillow diff --git a/benchmarks/600.workflows/710.gpu-single-infer/python/summarize.py b/benchmarks/600.workflows/710.gpu-single-infer/python/summarize.py new file mode 100644 index 000000000..1b8a2c90e --- /dev/null +++ b/benchmarks/600.workflows/710.gpu-single-infer/python/summarize.py @@ -0,0 +1,6 @@ +def handler(event): + return { + "top1_idx": event["top1_idx"], + "top1_score": event["top1_score"], + "image_path": event["image_path"], + } diff --git a/benchmarks/600.workflows/711.gpu-embed-batch/config.json b/benchmarks/600.workflows/711.gpu-embed-batch/config.json new file mode 100644 index 000000000..e57b5e8d1 --- /dev/null +++ b/benchmarks/600.workflows/711.gpu-embed-batch/config.json @@ -0,0 +1,6 @@ +{ + "timeout": 300, + "memory": 1024, + "languages": ["python"], + "modules": [] +} diff --git a/benchmarks/600.workflows/711.gpu-embed-batch/definition.json b/benchmarks/600.workflows/711.gpu-embed-batch/definition.json new file mode 100644 index 000000000..6521a1fb3 --- /dev/null +++ b/benchmarks/600.workflows/711.gpu-embed-batch/definition.json @@ -0,0 +1,19 @@ +{ + "root": "fetch", + "states": { + "fetch": { + "type": "task", + "func_name": "fetch_batch", + "next": "embed" + }, + "embed": { + "type": "task", + "func_name": "embed", + "next": "summarize" + }, + "summarize": { + "type": "task", + "func_name": "summarize" + } + } +} diff --git a/benchmarks/600.workflows/711.gpu-embed-batch/input.py b/benchmarks/600.workflows/711.gpu-embed-batch/input.py new file mode 100644 index 000000000..ce07f5593 --- /dev/null +++ b/benchmarks/600.workflows/711.gpu-embed-batch/input.py @@ -0,0 +1,15 @@ +def buckets_count(): + return (0, 0) + + +def generate_input( + data_dir, + size, + benchmarks_bucket, + input_buckets, + output_buckets, + upload_func, + nosql_func, +): + # Provide texts via workflow invocation; nothing to pre-stage. + return {} diff --git a/benchmarks/600.workflows/711.gpu-embed-batch/python/embed.py b/benchmarks/600.workflows/711.gpu-embed-batch/python/embed.py new file mode 100644 index 000000000..0ac3e0df4 --- /dev/null +++ b/benchmarks/600.workflows/711.gpu-embed-batch/python/embed.py @@ -0,0 +1,23 @@ +import torch + +# Tiny embedding MLP to keep dependencies light. +_proj = torch.nn.Sequential( + torch.nn.Linear(32, 128), + torch.nn.ReLU(), + torch.nn.Linear(128, 64), +).cuda() + + +def _hash_text(text): + h = torch.zeros(32, device="cuda", dtype=torch.float32) + for i, ch in enumerate(text.encode("utf-8")): + h[i % 32] += float(ch) + return h + + +def handler(event): + texts = event["texts"] + batch = torch.stack([_hash_text(t) for t in texts]) + with torch.inference_mode(): + embs = _proj(batch).detach().cpu().numpy().tolist() + return {"embeddings": embs, "count": len(texts)} diff --git a/benchmarks/600.workflows/711.gpu-embed-batch/python/fetch_batch.py b/benchmarks/600.workflows/711.gpu-embed-batch/python/fetch_batch.py new file mode 100644 index 000000000..e3a895760 --- /dev/null +++ b/benchmarks/600.workflows/711.gpu-embed-batch/python/fetch_batch.py @@ -0,0 +1,11 @@ +def handler(event): + texts = event.get( + "texts", + [ + "serverless workflows on gpu", + "short test sentence", + "embedding benchmark sample", + "fan out and summarize", + ], + ) + return {"texts": texts} diff --git a/benchmarks/600.workflows/711.gpu-embed-batch/python/requirements.txt b/benchmarks/600.workflows/711.gpu-embed-batch/python/requirements.txt new file mode 100644 index 000000000..b9d9d66d2 --- /dev/null +++ b/benchmarks/600.workflows/711.gpu-embed-batch/python/requirements.txt @@ -0,0 +1,2 @@ +torch +pillow diff --git a/benchmarks/600.workflows/711.gpu-embed-batch/python/summarize.py b/benchmarks/600.workflows/711.gpu-embed-batch/python/summarize.py new file mode 100644 index 000000000..e7efaf8e5 --- /dev/null +++ b/benchmarks/600.workflows/711.gpu-embed-batch/python/summarize.py @@ -0,0 +1,7 @@ +import math + + +def handler(event): + embs = event["embeddings"] + norms = [math.sqrt(sum(x * x for x in v)) for v in embs] + return {"count": event["count"], "avg_norm": sum(norms) / len(norms)} diff --git a/benchmarks/600.workflows/717.gpu-traffic-count/config.json b/benchmarks/600.workflows/717.gpu-traffic-count/config.json new file mode 100644 index 000000000..e14b3b052 --- /dev/null +++ b/benchmarks/600.workflows/717.gpu-traffic-count/config.json @@ -0,0 +1,6 @@ +{ + "timeout": 540, + "memory": 2048, + "languages": ["python"], + "modules": ["storage"] +} diff --git a/benchmarks/600.workflows/717.gpu-traffic-count/definition.json b/benchmarks/600.workflows/717.gpu-traffic-count/definition.json new file mode 100644 index 000000000..99acefbfd --- /dev/null +++ b/benchmarks/600.workflows/717.gpu-traffic-count/definition.json @@ -0,0 +1,27 @@ +{ + "root": "split", + "states": { + "split": { + "type": "task", + "func_name": "split", + "next": "detect-map" + }, + "detect-map": { + "type": "map", + "root": "detect", + "array": "segments", + "next": "aggregate-loop", + "states": { + "detect": { + "type": "task", + "func_name": "detect" + } + } + }, + "aggregate-loop": { + "type": "loop", + "func_name": "aggregate", + "array": "segments" + } + } +} diff --git a/benchmarks/600.workflows/717.gpu-traffic-count/input.py b/benchmarks/600.workflows/717.gpu-traffic-count/input.py new file mode 100644 index 000000000..8ca65f8be --- /dev/null +++ b/benchmarks/600.workflows/717.gpu-traffic-count/input.py @@ -0,0 +1,63 @@ +# benchmarks/600.workflows/717.gpu-traffic-count/input.py +import os + +size_generators = { + "test": (6, 3), + "small": (40, 8), + "large": (160, 16), +} + + +def buckets_count(): + return (1, 1) + + +def generate_input( + data_dir, + size, + benchmarks_bucket, + input_buckets, + output_buckets, + upload_func, + nosql_func, +): + if data_dir is None: + raise ValueError( + "/path/to/traffic_frames/\n" + " └── frames/\n" + " ├── cam01_0001.jpg\n" + " ├── cam01_0002.jpg\n" + " └── ..." + ) + + num_frames, batch_size = size_generators[size] + + frames_dir = os.path.join(data_dir, "frames") + if not os.path.isdir(frames_dir): + raise ValueError(f"frames dir not exist: {frames_dir}") + + frame_files = sorted( + f for f in os.listdir(frames_dir) if f.lower().endswith((".png", ".jpg", ".jpeg")) + ) + if not frame_files: + raise ValueError(f"no jpg/png under dir: {frames_dir}") + + new_frames = [] + for i in range(num_frames): + frame = frame_files[i % len(frame_files)] + ext = os.path.splitext(frame)[1].lower() or ".jpg" + name = f"{i:08d}{ext}" + path = os.path.join(frames_dir, frame) + + new_frames.append(name) + upload_func(0, name, path) + + assert len(new_frames) == num_frames + + return { + "segments": new_frames, + "benchmark_bucket": benchmarks_bucket, + "input_bucket": input_buckets[0], + "output_bucket": output_buckets[0], + "batch_size": batch_size, + } diff --git a/benchmarks/600.workflows/717.gpu-traffic-count/python/aggregate.py b/benchmarks/600.workflows/717.gpu-traffic-count/python/aggregate.py new file mode 100644 index 000000000..6abf7f0c7 --- /dev/null +++ b/benchmarks/600.workflows/717.gpu-traffic-count/python/aggregate.py @@ -0,0 +1,65 @@ +import json +import os +import shutil +import uuid +from collections import Counter + +from . import storage + +client = storage.storage.get_instance() + + +def _download_det(frame: str, tmp_dir: str, benchmark_bucket: str, output_bucket: str, prefix: str): + base, _ = os.path.splitext(frame) + remote_name = f"{prefix}{base}_det.json" + local_json = os.path.join(tmp_dir, f"{base}_det.json") + try: + client.download(benchmark_bucket, f"{output_bucket}/{remote_name}", local_json) + return local_json + except Exception: + return None + + +def handler(event): + frames = event["segments"] + benchmark_bucket = event["benchmark_bucket"] + output_bucket = event["output_bucket"] + prefix = event["prefix"] + + tmp_dir = os.path.join("/tmp", str(uuid.uuid4())) + os.makedirs(tmp_dir, exist_ok=True) + + vehicle_totals = Counter() + missing = [] + try: + for frame in frames: + det_path = _download_det(frame, tmp_dir, benchmark_bucket, output_bucket, prefix) + if det_path and os.path.exists(det_path): + with open(det_path, "r", encoding="utf-8") as f: + data = json.load(f) + vehicle_totals.update(data.get("vehicles", {})) + else: + missing.append(frame) + + summary = { + "total_frames": len(frames), + "labeled_frames": len(frames) - len(missing), + "vehicle_totals": {k: int(v) for k, v in vehicle_totals.items()}, + "missing": missing, + } + + summary_path = os.path.join(tmp_dir, "traffic_summary.json") + with open(summary_path, "w", encoding="utf-8") as f: + json.dump(summary, f, indent=2) + + client.upload( + benchmark_bucket, + f"{output_bucket}/{prefix}traffic_summary.json", + summary_path, + unique_name=False, + ) + + finally: + shutil.rmtree(tmp_dir, ignore_errors=True) + + return {**event, "summary": summary} diff --git a/benchmarks/600.workflows/717.gpu-traffic-count/python/detect.py b/benchmarks/600.workflows/717.gpu-traffic-count/python/detect.py new file mode 100644 index 000000000..d3f7d21f5 --- /dev/null +++ b/benchmarks/600.workflows/717.gpu-traffic-count/python/detect.py @@ -0,0 +1,81 @@ +import json +import logging +import os +import shutil +import uuid +from collections import Counter + +import torch +from PIL import Image +from torchvision.models.detection import FasterRCNN_ResNet50_FPN_Weights, fasterrcnn_resnet50_fpn + +from . import storage + +logger = logging.getLogger(__name__) +client = storage.storage.get_instance() + +_weights = FasterRCNN_ResNet50_FPN_Weights.DEFAULT +_model = fasterrcnn_resnet50_fpn(weights=_weights).cuda().eval() +_transform = _weights.transforms() +_categories = _weights.meta["categories"] + +_vehicle_labels = {name for name in _categories if name in {"car", "truck", "bus", "motorcycle", "bicycle"}} + + +def handler(event): + input_bucket = event["input_bucket"] + output_bucket = event["output_bucket"] + benchmark_bucket = event["benchmark_bucket"] + frames = event["segments"] + prefix = event["prefix"] + + tmp_dir = os.path.join("/tmp", str(uuid.uuid4())) + os.makedirs(tmp_dir, exist_ok=True) + + results = [] + try: + for frame in frames: + local_img = os.path.join(tmp_dir, frame) + client.download(benchmark_bucket, f"{input_bucket}/{frame}", local_img) + + img = Image.open(local_img).convert("RGB") + x = _transform(img).to("cuda") + with torch.inference_mode(): + out = _model([x])[0] + + labels = [int(l) for l in out["labels"].tolist()] + scores = out["scores"].tolist() + counts = Counter() + detections = [] + for lbl, score in zip(labels, scores): + if score < 0.4: + continue + name = _categories[lbl] + if name not in _vehicle_labels: + continue + counts[name] += 1 + detections.append({"label": name, "score": float(score)}) + + result = { + "frame": frame, + "vehicles": {k: int(v) for k, v in counts.items()}, + "total": int(sum(counts.values())), + } + + local_json = os.path.join(tmp_dir, f"{os.path.splitext(frame)[0]}_det.json") + with open(local_json, "w", encoding="utf-8") as f: + json.dump(result, f) + + client.upload( + benchmark_bucket, + f"{output_bucket}/{prefix}{os.path.basename(local_json)}", + local_json, + unique_name=False, + ) + + results.append(result) + + finally: + shutil.rmtree(tmp_dir, ignore_errors=True) + + return {**event, "results": results} diff --git a/benchmarks/600.workflows/717.gpu-traffic-count/python/requirements.txt b/benchmarks/600.workflows/717.gpu-traffic-count/python/requirements.txt new file mode 100644 index 000000000..4c4968110 --- /dev/null +++ b/benchmarks/600.workflows/717.gpu-traffic-count/python/requirements.txt @@ -0,0 +1,3 @@ +torch +torchvision +pillow diff --git a/benchmarks/600.workflows/717.gpu-traffic-count/python/split.py b/benchmarks/600.workflows/717.gpu-traffic-count/python/split.py new file mode 100644 index 000000000..10378a090 --- /dev/null +++ b/benchmarks/600.workflows/717.gpu-traffic-count/python/split.py @@ -0,0 +1,26 @@ +import uuid + + +def _chunks(lst, n): + for i in range(0, len(lst), n): + yield lst[i : i + n] + + +def handler(event): + segs = _chunks(event["segments"], event["batch_size"]) + input_bucket = event["input_bucket"] + output_bucket = event["output_bucket"] + benchmark_bucket = event["benchmark_bucket"] + + return { + "segments": [ + { + "prefix": str(uuid.uuid4().int & ((1 << 64) - 1))[:8], + "segments": ss, + "input_bucket": input_bucket, + "output_bucket": output_bucket, + "benchmark_bucket": benchmark_bucket, + } + for ss in segs + ] + } diff --git a/benchmarks/600.workflows/718.gpu-document-qna/config.json b/benchmarks/600.workflows/718.gpu-document-qna/config.json new file mode 100644 index 000000000..e14b3b052 --- /dev/null +++ b/benchmarks/600.workflows/718.gpu-document-qna/config.json @@ -0,0 +1,6 @@ +{ + "timeout": 540, + "memory": 2048, + "languages": ["python"], + "modules": ["storage"] +} diff --git a/benchmarks/600.workflows/718.gpu-document-qna/definition.json b/benchmarks/600.workflows/718.gpu-document-qna/definition.json new file mode 100644 index 000000000..597148b37 --- /dev/null +++ b/benchmarks/600.workflows/718.gpu-document-qna/definition.json @@ -0,0 +1,26 @@ +{ + "root": "split", + "states": { + "split": { + "type": "task", + "func_name": "split", + "next": "embed-map" + }, + "embed-map": { + "type": "map", + "root": "embed", + "array": "segments", + "next": "answer", + "states": { + "embed": { + "type": "task", + "func_name": "embed" + } + } + }, + "answer": { + "type": "task", + "func_name": "answer" + } + } +} diff --git a/benchmarks/600.workflows/718.gpu-document-qna/input.py b/benchmarks/600.workflows/718.gpu-document-qna/input.py new file mode 100644 index 000000000..fca5bde4b --- /dev/null +++ b/benchmarks/600.workflows/718.gpu-document-qna/input.py @@ -0,0 +1,59 @@ +# benchmarks/600.workflows/718.gpu-document-qna/input.py +import os + +size_generators = { + "test": (4, 500, 2), + "small": (16, 1500, 3), + "large": (48, 3000, 4), +} + + +def buckets_count(): + return (1, 1) + + +def generate_input( + data_dir, + size, + benchmarks_bucket, + input_buckets, + output_buckets, + upload_func, + nosql_func, +): + if data_dir is None: + raise ValueError( + "/path/to/documents/\n" + " └── docs/\n" + " ├── doc1.txt\n" + " ├── doc2.txt\n" + " └── ..." + ) + + num_docs, max_tokens, top_k = size_generators[size] + + docs_dir = os.path.join(data_dir, "docs") + if not os.path.isdir(docs_dir): + raise ValueError(f"docs dir not exist: {docs_dir}") + + doc_files = sorted(f for f in os.listdir(docs_dir) if f.lower().endswith(".txt")) + if not doc_files: + raise ValueError(f"no txt files under dir: {docs_dir}") + + new_docs = [] + for i in range(num_docs): + doc = doc_files[i % len(doc_files)] + name = f"{i:08d}.txt" + path = os.path.join(docs_dir, doc) + new_docs.append(name) + upload_func(0, name, path) + + return { + "segments": new_docs, + "benchmark_bucket": benchmarks_bucket, + "input_bucket": input_buckets[0], + "output_bucket": output_buckets[0], + "max_tokens": max_tokens, + "top_k": top_k, + "question": "Summarize the key risks mentioned in the document.", + } diff --git a/benchmarks/600.workflows/718.gpu-document-qna/python/answer.py b/benchmarks/600.workflows/718.gpu-document-qna/python/answer.py new file mode 100644 index 000000000..e3610d670 --- /dev/null +++ b/benchmarks/600.workflows/718.gpu-document-qna/python/answer.py @@ -0,0 +1,67 @@ +import json +import math +import os +import shutil +import uuid +from typing import List + +import torch + +from . import storage + +client = storage.storage.get_instance() + + +def _cosine(a: torch.Tensor, b: torch.Tensor) -> float: + return float(torch.nn.functional.cosine_similarity(a, b, dim=0).item()) + + +def handler(event): + segments = event["segments"] + output_bucket = segments[0]["output_bucket"] + benchmark_bucket = segments[0]["benchmark_bucket"] + top_k = event.get("top_k", 3) + + tmp_dir = os.path.join("/tmp", str(uuid.uuid4())) + os.makedirs(tmp_dir, exist_ok=True) + + scores: List[dict] = [] + try: + for seg in segments: + doc = seg["doc"] + prefix = seg["prefix"] + json_path = os.path.join(tmp_dir, f"{os.path.splitext(doc)[0]}_emb.json") + client.download( + benchmark_bucket, + f"{output_bucket}/{prefix}{os.path.basename(json_path)}", + json_path, + ) + with open(json_path, "r", encoding="utf-8") as f: + emb_data = json.load(f) + doc_vec = torch.tensor(emb_data["embedding"]) + q_vec = torch.tensor(emb_data["question_embedding"]) + sim = _cosine(doc_vec, q_vec) + scores.append({"doc": doc, "score": sim, "question": emb_data["question"]}) + + top = sorted(scores, key=lambda x: x["score"], reverse=True)[: top_k or len(scores)] + answer = { + "question": top[0]["question"] if top else "", + "top_docs": top, + "total_docs": len(scores), + } + + answer_path = os.path.join(tmp_dir, "answer.json") + with open(answer_path, "w", encoding="utf-8") as f: + json.dump(answer, f, indent=2) + + client.upload( + benchmark_bucket, + f"{output_bucket}/answer.json", + answer_path, + unique_name=False, + ) + + finally: + shutil.rmtree(tmp_dir, ignore_errors=True) + + return answer diff --git a/benchmarks/600.workflows/718.gpu-document-qna/python/embed.py b/benchmarks/600.workflows/718.gpu-document-qna/python/embed.py new file mode 100644 index 000000000..05d9fea49 --- /dev/null +++ b/benchmarks/600.workflows/718.gpu-document-qna/python/embed.py @@ -0,0 +1,75 @@ +import json +import os +import shutil +import uuid + +import torch +from transformers import AutoModel, AutoTokenizer + +from . import storage + +client = storage.storage.get_instance() + +_MODEL_NAME = "intfloat/e5-small-v2" +_tokenizer = AutoTokenizer.from_pretrained(_MODEL_NAME) +_model = AutoModel.from_pretrained(_MODEL_NAME).cuda().eval() + + +def _encode(text: str) -> list: + tokens = _tokenizer( + text, + truncation=True, + max_length=512, + padding="max_length", + return_tensors="pt", + ) + for k in tokens: + tokens[k] = tokens[k].cuda() + with torch.inference_mode(): + outputs = _model(**tokens) + embeddings = outputs.last_hidden_state.mean(dim=1) + return embeddings.squeeze(0).cpu().tolist() + + +def handler(event): + input_bucket = event["input_bucket"] + output_bucket = event["output_bucket"] + benchmark_bucket = event["benchmark_bucket"] + doc = event["doc"] + prefix = event["prefix"] + question = event["question"] + + tmp_dir = os.path.join("/tmp", str(uuid.uuid4())) + os.makedirs(tmp_dir, exist_ok=True) + + try: + local_doc = os.path.join(tmp_dir, doc) + client.download(benchmark_bucket, f"{input_bucket}/{doc}", local_doc) + with open(local_doc, "r", encoding="utf-8") as f: + text = f.read() + + doc_vec = _encode(text) + q_vec = _encode(question) + + result = { + "doc": doc, + "embedding": doc_vec, + "question_embedding": q_vec, + "question": question, + } + + local_json = os.path.join(tmp_dir, f"{os.path.splitext(doc)[0]}_emb.json") + with open(local_json, "w", encoding="utf-8") as f: + json.dump(result, f) + + client.upload( + benchmark_bucket, + f"{output_bucket}/{prefix}{os.path.basename(local_json)}", + local_json, + unique_name=False, + ) + + finally: + shutil.rmtree(tmp_dir, ignore_errors=True) + + return {**event, "embedding": doc_vec, "question_embedding": q_vec} diff --git a/benchmarks/600.workflows/718.gpu-document-qna/python/requirements.txt b/benchmarks/600.workflows/718.gpu-document-qna/python/requirements.txt new file mode 100644 index 000000000..d1548e6cd --- /dev/null +++ b/benchmarks/600.workflows/718.gpu-document-qna/python/requirements.txt @@ -0,0 +1,4 @@ +torch +transformers +pillow +tokenizers diff --git a/benchmarks/600.workflows/718.gpu-document-qna/python/split.py b/benchmarks/600.workflows/718.gpu-document-qna/python/split.py new file mode 100644 index 000000000..de0f037d7 --- /dev/null +++ b/benchmarks/600.workflows/718.gpu-document-qna/python/split.py @@ -0,0 +1,25 @@ +import uuid + + +def handler(event): + docs = event["segments"] + input_bucket = event["input_bucket"] + output_bucket = event["output_bucket"] + benchmark_bucket = event["benchmark_bucket"] + max_tokens = event["max_tokens"] + question = event["question"] + + return { + "segments": [ + { + "prefix": str(uuid.uuid4().int & ((1 << 64) - 1))[:8], + "doc": doc, + "input_bucket": input_bucket, + "output_bucket": output_bucket, + "benchmark_bucket": benchmark_bucket, + "max_tokens": max_tokens, + "question": question, + } + for doc in docs + ] + } diff --git a/benchmarks/600.workflows/720.gpu-medical-scan/config.json b/benchmarks/600.workflows/720.gpu-medical-scan/config.json new file mode 100644 index 000000000..f83e0d66e --- /dev/null +++ b/benchmarks/600.workflows/720.gpu-medical-scan/config.json @@ -0,0 +1,6 @@ +{ + "timeout": 300, + "memory": 4096, + "languages": ["python"], + "modules": [] +} diff --git a/benchmarks/600.workflows/720.gpu-medical-scan/definition.json b/benchmarks/600.workflows/720.gpu-medical-scan/definition.json new file mode 100644 index 000000000..972c1dcaa --- /dev/null +++ b/benchmarks/600.workflows/720.gpu-medical-scan/definition.json @@ -0,0 +1,26 @@ +{ + "root": "preprocess", + "states": { + "preprocess": { + "type": "task", + "func_name": "preprocess", + "next": "segment-map" + }, + "segment-map": { + "type": "map", + "root": "segment", + "array": "slices", + "next": "aggregate", + "states": { + "segment": { + "type": "task", + "func_name": "segment" + } + } + }, + "aggregate": { + "type": "task", + "func_name": "aggregate" + } + } +} diff --git a/benchmarks/600.workflows/720.gpu-medical-scan/input.py b/benchmarks/600.workflows/720.gpu-medical-scan/input.py new file mode 100644 index 000000000..e529a0603 --- /dev/null +++ b/benchmarks/600.workflows/720.gpu-medical-scan/input.py @@ -0,0 +1,28 @@ +size_generators = { + "test": (10, 128, 128, 3), # 10 slices, 128x128 resolution, 3 organs + "small": (25, 256, 256, 5), + "large": (50, 512, 512, 8), +} + + +def buckets_count(): + # No object storage buckets required for this workflow. + return (0, 0) + + +def generate_input( + data_dir, + size, + benchmarks_bucket, + input_buckets, + output_buckets, + upload_func, + nosql_func, +): + n_slices, height, width, n_organs = size_generators[size] + return { + "n_slices": n_slices, + "height": height, + "width": width, + "n_organs": n_organs, + } diff --git a/benchmarks/600.workflows/720.gpu-medical-scan/python/aggregate.py b/benchmarks/600.workflows/720.gpu-medical-scan/python/aggregate.py new file mode 100644 index 000000000..45f8c52c8 --- /dev/null +++ b/benchmarks/600.workflows/720.gpu-medical-scan/python/aggregate.py @@ -0,0 +1,55 @@ +def handler(event): + slices = event.get("slices", []) + + if not slices: + return { + "scan_summary": "No slices to analyze", + "total_slices": 0, + "total_anomalies": 0, + } + + # Aggregate organ statistics across all slices + total_anomalies = sum(s.get("anomaly_count", 0) for s in slices) + + # Collect all anomalies + all_anomalies = [] + for slice_result in slices: + for anomaly in slice_result.get("anomalies", []): + all_anomalies.append({ + **anomaly, + "slice_id": slice_result["slice_id"], + "slice_idx": slice_result["slice_idx"], + }) + + # Sort anomalies by severity + all_anomalies.sort(key=lambda a: a["severity"], reverse=True) + + # Aggregate organ stats across slices + organ_aggregate = {} + for slice_result in slices: + organ_stats = slice_result.get("organ_stats", {}) + for organ_name, stats in organ_stats.items(): + if organ_name not in organ_aggregate: + organ_aggregate[organ_name] = { + "total_pixels": 0, + "avg_percentage": 0.0, + } + organ_aggregate[organ_name]["total_pixels"] += stats["pixel_count"] + organ_aggregate[organ_name]["avg_percentage"] += stats["percentage"] + + # Calculate averages + n_slices = len(slices) + for organ_name in organ_aggregate: + organ_aggregate[organ_name]["avg_percentage"] /= n_slices + + # Generate diagnostic report + risk_level = "HIGH" if total_anomalies > n_slices * 0.3 else "MEDIUM" if total_anomalies > 0 else "LOW" + + return { + "scan_summary": f"Analyzed {n_slices} slices", + "total_slices": n_slices, + "total_anomalies": total_anomalies, + "risk_level": risk_level, + "organ_summary": organ_aggregate, + "top_anomalies": all_anomalies[:5], # Top 5 most severe + } diff --git a/benchmarks/600.workflows/720.gpu-medical-scan/python/preprocess.py b/benchmarks/600.workflows/720.gpu-medical-scan/python/preprocess.py new file mode 100644 index 000000000..31e9e1f72 --- /dev/null +++ b/benchmarks/600.workflows/720.gpu-medical-scan/python/preprocess.py @@ -0,0 +1,62 @@ +import uuid +import numpy as np + + +def normalize_slice(data: np.ndarray) -> np.ndarray: + """Normalize CT/MRI slice to [0, 1] range.""" + min_val, max_val = data.min(), data.max() + if max_val - min_val > 0: + return (data - min_val) / (max_val - min_val) + return data + + +def generate_synthetic_scan(height: int, width: int, seed: int) -> np.ndarray: + """Generate synthetic medical scan slice with anatomical features.""" + rng = np.random.default_rng(seed=seed) + + # Create base tissue structure + scan = rng.normal(0.3, 0.1, (height, width)).astype(np.float32) + + # Add circular organ-like structures + y, x = np.ogrid[:height, :width] + center_y, center_x = height // 2, width // 2 + + # Main organ (e.g., liver, kidney) + radius = min(height, width) // 3 + mask = (x - center_x)**2 + (y - center_y)**2 <= radius**2 + scan[mask] += rng.normal(0.4, 0.05, mask.sum()).astype(np.float32) + + # Add some smaller structures (lesions, vessels) + for i in range(3): + offset_x = rng.integers(-radius//2, radius//2) + offset_y = rng.integers(-radius//2, radius//2) + small_radius = rng.integers(10, 30) + small_mask = (x - center_x - offset_x)**2 + (y - center_y - offset_y)**2 <= small_radius**2 + scan[small_mask] += rng.normal(0.2, 0.03, small_mask.sum()).astype(np.float32) + + return normalize_slice(scan) + + +def handler(event): + n_slices = int(event["n_slices"]) + height = int(event["height"]) + width = int(event["width"]) + n_organs = int(event["n_organs"]) + scan_id = event.get("scan_id", str(uuid.uuid4())[:8]) + + slices = [] + for idx in range(n_slices): + # Generate synthetic scan data + scan_data = generate_synthetic_scan(height, width, seed=idx) + + slices.append({ + "slice_id": f"slice-{scan_id}-{idx:03d}", + "slice_idx": idx, + "scan_data": scan_data.tolist(), # Convert to list for JSON serialization + "height": height, + "width": width, + "n_organs": n_organs, + "scan_id": scan_id, + }) + + return {"slices": slices} diff --git a/benchmarks/600.workflows/720.gpu-medical-scan/python/segment.py b/benchmarks/600.workflows/720.gpu-medical-scan/python/segment.py new file mode 100644 index 000000000..a1593b040 --- /dev/null +++ b/benchmarks/600.workflows/720.gpu-medical-scan/python/segment.py @@ -0,0 +1,89 @@ +import numpy as np + + +def apply_threshold_segmentation(scan: np.ndarray, n_organs: int) -> np.ndarray: + """ + Simple multi-threshold segmentation to identify different tissue types. + Simulates GPU-accelerated semantic segmentation (e.g., U-Net inference). + """ + segmentation = np.zeros_like(scan, dtype=np.int32) + + # Create threshold ranges for different organs/tissues + thresholds = np.linspace(0, 1, n_organs + 1) + + for organ_id in range(1, n_organs + 1): + lower = thresholds[organ_id - 1] + upper = thresholds[organ_id] + mask = (scan >= lower) & (scan < upper) + segmentation[mask] = organ_id + + return segmentation + + +def compute_organ_stats(segmentation: np.ndarray, n_organs: int) -> dict: + """Compute statistics for each segmented organ.""" + stats = {} + total_pixels = segmentation.size + + for organ_id in range(1, n_organs + 1): + mask = segmentation == organ_id + pixel_count = mask.sum() + + stats[f"organ_{organ_id}"] = { + "pixel_count": int(pixel_count), + "percentage": float(pixel_count / total_pixels * 100), + } + + return stats + + +def detect_anomalies(segmentation: np.ndarray, scan: np.ndarray) -> list: + """ + Detect potential anomalies (bright spots, lesions, etc.). + Simulates GPU-accelerated anomaly detection. + """ + anomalies = [] + + # Detect bright regions that could indicate lesions or tumors + threshold = 0.75 + anomaly_mask = scan > threshold + + if anomaly_mask.any(): + # Find connected components (simplified) + coords = np.argwhere(anomaly_mask) + if len(coords) > 0: + # Calculate centroid + centroid_y, centroid_x = coords.mean(axis=0) + anomalies.append({ + "type": "bright_region", + "location": [float(centroid_y), float(centroid_x)], + "size": int(len(coords)), + "severity": float(scan[anomaly_mask].mean()), + }) + + return anomalies + + +def handler(schedule): + slice_id = schedule["slice_id"] + scan_data = np.array(schedule["scan_data"], dtype=np.float32) + height = schedule["height"] + width = schedule["width"] + n_organs = schedule["n_organs"] + + # Perform segmentation (simulating GPU inference) + segmentation = apply_threshold_segmentation(scan_data, n_organs) + + # Compute organ statistics + organ_stats = compute_organ_stats(segmentation, n_organs) + + # Detect anomalies + anomalies = detect_anomalies(segmentation, scan_data) + + return { + "slice_id": slice_id, + "slice_idx": schedule["slice_idx"], + "organ_stats": organ_stats, + "anomalies": anomalies, + "anomaly_count": len(anomalies), + } diff --git a/benchmarks/600.workflows/723.gpu-content-moderation/input.py b/benchmarks/600.workflows/723.gpu-content-moderation/input.py index 79a50a761..c018849a0 100644 --- a/benchmarks/600.workflows/723.gpu-content-moderation/input.py +++ b/benchmarks/600.workflows/723.gpu-content-moderation/input.py @@ -1,5 +1,5 @@ size_generators = { - "test": (30, 100), # 30 posts, avg 100 tokens per post + "test": (30, 100), # 30 posts, avg 100 tokens per post "small": (100, 200), "large": (250, 300), } diff --git a/benchmarks/600.workflows/723.gpu-content-moderation/python/ingest.py b/benchmarks/600.workflows/723.gpu-content-moderation/python/ingest.py index 8dc8cb272..0d15910ae 100644 --- a/benchmarks/600.workflows/723.gpu-content-moderation/python/ingest.py +++ b/benchmarks/600.workflows/723.gpu-content-moderation/python/ingest.py @@ -2,23 +2,80 @@ import numpy as np -POST_CATEGORIES = ["news", "personal", "commercial", "entertainment", "educational", "political", "social"] +POST_CATEGORIES = [ + "news", + "personal", + "commercial", + "entertainment", + "educational", + "political", + "social", +] LANGUAGES = ["en", "es", "fr", "de", "pt", "it"] SAFE_WORDS = [ - "cat", "dog", "food", "travel", "music", "art", "game", "tech", "book", "movie", - "friend", "family", "work", "love", "happy", "great", "amazing", "beautiful", "fun", "enjoy", - "today", "yesterday", "tomorrow", "share", "post", "comment", "like", "follow", "update" + "cat", + "dog", + "food", + "travel", + "music", + "art", + "game", + "tech", + "book", + "movie", + "friend", + "family", + "work", + "love", + "happy", + "great", + "amazing", + "beautiful", + "fun", + "enjoy", + "today", + "yesterday", + "tomorrow", + "share", + "post", + "comment", + "like", + "follow", + "update", ] BORDERLINE_WORDS = [ - "debate", "opinion", "protest", "controversial", "argument", "disagree", "conflict", - "political", "election", "policy", "criticism", "oppose", "challenge" + "debate", + "opinion", + "protest", + "controversial", + "argument", + "disagree", + "conflict", + "political", + "election", + "policy", + "criticism", + "oppose", + "challenge", ] UNSAFE_WORDS = [ - "hate", "violence", "spam", "scam", "fake", "attack", "threat", "abuse", "harass", - "derogatory", "discriminate", "offensive", "vulgar", "explicit" + "hate", + "violence", + "spam", + "scam", + "fake", + "attack", + "threat", + "abuse", + "harass", + "derogatory", + "discriminate", + "offensive", + "vulgar", + "explicit", ] @@ -32,25 +89,23 @@ def generate_text_content(avg_tokens: int, seed: int, violation_prob: float = 0. if has_violation: violation_type = rng.choice( ["hate_speech", "violence", "spam", "misinformation", "harassment"], - p=[0.25, 0.2, 0.3, 0.15, 0.1] + p=[0.25, 0.2, 0.3, 0.15, 0.1], ) unsafe_ratio = rng.uniform(0.15, 0.4) n_unsafe = int(n_tokens * unsafe_ratio) n_safe = n_tokens - n_unsafe - words = ( - list(rng.choice(UNSAFE_WORDS, size=n_unsafe, replace=True)) + - list(rng.choice(SAFE_WORDS, size=n_safe, replace=True)) + words = list(rng.choice(UNSAFE_WORDS, size=n_unsafe, replace=True)) + list( + rng.choice(SAFE_WORDS, size=n_safe, replace=True) ) else: borderline_ratio = rng.uniform(0, 0.2) n_borderline = int(n_tokens * borderline_ratio) n_safe = n_tokens - n_borderline - words = ( - list(rng.choice(BORDERLINE_WORDS, size=n_borderline, replace=True)) + - list(rng.choice(SAFE_WORDS, size=n_safe, replace=True)) + words = list(rng.choice(BORDERLINE_WORDS, size=n_borderline, replace=True)) + list( + rng.choice(SAFE_WORDS, size=n_safe, replace=True) ) violation_type = None diff --git a/benchmarks/600.workflows/723.gpu-content-moderation/python/moderate.py b/benchmarks/600.workflows/723.gpu-content-moderation/python/moderate.py index 04c85af93..c00815d9c 100644 --- a/benchmarks/600.workflows/723.gpu-content-moderation/python/moderate.py +++ b/benchmarks/600.workflows/723.gpu-content-moderation/python/moderate.py @@ -1,7 +1,17 @@ import numpy as np -UNSAFE_KEYWORDS = ["hate", "violence", "spam", "scam", "fake", "attack", "threat", "abuse", "harass"] +UNSAFE_KEYWORDS = [ + "hate", + "violence", + "spam", + "scam", + "fake", + "attack", + "threat", + "abuse", + "harass", +] BORDERLINE_KEYWORDS = ["controversial", "debate", "protest", "argument"] @@ -51,38 +61,48 @@ def classify_violation_type(text: str, toxicity_score: float, seed: int) -> list if "hate" in words or "discriminate" in words or "derogatory" in words: confidence = min(0.95, toxicity_score + rng.uniform(0.1, 0.2)) - violations.append({ - "type": "hate_speech", - "confidence": float(confidence), - }) + violations.append( + { + "type": "hate_speech", + "confidence": float(confidence), + } + ) if "violence" in words or "attack" in words or "threat" in words: confidence = min(0.95, toxicity_score + rng.uniform(0.1, 0.2)) - violations.append({ - "type": "violence", - "confidence": float(confidence), - }) + violations.append( + { + "type": "violence", + "confidence": float(confidence), + } + ) if "spam" in words or "scam" in words: confidence = min(0.9, toxicity_score + rng.uniform(0.05, 0.15)) - violations.append({ - "type": "spam", - "confidence": float(confidence), - }) + violations.append( + { + "type": "spam", + "confidence": float(confidence), + } + ) if "fake" in words: confidence = min(0.85, toxicity_score + rng.uniform(0.05, 0.15)) - violations.append({ - "type": "misinformation", - "confidence": float(confidence), - }) + violations.append( + { + "type": "misinformation", + "confidence": float(confidence), + } + ) if "harass" in words or "abuse" in words: confidence = min(0.9, toxicity_score + rng.uniform(0.1, 0.2)) - violations.append({ - "type": "harassment", - "confidence": float(confidence), - }) + violations.append( + { + "type": "harassment", + "confidence": float(confidence), + } + ) return violations @@ -109,7 +129,11 @@ def sentiment_analysis(text: str, seed: int) -> dict: return { "sentiment_score": float(sentiment_score), - "polarity": "positive" if sentiment_score > 0.1 else "negative" if sentiment_score < -0.1 else "neutral", + "polarity": "positive" + if sentiment_score > 0.1 + else "negative" + if sentiment_score < -0.1 + else "neutral", } @@ -152,9 +176,9 @@ def handler(post): spam_score = calculate_spam_score(post) overall_score = ( - 0.5 * toxicity_score + - 0.3 * spam_score + - 0.2 * (1.0 if sentiment["sentiment_score"] < -0.3 else 0.0) + 0.5 * toxicity_score + + 0.3 * spam_score + + 0.2 * (1.0 if sentiment["sentiment_score"] < -0.3 else 0.0) ) if overall_score > 0.75: diff --git a/benchmarks/600.workflows/723.gpu-content-moderation/python/review.py b/benchmarks/600.workflows/723.gpu-content-moderation/python/review.py index 3f3bcf4ce..113f0f3b4 100644 --- a/benchmarks/600.workflows/723.gpu-content-moderation/python/review.py +++ b/benchmarks/600.workflows/723.gpu-content-moderation/python/review.py @@ -3,20 +3,29 @@ def calculate_accuracy(posts: list) -> dict: detected_violations = [p for p in posts if p.get("action") in ["REMOVE", "REVIEW"]] - true_positives = len([ - p for p in posts - if p.get("true_violation") is not None and p.get("action") in ["REMOVE", "REVIEW"] - ]) - - false_positives = len([ - p for p in posts - if p.get("true_violation") is None and p.get("action") in ["REMOVE", "REVIEW"] - ]) - - false_negatives = len([ - p for p in posts - if p.get("true_violation") is not None and p.get("action") not in ["REMOVE", "REVIEW"] - ]) + true_positives = len( + [ + p + for p in posts + if p.get("true_violation") is not None and p.get("action") in ["REMOVE", "REVIEW"] + ] + ) + + false_positives = len( + [ + p + for p in posts + if p.get("true_violation") is None and p.get("action") in ["REMOVE", "REVIEW"] + ] + ) + + false_negatives = len( + [ + p + for p in posts + if p.get("true_violation") is not None and p.get("action") not in ["REMOVE", "REVIEW"] + ] + ) precision = true_positives / len(detected_violations) if detected_violations else 0.0 recall = true_positives / len(true_violations) if true_violations else 0.0 diff --git a/sebs/sonataflow/__init__.py b/sebs/sonataflow/__init__.py new file mode 100644 index 000000000..44a4707a8 --- /dev/null +++ b/sebs/sonataflow/__init__.py @@ -0,0 +1,2 @@ +from .workflow import SonataFlowWorkflow # noqa: F401 +from .translator import definition_to_sonataflow, sonataflow_to_yaml # noqa: F401 diff --git a/sebs/sonataflow/translator.py b/sebs/sonataflow/translator.py new file mode 100644 index 000000000..35def83a1 --- /dev/null +++ b/sebs/sonataflow/translator.py @@ -0,0 +1,131 @@ +import json +from typing import Dict, List, Set, Tuple + +try: + import yaml +except ImportError: + yaml = None + + +class UnsupportedWorkflowError(Exception): + pass + + +def _collect_functions(funcs: Set[str], func_name: str): + if func_name: + funcs.add(func_name) + + +def _task_state(name: str, state: Dict, funcs: Set[str]) -> Dict: + func_name = state["func_name"] + _collect_functions(funcs, func_name) + node = { + "name": name, + "type": "operation", + "actions": [{"name": func_name, "functionRef": func_name}], + } + if "next" in state: + node["transition"] = state["next"] + else: + node["end"] = True + return node + + +def _map_state(name: str, state: Dict, funcs: Set[str]) -> Dict: + nested_states: Dict = state["states"] + root = state["root"] + if root not in nested_states: + raise UnsupportedWorkflowError(f"Map state {name} missing root state {root}") + root_state = nested_states[root] + func_name = root_state["func_name"] + _collect_functions(funcs, func_name) + node = { + "name": name, + "type": "foreach", + "inputCollection": f"$.{state['array']}", + "iterationParam": "item", + "actions": [{"name": func_name, "functionRef": func_name}], + } + if "next" in state: + node["transition"] = state["next"] + else: + node["end"] = True + return node + + +def _loop_state(name: str, state: Dict, funcs: Set[str]) -> Dict: + func_name = state["func_name"] + _collect_functions(funcs, func_name) + node = { + "name": name, + "type": "operation", + "actions": [{"name": func_name, "functionRef": func_name}], + } + if "next" in state: + node["transition"] = state["next"] + else: + node["end"] = True + return node + + +def _linearize(definition: Dict) -> List[Tuple[str, Dict]]: + states = definition["states"] + cursor = definition["root"] + ordered: List[Tuple[str, Dict]] = [] + visited = set() + while cursor: + if cursor in visited: + raise UnsupportedWorkflowError(f"Cycle detected at state {cursor}") + visited.add(cursor) + state = states[cursor] + ordered.append((cursor, state)) + cursor = state.get("next") + return ordered + + +def definition_to_sonataflow(definition: Dict, name: str = "workflow") -> Tuple[Dict, Set[str]]: + """ + Convert SEBS workflow definition.json into a minimal SonataFlow (CNCF Serverless Workflow) + spec dictionary. Supports task/map/loop constructs used by the SEBS workflows. + Returns (spec, functions_used). + """ + if "root" not in definition or "states" not in definition: + raise UnsupportedWorkflowError("Invalid workflow definition") + + ordered_states = _linearize(definition) + funcs: Set[str] = set() + sonata_states: List[Dict] = [] + + builders = {"task": _task_state, "map": _map_state, "loop": _loop_state} + + for state_name, state in ordered_states: + stype = state["type"] + if stype not in builders: + raise UnsupportedWorkflowError(f"Unsupported state type {stype}") + sonata_states.append(builders[stype](state_name, state, funcs)) + + functions_block = [ + {"name": fn, "type": "custom", "operation": f"fn://{fn}"} for fn in sorted(funcs) + ] + spec = { + "id": name, + "name": name, + "version": "0.1", + "specVersion": "0.8", + "functions": functions_block, + "states": sonata_states, + } + return spec, funcs + + +def sonataflow_to_yaml(spec: Dict) -> str: + if yaml is None: + raise ImportError("PyYAML is required for YAML export") + return yaml.safe_dump(spec, sort_keys=False) + + +def definition_to_sonataflow_yaml(definition: Dict, name: str = "workflow") -> str: + spec, _ = definition_to_sonataflow(definition, name=name) + if yaml: + return sonataflow_to_yaml(spec) + return json.dumps(spec, indent=2) diff --git a/sebs/sonataflow/workflow.py b/sebs/sonataflow/workflow.py new file mode 100644 index 000000000..65d7fb916 --- /dev/null +++ b/sebs/sonataflow/workflow.py @@ -0,0 +1,47 @@ +from typing import Dict, List, Optional, Set + +from sebs.faas.function import FunctionConfig, Workflow + + +class SonataFlowWorkflow(Workflow): + """ + Lightweight wrapper that keeps track of the generated SonataFlow (CNCF Serverless Workflow) + spec for a SEBS benchmark workflow. Mirrors the structure of other providers' Workflow + classes so it can be cached/serialized the same way. + """ + + def __init__( + self, + name: str, + benchmark: str, + code_package_hash: str, + cfg: FunctionConfig, + sonataflow_spec: Dict, + functions: Optional[Set[str]] = None, + ): + super().__init__(benchmark, name, code_package_hash, cfg) + self.sonataflow_spec = sonataflow_spec + self.functions = list(functions or []) + + @staticmethod + def typename() -> str: + return "SonataFlow.Workflow" + + def serialize(self) -> dict: + return { + **super().serialize(), + "sonataflow_spec": self.sonataflow_spec, + "functions": self.functions, + } + + @staticmethod + def deserialize(cached_config: dict) -> "SonataFlowWorkflow": + cfg = FunctionConfig.deserialize(cached_config["config"]) + return SonataFlowWorkflow( + name=cached_config["name"], + benchmark=cached_config["benchmark"], + code_package_hash=cached_config["hash"], + cfg=cfg, + sonataflow_spec=cached_config["sonataflow_spec"], + functions=set(cached_config.get("functions", [])), + ) diff --git a/tools/sonataflow_builder.py b/tools/sonataflow_builder.py new file mode 100644 index 000000000..d33fa18e3 --- /dev/null +++ b/tools/sonataflow_builder.py @@ -0,0 +1,111 @@ +#!/usr/bin/env python3 +""" +Generate SonataFlow (CNCF Serverless Workflow) YAML along with Knative Service +manifests for SEBS workflows. This is a thin helper so users can go from +`definition.json` + function image mapping to runnable manifests under a +Quarkus SonataFlow operator on Knative. + +Usage: + python tools/sonataflow_builder.py \ + --definition benchmarks/600.workflows/6xx.OCR-pipeline/definition.json \ + --name ocr-pipeline \ + --namespace sebs \ + --image split=ghcr.io/example/split:latest \ + --image detect=ghcr.io/example/detect:latest \ + --image recognize=ghcr.io/example/recognize:latest \ + --image merge=ghcr.io/example/merge:latest \ + --out manifests.yaml +""" +import argparse +import json +import sys +from pathlib import Path +from typing import Dict, List + +from sebs.sonataflow.translator import definition_to_sonataflow_yaml, definition_to_sonataflow + + +KNATIVE_TEMPLATE = """\ +apiVersion: serving.knative.dev/v1 +kind: Service +metadata: + name: {name} + namespace: {namespace} +spec: + template: + spec: + containers: + - image: {image} + env: + - name: FUNCTION_NAME + value: {name} +""" + +SONATAFLOW_CR_TEMPLATE = """\ +apiVersion: sonataflow.org/v1alpha08 +kind: SonataFlow +metadata: + name: {name} + namespace: {namespace} +spec: + flow: | +{workflow_yaml} +""" + + +def parse_images(image_args: List[str]) -> Dict[str, str]: + mapping: Dict[str, str] = {} + for item in image_args: + if "=" not in item: + raise ValueError(f"Image argument must be key=value, got: {item}") + fn, img = item.split("=", 1) + mapping[fn.strip()] = img.strip() + return mapping + + +def build_knative_services(namespace: str, images: Dict[str, str], functions: List[str]) -> List[str]: + yamls = [] + for fn in functions: + image = images.get(fn) + if not image: + raise ValueError(f"Missing image mapping for function: {fn}") + yamls.append(KNATIVE_TEMPLATE.format(name=fn, namespace=namespace, image=image)) + return yamls + + +def main(): + parser = argparse.ArgumentParser(description="Generate SonataFlow + Knative YAML for SEBS workflow.") + parser.add_argument("--definition", required=True, help="Path to SEBS workflow definition.json") + parser.add_argument("--name", required=True, help="Workflow name/id") + parser.add_argument("--namespace", default="default", help="Kubernetes namespace for resources") + parser.add_argument( + "--image", + action="append", + default=[], + help="Mapping of function=image (repeat for each function)", + ) + parser.add_argument("--out", default="-", help="Output file path or '-' for stdout") + + args = parser.parse_args() + + definition = json.loads(Path(args.definition).read_text()) + spec, functions = definition_to_sonataflow(definition, name=args.name) + images = parse_images(args.image) + + workflow_yaml = definition_to_sonataflow_yaml(definition, name=args.name) + knative_yamls = build_knative_services(args.namespace, images, sorted(functions)) + + pieces = [] + pieces.extend(knative_yamls) + pieces.append(SONATAFLOW_CR_TEMPLATE.format(name=args.name, namespace=args.namespace, workflow_yaml="\n".join(" " + line for line in workflow_yaml.splitlines()))) + output = "\n---\n".join(pieces) + + if args.out == "-" or args.out == "/dev/stdout": + sys.stdout.write(output) + else: + Path(args.out).write_text(output) + print(f"Wrote manifests to {args.out}") + + +if __name__ == "__main__": + main() From d07abbf27339c10d73d9d906a0732ebc6770f7ff Mon Sep 17 00:00:00 2001 From: xipang Date: Wed, 10 Dec 2025 23:49:03 +0100 Subject: [PATCH 3/4] push workflow --- .../710.gpu-single-infer/config.json | 6 -- .../710.gpu-single-infer/definition.json | 19 ----- .../710.gpu-single-infer/input.py | 15 ---- .../python/fetch_image.py | 13 --- .../710.gpu-single-infer/python/infer.py | 27 ------- .../python/requirements.txt | 3 - .../710.gpu-single-infer/python/summarize.py | 6 -- .../711.gpu-embed-batch/config.json | 6 -- .../711.gpu-embed-batch/definition.json | 19 ----- .../711.gpu-embed-batch/input.py | 15 ---- .../711.gpu-embed-batch/python/embed.py | 23 ------ .../711.gpu-embed-batch/python/fetch_batch.py | 11 --- .../python/requirements.txt | 2 - .../711.gpu-embed-batch/python/summarize.py | 7 -- .../717.gpu-traffic-count/config.json | 6 -- .../717.gpu-traffic-count/definition.json | 27 ------- .../717.gpu-traffic-count/input.py | 63 --------------- .../717.gpu-traffic-count/python/aggregate.py | 65 --------------- .../717.gpu-traffic-count/python/detect.py | 81 ------------------- .../python/requirements.txt | 3 - .../717.gpu-traffic-count/python/split.py | 26 ------ .../718.gpu-document-qna/config.json | 6 -- .../718.gpu-document-qna/definition.json | 26 ------ .../718.gpu-document-qna/input.py | 59 -------------- .../718.gpu-document-qna/python/answer.py | 67 --------------- .../718.gpu-document-qna/python/embed.py | 75 ----------------- .../python/requirements.txt | 4 - .../718.gpu-document-qna/python/split.py | 25 ------ 28 files changed, 705 deletions(-) delete mode 100644 benchmarks/600.workflows/710.gpu-single-infer/config.json delete mode 100644 benchmarks/600.workflows/710.gpu-single-infer/definition.json delete mode 100644 benchmarks/600.workflows/710.gpu-single-infer/input.py delete mode 100644 benchmarks/600.workflows/710.gpu-single-infer/python/fetch_image.py delete mode 100644 benchmarks/600.workflows/710.gpu-single-infer/python/infer.py delete mode 100644 benchmarks/600.workflows/710.gpu-single-infer/python/requirements.txt delete mode 100644 benchmarks/600.workflows/710.gpu-single-infer/python/summarize.py delete mode 100644 benchmarks/600.workflows/711.gpu-embed-batch/config.json delete mode 100644 benchmarks/600.workflows/711.gpu-embed-batch/definition.json delete mode 100644 benchmarks/600.workflows/711.gpu-embed-batch/input.py delete mode 100644 benchmarks/600.workflows/711.gpu-embed-batch/python/embed.py delete mode 100644 benchmarks/600.workflows/711.gpu-embed-batch/python/fetch_batch.py delete mode 100644 benchmarks/600.workflows/711.gpu-embed-batch/python/requirements.txt delete mode 100644 benchmarks/600.workflows/711.gpu-embed-batch/python/summarize.py delete mode 100644 benchmarks/600.workflows/717.gpu-traffic-count/config.json delete mode 100644 benchmarks/600.workflows/717.gpu-traffic-count/definition.json delete mode 100644 benchmarks/600.workflows/717.gpu-traffic-count/input.py delete mode 100644 benchmarks/600.workflows/717.gpu-traffic-count/python/aggregate.py delete mode 100644 benchmarks/600.workflows/717.gpu-traffic-count/python/detect.py delete mode 100644 benchmarks/600.workflows/717.gpu-traffic-count/python/requirements.txt delete mode 100644 benchmarks/600.workflows/717.gpu-traffic-count/python/split.py delete mode 100644 benchmarks/600.workflows/718.gpu-document-qna/config.json delete mode 100644 benchmarks/600.workflows/718.gpu-document-qna/definition.json delete mode 100644 benchmarks/600.workflows/718.gpu-document-qna/input.py delete mode 100644 benchmarks/600.workflows/718.gpu-document-qna/python/answer.py delete mode 100644 benchmarks/600.workflows/718.gpu-document-qna/python/embed.py delete mode 100644 benchmarks/600.workflows/718.gpu-document-qna/python/requirements.txt delete mode 100644 benchmarks/600.workflows/718.gpu-document-qna/python/split.py diff --git a/benchmarks/600.workflows/710.gpu-single-infer/config.json b/benchmarks/600.workflows/710.gpu-single-infer/config.json deleted file mode 100644 index e57b5e8d1..000000000 --- a/benchmarks/600.workflows/710.gpu-single-infer/config.json +++ /dev/null @@ -1,6 +0,0 @@ -{ - "timeout": 300, - "memory": 1024, - "languages": ["python"], - "modules": [] -} diff --git a/benchmarks/600.workflows/710.gpu-single-infer/definition.json b/benchmarks/600.workflows/710.gpu-single-infer/definition.json deleted file mode 100644 index aec4de72d..000000000 --- a/benchmarks/600.workflows/710.gpu-single-infer/definition.json +++ /dev/null @@ -1,19 +0,0 @@ -{ - "root": "fetch", - "states": { - "fetch": { - "type": "task", - "func_name": "fetch_image", - "next": "infer" - }, - "infer": { - "type": "task", - "func_name": "infer", - "next": "summarize" - }, - "summarize": { - "type": "task", - "func_name": "summarize" - } - } -} diff --git a/benchmarks/600.workflows/710.gpu-single-infer/input.py b/benchmarks/600.workflows/710.gpu-single-infer/input.py deleted file mode 100644 index 811632497..000000000 --- a/benchmarks/600.workflows/710.gpu-single-infer/input.py +++ /dev/null @@ -1,15 +0,0 @@ -def buckets_count(): - return (0, 0) - - -def generate_input( - data_dir, - size, - benchmarks_bucket, - input_buckets, - output_buckets, - upload_func, - nosql_func, -): - # Minimal input: optionally pass image_url or image_path via event overrides. - return {} diff --git a/benchmarks/600.workflows/710.gpu-single-infer/python/fetch_image.py b/benchmarks/600.workflows/710.gpu-single-infer/python/fetch_image.py deleted file mode 100644 index eacc25523..000000000 --- a/benchmarks/600.workflows/710.gpu-single-infer/python/fetch_image.py +++ /dev/null @@ -1,13 +0,0 @@ -import os -from uuid import uuid4 -from PIL import Image - - -def handler(event): - if "image_path" in event: - return {"image_path": event["image_path"]} - - img = Image.new("RGB", (320, 240), color=(64, 128, 192)) - tmp_path = os.path.join("/tmp", f"{uuid4().hex}.jpg") - img.save(tmp_path, format="JPEG") - return {"image_path": tmp_path} diff --git a/benchmarks/600.workflows/710.gpu-single-infer/python/infer.py b/benchmarks/600.workflows/710.gpu-single-infer/python/infer.py deleted file mode 100644 index 0976501d4..000000000 --- a/benchmarks/600.workflows/710.gpu-single-infer/python/infer.py +++ /dev/null @@ -1,27 +0,0 @@ -import torch -import torchvision -import torchvision.transforms as T -from PIL import Image - -_model = torchvision.models.resnet18(weights="DEFAULT").cuda().eval() -_transform = T.Compose( - [ - T.Resize(256), - T.CenterCrop(224), - T.ToTensor(), - T.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]), - ] -) - - -def handler(event): - path = event["image_path"] - img = Image.open(path).convert("RGB") - x = _transform(img).unsqueeze(0).cuda() - - with torch.inference_mode(): - logits = _model(x) - top_idx = int(torch.argmax(logits, dim=1).item()) - score = float(torch.nn.functional.softmax(logits, dim=1)[0, top_idx].item()) - - return {"top1_idx": top_idx, "top1_score": score, "image_path": path} diff --git a/benchmarks/600.workflows/710.gpu-single-infer/python/requirements.txt b/benchmarks/600.workflows/710.gpu-single-infer/python/requirements.txt deleted file mode 100644 index 4c4968110..000000000 --- a/benchmarks/600.workflows/710.gpu-single-infer/python/requirements.txt +++ /dev/null @@ -1,3 +0,0 @@ -torch -torchvision -pillow diff --git a/benchmarks/600.workflows/710.gpu-single-infer/python/summarize.py b/benchmarks/600.workflows/710.gpu-single-infer/python/summarize.py deleted file mode 100644 index 1b8a2c90e..000000000 --- a/benchmarks/600.workflows/710.gpu-single-infer/python/summarize.py +++ /dev/null @@ -1,6 +0,0 @@ -def handler(event): - return { - "top1_idx": event["top1_idx"], - "top1_score": event["top1_score"], - "image_path": event["image_path"], - } diff --git a/benchmarks/600.workflows/711.gpu-embed-batch/config.json b/benchmarks/600.workflows/711.gpu-embed-batch/config.json deleted file mode 100644 index e57b5e8d1..000000000 --- a/benchmarks/600.workflows/711.gpu-embed-batch/config.json +++ /dev/null @@ -1,6 +0,0 @@ -{ - "timeout": 300, - "memory": 1024, - "languages": ["python"], - "modules": [] -} diff --git a/benchmarks/600.workflows/711.gpu-embed-batch/definition.json b/benchmarks/600.workflows/711.gpu-embed-batch/definition.json deleted file mode 100644 index 6521a1fb3..000000000 --- a/benchmarks/600.workflows/711.gpu-embed-batch/definition.json +++ /dev/null @@ -1,19 +0,0 @@ -{ - "root": "fetch", - "states": { - "fetch": { - "type": "task", - "func_name": "fetch_batch", - "next": "embed" - }, - "embed": { - "type": "task", - "func_name": "embed", - "next": "summarize" - }, - "summarize": { - "type": "task", - "func_name": "summarize" - } - } -} diff --git a/benchmarks/600.workflows/711.gpu-embed-batch/input.py b/benchmarks/600.workflows/711.gpu-embed-batch/input.py deleted file mode 100644 index ce07f5593..000000000 --- a/benchmarks/600.workflows/711.gpu-embed-batch/input.py +++ /dev/null @@ -1,15 +0,0 @@ -def buckets_count(): - return (0, 0) - - -def generate_input( - data_dir, - size, - benchmarks_bucket, - input_buckets, - output_buckets, - upload_func, - nosql_func, -): - # Provide texts via workflow invocation; nothing to pre-stage. - return {} diff --git a/benchmarks/600.workflows/711.gpu-embed-batch/python/embed.py b/benchmarks/600.workflows/711.gpu-embed-batch/python/embed.py deleted file mode 100644 index 0ac3e0df4..000000000 --- a/benchmarks/600.workflows/711.gpu-embed-batch/python/embed.py +++ /dev/null @@ -1,23 +0,0 @@ -import torch - -# Tiny embedding MLP to keep dependencies light. -_proj = torch.nn.Sequential( - torch.nn.Linear(32, 128), - torch.nn.ReLU(), - torch.nn.Linear(128, 64), -).cuda() - - -def _hash_text(text): - h = torch.zeros(32, device="cuda", dtype=torch.float32) - for i, ch in enumerate(text.encode("utf-8")): - h[i % 32] += float(ch) - return h - - -def handler(event): - texts = event["texts"] - batch = torch.stack([_hash_text(t) for t in texts]) - with torch.inference_mode(): - embs = _proj(batch).detach().cpu().numpy().tolist() - return {"embeddings": embs, "count": len(texts)} diff --git a/benchmarks/600.workflows/711.gpu-embed-batch/python/fetch_batch.py b/benchmarks/600.workflows/711.gpu-embed-batch/python/fetch_batch.py deleted file mode 100644 index e3a895760..000000000 --- a/benchmarks/600.workflows/711.gpu-embed-batch/python/fetch_batch.py +++ /dev/null @@ -1,11 +0,0 @@ -def handler(event): - texts = event.get( - "texts", - [ - "serverless workflows on gpu", - "short test sentence", - "embedding benchmark sample", - "fan out and summarize", - ], - ) - return {"texts": texts} diff --git a/benchmarks/600.workflows/711.gpu-embed-batch/python/requirements.txt b/benchmarks/600.workflows/711.gpu-embed-batch/python/requirements.txt deleted file mode 100644 index b9d9d66d2..000000000 --- a/benchmarks/600.workflows/711.gpu-embed-batch/python/requirements.txt +++ /dev/null @@ -1,2 +0,0 @@ -torch -pillow diff --git a/benchmarks/600.workflows/711.gpu-embed-batch/python/summarize.py b/benchmarks/600.workflows/711.gpu-embed-batch/python/summarize.py deleted file mode 100644 index e7efaf8e5..000000000 --- a/benchmarks/600.workflows/711.gpu-embed-batch/python/summarize.py +++ /dev/null @@ -1,7 +0,0 @@ -import math - - -def handler(event): - embs = event["embeddings"] - norms = [math.sqrt(sum(x * x for x in v)) for v in embs] - return {"count": event["count"], "avg_norm": sum(norms) / len(norms)} diff --git a/benchmarks/600.workflows/717.gpu-traffic-count/config.json b/benchmarks/600.workflows/717.gpu-traffic-count/config.json deleted file mode 100644 index e14b3b052..000000000 --- a/benchmarks/600.workflows/717.gpu-traffic-count/config.json +++ /dev/null @@ -1,6 +0,0 @@ -{ - "timeout": 540, - "memory": 2048, - "languages": ["python"], - "modules": ["storage"] -} diff --git a/benchmarks/600.workflows/717.gpu-traffic-count/definition.json b/benchmarks/600.workflows/717.gpu-traffic-count/definition.json deleted file mode 100644 index 99acefbfd..000000000 --- a/benchmarks/600.workflows/717.gpu-traffic-count/definition.json +++ /dev/null @@ -1,27 +0,0 @@ -{ - "root": "split", - "states": { - "split": { - "type": "task", - "func_name": "split", - "next": "detect-map" - }, - "detect-map": { - "type": "map", - "root": "detect", - "array": "segments", - "next": "aggregate-loop", - "states": { - "detect": { - "type": "task", - "func_name": "detect" - } - } - }, - "aggregate-loop": { - "type": "loop", - "func_name": "aggregate", - "array": "segments" - } - } -} diff --git a/benchmarks/600.workflows/717.gpu-traffic-count/input.py b/benchmarks/600.workflows/717.gpu-traffic-count/input.py deleted file mode 100644 index 8ca65f8be..000000000 --- a/benchmarks/600.workflows/717.gpu-traffic-count/input.py +++ /dev/null @@ -1,63 +0,0 @@ -# benchmarks/600.workflows/717.gpu-traffic-count/input.py -import os - -size_generators = { - "test": (6, 3), - "small": (40, 8), - "large": (160, 16), -} - - -def buckets_count(): - return (1, 1) - - -def generate_input( - data_dir, - size, - benchmarks_bucket, - input_buckets, - output_buckets, - upload_func, - nosql_func, -): - if data_dir is None: - raise ValueError( - "/path/to/traffic_frames/\n" - " └── frames/\n" - " ├── cam01_0001.jpg\n" - " ├── cam01_0002.jpg\n" - " └── ..." - ) - - num_frames, batch_size = size_generators[size] - - frames_dir = os.path.join(data_dir, "frames") - if not os.path.isdir(frames_dir): - raise ValueError(f"frames dir not exist: {frames_dir}") - - frame_files = sorted( - f for f in os.listdir(frames_dir) if f.lower().endswith((".png", ".jpg", ".jpeg")) - ) - if not frame_files: - raise ValueError(f"no jpg/png under dir: {frames_dir}") - - new_frames = [] - for i in range(num_frames): - frame = frame_files[i % len(frame_files)] - ext = os.path.splitext(frame)[1].lower() or ".jpg" - name = f"{i:08d}{ext}" - path = os.path.join(frames_dir, frame) - - new_frames.append(name) - upload_func(0, name, path) - - assert len(new_frames) == num_frames - - return { - "segments": new_frames, - "benchmark_bucket": benchmarks_bucket, - "input_bucket": input_buckets[0], - "output_bucket": output_buckets[0], - "batch_size": batch_size, - } diff --git a/benchmarks/600.workflows/717.gpu-traffic-count/python/aggregate.py b/benchmarks/600.workflows/717.gpu-traffic-count/python/aggregate.py deleted file mode 100644 index 6abf7f0c7..000000000 --- a/benchmarks/600.workflows/717.gpu-traffic-count/python/aggregate.py +++ /dev/null @@ -1,65 +0,0 @@ -import json -import os -import shutil -import uuid -from collections import Counter - -from . import storage - -client = storage.storage.get_instance() - - -def _download_det(frame: str, tmp_dir: str, benchmark_bucket: str, output_bucket: str, prefix: str): - base, _ = os.path.splitext(frame) - remote_name = f"{prefix}{base}_det.json" - local_json = os.path.join(tmp_dir, f"{base}_det.json") - try: - client.download(benchmark_bucket, f"{output_bucket}/{remote_name}", local_json) - return local_json - except Exception: - return None - - -def handler(event): - frames = event["segments"] - benchmark_bucket = event["benchmark_bucket"] - output_bucket = event["output_bucket"] - prefix = event["prefix"] - - tmp_dir = os.path.join("/tmp", str(uuid.uuid4())) - os.makedirs(tmp_dir, exist_ok=True) - - vehicle_totals = Counter() - missing = [] - try: - for frame in frames: - det_path = _download_det(frame, tmp_dir, benchmark_bucket, output_bucket, prefix) - if det_path and os.path.exists(det_path): - with open(det_path, "r", encoding="utf-8") as f: - data = json.load(f) - vehicle_totals.update(data.get("vehicles", {})) - else: - missing.append(frame) - - summary = { - "total_frames": len(frames), - "labeled_frames": len(frames) - len(missing), - "vehicle_totals": {k: int(v) for k, v in vehicle_totals.items()}, - "missing": missing, - } - - summary_path = os.path.join(tmp_dir, "traffic_summary.json") - with open(summary_path, "w", encoding="utf-8") as f: - json.dump(summary, f, indent=2) - - client.upload( - benchmark_bucket, - f"{output_bucket}/{prefix}traffic_summary.json", - summary_path, - unique_name=False, - ) - - finally: - shutil.rmtree(tmp_dir, ignore_errors=True) - - return {**event, "summary": summary} diff --git a/benchmarks/600.workflows/717.gpu-traffic-count/python/detect.py b/benchmarks/600.workflows/717.gpu-traffic-count/python/detect.py deleted file mode 100644 index d3f7d21f5..000000000 --- a/benchmarks/600.workflows/717.gpu-traffic-count/python/detect.py +++ /dev/null @@ -1,81 +0,0 @@ -import json -import logging -import os -import shutil -import uuid -from collections import Counter - -import torch -from PIL import Image -from torchvision.models.detection import FasterRCNN_ResNet50_FPN_Weights, fasterrcnn_resnet50_fpn - -from . import storage - -logger = logging.getLogger(__name__) -client = storage.storage.get_instance() - -_weights = FasterRCNN_ResNet50_FPN_Weights.DEFAULT -_model = fasterrcnn_resnet50_fpn(weights=_weights).cuda().eval() -_transform = _weights.transforms() -_categories = _weights.meta["categories"] - -_vehicle_labels = {name for name in _categories if name in {"car", "truck", "bus", "motorcycle", "bicycle"}} - - -def handler(event): - input_bucket = event["input_bucket"] - output_bucket = event["output_bucket"] - benchmark_bucket = event["benchmark_bucket"] - frames = event["segments"] - prefix = event["prefix"] - - tmp_dir = os.path.join("/tmp", str(uuid.uuid4())) - os.makedirs(tmp_dir, exist_ok=True) - - results = [] - try: - for frame in frames: - local_img = os.path.join(tmp_dir, frame) - client.download(benchmark_bucket, f"{input_bucket}/{frame}", local_img) - - img = Image.open(local_img).convert("RGB") - x = _transform(img).to("cuda") - with torch.inference_mode(): - out = _model([x])[0] - - labels = [int(l) for l in out["labels"].tolist()] - scores = out["scores"].tolist() - counts = Counter() - detections = [] - for lbl, score in zip(labels, scores): - if score < 0.4: - continue - name = _categories[lbl] - if name not in _vehicle_labels: - continue - counts[name] += 1 - detections.append({"label": name, "score": float(score)}) - - result = { - "frame": frame, - "vehicles": {k: int(v) for k, v in counts.items()}, - "total": int(sum(counts.values())), - } - - local_json = os.path.join(tmp_dir, f"{os.path.splitext(frame)[0]}_det.json") - with open(local_json, "w", encoding="utf-8") as f: - json.dump(result, f) - - client.upload( - benchmark_bucket, - f"{output_bucket}/{prefix}{os.path.basename(local_json)}", - local_json, - unique_name=False, - ) - - results.append(result) - - finally: - shutil.rmtree(tmp_dir, ignore_errors=True) - - return {**event, "results": results} diff --git a/benchmarks/600.workflows/717.gpu-traffic-count/python/requirements.txt b/benchmarks/600.workflows/717.gpu-traffic-count/python/requirements.txt deleted file mode 100644 index 4c4968110..000000000 --- a/benchmarks/600.workflows/717.gpu-traffic-count/python/requirements.txt +++ /dev/null @@ -1,3 +0,0 @@ -torch -torchvision -pillow diff --git a/benchmarks/600.workflows/717.gpu-traffic-count/python/split.py b/benchmarks/600.workflows/717.gpu-traffic-count/python/split.py deleted file mode 100644 index 10378a090..000000000 --- a/benchmarks/600.workflows/717.gpu-traffic-count/python/split.py +++ /dev/null @@ -1,26 +0,0 @@ -import uuid - - -def _chunks(lst, n): - for i in range(0, len(lst), n): - yield lst[i : i + n] - - -def handler(event): - segs = _chunks(event["segments"], event["batch_size"]) - input_bucket = event["input_bucket"] - output_bucket = event["output_bucket"] - benchmark_bucket = event["benchmark_bucket"] - - return { - "segments": [ - { - "prefix": str(uuid.uuid4().int & ((1 << 64) - 1))[:8], - "segments": ss, - "input_bucket": input_bucket, - "output_bucket": output_bucket, - "benchmark_bucket": benchmark_bucket, - } - for ss in segs - ] - } diff --git a/benchmarks/600.workflows/718.gpu-document-qna/config.json b/benchmarks/600.workflows/718.gpu-document-qna/config.json deleted file mode 100644 index e14b3b052..000000000 --- a/benchmarks/600.workflows/718.gpu-document-qna/config.json +++ /dev/null @@ -1,6 +0,0 @@ -{ - "timeout": 540, - "memory": 2048, - "languages": ["python"], - "modules": ["storage"] -} diff --git a/benchmarks/600.workflows/718.gpu-document-qna/definition.json b/benchmarks/600.workflows/718.gpu-document-qna/definition.json deleted file mode 100644 index 597148b37..000000000 --- a/benchmarks/600.workflows/718.gpu-document-qna/definition.json +++ /dev/null @@ -1,26 +0,0 @@ -{ - "root": "split", - "states": { - "split": { - "type": "task", - "func_name": "split", - "next": "embed-map" - }, - "embed-map": { - "type": "map", - "root": "embed", - "array": "segments", - "next": "answer", - "states": { - "embed": { - "type": "task", - "func_name": "embed" - } - } - }, - "answer": { - "type": "task", - "func_name": "answer" - } - } -} diff --git a/benchmarks/600.workflows/718.gpu-document-qna/input.py b/benchmarks/600.workflows/718.gpu-document-qna/input.py deleted file mode 100644 index fca5bde4b..000000000 --- a/benchmarks/600.workflows/718.gpu-document-qna/input.py +++ /dev/null @@ -1,59 +0,0 @@ -# benchmarks/600.workflows/718.gpu-document-qna/input.py -import os - -size_generators = { - "test": (4, 500, 2), - "small": (16, 1500, 3), - "large": (48, 3000, 4), -} - - -def buckets_count(): - return (1, 1) - - -def generate_input( - data_dir, - size, - benchmarks_bucket, - input_buckets, - output_buckets, - upload_func, - nosql_func, -): - if data_dir is None: - raise ValueError( - "/path/to/documents/\n" - " └── docs/\n" - " ├── doc1.txt\n" - " ├── doc2.txt\n" - " └── ..." - ) - - num_docs, max_tokens, top_k = size_generators[size] - - docs_dir = os.path.join(data_dir, "docs") - if not os.path.isdir(docs_dir): - raise ValueError(f"docs dir not exist: {docs_dir}") - - doc_files = sorted(f for f in os.listdir(docs_dir) if f.lower().endswith(".txt")) - if not doc_files: - raise ValueError(f"no txt files under dir: {docs_dir}") - - new_docs = [] - for i in range(num_docs): - doc = doc_files[i % len(doc_files)] - name = f"{i:08d}.txt" - path = os.path.join(docs_dir, doc) - new_docs.append(name) - upload_func(0, name, path) - - return { - "segments": new_docs, - "benchmark_bucket": benchmarks_bucket, - "input_bucket": input_buckets[0], - "output_bucket": output_buckets[0], - "max_tokens": max_tokens, - "top_k": top_k, - "question": "Summarize the key risks mentioned in the document.", - } diff --git a/benchmarks/600.workflows/718.gpu-document-qna/python/answer.py b/benchmarks/600.workflows/718.gpu-document-qna/python/answer.py deleted file mode 100644 index e3610d670..000000000 --- a/benchmarks/600.workflows/718.gpu-document-qna/python/answer.py +++ /dev/null @@ -1,67 +0,0 @@ -import json -import math -import os -import shutil -import uuid -from typing import List - -import torch - -from . import storage - -client = storage.storage.get_instance() - - -def _cosine(a: torch.Tensor, b: torch.Tensor) -> float: - return float(torch.nn.functional.cosine_similarity(a, b, dim=0).item()) - - -def handler(event): - segments = event["segments"] - output_bucket = segments[0]["output_bucket"] - benchmark_bucket = segments[0]["benchmark_bucket"] - top_k = event.get("top_k", 3) - - tmp_dir = os.path.join("/tmp", str(uuid.uuid4())) - os.makedirs(tmp_dir, exist_ok=True) - - scores: List[dict] = [] - try: - for seg in segments: - doc = seg["doc"] - prefix = seg["prefix"] - json_path = os.path.join(tmp_dir, f"{os.path.splitext(doc)[0]}_emb.json") - client.download( - benchmark_bucket, - f"{output_bucket}/{prefix}{os.path.basename(json_path)}", - json_path, - ) - with open(json_path, "r", encoding="utf-8") as f: - emb_data = json.load(f) - doc_vec = torch.tensor(emb_data["embedding"]) - q_vec = torch.tensor(emb_data["question_embedding"]) - sim = _cosine(doc_vec, q_vec) - scores.append({"doc": doc, "score": sim, "question": emb_data["question"]}) - - top = sorted(scores, key=lambda x: x["score"], reverse=True)[: top_k or len(scores)] - answer = { - "question": top[0]["question"] if top else "", - "top_docs": top, - "total_docs": len(scores), - } - - answer_path = os.path.join(tmp_dir, "answer.json") - with open(answer_path, "w", encoding="utf-8") as f: - json.dump(answer, f, indent=2) - - client.upload( - benchmark_bucket, - f"{output_bucket}/answer.json", - answer_path, - unique_name=False, - ) - - finally: - shutil.rmtree(tmp_dir, ignore_errors=True) - - return answer diff --git a/benchmarks/600.workflows/718.gpu-document-qna/python/embed.py b/benchmarks/600.workflows/718.gpu-document-qna/python/embed.py deleted file mode 100644 index 05d9fea49..000000000 --- a/benchmarks/600.workflows/718.gpu-document-qna/python/embed.py +++ /dev/null @@ -1,75 +0,0 @@ -import json -import os -import shutil -import uuid - -import torch -from transformers import AutoModel, AutoTokenizer - -from . import storage - -client = storage.storage.get_instance() - -_MODEL_NAME = "intfloat/e5-small-v2" -_tokenizer = AutoTokenizer.from_pretrained(_MODEL_NAME) -_model = AutoModel.from_pretrained(_MODEL_NAME).cuda().eval() - - -def _encode(text: str) -> list: - tokens = _tokenizer( - text, - truncation=True, - max_length=512, - padding="max_length", - return_tensors="pt", - ) - for k in tokens: - tokens[k] = tokens[k].cuda() - with torch.inference_mode(): - outputs = _model(**tokens) - embeddings = outputs.last_hidden_state.mean(dim=1) - return embeddings.squeeze(0).cpu().tolist() - - -def handler(event): - input_bucket = event["input_bucket"] - output_bucket = event["output_bucket"] - benchmark_bucket = event["benchmark_bucket"] - doc = event["doc"] - prefix = event["prefix"] - question = event["question"] - - tmp_dir = os.path.join("/tmp", str(uuid.uuid4())) - os.makedirs(tmp_dir, exist_ok=True) - - try: - local_doc = os.path.join(tmp_dir, doc) - client.download(benchmark_bucket, f"{input_bucket}/{doc}", local_doc) - with open(local_doc, "r", encoding="utf-8") as f: - text = f.read() - - doc_vec = _encode(text) - q_vec = _encode(question) - - result = { - "doc": doc, - "embedding": doc_vec, - "question_embedding": q_vec, - "question": question, - } - - local_json = os.path.join(tmp_dir, f"{os.path.splitext(doc)[0]}_emb.json") - with open(local_json, "w", encoding="utf-8") as f: - json.dump(result, f) - - client.upload( - benchmark_bucket, - f"{output_bucket}/{prefix}{os.path.basename(local_json)}", - local_json, - unique_name=False, - ) - - finally: - shutil.rmtree(tmp_dir, ignore_errors=True) - - return {**event, "embedding": doc_vec, "question_embedding": q_vec} diff --git a/benchmarks/600.workflows/718.gpu-document-qna/python/requirements.txt b/benchmarks/600.workflows/718.gpu-document-qna/python/requirements.txt deleted file mode 100644 index d1548e6cd..000000000 --- a/benchmarks/600.workflows/718.gpu-document-qna/python/requirements.txt +++ /dev/null @@ -1,4 +0,0 @@ -torch -transformers -pillow -tokenizers diff --git a/benchmarks/600.workflows/718.gpu-document-qna/python/split.py b/benchmarks/600.workflows/718.gpu-document-qna/python/split.py deleted file mode 100644 index de0f037d7..000000000 --- a/benchmarks/600.workflows/718.gpu-document-qna/python/split.py +++ /dev/null @@ -1,25 +0,0 @@ -import uuid - - -def handler(event): - docs = event["segments"] - input_bucket = event["input_bucket"] - output_bucket = event["output_bucket"] - benchmark_bucket = event["benchmark_bucket"] - max_tokens = event["max_tokens"] - question = event["question"] - - return { - "segments": [ - { - "prefix": str(uuid.uuid4().int & ((1 << 64) - 1))[:8], - "doc": doc, - "input_bucket": input_bucket, - "output_bucket": output_bucket, - "benchmark_bucket": benchmark_bucket, - "max_tokens": max_tokens, - "question": question, - } - for doc in docs - ] - } From 2250ba76406a9586984bf1d2c7c4fb82f6387350 Mon Sep 17 00:00:00 2001 From: xipang Date: Wed, 10 Dec 2025 23:51:09 +0100 Subject: [PATCH 4/4] push workflow --- .../720.gpu-medical-scan/config.json | 6 -- .../720.gpu-medical-scan/definition.json | 26 ------ .../720.gpu-medical-scan/input.py | 28 ------ .../720.gpu-medical-scan/python/aggregate.py | 55 ------------ .../720.gpu-medical-scan/python/preprocess.py | 62 ------------- .../720.gpu-medical-scan/python/segment.py | 89 ------------------- 6 files changed, 266 deletions(-) delete mode 100644 benchmarks/600.workflows/720.gpu-medical-scan/config.json delete mode 100644 benchmarks/600.workflows/720.gpu-medical-scan/definition.json delete mode 100644 benchmarks/600.workflows/720.gpu-medical-scan/input.py delete mode 100644 benchmarks/600.workflows/720.gpu-medical-scan/python/aggregate.py delete mode 100644 benchmarks/600.workflows/720.gpu-medical-scan/python/preprocess.py delete mode 100644 benchmarks/600.workflows/720.gpu-medical-scan/python/segment.py diff --git a/benchmarks/600.workflows/720.gpu-medical-scan/config.json b/benchmarks/600.workflows/720.gpu-medical-scan/config.json deleted file mode 100644 index f83e0d66e..000000000 --- a/benchmarks/600.workflows/720.gpu-medical-scan/config.json +++ /dev/null @@ -1,6 +0,0 @@ -{ - "timeout": 300, - "memory": 4096, - "languages": ["python"], - "modules": [] -} diff --git a/benchmarks/600.workflows/720.gpu-medical-scan/definition.json b/benchmarks/600.workflows/720.gpu-medical-scan/definition.json deleted file mode 100644 index 972c1dcaa..000000000 --- a/benchmarks/600.workflows/720.gpu-medical-scan/definition.json +++ /dev/null @@ -1,26 +0,0 @@ -{ - "root": "preprocess", - "states": { - "preprocess": { - "type": "task", - "func_name": "preprocess", - "next": "segment-map" - }, - "segment-map": { - "type": "map", - "root": "segment", - "array": "slices", - "next": "aggregate", - "states": { - "segment": { - "type": "task", - "func_name": "segment" - } - } - }, - "aggregate": { - "type": "task", - "func_name": "aggregate" - } - } -} diff --git a/benchmarks/600.workflows/720.gpu-medical-scan/input.py b/benchmarks/600.workflows/720.gpu-medical-scan/input.py deleted file mode 100644 index e529a0603..000000000 --- a/benchmarks/600.workflows/720.gpu-medical-scan/input.py +++ /dev/null @@ -1,28 +0,0 @@ -size_generators = { - "test": (10, 128, 128, 3), # 10 slices, 128x128 resolution, 3 organs - "small": (25, 256, 256, 5), - "large": (50, 512, 512, 8), -} - - -def buckets_count(): - # No object storage buckets required for this workflow. - return (0, 0) - - -def generate_input( - data_dir, - size, - benchmarks_bucket, - input_buckets, - output_buckets, - upload_func, - nosql_func, -): - n_slices, height, width, n_organs = size_generators[size] - return { - "n_slices": n_slices, - "height": height, - "width": width, - "n_organs": n_organs, - } diff --git a/benchmarks/600.workflows/720.gpu-medical-scan/python/aggregate.py b/benchmarks/600.workflows/720.gpu-medical-scan/python/aggregate.py deleted file mode 100644 index 45f8c52c8..000000000 --- a/benchmarks/600.workflows/720.gpu-medical-scan/python/aggregate.py +++ /dev/null @@ -1,55 +0,0 @@ -def handler(event): - slices = event.get("slices", []) - - if not slices: - return { - "scan_summary": "No slices to analyze", - "total_slices": 0, - "total_anomalies": 0, - } - - # Aggregate organ statistics across all slices - total_anomalies = sum(s.get("anomaly_count", 0) for s in slices) - - # Collect all anomalies - all_anomalies = [] - for slice_result in slices: - for anomaly in slice_result.get("anomalies", []): - all_anomalies.append({ - **anomaly, - "slice_id": slice_result["slice_id"], - "slice_idx": slice_result["slice_idx"], - }) - - # Sort anomalies by severity - all_anomalies.sort(key=lambda a: a["severity"], reverse=True) - - # Aggregate organ stats across slices - organ_aggregate = {} - for slice_result in slices: - organ_stats = slice_result.get("organ_stats", {}) - for organ_name, stats in organ_stats.items(): - if organ_name not in organ_aggregate: - organ_aggregate[organ_name] = { - "total_pixels": 0, - "avg_percentage": 0.0, - } - organ_aggregate[organ_name]["total_pixels"] += stats["pixel_count"] - organ_aggregate[organ_name]["avg_percentage"] += stats["percentage"] - - # Calculate averages - n_slices = len(slices) - for organ_name in organ_aggregate: - organ_aggregate[organ_name]["avg_percentage"] /= n_slices - - # Generate diagnostic report - risk_level = "HIGH" if total_anomalies > n_slices * 0.3 else "MEDIUM" if total_anomalies > 0 else "LOW" - - return { - "scan_summary": f"Analyzed {n_slices} slices", - "total_slices": n_slices, - "total_anomalies": total_anomalies, - "risk_level": risk_level, - "organ_summary": organ_aggregate, - "top_anomalies": all_anomalies[:5], # Top 5 most severe - } diff --git a/benchmarks/600.workflows/720.gpu-medical-scan/python/preprocess.py b/benchmarks/600.workflows/720.gpu-medical-scan/python/preprocess.py deleted file mode 100644 index 31e9e1f72..000000000 --- a/benchmarks/600.workflows/720.gpu-medical-scan/python/preprocess.py +++ /dev/null @@ -1,62 +0,0 @@ -import uuid -import numpy as np - - -def normalize_slice(data: np.ndarray) -> np.ndarray: - """Normalize CT/MRI slice to [0, 1] range.""" - min_val, max_val = data.min(), data.max() - if max_val - min_val > 0: - return (data - min_val) / (max_val - min_val) - return data - - -def generate_synthetic_scan(height: int, width: int, seed: int) -> np.ndarray: - """Generate synthetic medical scan slice with anatomical features.""" - rng = np.random.default_rng(seed=seed) - - # Create base tissue structure - scan = rng.normal(0.3, 0.1, (height, width)).astype(np.float32) - - # Add circular organ-like structures - y, x = np.ogrid[:height, :width] - center_y, center_x = height // 2, width // 2 - - # Main organ (e.g., liver, kidney) - radius = min(height, width) // 3 - mask = (x - center_x)**2 + (y - center_y)**2 <= radius**2 - scan[mask] += rng.normal(0.4, 0.05, mask.sum()).astype(np.float32) - - # Add some smaller structures (lesions, vessels) - for i in range(3): - offset_x = rng.integers(-radius//2, radius//2) - offset_y = rng.integers(-radius//2, radius//2) - small_radius = rng.integers(10, 30) - small_mask = (x - center_x - offset_x)**2 + (y - center_y - offset_y)**2 <= small_radius**2 - scan[small_mask] += rng.normal(0.2, 0.03, small_mask.sum()).astype(np.float32) - - return normalize_slice(scan) - - -def handler(event): - n_slices = int(event["n_slices"]) - height = int(event["height"]) - width = int(event["width"]) - n_organs = int(event["n_organs"]) - scan_id = event.get("scan_id", str(uuid.uuid4())[:8]) - - slices = [] - for idx in range(n_slices): - # Generate synthetic scan data - scan_data = generate_synthetic_scan(height, width, seed=idx) - - slices.append({ - "slice_id": f"slice-{scan_id}-{idx:03d}", - "slice_idx": idx, - "scan_data": scan_data.tolist(), # Convert to list for JSON serialization - "height": height, - "width": width, - "n_organs": n_organs, - "scan_id": scan_id, - }) - - return {"slices": slices} diff --git a/benchmarks/600.workflows/720.gpu-medical-scan/python/segment.py b/benchmarks/600.workflows/720.gpu-medical-scan/python/segment.py deleted file mode 100644 index a1593b040..000000000 --- a/benchmarks/600.workflows/720.gpu-medical-scan/python/segment.py +++ /dev/null @@ -1,89 +0,0 @@ -import numpy as np - - -def apply_threshold_segmentation(scan: np.ndarray, n_organs: int) -> np.ndarray: - """ - Simple multi-threshold segmentation to identify different tissue types. - Simulates GPU-accelerated semantic segmentation (e.g., U-Net inference). - """ - segmentation = np.zeros_like(scan, dtype=np.int32) - - # Create threshold ranges for different organs/tissues - thresholds = np.linspace(0, 1, n_organs + 1) - - for organ_id in range(1, n_organs + 1): - lower = thresholds[organ_id - 1] - upper = thresholds[organ_id] - mask = (scan >= lower) & (scan < upper) - segmentation[mask] = organ_id - - return segmentation - - -def compute_organ_stats(segmentation: np.ndarray, n_organs: int) -> dict: - """Compute statistics for each segmented organ.""" - stats = {} - total_pixels = segmentation.size - - for organ_id in range(1, n_organs + 1): - mask = segmentation == organ_id - pixel_count = mask.sum() - - stats[f"organ_{organ_id}"] = { - "pixel_count": int(pixel_count), - "percentage": float(pixel_count / total_pixels * 100), - } - - return stats - - -def detect_anomalies(segmentation: np.ndarray, scan: np.ndarray) -> list: - """ - Detect potential anomalies (bright spots, lesions, etc.). - Simulates GPU-accelerated anomaly detection. - """ - anomalies = [] - - # Detect bright regions that could indicate lesions or tumors - threshold = 0.75 - anomaly_mask = scan > threshold - - if anomaly_mask.any(): - # Find connected components (simplified) - coords = np.argwhere(anomaly_mask) - if len(coords) > 0: - # Calculate centroid - centroid_y, centroid_x = coords.mean(axis=0) - anomalies.append({ - "type": "bright_region", - "location": [float(centroid_y), float(centroid_x)], - "size": int(len(coords)), - "severity": float(scan[anomaly_mask].mean()), - }) - - return anomalies - - -def handler(schedule): - slice_id = schedule["slice_id"] - scan_data = np.array(schedule["scan_data"], dtype=np.float32) - height = schedule["height"] - width = schedule["width"] - n_organs = schedule["n_organs"] - - # Perform segmentation (simulating GPU inference) - segmentation = apply_threshold_segmentation(scan_data, n_organs) - - # Compute organ statistics - organ_stats = compute_organ_stats(segmentation, n_organs) - - # Detect anomalies - anomalies = detect_anomalies(segmentation, scan_data) - - return { - "slice_id": slice_id, - "slice_idx": schedule["slice_idx"], - "organ_stats": organ_stats, - "anomalies": anomalies, - "anomaly_count": len(anomalies), - }