下载地址：http://lanzou.com.cn/i096b4281

📁 output/zhuanzhangzhenghedamoxingai/
├── 📄 README.md204 B
├── 📄 pom.xml1.5 KB
├── 📄 package.json708 B
├── 📄 common/Loader.js2.6 KB
├── 📄 impl/Queue.ts3.1 KB
├── 📄 src/main/java/Repository.java6.7 KB
├── 📄 common/Controller.js3.9 KB
├── 📄 config/Util.xml1.6 KB
├── 📄 entity/Builder.py4.2 KB
├── 📄 src/main/java/Executor.java4.8 KB
├── 📄 common/Pool.go3.5 KB
├── 📄 graphql/Processor.py4.2 KB
├── 📄 config/Worker.json708 B
├── 📄 impl/Engine.ts2.5 KB
├── 📄 entity/Proxy.js3.9 KB
├── 📄 graphql/Handler.php3.3 KB
├── 📄 graphql/Manager.py5.8 KB
├── 📄 common/Wrapper.php2.8 KB
├── 📄 src/main/java/Resolver.java5.5 KB
├── 📄 src/main/java/Factory.java4.9 KB
├── 📄 common/Dispatcher.go3.1 KB
├── 📄 entity/Scheduler.js4.7 KB
├── 📄 common/Adapter.ts2.6 KB
├── 📄 src/main/java/Helper.java7.1 KB
├── 📄 common/Parser.js2.9 KB

编译入口目录：

Project Structure

Project : 转账凭证审核大模型ai

Folder : zhuanzhangzhenghedamoxingai

Files : 26

Size : 86.6 KB

Generated: 2026-03-22 18:19:45

zhuanzhangzhenghedamoxingai/
├── README.md [204 B]
├── common/
│ ├── Adapter.ts [2.6 KB]
│ ├── Controller.js [3.9 KB]
│ ├── Dispatcher.go [3.1 KB]
│ ├── Loader.js [2.6 KB]
│ ├── Parser.js [2.9 KB]
│ ├── Pool.go [3.5 KB]
│ └── Wrapper.php [2.8 KB]
├── config/
│ ├── Util.xml [1.6 KB]
│ └── Worker.json [708 B]
├── entity/
│ ├── Builder.py [4.2 KB]
│ ├── Proxy.js [3.9 KB]
│ └── Scheduler.js [4.7 KB]
├── graphql/
│ ├── Handler.php [3.3 KB]
│ ├── Manager.py [5.8 KB]
│ └── Processor.py [4.2 KB]
├── impl/
│ ├── Engine.ts [2.5 KB]
│ └── Queue.ts [3.1 KB]
├── package.json [708 B]
├── pom.xml [1.5 KB]
└── src/
├── main/
│ ├── java/
│ │ ├── Executor.java [4.8 KB]
│ │ ├── Factory.java [4.9 KB]
│ │ ├── Helper.java [7.1 KB]
│ │ ├── Repository.java [6.7 KB]
│ │ └── Resolver.java [5.5 KB]
│ └── resources/
└── test/
└── java/

============================================

转账凭证审核大模型AI (Lingo版)

模块: 核心审核引擎

功能: 基于规则与轻量级语义分析的转账凭证审核

============================================

import re
import json
from datetime import datetime
from typing import Dict, List, Tuple, Optional
from dataclasses import dataclass, asdict
from enum import Enum

------------------------------

1. 数据模型定义

------------------------------

class RiskLevel(Enum):
"""风险等级枚举"""
LOW = "低风险"
MEDIUM = "中风险"
HIGH = "高风险"
CRITICAL = "严重风险"

class AuditStatus(Enum):
"""审核状态枚举"""
PASS = "审核通过"
FLAG = "需人工复核"
REJECT = "审核拒绝"

@dataclass
class TransferVoucher:
"""转账凭证数据模型"""
voucher_id: str # 凭证编号
date: str # 交易日期 YYYY-MM-DD
payer_account: str # 付款方账号
payer_name: str # 付款方名称
payee_account: str # 收款方账号
payee_name: str # 收款方名称
amount: float # 交易金额
currency: str # 币种
purpose: str # 转账用途
approval_status: str # 审批状态 (已审批/未审批)
attachment_flag: bool # 是否有附件
historical_risk: bool # 历史风险记录

@dataclass
class AuditResult:
"""审核结果"""
voucher_id: str
status: AuditStatus
risk_level: RiskLevel
score: float # 综合评分 0-100
rules_violations: List[str] # 违反的规则列表
risk_factors: List[str] # 风险因素
suggestion: str # 审核建议
audit_time: str

------------------------------

2. 规则引擎 (模拟大模型规则层)

------------------------------

class RuleEngine:
"""基于规则的审核引擎，模拟大模型的部分确定性逻辑"""

# 高风险账号黑名单（模拟）
BLACKLIST_ACCOUNTS = {
    "6217000012345678": "涉诈账户",
    "6212000098765432": "高风险对公账户"
}

# 高风险用途关键词
RISK_PURPOSE_KEYWORDS = [
    "虚拟货币", "比特币", "数字货币", "网络赌博", "地下钱庄",
    "非法外汇", "洗钱", "诈骗", "传销"
]

# 可疑用途关键词（中风险）
SUSPICIOUS_PURPOSE = [
    "投资", "理财", "保证金", "贷款", "借款", "预付", "代理"
]

@staticmethod
def check_blacklist(account: str) -> Tuple[bool, str]:
    """检查黑名单"""
    if account in RuleEngine.BLACKLIST_ACCOUNTS:
        return True, RuleEngine.BLACKLIST_ACCOUNTS[account]
    return False, ""

@staticmethod
def check_amount_threshold(amount: float) -> Tuple[bool, RiskLevel]:
    """大额交易阈值检查"""
    if amount >= 500000:      # 50万以上
        return True, RiskLevel.HIGH
    elif amount >= 100000:    # 10万以上
        return True, RiskLevel.MEDIUM
    return False, RiskLevel.LOW

@staticmethod
def analyze_purpose(purpose: str) -> Tuple[RiskLevel, List[str]]:
    """分析转账用途风险"""
    risk_factors = []
    level = RiskLevel.LOW

    purpose_lower = purpose.lower()

    # 高风险关键词匹配
    for kw in RuleEngine.RISK_PURPOSE_KEYWORDS:
        if kw in purpose_lower:
            risk_factors.append(f"用途包含高风险词: {kw}")
            level = RiskLevel.HIGH
            break

    # 中风险关键词匹配（如果还没到高风险）
    if level != RiskLevel.HIGH:
        for kw in RuleEngine.SUSPICIOUS_PURPOSE:
            if kw in purpose_lower:
                risk_factors.append(f"用途包含敏感词: {kw}")
                level = RiskLevel.MEDIUM
                break

    # 空用途或过于简短
    if len(purpose.strip()) < 3:
        risk_factors.append("用途描述过于简略或缺失")
        if level == RiskLevel.LOW:
            level = RiskLevel.MEDIUM

    return level, risk_factors

@staticmethod
def check_approval(approval_status: str) -> Tuple[bool, str]:
    """检查审批状态"""
    if approval_status != "已审批":
        return False, "转账凭证未经有效审批"
    return True, ""

@staticmethod
def check_attachment(has_attachment: bool, amount: float) -> Tuple[bool, str]:
    """大额交易必须有附件"""
    if amount >= 50000 and not has_attachment:
        return False, "大额交易（≥5万）缺少支持性附件"
    return True, ""

------------------------------

3. 语义分析模块 (模拟大模型Lingo核心)

------------------------------

class LingoAnalyzer:
"""
模拟Lingo版大模型的语义分析能力
实际生产环境可替换为真实的LLM API调用
"""

def __init__(self):
    # 模拟的语义规则库
    self.semantic_patterns = {
        "关联交易": r"(关联方|子公司|母公司|兄弟公司|同一控制)",
        "循环转账": r"(循环|倒账|过桥|走账|回流)",
        "拆分交易": r"(拆分|分批|规避|限额)"
    }

def analyze_semantic_risk(self, voucher: TransferVoucher) -> Tuple[RiskLevel, List[str]]:
    """
    对凭证进行语义风险分析
    返回: (风险等级, 风险因素列表)
    """
    risk_factors = []
    risk_level = RiskLevel.LOW

    # 1. 分析名称与账户的语义匹配性（简单模拟）
    if voucher.payer_name == voucher.payee_name:
        risk_factors.append("付款方与收款方名称相同，可能存在自转风险")
        risk_level = RiskLevel.HIGH

    # 2. 分析用途中的语义模式
    purpose = voucher.purpose.lower()
    for pattern_name, pattern_regex in self.semantic_patterns.items():
        if re.search(pattern_regex, purpose, re.IGNORECASE):
            risk_factors.append(f"用途语义命中异常模式: {pattern_name}")
            if risk_level != RiskLevel.HIGH:
                risk_level = RiskLevel.MEDIUM

    # 3. 模拟的上下文异常检测（例如同一付款方短时间内多笔）
    # 实际场景中会结合历史数据，这里仅作示例
    if "紧急" in purpose and voucher.amount > 100000:
        risk_factors.append("紧急大额转账，需关注真实性")
        if risk_level == RiskLevel.LOW:
            risk_level = RiskLevel.MEDIUM

    # 4. 币种异常提醒（非人民币）
    if voucher.currency != "CNY":
        risk_factors.append("跨境或外币交易，需额外合规审查")
        if risk_level == RiskLevel.LOW:
            risk_level = RiskLevel.MEDIUM

    return risk_level, risk_factors

def generate_suggestion(self, risk_level: RiskLevel, violations: List[str]) -> str:
    """根据风险生成审核建议"""
    if risk_level == RiskLevel.CRITICAL:
        return "立即拒绝交易，上报反洗钱部门"
    elif risk_level == RiskLevel.HIGH:
        return "建议拒绝，需补充合规材料或重新审批"
    elif risk_level == RiskLevel.MEDIUM:
        return "转人工复核，重点关注用途真实性与关联关系"
    else:
        return "自动审核通过，建议抽查留档"

------------------------------

4. 综合审核器 (整合规则与语义)

------------------------------

class TransferAuditor:
"""转账凭证审核器 - 整合规则引擎与Lingo语义分析"""

def __init__(self):
    self.rule_engine = RuleEngine()
    self.lingo = LingoAnalyzer()

def audit(self, voucher: TransferVoucher) -> AuditResult:
    """
    执行完整审核流程
    """
    violations = []
    risk_factors = []
    total_risk_score = 0.0  # 分数越高风险越大

    # ---------- 1. 规则检查 ----------
    # 黑名单检查
    is_black, black_reason = self.rule_engine.check_blacklist(voucher.payer_account)
    if is_black:
        violations.append(f"付款方黑名单: {black_reason}")
        total_risk_score += 40

    is_black_payee, black_reason_payee = self.rule_engine.check_blacklist(voucher.payee_account)
    if is_black_payee:
        violations.append(f"收款方黑名单: {black_reason_payee}")
        total_risk_score += 40

    # 金额阈值
    over_threshold, amount_risk = self.rule_engine.check_amount_threshold(voucher.amount)
    if over_threshold:
        violations.append(f"交易金额超过阈值: {voucher.amount}元")
        risk_factors.append(f"大额交易({voucher.amount}元)")
        total_risk_score += 20 if amount_risk == RiskLevel.MEDIUM else 30

    # 用途分析（规则版）
    purpose_risk, purpose_factors = self.rule_engine.analyze_purpose(voucher.purpose)
    risk_factors.extend(purpose_factors)
    if purpose_risk == RiskLevel.HIGH:
        total_risk_score += 25
    elif purpose_risk == RiskLevel.MEDIUM:
        total_risk_score += 15

    # 审批检查
    approval_ok, approval_msg = self.rule_engine.check_approval(voucher.approval_status)
    if not approval_ok:
        violations.append(approval_msg)
        total_risk_score += 35

    # 附件检查
    attach_ok, attach_msg = self.rule_engine.check_attachment(voucher.attachment_flag, voucher.amount)
    if not attach_ok:
        violations.append(attach_msg)
        total_risk_score += 20

    # ---------- 2. 语义分析 (Lingo核心) ----------
    semantic_level, semantic_factors = self.lingo.analyze_semantic_risk(voucher)
    risk_factors.extend(semantic_factors)
    if semantic_level == RiskLevel.HIGH:
        total_risk_score += 30
    elif semantic_level == RiskLevel.MEDIUM:
        total_risk_score += 15

    # ---------- 3. 综合评分与决策 ----------
    # 分数归一化到0-100 (最大理论分约180，做上限100处理)
    final_score = min(100, total_risk_score)

    # 决策逻辑
    if final_score >= 70 or any("黑名单" in v for v in violations):
        status = AuditStatus.REJECT
        risk_level = RiskLevel.CRITICAL if final_score >= 85 else RiskLevel.HIGH
    elif final_score >= 40:
        status = AuditStatus.FLAG
        risk_level = RiskLevel.MEDIUM
    else:
        status = AuditStatus.PASS
        risk_level = RiskLevel.LOW

    # 生成建议
    suggestion = self.lingo.generate_suggestion(risk_level, violations)

    return AuditResult(
        voucher_id=voucher.voucher_id,
        status=status,
        risk_level=risk_level,
        score=final_score,
        rules_violations=violations,
        risk_factors=risk_factors,
        suggestion=suggestion,
        audit_time=datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    )

------------------------------

5. 测试与示例

------------------------------

def run_demo():
"""运行演示案例"""
print("=" 60)
print("转账凭证审核大模型AI (Lingo版) 演示")
print("=" 60)

# 创建审核器
auditor = TransferAuditor()

# 案例1: 正常低风险凭证
voucher1 = TransferVoucher(
    voucher_id="V20240001",
    date="2024-03-20",
    payer_account="6217000011111111",
    payer_name="XX科技有限公司",
    payee_account="6212000022222222",
    payee_name="YY商贸有限公司",
    amount=25000.00,
    currency="CNY",
    purpose="采购货款",
    approval_status="已审批",
    attachment_flag=True,
    historical_risk=False
)

# 案例2: 高风险凭证（黑名单+大额+可疑用途）
voucher2 = TransferVoucher(
    voucher_id="V20240002",
    date="2024-03-21",
    payer_account="6217000012345678",  # 黑名单账户
    payer_name="李某某",
    payee_account="6212000098765432",  # 黑名单账户
    payee_name="未知公司",
    amount=880000.00,
    currency="USD",
    purpose="虚拟货币投资，紧急",
    approval_status="未审批",
    attachment_flag=False,
    historical_risk=True
)

# 案例3: 中风险凭证（用途语义异常）
voucher3 = TransferVoucher(
    voucher_id="V20240003",
    date="2024-03-22",
    payer_account="6217000099999999",
    payer_name="王五",
    payee_account="6212000088888888",
    payee_name="王五",  # 同名
    amount=120000.00,
    currency="CNY",
    purpose="关联方资金拆借，循环走账",
    approval_status="已审批",
    attachment_flag=False,
    historical_risk=False
)

# 执行审核
results = []
for v in [voucher1, voucher2, voucher3]:
    result = auditor.audit(v)
    results.append(result)

# 输出结果
for r in results:
    print(f"\n凭证ID: {r.voucher_id}")
    print(f"审核状态: {r.status.value}")
    print(f"风险等级: {r.risk_level.value}")
    print(f"综合评分: {r.score}分")
    if r.rules_violations:
        print("违反规则:")
        for v in r.rules_violations:
            print(f"  - {v}")
    if r.risk_factors:
        print("风险因素:")
        for f in r.risk_factors:
            print(f"  - {f}")
    print(f"审核建议: {r.suggestion}")
    print(f"审核时间: {r.audit_time}")
    print("-" * 40)

# 模拟批量导出JSON结果
export_data = [asdict(r) for r in results]
with open("audit_results.json", "w", encoding="utf-8") as f:
    json.dump(export_data, f, ensure_ascii=False, indent=2)
print("\n审核结果已导出至 audit_results.json")

if name == "main":
run_demo()

text
┌─────────────────────────────────────────────────┐
│ 输入层 │
│ (转账凭证数据模型 TransferVoucher) │
└─────────────────────┬───────────────────────────┘
▼
┌─────────────────────────────────────────────────┐
│ 规则引擎 (RuleEngine) │
│ - 黑名单校验 - 金额阈值 - 用途关键词匹配 │
│ - 审批状态 - 附件合规 │
└─────────────────────┬───────────────────────────┘
▼
┌─────────────────────────────────────────────────┐
│ 语义分析模块 (LingoAnalyzer) │
│ - 关联交易识别 - 循环转账检测 │
│ - 拆分交易预警 - 上下文异常判断 │
└─────────────────────┬───────────────────────────┘
▼
┌─────────────────────────────────────────────────┐
│ 综合审核器 (TransferAuditor) │
│ - 评分加权 - 决策逻辑 │
│ - 结果生成 - 建议输出 │
└─────────────────────┬───────────────────────────┘
▼
┌─────────────────────────────────────────────────┐
│ 输出层 │
│ (AuditResult: 状态/等级/建议/风险因素) │
└─────────────────────────────────────────────────┘