비동기 신원 확인 워크플로우의 고급 오류 처리 기법 (KO)
비동기 신원 확인 시스템에서 강력한 오류 처리는 매우 중요합니다. 이 가이드는 재시도(백오프 포함), 서킷 브레이커, 포괄적인 로깅과 같은 Python 전략을 탐구하여 시스템의 복원력을 보장합니다.
강력한 재시도외부 신원 확인 서비스에 대한 API 호출 시 일시적인 오류에 대해 지수 백오프 및 지터(jitter)를 구현하여 시스템 과부하를 방지하고 성공률을 높입니다.
서킷 브레이커 패턴실패하는 서비스에 대한 요청을 일시적으로 중단하여 시스템이 복구될 시간을 벌어주고 전체 애플리케이션 안정성을 유지함으로써 연쇄적인 장애로부터 시스템을 보호합니다.
포괄적인 로깅 및 모니터링분산 비동기 신원 확인 파이프라인 내에서 문제를 신속하게 식별, 진단 및 해결하기 위해 구조화된 로깅, 상관 ID 및 실시간 모니터링을 활용합니다.
Didit의 내장된 복원력Didit의 AI 기반 모듈형 플랫폼은 오케스트레이션된 워크플로우와 강력한 API 설계를 제공하여 핵심 KYC, 생체 확인 및 AML 검사에 대한 복잡한 오류 처리를 추상화하여 신뢰성과 개발자 경험을 향상시킵니다.
신원 확인의 세계에서는 속도와 신뢰성이 가장 중요합니다. 기업이 성장함에 따라 비동기 워크플로우는 주 애플리케이션 스레드를 차단하지 않고 많은 양의 요청을 처리하는 데 필수적입니다. 그러나 이러한 분산 및 비차단 특성은 특히 오류 처리와 관련하여 상당한 복잡성을 야기합니다. 네트워크 문제, 서비스 중단, 데이터 불일치 및 예기치 않은 API 응답은 모두 신원 확인 프로세스를 방해하여 사용자 경험 저하, 규정 준수 위험 및 운영 비효율성으로 이어질 수 있습니다.
이 블로그 게시물은 비동기 신원 확인 워크플로우를 위한 고급 오류 처리 전략, 특히 Python 구현에 중점을 둡니다. 우리는 문제가 발생하더라도 확인 프로세스가 강력하게 유지되도록 보다 탄력적이고 내결함성 있는 시스템을 구축하는 방법을 탐구할 것입니다.
신원 확인의 비동기 오류 문제
비동기 신원 확인은 종종 여러 외부 서비스를 포함합니다. OCR 및 생체 확인을 위한 Didit과 같은 신원 확인 공급자, AML 심사 서비스, 주소 증명 데이터베이스, 그리고 잠재적으로 다른 데이터 소스들이 있습니다. 이러한 각 상호 작용은 잠재적인 실패 지점입니다. 작업이 훨씬 나중에 다른 프로세스에서 완료되거나 즉각적인 피드백 없이 조용히 실패할 수 있는 경우 전통적인 동기식 오류 처리(예: 간단한 try-except 블록)는 불충분합니다.
일반적인 KYC 워크플로우를 고려해봅시다. 사용자가 신분증을 업로드하고, 생체 확인이 수행된 다음, AML 심사가 시작됩니다. 생체 확인 서비스에 일시적인 네트워크 문제가 발생하면 즉시 재시도하는 것이 문제를 악화시킬 수 있습니다. AML 서비스가 완전히 중단된 경우 반복적인 시도는 리소스만 낭비하고 사용자 온보딩을 지연시킬 뿐입니다.
지수 백오프 및 지터(Jitter)를 사용한 강력한 재시도 구현
분산 시스템에서 가장 흔한 오류 유형 중 하나는 일시적인 장애입니다. 이는 네트워크 글리치, 서비스 과부하 오류 또는 데이터 경합과 같이 짧은 시간 후에 자체적으로 해결되는 임시적인 문제입니다. 장애 발생 직후 맹목적으로 재시도하면 어려움을 겪고 있는 서비스에 과부하를 주어 연쇄적인 장애로 이어질 수 있습니다. 해결책은 지수 백오프 및 지터(jitter)를 사용하는 지능형 재시도입니다.
지수 백오프는 재시도 간격 시간을 지수적으로 늘리는 것을 포함합니다. 예를 들어, 1초, 2초, 4초, 8초 등으로 대기 시간을 늘립니다. 이는 서비스가 복구될 시간을 제공합니다. 지터는 백오프 시간에 작고 무작위적인 지연을 추가하여 모든 클라이언트가 정확히 동시에 재시도하는 것을 방지하여 'thundering herd problem'을 일으킬 수 있는 상황을 막습니다.
import asyncio
import random
async def call_didit_api(data, attempt=0):
max_retries = 5
base_delay = 1 # seconds
try:
# Didit의 신원 확인 또는 생체 확인 서비스에 대한 API 호출 시뮬레이션
if random.random() < 0.6 and attempt < 3: # 일시적인 장애 시뮬레이션
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) # 지수 백오프 + 지터
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 # 모든 재시도가 실패하면 마지막 예외 다시 발생
async def main():
try:
# Didit의 신원 확인 예시 사용
result = await call_didit_api({"document_image": "base64_id_scan"})
print(f"Final result: {result}")
# Didit의 생체 확인 예시 사용
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())
이 패턴은 Didit의 신원 확인, 수동 및 능동 생체 확인, 또는 AML 심사 API를 포함한 외부 서비스와 통합할 때 매우 유용하며, 이 모든 것은 탄력적인 통신으로부터 이점을 얻습니다.
서킷 브레이커 패턴 구현
재시도는 일시적인 오류에 도움이 되지만, 서비스가 장기간 중단을 겪고 있는 경우 상황을 악화시킬 수 있습니다. 서킷 브레이커 패턴은 애플리케이션이 실패할 가능성이 있는 서비스를 반복적으로 호출하는 것을 방지합니다. 이 패턴은 실패를 모니터링하여, 주어진 시간 내에 실패가 특정 임계값을 초과하면 회로를 '트립'하여 실패하는 서비스에 대한 추가 호출을 차단합니다. 구성 가능한 시간 초과 후에는 '반개방(half-open)' 상태로 전환되어 몇 가지 테스트 요청을 허용하여 서비스가 복구되었는지 확인합니다.
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 # 닫힌 상태에서 성공 시 실패 횟수 초기화
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
# Didit AML 심사 호출 시뮬레이션을 사용한 예시 사용
async def simulate_aml_screening():
if random.random() < 0.7: # 잦은 실패 시뮬레이션
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) # 회로가 열려 있으면 다음 시도 전에 잠시 대기
except ConnectionError as e:
print(f"Caught: {e}")
await asyncio.sleep(0.5)
if __name__ == "__main__":
asyncio.run(main())
이 패턴은 Didit의 AML 심사 또는 대규모 얼굴 검색 작업과 같이 실패하는 종속성이 많은 사용자에게 영향을 미칠 수 있는 중요한 서비스에 특히 유용합니다.
포괄적인 로깅, 모니터링 및 경고
강력한 재시도와 서킷 브레이커가 있더라도 오류는 발생할 것입니다. 핵심은 오류가 언제 발생하는지, 왜 발생하는지 이해하고 신속하게 대응하는 것입니다. 포괄적인 로깅, 실시간 모니터링 및 사전 예방적 경고는 비동기 워크플로우에 필수적입니다.
- 구조화된 로깅: 로그 메시지는 기계가 읽을 수 있는 형식(예: JSON)이어야 하며
session_id,workflow_id, 서비스 이름, 타임스탬프 및 오류 유형과 같은 컨텍스트를 포함해야 합니다. 이를 통해 쉽게 집계 및 분석할 수 있습니다. - 상관 ID: 각 신원 확인 요청에 진입점에서 고유한 상관 ID를 할당하고 모든 후속 서비스 호출을 통해 전달합니다. 이를 통해 Didit의 신원 확인 및 연령 추정 모듈형 서비스와 같은 복잡하고 분산된 시스템을 통해 단일 사용자의 여정을 추적할 수 있습니다.
- 모니터링 대시보드: 각 워크플로우 구성 요소에 대한 API 성공률, 지연 시간, 오류율 및 큐 길이와 같은 주요 메트릭을 시각화합니다. Prometheus, Grafana 또는 클라우드 네이티브 모니터링 서비스와 같은 도구가 매우 유용합니다.
- 경고: 임계값(예: 5분 동안 오류율이 5%를 초과하거나 특정 서비스에 연결할 수 없는 경우)에 대한 경고를 설정합니다. 경고는 PagerDuty, Slack 또는 이메일을 통해 적절한 팀에 전달되어 즉각적인 조치를 가능하게 해야 합니다.
예를 들어, 사용자가 Didit의 오케스트레이션된 워크플로우를 사용하여 확인 세션을 시작할 때 session_id가 생성됩니다. 이 ID는 Didit에 대한 초기 API 호출부터 확인 결과가 포함된 최종 웹훅 콜백까지 모든 단계의 로그에 캡처되어야 합니다. 문제가 발생하면 이 session_id로 로그를 빠르게 필터링하여 정확한 실패 지점을 찾아낼 수 있습니다.
Didit이 도움이 되는 방법
Didit은 AI 기반의 개발자 우선 신원 플랫폼으로서, 고유한 오류 처리 문제를 포함하여 복잡한 신원 확인 워크플로우를 단순화하도록 설계되었습니다. 당사의 모듈형 아키텍처는 고도로 맞춤화된 솔루션을 구축할 수 있지만, 기본 복원력의 상당 부분이 자동으로 처리된다는 것을 의미합니다.
- 오케스트레이션된 워크플로우: Didit의 노코드 워크플로우 엔진을 사용하면 오케스트레이션 또는 상태 관리를 위한 광범위한 코드를 작성할 필요 없이 복잡한 확인 시퀀스(예: 신원 확인 + 생체 확인 + AML 심사)를 정의할 수 있습니다. Didit은 내부 재시도 및 상태 전환을 처리하여 오류 처리 부담을 크게 줄여줍니다.
- 강력한 API 및 웹훅: 당사의 깔끔한 API는 신뢰성을 위해 구축되었으며, 웹훅 시스템은 확인 상태에 대한 실시간 업데이트를 제공합니다. Didit은 내장된 재시도 메커니즘을 통해 이러한 웹훅 전달을 관리하여 엔드포인트가 일시적으로 사용 불가능하더라도 중요한 업데이트를 받을 수 있도록 합니다.
- 무료 핵심 KYC: 신원 확인(OCR, MRZ, 바코드) 및 수동 및 능동 생체 확인을 포함한 필수 신원 확인을 선불 비용 없이 시작할 수 있습니다. 이를 통해 기본 인프라의 복원력에 대해 걱정할 필요 없이 강력한 확인을 구현할 수 있습니다.
- AI 기반 안정성: 당사의 AI 기반 시스템은 본질적으로 높은 가용성과 성능을 위해 설계되어 내부 오류를 최소화하고 1:1 얼굴 매칭 및 연령 추정과 같은 제품에 대해 일관된 결과를 제공합니다.
- 구조화된 신원 데이터: 모든 확인 결과는 구조화된 데이터로 제공되므로 시스템에서 결과를 처리하고 예외를 프로그래밍 방식으로 처리하기가 더 쉽습니다.
Didit 플랫폼을 활용하면 비동기 오류 처리의 복잡성을 상당 부분 줄일 수 있으며, 사용자를 위한 안정적이고 안전한 신원 확인 경험을 보장하면서 핵심 비즈니스 로직에 집중할 수 있습니다.
시작할 준비가 되셨습니까?
Didit의 작동 방식을 확인하고 싶으신가요? 지금 무료 데모를 받아보세요.
Didit의 무료 티어로 무료로 신원 확인을 시작하세요.