Flyte HITL Plugin
Human-in-the-Loop (HITL) plugin for Flyte. This plugin provides an event-based API for pausing workflows and waiting for human input.
Installation
pip install flyteplugins-hitl
Usage
import flyte
import flyteplugins.hitl as hitl
task_env = flyte.TaskEnvironment(
name="my-hitl-workflow",
image=flyte.Image.from_debian_base(python_version=(3, 12)),
resources=flyte.Resources(cpu=1, memory="512Mi"),
depends_on=[hitl.env],
)
@task_env.task
async def task1() -> int:
"""First task - returns an automated value."""
return 42
@task_env.task
async def task2(x: int, y: int) -> int:
"""Second task - combines automated and human input."""
return x + y
@task_env.task(report=True)
async def main() -> int:
"""
Main workflow that orchestrates automated and human-in-the-loop tasks.
Flow:
1. task1() runs and returns an automated value (x)
2. Create an Event (serves the app) and wait for human input (y)
3. task2(x, y) combines both values and returns the result
"""
print("Starting HITL workflow...")
# Step 1: Automated task
x = await task1()
print(f"task1 completed: x = {x}")
# Step 2: Human-in-the-loop using the Event-based API
# Create an event (this serves the app if not already running)
event = await hitl.new_event.aio(
"integer_input_event",
data_type=int,
scope="run",
prompt="What should I add to x?",
)
y = await event.wait.aio()
print(f"Event completed: y = {y}")
# Step 3: Combine results
result = await task2(x, y)
print(f"task2 completed: result = {result}")
return result
if __name__ == "__main__":
flyte.init_from_config()
run = flyte.run(main)
print(f"Run URL: {run.url}")
run.wait()
print(f"Result: {run.outputs()}")
Features
- Event-based API: Create events that pause workflow execution until human input is received
- Web Form: Automatically generates a web form for human input
- Programmatic API: Submit input via curl or any HTTP client
- Type Safety: Supports int, float, str, and bool data types
API Reference
hitl.new_event(name, data_type, scope, prompt, ...)
Create a new human-in-the-loop event.
Parameters:
name(str): A descriptive name for the eventdata_type(Type): The expected type of the input (int, float, str, bool)scope(str): The scope of the event. Currently only "run" is supported.prompt(str): The prompt to display to the humantimeout_seconds(int): Maximum time to wait for human input (default: 3600)poll_interval_seconds(int): How often to check for a response (default: 5)
Returns: An Event object
event.wait() / await event.wait.aio()
Wait for human input and return the result.
Returns: The value provided by the human, converted to the event's data_type
hitl.env
The TaskEnvironment that provides the HITL app infrastructure. Add this to your task environment's depends_on list.
hitl.Event
The Event class for type annotations and advanced usage.
1 maintainer
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distributions
Built Distribution
Filter files by name, interpreter, ABI, and platform.
If you're not sure about the file name format, learn more about wheel file names.
Copy a direct link to the current filters
File details
Details for the file flyteplugins_hitl-2.5.9-py3-none-any.whl.
File metadata
- Download URL: flyteplugins_hitl-2.5.9-py3-none-any.whl
- Upload date:
- Size: 13.7 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? No
- Uploaded via:
twine/6.2.0 CPython/3.13.14
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
42892f237714eb3ba305ac6a2b1a215f1a8be015bf4f29f8db0121628e64523f
|
|
| MD5 |
ecf164deb65e8c4bae6d00a6821c8785
|
|
| BLAKE2b-256 |
437a3c74934c2d21d7f2b4b4f5cf6fde9a0b8e5799042e4fd1b54c124a086003
|
