确定性 AI 系统实现：架构设计与技术路径

from typing import List, Dict, Any, Tuple
from pydantic import BaseModel

# 规则接口
class Rule(BaseModel):
    def evaluate(self, data: Any) -> bool:
        """评估数据是否符合规则"""
        pass

    def get_description(self) -> str:
        """获取规则描述"""
        pass

# 具体规则实现
class MinAgeRule(Rule):
    min_age: int = 18

    def evaluate(self, employee: Dict) -> bool:
        return employee.get("age", 0) >= self.min_age

    def get_description(self) -> str:
        return f"员工年龄必须至少为{self.min_age}岁"

# 推理引擎
class RulesEngine:
    def __init__(self, rules: List[Rule]):
        self.rules = rules
        self.trace = []

    def validate(self, data: Any) -> Tuple[bool, List[str]]:
        """验证数据是否符合所有规则"""
        self.trace = []
        violations = []

        for rule in self.rules:
            self.trace.append(f"评估规则: {rule.get_description()}")
            if not rule.evaluate(data):
                violations.append(rule.get_description())
                self.trace.append(f"违反规则: {rule.get_description()}")
            else:
                self.trace.append(f"通过规则: {rule.get_description()}")

        return len(violations) == 0, violations

    def get_explanation(self) -> List[str]:
        """获取推理过程的解释"""
        return self.trace

from typing import Dict, List, Callable, Any, Optional
from enum import Enum, auto

# 状态枚举
class CustomerServiceState(Enum):
    INITIALIZE = auto()
    GREETING = auto()
    IDENTIFY_NEEDS = auto()
    PROVIDE_INFO = auto()
    HANDLE_COMPLAINT = auto()
    RECOMMEND_PRODUCT = auto()
    CONFIRM_SATISFACTION = auto()
    ESCALATE = auto()
    END_SERVICE = auto()

# 事件枚举
class CustomerServiceEvent(Enum):
    START = auto()
    GREETED = auto()
    NEED_IDENTIFIED = auto()
    INFO_REQUESTED = auto()
    COMPLAINT_REPORTED = auto()
    PRODUCT_INQUIRED = auto()
    SERVICE_COMPLETED = auto()
    CUSTOMER_SATISFIED = auto()
    CUSTOMER_UNSATISFIED = auto()
    ESCALATION_COMPLETED = auto()

# 状态机实现
class StateMachine:
    def __init__(self):
        self.current_state = None
        self.states = {}
        self.transitions = {}
        self.history = []

    def add_state(self, state: Enum, entry_action: Optional[Callable] = None,
                 exit_action: Optional[Callable] = None):
        """添加状态"""
        self.states[state] = {
            'entry_action': entry_action,
            'exit_action': exit_action
        }

    def add_transition(self, from_state: Enum, event: Enum, to_state: Enum):
        """添加状态转换规则"""
        if from_state not in self.transitions:
            self.transitions[from_state] = {}
        self.transitions[from_state][event] = to_state

    def initialize(self, initial_state: Enum):
        """初始化状态机"""
        self.current_state = initial_state
        self.history.append(initial_state)
        entry_action = self.states[initial_state]['entry_action']
        if entry_action:
            entry_action()

    def transition(self, event: Enum) -> bool:
        """执行状态转换"""
        if self.current_state not in self.transitions or event not in self.transitions[self.current_state]:
            return False

        # 执行当前状态的退出动作
        exit_action = self.states[self.current_state]['exit_action']
        if exit_action:
            exit_action()

        # 状态转换
        old_state = self.current_state
        self.current_state = self.transitions[old_state][event]
        self.history.append(self.current_state)

        # 执行新状态的进入动作
        entry_action = self.states[self.current_state]['entry_action']
        if entry_action:
            entry_action()

        return True

    def get_current_state(self) -> Enum:
        """获取当前状态"""
        return self.current_state

    def get_available_events(self) -> List[Enum]:
        """获取当前可用的事件"""
        if self.current_state not in self.transitions:
            return []
        return list(self.transitions[self.current_state].keys())

    def get_history(self) -> List[Enum]:
        """获取状态历史"""
        return self.history

# 不良实践：复杂、难以理解的规则
def complex_rule(data):
    if data.get('age') >= 18 and data.get('country') in ['US', 'CA'] and \
       (data.get('role') == 'manager' or (data.get('role') == 'worker' and data.get('experience') > 5)) and \
       data.get('certification') is not None:
        return True
    return False

# 最佳实践：原子性、可读性好的规则
class AgeRule(Rule):
    min_age: int = 18

    def evaluate(self, data: Dict) -> bool:
        return data.get('age', 0) >= self.min_age

    def get_description(self) -> str:
        return f"年龄必须至少为{self.min_age}岁"

class CountryRule(Rule):
    allowed_countries: List[str] = ['US', 'CA']

    def evaluate(self, data: Dict) -> bool:
        return data.get('country', '') in self.allowed_countries

    def get_description(self) -> str:
        return f"国家必须是以下之一: {', '.join(self.allowed_countries)}"

# 规则冲突检测示例
def detect_conflicts(rules: List[Rule]) -> List[Tuple[Rule, Rule]]:
    conflicts = []
    for i, rule1 in enumerate(rules):
        for rule2 in rules[i+1:]:
            if is_conflicting(rule1, rule2):
                conflicts.append((rule1, rule2))
    return conflicts

def is_conflicting(rule1: Rule, rule2: Rule) -> bool:
    # 实现冲突检测逻辑
    # 例如，检查两个规则是否对相同条件有不同结论
    pass

# 解释生成示例
class Explanation:
    def __init__(self):
        self.steps = []
        self.rules_applied = []
        self.key_factors = []

    def add_step(self, step_description: str):
        self.steps.append(step_description)

    def add_rule_applied(self, rule: Rule, result: bool):
        self.rules_applied.append({
            'rule': rule.get_description(),
            'result': result
        })

    def add_key_factor(self, factor: str, importance: float):
        self.key_factors.append({
            'factor': factor,
            'importance': importance
        })

    def generate_natural_language(self) -> str:
        """生成自然语言解释"""
        explanation = "系统决策过程如下：\n\n"

        # 添加步骤说明
        explanation += "处理步骤：\n"
        for i, step in enumerate(self.steps):
            explanation += f"{i+1}. {step}\n"
        explanation += "\n"

        # 添加规则应用说明
        explanation += "应用的规则：\n"
        for rule_info in self.rules_applied:
            result = "通过" if rule_info['result'] else "未通过"
            explanation += f"- {rule_info['rule']}: {result}\n"
        explanation += "\n"

        # 添加关键因素说明
        if self.key_factors:
            explanation += "关键影响因素：\n"
            sorted_factors = sorted(self.key_factors, key=lambda x: x['importance'], reverse=True)
            for factor in sorted_factors:
                explanation += f"- {factor['factor']}: 重要性 {factor['importance']:.2f}\n"

        return explanation

# 可解释性评估示例
def evaluate_explanation(explanation: str, ground_truth: Dict, user_feedback: Dict) -> Dict:
    """评估解释的质量"""
    scores = {}

    # 评估完整性
    completeness = calculate_completeness(explanation, ground_truth)
    scores['completeness'] = completeness

    # 评估准确性
    accuracy = calculate_accuracy(explanation, ground_truth)
    scores['accuracy'] = accuracy

    # 评估一致性
    consistency = calculate_consistency(explanation, historical_explanations)
    scores['consistency'] = consistency

    # 评估可理解性
    understandability = user_feedback.get('understandability', 0)
    scores['understandability'] = understandability

    # 评估简洁性
    conciseness = user_feedback.get('conciseness', 0)
    scores['conciseness'] = conciseness

    # 计算总分
    total_score = (completeness * 0.3 + accuracy * 0.3 +
                  consistency * 0.2 + understandability * 0.1 +
                  conciseness * 0.1)
    scores['total'] = total_score

    return scores

class AuditRuleEngine:
    def __init__(self, rules_config):
        self.rules = self._load_rules(rules_config)
        self.trace = []

    def _load_rules(self, config):
        # 从配置加载规则
        rules = []
        for rule_config in config:
            rule = self._create_rule(rule_config)
            rules.append(rule)
        return rules

    def _create_rule(self, config):
        rule_type = config['type']
        if rule_type == 'keyword':
            return KeywordRule(config)
        elif rule_type == 'category':
            return CategoryRule(config)
        elif rule_type == 'image':
            return ImageRule(config)
        # 更多规则类型...

    def evaluate(self, product):
        self.trace = []
        violations = []

        for rule in self.rules:
            self.trace.append(f"评估规则: {rule.get_description()}")
            if not rule.evaluate(product):
                violations.append({
                    'rule': rule.get_description(),
                    'suggestion': rule.get_suggestion()
                })
                self.trace.append(f"违反规则: {rule.get_description()}")
            else:
                self.trace.append(f"通过规则: {rule.get_description()}")

        return len(violations) == 0, violations

class AuditStateMachine:
    def __init__(self):
        self.states = {
            'SUBMITTED': self._handle_submitted,
            'RULE_CHECK': self._handle_rule_check,
            'RISK_ASSESSMENT': self._handle_risk_assessment,
            'DECISION': self._handle_decision,
            'MANUAL_REVIEW': self._handle_manual_review,
            'APPROVED': self._handle_approved,
            'REJECTED': self._handle_rejected
        }
        self.transitions = {
            'SUBMITTED': ['RULE_CHECK'],
            'RULE_CHECK': ['RISK_ASSESSMENT', 'REJECTED'],
            'RISK_ASSESSMENT': ['DECISION', 'MANUAL_REVIEW'],
            'DECISION': ['APPROVED', 'REJECTED', 'MANUAL_REVIEW'],
            'MANUAL_REVIEW': ['APPROVED', 'REJECTED'],
            'APPROVED': [],
            'REJECTED': []
        }
        self.current_state = 'SUBMITTED'
        self.history = []

    def transition(self, to_state, context):
        if to_state not in self.transitions[self.current_state]:
            raise ValueError(f"Invalid transition from {self.current_state} to {to_state}")

        self.history.append({
            'from': self.current_state,
            'to': to_state,
            'timestamp': datetime.now(),
            'context': context
        })

        self.current_state = to_state
        handler = self.states[to_state]
        return handler(context)

    def _handle_submitted(self, context):
        # 处理提交状态
        return {'next_state': 'RULE_CHECK'}

    def _handle_rule_check(self, context):
        # 处理规则检查状态
        rule_engine = context.get('rule_engine')
        product = context.get('product')
        passed, violations = rule_engine.evaluate(product)

        if passed:
            return {'next_state': 'RISK_ASSESSMENT'}
        else:
            return {
                'next_state': 'REJECTED',
                'reason': 'rule_violation',
                'violations': violations
            }

    # 其他状态处理方法...

class ExplanationGenerator:
    def __init__(self, templates):
        self.templates = templates

    def generate(self, audit_result):
        explanation = {
            'summary': self._generate_summary(audit_result),
            'details': self._generate_details(audit_result),
            'suggestions': self._generate_suggestions(audit_result)
        }
        return explanation

    def _generate_summary(self, audit_result):
        template = self.templates['summary']
        reason = audit_result.get('reason', '')

        if reason == 'rule_violation':
            return template['rule_violation'].format(
                violation_count=len(audit_result.get('violations', []))
            )
        elif reason == 'high_risk':
            return template['high_risk'].format(
                risk_score=audit_result.get('risk_score', 0)
            )
        # 其他原因...

    def _generate_details(self, audit_result):
        details = []

        if audit_result.get('reason') == 'rule_violation':
            for violation in audit_result.get('violations', []):
                details.append({
                    'title': violation['rule'],
                    'description': self.templates['details']['rule_violation'].format(
                        rule=violation['rule']
                    )
                })

        # 其他详情生成...

        return details

    def _generate_suggestions(self, audit_result):
        suggestions = []

        if audit_result.get('reason') == 'rule_violation':
            for violation in audit_result.get('violations', []):
                if 'suggestion' in violation:
                    suggestions.append(violation['suggestion'])

        # 其他建议生成...

        return suggestions