Pause Your ML Pipelines for Human Review Using AWS Step Functions + Slack

wished to pause an automatic workflow to attend for a human choice?

Perhaps you want approval earlier than provisioning cloud assets, selling a machine studying mannequin to manufacturing, or charging a buyer’s bank card.

In lots of knowledge science and machine studying workflows, automation will get you 90% of the way in which — however that important final step typically wants human judgment.

Particularly in manufacturing environments, mannequin retraining, anomaly overrides, or giant knowledge actions require cautious human assessment to keep away from costly errors.

In my case, I wanted to manually assessment conditions the place my system flagged greater than 6% of buyer knowledge for anomalies — typically because of unintended pushes by prospects.

Earlier than I applied a correct workflow, this was dealt with informally: builders would straight replace manufacturing databases (!) — dangerous, error-prone, and unscalable.

To unravel this, I constructed a scalable handbook approval system utilizing AWS Step Features, Slack, Lambda, and SNS — a cloud-native, low-cost structure that cleanly paused workflows for human approvals with out spinning up idle compute.

On this submit, I’ll stroll you thru the total design, the AWS assets concerned, and how one can apply it to your individual important workflows.

Let’s get into it 👇

The Answer

My software is deployed within the AWS ecosystem, so we’ll use Aws Step Functions to construct a state machine that:

1. Executes enterprise logic
2. Lambda with WaitForTaskToken to pause till approval
3. Sends a Slack message requesting approval (could be an e mail/)
4. Waits for a human to click on “Approve” or “Reject”
5. Resumes routinely from the identical level

Here’s a youtube video exhibiting the demo and precise software in motion:

I’ve additionally hosted the stay demo app right here →
👉 https://v0-manual-review-app-fwtjca.vercel.app
All code is hosted here with the appropriate set of IAM permissions.

Step-by-Step Implementation

1. Now we’ll create the Step Perform with a handbook assessment circulate step. Right here is the step perform definition:

The circulate above generates a dataset, uploads it to AWS S3 and if a assessment is required, then invokes the Handbook Overview lambda. On the handbook assessment step, we’ll use a Process lambda with an invoke on WaitForTaskToken, which pauses execution till resumed. The lambda reads the token this manner:

Python">def lambda_handler(occasion, context):

  config = occasion["Payload"]["config"]
  task_token = occasion["Payload"]["taskToken"] # Step Features auto-generates this

  reviewer = ManualReview(config, task_token)
  reviewer.send_notification()

  return config

This Lambda sends a Slack message that features the duty token so the perform is aware of what execution to renew.

2. Earlier than the we ship out the slack notification, we have to

1. setup an SNS Subject that receives assessment messages from the lambda
2. a slack workflow with a web-hook subscribed to the SNS matter, and a confirmed subscription
3. an https API Gateway with approval and rejection endpoints.
4. a lambda perform that processes the API Gateway requests: code

I adopted the youtube video right here for my setup.

3. As soon as the above is setup, setup the variables into the web-hook step of the slack workflow:

And use the variables with a useful notice within the following step:

The ultimate workflow will seem like this:

4. Ship a Slack Notification printed to an SNS matter (you’ll be able to alternately use slack-sdk as effectively) with job parameters. Here’s what the message will seem like:

def publish_message(self, bucket_name: str, s3_file: str, topic: str = "Handbook Overview") -> dict:

    presigned_url = S3.generate_presigned_url(bucket_name, s3_file, expiration=86400)  # 1 day expiration

    message = {
        "approval_link": self.approve_link,
        "rejection_link": self.reject_link,
        "s3_file": presigned_url if presigned_url else s3_file
    }

    logging.data(f"Publishing message to <{self.topic_arn}>, with topic: {topic}, message: {message}")

    response = self.consumer.publish(
        TopicArn=self.topic_arn,
        Message=json.dumps(message),
        Topic=topic
    )

    logging.data(f"Response: {response}")
    return response

This Lambda sends a Slack message that features the duty token so the perform is aware of what execution to renew.

def send_notification(self):

    # As quickly as this message is distributed out, this callback lambda will go right into a wait state,
    # till an express name to this Lambda perform execution is triggered.

    # If you do not need this perform to attend perpetually (or the default Steps timeout), make sure you setup
    # an express timeout on this
    self.sns.publish_message(self.s3_bucket_name, self.s3_key)

def lambda_handler(occasion, context):

    config = occasion["Payload"]["config"]
    task_token = occasion["Payload"]["taskToken"]  # Step Features auto-generates this

    reviewer = ManualReview(config, task_token)
    reviewer.send_notification()

5. As soon as a assessment notification is acquired in slack, the consumer can approve or reject it. The step perform goes right into a wait state till it receives a consumer response; nevertheless the duty token is ready to run out in 24 hours, so inactivity will timeout the step perform.

Based mostly on whether or not the consumer approves or rejects the assessment request, the rawPath will get set and could be parsed right here: code

motion = occasion.get("rawPath", "").strip("/").decrease()  
# Extracts 'approve' or 'reject'

The receiving API Gateway + Lambda combo:

• Parses the Slack payload
• Extracts taskToken + choice
• Makes use of StepFunctions.send_task_success() or send_task_failure()

Instance:

match motion:
    case "approve":
        output_dict["is_manually_approved"] = True
        response_message = "Approval processed efficiently."
    case "reject":
        output_dict["is_manually_rejected"] = True
        response_message = "Rejection processed efficiently."
    case _:
        return {
            "statusCode": 400,
            "physique": json.dumps({"error": "Invalid motion. Use '/approve' or '/reject' in URL."})
        }

...

sfn_client.send_task_success(
    taskToken=task_token,
    output=output
)

Word: Lambda configured with WaitForTaskToken should wait. In the event you don’t ship the token, your workflow simply stalls.

Bonus: In the event you want e mail or SMS alerts, use SNS to inform a broader group.
Simply sns.publish() from inside your Lambda or Step Perform.

Testing

As soon as the handbook approval system was wired up, it was time to kick the tires. Right here’s how I examined it:

• Proper after publishing the slack workflow, I confirmed the SNS subscription earlier than messages get forwarded. Don’t skip this step.
• Then, I triggered the Step Perform manually with a faux payload simulating a knowledge flagging occasion.
• When the workflow hit the handbook approval step, it despatched a Slack message with Approve/Reject buttons.

I examined all main paths:

• Approve: Clicked Approve — noticed the Step Perform resume and full efficiently.
• Reject: Clicked Reject — Step Perform moved cleanly right into a failure state.
• Timeout: Ignored the Slack message — Step Perform waited for the configured timeout after which gracefully timed out with out hanging.

Behind the scenes, I additionally verified that:

• The Lambda receiving Slack responses was accurately parsing motion payloads.
• No rogue process tokens had been left hanging.
• Step Features metrics and Slack error logs had been clear.

I extremely advocate testing not simply glad paths, but in addition “what if no one clicks?” and “what if Slack glitches?” — catching these edge instances early saved me complications later.

Classes Discovered

• At all times use timeouts: Set a timeout each on the WaitForTaskToken step and on the whole Step Perform. With out it, workflows can get caught indefinitely if nobody responds.
• Go essential context: In case your Step Perform wants sure information, paths, or config settings after resuming, be sure you encode and ship them alongside within the SNS notification.
  Step Features don’t routinely retain earlier in-memory context when resuming from a Process Token.
• Handle Slack noise: Watch out about spamming a Slack channel with too many assessment requests. I like to recommend creating separate channels for improvement, UAT, and manufacturing flows to maintain issues clear.
• Lock down permissions early: Be sure that all of your AWS assets (Lambda features, API Gateway, S3 buckets, SNS Matters) have appropriate and minimal permissions following the precept of least privilege. The place I wanted to customise past AWS’s defaults, I wrote and posted inline IAM insurance policies as JSON. (You’ll discover examples within the GitHub repo).
• Pre-sign and shorten URLs: In the event you’re sending hyperlinks (e.g., to S3 information) in Slack messages, pre-sign the URLs for safe entry — and shorten them for a cleaner Slack UI. Right here’s a fast instance I used:

shorten_url = requests.get(f"http://tinyurl.com/api-create.php?url={presigned_url}").textual content
default_links[key] = shorten_url if shorten_url else presigned_url

Wrapping Up

Including human-in-the-loop logic doesn’t need to imply duct tape and cron jobs. With Step Features + Slack, you’ll be able to construct reviewable, traceable, and production-safe approval flows.

If this helped, otherwise you’re attempting one thing comparable, drop a notice within the feedback! Let’s construct higher workflows. 

Word: All pictures on this article had been created by the writer