Update to Flyte

{
  "description": "Migrates Flyte 1 machine learning code to Flyte 2 and unlocks net-new v2 patterns. Use when migrating Flyte 1 ML workloads (training, HPO, GPU/deep learning, batch inference) to Flyte 2, specifying GPU resources, or building the end-to-end pipeline pattern. Trigger words - migrate training, HPO, GPU, deep learning, batch inference, model serving, pytorch.",
  "included_files": [],
  "name": "flyte-migrate-ml",
  "skill_md_contents": "---\nname: flyte-migrate-ml\ndescription: Migrates Flyte 1 machine learning code to Flyte 2 and unlocks net-new v2 patterns. Use when migrating Flyte 1 ML workloads (training, HPO, GPU/deep learning, batch inference) to Flyte 2, specifying GPU resources, or building the end-to-end pipeline pattern. Trigger words - migrate training, HPO, GPU, deep learning, batch inference, model serving, pytorch.\n---\n\n# Flyte 1 to Flyte 2 ML Migration Skill\n\nMigrate existing Flyte 1 ML workloads — small-model training, hyperparameter optimization, deep learning on GPUs, and batch inference — to Flyte 2, then take advantage of patterns that were not possible in Flyte 1 (real-time serving, apps, sandboxed execution).\n\nThis skill is specifically about **migrating existing v1 ML code**. For greenfield authoring in Flyte 2, use the companion skills:\n\n- `flyte-sdk-ml` — writing new ML training / inference tasks in Flyte 2.\n- `flyte-sdk-app` — writing new apps and serving endpoints.\n- `flyte-sdk-agent` — writing new agents and sandboxed / code-mode workloads.\n\n## Grounding References\n\n| Resource | URL |\n|---|---|\n| Migration guide (ML workloads) | https://www.union.ai/docs/v2/flyte/user-guide/migration/flyte-2/ml-workloads/ |\n| Migration guide (New in Flyte 2) | https://www.union.ai/docs/v2/flyte/user-guide/migration/flyte-2/new-in-flyte-2/ |\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| Example code | https://github.com/unionai/unionai-examples |\n| Flyte MCP tools | Available via `flyte-mcp` server |\n\n## Migration Cheat Sheet\n\n| Flyte 1 | Flyte 2 |\n|---|---|\n| `ImageSpec(name=..., packages=[...])` | `flyte.Image.from_debian_base().with_pip_packages(...)` |\n| `@task(container_image=..., requests=..., cache=...)` | Set `image`, `resources`, `cache` once on `flyte.TaskEnvironment`, then `@env.task` |\n| `Resources(cpu=..., mem=...)` | `flyte.Resources(cpu=..., memory=...)` (note `mem` becomes `memory`) |\n| `Resources(gpu=\"1\")` + `accelerator=T4` | `flyte.Resources(gpu=\"T4:1\")` |\n| `FlyteFile` / `FlyteFile(path=...)` | `flyte.io.File` / `await File.from_local(...)` |\n| `model_file.download()` | `await model_file.download()` |\n| `current_context().working_directory` | `os.getcwd()` |\n| `@workflow` | An orchestrating `@env.task` (plain `async` Python) |\n| `map_task(fn)(x=xs)` | `await asyncio.gather(*[fn(x) for x in xs])` |\n| A \"pick the best\" task | Plain Python after `gather` |\n\n## Small model training (scikit-learn / XGBoost)\n\nTrain a model, persist it as a `File`, and evaluate it. Image, resources, and caching move to the `TaskEnvironment`; `FlyteFile` becomes `flyte.io.File`.\n\n### Flyte 1\n\n```python\nimport os\n\nimport joblib\nfrom flytekit import task, workflow, ImageSpec, Resources, current_context\nfrom flytekit.types.file import FlyteFile\nfrom sklearn.datasets import load_breast_cancer\nfrom sklearn.model_selection import train_test_split\nfrom xgboost import XGBClassifier\n\nimage = ImageSpec(\n    name=\"xgb-image\",\n    packages=[\"xgboost\", \"scikit-learn\", \"joblib\"],\n)\n\n@task(container_image=image, requests=Resources(cpu=\"2\", mem=\"4Gi\"))\ndef train_model(n_estimators: int, max_depth: int) -> FlyteFile:\n    data = load_breast_cancer()\n    X_train, _, y_train, _ = train_test_split(data.data, data.target, random_state=42)\n    model = XGBClassifier(n_estimators=n_estimators, max_depth=max_depth)\n    model.fit(X_train, y_train)\n\n    model_path = os.path.join(current_context().working_directory, \"model.json\")\n    joblib.dump(model, model_path)\n    return FlyteFile(path=model_path)\n\n@task(container_image=image)\ndef evaluate(model_file: FlyteFile) -> float:\n    model = joblib.load(model_file.download())\n    data = load_breast_cancer()\n    _, X_test, _, y_test = train_test_split(data.data, data.target, random_state=42)\n    return float(model.score(X_test, y_test))\n\n@workflow\ndef main(n_estimators: int, max_depth: int) -> float:\n    model = train_model(n_estimators=n_estimators, max_depth=max_depth)\n    return evaluate(model_file=model)\n```\n\n### Flyte 2\n\n```python\nimport os\n\nimport joblib\nimport flyte\nfrom flyte.io import File\nfrom sklearn.datasets import load_breast_cancer\nfrom sklearn.model_selection import train_test_split\nfrom xgboost import XGBClassifier\n\nenv = flyte.TaskEnvironment(\n    name=\"train_xgboost\",\n    image=flyte.Image.from_debian_base().with_pip_packages(\n        \"xgboost\", \"scikit-learn\", \"joblib\"\n    ),\n    resources=flyte.Resources(cpu=\"2\", memory=\"4Gi\"),\n)\n\n@env.task\nasync def train_model(n_estimators: int, max_depth: int) -> File:\n    data = load_breast_cancer()\n    X_train, _, y_train, _ = train_test_split(data.data, data.target, random_state=42)\n    model = XGBClassifier(n_estimators=n_estimators, max_depth=max_depth)\n    model.fit(X_train, y_train)\n\n    model_path = os.path.join(os.getcwd(), \"model.json\")\n    joblib.dump(model, model_path)\n    return await File.from_local(model_path)\n\n@env.task\nasync def evaluate(model_file: File) -> float:\n    local_path = await model_file.download()\n    model = joblib.load(local_path)\n    data = load_breast_cancer()\n    _, X_test, _, y_test = train_test_split(data.data, data.target, random_state=42)\n    return float(model.score(X_test, y_test))\n\n@env.task\nasync def main(n_estimators: int, max_depth: int) -> float:\n    model = await train_model(n_estimators, max_depth)\n    return await evaluate(model)\n```\n\n## Hyperparameter optimization\n\nFan out one training run per hyperparameter, then pick the best. In Flyte 1 the grid search runs through `map_task` and the \"pick the best\" step must itself be a task. In Flyte 2 you `gather` the runs and select the winner in plain Python.\n\n### Flyte 1\n\n```python\nfrom flytekit import task, workflow, map_task\nfrom sklearn.datasets import load_iris\nfrom sklearn.ensemble import RandomForestClassifier\nfrom sklearn.model_selection import cross_val_score\n\n@task\ndef get_grid() -> list[int]:\n    return [2, 4, 8, 16]\n\n@task\ndef train_eval(max_depth: int) -> float:\n    data = load_iris()\n    model = RandomForestClassifier(max_depth=max_depth, random_state=42)\n    scores = cross_val_score(model, data.data, data.target, cv=3)\n    return float(scores.mean())\n\n@task\ndef best_score(scores: list[float]) -> float:\n    return max(scores)\n\n@workflow\ndef main() -> float:\n    grid = get_grid()\n    # Fan out one training run per hyperparameter value.\n    scores = map_task(train_eval)(max_depth=grid)\n    return best_score(scores=scores)\n```\n\n### Flyte 2\n\n```python\nimport asyncio\n\nimport flyte\nfrom sklearn.datasets import load_iris\nfrom sklearn.ensemble import RandomForestClassifier\nfrom sklearn.model_selection import cross_val_score\n\nenv = flyte.TaskEnvironment(\n    name=\"hpo\",\n    image=flyte.Image.from_debian_base().with_pip_packages(\"scikit-learn\"),\n)\n\n@env.task\nasync def train_eval(max_depth: int) -> float:\n    data = load_iris()\n    model = RandomForestClassifier(max_depth=max_depth, random_state=42)\n    scores = cross_val_score(model, data.data, data.target, cv=3)\n    return float(scores.mean())\n\n@env.task\nasync def main() -> dict:\n    grid = [2, 4, 8, 16]\n    # Fan out one training run per hyperparameter value...\n    scores = await asyncio.gather(*[train_eval(d) for d in grid])\n    # ...then pick the best in plain Python (impossible in a Flyte 1 workflow).\n    best_idx = max(range(len(scores)), key=lambda i: scores[i])\n    return {\"best_max_depth\": grid[best_idx], \"best_score\": scores[best_idx]}\n```\n\n## Large model training (deep learning)\n\nGPU configuration moves to the `TaskEnvironment`: the Flyte 1 `Resources(gpu=\"1\")` plus a separate `accelerator=T4` become a single `gpu=\"T4:1\"` string on `flyte.Resources`.\n\n### Flyte 1\n\n```python\nfrom flytekit import task, workflow, ImageSpec, Resources\nfrom flytekit.extras.accelerators import T4\nimport torch\nimport torch.nn as nn\n\nimage = ImageSpec(\n    name=\"dl-image\",\n    packages=[\"torch\"],\n)\n\n@task(\n    container_image=image,\n    requests=Resources(cpu=\"4\", mem=\"16Gi\", gpu=\"1\"),\n    accelerator=T4,\n)\ndef train(epochs: int) -> float:\n    device = \"cuda\" if torch.cuda.is_available() else \"cpu\"\n    model = nn.Linear(10, 1).to(device)\n    optimizer = torch.optim.SGD(model.parameters(), lr=0.01)\n    loss_fn = nn.MSELoss()\n\n    X = torch.randn(128, 10, device=device)\n    y = torch.randn(128, 1, device=device)\n\n    loss = torch.tensor(0.0)\n    for _ in range(epochs):\n        optimizer.zero_grad()\n        loss = loss_fn(model(X), y)\n        loss.backward()\n        optimizer.step()\n    return float(loss.item())\n\n@workflow\ndef main(epochs: int) -> float:\n    return train(epochs=epochs)\n```\n\n### Flyte 2\n\n```python\nimport flyte\nimport torch\nimport torch.nn as nn\n\n# GPU type and count go in a single \"T4:1\"-style string. For multi-node\n# distributed training, wrap the training task with the torch elastic plugin.\nenv = flyte.TaskEnvironment(\n    name=\"train_deep_learning\",\n    image=flyte.Image.from_debian_base().with_pip_packages(\"torch\"),\n    resources=flyte.Resources(cpu=\"4\", memory=\"16Gi\", gpu=\"T4:1\"),\n)\n\n@env.task\nasync def train(epochs: int) -> float:\n    device = \"cuda\" if torch.cuda.is_available() else \"cpu\"\n    model = nn.Linear(10, 1).to(device)\n    optimizer = torch.optim.SGD(model.parameters(), lr=0.01)\n    loss_fn = nn.MSELoss()\n\n    X = torch.randn(128, 10, device=device)\n    y = torch.randn(128, 1, device=device)\n\n    loss = torch.tensor(0.0)\n    for _ in range(epochs):\n        optimizer.zero_grad()\n        loss = loss_fn(model(X), y)\n        loss.backward()\n        optimizer.step()\n    return float(loss.item())\n\n@env.task\nasync def main(epochs: int) -> float:\n    return await train(epochs)\n```\n\nFor multi-node distributed training (PyTorch elastic, etc.), wrap the training task with the torch elastic plugin. See the Resources docs and plugin integrations at https://www.union.ai/docs/v2/flyte/user-guide/task-configuration/resources.\n\n## Batch inference\n\nLoad a trained model once and score many batches in parallel. `map_task` with a `partial`-bound model becomes `asyncio.gather` over the batches, reusing the same model reference.\n\n### Flyte 1\n\n```python\nimport os\nfrom functools import partial\n\nimport joblib\nfrom flytekit import task, workflow, map_task, ImageSpec, current_context\nfrom flytekit.types.file import FlyteFile\nfrom sklearn.datasets import load_iris\nfrom sklearn.ensemble import RandomForestClassifier\n\nimage = ImageSpec(name=\"inference-image\", packages=[\"scikit-learn\", \"joblib\"])\n\n@task(container_image=image)\ndef train_model() -> FlyteFile:\n    data = load_iris()\n    model = RandomForestClassifier().fit(data.data, data.target)\n    model_path = os.path.join(current_context().working_directory, \"model.joblib\")\n    joblib.dump(model, model_path)\n    return FlyteFile(path=model_path)\n\n@task(container_image=image)\ndef get_batches() -> list[list[list[float]]]:\n    data = load_iris()\n    rows = data.data.tolist()\n    # Split the rows into batches of 30.\n    return [rows[i : i + 30] for i in range(0, len(rows), 30)]\n\n@task(container_image=image)\ndef score_batch(model_file: FlyteFile, batch: list[list[float]]) -> list[int]:\n    model = joblib.load(model_file.download())\n    return [int(p) for p in model.predict(batch)]\n\n@workflow\ndef main() -> list[list[int]]:\n    model = train_model()\n    batches = get_batches()\n    return map_task(partial(score_batch, model_file=model))(batch=batches)\n```\n\n### Flyte 2\n\n```python\nimport asyncio\nimport os\n\nimport joblib\nimport flyte\nfrom flyte.io import File\nfrom sklearn.datasets import load_iris\nfrom sklearn.ensemble import RandomForestClassifier\n\nenv = flyte.TaskEnvironment(\n    name=\"batch_inference\",\n    image=flyte.Image.from_debian_base().with_pip_packages(\"scikit-learn\", \"joblib\"),\n)\n\n@env.task\nasync def train_model() -> File:\n    data = load_iris()\n    model = RandomForestClassifier().fit(data.data, data.target)\n    model_path = os.path.join(os.getcwd(), \"model.joblib\")\n    joblib.dump(model, model_path)\n    return await File.from_local(model_path)\n\n@env.task\nasync def score_batch(model_file: File, batch: list[list[float]]) -> list[int]:\n    local_path = await model_file.download()\n    model = joblib.load(local_path)\n    return [int(p) for p in model.predict(batch)]\n\n@env.task\nasync def main() -> list[list[int]]:\n    model = await train_model()\n    rows = load_iris().data.tolist()\n    batches = [rows[i : i + 30] for i in range(0, len(rows), 30)]\n    # Score every batch in parallel, reusing the same model reference.\n    coros = [score_batch(model, batch) for batch in batches]\n    return list(await asyncio.gather(*coros))\n```\n\n## A complete example: end-to-end ML pipeline\n\nPutting it together — a load / train / evaluate pipeline shows the image, resources, caching, file I/O, and orchestration changes in one place. Image, resources, and cache are set **once** on the `TaskEnvironment`, and the \"workflow\" is just an orchestrating task.\n\n### Flyte 1\n\n```python\nimport os\n\nimport joblib\nimport pandas as pd\nfrom flytekit import task, workflow, ImageSpec, Resources, current_context\nfrom flytekit.types.file import FlyteFile\nfrom sklearn.datasets import load_iris\nfrom sklearn.ensemble import RandomForestClassifier\n\nimage = ImageSpec(\n    name=\"ml-image\",\n    packages=[\"pandas\", \"scikit-learn\", \"joblib\"],\n)\n\n@task(\n    container_image=image,\n    requests=Resources(cpu=\"2\", mem=\"4Gi\"),\n    cache=True,\n    cache_version=\"1.0\",\n)\ndef load_data() -> pd.DataFrame:\n    data = load_iris(as_frame=True)\n    df = data.frame\n    df[\"species\"] = data.target\n    return df\n\n@task(container_image=image)\ndef train_model(data: pd.DataFrame) -> FlyteFile:\n    model = RandomForestClassifier()\n    X = data.drop(\"species\", axis=1)\n    y = data[\"species\"]\n    model.fit(X, y)\n\n    model_path = os.path.join(current_context().working_directory, \"model.joblib\")\n    joblib.dump(model, model_path)\n    return FlyteFile(path=model_path)\n\n@task(container_image=image)\ndef evaluate(model_file: FlyteFile, data: pd.DataFrame) -> float:\n    model = joblib.load(model_file.download())\n    X = data.drop(\"species\", axis=1)\n    y = data[\"species\"]\n    return float(model.score(X, y))\n\n@workflow\ndef main() -> float:\n    data = load_data()\n    model = train_model(data=data)\n    return evaluate(model_file=model, data=data)\n```\n\n### Flyte 2\n\n```python\nimport os\n\nimport joblib\nimport pandas as pd\nimport flyte\nfrom flyte.io import File\nfrom sklearn.datasets import load_iris\nfrom sklearn.ensemble import RandomForestClassifier\n\n# Image, resources, and cache are set once on the TaskEnvironment.\nenv = flyte.TaskEnvironment(\n    name=\"ml_pipeline\",\n    image=flyte.Image.from_debian_base().with_pip_packages(\n        \"pandas\", \"scikit-learn\", \"joblib\"\n    ),\n    resources=flyte.Resources(cpu=\"2\", memory=\"4Gi\"),\n    cache=\"auto\",\n)\n\n@env.task\nasync def load_data() -> pd.DataFrame:\n    data = load_iris(as_frame=True)\n    df = data.frame\n    df[\"species\"] = data.target\n    return df\n\n@env.task\nasync def train_model(data: pd.DataFrame) -> File:\n    model = RandomForestClassifier()\n    X = data.drop(\"species\", axis=1)\n    y = data[\"species\"]\n    model.fit(X, y)\n\n    model_path = os.path.join(os.getcwd(), \"model.joblib\")\n    joblib.dump(model, model_path)\n    return await File.from_local(model_path)\n\n@env.task\nasync def evaluate(model_file: File, data: pd.DataFrame) -> float:\n    local_path = await model_file.download()\n    model = joblib.load(local_path)\n    X = data.drop(\"species\", axis=1)\n    y = data[\"species\"]\n    return float(model.score(X, y))\n\n# The \"workflow\" is just an orchestrating task.\n@env.task\nasync def main() -> float:\n    data = await load_data()\n    model = await train_model(data)\n    return await evaluate(model, data)\n```\n\n## New in Flyte 2\n\nFlyte 1 was a batch orchestration system: everything ran as a finite DAG that started, did work, and finished. Flyte 2 keeps all of that and adds long-running services, high-throughput batch inference, and sandboxed code execution — so the same project that trains your model can also serve it, host a dashboard, saturate a GPU, or safely run LLM-generated code. There is no v1 counterpart to migrate here; these are net-new capabilities that your migrated training code unlocks. For greenfield authoring of these, see the `flyte-sdk-app` and `flyte-sdk-agent` skills.\n\n### Real-time inference and model serving\n\nInstead of scoring a batch and exiting, you can stand up an always-on REST endpoint from a `FastAPIAppEnvironment` and deploy it with `flyte.deploy`. The app can load a model artifact produced by one of your migrated training tasks.\n\n```python\napp = FastAPI(title=\"ML Model API\")\n\n# Define request/response models\nclass PredictionRequest(BaseModel):\n    feature1: float\n    feature2: float\n    feature3: float\n\nclass PredictionResponse(BaseModel):\n    prediction: float\n    probability: float\n\n# Load model (you would typically load this from storage)\nmodel = None\n\n@asynccontextmanager\nasync def lifespan(app: FastAPI):\n    global model\n    model_path = os.getenv(\"MODEL_PATH\", \"/app/models/model.joblib\")\n    # In production, load from your storage\n    if os.path.exists(model_path):\n        with open(model_path, \"rb\") as f:\n            model = joblib.load(f)\n    yield\n\n@app.post(\"/predict\", response_model=PredictionResponse)\nasync def predict(request: PredictionRequest):\n    # Make prediction\n    # prediction = model.predict([[request.feature1, request.feature2, request.feature3]])\n\n    # Dummy prediction for demo\n    prediction = 0.85\n    probability = 0.92\n\n    return PredictionResponse(\n        prediction=prediction,\n        probability=probability,\n    )\n\nenv = FastAPIAppEnvironment(\n    name=\"ml-model-api\",\n    app=app,\n    image=flyte.Image.from_debian_base(python_version=(3, 12)).with_pip_packages(\n        \"fastapi\",\n        \"uvicorn\",\n        \"scikit-learn\",\n        \"pydantic\",\n        \"joblib\",\n    ),\n    parameters=[\n        flyte.app.Parameter(\n            name=\"model_file\",\n            value=flyte.io.File.from_existing_remote(\"s3://bucket/models/model.joblib\"),\n            mount=\"/app/models\",\n            env_var=\"MODEL_PATH\",\n        ),\n    ],\n    resources=flyte.Resources(cpu=2, memory=\"2Gi\"),\n    requires_auth=False,\n)\n```\n\nFor serving large language models, the `flyteplugins-vllm` integration gives you a production-grade vLLM server (with autoscaling to zero) via `VLLMAppEnvironment`. Any web app — a Streamlit dashboard, a Gradio demo, a Flask backend — runs as a `flyte.app.AppEnvironment` that you configure with image, resources, port, autoscaling, and a custom subdomain, then `flyte.serve`.\n\n### Dynamic batching for GPU inference\n\nFor in-process batch inference, `DynamicBatcher` from `flyte.extras` keeps an expensive GPU saturated: async producers load and preprocess data concurrently while a single consumer feeds the model in optimally-sized batches, with built-in backpressure. This replaces the Flyte 1 pattern of standing up a separate inference server just to get request batching.\n\n```python\nimport asyncio\nfrom flyte.extras import DynamicBatcher\n\nasync with DynamicBatcher(\n    process_fn=run_inference,   # takes a batch, returns results in the same order\n    target_batch_cost=1000,     # cost budget per batch\n    max_batch_size=64,          # hard cap on records per batch\n    batch_timeout_s=0.05,       # max wait before dispatching a partial batch\n) as batcher:\n    futures = [await batcher.submit(record) for record in records]\n    results = await asyncio.gather(*futures)\n```\n\n`submit()` is non-blocking and returns a `Future`; when the queue is full it applies backpressure automatically. See the batch inference docs for `TokenBatcher` (token-aware LLM batching).\n\n### Sandboxed code execution\n\n`flyte.sandbox.create()` runs arbitrary Python code or shell commands inside an ephemeral, single-use Docker container — built on demand from declared dependencies, executed once, then discarded. Only declared inputs go in and only declared outputs come back, which makes it the safe way to run untrusted code, most importantly code generated by an LLM.\n\n```python\n# sandbox_environment provides the base runtime for code sandboxes.\n# Include it in depends_on so the sandbox runtime is available when tasks execute.\nenv = flyte.TaskEnvironment(\n    name=\"sandbox-demo\",\n    image=flyte.Image.from_debian_base(name=\"sandbox-demo\"),\n    depends_on=[sandbox_environment],\n)\n\n# Auto-IO mode: pure computation. The code string runs in an isolated sandbox;\n# only the declared inputs go in and only the declared outputs come back.\nsum_sandbox = flyte.sandbox.create(\n    name=\"sum-to-n\",\n    code=\"total = sum(range(n + 1)) if conditional else 0\",\n    inputs={\"n\": int, \"conditional\": bool},\n    outputs={\"total\": int},\n)\n```\n\nCall it from a task with `await sum_sandbox.run.aio(n=10, conditional=True)`. This also powers **code mode** (programmatic tool calling), where an agent writes a whole program instead of emitting one tool call at a time.\n\n## Anti-Patterns\n\n1. **Don't keep `@task` / `@workflow` per-task config** — move `image`, `resources`, and `cache` onto a single `flyte.TaskEnvironment` and decorate with `@env.task`.\n2. **Don't leave a separate \"pick the best\" task** — after `asyncio.gather`, select the winner in plain Python inside the orchestrating task.\n3. **Don't carry `map_task` + `partial` into v2** — fan out with `asyncio.gather` over coroutines, reusing the same model reference.\n4. **Don't split GPU type and count** — replace `Resources(gpu=\"1\")` + `accelerator=T4` with a single `gpu=\"T4:1\"` string on `flyte.Resources`.\n5. **Don't use `mem=` or `current_context().working_directory`** — use `memory=` on `flyte.Resources` and `os.getcwd()` for local paths.\n6. **Don't forget `await`** — `File.from_local`, `download`, and task calls are all async in v2.\n7. **Don't hand-roll a serving container or a request-batching server** — use a `FastAPIAppEnvironment` / `AppEnvironment` for serving and `DynamicBatcher` for GPU batching.\n8. **Don't run untrusted or LLM-generated code inline** — use `flyte.sandbox.create()` with `sandbox_environment` in `depends_on`.\n"
}