实施多代理KYC系统

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

import uuid
import asyncio
import logging
from dataclasses import dataclass, field
from typing import Protocol, Optional, Literal, Dict, Any, List

# ----- 在代理之间传递的类型化结果 -----------------------------------
@dataclass(frozen=True)
class OnboardingResult:
    submitted_documents: Dict[str, Any]
    kyc_level: Literal["L1", "L2", "L3"]

@dataclass(frozen=True)
class VerificationResult:
    is_complete: bool
    normalized_data: Dict[str, Any]


@dataclass(frozen=True)
class RiskResult:
    risk_score: float
    drivers: List[str] = field(default_factory=list)  # 例如："PEP匹配"、"设备不匹配"

@dataclass(frozen=True)
class ComplianceDecision:
    status: Literal["approved", "manual_review", "rejected"]
    rationale: str
    case_id: str

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

# ----- 代理接口（协议）------------------------------------------

class OnboardingAgent(Protocol):
    async def determine_requirements(self, customer_data: Dict[str, Any], case_id: str)

class DocumentAgent(Protocol):
    async def process_documents(self, submitted_documents: Dict[str, Any], case_id: str)

class RiskAgent(Protocol):
    async def calculate_risk_profile(self, verification: VerificationResult, customer_data: Dict[str, Any], case_id: str
    )

class ComplianceAgent(Protocol):
    async def make_final_decision(self, verification: VerificationResult, risk: RiskResult, case_id: str
    )

class MonitoringAgent(Protocol):
    async def initialize_monitoring(self, customer_data: Dict[str, Any], risk_score: float, case_id: str)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76

# ----- 具有决策、超时和日志记录的协调器 ---------------------------
class KYCOrchestrator:
    def __init__(
        self,
        *,
        onboarding: OnboardingAgent,
        documents: DocumentAgent,
        risk: RiskAgent,
        compliance: ComplianceAgent,
        monitoring: MonitoringAgent,
            
        step_timeout_sec: float = 30.0,
        logger: Optional[logging.Logger] = None,

        ) -> None:

        self.onboarding = onboarding
        self.documents = documents
        self.risk = risk
        self.compliance = compliance
        self.monitoring = monitoring
        self.step_timeout_sec = step_timeout_sec
        self.log = logger or logging.getLogger("kyc.orchestrator")
    
    async def process_kyc_application(self, customer_data: Dict[str, Any]):

        case_id = str(uuid.uuid4())

        self.log.info("KYC案例已创建", extra={"case_id": case_id, "customer_id": customer_data.get("customer_id")})
        try:
            onboarding_result = await asyncio.wait_for(
                self.onboarding.determine_requirements(customer_data, case_id),
                timeout=self.step_timeout_sec,
            )
                
            self.log.debug("入职完成",
                extra={"case_id": case_id, "kyc_level": onboarding_result.kyc_level})
            verification_result = await asyncio.wait_for(
                self.documents.process_documents(onboarding_result.submitted_documents, case_id),
                timeout=self.step_timeout_sec,
            )
                
            self.log.debug("文档验证完成",

            risk_result = await asyncio.wait_for(
                self.risk.calculate_risk_profile(verification_result, customer_data, case_id),
                timeout=self.step_timeout_sec,
            )
                
            self.log.debug("风险评估完成",
                extra={"case_id": case_id, "risk_score": risk_result.risk_score})
            decision = await asyncio.wait_for(
                self.compliance.make_final_decision(verification_result, risk_result, case_id),
                timeout=self.step_timeout_sec,
            )
                
            self.log.info("合规决策",
                extra={"case_id": case_id, "status": decision.status, "rationale": decision.rationale})
            if decision.status == "approved":
                await self.monitoring.initialize_monitoring(customer_data, risk_result.risk_score, case_id)
            return decision
        except asyncio.TimeoutError as e:
            self.log.error("KYC流程步骤超时", extra={"case_id": case_id})
            return ComplianceDecision(
                status="manual_review",
                rationale="处理超时。路由至L3审核。",
                case_id=case_id,
            )
        except Exception as e:
            # 全面捕获以避免丢失案例；保持审计跟踪完整
            self.log.exception("KYC流程中出现未处理错误", extra={"case_id": case_id})
            return ComplianceDecision(
                status="manual_review",
                rationale=f"意外错误：{type(e).__name__}。路由至L3审核。",
                case_id=case_id,
            )