flyte-sdk-eval

Flyte 2 SDK Eval Skill

Build evaluation harnesses, unit tests, and validation pipelines for Flyte 2 workflows.

Grounding References

Resource	URL
Official docs	https://www.union.ai/docs/v2/flyte
Docs index (LLMs)	https://www.union.ai/docs/v2/flyte/llms.txt
SDK API reference	https://www.union.ai/docs/v2/union/api-reference/flyte-sdk/
CLI API reference	https://www.union.ai/docs/v2/union/api-reference/flyte-cli/
flyte-sdk source	https://github.com/flyteorg/flyte-sdk
Example code	https://github.com/unionai/unionai-examples
Flyte MCP tools	Available via the flyte-cluster and flyte-docs MCP servers

Testing Patterns

Direct Task Invocation (unit testing)

Test task logic directly without remote execution:

import pytest
from pipeline import preprocess, train, evaluate

def test_preprocess():
    """Test preprocessing logic in isolation."""
    result = preprocess(["a", "b", "c"])
    assert result is not None
    assert len(result) == 3

def test_train():
    """Test training with a small dataset."""
    import flyte
    import flyte.io
    data = flyte.io.DataFrame(polars.DataFrame({"x": [1, 2, 3], "y": [4, 5, 6]}))
    model = train(data)
    assert model is not None

def test_evaluate():
    """Test evaluation metrics."""
    import flyte
    model = flyte.io.File(path="/tmp/mock_model.pt")
    metrics = evaluate(model)
    assert "accuracy" in metrics
    assert 0 <= metrics["accuracy"] <= 1

Using flyte.run() for Integration Testing

Test the full workflow execution locally:

import pytest
import flyte
from pipeline import main

def test_full_pipeline():
    """Run the full pipeline locally with test data."""
    result = flyte.run(main, inputs={"data": ["test1", "test2"]})
    assert result is not None
    assert "accuracy" in result.outputs

def test_full_pipeline_with_inputs():
    """Test with specific inputs via flyte.run()."""
    result = flyte.run(
        main,
        inputs={"data": ["a", "b", "c"]},
    )
    assert result.status == "SUCCEEDED"