# 05.2 伦理与责任

## 核心问题

> 空间决策如何影响不同的人群和环境？
> 当AI参与空间规划时，如何确保过程的公平和透明？
> 出错的决策责任应该由谁来承担？

---

## 概念讲解

### 空间决策的伦理维度

空间决策不是价值中立的，它们分配资源、机会和风险：

```
空间决策的伦理影响

┌─────────────────────────────────────────────────────────────┐
│                                                             │
│  1. 分配正义 (Distributive Justice)                         │
│     - 谁获得绿地、公园等正面空间资源？                        │
│     - 谁承受污染、噪声等负面影响？                            │
│     - 空间资源的公平分配原则是什么？                          │
│                                                             │
│  2. 程序正义 (Procedural Justice)                           │
│     - 决策过程是否透明？                                      │
│     - 受影响者能否参与决策？                                  │
│     - 决策依据是否可审查？                                    │
│                                                             │
│  3. 承认正义 (Recognition Justice)                          │
│     - 不同群体的需求和价值观是否被认可？                      │
│     - 弱势群体的空间权利是否被尊重？                          │
│     - 文化多样性在空间中如何体现？                            │
│                                                             │
│  4. 生态正义 (Ecological Justice)                           │
│     - 当代人与未来世代之间的公平？                            │
│     - 人类活动对生态系统的责任？                              │
│     - 非人类物种的空间权利？                                  │
│                                                             │
└─────────────────────────────────────────────────────────────┘
```

### AI带来的新伦理挑战

```
AI空间决策的特有问题

1. 算法偏见 (Algorithmic Bias)
   训练数据中的社会偏见被编码进模型
   → 历史上的红线政策可能影响现在的预测

2. 黑箱决策 (Black Box Decision)
   复杂模型的决策过程难以解释
   → 利益相关者无法质疑或理解决策

3. 责任分散 (Diffused Responsibility)
   涉及多个主体：开发者、用户、数据提供者
   → 出错时责任难以界定

4. 规模效应 (Scale Effects)
   AI可以大规模应用决策
   → 小偏差在大规模下产生大影响

5. 路径依赖 (Path Dependence)
   早期决策影响后续数据收集
   → 偏见自我强化
```

### 空间正义的典型问题

| 问题类型 | 说明 | AI相关风险 |
|---------|------|-----------|
| **环境种族主义** | 有害设施更多位于少数族裔社区 | AI可能复制历史模式 |
| **绿色绅士化** | 绿地改善导致原住民被迫搬迁 | AI优化可能加剧此问题 |
| **数字鸿沟** | 缺乏数据地区被忽视 | AI只关注数据丰富的区域 |
| **代表性不足** | 某些群体的需求未被考虑 | 训练数据偏差 |

---

## 设计原理

### 可解释性设计原则

```python
"""
可解释的空间AI设计
"""
from typing import Dict, List, Any, Optional
from dataclasses import dataclass
from abc import ABC, abstractmethod

@dataclass
class DecisionExplanation:
    """决策解释"""
    decision: str                      # 做出的决策
    rationale: List[str]                # 决策理由
    key_factors: Dict[str, float]       # 关键因素及权重
    alternatives: List[Dict]            # 考虑过的替代方案
    uncertainties: List[str]            # 不确定性说明
    assumptions: List[str]              # 假设条件

class ExplainableSpatialAI(ABC):
    """可解释的空间AI基类"""

    @abstractmethod
    def make_decision(self, context: Dict) -> Any:
        """做出决策"""
        pass

    @abstractmethod
    def explain_decision(self, decision: Any, context: Dict) -> DecisionExplanation:
        """解释决策"""
        pass

    def audit_trail(self) -> List[Dict]:
        """返回审计轨迹"""
        return self._audit_log

class ExplainableSiteSelector(ExplainableSpatialAI):
    """可解释的选址AI"""

    def __init__(self):
        self.criteria_weights = {}
        self._audit_log = []

    def set_criteria(self, criteria: Dict[str, float], justification: str):
        """
        设置评判标准

        Args:
            criteria: {标准名: 权重}
            justification: 权重选择的理由
        """
        self.criteria_weights = criteria.copy()
        self._log({
            'action': 'set_criteria',
            'criteria': criteria,
            'justification': justification
        })

    def make_decision(self, context: Dict) -> Dict:
        """
        做出选址决策

        返回选中的地点及其评分
        """
        sites = context['sites']
        constraints = context.get('constraints', {})

        # 评估每个候选地
        scored_sites = []
        for site in sites:
            score, details = self._evaluate_site(site, context)
            scored_sites.append({
                'site': site,
                'score': score,
                'details': details
            })

        # 排序并选择最高分
        scored_sites.sort(key=lambda x: x['score'], reverse=True)
        selected = scored_sites[0]

        # 记录决策
        self._log({
            'action': 'make_decision',
            'selected': selected['site'],
            'score': selected['score'],
            'alternatives': scored_sites[1:4]  # 保存前几个备选
        })

        return selected

    def explain_decision(self, decision: Any, context: Dict) -> DecisionExplanation:
        """生成决策解释"""
        selected_site = decision['site']
        score_details = decision['details']

        # 生成解释
        return DecisionExplanation(
            decision=f"选择地点 {selected_site['name']}",
            rationale=[
                f"该地点综合评分最高 ({decision['score']:.2f})",
                "评分基于预定义的标准和权重",
                "所有候选地点已被系统评估"
            ],
            key_factors=score_details,
            alternatives=[
                {
                    'site': alt['site']['name'],
                    'score': alt['score'],
                    'reason': '评分较低'
                }
                for alt in context.get('alternatives', [])[:3]
            ],
            uncertainties=[
                "评分依赖输入数据的准确性",
                "权重选择包含主观判断",
                "未量化的因素可能影响实际适用性"
            ],
            assumptions=[
                "所有标准可以用数值表示",
                "各标准相互独立",
                "当前条件在未来保持稳定"
            ]
        )

    def _evaluate_site(self, site: Dict, context: Dict) -> tuple:
        """评估单个地点"""
        scores = {}

        for criterion, weight in self.criteria_weights.items():
            # 从地点数据中获取该标准的值
            value = site.get(criterion, 0)

            # 标准化（简化版）
            normalized = self._normalize(criterion, value)

            # 加权
            scores[criterion] = normalized * weight

        total_score = sum(scores.values())

        return total_score, scores

    def _normalize(self, criterion: str, value: float) -> float:
        """标准化准则值"""
        # 简化：假设越大越好，范围0-100
        return min(max(value / 100, 0), 1)

    def _log(self, entry: Dict):
        """记录日志"""
        entry['timestamp'] = self._get_timestamp()
        self._audit_log.append(entry)

    def _get_timestamp(self) -> str:
        """获取时间戳"""
        from datetime import datetime
        return datetime.now().isoformat()

# === 伦理检查 ===

class EthicsChecker:
    """伦理检查器"""

    def __init__(self):
        self.checks = []

    def add_check(self, check_fn, name: str):
        """添加检查"""
        self.checks.append((check_fn, name))
        return self

    def check_decision(self, decision: Any, context: Dict) -> Dict:
        """执行所有伦理检查"""
        results = {
            'passed': True,
            'issues': [],
            'warnings': []
        }

        for check_fn, name in self.checks:
            try:
                result = check_fn(decision, context)
                if not result['passed']:
                    results['passed'] = False
                    results['issues'].append({
                        'check': name,
                        'reason': result['reason']
                    })
                elif result.get('warning'):
                    results['warnings'].append({
                        'check': name,
                        'warning': result['warning']
                    })
            except Exception as e:
                results['issues'].append({
                    'check': name,
                    'reason': f"检查失败: {str(e)}"
                })

        return results

# 预定义的伦理检查

def check_environmental_justice(decision, context) -> Dict:
    """检查环境正义：确保不将负面影响集中到弱势社区"""
    selected_site = decision['site']

    # 检查是否有弱势群体数据
    vulnerable_communities = context.get('vulnerable_communities', [])

    for community in vulnerable_communities:
        if selected_site.get('near_community') == community['id']:
            # 如果项目有负面影响，需要特别审查
            if context.get('project_type') == 'negative_impact':
                return {
                    'passed': False,
                    'reason': f"选址靠近弱势社区 {community['name']}，需要额外的环境正义审查"
                }

    return {'passed': True}

def check_transparency(decision, context) -> Dict:
    """检查透明度：确保决策过程可记录和审查"""
    if not decision.get('details'):
        return {
            'passed': False,
            'reason': '决策缺乏详细评分信息，无法审查'
        }

    return {'passed': True}

def check_public_participation(decision, context) -> Dict:
    """检查公众参与：确保受影响者有机会表达意见"""
    if context.get('affects_public', False):
        participation = context.get('public_participation')
        if not participation or participation == 'none':
            return {
                'passed': False,
                'reason': '项目影响公众但缺乏公众参与程序'
            }
        elif participation == 'minimal':
            return {
                'passed': True,
                'warning': '公众参与程度较低，建议加强'
            }

    return {'passed': True}

# === 使用示例 ===

if __name__ == "__main__":
    print("=== 可解释的空间AI ===\n")

    # 创建选址器
    selector = ExplainableSiteSelector()

    # 设置评判标准（带理由）
    selector.set_criteria(
        criteria={
            'accessibility': 0.3,
            'environmental_quality': 0.25,
            'cost_effectiveness': 0.2,
            'community_support': 0.15,
            'future_potential': 0.1
        },
        justification="基于项目目标和利益相关者访谈"
    )

    # 模拟候选地点
    sites = [
        {'name': 'Site A', 'accessibility': 85, 'environmental_quality': 70,
         'cost_effectiveness': 60, 'community_support': 80, 'future_potential': 75},
        {'name': 'Site B', 'accessibility': 70, 'environmental_quality': 85,
         'cost_effectiveness': 75, 'community_support': 60, 'future_potential': 70},
        {'name': 'Site C', 'accessibility': 90, 'environmental_quality': 60,
         'cost_effectiveness': 80, 'community_support': 70, 'future_potential': 65},
    ]

    # 创建伦理检查器
    ethics_checker = EthicsChecker()
    ethics_checker.add_check(check_environmental_justice, "环境正义检查")
    ethics_checker.add_check(check_transparency, "透明度检查")
    ethics_checker.add_check(check_public_participation, "公众参与检查")

    # 上下文
    context = {
        'sites': sites,
        'vulnerable_communities': [],
        'project_type': 'neutral',
        'affects_public': True,
        'public_participation': 'moderate'
    }

    # 做出决策
    decision = selector.make_decision(context)

    print(f"选中地点: {decision['site']['name']}")
    print(f"综合评分: {decision['score']:.2f}\n")

    # 获取解释
    explanation = selector.explain_decision(decision, context)
    print("=== 决策解释 ===")
    print(f"决策: {explanation.decision}")
    print(f"\n理由:")
    for r in explanation.rationale:
        print(f"  - {r}")
    print(f"\n关键因素:")
    for factor, value in explanation.key_factors.items():
        print(f"  - {factor}: {value:.3f}")

    # 伦理检查
    print(f"\n=== 伦理检查 ===")
    ethics_result = ethics_checker.check_decision(decision, context)
    if ethics_result['passed']:
        print("所有伦理检查通过")
    else:
        print("伦理检查发现问题:")
        for issue in ethics_result['issues']:
            print(f"  - [{issue['check']}] {issue['reason']}")
```

### 问责机制设计

```
AI空间决策的问责框架

┌─────────────────────────────────────────────────────────────┐
│                                                             │
│  责任链                                                     │
│                                                             │
│  数据提供者 ──→ 模型开发者 ──→ 系统集成者 ──→ 最终用户      │
│      │            │              │             │            │
│      │            │              │             └── 决策责任 │
│      │            │              └── 集成责任               │
│      │            └── 模型责任                              │
│      └── 数据质量责任                                         │
│                                                             │
│  问责机制                                                   │
│                                                             │
│  1. 文档化 (Documentation)                                  │
│     - 记录所有决策和数据来源                                 │
│     - 保存模型版本和参数                                     │
│     - 维护变更历史                                           │
│                                                             │
│  2. 审计 (Audit)                                            │
│     - 定期审查决策                                           │
│     - 检查偏见和公平性                                       │
│     - 验证技术正确性                                         │
│                                                             │
│  3. 申诉 (Appeal)                                           │
│     - 提供质疑决策的渠道                                     │
│     - 建立复核机制                                           │
│     - 允许人工干预                                           │
│                                                             │
│  4. 纠正 (Remedy)                                           │
│     - 发现错误后的补救措施                                   │
│     - 对受影响方的补偿                                       │
│     - 系统改进和预防                                         │
│                                                             │
└─────────────────────────────────────────────────────────────┘
```

---

## 案例分析

### 案例1：城市绿地分布的算法偏见

**问题背景**：某城市使用AI优化绿地布局，但结果加剧了既有不平等。

```python
"""
问题代码示例：训练数据中的历史偏见
"""

# 问题：使用历史公园使用数据来优化新公园位置
def optimize_park_locations_biased(historical_usage_data, new_sites):
    """
    有偏见的优化算法

    问题：历史使用数据反映的是历史可达性，
    而非真实需求。服务不足的区域数据少，
    因此被算法继续忽视。
    """
    # 简单优化：在历史使用高的地方附近选址
    scored = []
    for site in new_sites:
        # 靠近高使用区域得分高
        score = sum(
            usage for nearby, usage in historical_usage_data
            if distance(site, nearby) < 1000
        )
        scored.append((site, score))

    # 选择得分最高的
    scored.sort(key=lambda x: x[1], reverse=True)
    return [s[0] for s in scored[:5]]

# 改进版本：考虑需求而非历史使用
def optimize_park_locations_fair(demand_indicators, new_sites,
                                 equity_weight=0.5):
    """
    公平的优化算法

    考虑：
    1. 当前服务不足程度（需求）
    2. 人口密度
    3. 弱势群体分布
    """
    scored = []
    for site in new_sites:
        # 服务不足得分
        underserved_score = calculate_underserved(site, demand_indicators)

        # 效率得分（可达人口）
        efficiency_score = calculate_accessible_population(site)

        # 综合得分，可调整公平权重
        score = (1 - equity_weight) * efficiency_score + \
                equity_weight * underserved_score

        scored.append((site, score, {
            'underserved': underserved_score,
            'efficiency': efficiency_score
        }))

    # 按综合得分排序
    scored.sort(key=lambda x: x[1], reverse=True)

    # 记录决策依据
    for site, score, details in scored:
        site['selection_score'] = score
        site['score_details'] = details

    return [s[0] for s in scored[:5]]

def calculate_underserved(site, indicators):
    """计算服务不足程度"""
    # 距离最近的现有设施
    distance_to_nearest = min_distance_to_parks(site)

    # 附近人口中的弱势群体比例
    vulnerable_ratio = get_vulnerable_population_ratio(site)

    # 服务不足 = 距离远 + 弱势群体多
    return distance_to_nearest * (1 + vulnerable_ratio)
```

### 案例2：生态保护区的社区影响

**问题背景**：AI优化的生态廊道选址忽略了当地社区权益。

```python
"""
考虑多方利益的生态廊道选址
"""

class EthicalCorridorSelector:
    """伦理导向的廊道选址器"""

    def __init__(self):
        self.stakeholders = {
            'ecology': {'weight': 0.4, 'concern': '生态连通性'},
            'community': {'weight': 0.3, 'concern': '社区利益'},
            'economy': {'weight': 0.2, 'concern': '经济成本'},
            'culture': {'weight': 0.1, 'concern': '文化价值'}
        }

    def evaluate_corridor_route(self, route, context):
        """
        评估廊道路线

        返回：综合评分和各利益相关方的影响
        """
        scores = {}

        # 生态评分
        scores['ecology'] = self._evaluate_ecological_value(route, context)

        # 社区评分
        scores['community'] = self._evaluate_community_impact(route, context)

        # 经济评分
        scores['economy'] = self._evaluate_economic_cost(route, context)

        # 文化评分
        scores['culture'] = self._evaluate_cultural_impact(route, context)

        # 加权综合
        total = sum(
            scores[stakeholder] * self.stakeholders[stakeholder]['weight']
            for stakeholder in self.stakeholders
        )

        # 检查任何一方的严重负面影响
        for stakeholder, score in scores.items():
            if score < 0.3:  # 阈值
                return {
                    'acceptable': False,
                    'reason': f"{stakeholder}评分过低: {score:.2f}",
                    'scores': scores,
                    'total': total
                }

        return {
            'acceptable': True,
            'total_score': total,
            'scores': scores,
            'breakdown': {
                stakeholder: {
                    'score': scores[stakeholder],
                    'weight': self.stakeholders[stakeholder]['weight'],
                    'concern': self.stakeholders[stakeholder]['concern']
                }
                for stakeholder in self.stakeholders
            }
        }

    def _evaluate_ecological_value(self, route, context):
        """评估生态价值"""
        # 连通的源地质量
        source_quality = self._connected_source_quality(route, context)

        # 廊道宽度
        width_score = min(route['width'] / 100, 1.0)

        # 栖息地适宜性
        habitat_score = self._habitat_suitability(route, context)

        return (source_quality + width_score + habitat_score) / 3

    def _evaluate_community_impact(self, route, context):
        """评估社区影响"""
        # 正面：休闲价值
        recreational_value = self._recreational_potential(route)

        # 负面：拆迁、限制使用
        negative_impact = self._negative_community_impact(route, context)

        return max(0, recreational_value - negative_impact)

    def _evaluate_economic_cost(self, route, context):
        """评估经济成本（分数越高表示成本越可接受）"""
        # 土地获取成本
        land_cost = route.get('land_cost', 0)

        # 建设成本
        construction_cost = route.get('construction_cost', 0)

        # 归一化：成本越低分数越高
        max_cost = context.get('max_budget', float('inf'))
        total_cost = land_cost + construction_cost

        if total_cost > max_cost:
            return 0  # 超预算
        else:
            return 1 - (total_cost / max_cost) * 0.5

    def _evaluate_cultural_impact(self, route, context):
        """评估文化影响"""
        # 是否涉及文化遗产
        cultural_sites = route.get('cultural_sites', [])
        if cultural_sites:
            return 0.3  # 低分，需要特别处理

        # 是否支持传统文化活动
        traditional_use = route.get('supports_traditional_use', False)
        if traditional_use:
            return 1.0

        return 0.7  # 中性

    # ... 其他辅助方法 ...

if __name__ == "__main__":
    print("=== 伦理导向的廊道选址 ===\n")

    selector = EthicalCorridorSelector()

    # 示例路线
    route = {
        'width': 80,
        'land_cost': 500000,
        'construction_cost': 1000000,
        'cultural_sites': [],
        'supports_traditional_use': True
    }

    context = {
        'max_budget': 2000000,
        'ecological_data': {},
        'community_data': {}
    }

    result = selector.evaluate_corridor_route(route, context)

    if result['acceptable']:
        print(f"路线可接受，综合评分: {result['total_score']:.2f}")
        print("\n各利益相关方评分:")
        for stakeholder, info in result['breakdown'].items():
            print(f"  {stakeholder}: {info['score']:.2f} "
                  f"(权重: {info['weight']}, 关注: {info['concern']})")
    else:
        print(f"路线不可接受: {result['reason']}")
```

### 案例3：透明度和可审计性

```python
"""
决策日志系统
"""
import json
from datetime import datetime
from typing import Dict, Any, List

class DecisionLog:
    """决策日志系统"""

    def __init__(self, project_id: str):
        self.project_id = project_id
        self.entries: List[Dict] = []

    def log_decision(self, decision_type: str, decision: Any,
                     rationale: str, alternatives: List[Dict],
                     metadata: Dict = None):
        """记录决策"""
        entry = {
            'timestamp': datetime.now().isoformat(),
            'project_id': self.project_id,
            'decision_type': decision_type,
            'decision': decision,
            'rationale': rationale,
            'alternatives': alternatives,
            'metadata': metadata or {}
        }
        self.entries.append(entry)

    def log_data_source(self, data_type: str, source: str,
                        quality: Dict, limitations: List[str]):
        """记录数据来源"""
        entry = {
            'timestamp': datetime.now().isoformat(),
            'project_id': self.project_id,
            'type': 'data_source',
            'data_type': data_type,
            'source': source,
            'quality': quality,
            'limitations': limitations
        }
        self.entries.append(entry)

    def log_model_info(self, model_name: str, version: str,
                      training_data: Dict, limitations: List[str]):
        """记录模型信息"""
        entry = {
            'timestamp': datetime.now().isoformat(),
            'project_id': self.project_id,
            'type': 'model_info',
            'model_name': model_name,
            'version': version,
            'training_data': training_data,
            'limitations': limitations
        }
        self.entries.append(entry)

    def export_audit_report(self) -> str:
        """导出审计报告"""
        report = {
            'project_id': self.project_id,
            'export_time': datetime.now().isoformat(),
            'entries': self.entries,
            'summary': self._generate_summary()
        }
        return json.dumps(report, indent=2, ensure_ascii=False)

    def _generate_summary(self) -> Dict:
        """生成摘要"""
        summary = {
            'total_entries': len(self.entries),
            'decision_types': {},
            'data_sources': [],
            'models_used': []
        }

        for entry in self.entries:
            if entry.get('type') == 'data_source':
                summary['data_sources'].append(entry['data_type'])
            elif entry.get('type') == 'model_info':
                summary['models_used'].append(entry['model_name'])
            elif 'decision_type' in entry:
                dt = entry['decision_type']
                summary['decision_types'][dt] = \
                    summary['decision_types'].get(dt, 0) + 1

        return summary
```

---

## 反思与延伸

### 思考问题

1. **价值权衡**：当生态目标和社会目标冲突时，应该如何权衡？谁有权决定？

2. **偏见识别**：你的空间分析可能隐含哪些偏见？如何检测？

3. **透明度边界**：哪些决策细节必须公开？哪些可以保密？

4. **长期责任**：AI辅助的决策出现问题后，如何追溯和纠正？

### 实践练习

1. **伦理审计**：对你做过的一个空间项目进行伦理审计

2. **利益相关者地图**：绘制项目的利益相关者及其关注点

3. **透明度检查**：为你的分析流程建立可审计的文档体系

### 延伸阅读

- **"Weapons of Math Destruction"** (Cathy O'Neil) - 算法的社会影响
- **"The Alignment Problem"** (Brian Christian) - AI对齐问题
- **"Spatial Justice"** (Edward Soja) - 空间正义理论
- UN-Habitat's ethics guidelines for spatial planning

---

## 关键要点

1. **空间决策具有深刻的伦理维度**，影响资源分配和社会正义
2. **AI可能放大既有偏见**，需要主动的公平性设计
3. **可解释性是负责任AI的基础**，决策过程应可审查
4. **建立清晰的问责机制**，明确各方责任
5. **伦理思考应该贯穿整个项目生命周期**，而非事后补充