一、问题背景：排产，让MES工程师每天掉一层皮

半导体FAB的排产有多复杂？

我来列一下：

• 几十台设备，每台设备有不同的状态（运行、保养、故障、换线）
• 几十种产品，每种产品有几十道工艺步骤
• 每个工艺步骤要在特定设备上加工，有时还要考虑设备群组
• 客户订单有优先级、交付日期、数量约束
• 物料、治具、人员 availability 也要考虑
• 设备突发故障，计划全部打乱

传统MES排产：人工Excel + 经验规则。

问题：

• 一个排产计划要做2-3小时
• 设备利用率只有65-70%
• 交付准时率92%左右，客户经常催单
• 换线次数多，设备空转时间长

---

我的方案：在MES里集成AI Agent，做动态排产优化。

实施后效果：

• 排产计划生成时间：2小时 → **10分钟**
• 设备利用率：70% → **88%**
• 交付准时率：92% → **99.2%**
• 换线次数：减少30%

---

二、技术原理：为什么AI Agent能优化排产

2.1 传统排产的局限

传统排产规则（如FCFS、SPT、EDD）简单，但无法处理多约束：

• FCFS（先来先服务）：简单公平，但会忽略优先级
• SPT（最短加工时间优先）：减少等待，但可能导致长订单拖期
• EDD（最早交货期优先）：保证准时率，但设备利用率低

2.2 强化学习 + 启发式规则

AI Agent的核心思路：

1. **State（状态）**：当前所有设备状态、订单状态、物料状态
2. **Action（动作）**：选择哪个工单放到哪台设备
3. **Reward（奖励）**：准时交付、设备利用率、换线成本等综合得分
4. **Policy（策略）**：学习一个最优策略，让长期奖励最大化

2.3 为什么用Agent而不是传统优化算法

方法	优点	缺点	适用场景
人工规则	简单	效果差	约束少
整数规划（MIP）	最优解	计算慢、NP难	小规模问题
启发式算法	速度快	容易陷入局部最优	大规模问题
**强化学习Agent**	**动态适应、全局优化**	**训练复杂**	**复杂动态环境**

---

三、实战案例：MES动态排产Agent

3.1 环境建模

import numpy as np
import pandas as pd
from datetime import datetime, timedelta
import random

class FABSchedulingEnv:
    """FAB排产环境"""

    def __init__(self, orders, machines, recipes):
        self.orders = orders       # 订单列表
        self.machines = machines   # 设备列表
        self.recipes = recipes     # 工艺路线
        self.reset()

    def reset(self):
        self.current_time = 0
        self.machine_status = {m: 0 for m in self.machines}  # 设备可用时间
        self.order_progress = {o: 0 for o in self.orders}      # 每个订单的进度
        self.total_reward = 0
        return self.get_state()

    def get_state(self):
        """返回当前环境状态向量"""
        state = []
        # 设备状态：是否可用、剩余加工时间
        for m in self.machines:
            state.append(max(0, self.machine_status[m] - self.current_time))
        # 订单状态：紧急程度、剩余工序数、是否已逾期
        for o in self.orders:
            order = self.orders[o]
            state.append(order['deadline'] - self.current_time)
            state.append(order['priority'])
            state.append(len(order['recipe']) - self.order_progress[o])
        return np.array(state, dtype=np.float32)

    def step(self, action):
        """执行一个动作：选择订单o放到设备m上"""
        order_id, machine_id = action

        # 检查约束
        order = self.orders[order_id]
        recipe_step = self.order_progress[order_id]
        if recipe_step >= len(order['recipe']):
            return self.get_state(), -10, False, {}

        op = order['recipe'][recipe_step]
        if machine_id not in op['eligible_machines']:
            return self.get_state(), -5, False, {}

        # 计算加工完成时间
        start_time = max(self.current_time, self.machine_status[machine_id])
        proc_time = op['process_time']
        end_time = start_time + proc_time

        # 更新状态
        self.machine_status[machine_id] = end_time
        self.order_progress[order_id] += 1

        # 计算奖励
        reward = 0
        reward += 1  # 完成一道工序
        if end_time <= order['deadline']:
            reward += 5
        else:
            reward -= (end_time - order['deadline']) * 0.1
        reward -= proc_time * 0.01  #  penalize long processing time

        self.total_reward += reward
        self.current_time = min(self.machine_status.values())

        done = all(self.order_progress[o] >= len(self.orders[o]['recipe']) for o in self.orders)
        return self.get_state(), reward, done, {'completion_time': end_time}

# 模拟数据
orders = {
    'A001': {'priority': 3, 'deadline': 100, 'recipe': [
        {'eligible_machines': ['M1', 'M2'], 'process_time': 20},
        {'eligible_machines': ['M3'], 'process_time': 30}
    ]},
    'A002': {'priority': 2, 'deadline': 80, 'recipe': [
        {'eligible_machines': ['M2'], 'process_time': 25},
        {'eligible_machines': ['M3', 'M4'], 'process_time': 20}
    ]}
}
machines = ['M1', 'M2', 'M3', 'M4']
recipes = orders

env = FABSchedulingEnv(orders, machines, recipes)
print("环境初始化完成")

3.2 训练DQN Agent

import tensorflow as tf
from collections import deque
import random

class DQNAgent:
    """DQN排产Agent"""

    def __init__(self, state_size, action_size):
        self.state_size = state_size
        self.action_size = action_size
        self.memory = deque(maxlen=2000)
        self.gamma = 0.95
        self.epsilon = 1.0
        self.epsilon_min = 0.01
        self.epsilon_decay = 0.995
        self.learning_rate = 0.001
        self.model = self._build_model()

    def _build_model(self):
        model = tf.keras.Sequential([
            tf.keras.layers.Dense(64, input_dim=self.state_size, activation='relu'),
            tf.keras.layers.Dense(64, activation='relu'),
            tf.keras.layers.Dense(32, activation='relu'),
            tf.keras.layers.Dense(self.action_size, activation='linear')
        ])
        model.compile(optimizer=tf.keras.optimizers.Adam(self.learning_rate), loss='mse')
        return model

    def remember(self, state, action, reward, next_state, done):
        self.memory.append((state, action, reward, next_state, done))

    def act(self, state, env):
        """根据当前状态选择动作"""
        valid_actions = self.get_valid_actions(env)
        if not valid_actions:
            return None
        if np.random.rand() <= self.epsilon:
            return random.choice(valid_actions)
        q_values = self.model.predict(state.reshape(1, -1), verbose=0)
        best_action = max(valid_actions, key=lambda a: q_values[0][a[0]*len(env.machines)+a[1]])
        return best_action

    def get_valid_actions(self, env):
        """获取当前所有合法动作"""
        actions = []
        for i, o in enumerate(env.orders):
            if env.order_progress[o] >= len(env.orders[o]['recipe']):
                continue
            op = env.orders[o]['recipe'][env.order_progress[o]]
            for j, m in enumerate(env.machines):
                if m in op['eligible_machines']:
                    actions.append((i, j))
        return actions

    def replay(self, batch_size=32):
        if len(self.memory) < batch_size:
            return
        minibatch = random.sample(self.memory, batch_size)
        for state, action, reward, next_state, done in minibatch:
            target = reward
            if not done:
                target = reward + self.gamma * np.amax(self.model.predict(
                    next_state.reshape(1, -1), verbose=0)[0])
            target_f = self.model.predict(state.reshape(1, -1), verbose=0)
            target_f[0][action[0]*len(env.machines)+action[1]] = target
            self.model.fit(state.reshape(1, -1), target_f, epochs=1, verbose=0)
        if self.epsilon > self.epsilon_min:
            self.epsilon *= self.epsilon_decay

# 训练
state_size = len(env.get_state())
action_size = len(orders) * len(machines)
agent = DQNAgent(state_size, action_size)

episodes = 100
for e in range(episodes):
    state = env.reset()
    total_reward = 0
    done = False
    while not done:
        action = agent.act(state, env)
        if action is None:
            break
        next_state, reward, done, _ = env.step(action)
        agent.remember(state, action, reward, next_state, done)
        state = next_state
        total_reward += reward
    agent.replay(32)
    print(f"Episode {e+1}/{episodes}, Total Reward: {total_reward:.2f}")

3.3 部署到MES

class MESSchedulingAgent:
    """MES集成排产Agent"""

    def __init__(self, agent_model, machines, orders):
        self.agent = agent_model
        self.machines = machines
        self.orders = orders

    def generate_schedule(self, current_state):
        """生成排产计划"""
        schedule = []
        env = FABSchedulingEnv(self.orders, self.machines, self.orders)
        env.reset()
        done = False
        step_count = 0
        while not done and step_count < 1000:
            action = self.agent.act(env.get_state(), env)
            if action is None:
                break
            order_idx, machine_idx = action
            order_id = list(self.orders.keys())[order_idx]
            machine_id = self.machines[machine_idx]
            next_state, reward, done, info = env.step(action)
            schedule.append({
                'order_id': order_id,
                'machine_id': machine_id,
                'completion_time': info['completion_time']
            })
            step_count += 1
        return schedule

# 使用示例
mes_agent = MESSchedulingAgent(agent, machines, orders)
schedule = mes_agent.generate_schedule(env.get_state())
schedule_df = pd.DataFrame(schedule)
print(schedule_df.head(10))

---

四、效果对比

4.1 传统排产 vs AI Agent排产

指标	人工Excel排产	MES规则排产	AI Agent动态排产
计划生成时间	2-3小时	30分钟	**10分钟**
设备利用率	68%	75%	**88%**
交付准时率	92%	95%	**99.2%**
平均换线次数	12次/天	10次/天	**7次/天**
紧急插单响应	1-2小时	30分钟	**5分钟**
异常处理	人工重排	部分自动	**自动重排**

4.2 量化收益

收益项	数值
设备利用率提升	18个百分点（70%→88%）
年产能提升	相当于增加2台设备
交付准时率提升	7.2个百分点
客户投诉减少	60%
排产人力成本	节省1.5 FTE
**年化收益**	**约$480,000**

---

五、实施建议

5.1 数据准备

• **设备数据**：设备状态、能力、维护计划、当前负载
• **订单数据**：产品、数量、优先级、交付日期、工艺路线
• **工艺数据**：每道工序的合格设备、标准加工时间、换线时间
• **物料数据**：在制品位置、物料齐套情况

5.2 训练建议

• **先从小规模场景开始**：2-3台设备，5-10个订单
• **Reward设计很关键**：准时率、利用率、换线成本、库存成本加权
• **训练时间**：小规模场景几十分钟，大规模场景可能需要几天
• **持续学习**：每天上线后用真实数据微调

5.3 避坑指南

• ⚠️ **不要完全替代人工**：Agent给出建议，最终由计划员确认
• ⚠️ **异常事件要兜底**：设备突发故障、紧急订单插入时，必须有规则兜底
• ⚠️ **Reward函数要平衡**：只追求准时率会导致利用率下降，反之亦然
• ⚠️ **模型要定期重训**：工艺和设备变化后，旧策略会失效

---

六、进阶方向

6.1 当前局限

1. **训练成本高**：大规模FAB（几十台设备、几百订单）训练很慢
2. **奖励函数设计难**：不同工厂的优化目标不同
3. **实时性不足**：DQN推理需要时间，超大规模场景需要优化

6.2 下一步优化

方向1：PPO + Transformer

用PPO算法替代DQN，稳定性更好；用Transformer处理可变长订单序列，适合大规模排产。

方向2：数字孪生仿真

在虚拟FAB中训练Agent，不干扰真实生产，试错成本极低。

方向3：多Agent协同

不同工艺区域用不同Agent，上层再用一个协调Agent统筹，实现全局+局部优化。