当前位置：首页 > news >正文

AMHS自动物料搬运系统实战

news 2026/6/8 22:25:19

在12英寸晶圆厂里，每天有数万片晶圆在各道工序间穿梭。这些晶圆价值昂贵（一片300mm晶圆售价可达数千元）、极易损坏（静电、颗粒污染都可能报废）、洁净度要求极高（Class 1环境）。靠人工搬运？不仅效率低，而且风险大。本文将深入剖析AMHS系统的技术原理、架构设计、算法实现，以及在某12英寸FAB的实际应用案例。

---

一、从"人肉物流"到"智慧物流"

1.1 一个真实的故事

2019年，某国产8英寸晶圆厂。

早晨8点，操作员老张推着一辆载有4片晶圆的小车，从清洗区走向光刻区。

"让一下！" 老张喊道。

通道狭窄，左右都是设备。两边操作员侧身让路。

"小心！" 老张的车轮卡了一下，整车晶圆差点翻倒。

好不容易送到光刻区，却发现设备还在运行上一个批次。

老张看了看手表：8:23。等吧。

...

等他回到清洗区时，已经是8:47。24分钟，就干了这一件事。

这一天，老张走了12趟，耗时接近5小时。

按每班10个操作员算，每天浪费50个工时。
一个月就是1500个工时。一年将近18000个工时。

这就是"人肉物流"的代价。

1.2 12英寸FAB的挑战

1.英寸（300mm）晶圆比8英寸大了将近一倍，重量增加了3倍多：

| 参数 | 8英寸 | 12英寸 | 差异 |
|------|:------:|:------:|:----:|
| 直径 | 200mm | 300mm | +50% |
| 面积 | 314cm² | 707cm² | +125% |
| 重量 | 约50g | 约125g | +150% |
| 传送方式 | 人工/简易传送带 | 必须自动搬运 | - |
| 洁净度要求 | Class 10 | Class 1 | 10倍 |

**关键问题：**
人推车搬运容易磕碰
人携带颗粒会污染FAB
人工响应速度慢
人工无法保证24小时运转

1.3 AMHS：FAB的"血管系统"

如果把FAB比作人体：
**机台** = 各个器官（执行加工任务）
**晶圆** = 血液（在器官间流转）
**AMHS** = 血管系统（自动传输血液）

AMHS = Automated Material Handling System = 自动物料搬运系统

┌─────────────────────────────────────────────────────────────┐
│ │
│ ┌─────┐ ┌─────┐ ┌─────┐ ┌─────┐ │
│ │机台1│◀───▶│机台2│◀───▶│机台3│◀───▶│机台4│ │
│ └──┬──┘ └──┬──┘ └──┬──┘ └──┬──┘ │
│ │ │ │ │ │
│ └─────────────┼─────────────┼─────────────┘ │
│ │ │
│ ═════════════════╪═════════════════════════ │
│ │ ← OHT轨道 │
│ ▼ │
│ ┌─────────────┐ │
│ │ OHT小车 │ │
│ │ ↑ FOUP │ │
│ │ ┌─────┐ │ │
│ │ │晶圆 │ │ │
│ │ └─────┘ │ │
│ └─────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘

---

二、AMHS核心组件详解

2.1 OHT：空中悬轨车

**OHT = Overhead Hoist Transport**

想象一下：工厂天花板上有一条"单轨铁路"，上面跑着一辆辆"空中吊车"。每辆吊车可以抓起晶圆的容器（FOUP），在空中自由移动，把晶圆送到任何工位。

┌─────────────────────┐
│ OHT小车 │
│ ┌───────────────┐ │
│ │ 行走轮组 │ │
│ └───────┬───────┘ │
│ │ │
│ ════════════ │ ← 悬轨
│ │ │
│ ┌───────┴───────┐ │
│ │ 升降机构 │ │
│ └───────┬───────┘ │
│ │ │
│ ┌───────┴───────┐ │
│ │ 夹取机构 │ │
│ └───────┬───────┘ │
│ │ │
│ ┌┴┐ │
│ │ │ ← FOUP│
│ └─┘ │
└─────────────────────┘

**OHT技术规格（典型值）：**

| 参数 | 规格 | 说明 |
|------|:-----:|------|
| 载重 | 25kg×4=100kg | 可同时搬运4个FOUP |
| 速度 | 1-3 m/s | 空载3m/s，满载1m/s |
| 定位精度 | ±1mm | 满足精密定位需求 |
| 洁净等级 | Class 1 | 无尘运行 |
| 行走方式 | 单轨双向 | 节省空间 |

**OHT的工作流程：**

1. 接收指令
MCS发送："去STOCKER-A取FOUP-001，送往CVD-01"

2. 行走到位
OHT从当前位置出发
沿轨道行驶到STOCKER-A上方
定位精度±1mm

3. 取货
下降升降机构
夹取机构抓住FOUP
提升FOUP离开端口

4. 运输
OHT携带FOUP沿轨道行驶
速度1-2m/s（满载状态）

5. 卸货
到达CVD-01上方
下降FOUP到设备端口
夹取机构释放
OHT上升离开

6. 返回
OHT沿轨道返回待命点
或执行下一个任务

全程耗时：约30-120秒（取决于距离）

2.2 Stocker：晶圆的"立体车库"

Stocker是AMHS系统中的"仓储节点"，类似于立体停车场的原理，只是存放的不是汽车，而是装有晶圆的FOUP。

俯视图

┌─────────────────────────────┐
│ ┌──┬──┬──┬──┬──┬──┬──┐ │ ← 货格阵列
│ │01│02│03│04│05│06│07│ │ （每个格子
│ ├──┼──┼──┼──┼──┼──┼──┤ │ 可放1个FOUP）
│ │08│09│10│11│12│13│14│ │
│ ├──┼──┼──┼──┼──┼──┼──┤ │
│ │15│16│17│18│19│20│21│ │
│ ├──┼──┼──┼──┼──┼──┼──┤ │
│ │22│23│24│25│26│27│28│ │
│ └──┴──┴──┴──┴──┴──┴──┘ │
│ │ │
│ 存取机 ◀─┼──▶ 交接位 │
│ (存取) │ (OHT/AGV) │
└────────────┴──────────────────┘

侧视图

┌─────────────────────────────┐
│ ┌─┐┌─┐┌─┐┌─┐┌─┐┌─┐┌─┐ │ ← 高度10-20米
│ └─┘└─┘└─┘└─┘└─┘└─┘└─┘ │ 可存放100-500个FOUP
│ ┌─┐┌─┐┌─┐┌─┐┌─┐┌─┐┌─┐ │
│ └─┘└─┘└─┘└─┘└─┘└─┘└─┘ │
│ ┌─┐┌─┐┌─┐┌─┐┌─┐┌─┐┌─┐ │
│ └─┘└─┘└─┘└─┘└─┘└─┘└─┘ │
│ │ │
│ ┌────────┴────────┐ │
│ │ 存取机 │ │ ← 垂直升降+水平移动
│ └────────┬────────┘ │
└─────────────┴──────────────────┘

**Stocker关键参数：**

| 参数 | 规格 | 说明 |
|------|:-----:|------|
| 存储容量 | 100-500 FOUP | 取决于高度和货格数 |
| 高度 | 10-20米 | 空间利用率高 |
| 取片时间 | 3-8秒 | 从命令到FOUP到达交接位 |
| 温度控制 | ±0.5°C | 满足工艺要求 |
| 湿度控制 | ±5% RH | 防止静电 |

**Stocker的三大功能：**

# Stocker的三大核心功能

class Stocker:
"""
晶圆仓储系统
"""

def store_fooup(self, fooup_id: str, port_id: str):
"""
1. 存储功能
- OHT送来FOUP
- 存取机取出FOUP
- 找到空闲货格
- 放入并记录位置
"""
pass

def retrieve_fooup(self, fooup_id: str, destination: str):
"""
2. 取货功能
- MCS请求取出FOUP
- 找到FOUP所在货格
- 存取机取出FOUP
- 送到交接位给OHT
"""
pass

def query_status(self) -> dict:
"""
3. 状态查询
- 各货格占用情况
- FOUP存放位置
- 温度湿度状态
- 存取机状态
"""
pass

2.3 AGV和RGV：地面搬运系统

**AGV = Automated Guided Vehicle（自动导引车）**
在地面行驶
适合较长距离运输
可以跨区域移动
速度较慢（约0.5-1m/s）

**RGV = Rail Guided Vehicle（轨道导引车）**
在固定轨道上行驶
比AGV更快更稳
适合相邻区域间的运输
速度1-2m/s

AGV（自动导引车） RGV（轨道导引车）

┌───────────────────┐ ══════════════════
│ │ │ │
│ ┌───────────┐ │ │ ┌───────┐ │
│ │ FOUP │ │ │ │ RGV │ │
│ └───────────┘ │ │ │ ↑ │ │
│ │ │ │ FOUP │ │
│ ○ ○ ○ ○ │ │ └───────┘ │
│ 轮子组 │ │ │
│ │ ══════════════════
│ ═══════════════ │ 固定轨道
│ 磁条/激光导航 │
└───────────────────┘

特点：自由路径特点：固定路径，更快更稳

2.4 MCS：系统大脑

**MCS = Material Control System（物料控制系统）**

MCS是AMHS的"指挥中心"，相当于城市交通的"交通指挥中心"：

┌─────────────────────────────────────────────────────────────┐
│ MCS主控系统 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 【任务管理层】 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 任务池 │ 任务队列 │ 任务历史 │ 异常处理 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │ │
│ 【调度核心】 │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌─────────┐│
│ │ 路径规划 │ │ 任务分配 │ │ 负载均衡 │ │ 优先级 ││
│ │ A*算法 │ │ 贪心/拍卖│ │ 动态调整 │ │ 调度 ││
│ └──────────┘ └──────────┘ └──────────┘ └─────────┘│
│ │ │
│ 【设备控制层】 │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌─────────┐│
│ │ OHT │ │ Stocker │ │ AGV │ │ RGV ││
│ │ 控制卡 │ │ 控制器 │ │ 控制卡 │ │ 控制器 ││
│ └──────────┘ └──────────┘ └──────────┘ └─────────┘│
│ │
│ 【通信接口】 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ SECS/GEM │ MQTT │ REST API │ OPC UA │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘

MCS与外部系统的交互：

# MCS与FAB其他系统的交互

class MCSInterface:
"""
MCS对外接口
"""

def receive_from_mes(self, transport_order: dict):
"""
接收来自MES的搬运指令
{
"order_id": "MO-20260607-001",
"wafer_id": "WAF-001",
"from_location": "STOCKER-A",
"to_location": "CVD-01",
"priority": "HIGH",
"due_time": "2026-06-07 14:00"
}
"""
pass

def send_to_apc(self, wafer_arrived_notification: dict):
"""
通知APC：物料已到位，可以开始加工
{
"equipment_id": "CVD-01",
"wafer_id": "WAF-001",
"arrival_time": "2026-06-07 10:30:15"
}
"""
pass

def report_to_mes(self, task_completed: dict):
"""
报告给MES：搬运任务完成
{
"order_id": "MO-20260607-001",
"status": "COMPLETED",
"completed_time": "2026-06-07 10:32:45",
"actual_transport_time": 150 # 秒
}
"""
pass

---

三、MCS调度算法深度解析

3.1 调度问题的数学建模

AMHS调度问题可以抽象为：

给定：
- N个搬运任务（晶圆需要从A位置搬到B位置）
- M个OHT车辆（可以在轨道上移动）
- K个优先级等级（URGENT/HIGH/NORMAL/LOW）
- 网络拓扑（轨道连接关系）

目标：
- 最小化平均任务响应时间
- 最大化设备利用率
- 确保高优先级任务优先处理

约束：
- 每个OHT同一时间只能执行一个任务
- 路径不能冲突
- 必须满足优先级约束

3.2 A*路径规划算法

在AMHS系统中，OHT需要在轨道网络中找最短路径。A*算法是经典的最优路径算法。

import heapq
from dataclasses import dataclass, field
from typing import List, Dict, Tuple, Optional
from enum import Enum
import time

class NodeType(Enum):
"""节点类型"""
STATION = "STATION" # 工位
STOCKER = "STOCKER" # 仓储
INTERSECTION =INTERSECTION" # 交汇点

@dataclass
class Node:
"""轨道网络节点"""
node_id: str
node_type: NodeType
position: Tuple[float, float] # (x, y) 坐标
neighbors: List[str] = field(default_factory=list) # 邻居节点列表
distance_to_goal: float = 0.0 # 到目标的启发式估计

@dataclass
class Edge:
"""轨道边"""
from_node: str
to_node: str
distance: float # 距离（米）
speed_limit: float = 2.0 # 速度限制（m/s）

@dataclass(order=True)
class PathNode:
"""A*算法中的节点"""
priority: float
current_node: str = field(compare=False)
path: List[str] = field(compare=False)
g_cost: float = field(compare=False) # 从起点到当前节点的实际代价
f_cost: float = field(compare=False) # g_cost + h_cost

def __lt__(self, other):
return self.priority < other.priority

class AStarPathFinder:
"""
A*路径规划算法
用于在AMHS轨道网络中找最优路径
"""

def __init__(self):
self.nodes: Dict[str, Node] = {}
self.edges: Dict[Tuple[str, str], Edge] = {}

def add_node(self, node: Node):
"""添加节点"""
self.nodes[node.node_id] = node

def add_edge(self, edge: Edge):
"""添加边（双向）"""
self.edges[(edge.from_node, edge.to_node)] = edge
# 双向添加
reverse_edge = Edge(
from_node=edge.to_node,
to_node=edge.from_node,
distance=edge.distance,
speed_limit=edge.speed_limit
)
self.edges[(edge.to_node, edge.from_node)] = reverse_edge

def _heuristic(self, node_id: str, goal_id: str) -> float:
"""
启发函数：使用欧几里得距离
"""
if node_id not in self.nodes or goal_id not in self.nodes:
return 0.0

node = self.nodes[node_id]
goal = self.nodes[goal_id]

dx = goal.position[0] - node.position[0]
dy = goal.position[1] - node.position[1]

return (dx**2 + dy**2) ** 0.5

def find_path(self, start: str, goal: str) -> Optional[List[str]]:
"""
A*算法找最优路径

返回：路径节点列表，如 ["STOCKER-A", "INT-01", "CVD-01"]
"""
if start not in self.nodes or goal not in self.nodes:
return None

# 初始化
open_set = []
closed_set = set()
g_costs = {start: 0.0}

start_node = self.nodes[start]
h_cost = self._heuristic(start, goal)
f_cost = g_costs[start] + h_cost

heapq.heappush(open_set, PathNode(
priority=f_cost,
current_node=start,
path=[start],
g_cost=g_costs[start],
f_cost=f_cost
))

while open_set:
# 取出f_cost最小的节点
current = heapq.heappop(open_set)

if current.current_node == goal:
return current.path

if current.current_node in closed_set:
continue

closed_set.add(current.current_node)

# 遍历邻居
for neighbor_id in self.nodes[current.current_node].neighbors:
if neighbor_id in closed_set:
continue

edge = self.edges.get((current.current_node, neighbor_id))
if not edge:
continue

# 计算新的g_cost
tentative_g = current.g_cost + edge.distance

if neighbor_id not in g_costs or tentative_g < g_costs[neighbor_id]:
g_costs[neighbor_id] = tentative_g
h_cost = self._heuristic(neighbor_id, goal)
f_cost = tentative_g + h_cost

new_path = current.path + [neighbor_id]

heapq.heappush(open_set, PathNode(
priority=f_cost,
current_node=neighbor_id,
path=new_path,
g_cost=tentative_g,
f_cost=f_cost
))

return None # 没有找到路径

def calculate_travel_time(self, path: List[str]) -> float:
"""
计算路径的预计行驶时间
"""
total_time = 0.0

for i in range(len(path) - 1):
edge = self.edges.get((path[i], path[i+1]))
if edge:
total_time += edge.distance / edge.speed_limit

return total_time

def demo_astar():
"""A*路径规划演示"""
print("=" * 60)
print(" A*路径规划算法演示")
print("=" * 60)

# 创建轨道网络
pathfinder = AStarPathFinder()

# 添加节点（模拟FAB布局）
network_layout = [
("STOCKER-A", "STOCKER", (0, 0)),
("STOCKER-B", "STOCKER", (100, 0)),
("CVD-01", "STATION", (50, 30)),
("CVD-02", "STATION", (80, 30)),
("ETCH-01", "STATION", (50, 60)),
("ETCH-02", "STATION", (80, 60)),
("PVD-01", "STATION", (65, 90)),
("INT-01", "INTERSECTION", (50, 15)),
("INT-02", "INTERSECTION", (65, 45)),
("INT-03", "INTERSECTION", (65, 75)),
]

for node_id, node_type, pos in network_layout:
pathfinder.add_node(Node(
node_id=node_id,
node_type=NodeType(node_type),
position=pos
))

# 添加边（轨道连接）
rail_connections = [
("STOCKER-A", "INT-01", 50),
("INT-01", "CVD-01", 30),
("INT-01", "CVD-02", 45),
("CVD-01", "INT-02", 20),
("CVD-02", "INT-02", 20),
("INT-02", "ETCH-01", 30),
("INT-02", "ETCH-02", 45),
("ETCH-01", "INT-03", 20),
("ETCH-02", "INT-03", 20),
("INT-03", "PVD-01", 25),
("STOCKER-B", "CVD-02", 80),
]

for from_node, to_node, distance in rail_connections:
pathfinder.add_edge(Edge(
from_node=from_node,
to_node=to_node,
distance=distance
))
# 自动添加邻居关系
pathfinder.nodes[from_node].neighbors.append(to_node)
pathfinder.nodes[to_node].neighbors.append(from_node)

print("\n[轨道网络]")
print(f" 节点数: {len(pathfinder.nodes)}")
print(f" 边数: {len(pathfinder.edges)//2}")

# 测试路径规划
test_cases = [
("STOCKER-A", "ETCH-01"),
("STOCKER-A", "PVD-01"),
("STOCKER-B", "PVD-01"),
("CVD-01", "ETCH-02"),
]

print("\n[路径规划测试]")
print("-" * 60)

for start, goal in test_cases:
path = pathfinder.find_path(start, goal)
travel_time = pathfinder.calculate_travel_time(path) if path else 0

print(f"\n{start} → {goal}")
if path:
path_str = " → ".join(path)
print(f" 路径: {path_str}")
print(f" 距离: {pathfinder.nodes[goal].position[0] - pathfinder.nodes[start].position[0]:.0f}m（估算）")
print(f" 预计时间: {travel_time:.1f}秒")
else:
print(" 无路径")

print("\n" + "=" * 60)
print(" A*算法特点：")
print(" 1. 保证找到最优路径")
print(" 2. 考虑速度限制")
print(" 3. 支持任意拓扑的网络")
print(" 4. 时间复杂度: O(E log V)")
print("=" * 60)

if __name__ == "__main__":
demo_astar()

3.3 任务调度：优先级队列

在AMHS中，往往有多个任务同时等待执行。MCS需要决定：先执行哪个？

from dataclasses import dataclass, field
from typing import List, Dict, Optional
from datetime import datetime, timedelta
from enum import Enum
import heapq
import uuid

class TaskPriority(Enum):
"""任务优先级"""
URGENT = 1 # 加急（VIP客户/测试片）
HIGH = 2 # 高优先级（紧急订单）
NORMAL = 3 # 普通（正常生产）
LOW = 4 # 低优先级（内部测试）

@dataclass
class TransportTask:
"""搬运任务"""
task_id: str
wafer_id: str
from_location: str
to_location: str
priority: TaskPriority
create_time: datetime
due_time: Optional[datetime] = None
assigned_vehicle: Optional[str] = None
status: str = "PENDING" # PENDING/ASSIGNED/IN_PROGRESS/COMPLETED/FAILED

def __lt__(self, other):
"""
优先级队列排序规则：
1. 优先级高的先执行
2. 同优先级，早创建的先执行
3. 有截止时间的，接近截止的先执行
"""
if self.priority.value != other.priority.value:
return self.priority.value < other.priority.value

# 有截止时间的优先
if self.due_time and other.due_time:
time_to_due_self = (self.due_time - datetime.now()).total_seconds()
time_to_due_other = (other.due_time - datetime.now()).total_seconds()
return time_to_due_self < time_to_due_other

return self.create_time < other.create_time

class MCSTaskScheduler:
"""
MCS任务调度器
使用优先级队列管理搬运任务
"""

def __init__(self):
self.tasks: List[TransportTask] = []
self.priority_queue: List[TransportTask] = []
self.task_history: List[TransportTask] = []

# 统计
self.stats = {
"total_created": 0,
"total_completed": 0,
"total_failed": 0,
"avg_wait_time": 0.0,
"avg_process_time": 0.0,
}

def create_task(self, wafer_id: str, from_loc: str, to_loc: str,
priority: TaskPriority = TaskPriority.NORMAL,
due_time: Optional[datetime] = None) -> TransportTask:
"""创建新任务"""
task = TransportTask(
task_id=f"TASK-{uuid.uuid4().hex[:8].upper()}",
wafer_id=wafer_id,
from_location=from_loc,
to_location=to_loc,
priority=priority,
create_time=datetime.now(),
due_time=due_time
)

self.tasks.append(task)
heapq.heappush(self.priority_queue, task)
self.stats["total_created"] += 1

return task

def get_next_task(self) -> Optional[TransportTask]:
"""
获取下一个待执行任务
按照优先级自动排序
"""
if not self.priority_queue:
return None

task = heapq.heappop(self.priority_queue)
return task

def assign_task(self, task: TransportTask, vehicle_id: str):
"""分配任务给指定车辆"""
task.assigned_vehicle = vehicle_id
task.status = "ASSIGNED"

def complete_task(self, task: TransportTask):
"""标记任务完成"""
task.status = "COMPLETED"
self.tasks.remove(task)
self.task_history.append(task)
self.stats["total_completed"] += 1

# 更新平均处理时间
process_time = (datetime.now() - task.create_time).total_seconds()
n = self.stats["total_completed"]
self.stats["avg_process_time"] = (
self.stats["avg_process_time"] * (n - 1) + process_time
) / n

def get_pending_tasks(self) -> List[TransportTask]:
"""获取所有待处理任务"""
return sorted(self.priority_queue, key=lambda t: t.priority.value)

def get_statistics(self) -> Dict:
"""获取调度统计"""
return {
"total_tasks": len(self.tasks),
"pending_tasks": len(self.priority_queue),
"completed_today": self.stats["total_completed"],
"failed_today": self.stats["total_failed"],
"avg_process_time": f"{self.stats['avg_process_time']:.1f}秒",
}

def demo_scheduler():
"""任务调度演示"""
print("=" * 60)
print(" MCS任务调度器演示")
print("=" * 60)

scheduler = MCSTaskScheduler()

# 模拟创建任务
print("\n[1] 创建搬运任务...")

tasks_to_create = [
# (wafer_id, from, to, priority, due_time)
("WAF-001", "STOCKER-A", "CVD-01", TaskPriority.URGENT,
datetime.now() + timedelta(minutes=10)),
("WAF-002", "CVD-01", "ETCH-01", TaskPriority.HIGH,
datetime.now() + timedelta(minutes=30)),
("WAF-003", "STOCKER-B", "PVD-01", TaskPriority.NORMAL, None),
("WAF-004", "ETCH-01", "METROLOGY-01", TaskPriority.LOW, None),
("WAF-005", "STOCKER-A", "CVD-02", TaskPriority.HIGH,
datetime.now() + timedelta(minutes=20)),
]

priority_names = {
TaskPriority.URGENT: "🔴加急",
TaskPriority.HIGH: "🟡高优先",
TaskPriority.NORMAL: "🟢普通",
TaskPriority.LOW: "⚪低优先"
}

for wafer, f, t, p, due in tasks_to_create:
task = scheduler.create_task(wafer, f, t, p, due)
print(f" {task.task_id}: {f}→{t} {priority_names[p]}")

# 显示待处理队列
print("\n[2] 任务调度顺序（按优先级）...")
print("-" * 60)

while True:
task = scheduler.get_next_task()
if not task:
break

print(f" {task.task_id}: {task.from_location}→{task.to_location} "
f"{priority_names[task.priority]}")

# 模拟分配给OHT
scheduler.assign_task(task, f"OHT-{hash(task.task_id) % 3 + 1:02d}")

# 模拟完成
scheduler.complete_task(task)

# 显示统计
print("\n[3] 调度统计...")
stats = scheduler.get_statistics()
print(f" 总任务数: {stats['total_tasks']}")
print(f" 待处理: {stats['pending_tasks']}")
print(f" 已完成: {stats['completed_today']}")
print(f" 平均处理时间: {stats['avg_process_time']}")

print("\n" + "=" * 60)
print(" 调度算法特点：")
print(" 1. 基于优先级队列，O(log n) 入队/出队")
print(" 2. 自动处理优先级排序")
print(" 3. 支持截止时间约束")
print(" 4. 保证高优先级任务优先执行")
print("=" * 60)

if __name__ == "__main__":
demo_scheduler()

---

四、实战案例：某12英寸FAB的AMHS实施

4.1 项目背景

**工厂概况：**
12英寸晶圆厂，月产能48,000片
主要产品：逻辑芯片（28nm-14nm）
FAB面积：约15,000平方米

**原有物流方式：**
人工搬运为主（约70%）
OHT覆盖约30%区域
操作员每班12人
OEE约73%

**痛点：**
人工搬运效率低（平均响应时间25分钟）
操作员劳动强度大
人为失误导致晶圆损坏（每月2-3次）
无法支撑产能扩张

4.2 实施方案

**第一阶段（3个月）：覆盖核心区域**

| 项目 | 数量 | 说明 |
|------|:----:|------|
| OHT轨道 | 1.5km | 覆盖CVD/ETCH/PVD核心区 |
| OHT车辆 | 12台 | 双轨OHT，可载2个FOUP |
| Stocker | 2台 | 每台300个存储位 |
| MCS系统 | 1套 | 主控系统 |

**第二阶段（6个月）：全面覆盖**

| 项目 | 数量 | 说明 |
|------|:----:|------|
| OHT轨道 | 3km | 覆盖全部工艺区 |
| OHT车辆 | 20台 | 增加备用车辆 |
| Stocker | 4台 | 增加存储容量 |
| AGV | 4台 | 跨区域运输 |

4.3 实施效果

**量化改善：**

| 指标 | 实施前 | 实施后 | 改善幅度 |
|------|:------:|:------:|:--------:|
| 平均响应时间 | 25分钟 | 1.5分钟 | **-94%** |
| 操作员数量 | 12人/班 | 4人/班 | **-67%** |
| OEE | 73% | 86% | **+13%** |
| 月产能 | 48,000片 | 56,000片 | **+17%** |
| 晶圆损坏率 | 0.05% | 0.01% | **-80%** |
| 每日搬运任务 | 800次 | 2500次 | **+213%** |

**年化经济效益计算：**

1. 人员节省
= (12 - 4) 人 × 3班 × 12万/人/年
= 8 × 3 × 12万
= 288万元/年

2. 产能提升
= (56000 - 48000) 片/月 × 12月 × 500元/片（利润）
= 8000 × 12 × 500
= 4800万元/年

3. 良率提升
= 48000片 × 12月 × 0.04% × 2000元/片
= 1152万元/年

4. 晶圆损坏减少
= (0.05% - 0.01%) × 48000片 × 12月 × 2000元/片
= 460.8万元/年

年化总效益：约 6700万元
投资回收期：约 8个月

---

五、与MES/APC系统的深度集成

5.1 系统交互全景

┌─────────────────────────────────────────────────────────────────┐
│ FAB智能制造系统集成 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────┐ │
│ │ MES │ ← 工单管理、生产计划、设备状态 │
│ └───┬─────┘ │
│ │ │
│ │ 搬运指令 │
│ ▼ │
│ ┌─────────┐ │
│ │ MCS │ ← 物料控制、任务调度、路径规划 │
│ └───┬─────┘ │
│ │ │
│ │ 物料到位/任务完成 │
│ ▼ │
│ ┌─────────┐ │
│ │ APC │ ← 工艺控制、R2R、FDC、VM │
│ └───┬─────┘ │
│ │ │
│ │ 开始加工/工艺参数 │
│ ▼ │
│ ┌─────────┐ │
│ │ 机台 │ ← CVD/ETCH/PVD/CMP等 │
│ └─────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘

5.2 关键交互场景

**场景1：新批次到达FAB入口**

1. MES创建工单
MES → "有新批次LOT-A001到达，共25片晶圆"

2. MCS分配Stocker仓位
MCS → "分配STOCKER-A的仓位12-15给LOT-A001"
OHT → 搬运晶圆到指定仓位

3. MES通知APC
MES → "批次LOT-A001已就位，可以开始加工"
APC → 准备工艺配方（R2R初始化）

4. MCS等待APC指令
APC → "第一道工序CVD完成，准备ETCH"
MCS → "正在搬运到ETCH设备"

**场景2：设备报警触发改道**

1. ETCH-01报警
机台 → "RF功率异常，设备停机"

2. APC通知MES
APC → "ETCH-01不可用，等待修复"

3. MES通知MCS
MES → "ETCH-01不可用，ETCH工序改到ETCH-02"

4. MCS重新规划路径
MCS → "原计划搬运到ETCH-01，现改为ETCH-02"
OHT → 调整路线，运往ETCH-02

5. 通知相关系统
MCS → MES: "搬运任务已重新分配"
MCS → APC: "物料将送达ETCH-02"

---

六、总结与展望

6.1 AMHS核心价值

| 价值维度 | 具体表现 |
|---------|---------|
| **效率提升** | 响应时间从25分钟降至1.5分钟（-94%） |
| **产能扩张** | 月产能从48K提升到56K（+17%） |
| **人员优化** | 操作员从12人减至4人（-67%） |
| **质量保障** | 晶圆损坏率降低80% |
| **OEE改善** | 从73%提升到86%（+13%） |

6.2 未来发展趋势

**1. AI调度智能化**
# 未来：使用强化学习优化调度策略
class AIRouteOptimizer:
"""
基于强化学习的路径优化
"""
def train(self, historical_data):
"""从历史数据中学习最优策略"""
# 状态：任务队列、设备状态、OHT位置
# 动作：分配任务给某OHT
# 奖励：任务完成时间、设备利用率
pass

def predict(self, current_state):
"""预测最优动作"""
pass

**2. 数字孪生**
建立FAB数字孪生模型
模拟各种场景下的物流效率
预测产能瓶颈
优化布局设计

**3. 全自动化FAB（灯塔工厂）**
晶圆从进厂到出厂全程自动搬运
与机器人集成
实现"熄灯工厂"（Lights Out Factory）

---

七、实施建议

7.1 实施路线图

| 阶段 | 时间 | 内容 | 投资 |
|------|:----:|------|:----:|
| 试点 | 3个月 | 核心区域OHT+MCS | 2000万 |
| 推广 | 6个月 | 全面覆盖 | 4000万 |
| 优化 | 3个月 | AI调度升级 | 500万 |

7.2 关键成功因素

2. **轨道规划要科学**：结合工艺流程，避免交叉
3. **MCS系统要稳定**：99.9%以上可用性
4. **与MES集成要顺畅**：实时数据同步
5. **运维团队要专业**：7×24小时支持

---

**更多资源：**
- 《AMHS系统设计完整指南》- VIP资源下载
- 《MCS调度算法深度解析》- 技术文档
- 《AMHS项目实施案例集》- 实战分享

---