2026_DesignAI/officefile/md/supplements/01-foundations/01.4-feedback-and-learning.md

01.4 反馈与学习

核心问题

系统如何从经验中改进？ 强化学习的基本直觉是什么？ 如何设计一个好的奖励函数？

---

概念讲解

反馈循环

反馈是系统学习的基础机制：

      ┌─────────────────────────────────────────────────┐
      │                                                 │
      │    ┌─────────┐      ┌─────────┐      ┌─────────┐│
      │    │  Action │ ───→ │ Effect  │ ───→ │Reward   ││
      │    └─────────┘      └─────────┘      └─────────┘│
      │         │                                 │    │
      │         │            ┌────────────┐       │    │
      │         └───────────→│   Update   │←──────┘    │
      │                            ↑                   │
      │                            │                   │
      │                      ┌──────┴──────┐           │
      │                      │  Policy     │           │
      │                      │  Improvement│           │
      │                      └─────────────┘           │
      │                                                 │
      └─────────────────────────────────────────────────┘

反馈的类型：

类型	说明	例子
正反馈	强化正确行为	生态廊道有效，增加类似策略
负反馈	抑制错误行为	阻力面不合理，调整权重
延迟反馈	效果滞后	生态工程几年后才见效
隐式反馈	未明确标注	用户不使用某功能 = 不好用

强化学习的直觉

强化学习（RL）是关于"如何通过试错学习"：

强化学习核心概念

智能体    ──→  动作    ──→  环境    ──→  奖励
   ↑                                       │
   │                                       │
   └─────────────── 观察状态 ←──────────────┘
                     │
                     ↓
                  更新策略

关键要素：

1. 状态 (State)：智能体看到的当前情况
2. 动作 (Action)：智能体能做的事情
3. 奖励 (Reward)：动作好坏的即时反馈
4. 策略 (Policy)：状态到动作的映射规则
5. 价值函数 (Value)：对长期收益的估计

# RL的数学直觉

# 策略：在状态s采取动作a的概率
π(a|s) = P(action=a | state=s)

# 价值函数：从状态s开始的期望累积奖励
V(s) = E[Σ γ^t * r_t | s_0 = s]
# γ是折扣因子，平衡即时和长期奖励

# 动作价值函数：在状态s采取动作a后的期望累积奖励
Q(s,a) = E[Σ γ^t * r_t | s_0 = s, a_0 = a]

# 目标：找到最优策略，最大化累积奖励
π* = argmax_π V^π(s)

探索与利用的权衡

RL中经典的困境：

探索 (Explore) vs 利用 (Exploit)

    利用                          探索
    ↓                             ↓
选择已知最好的动作              尝试新动作
获得稳定奖励                    可能发现更好动作
可能错过最优                   可能浪费资源

策略：

策略	方法	适用场景
ε-greedy	以ε概率随机探索	通用，简单
Boltzmann	按价值概率选择	需要细粒度控制
UCB	上置信界选择	需要理论保证
Thompson Sampling	采样后验概率	贝叶斯框架

def epsilon_greedy_action(q_values, epsilon, n_actions):
    """
    ε-greedy策略

    Args:
        q_values: 各动作的估计价值
        epsilon: 探索概率
        n_actions: 动作数量

    Returns:
        选择的动作
    """
    if np.random.random() < epsilon:
        # 探索：随机选择
        return np.random.randint(n_actions)
    else:
        # 利用：选择价值最高的
        return np.argmax(q_values)

# ε的衰减策略
def epsilon_schedule(initial_epsilon, final_epsilon, total_steps, current_step):
    """线性衰减ε"""
    decay = (initial_epsilon - final_epsilon) / total_steps
    return max(final_epsilon, initial_epsilon - decay * current_step)

---

设计原理

奖励函数设计

奖励函数定义了"什么是好的行为"：

"""
奖励函数设计原则
"""

# 原则1: 清晰明确
# 好的奖励
def good_reward(ecological_quality):
    """生态质量越高，奖励越高"""
    return ecological_quality

# 不好的奖励（有歧义）
def bad_reward(ecological_quality, cost):
    """混合多个目标，可能冲突"""
    return ecological_quality - cost * 0.001

# 原则2: 适度塑形 (Reward Shaping)
# 不要过度引导，让智能体自己探索

def shaped_reward(base_reward, intermediate_metric):
    """
    基础奖励 + 形状奖励

    基础奖励：定义最终目标
    形状奖励：引导到达目标（权重较小）
    """
    return base_reward + 0.1 * intermediate_metric

# 原则3: 避免奖励黑客 (Reward Hacking)
# 防止智能体找到"作弊"方法

def safe_reward_with_constraints(action_result):
    """
    带约束的奖励
    """
    base_reward = action_result['quality']

    # 如果违反约束，给予惩罚
    if action_result['violates_constraint']:
        base_reward -= 100  # 大惩罚

    # 如果使用"作弊"方法，给予惩罚
    if action_result['uses_exploit']:
        base_reward -= 50

    return base_reward

# 原则4: 多目标平衡
def multi_objective_reward(ecological, economic, social, weights):
    """
    多目标加权

    Args:
        ecological: 生态效益
        economic: 经济效益
        social: 社会效益
        weights: 各目标权重

    Returns:
        综合奖励
    """
    # 归一化到[0,1]
    normalized = {
        'eco': min(ecological / 100, 1.0),
        'eco': min(economic / 1000, 1.0),
        'soc': min(social / 100, 1.0)
    }

    total = (weights['eco'] * normalized['eco'] +
             weights['eco'] * normalized['eco'] +
             weights['soc'] * normalized['soc'])

    return total

在空间分析中的应用

class SpatialOptimizerRL:
    """
    用强化学习优化空间布局

    场景：给定区域内选择最优生态廊道路线
    """

    def __init__(self, landscape, constraints):
        self.landscape = landscape
        self.constraints = constraints

        # 状态空间：当前的廊道路线
        # 动作空间：下一步走向哪个像元
        # 奖励：连通性、距离、穿越地类的综合

    def state_representation(self):
        """将当前空间格局转换为状态表示"""
        return {
            'current_position': self.current_position,
            'visited_cells': self.visited_cells,
            'local_context': self._get_local_context()
        }

    def available_actions(self):
        """获取可用的动作"""
        # 可以向8个方向移动
        directions = [
            (0, 1), (1, 0), (0, -1), (-1, 0),  # 上下左右
            (1, 1), (1, -1), (-1, 1), (-1, -1)  # 对角
        ]

        actions = []
        for dx, dy in directions:
            new_x = self.current_position[0] + dx
            new_y = self.current_position[1] + dy

            if self._is_valid_move(new_x, new_y):
                actions.append((new_x, new_y))

        return actions

    def reward_function(self, action, new_state):
        """
        定义奖励函数

        考虑：
        1. 穿越的土地类型（林地奖励，城市惩罚）
        2. 距离目标的远近（越近越好）
        3. 是否到达目标（大奖励）
        """
        x, y = new_state['position']

        # 1. 土地类型奖励/惩罚
        land_type = self.landscape[y, x]
        land_rewards = {
            'forest': 10,
            'grassland': 5,
            'wetland': 8,
            'agriculture': 0,
            'urban': -50,
            'water': -20
        }
        land_reward = land_rewards.get(land_type, -10)

        # 2. 距离奖励（离目标越近越好）
        dist_to_goal = self._distance_to_goal(new_state['position'])
        distance_reward = -dist_to_goal * 0.1

        # 3. 目标到达奖励
        goal_reward = 0
        if new_state['position'] == self.goal_position:
            goal_reward = 1000

        # 4. 约束惩罚
        constraint_penalty = 0
        if self._violates_constraint(new_state):
            constraint_penalty = -100

        # 总奖励
        total_reward = (land_reward + distance_reward +
                       goal_reward + constraint_penalty)

        return total_reward

    def train(self, n_episodes=1000):
        """
        训练智能体

        使用Q-learning
        """
        q_table = {}  # Q值表

        for episode in range(n_episodes):
            state = self._reset()
            epsilon = self._epsilon_schedule(episode)

            done = False
            while not done:
                # ε-greedy选择动作
                if np.random.random() < epsilon:
                    action = np.random.choice(self.available_actions())
                else:
                    # 选择Q值最高的动作
                    q_values = [q_table.get((state, a), 0)
                               for a in self.available_actions()]
                    action = self.available_actions()[np.argmax(q_values)]

                # 执行动作
                new_state, reward, done = self._step(action)

                # 更新Q值
                old_q = q_table.get((state, action), 0)
                max_next_q = max([q_table.get((new_state, a), 0)
                                 for a in self.available_actions()] + [0])

                # Q-learning更新公式
                q_table[(state, action)] = old_q + 0.1 * (
                    reward + 0.99 * max_next_q - old_q
                )

                state = new_state

        return q_table

---

代码示例

简化的生态网络优化RL

"""
简化版：用Q-learning优化生态源地选择
"""
import numpy as np
from typing import List, Dict, Tuple
import random

class EcologicalNetworkOptimizer:
    """
    生态网络优化器（RL简化版）

    问题：从候选源地中选择最优组合
    - 最大化总生态价值
    - 满足连通性要求
    - 预算约束
    """

    def __init__(self, candidate_sites: List[Dict], budget: float):
        self.candidate_sites = candidate_sites
        self.budget = budget

        # 动作：选择或不选择某个源地
        self.n_actions = len(candidate_sites)

        # 状态：已选源地列表
        # 简化：用位掩码表示状态
        self.n_states = 2 ** self.n_actions

        # Q表
        self.q_table = np.zeros((self.n_states, self.n_actions))

    def state_to_mask(self, state: int) -> List[bool]:
        """状态索引转位掩码"""
        return [(state >> i) & 1 for i in range(self.n_actions)]

    def mask_to_state(self, mask: List[bool]) -> int:
        """位掩码转状态索引"""
        state = 0
        for i, bit in enumerate(mask):
            if bit:
                state |= (1 << i)
        return state

    def available_actions(self, state_mask: List[bool]) -> List[int]:
        """获取可用动作（未选的源地）"""
        return [i for i, selected in enumerate(state_mask) if not selected]

    def reward_function(self, state_mask: List[bool]) -> float:
        """
        计算当前选择的奖励

        考虑：
        1. 总生态价值
        2. 连通性
        3. 预算约束
        """
        # 选中源地
        selected_sites = [self.candidate_sites[i]
                          for i, selected in enumerate(state_mask) if selected]

        if not selected_sites:
            return 0

        # 1. 总生态价值
        total_value = sum(site['value'] for site in selected_sites)

        # 2. 连通性（简化：已选源地之间的平均距离）
        if len(selected_sites) > 1:
            positions = [(site['x'], site['y']) for site in selected_sites]
            distances = []
            for i in range(len(positions)):
                for j in range(i + 1, len(positions)):
                    dist = np.sqrt((positions[i][0] - positions[j][0])**2 +
                                  (positions[i][1] - positions[j][1])**2)
                    distances.append(dist)
            avg_distance = np.mean(distances)
            connectivity_reward = -0.1 * avg_distance  # 距离越近越好
        else:
            connectivity_reward = 0

        # 3. 预算惩罚
        total_cost = sum(site['cost'] for site in selected_sites)
        budget_penalty = 0
        if total_cost > self.budget:
            budget_penalty = -100 * (total_cost - self.budget) / self.budget

        # 总奖励
        total_reward = total_value + connectivity_reward + budget_penalty

        return total_reward

    def step(self, state: int, action: int) -> Tuple[int, float, bool]:
        """
        执行一步

        Returns:
            next_state: 下一个状态
            reward: 奖励
            done: 是否结束
        """
        state_mask = self.state_to_mask(state)

        # 执行动作（选择一个源地）
        if state_mask[action]:  # 已经选过了
            return state, -100, True  # 惩罚并结束

        new_mask = state_mask.copy()
        new_mask[action] = True

        # 计算奖励
        reward = self.reward_function(new_mask)

        # 检查是否结束（预算用完或所有源地都选了）
        total_cost = sum(self.candidate_sites[i]['cost']
                        for i, selected in enumerate(new_mask) if selected)
        done = (total_cost >= self.budget) or (sum(new_mask) == len(new_mask))

        next_state = self.mask_to_state(new_mask)

        return next_state, reward, done

    def train(self, n_episodes=1000, alpha=0.1, gamma=0.99,
              epsilon_start=1.0, epsilon_end=0.01):
        """
        Q-learning训练

        Args:
            n_episodes: 训练回合数
            alpha: 学习率
            gamma: 折扣因子
            epsilon_start: 初始探索率
            epsilon_end: 最终探索率
        """
        for episode in range(n_episodes):
            # 线性衰减ε
            epsilon = epsilon_start - (epsilon_start - epsilon_end) * episode / n_episodes

            state = 0  # 初始状态（空）
            done = False

            while not done:
                state_mask = self.state_to_mask(state)
                available = self.available_actions(state_mask)

                if not available:
                    break

                # ε-greedy
                if np.random.random() < epsilon:
                    action = random.choice(available)
                else:
                    q_values = [self.q_table[state, a] for a in available]
                    action = available[np.argmax(q_values)]

                # 执行动作
                next_state, reward, done = self.step(state, action)

                # Q-learning更新
                old_q = self.q_table[state, action]
                next_max = np.max(self.q_table[next_state])

                self.q_table[state, action] = old_q + alpha * (
                    reward + gamma * next_max - old_q
                )

                state = next_state

            # 定期报告
            if episode % 100 == 0:
                current_epsilon = epsilon_start - (epsilon_start - epsilon_end) * episode / n_episodes
                print(f"Episode {episode}, ε={current_epsilon:.3f}, "
                      f"Best Q: {np.max(self.q_table[0]):.2f}")

        return self.q_table

    def get_solution(self) -> List[Dict]:
        """获取学习到的最优解"""
        state = 0
        state_mask = self.state_to_mask(state)
        solution = []

        while True:
            available = self.available_actions(state_mask)
            if not available:
                break

            # 选择Q值最高的动作
            q_values = [self.q_table[state, a] for a in available]
            action = available[np.argmax(q_values)]

            solution.append(self.candidate_sites[action])
            state, _, done = self.step(state, action)

            if done:
                break

        return solution

# 示例使用
def example_usage():
    """示例使用"""
    print("=== 生态网络优化：Q-learning ===\n")

    # 创建候选源地
    np.random.seed(42)
    n_candidates = 10
    candidates = []
    for i in range(n_candidates):
        candidates.append({
            'id': i,
            'x': np.random.randint(0, 100),
            'y': np.random.randint(0, 100),
            'value': np.random.randint(50, 150),
            'cost': np.random.randint(20, 80)
        })

    budget = 200

    print(f"候选源地数: {n_candidates}")
    print(f"预算: {budget}\n")

    # 创建优化器并训练
    optimizer = EcologicalNetworkOptimizer(candidates, budget)
    optimizer.train(n_episodes=500)

    # 获取解
    solution = optimizer.get_solution()

    print("\n=== 最优解 ===")
    print(f"选择源地数: {len(solution)}")
    total_value = sum(s['value'] for s in solution)
    total_cost = sum(s['cost'] for s in solution)
    print(f"总价值: {total_value}")
    print(f"总成本: {total_cost}")

    print("\n选择的源地:")
    for s in solution:
        print(f"  源地 {s['id']}: 价值={s['value']}, 成本={s['cost']}")

if __name__ == "__main__":
    example_usage()

---

案例分析

ENAgent中的反馈机制

class ENAgentFeedback:
    """
    ENAgent的反馈机制

    场景：生态网络迭代的改进
    """

    def __init__(self):
        self.iteration_history = []
        self.performance_metrics = []

    def collect_feedback(self, iteration, result, human_feedback):
        """
        收集每轮的反馈

        Args:
            iteration: 迭代次数
            result: 本轮结果
            human_feedback: 人类专家的反馈
        """
        feedback_record = {
            'iteration': iteration,
            'result': result,
            'human_feedback': human_feedback,
            'timestamp': time.time()
        }

        self.iteration_history.append(feedback_record)

    def analyze_feedback(self) -> Dict:
        """
        分析反馈，提取改进建议

        Returns:
            改进建议
        """
        if not self.iteration_history:
            return {}

        # 分析模式
        suggestions = {}

        # 1. 常见问题
        problem_counts = {}
        for record in self.iteration_history:
            for problem in record['human_feedback'].get('problems', []):
                problem_counts[problem] = problem_counts.get(problem, 0) + 1

        if problem_counts:
            common_problems = sorted(problem_counts.items(),
                                    key=lambda x: x[1], reverse=True)
            suggestions['common_problems'] = common_problems

        # 2. 趋势分析
        if len(self.iteration_history) > 1:
            recent_quality = self.iteration_history[-1]['result']['quality']
            previous_quality = self.iteration_history[-2]['result']['quality']

            if recent_quality > previous_quality:
                suggestions['trend'] = 'improving'
            else:
                suggestions['trend'] = 'stagnant_or_degrading'

        # 3. 参数调整建议
        suggestions['parameter_adjustments'] = self._suggest_adjustments()

        return suggestions

    def _suggest_adjustments(self) -> Dict:
        """建议参数调整"""
        # 基于反馈历史，建议如何调整参数
        # 这是一个简化示例
        return {
            'resistance_weights': 'consider adjusting urban weight',
            'source_threshold': 'might be too high/low'
        }

---

反思与延伸

思考问题

1. 延迟奖励：生态工程的效果多年后才显现，如何设计奖励函数？

2. 稀疏奖励：当大多数步骤没有明确反馈时，如何学习？

3. 多目标冲突：生态目标和经济目标冲突时，奖励函数如何平衡？

4. 人类反馈：如何整合人类专家的定性反馈？

实践练习

1. 奖励设计：为一个你熟悉的任务设计奖励函数

2. 调试RL：观察Q表的变化，理解学习过程

3. 探索策略：比较不同ε衰减策略的效果

延伸阅读

• "Reinforcement Learning: An Introduction" (Sutton & Barto) - RL圣经
• "Algorithms for Decision Making" (Mykel Kochenderfer) - 决策与RL
• **"Reward Shaping"**论文 - 奖励塑形理论

---

关键要点

1. 反馈是学习的基础机制，正反馈强化正确行为，负反馈纠正错误
2. 强化学习核心：状态、动作、奖励、策略、价值函数
3. 探索vs利用：经典困境，需要平衡策略
4. 奖励函数设计是RL的关键，定义了"什么是好的行为"
5. 在空间分析中：RL可用于优化布局、路径选择、参数调整