章节导语

AI的价值最终要落实到具体的行业场景中。

制造业引入AI进行预测性维护，一年省下数千万的设备维修费用；医疗行业用AI辅助影像诊断，帮助医生更早发现癌症；零售业用AI优化库存和推荐，年营收提升15%……这些不是想象，而是正在发生的现实。

本文选取制造业、医疗、零售、金融四个典型行业，深入讲解AI落地的实际方案、关键技术挑战、ROI评估方法，以及行业特有的合规要求。

一、制造业：智能制造与预测性维护

1.1 行业痛点与AI机会

制造业的核心痛点：设备意外停机造成的损失巨大。据估计，制造业每年因设备停机损失约500亿美元。

AI在制造业的主要应用：

预测性维护：通过传感器数据分析，预测设备故障，提前安排维修。

质量控制：用计算机视觉自动检测产品缺陷，比人工更准确更稳定。

工艺优化：AI分析生产数据，优化工艺参数，提升良率和效率。

供应链优化：预测需求，优化库存，减少积压和缺货。

1.2 预测性维护实战

import numpy as np
import pandas as pd
from typing import List, Dict, Tuple
from dataclasses import dataclass
import warnings
warnings.filterwarnings('ignore')

@dataclass
class SensorReading:
    timestamp: float
    temperature: float  # 摄氏度
    vibration: float    # 振动幅度 (mm/s)
    pressure: float     # 压力 (bar)
    rotation_speed: float  # 转速 (RPM)
    power_consumption: float  # 功耗 (kW)

class PredictiveMaintenanceSystem:
    """预测性维护系统"""
    
    def __init__(self):
        self.baseline_normal = None
        self.baseline_anomaly = None
        self.is_trained = False
    
    def preprocess_data(self, readings: List[SensorReading]) -> pd.DataFrame:
        """数据预处理"""
        df = pd.DataFrame([
            {
                'temperature': r.temperature,
                'vibration': r.vibration,
                'pressure': r.pressure,
                'rotation_speed': r.rotation_speed,
                'power_consumption': r.power_consumption
            }
            for r in readings
        ])
        
        # 添加时序特征
        df['temp_trend'] = df['temperature'].diff()
        df['vib_trend'] = df['vibration'].diff()
        df['rolling_mean_temp'] = df['temperature'].rolling(5).mean()
        df['rolling_std_vib'] = df['vibration'].rolling(5).std()
        
        return df.fillna(0)
    
    def train(self, normal_readings: List[SensorReading],
              anomaly_readings: List[SensorReading]):
        """训练异常检测模型"""
        normal_df = self.preprocess_data(normal_readings)
        anomaly_df = self.preprocess_data(anomaly_readings)
        
        # 计算正常状态的统计特性
        self.baseline_normal = {
            'temp_mean': normal_df['temperature'].mean(),
            'temp_std': normal_df['temperature'].std(),
            'vib_mean': normal_df['vibration'].mean(),
            'vib_std': normal_df['vibration'].std(),
            'power_mean': normal_df['power_consumption'].mean(),
            'power_std': normal_df['power_consumption'].std(),
        }
        
        # 计算异常状态的统计特性
        self.baseline_anomaly = {
            'temp_mean': anomaly_df['temperature'].mean(),
            'temp_std': anomaly_df['temperature'].std(),
            'vib_mean': anomaly_df['vibration'].mean(),
            'vib_std': anomaly_df['vibration'].std(),
        }
        
        self.is_trained = True
        print("模型训练完成")
        print(f"正常温度: {self.baseline_normal['temp_mean']:.1f}°C ± {self.baseline_normal['temp_std']:.1f}")
        print(f"异常温度: {self.baseline_anomaly['temp_mean']:.1f}°C ± {self.baseline_anomaly['temp_std']:.1f}")
    
    def detect_anomaly(self, reading: SensorReading) -> Tuple[bool, float, str]:
        """检测异常
        
        Returns:
            (是否异常, 风险分数, 原因)
        """
        if not self.is_trained:
            return False, 0.0, "模型未训练"
        
        # 计算各指标的风险分数
        temp_z = abs(reading.temperature - self.baseline_normal['temp_mean']) / max(self.baseline_normal['temp_std'], 0.1)
        vib_z = abs(reading.vibration - self.baseline_normal['vib_mean']) / max(self.baseline_normal['vib_std'], 0.1)
        power_z = abs(reading.power_consumption - self.baseline_normal['power_mean']) / max(self.baseline_normal['power_std'], 0.1)
        
        # 综合风险分数
        risk_score = min(1.0, (temp_z + vib_z + power_z) / 3 / 3)
        
        reasons = []
        if temp_z > 2:
            reasons.append(f"温度异常(偏高{reading.temperature:.1f}°C)")
        if vib_z > 2:
            reasons.append(f"振动异常(偏高{reading.vibration:.1f}mm/s)")
        if power_z > 2:
            reasons.append(f"功耗异常(偏高{reading.power_consumption:.1f}kW)")
        
        is_anomaly = risk_score > 0.7 or len(reasons) >= 2
        
        return is_anomaly, risk_score, "; ".join(reasons) if reasons else "正常"
    
    def predict_remaining_useful_life(self, readings: List[SensorReading]) -> float:
        """预测剩余使用寿命（天）"""
        if len(readings) < 10:
            return 30.0  # 数据不足，默认30天
        
        df = self.preprocess_data(readings)
        
        # 简化实现：基于温度趋势预测
        recent_temps = df['temperature'].tail(10)
        temp_increase_rate = recent_temps.diff().mean()
        
        if temp_increase_rate <= 0:
            return 90.0  # 温度稳定或下降，剩余寿命长
        
        # 计算达到异常阈值还需要多少天
        current_temp = recent_temps.iloc[-1]
        temp_threshold = self.baseline_normal['temp_mean'] + 3 * self.baseline_normal['temp_std']
        
        days_to_failure = (temp_threshold - current_temp) / temp_increase_rate
        
        return max(1, min(90, days_to_failure))

# 使用示例
def main():
    # 模拟正常数据
    np.random.seed(42)
    normal_readings = []
    for i in range(100):
        normal_readings.append(SensorReading(
            timestamp=i,
            temperature=65 + np.random.randn() * 3,
            vibration=2.5 + np.random.randn() * 0.5,
            pressure=10 + np.random.randn() * 0.2,
            rotation_speed=3000 + np.random.randn() * 50,
            power_consumption=150 + np.random.randn() * 5
        ))
    
    # 模拟异常数据（温度和振动偏高）
    anomaly_readings = []
    for i in range(50):
        anomaly_readings.append(SensorReading(
            timestamp=i,
            temperature=80 + np.random.randn() * 5,
            vibration=4.0 + np.random.randn() * 1,
            pressure=10 + np.random.randn() * 0.2,
            rotation_speed=3000 + np.random.randn() * 50,
            power_consumption=170 + np.random.randn() * 8
        ))
    
    # 训练
    system = PredictiveMaintenanceSystem()
    system.train(normal_readings, anomaly_readings)
    
    # 测试
    print("\n=== 异常检测 ===")
    test_reading = SensorReading(
        timestamp=0,
        temperature=78,  # 偏高
        vibration=3.8,    # 偏高
        pressure=10.1,
        rotation_speed=3050,
        power_consumption=165
    )
    
    is_anomaly, risk, reason = system.detect_anomaly(test_reading)
    print(f"检测结果: {'异常' if is_anomaly else '正常'}")
    print(f"风险分数: {risk:.2%}")
    print(f"原因: {reason}")
    
    # 预测剩余寿命
    rul = system.predict_remaining_useful_life(normal_readings[-20:] + [test_reading])
    print(f"\n预测剩余使用寿命: {rul:.0f} 天")

main()

1.3 质量控制实战

import numpy as np
from typing import List, Tuple
from dataclasses import dataclass

@dataclass
class DefectInfo:
    """缺陷信息"""
    defect_type: str
    severity: str  # critical/major/minor
    location: Tuple[int, int]  # x, y坐标
    confidence: float

class VisualQualityInspector:
    """视觉质量检测系统"""
    
    def __init__(self, model=None):
        self.model = model  # 实际使用中加载CNN模型
        self.defect_types = ['划痕', '凹坑', '裂纹', '污渍', '变形']
    
    def preprocess_image(self, image: np.ndarray) -> np.ndarray:
        """图像预处理"""
        # 调整大小
        image = self.resize_image(image, (224, 224))
        
        # 归一化
        image = image.astype(np.float32) / 255.0
        
        # 标准化
        mean = np.array([0.485, 0.456, 0.406])
        std = np.array([0.229, 0.224, 0.225])
        image = (image - mean) / std
        
        return image
    
    def resize_image(self, image: np.ndarray, size: Tuple[int, int]) -> np.ndarray:
        """调整图像大小"""
        # 简化实现
        from scipy.ndimage import zoom
        factors = (size[0]/image.shape[0], size[1]/image.shape[1], 1)
        return zoom(image, factors)
    
    def detect_defects(self, image: np.ndarray) -> List[DefectInfo]:
        """检测缺陷"""
        if self.model is None:
            # 简化实现：模拟检测结果
            np.random.seed(int(np.mean(image) * 1000) % 100)
            
            if np.random.random() > 0.7:
                # 有缺陷
                n_defects = np.random.randint(1, 4)
                defects = []
                
                for _ in range(n_defects):
                    defects.append(DefectInfo(
                        defect_type=np.random.choice(self.defect_types),
                        severity=np.random.choice(['critical', 'major', 'minor'], p=[0.1, 0.3, 0.6]),
                        location=(np.random.randint(0, 224), np.random.randint(0, 224)),
                        confidence=np.random.uniform(0.7, 0.99)
                    ))
                
                return defects
            else:
                return []
        
        # 实际使用中调用模型
        # ...
        return []
    
    def classify_defect_severity(self, defect_size: float, defect_type: str) -> str:
        """判断缺陷严重程度"""
        # 缺陷面积阈值（像素）
        thresholds = {
            '划痕': {'critical': 100, 'major': 50},
            '凹坑': {'critical': 80, 'major': 40},
            '裂纹': {'critical': 60, 'major': 30},
            '污渍': {'critical': 150, 'major': 80},
            '变形': {'critical': 40, 'major': 20},
        }
        
        thresh = thresholds.get(defect_type, {'critical': 100, 'major': 50})
        
        if defect_size >= thresh['critical']:
            return 'critical'
        elif defect_size >= thresh['major']:
            return 'major'
        else:
            return 'minor'
    
    def make_decision(self, defects: List[DefectInfo]) -> str:
        """做出质检决策"""
        if not defects:
            return "PASS"  # 无缺陷，通过
        
        # 检查是否有严重缺陷
        has_critical = any(d.severity == 'critical' for d in defects)
        has_major = any(d.severity == 'major' for d in defects)
        
        if has_critical:
            return "FAIL_CRITICAL"  # 严重缺陷，不合格
        elif has_major and len(defects) >= 2:
            return "FAIL_MAJOR"  # 多个主要缺陷，不合格
        elif len(defects) >= 5:
            return "FAIL_MINOR"  # 缺陷过多，不合格
        else:
            return "PASS_WITH_DEFECTS"  # 有缺陷但可接受

# 使用示例
inspector = VisualQualityInspector()

# 模拟一张产品图像
fake_image = np.random.randint(0, 255, (480, 640, 3), dtype=np.uint8)

# 检测
defects = inspector.detect_defects(fake_image)
decision = inspector.make_decision(defects)

print(f"检测到 {len(defects)} 个缺陷")
for d in defects:
    print(f"  - {d.defect_type}: {d.severity}, 置信度{d.confidence:.0%}")
print(f"\n质检决策: {decision}")

二、医疗行业：AI辅助诊断

2.1 行业背景与监管

医疗AI的特殊性：

监管严格：医疗AI属于医疗器械，需要获得药监局审批。

容错率极低：诊断错误可能危及生命。

数据隐私：患者隐私必须严格保护。

可解释性要求：医生需要理解AI的判断依据。

2.2 医疗影像AI辅助诊断

from typing import List, Dict, Tuple, Optional
from dataclasses import dataclass
from enum import Enum
import numpy as np

class Modality(Enum):
    X_RAY = "x_ray"
    CT = "ct"
    MRI = "mri"
    ULTRASOUND = "ultrasound"

class BodyPart(Enum):
    CHEST = "chest"
    BRAIN = "brain"
    ABDOMEN = "abdomen"
    BONE = "bone"

@dataclass
class AIDiagnosisResult:
    """AI诊断结果"""
    finding: str  # 发现的征象
    diagnosis: str  # 诊断意见
    confidence: float  # 置信度
    bbox: Optional[Tuple[int, int, int, int]] = None  # 病灶位置
    recommendation: str = ""  # 建议

class MedicalImageAI:
    """医疗影像AI辅助诊断系统"""
    
    def __init__(self):
        self.model = None  # 加载训练好的模型
        self.supported_findings = {
            Modality.X_RAY: ['肺炎', '肺结节', '气胸', '胸腔积液', '骨折'],
            Modality.CT: ['脑出血', '脑梗死', '肺肿瘤', '肝囊肿'],
            Modality.MRI: ['脑肿瘤', '脊髓病变', '关节损伤'],
        }
    
    def preprocess(self, image: np.ndarray, modality: Modality) -> np.ndarray:
        """预处理"""
        # 调整大小
        image = self.resize_to_standard(image, modality)
        
        # 归一化
        image = image.astype(np.float32)
        
        # 窗宽窗位调整（CT/MRI特有）
        if modality == Modality.CT:
            image = self.apply_window(image, window_center=40, window_width=400)
        
        return image
    
    def resize_to_standard(self, image: np.ndarray, modality: Modality) -> np.ndarray:
        """调整到标准尺寸"""
        standards = {
            Modality.X_RAY: (512, 512),
            Modality.CT: (512, 512, 64),  # 3D
            Modality.MRI: (256, 256, 128),
            Modality.ULTRASOUND: (640, 480),
        }
        # 简化实现
        return image
    
    def apply_window(self, image: np.ndarray, window_center: float, window_width: float) -> np.ndarray:
        """窗宽窗位调整"""
        min_val = window_center - window_width / 2
        max_val = window_center + window_width / 2
        
        image = np.clip(image, min_val, max_val)
        image = (image - min_val) / (max_val - min_val)
        
        return image
    
    def analyze(self, image: np.ndarray, modality: Modality, 
                body_part: BodyPart) -> List[AIDiagnosisResult]:
        """分析影像"""
        # 预处理
        processed = self.preprocess(image, modality)
        
        # 获取可疑区域
        regions = self.detect_suspicious_regions(processed)
        
        # 分析每个区域
        results = []
        for region in regions:
            finding = self.classify_region(region, modality)
            if finding:
                confidence = self.calculate_confidence(region, finding)
                
                results.append(AIDiagnosisResult(
                    finding=finding,
                    diagnosis=self.generate_diagnosis(finding, body_part),
                    confidence=confidence,
                    bbox=self.extract_bbox(region),
                    recommendation=self.get_recommendation(finding, confidence)
                ))
        
        # 按置信度排序
        results.sort(key=lambda x: x.confidence, reverse=True)
        
        return results
    
    def detect_suspicious_regions(self, image: np.ndarray) -> List[np.ndarray]:
        """检测可疑区域"""
        # 简化实现：返回模拟结果
        if np.random.random() > 0.7:
            return [np.random.randn(50, 50)]
        return []
    
    def classify_region(self, region: np.ndarray, modality: Modality) -> Optional[str]:
        """分类区域"""
        findings = self.supported_findings.get(modality, [])
        if not findings:
            return None
        
        # 简化：随机选择一个
        return np.random.choice(findings) if np.random.random() > 0.5 else None
    
    def calculate_confidence(self, region: np.ndarray, finding: str) -> float:
        """计算置信度"""
        # 简化实现
        return np.random.uniform(0.6, 0.95)
    
    def extract_bbox(self, region: np.ndarray) -> Tuple[int, int, int, int]:
        """提取边界框"""
        # 返回 (x, y, w, h)
        return (0, 0, 50, 50)
    
    def generate_diagnosis(self, finding: str, body_part: BodyPart) -> str:
        """生成诊断意见"""
        templates = {
            '肺炎': f'双肺{finding}，建议结合临床症状进一步检查',
            '肺结节': f'右肺见{finding}，建议定期复查或进一步CT检查',
            '脑出血': f'颅内{finding}，建议急诊处理',
            '脑梗死': f'脑部{finding}，建议MRI进一步检查',
        }
        return templates.get(finding, f'发现{finding}，建议进一步检查')
    
    def get_recommendation(self, finding: str, confidence: float) -> str:
        """获取建议"""
        if confidence > 0.9:
            return "高置信度，建议优先处理"
        elif confidence > 0.7:
            return "建议结合临床信息综合判断"
        else:
            return "建议进一步检查确认"

# 使用示例
ai = MedicalImageAI()

# 模拟CT图像
fake_ct = np.random.randint(0, 255, (512, 512, 64), dtype=np.uint8)

# 分析
results = ai.analyze(fake_ct, Modality.CT, BodyPart.BRAIN)

print(f"发现 {len(results)} 个可疑区域")
for r in results:
    print(f"\n征象: {r.finding}")
    print(f"诊断: {r.diagnosis}")
    print(f"置信度: {r.confidence:.1%}")
    print(f"建议: {r.recommendation}")

三、零售行业：智能推荐与库存优化

3.1 行业痛点与AI机会

零售行业的核心挑战：

库存管理：积压和缺货都会造成损失。

客户获取成本高：获客难，留客更难。

个性化需求：消费者期望个性化的服务和推荐。

3.2 智能推荐系统

import numpy as np
from typing import List, Dict, Tuple
from dataclasses import dataclass

@dataclass
class User:
    user_id: str
    features: np.ndarray  # 用户特征向量

@dataclass
class Product:
    product_id: str
    category: str
    features: np.ndarray  # 商品特征向量
    price: float
    popularity: float  # 热度

class RecommenderSystem:
    """智能推荐系统"""
    
    def __init__(self):
        self.user_embeddings: Dict[str, np.ndarray] = {}
        self.product_embeddings: Dict[str, np.ndarray] = {}
        self.user_interactions: Dict[str, List[str]] = {}  # user_id -> [product_ids]
    
    def train(self, users: List[User], products: List[Product], interactions: List[Tuple[str, str]]):
        """训练推荐模型"""
        # 保存用户和商品embedding
        for user in users:
            self.user_embeddings[user.user_id] = user.features
        
        for product in products:
            self.product_embeddings[product.product_id] = product.features
        
        # 记录交互历史
        for user_id, product_id in interactions:
            if user_id not in self.user_interactions:
                self.user_interactions[user_id] = []
            self.user_interactions[user_id].append(product_id)
        
        print(f"训练完成: {len(users)} 用户, {len(products)} 商品, {len(interactions)} 交互")
    
    def recommend(self, user_id: str, n: int = 10) -> List[Tuple[str, float]]:
        """为用户推荐商品
        
        Returns:
            [(product_id, score), ...]
        """
        if user_id not in self.user_embeddings:
            return self.recommend_popular(n)
        
        user_emb = self.user_embeddings[user_id]
        
        # 计算用户对所有商品的分数
        scores = []
        for product_id, product_emb in self.product_embeddings.items():
            # 协同过滤分数
            cf_score = self._compute_cf_score(user_id, product_id)
            
            # 内容相似度分数
            content_score = np.dot(user_emb, product_emb) / (np.linalg.norm(user_emb) * np.linalg.norm(product_emb) + 1e-8)
            
            # 综合分数
            final_score = 0.7 * cf_score + 0.3 * content_score
            
            scores.append((product_id, final_score))
        
        # 排序
        scores.sort(key=lambda x: x[1], reverse=True)
        
        # 过滤掉用户已经交互过的
        interacted = set(self.user_interactions.get(user_id, []))
        scores = [(p, s) for p, s in scores if p not in interacted][:n]
        
        return scores
    
    def _compute_cf_score(self, user_id: str, product_id: str) -> float:
        """计算协同过滤分数"""
        if user_id not in self.user_interactions:
            return 0.5
        
        # 简化：使用商品热度
        product_emb = self.product_embeddings.get(product_id)
        if product_emb is None:
            return 0.5
        
        # 计算与用户历史商品的相似度
        user_history = self.user_interactions[user_id]
        similarities = []
        
        for hist_product_id in user_history[-10:]:  # 最近10个
            hist_emb = self.product_embeddings.get(hist_product_id)
            if hist_emb is not None:
                sim = np.dot(product_emb, hist_emb) / (np.linalg.norm(product_emb) * np.linalg.norm(hist_emb) + 1e-8)
                similarities.append(sim)
        
        return np.mean(similarities) if similarities else 0.5
    
    def recommend_popular(self, n: int = 10) -> List[Tuple[str, float]]:
        """推荐热门商品"""
        popular = sorted(self.product_embeddings.items(), 
                        key=lambda x: np.random.random(), reverse=True)[:n]
        return [(p, 0.5) for p, _ in popular]

# 使用示例
np.random.seed(42)

# 创建模拟用户
users = [
    User(f"user_{i}", np.random.randn(50))
    for i in range(100)
]

# 创建模拟商品
categories = ['电子产品', '服装', '食品', '图书', '美妆']
products = [
    Product(f"prod_{i}", np.random.choice(categories), np.random.randn(50), 
           np.random.uniform(10, 1000), np.random.random())
    for i in range(500)
]

# 创建模拟交互
interactions = [
    (f"user_{np.random.randint(0, 100)}", f"prod_{np.random.randint(0, 500)}")
    for _ in range(5000)
]

# 训练
rec = RecommenderSystem()
rec.train(users, products, interactions)

# 推荐
recommendations = rec.recommend("user_0", n=5)

print("\n为 user_0 的推荐:")
for product_id, score in recommendations:
    product = next(p for p in products if p.product_id == product_id)
    print(f"  {product_id} ({product.category}): {score:.3f}")

四、金融行业：智能风控

4.1 行业痛点

金融行业的核心挑战：

欺诈风险：每年因欺诈损失巨大。

信用评估：需要准确评估借款人信用。

合规要求：监管要求严格，需要可解释性。

4.2 智能风控实战

from typing import List, Dict, Tuple, Optional
from dataclasses import dataclass
from enum import Enum
import numpy as np

class RiskLevel(Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"

@dataclass
class Transaction:
    transaction_id: str
    user_id: str
    amount: float
    merchant_category: str
    location: Tuple[str, str]  # (城市, 国家)
    timestamp: float
    is_online: bool

@dataclass
class RiskAssessment:
    risk_level: RiskLevel
    risk_score: float
    reasons: List[str]
    recommended_action: str  # approve/deny/review

class AntiFraudSystem:
    """反欺诈系统"""
    
    def __init__(self):
        self.user_profiles: Dict[str, Dict] = {}
        self.fraud_rules: List[Dict] = []
        self.ml_model = None  # ML模型
    
    def load_user_profile(self, user_id: str, profile: Dict):
        """加载用户画像"""
        self.user_profiles[user_id] = profile
    
    def add_rule(self, name: str, condition: callable, risk_delta: float):
        """添加规则"""
        self.fraud_rules.append({
            'name': name,
            'condition': condition,
            'risk_delta': risk_delta
        })
    
    def assess_transaction(self, txn: Transaction) -> RiskAssessment:
        """评估交易风险"""
        risk_score = 0.0
        reasons = []
        
        # 1. 规则引擎
        for rule in self.fraud_rules:
            if rule['condition'](txn):
                risk_score += rule['risk_delta']
                reasons.append(rule['name'])
        
        # 2. 用户历史检查
        if txn.user_id in self.user_profiles:
            profile = self.user_profiles[txn.user_id]
            
            # 检查是否是异常金额
            avg_amount = profile.get('avg_amount', txn.amount)
            if txn.amount > avg_amount * 5:
                risk_score += 0.3
                reasons.append("异常大额交易")
            
            # 检查是否是异常地点
            usual_locations = profile.get('usual_locations', set())
            if txn.location not in usual_locations:
                risk_score += 0.2
                reasons.append("异常交易地点")
            
            # 检查是否是异常时间
            usual_hours = profile.get('usual_hours', set())
            txn_hour = int(txn.timestamp % 24)
            if txn_hour not in usual_hours:
                risk_score += 0.1
                reasons.append("异常交易时间")
        
        # 3. ML模型评分
        if self.ml_model:
            ml_score = self.ml_model.predict(txn)
            risk_score = 0.7 * risk_score + 0.3 * ml_score
        
        # 4. 综合判断
        risk_score = min(1.0, risk_score)
        
        if risk_score >= 0.8:
            risk_level = RiskLevel.CRITICAL
            action = "deny"
        elif risk_score >= 0.6:
            risk_level = RiskLevel.HIGH
            action = "deny"
        elif risk_score >= 0.4:
            risk_level = RiskLevel.MEDIUM
            action = "review"
        else:
            risk_level = RiskLevel.LOW
            action = "approve"
        
        return RiskAssessment(
            risk_level=risk_level,
            risk_score=risk_score,
            reasons=reasons,
            recommended_action=action
        )

# 使用示例
fraud_system = AntiFraudSystem()

# 添加规则
fraud_system.add_rule(
    "大额交易",
    lambda t: t.amount > 10000,
    0.3
)

fraud_system.add_rule(
    "异常商户",
    lambda t: t.merchant_category in ['赌博', '虚拟货币'],
    0.4
)

fraud_system.add_rule(
    "跨境交易",
    lambda t: t.location[1] != '中国',
    0.2
)

# 加载用户画像
fraud_system.load_user_profile("user_123", {
    'avg_amount': 500,
    'usual_locations': {('北京', '中国'), ('上海', '中国')},
    'usual_hours': set(range(8, 23))
})

# 测试交易
test_txn = Transaction(
    transaction_id="txn_001",
    user_id="user_123",
    amount=5000,
    merchant_category="电子产品",
    location=("深圳", "中国"),
    timestamp=14.5,  # 下午2点
    is_online=False
)

result = fraud_system.assess_transaction(test_txn)

print(f"风险等级: {result.risk_level.value}")
print(f"风险分数: {result.risk_score:.1%}")
print(f"原因: {', '.join(result.reasons)}")
print(f"建议: {result.recommended_action}")

五、ROI评估与落地建议

5.1 AI项目ROI评估框架

5.2 行业落地建议

制造业：从预测性维护切入，ROI明显，数据基础好。

医疗：从影像辅助诊断入手，但要注意监管合规。

零售：从推荐和库存优化开始，数据丰富，见效快。

金融：反欺诈和信用评估ROI最高，但监管严格。

六、总结

行业AI落地的关键是找准场景。不是所有场景都适合AI，要找ROI明确、数据可得、监管允许的场景。

行业知识比技术更重要。AI工程师必须深入理解行业，才能设计出真正解决问题的方案。

合规是前提。医疗、金融等强监管行业，合规审查必须前置。

持续优化是关键。AI系统上线后需要持续优化，才能保持效果。

延伸阅读

• 麦肯锡AI行业报告
• Gartner行业AI成熟度模型
• 各行业AI最佳实践案例

课后练习

基础题：选择一个你熟悉的行业，分析AI可以如何落地。

进阶题：为一个具体场景设计完整的AI解决方案。

挑战题：评估你选定方案的ROI和实施风险。