¿Quería detener un flujo de trabajo automatizado para esperar una decisión humana?
Tal vez necesite aprobación antes de aprovisionar los recursos en la nube, promover un modelo de aprendizaje automático a la producción o cobrar la tarjeta de crédito de un cliente.
En muchos Ciencia de datos y flujos de trabajo de aprendizaje automáticoLa automatización le brinda el 90% del camino, pero ese último paso crítico a menudo necesita un juicio humano.
Especialmente en entornos de producción, reentrenamiento modelo, anomalías anulacioneso Grandes movimientos de datos requiere una revisión humana cuidadosa para evitar errores costosos.
En mi caso, necesitaba revisar manualmente las situaciones donde mi sistema marcaba Más del 6% de los datos del cliente para anomalías – a menudo debido a los empujes accidentales por parte de los clientes.
Antes de implementar un flujo de trabajo adecuado, esto se manejó informalmente: los desarrolladores actualizarían directamente las bases de datos de producción (!): Riesgosos, propensos a errores e indescaltables.
Para resolver esto, construí un Sistema de aprobación manual escalable usando AWS Funciones de paso, Flojo, Lambday SNS -Una arquitectura de bajo costo nativo de la nube que detuvo limpiamente los flujos de trabajo para las aprobaciones humanas sin girar el cálculo inactivo.
En esta publicación, lo guiaré a través del diseño completo, los recursos de AWS involucrados y cómo puede aplicarlo a sus propios flujos de trabajo críticos.
Vamos a entrar en eso 👇
La solución
Mi aplicación se implementa en el ecosistema de AWS, por lo que usaremos Funciones del paso de AWS para construir una máquina de estado que:
- Ejecuta la lógica de negocios
- Lambda con
WaitForTaskTokenhacer una pausa hasta la aprobación - Envía un mensaje de Slack solicitando la aprobación (puede ser un correo electrónico/)
- Espera a que un humano haga clic en “aprobar” o “rechazar”
- Se reanuda automáticamente desde el mismo punto
Aquí hay un video de YouTube que muestra la demostración y la aplicación real en acción:
También he anfitrión el en vivo aplicación de demostración aquí →
👉 https://v0-manual-review-app-fwtjca.vercel.app
Todo el código está alojado aquí con el conjunto correcto de permisos de IAM.
Implementación paso a paso
- Ahora crearemos la función de paso con un paso de flujo de revisión manual. Aquí está la definición de la función de paso:
El flujo anterior genera un conjunto de datos, lo sube a AWS S3 y si se requiere una revisión, invoca la revisión manual de Lambda. En el paso de revisión manual, utilizaremos una tarea de lambda con un invoke en WaitForTaskTokenque detiene la ejecución hasta que se reanude. El Lambda lee el token de esta manera:
Python">def lambda_handler(event, context):
config = event["Payload"]["config"]
task_token = event["Payload"]["taskToken"] # Step Functions auto-generates this
reviewer = ManualReview(config, task_token)
reviewer.send_notification()
return configEste lambda envía un mensaje flojo que incluye el token de la tarea para que la función sepa qué ejecución reanudar.
2. Antes de que enviemos la notificación Slack, necesitamos
- Configurar un tema de SNS que reciba mensajes de revisión de la Lambda
- Un flujo de trabajo flojo con un gancho web suscrito al tema SNS, y una suscripción confirmada
- una puerta de enlace de la API HTTPS con
approvalyrejectionpuntos finales. - Una función lambda que procesa las solicitudes de puerta de enlace de la API: código
Seguí el video de YouTube aquí para mi configuración.
3. Una vez que se configure lo anterior, configure las variables en el paso de Hook Web del flujo de trabajo:
Y use las variables con una nota útil en el siguiente paso:
El flujo de trabajo final se verá así:
4. Envíe una notificación Slack publicada a un tema SNS (también puede usar alternativamente Slack-SDK) con los parámetros de trabajo. Así es como se verá el mensaje:
def publish_message(self, bucket_name: str, s3_file: str, subject: str = "Manual Review") -> 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.info(f"Publishing message to <{self.topic_arn}>, with subject: {subject}, message: {message}")
response = self.client.publish(
TopicArn=self.topic_arn,
Message=json.dumps(message),
Subject=subject
)
logging.info(f"Response: {response}")
return responseEste lambda envía un mensaje flojo que incluye el token de la tarea para que la función sepa qué ejecución reanudar.
def send_notification(self):
# As soon as this message is sent out, this callback lambda will go into a wait state,
# until an explicit call to this Lambda function execution is triggered.
# If you don't want this function to wait forever (or the default Steps timeout), ensure you setup
# an explicit timeout on this
self.sns.publish_message(self.s3_bucket_name, self.s3_key)
def lambda_handler(event, context):
config = event["Payload"]["config"]
task_token = event["Payload"]["taskToken"] # Step Functions auto-generates this
reviewer = ManualReview(config, task_token)
reviewer.send_notification()5. Una vez que se recibe una notificación de revisión en Slack, el usuario puede aprobarlo o rechazarlo. La función de paso entra en un estado de espera hasta que recibe una respuesta del usuario; Sin embargo, el token de la tarea expira en 24 horas, por lo que la inactividad temporiza la función de paso.
Según si el usuario aprueba o rechaza la solicitud de revisión, el RawPath se establece y se puede analizar aquí: código
action = event.get("rawPath", "").strip("/").lower()
# Extracts 'approve' or 'reject'El API Gateway + Lambda Combo:
- Analiza la carga útil
- Extractos
taskToken+ decisión - Usos
StepFunctions.send_task_success()osend_task_failure()
Ejemplo:
match action:
case "approve":
output_dict["is_manually_approved"] = True
response_message = "Approval processed successfully."
case "reject":
output_dict["is_manually_rejected"] = True
response_message = "Rejection processed successfully."
case _:
return {
"statusCode": 400,
"body": json.dumps({"error": "Invalid action. Use '/approve' or '/reject' in URL."})
}
...
sfn_client.send_task_success(
taskToken=task_token,
output=output
)Nota: Lambda configurado con WaitForTaskToken debe esperar. Si no envía el token, su flujo de trabajo solo se detiene.
Bonificación: si necesita alertas por correo electrónico o SMS, use SNS Para notificar a un grupo más amplio.
Justosns.publish()desde dentro de su lambda o función de paso.
Pruebas
Una vez que se conectó el sistema de aprobación manual, era hora de patear los neumáticos. Así es como lo probé:
- Justo después de publicar el flujo de trabajo Slack, confirmé la suscripción SNS antes de que se reenvíen los mensajes. No te saltes este paso.
- Luego, activé la función de paso manualmente con una carga útil falsa que simula un evento de marcado de datos.
- Cuando el flujo de trabajo presionó el paso de aprobación manual, envió un mensaje flojo con botones de aprobación/rechazo.
Probé Todos los caminos principales:
- Aprobar: Haga clic en Aprueba: vio el currículum de la función de paso y se complete con éxito.
- Rechazar: Haga clic en Rechazo: la función de paso se movió limpiamente en un estado de falla.
- Se acabó el tiempo: Ignoró el mensaje Slack: la función de paso esperó el tiempo de espera configurado y luego se agotó con gracia sin colgar.
Detrás de escena, también verifiqué que:
- La Lambda que recibió respuestas flojas estaba analizando correctamente las cargas de acción.
- No quedaron tokens de tareas pícaros colgando.
- Funciones de paso Las métricas y los registros de errores de holgura estaban limpios.
Recomiendo probar no solo rutas felices, sino también “¿Qué pasa si nadie hace clic?” y “¿Qué pasa si Slack se aplica?” – Atrapar estos casos de borde temprano me salvó dolores de cabeza más tarde.
Lecciones aprendidas
- Siempre use tiempos de espera: Establezca un tiempo de espera tanto en el
WaitForTaskTokenpaso y en toda la función de paso. Sin él, los flujos de trabajo pueden atascarse indefinidamente si nadie responde. - Pasar el contexto necesario: Si su función de paso necesita ciertos archivos, rutas o configuraciones de configuración después de reanudar, asegúrese de codificarlos y enviarlos En la notificación SNS.
Funciones de paso no Conserve automáticamente el contexto anterior en la memoria al reanudarse de una token de tarea. - Administre el ruido flojo: Tenga cuidado al spam a un canal de holgura con demasiadas solicitudes de revisión. Recomiendo crear canales separados Para el desarrollo, la UAT y la producción fluyen para mantener las cosas limpias.
- Bloquear los permisos temprano: Asegúrese de que todos sus recursos de AWS (funciones Lambda, API Gateway, cubos S3, temas de SNS) tengan permisos correctos y mínimos siguiendo el principio de menor privilegio. Donde necesitaba personalizar más allá de los valores predeterminados de AWS, escribí y publiqué las políticas de IAM en línea como JSON. (Encontrarás ejemplos en el Repositorio de Github).
- Firmar y acortar las URL: Si está enviando enlaces (por ejemplo, a los archivos S3) en mensajes Slack, pre-firmar las URL para el acceso seguro- y Acortarlos para una interfaz de usuario floja más limpia. Aquí hay un ejemplo rápido que usé:
shorten_url = requests.get(f"http://tinyurl.com/api-create.php?url={presigned_url}").text
default_links[key] = shorten_url if shorten_url else presigned_urlConcluir
Agregar lógica humana en el bucle no tiene que significar cinta adhesiva y trabajos cron. Con Funciones de paso + Slackpuede construir revisable, rastreable y Flujos de aprobación a prueba de producción.
Si esto ayudó, o estás intentando algo similar, ¡suelta una nota en los comentarios! Construyamos mejores flujos de trabajo.
Nota: Todas las imágenes en este artículo fueron creadas por el autor