返回博客

推理结果可解释性:为什么系统做了这个决策

在金融合规、医疗决策和司法领域,"系统为什么做了这个决策"比"系统做了什么决策"更重要。coomia-dip 构建了完整的推理可解释性框架,涵盖规则追踪链、ML 模型解释(SHAP/LIME)、因果图构建以及自然语言解释生成。本文深入解析可解释性的四个层次、解释数据模型、实时解释生成引擎以及面向不同角色的解释视图。

Coomia发布于 2025年8月28日18 分钟阅读
分享本文Twitter / X

系列:S5 智能决策 · 第 6 篇 | 难度:高级 | 阅读时间:20 分钟

推理结果可解释性:为什么系统做了这个决策

#TL;DR

在金融合规、医疗决策和司法领域,"系统为什么做了这个决策"比"系统做了什么决策"更重要。coomia-dip 构建了完整的推理可解释性框架,涵盖规则追踪链、ML 模型解释(SHAP/LIME)、因果图构建以及自然语言解释生成。本文深入解析可解释性的四个层次、解释数据模型、实时解释生成引擎以及面向不同角色的解释视图。

#1. 为什么需要可解释性

#1.1 不可解释的风险

Code
不可解释决策的后果:

  场景 1: 贷款拒绝
  +-----------------------------------+
  | "您的贷款申请被拒绝"               |
  | 原因: [无]                         |
  +-----------------------------------+
  -> 用户投诉, 监管罚款, 品牌受损

  场景 2: 可解释决策
  +-----------------------------------+
  | "您的贷款申请被拒绝"               |
  | 原因:                              |
  |   1. 信用评分 580 低于最低要求 620  |
  |   2. 近 12 个月有 2 次逾期记录      |
  |   3. 负债收入比 0.65 超过阈值 0.5   |
  | 建议: 降低负债后重新申请            |
  +-----------------------------------+
  -> 用户理解, 合规达标, 信任建立

#1.2 法规要求

法规区域要求
GDPR Art.22欧盟自动化决策必须提供有意义的解释
ECOA美国信贷拒绝必须提供具体原因
《个人信息保护法》中国自动化决策应提供说明
AI Act欧盟高风险 AI 系统必须可解释

#1.3 可解释性四层模型

Code
可解释性四层模型:

  Level 4: 自然语言解释
  +------------------------------------------+
  | "因为您的信用评分低于阈值且负债率偏高,   |
  |  系统决定拒绝本次贷款申请"               |
  +------------------------------------------+
                    |
  Level 3: 因果图
  +------------------------------------------+
  | credit_score(580) ---> below_threshold    |
  | debt_ratio(0.65) ----> over_limit         |
  | below_threshold + over_limit --> REJECT   |
  +------------------------------------------+
                    |
  Level 2: 特征归因
  +------------------------------------------+
  | credit_score: -0.35 (最大负面影响)        |
  | debt_ratio: -0.28                         |
  | income: +0.15                             |
  | employment: +0.08                         |
  +------------------------------------------+
                    |
  Level 1: 规则/模型追踪
  +------------------------------------------+
  | Rule CR-002 fired (priority 90)           |
  | Model: credit_v3 (confidence 0.23)        |
  | Path: PARALLEL_FUSION -> conflict -> rule |
  +------------------------------------------+

#2. 解释数据模型

#2.1 核心数据结构

Python
from dataclasses import dataclass, field
from typing import Any
from enum import Enum
from datetime import datetime


class ExplanationType(str, Enum):
    RULE_TRACE = "rule_trace"
    FEATURE_ATTRIBUTION = "feature_attribution"
    CAUSAL_GRAPH = "causal_graph"
    NATURAL_LANGUAGE = "natural_language"
    COUNTERFACTUAL = "counterfactual"


@dataclass
class FactorContribution:
    """单因素贡献"""
    factor_name: str
    factor_value: Any
    contribution: float       # -1.0 到 1.0
    direction: str            # "positive", "negative", "neutral"
    description: str

    @property
    def impact_level(self) -> str:
        abs_c = abs(self.contribution)
        if abs_c >= 0.3:
            return "high"
        if abs_c >= 0.1:
            return "medium"
        return "low"


@dataclass
class RuleTraceNode:
    """规则追踪节点"""
    node_type: str            # "fact", "condition", "rule", "conclusion"
    node_id: str
    label: str
    value: Any = None
    children: list["RuleTraceNode"] = field(default_factory=list)
    metadata: dict = field(default_factory=dict)


@dataclass
class CausalLink:
    """因果链接"""
    source: str
    target: str
    strength: float           # 0.0 到 1.0
    relationship: str         # "causes", "contributes", "blocks"


@dataclass
class Explanation:
    """完整解释对象"""
    explanation_id: str
    decision_id: str
    timestamp: datetime
    decision: str
    confidence: float

    # Level 1: 规则追踪
    rule_trace: RuleTraceNode | None = None
    fired_rules: list[str] = field(default_factory=list)

    # Level 2: 特征归因
    factors: list[FactorContribution] = field(default_factory=list)

    # Level 3: 因果图
    causal_links: list[CausalLink] = field(default_factory=list)

    # Level 4: 自然语言
    summary: str = ""
    detailed_explanation: str = ""

    # 反事实
    counterfactuals: list[dict] = field(default_factory=list)

    def top_factors(self, n: int = 3) -> list[FactorContribution]:
        """返回影响最大的 N 个因素"""
        return sorted(
            self.factors,
            key=lambda f: abs(f.contribution),
            reverse=True,
        )[:n]

#3. Level 1:规则追踪链

#3.1 规则触发追踪

Python
class RuleTraceBuilder:
    """构建规则推理追踪链"""

    def build_trace(self, fired_rules: list, facts: dict,
                    working_memory_snapshot: dict) -> RuleTraceNode:
        """构建完整的规则追踪树"""
        root = RuleTraceNode(
            node_type="root",
            node_id="trace_root",
            label="推理过程",
        )

        # 输入事实节点
        facts_node = RuleTraceNode(
            node_type="facts",
            node_id="input_facts",
            label="输入事实",
        )
        for key, value in facts.items():
            facts_node.children.append(RuleTraceNode(
                node_type="fact",
                node_id=f"fact_{key}",
                label=key,
                value=value,
            ))
        root.children.append(facts_node)

        # 规则匹配节点
        for rule in fired_rules:
            rule_node = RuleTraceNode(
                node_type="rule",
                node_id=rule.rule_id,
                label=rule.name,
                metadata={
                    "priority": rule.priority,
                    "cycle": rule.fired_at_cycle,
                },
            )

            # 条件匹配详情
            for cond in rule.conditions:
                fact_value = facts.get(cond.attribute)
                matched = cond.evaluate(fact_value)
                cond_node = RuleTraceNode(
                    node_type="condition",
                    node_id=f"{rule.rule_id}_{cond.attribute}",
                    label=f"{cond.attribute} {cond.operator} {cond.value}",
                    value={"actual": fact_value, "matched": matched},
                )
                rule_node.children.append(cond_node)

            # 结论
            conclusion = RuleTraceNode(
                node_type="conclusion",
                node_id=f"{rule.rule_id}_conclusion",
                label=rule.conclusion,
                value=rule.action_result,
            )
            rule_node.children.append(conclusion)
            root.children.append(rule_node)

        return root

    def to_ascii_tree(self, node: RuleTraceNode,
                      prefix: str = "", is_last: bool = True) -> str:
        """生成 ASCII 格式的追踪树"""
        connector = "+-- " if is_last else "|-- "
        lines = [f"{prefix}{connector}[{node.node_type}] {node.label}"]

        if node.value is not None:
            val_prefix = prefix + ("    " if is_last else "|   ")
            lines.append(f"{val_prefix}= {node.value}")

        child_prefix = prefix + ("    " if is_last else "|   ")
        for i, child in enumerate(node.children):
            is_child_last = (i == len(node.children) - 1)
            lines.append(
                self.to_ascii_tree(child, child_prefix, is_child_last)
            )

        return "\n".join(lines)

#3.2 追踪示例输出

Code
规则追踪树:

+-- [root] 推理过程
    |-- [facts] 输入事实
    |   |-- [fact] credit_score
    |   |   = 580
    |   |-- [fact] annual_income
    |   |   = 180000
    |   |-- [fact] debt_ratio
    |   |   = 0.65
    |   +-- [fact] previous_defaults
    |       = 2
    |
    |-- [rule] 低信用拒绝 (CR-002, priority=90)
    |   |-- [condition] credit_score < 500
    |   |   = {"actual": 580, "matched": false}
    |   |-- [condition] previous_defaults >= 3
    |   |   = {"actual": 2, "matched": false}
    |   +-- [conclusion] NO MATCH
    |
    +-- [rule] 中等信用条件审批 (CR-003, priority=50)
        |-- [condition] credit_score >= 600
        |   = {"actual": 580, "matched": false}
        +-- [conclusion] NO MATCH (first condition failed)

最终: 无规则匹配 -> 交由 ML 层处理

#4. Level 2:特征归因

#4.1 SHAP 值计算

Python
import numpy as np
from itertools import combinations


class SHAPExplainer:
    """SHAP (SHapley Additive exPlanations) 计算器"""

    def __init__(self, model, feature_names: list[str]):
        self._model = model
        self._features = feature_names

    def explain(self, instance: dict,
                background: list[dict] | None = None) -> list[FactorContribution]:
        """计算 SHAP 值"""
        n = len(self._features)
        shap_values = {}

        for feature in self._features:
            shap_values[feature] = self._shapley_value(
                instance, feature, background
            )

        # 归一化
        total = sum(abs(v) for v in shap_values.values()) or 1.0

        contributions = []
        for name in self._features:
            raw_shap = shap_values[name]
            normalized = raw_shap / total
            direction = "positive" if normalized > 0.01 else (
                "negative" if normalized < -0.01 else "neutral"
            )
            contributions.append(FactorContribution(
                factor_name=name,
                factor_value=instance.get(name),
                contribution=normalized,
                direction=direction,
                description=self._describe_factor(
                    name, instance.get(name), normalized
                ),
            ))

        return sorted(contributions,
                      key=lambda c: abs(c.contribution), reverse=True)

    def _shapley_value(self, instance: dict, feature: str,
                       background: list[dict] | None) -> float:
        """计算单个特征的 Shapley 值"""
        other_features = [f for f in self._features if f != feature]
        n = len(other_features)
        shapley = 0.0

        for size in range(n + 1):
            for subset in combinations(other_features, size):
                subset_set = set(subset)

                # 有该特征时的预测
                with_input = {f: instance[f] for f in subset_set | {feature}}
                pred_with, _ = self._model.predict(with_input)

                # 无该特征时的预测
                without_input = {f: instance[f] for f in subset_set}
                pred_without, _ = self._model.predict(without_input)

                # 边际贡献
                marginal = self._to_numeric(pred_with) - self._to_numeric(pred_without)

                # Shapley 权重
                weight = (
                    np.math.factorial(size) *
                    np.math.factorial(n - size - 1)
                ) / np.math.factorial(n)

                shapley += weight * marginal

        return shapley

    def _to_numeric(self, prediction: str) -> float:
        mapping = {"approve": 1.0, "conditional_approve": 0.5, "reject": 0.0}
        return mapping.get(prediction, 0.5)

    def _describe_factor(self, name: str, value: Any,
                         contribution: float) -> str:
        abs_c = abs(contribution)
        impact = "强烈" if abs_c >= 0.3 else ("中等" if abs_c >= 0.1 else "轻微")
        direction = "正面" if contribution > 0 else "负面"
        return f"{name}={value} 对决策有{impact}{direction}影响"

#4.2 特征归因可视化

Code
特征归因(SHAP 值):

  因素              值       贡献度    方向      影响
  ---------------------------------------------------------------
  credit_score      580     -0.35     <<<<<<  [==========] 高-负面
  debt_ratio        0.65    -0.28     <<<<    [========]   高-负面
  annual_income     180000  +0.15       >>>   [====]       中-正面
  employment_yrs    5       +0.08       >>    [==]         低-正面
  previous_defaults 2       -0.12     <<<     [===]        中-负面
  industry          tech    +0.02       >     [=]          低-正面

  负面 <<<<<<<<<<|>>>>>>>>> 正面

#5. Level 3:因果图构建

#5.1 因果关系提取

Python
@dataclass
class CausalNode:
    """因果图节点"""
    node_id: str
    label: str
    node_type: str     # "input", "intermediate", "output"
    value: Any = None


class CausalGraphBuilder:
    """因果图构建器"""

    def __init__(self):
        self._nodes: dict[str, CausalNode] = {}
        self._edges: list[CausalLink] = []

    def build_from_explanation(
        self, factors: list[FactorContribution],
        rules_fired: list, decision: str
    ) -> tuple[list[CausalNode], list[CausalLink]]:
        """从解释数据构建因果图"""
        self._nodes.clear()
        self._edges.clear()

        # 输入因素节点
        for factor in factors:
            node = CausalNode(
                node_id=f"input_{factor.factor_name}",
                label=f"{factor.factor_name}={factor.factor_value}",
                node_type="input",
                value=factor.factor_value,
            )
            self._nodes[node.node_id] = node

        # 中间推理节点(从规则条件提取)
        for rule in rules_fired:
            for cond in rule.conditions:
                mid_id = f"check_{cond.attribute}_{cond.operator}"
                mid_node = CausalNode(
                    node_id=mid_id,
                    label=f"{cond.attribute} {cond.operator} {cond.value}",
                    node_type="intermediate",
                    value=cond.evaluate(
                        next(f.factor_value for f in factors
                             if f.factor_name == cond.attribute)
                    ),
                )
                self._nodes[mid_id] = mid_node

                # 输入 -> 中间
                self._edges.append(CausalLink(
                    source=f"input_{cond.attribute}",
                    target=mid_id,
                    strength=abs(next(
                        f.contribution for f in factors
                        if f.factor_name == cond.attribute
                    )),
                    relationship="causes",
                ))

            # 中间 -> 规则结论
            rule_node = CausalNode(
                node_id=f"rule_{rule.rule_id}",
                label=f"Rule: {rule.name}",
                node_type="intermediate",
            )
            self._nodes[rule_node.node_id] = rule_node

        # 最终决策节点
        decision_node = CausalNode(
            node_id="decision",
            label=f"Decision: {decision}",
            node_type="output",
            value=decision,
        )
        self._nodes["decision"] = decision_node

        return list(self._nodes.values()), self._edges

    def to_ascii(self) -> str:
        """生成 ASCII 因果图"""
        lines = ["因果关系图:", ""]
        inputs = [n for n in self._nodes.values() if n.node_type == "input"]
        mids = [n for n in self._nodes.values() if n.node_type == "intermediate"]
        outputs = [n for n in self._nodes.values() if n.node_type == "output"]

        lines.append("  输入因素:")
        for n in inputs:
            lines.append(f"    [{n.label}]")
        lines.append("        |")
        lines.append("        v")
        lines.append("  推理过程:")
        for n in mids:
            result = "PASS" if n.value else "FAIL"
            lines.append(f"    [{n.label}] -> {result}")
        lines.append("        |")
        lines.append("        v")
        lines.append("  决策结果:")
        for n in outputs:
            lines.append(f"    [{n.label}]")

        return "\n".join(lines)

#5.2 因果图示例

Code
因果关系图:

  输入因素:
    [credit_score=580] --+
    [debt_ratio=0.65] ---+---> [credit_score >= 600?] -> FAIL
    [income=180000] -----+
    [defaults=2] --------+---> [defaults < 2?] -> FAIL
                          |
                          v
  推理过程:
    [Rule: CR-003 中等信用] -> NOT MATCHED (条件不满足)
    [ML Model: credit_v3] -> REJECT (confidence: 0.77)
                          |
                          v
  融合决策:
    [Fusion: ML wins] -> confidence: 0.62
                          |
                          v
  最终:
    [Decision: REJECT] -- 需要人工审核 (confidence < 0.7)

#6. Level 4:自然语言解释生成

#6.1 模板引擎

Python
class NaturalLanguageExplainer:
    """自然语言解释生成器"""

    def __init__(self):
        self._templates = {
            "approve": {
                "zh": "系统批准了此{domain}申请,主要因为{top_positive_factors}。"
                      "综合置信度为 {confidence:.0%}。",
                "en": "The system approved this {domain} application, primarily because "
                      "{top_positive_factors}. Overall confidence: {confidence:.0%}.",
            },
            "reject": {
                "zh": "系统拒绝了此{domain}申请。主要原因:{top_negative_factors}。"
                      "{suggestion}",
                "en": "The system rejected this {domain} application. Main reasons: "
                      "{top_negative_factors}. {suggestion}",
            },
            "conditional_approve": {
                "zh": "系统有条件地批准了此{domain}申请。{conditions_text}。"
                      "需要满足以下条件后方可最终通过。",
                "en": "The system conditionally approved this {domain} application. "
                      "{conditions_text}. The following conditions must be met.",
            },
        }
        self._factor_templates = {
            "zh": {
                "positive": "{name} 为 {value},对决策有正面影响",
                "negative": "{name} 为 {value},低于要求标准",
            },
            "en": {
                "positive": "{name} is {value}, positively impacting the decision",
                "negative": "{name} is {value}, below required threshold",
            },
        }

    def generate(self, explanation: Explanation, domain: str,
                 lang: str = "zh") -> str:
        """生成自然语言解释"""
        template = self._templates.get(
            explanation.decision, self._templates["reject"]
        )[lang]

        top_factors = explanation.top_factors(3)

        positive = [f for f in top_factors if f.direction == "positive"]
        negative = [f for f in top_factors if f.direction == "negative"]

        pos_text = self._format_factors(positive, "positive", lang)
        neg_text = self._format_factors(negative, "negative", lang)

        suggestion = self._generate_suggestion(
            explanation.decision, negative, lang
        )

        result = template.format(
            domain=domain,
            confidence=explanation.confidence,
            top_positive_factors=pos_text or ("各项指标达标" if lang == "zh"
                                               else "all metrics met"),
            top_negative_factors=neg_text or ("无" if lang == "zh" else "none"),
            suggestion=suggestion,
            conditions_text=neg_text,
        )

        return result

    def _format_factors(self, factors: list[FactorContribution],
                        direction: str, lang: str) -> str:
        if not factors:
            return ""
        tpl = self._factor_templates[lang][direction]
        parts = [
            tpl.format(name=f.factor_name, value=f.factor_value)
            for f in factors
        ]
        sep = ";" if lang == "zh" else "; "
        return sep.join(parts)

    def _generate_suggestion(self, decision: str,
                             negative_factors: list[FactorContribution],
                             lang: str) -> str:
        if decision != "reject" or not negative_factors:
            return ""
        if lang == "zh":
            suggestions = [
                f"建议改善 {f.factor_name}"
                for f in negative_factors
            ]
            return "改善建议:" + "、".join(suggestions) + " 后重新申请。"
        else:
            suggestions = [
                f"improve {f.factor_name}"
                for f in negative_factors
            ]
            return "Suggestions: " + ", ".join(suggestions) + " before reapplying."

#6.2 反事实解释

Python
class CounterfactualExplainer:
    """反事实解释:'如果 X 不同,结果会怎样?'"""

    def __init__(self, reasoning_engine):
        self._engine = reasoning_engine

    def generate_counterfactuals(
        self, original_input: dict,
        original_decision: str,
        target_decision: str,
        max_changes: int = 3,
    ) -> list[dict]:
        """生成反事实解释"""
        counterfactuals = []

        # 对每个输入变量尝试调整
        for key in original_input:
            adjusted = self._find_threshold(
                original_input, key, target_decision
            )
            if adjusted is not None:
                counterfactuals.append({
                    "changed_factor": key,
                    "original_value": original_input[key],
                    "required_value": adjusted,
                    "new_decision": target_decision,
                    "description": (
                        f"如果 {key}{original_input[key]} "
                        f"变为 {adjusted},决策将变为 {target_decision}"
                    ),
                })

        # 按变化幅度排序
        counterfactuals.sort(
            key=lambda c: abs(
                c["required_value"] - c["original_value"]
            ) if isinstance(c["original_value"], (int, float)) else 0
        )

        return counterfactuals[:max_changes]

    def _find_threshold(self, inputs: dict, key: str,
                        target: str) -> Any:
        """二分搜索找到使决策改变的阈值"""
        value = inputs[key]
        if not isinstance(value, (int, float)):
            return None

        # 搜索范围
        low = value * 0.5
        high = value * 2.0

        for _ in range(20):  # 最多 20 次二分
            mid = (low + high) / 2
            test_input = {**inputs, key: mid}
            result = self._engine.evaluate(test_input)
            if result and result.get("decision") == target:
                high = mid
            else:
                low = mid

        # 验证
        final_input = {**inputs, key: high}
        result = self._engine.evaluate(final_input)
        if result and result.get("decision") == target:
            return round(high, 2) if isinstance(value, float) else int(high)
        return None

#7. 解释服务架构

#7.1 完整服务流程

Code
解释服务架构:

  推理请求
      |
      v
  +---+---+
  | 推理   |---> 决策结果
  | 引擎   |        |
  +---+---+         |
      |              |
      v              v
  +---+---+    +-----+-----+
  | 追踪   |    | 归因计算  |
  | 收集器 |    | (SHAP)   |
  +---+---+    +-----+-----+
      |              |
      +------+-------+
             |
             v
      +------+------+
      | 因果图构建   |
      +------+------+
             |
             v
      +------+------+
      | NL 生成器   |
      +------+------+
             |
             v
      +------+------+
      | Explanation |
      | Store       |
      +------+------+
             |
             v
        gRPC 响应

#7.2 gRPC 接口

PROTOBUF
syntax = "proto3";
package onto.explainability.v1;

service ExplainabilityService {
    rpc GetExplanation(GetExplanationRequest)
        returns (ExplanationResponse);
    rpc GetCounterfactuals(CounterfactualRequest)
        returns (CounterfactualResponse);
    rpc GetFactorAttribution(AttributionRequest)
        returns (AttributionResponse);
}

message GetExplanationRequest {
    string decision_id = 1;
    string language = 2;           // "zh", "en"
    repeated string levels = 3;    // "rule_trace", "attribution", etc.
}

message ExplanationResponse {
    string decision = 1;
    double confidence = 2;
    string summary = 3;
    string detailed_explanation = 4;
    repeated Factor factors = 5;
    repeated CausalLink causal_links = 6;
    string rule_trace_ascii = 7;
}

message Factor {
    string name = 1;
    string value = 2;
    double contribution = 3;
    string direction = 4;
    string description = 5;
}

#8. 面向角色的解释视图

#8.1 多角色视图

Code
角色视图矩阵:

角色          | 需要的解释层次        | 详细程度 | 格式
-------------|---------------------|---------|--------
终端用户      | Level 4 (自然语言)   | 简要     | 文本
业务经理      | Level 3+4 (因果图+NL)| 中等     | 图+文本
合规审计      | Level 1+2 (追踪+归因)| 详细     | 完整报告
数据科学家    | Level 1+2+3 (全部)   | 详细     | JSON/图
系统管理员    | Level 1 (追踪)       | 技术     | 日志

#8.2 视图生成器

Python
class ExplanationViewGenerator:
    """按角色生成不同的解释视图"""

    def __init__(self, nl_explainer: NaturalLanguageExplainer):
        self._nl = nl_explainer

    def generate_view(self, explanation: Explanation,
                      role: str, domain: str,
                      lang: str = "zh") -> dict:
        """按角色生成解释视图"""
        if role == "end_user":
            return self._end_user_view(explanation, domain, lang)
        elif role == "business_manager":
            return self._manager_view(explanation, domain, lang)
        elif role == "compliance":
            return self._compliance_view(explanation, domain, lang)
        elif role == "data_scientist":
            return self._scientist_view(explanation, domain, lang)
        else:
            return self._default_view(explanation, domain, lang)

    def _end_user_view(self, exp: Explanation,
                       domain: str, lang: str) -> dict:
        return {
            "summary": self._nl.generate(exp, domain, lang),
            "decision": exp.decision,
            "top_reasons": [
                f.description for f in exp.top_factors(3)
            ],
            "suggestions": [
                c["description"] for c in exp.counterfactuals[:2]
            ],
        }

    def _compliance_view(self, exp: Explanation,
                         domain: str, lang: str) -> dict:
        trace_builder = RuleTraceBuilder()
        return {
            "decision_id": exp.decision_id,
            "timestamp": exp.timestamp.isoformat(),
            "decision": exp.decision,
            "confidence": exp.confidence,
            "fired_rules": exp.fired_rules,
            "rule_trace": (trace_builder.to_ascii_tree(exp.rule_trace)
                          if exp.rule_trace else "N/A"),
            "all_factors": [
                {
                    "name": f.factor_name,
                    "value": f.factor_value,
                    "contribution": f.contribution,
                    "impact": f.impact_level,
                }
                for f in exp.factors
            ],
            "causal_links": [
                {
                    "from": l.source,
                    "to": l.target,
                    "strength": l.strength,
                    "type": l.relationship,
                }
                for l in exp.causal_links
            ],
            "audit_trail": {
                "reasoning_path": "PARALLEL_FUSION",
                "rule_layer_result": exp.fired_rules,
                "ml_model_used": "credit_v3",
                "fusion_method": "weighted_agreement",
            },
        }

    def _manager_view(self, exp: Explanation,
                      domain: str, lang: str) -> dict:
        return {
            "summary": self._nl.generate(exp, domain, lang),
            "decision": exp.decision,
            "confidence": f"{exp.confidence:.0%}",
            "key_factors": [
                {
                    "factor": f.factor_name,
                    "value": f.factor_value,
                    "impact": f.impact_level,
                    "direction": f.direction,
                }
                for f in exp.top_factors(5)
            ],
            "counterfactuals": exp.counterfactuals[:3],
        }

    def _scientist_view(self, exp: Explanation,
                        domain: str, lang: str) -> dict:
        return {
            "decision_id": exp.decision_id,
            "decision": exp.decision,
            "confidence": exp.confidence,
            "rule_trace": exp.rule_trace,
            "shap_values": {f.factor_name: f.contribution for f in exp.factors},
            "causal_graph": {
                "nodes": [vars(n) for n in []],
                "edges": [vars(l) for l in exp.causal_links],
            },
            "counterfactuals": exp.counterfactuals,
            "raw_explanation": exp,
        }

    def _default_view(self, exp: Explanation,
                      domain: str, lang: str) -> dict:
        return self._end_user_view(exp, domain, lang)

#9. 性能与存储

#9.1 解释生成延迟

解释层次延迟占推理延迟比例
Level 1 规则追踪2-5ms< 5%
Level 2 SHAP 归因50-200ms20-40%
Level 3 因果图10-30ms5-10%
Level 4 NL 生成5-15ms3-5%
反事实计算200-500ms异步

#9.2 存储方案

Code
解释存储方案:

  实时查询              归档存储
  (< 30 天)             (> 30 天)
  +----------+          +----------+
  | PostgreSQL|          | Iceberg  |
  | (JSON列)  |          | (Parquet)|
  +----------+          +----------+
       |                      |
       v                      v
  索引: decision_id      分区: year/month
  索引: timestamp        保留: 7 年 (合规)
  索引: domain

#10. 实战案例

Python
# 完整的可解释性流程

# 1. 推理完成后,构建解释
explanation = Explanation(
    explanation_id="EXP-2026-001",
    decision_id="DEC-2026-001",
    timestamp=datetime.utcnow(),
    decision="reject",
    confidence=0.62,
    fired_rules=["CR-002"],
    factors=[
        FactorContribution("credit_score", 580, -0.35, "negative",
                          "信用评分 580 低于最低要求 620"),
        FactorContribution("debt_ratio", 0.65, -0.28, "negative",
                          "负债收入比 0.65 超过阈值 0.5"),
        FactorContribution("annual_income", 180000, 0.15, "positive",
                          "年收入 18 万符合基本要求"),
        FactorContribution("previous_defaults", 2, -0.12, "negative",
                          "近期有 2 次逾期记录"),
    ],
    counterfactuals=[
        {
            "changed_factor": "credit_score",
            "original_value": 580,
            "required_value": 650,
            "new_decision": "conditional_approve",
            "description": "如果信用评分提升至 650,可获得条件审批",
        },
    ],
)

# 2. 生成自然语言解释
nl = NaturalLanguageExplainer()
summary = nl.generate(explanation, "credit", "zh")
# -> "系统拒绝了此信用申请。主要原因:credit_score 为 580,低于要求标准;
#     debt_ratio 为 0.65,低于要求标准。建议改善 credit_score、debt_ratio 后重新申请。"

# 3. 生成角色视图
view_gen = ExplanationViewGenerator(nl)
user_view = view_gen.generate_view(explanation, "end_user", "credit", "zh")
audit_view = view_gen.generate_view(explanation, "compliance", "credit", "zh")

#Key Takeaways

  1. 四层可解释性模型从技术追踪到自然语言覆盖不同深度需求
  2. 规则追踪链完整记录每条规则的条件匹配和触发过程
  3. SHAP 特征归因量化每个输入因素对最终决策的贡献度
  4. 因果图可视化展示从输入到决策的完整推理路径
  5. 自然语言生成让非技术用户也能理解决策原因
  6. 反事实解释回答"如果改变什么,结果会不同"的关键问题
  7. 多角色视图(终端用户/经理/合规/数据科学家)满足不同场景需求

#Next Article

下一篇 S5-07 决策引擎架构:决策树 + 约束求解双引擎 将深入解析 coomia-dip DecisionEngine 的双引擎设计。

tags: #explainability #shap #causal-graph #natural-language #counterfactual #audit #coomia-dip