Update to Flyte

{
  "description": "Builds minimal evaluation harnesses (unit tests + small-run workflows) and suggests ways to validate correctness and performance early. Use when the user wants to test Flyte tasks, validate pipeline outputs, set up evaluation pipelines, or write unit tests for ML/data workflows. Trigger words: \"test\", \"evaluate\", \"validation\", \"unit test\", \"verify\", \"assert\", \"data quality\", \"metrics\", \"benchmark\".",
  "included_files": [],
  "name": "flyte-sdk-eval",
  "skill_md_contents": "---\nname: flyte-sdk-eval\ndescription: 'Builds minimal evaluation harnesses (unit tests + small-run workflows) and suggests ways to validate correctness and performance early. Use when the user wants to test Flyte tasks, validate pipeline outputs, set up evaluation pipelines, or write unit tests for ML/data workflows. Trigger words: \"test\", \"evaluate\", \"validation\", \"unit test\", \"verify\", \"assert\", \"data quality\", \"metrics\", \"benchmark\".'\n---\n\n# Flyte 2 SDK Eval Skill\n\nBuild evaluation harnesses, unit tests, and validation pipelines for Flyte 2 workflows.\n\n## Grounding References\n\n| Resource | URL |\n|---|---|\n| Official docs | https://www.union.ai/docs/v2/flyte |\n| Docs index (LLMs) | https://www.union.ai/docs/v2/flyte/llms.txt |\n| SDK API reference | https://www.union.ai/docs/v2/union/api-reference/flyte-sdk/ |\n| CLI API reference | https://www.union.ai/docs/v2/union/api-reference/flyte-cli/ |\n| flyte-sdk source | https://github.com/flyteorg/flyte-sdk |\n| Example code | https://github.com/unionai/unionai-examples |\n| Flyte MCP tools | Available via the `flyte-cluster` and `flyte-docs` MCP servers |\n\n## Testing Patterns\n\n### Direct Task Invocation (unit testing)\n\nTest task logic directly without remote execution:\n\n```python\nimport pytest\nfrom pipeline import preprocess, train, evaluate\n\ndef test_preprocess():\n    \"\"\"Test preprocessing logic in isolation.\"\"\"\n    result = preprocess([\"a\", \"b\", \"c\"])\n    assert result is not None\n    assert len(result) == 3\n\ndef test_train():\n    \"\"\"Test training with a small dataset.\"\"\"\n    import flyte\n    import flyte.io\n    data = flyte.io.DataFrame(polars.DataFrame({\"x\": [1, 2, 3], \"y\": [4, 5, 6]}))\n    model = train(data)\n    assert model is not None\n\ndef test_evaluate():\n    \"\"\"Test evaluation metrics.\"\"\"\n    import flyte\n    model = flyte.io.File(path=\"/tmp/mock_model.pt\")\n    metrics = evaluate(model)\n    assert \"accuracy\" in metrics\n    assert 0 <= metrics[\"accuracy\"] <= 1\n```\n\n### Using flyte.run() for Integration Testing\n\nTest the full workflow execution locally:\n\n```python\nimport pytest\nimport flyte\nfrom pipeline import main\n\ndef test_full_pipeline():\n    \"\"\"Run the full pipeline locally with test data.\"\"\"\n    result = flyte.run(main, inputs={\"data\": [\"test1\", \"test2\"]})\n    assert result is not None\n    assert \"accuracy\" in result.outputs\n\ndef test_full_pipeline_with_inputs():\n    \"\"\"Test with specific inputs via flyte.run().\"\"\"\n    result = flyte.run(\n        main,\n        inputs={\"data\": [\"a\", \"b\", \"c\"]},\n    )\n    assert result.status == \"SUCCEEDED\"\n```\n\n### Testing Async Tasks\n\n```python\nimport asyncio\nimport pytest\n\ndef test_async_task():\n    \"\"\"Test async task by running in event loop.\"\"\"\n    result = asyncio.run(preprocess([\"a\", \"b\"]))\n    assert result is not None\n\n@pytest.mark.asyncio\nasync def test_async_task_mark():\n    \"\"\"Test async task with pytest-asyncio.\"\"\"\n    result = await preprocess([\"a\", \"b\"])\n    assert len(result) == 2\n```\n\n## Evaluation Pipeline Patterns\n\n### ML Model Evaluation\n\n```python\nimport flyte\nimport flyte.io\n\nenv = flyte.TaskEnvironment(\n    name=\"eval\",\n    image=flyte.Image.from_debian_base(python_version=(3, 12)).with_pip_packages(\n        \"scikit-learn\", \"scipy\", \"pandas\", \"matplotlib\",\n    ),\n)\n\n@env.task\nasync def load_test_data() -> flyte.io.DataFrame:\n    \"\"\"Load ground truth test data.\"\"\"\n    ...\n    return flyte.io.DataFrame(df)\n\n@env.task\nasync def load_model(model_uri: str) -> object:\n    \"\"\"Load a trained model.\"\"\"\n    ...\n\n@env.task\nasync def predict(model: object, data: flyte.io.DataFrame) -> flyte.io.DataFrame:\n    \"\"\"Run model predictions on test data.\"\"\"\n    ...\n\n@env.task\nasync def compute_metrics(\n    predictions: flyte.io.DataFrame,\n    ground_truth: flyte.io.DataFrame,\n) -> dict:\n    \"\"\"Compute evaluation metrics.\"\"\"\n    from sklearn.metrics import (\n        accuracy_score, f1_score, precision_score, recall_score,\n        roc_auc_score, mean_squared_error,\n    )\n    y_true = ground_truth.to_polars()[\"label\"].to_list()\n    y_pred = predictions.to_polars()[\"prediction\"].to_list()\n    y_prob = predictions.to_polars()[\"probability\"].to_list()\n\n    return {\n        \"accuracy\": accuracy_score(y_true, y_pred),\n        \"f1\": f1_score(y_true, y_pred),\n        \"precision\": precision_score(y_true, y_pred),\n        \"recall\": recall_score(y_true, y_pred),\n        \"auc\": roc_auc_score(y_true, y_prob),\n    }\n\n@env.task\nasync def generate_report(metrics: dict) -> flyte.io.File:\n    \"\"\"Generate an evaluation report.\"\"\"\n    import matplotlib.pyplot as plt\n    fig, axes = plt.subplots(1, 2, figsize=(12, 4))\n    # Confusion matrix, ROC curve, etc.\n    path = \"/tmp/eval_report.png\"\n    fig.savefig(path)\n    return flyte.io.File(path=path)\n\n@env.task\nasync def evaluate_pipeline(\n    model_uri: str,\n    test_data_uri: str,\n) -> dict:\n    \"\"\"Full evaluation pipeline.\"\"\"\n    data = await load_test_data()\n    model = await load_model(model_uri)\n    preds = await predict(model, data)\n    metrics = await compute_metrics(preds, data)\n    report = await generate_report(metrics)\n    return {\"metrics\": metrics, \"report\": report}\n```\n\n### A/B Model Comparison\n\n```python\n@env.task\nasync def compare_models(\n    model_a_uri: str,\n    model_b_uri: str,\n    test_data: flyte.io.DataFrame,\n) -> dict:\n    \"\"\"Compare two models on the same test data.\"\"\"\n    model_a = await load_model(model_a_uri)\n    model_b = await load_model(model_b_uri)\n    preds_a = await predict(model_a, test_data)\n    preds_b = await predict(model_b, test_data)\n    metrics_a = await compute_metrics(preds_a, test_data)\n    metrics_b = await compute_metrics(preds_b, test_data)\n\n    winner = \"A\" if metrics_a[\"accuracy\"] > metrics_b[\"accuracy\"] else \"B\"\n    return {\n        \"model_a_metrics\": metrics_a,\n        \"model_b_metrics\": metrics_b,\n        \"winner\": winner,\n        \"improvement\": metrics_a[\"accuracy\"] - metrics_b[\"accuracy\"],\n    }\n```\n\n### Data Quality Validation\n\n```python\n@env.task\nasync def validate_data(df: flyte.io.DataFrame) -> dict:\n    \"\"\"Run data quality checks.\"\"\"\n    inner = df.to_polars()\n    checks = {}\n\n    # Row count check\n    row_count = len(inner)\n    checks[\"row_count\"] = row_count\n    if row_count == 0:\n        raise ValueError(\"Dataset is empty\")\n\n    # Null check\n    null_counts = inner.null_count().to_dict()\n    checks[\"null_counts\"] = null_counts\n    for col, count in null_counts.items():\n        if count > 0 and count / row_count > 0.5:\n            raise ValueError(f\"Column {col} has >50% nulls\")\n\n    # Type check\n    checks[\"dtypes\"] = {str(k): str(v) for k, v in inner.schema.items()}\n\n    # Value range check\n    for col in inner.columns:\n        if inner[col].dtype.is_float64():\n            min_val = inner[col].min()\n            max_val = inner[col].max()\n            if min_val < 0 or max_val > 1:\n                checks[f\"range_{col}\"] = {\"min\": min_val, \"max\": max_val}\n\n    return {\"passed\": True, \"checks\": checks}\n\n@env.task\nasync def data_quality_gate(\n    data: flyte.io.DataFrame,\n    threshold: float = 0.9,\n) -> bool:\n    \"\"\"Pass/fail gate based on data quality score.\"\"\"\n    result = await validate_data(data)\n    score = result[\"checks\"].get(\"quality_score\", 1.0)\n    if score < threshold:\n        raise ValueError(f\"Data quality gate failed: {score} < {threshold}\")\n    return True\n```\n\n### Pipeline Output Validation\n\n```python\n@env.task\nasync def validate_output(\n    model_path: flyte.io.File,\n    metrics: dict,\n    min_accuracy: float = 0.8,\n) -> dict:\n    \"\"\"Validate that pipeline outputs meet quality thresholds.\"\"\"\n    validation = {\n        \"model_exists\": model_path is not None,\n        \"metrics_valid\": all(v >= 0 and v <= 1 for v in metrics.values()),\n        \"accuracy_threshold\": metrics.get(\"accuracy\", 0) >= min_accuracy,\n    }\n\n    if not validation[\"accuracy_threshold\"]:\n        raise ValueError(\n            f\"Model accuracy {metrics['accuracy']} below threshold {min_accuracy}\"\n        )\n\n    return validation\n```\n\n## Experiment Tracking\n\n### Manual Experiment Tracking\n\n```python\nimport json\nimport datetime\nimport flyte\nimport flyte.io\n\n@env.task\nasync def track_experiment(\n    experiment_name: str,\n    hyperparams: dict,\n    metrics: dict,\n) -> flyte.io.File:\n    \"\"\"Track experiment results as a JSON file.\"\"\"\n    record = {\n        \"experiment\": experiment_name,\n        \"timestamp\": datetime.datetime.now().isoformat(),\n        \"hyperparameters\": hyperparams,\n        \"metrics\": metrics,\n    }\n    path = f\"/tmp/experiments/{experiment_name}.json\"\n    with open(path, \"w\") as f:\n        json.dump(record, f, indent=2)\n    return flyte.io.File(path=path)\n\n@env.task\nasync def run_experiment(\n    config: dict,\n    data: flyte.io.DataFrame,\n) -> dict:\n    \"\"\"Run a single experiment and track results.\"\"\"\n    model = await train(data, config)\n    metrics = await evaluate(model, data)\n    await track_experiment(config[\"name\"], config, metrics)\n    return metrics\n```\n\n### Hyperparameter Search with Tracking\n\n```python\n@env.task\nasync def hpo_search(\n    param_grid: list[dict],\n    data: flyte.io.DataFrame,\n) -> dict:\n    \"\"\"Run hyperparameter search with experiment tracking.\"\"\"\n    results = await flyte.map(\n        lambda cfg: run_experiment(cfg, data),\n        param_grid,\n    )\n    best = max(results, key=lambda r: r[\"accuracy\"])\n    return best\n```\n\n## Performance Benchmarking\n\n### Task-level Benchmarking\n\n```python\nimport time\nimport flyte\n\n@env.task\nasync def benchmark_task(\n    task_fn,\n    inputs: dict,\n    num_runs: int = 5,\n) -> dict:\n    \"\"\"Benchmark a task's performance.\"\"\"\n    durations = []\n    for _ in range(num_runs):\n        start = time.time()\n        await task_fn(**inputs)\n        durations.append(time.time() - start)\n\n    return {\n        \"mean_ms\": sum(durations) / len(durations) * 1000,\n        \"min_ms\": min(durations) * 1000,\n        \"max_ms\": max(durations) * 1000,\n        \"p95_ms\": sorted(durations)[int(len(durations) * 0.95)] * 1000,\n    }\n```\n\n### Throughput Testing\n\n```python\n@env.task\nasync def throughput_test(\n    batch_sizes: list[int],\n) -> dict:\n    \"\"\"Test throughput at different batch sizes.\"\"\"\n    results = {}\n    for size in batch_sizes:\n        data = create_batch(size)\n        start = time.time()\n        await process_batch(data)\n        elapsed = time.time() - start\n        results[size] = {\n            \"throughput\": size / elapsed if elapsed > 0 else 0,\n            \"latency_ms\": elapsed * 1000 / size,\n        }\n    return results\n```\n\n## Testing with Flyte MCP\n\n### Inspecting runs with MCP\n\nIf Flyte MCP tools are available, use them to read a past run's inputs and outputs for\nvalidation, list recent runs for a task to compare against, and block until a run finishes\nbefore checking its status.\n\n\n## pytest Configuration\n\n```ini\n# pytest.ini\n[pytest]\ntestpaths = tests\nasyncio_mode = auto\nmarkers =\n    slow: marks tests as slow (deselect with '-m \"not slow\"')\n    integration: marks tests as integration tests\n```\n\n```toml\n# pyproject.toml\n[tool.pytest.ini_options]\ntestpaths = [\"tests\"]\nasyncio_mode = \"auto\"\nmarkers = [\n    \"slow: marks tests as slow\",\n    \"integration: marks tests as integration tests\",\n]\n```\n\n## Test Structure\n\n```\ntests/\n  __init__.py\n  test_preprocess.py     # unit tests for preprocessing\n  test_train.py           # unit tests for training\n  test_evaluate.py        # unit tests for evaluation\n  test_integration.py     # integration tests (flyte.run)\n  test_data_quality.py    # data quality validation\n  conftest.py             # shared fixtures\n```\n\n## Anti-Patterns\n\n1. **Don't test against remote runs in unit tests** — use direct function invocation for unit tests. Reserve `flyte.run()` for integration tests.\n2. **Don't hardcode test data paths** — use `flyte.io.File(path=\"/tmp/test_data\")` with temp directories.\n3. **Don't skip data quality gates** — always validate data before and after transformations.\n4. **Don't use Union-only features** — avoid `ReusePolicy` and other Union-specific APIs.\n5. **Don't test ML models with random data** — use representative test datasets that match production distribution.\n"
}