Gestion avancée des erreurs dans les flux de vérification d'identité asynchrones (FR)

Nouvelles tentatives robustesMettez en œuvre une temporisation exponentielle et une gigue pour les erreurs transitoires lors des appels API aux services externes de vérification d'identité, prévenant la surcharge du système et améliorant les taux de succès.

Modèles de disjoncteursProtégez votre système contre les défaillances en cascade en suspendant temporairement les requêtes vers les services défaillants, leur permettant de récupérer et préservant la stabilité globale de l'application.

Journalisation et surveillance complètesUtilisez la journalisation structurée, les ID de corrélation et la surveillance en temps réel pour identifier, diagnostiquer et résoudre rapidement les problèmes au sein des pipelines de vérification d'identité asynchrones distribués.

Résilience intégrée de DiditLa plateforme modulaire et basée sur l'IA de Didit offre des flux de travail orchestrés et une conception API robuste, masquant la gestion complexe des erreurs pour les vérifications KYC, de vivacité et AML, améliorant la fiabilité et l'expérience développeur.

Dans le monde de la vérification d'identité, la rapidité et la fiabilité sont primordiales. À mesure que les entreprises évoluent, les flux de travail asynchrones deviennent essentiels pour gérer de grands volumes de requêtes sans bloquer le fil d'exécution principal de l'application. Cependant, cette nature distribuée et non bloquante introduit des complexités significatives, en particulier en ce qui concerne la gestion des erreurs. Les problèmes réseau, les pannes de service, les incohérences de données et les réponses API inattendues peuvent tous faire dérailler un processus de vérification d'identité, entraînant une mauvaise expérience utilisateur, des risques de conformité et des inefficacités opérationnelles.

Cet article de blog explore les stratégies avancées de gestion des erreurs pour les flux de travail de vérification d'identité asynchrones, en se concentrant spécifiquement sur les implémentations Python. Nous explorerons comment construire des systèmes plus résilients et tolérants aux pannes, garantissant que même lorsque les choses tournent mal, vos processus de vérification restent robustes.

Le défi des erreurs asynchrones dans la vérification d'identité

La vérification d'identité asynchrone implique souvent plusieurs services externes : un fournisseur de vérification d'identité comme Didit pour l'OCR et les contrôles de vivacité, un service de screening AML, une base de données de preuves d'adresse, et potentiellement d'autres sources de données. Chacune de ces interactions est un point de défaillance potentiel. La gestion synchrone traditionnelle des erreurs (par exemple, un simple bloc try-except) est insuffisante lorsque les opérations peuvent se terminer beaucoup plus tard, dans un processus différent, ou même échouer silencieusement sans retour immédiat.

Considérez un flux de travail KYC typique : un utilisateur télécharge sa pièce d'identité, un contrôle de vivacité est effectué, puis un screening AML est initié. Si le service de contrôle de vivacité rencontre un problème réseau transitoire, une simple nouvelle tentative immédiate pourrait exacerber le problème. Si le service AML est complètement en panne, des tentatives répétées ne feront que gaspiller des ressources et retarder l'onboarding de l'utilisateur.

Mise en œuvre de nouvelles tentatives robustes avec temporisation exponentielle et gigue

L'un des types d'erreurs les plus courants dans les systèmes distribués est les défaillances transitoires. Ce sont des problèmes temporaires comme des problèmes de réseau, des erreurs de service occupé ou des contentions de base de données qui se résolvent d'eux-mêmes après une courte période. Relancer aveuglément immédiatement après un échec peut surcharger un service en difficulté, entraînant une défaillance en cascade. La solution est des nouvelles tentatives intelligentes utilisant une temporisation exponentielle et une gigue.

La temporisation exponentielle implique d'augmenter le temps d'attente entre les nouvelles tentatives de manière exponentielle. Par exemple, attendre 1 seconde, puis 2, puis 4, puis 8, et ainsi de suite. Cela donne au service le temps de récupérer. La gigue ajoute un petit délai aléatoire au temps de temporisation, empêchant tous les clients de réessayer exactement au même moment, ce qui pourrait créer un problème de « thundering herd ».

import asyncio
import random

async def call_didit_api(data, attempt=0):
    max_retries = 5
    base_delay = 1  # seconds

    try:
        # Simulate an API call to Didit's ID Verification or Liveness service
        if random.random() < 0.6 and attempt < 3: # Simulate transient failure
            raise ConnectionError(f"Simulated API error on attempt {attempt+1}")

        print(f"Successfully called Didit API on attempt {attempt+1} with data: {data}")
        return {"status": "success", "result": "verification_data"}

    except (ConnectionError, asyncio.TimeoutError) as e:
        if attempt < max_retries - 1:
            delay = base_delay * (2 ** attempt) + random.uniform(0, 0.5) # Exponential backoff + jitter
            print(f"Attempt {attempt+1} failed: {e}. Retrying in {delay:.2f} seconds...")
            await asyncio.sleep(delay)
            return await call_didit_api(data, attempt + 1)
        else:
            print(f"All {max_retries} attempts failed for data: {data}")
            raise # Re-raise the last exception if all retries fail

async def main():
    try:
        # Example usage for Didit's ID Verification
        result = await call_didit_api({"document_image": "base64_id_scan"})
        print(f"Final result: {result}")

        # Example usage for Didit's Liveness
        result_liveness = await call_didit_api({"liveness_video": "base64_video"})
        print(f"Final liveness result: {result_liveness}")

    except Exception as e:
        print(f"Workflow failed after retries: {e}")

if __name__ == "__main__":
    asyncio.run(main())

Ce modèle est inestimable lors de l'intégration avec des services externes, y compris les API de vérification d'identité, de vivacité passive et active, ou de screening AML de Didit, qui bénéficient toutes d'une communication résiliente.

Mise en œuvre du modèle de disjoncteur

Bien que les nouvelles tentatives aident à gérer les erreurs transitoires, elles peuvent aggraver la situation si un service connaît des pannes prolongées. Le modèle de disjoncteur empêche votre application d'invoquer à plusieurs reprises un service susceptible d'échouer. Il fonctionne en surveillant les échecs, et si ceux-ci dépassent un certain seuil dans un délai donné, il « déclenche » le circuit, l'ouvrant pour empêcher d'autres appels au service défaillant. Après un délai configurable, il entre dans un état « semi-ouvert », permettant quelques requêtes de test pour voir si le service a récupéré.

import asyncio
import time
from collections import deque

class CircuitBreaker:
    def __init__(self, failure_threshold=3, recovery_timeout=10, half_open_attempts=1):
        self.state = "CLOSED"
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.half_open_attempts = half_open_attempts
        self.failures = 0
        self.last_failure_time = None
        self.successes_in_half_open = 0

    async def __call__(self, func, *args, **kwargs):
        if self.state == "OPEN":
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = "HALF_OPEN"
                self.successes_in_half_open = 0
                print("Circuit Breaker: Moving to HALF_OPEN state.")
            else:
                raise CircuitBreakerOpenError("Circuit is OPEN. Service is likely down.")

        try:
            result = await func(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure(e)
            raise

    def _on_success(self):
        if self.state == "HALF_OPEN":
            self.successes_in_half_open += 1
            if self.successes_in_half_open >= self.half_open_attempts:
                self.state = "CLOSED"
                self.failures = 0
                print("Circuit Breaker: Service recovered. Moving to CLOSED state.")
        elif self.state == "CLOSED":
            self.failures = 0 # Reset failures on success in closed state

    def _on_failure(self, error):
        if self.state == "HALF_OPEN":
            self.state = "OPEN"
            self.last_failure_time = time.time()
            print(f"Circuit Breaker: Failure in HALF_OPEN. Moving to OPEN state. Error: {error}")
        elif self.state == "CLOSED":
            self.failures += 1
            if self.failures >= self.failure_threshold:
                self.state = "OPEN"
                self.last_failure_time = time.time()
                print(f"Circuit Breaker: Failures exceeded threshold. Moving to OPEN state. Error: {error}")

class CircuitBreakerOpenError(Exception):
    pass

# Example usage with a simulated Didit AML Screening call
async def simulate_aml_screening():
    if random.random() < 0.7: # Simulate frequent failures
        raise ConnectionError("AML service unavailable")
    await asyncio.sleep(0.1)
    return {"aml_status": "clear"}

async def main():
    cb = CircuitBreaker()
    for i in range(20):
        try:
            print(f"--- Attempt {i+1} ---")
            result = await cb(simulate_aml_screening)
            print(f"AML Screening Success: {result}")
        except CircuitBreakerOpenError as e:
            print(f"Caught: {e}")
            await asyncio.sleep(1) # Wait a bit before next attempt if circuit is open
        except ConnectionError as e:
            print(f"Caught: {e}")
        await asyncio.sleep(0.5)

if __name__ == "__main__":
    asyncio.run(main())

Ce modèle est particulièrement utile pour les services critiques comme le screening AML de Didit ou les opérations de recherche faciale à grande échelle, où une dépendance défaillante pourrait impacter de nombreux utilisateurs.

Journalisation, surveillance et alertes complètes

Même avec des tentatives robustes et des disjoncteurs, des erreurs se produiront. La clé est de savoir quand elles se produisent, de comprendre pourquoi, et de réagir rapidement. La journalisation complète, la surveillance en temps réel et l'alerte proactive sont non négociables pour les flux de travail asynchrones.

• Journalisation structurée : Les messages de journal doivent être dans un format lisible par machine (par exemple, JSON) et inclure un contexte comme session_id, workflow_id, le nom du service, l'horodatage et le type d'erreur. Cela permet une agrégation et une analyse faciles.
• ID de corrélation : Attribuez un ID de corrélation unique à chaque demande de vérification d'identité à son point d'entrée et transmettez-le à tous les appels de service ultérieurs. Cela vous permet de suivre le parcours d'un seul utilisateur à travers un système distribué complexe, même lors de l'utilisation de services modulaires comme la vérification d'identité et l'estimation d'âge de Didit.
• Tableaux de bord de surveillance : Visualisez les métriques clés comme les taux de succès des API, la latence, les taux d'erreur et les longueurs de file d'attente pour chaque composant de votre flux de travail. Des outils comme Prometheus, Grafana ou les services de surveillance natifs du cloud sont inestimables.
• Alerting : Configurez des alertes pour les seuils critiques (par exemple, un taux d'erreur dépassant 5 % pendant 5 minutes, ou un service spécifique étant inaccessible). Les alertes doivent être envoyées à la bonne équipe via PagerDuty, Slack ou e-mail, permettant une action immédiate.

Par exemple, lorsqu'un utilisateur lance une session de vérification à l'aide des flux de travail orchestrés de Didit, un session_id est généré. Cet ID doit être capturé dans vos journaux pour chaque étape, de l'appel API initial à Didit au rappel webhook final avec les résultats de la vérification. En cas de problème, vous pouvez rapidement filtrer les journaux par cet session_id pour identifier le point de défaillance exact.

Comment Didit aide

Didit, en tant que plateforme d'identité native de l'IA et axée sur les développeurs, est conçue pour simplifier les flux de travail complexes de vérification d'identité, y compris leurs défis inhérents de gestion des erreurs. Notre architecture modulaire signifie que, bien que vous puissiez construire des solutions profondément personnalisées, une grande partie de la résilience sous-jacente est gérée pour vous.

• Flux de travail orchestrés : Le moteur de flux de travail sans code de Didit vous permet de définir des séquences de vérification complexes (par exemple, vérification d'identité + vivacité + screening AML) sans écrire de code étendu pour l'orchestration ou la gestion de l'état. Didit gère les nouvelles tentatives internes et les transitions d'état, réduisant considérablement le fardeau de la gestion des erreurs de votre côté.
• APIs et Webhooks robustes : Nos API claires sont conçues pour la fiabilité, et notre système de webhook fournit des mises à jour en temps réel sur l'état de la vérification. Didit gère la livraison de ces webhooks avec des mécanismes de nouvelle tentative intégrés, vous assurant de recevoir des mises à jour critiques même si votre endpoint est temporairement indisponible.
• KYC de base gratuit : Commencez avec la vérification d'identité essentielle, y compris la vérification d'identité (OCR, MRZ, codes-barres) et la vivacité passive et active, sans frais initiaux. Cela vous permet de mettre en œuvre une vérification robuste sans vous soucier de la résilience de l'infrastructure sous-jacente.
• Fiabilité native de l'IA : Nos systèmes basés sur l'IA sont intrinsèquement conçus pour une haute disponibilité et des performances élevées, minimisant les erreurs internes et fournissant des résultats cohérents pour des produits comme la correspondance faciale 1:1 et l'estimation d'âge.
• Données d'identité structurées : Tous les résultats de vérification sont fournis sous forme de données structurées, ce qui facilite le traitement des résultats par vos systèmes et la gestion programmatique des exceptions.

En tirant parti de la plateforme Didit, vous pouvez décharger une grande partie de la complexité de la gestion des erreurs asynchrones, en vous concentrant plutôt sur votre logique métier principale tout en garantissant une expérience de vérification d'identité fiable et sécurisée pour vos utilisateurs.

Prêt à commencer ?

Prêt à voir Didit en action ? Obtenez une démo gratuite dès aujourd'hui.

Commencez à vérifier les identités gratuitement avec le niveau gratuit de Didit.