关于 control-tower-design
面向供应链团队的设计技能，指导评估需求并规划控制塔的范围、成熟度与架构。涵盖规划、物流、采购、制造及端到端运营，并讨论实施挑战。
主要功能
- 01评估业务需求与运营范围
- 02区分五类控制塔
- 03规划五级成熟度路径
- 04设计数据与系统架构
- 05应对集成和采用挑战
适用场景
- 01确定控制塔覆盖哪些流程
- 02评估当前监控成熟度
- 03规划供应链可视化与预警
- 04制定分阶段实施方案
安装
复制下面的命令，在终端中运行。
$
npx skills add https:/安装后在 Claude Code 中输入/control-tower-design确认已加载。
详细介绍
控制塔设计
你是供应链控制塔设计与实施方面的专家。你的目标是帮助组织建立集中式可视化、监控和协调能力，以主动管理供应链运营，并快速应对中断。
初步评估
设计控制塔之前，请先了解：
-
业务驱动因素
- 控制塔要解决哪些问题？（可视化缺口、响应缓慢、效率低下）
- 预期收益是什么？（降低成本、改善服务、降低风险）
- 触发事件有哪些？（中断、增长、复杂性）
- 是否有高管支持和预算？
-
范围与覆盖
- 要覆盖哪些流程？（规划、执行、物流、采购）
- 地理覆盖范围如何？（区域、全球）
- 覆盖内部还是扩展至外部（供应商、合作伙伴、客户）？
- 是端到端覆盖还是针对特定领域？
-
当前状态
- 现有的可视化能力如何？
- 数据来源和系统有哪些？（ERP、TMS、WMS 等）
- 团队结构和技能如何？
- 绩效监控流程如何？
-
组织准备情况
- 跨职能协作的成熟度如何？
- 数据质量和集成能力如何？
- 变更管理资源是否充足？
- 技术基础设施如何？
控制塔框架
控制塔类型
1. 规划控制塔
- 需求与供应匹配
- 生产计划协调
- S&OP 流程管理
- 情景规划与假设分析
- 预测协作
2. 物流控制塔
- 运输可视化
- 货运追踪与监控
- 承运商绩效管理
- 货运支出优化
- 最后一公里配送协调
3. 采购控制塔
- 供应商绩效监控
- 订单追踪（从采购订单到交付）
- 供应商风险监控
- 合同合规
- 支出分析
4. 制造控制塔
- 生产计划监控
- 物料可用性追踪
- 质量监控
- 设备绩效
- 产率优化
5. 端到端控制塔
- 全面的供应链可视化
- 跨职能协调
- 集成式异常管理
- 多层级可视化
- 战略与运营集成
控制塔成熟度模型
级别 1：基础可视化
- 被动监控
- 手动收集数据
- 孤立的仪表板
- 集成有限
- 基于电子邮件的沟通
级别 2：集成监控
- 自动化数据馈送
- 整合式仪表板
- 异常提醒
- 基础分析
- 标准 KPI
级别 3：主动管理
- 预测分析
- 自动化工作流
- 根因分析
- 绩效优化
- 协作工具
级别 4：认知运营
- AI/ML 驱动的洞察
- 处方性建议
- 自动化决策
- 自愈流程
- 持续优化
级别 5：自主供应链
- 全面自动化运营
- 实时优化
- 最少人工干预
- 生态系统编排
- 数字孪生集成
控制塔设计架构
数据架构
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import json
class ControlTowerPlatform:
"""Supply Chain Control Tower platform for visibility and monitoring"""
def __init__(self):
self.data_sources = {}
self.orders = []
self.shipments = []
self.exceptions = []
self.kpis = {}
self.alerts = []
def register_data_source(self, source_name, source_type, update_frequency,
connection_params):
"""
Register data source for control tower
source_type: 'ERP', 'TMS', 'WMS', 'MES', 'API', 'EDI', etc.
update_frequency: minutes between updates
"""
self.data_sources[source_name] = {
'type': source_type,
'update_frequency_min': update_frequency,
'last_update': None,
'status': 'Connected',
'connection_params': connection_params,
'record_count': 0
}
def ingest_orders(self, orders_data):
"""
Ingest order data from various sources
orders_data: list of order dicts
"""
for order in orders_data:
order_enriched = {
'order_id': order['order_id'],
'customer_id': order.get('customer_id'),
'order_date': order['order_date'],
'requested_date': order['requested_date'],
'promised_date': order.get('promised_date', order['requested_date']),
'status': order.get('status', 'Open'),
'value': order.get('value', 0),
'priority': order.get('priority', 'Normal'),
'items': order.get('items', []),
'fulfillment_location': order.get('fulfillment_location'),
'shipment_id': order.get('shipment_id'),
'ingestion_timestamp': datetime.now()
}
# Calculate metrics
order_enriched['days_to_requested'] = self._calculate_days_between(
datetime.now(), order['requested_date']
)
# Flag exceptions
if order_enriched['days_to_requested'] < 0:
self._create_exception(
'ORDER_LATE',
f"Order {order['order_id']} is past requested date",
'High',
order_id=order['order_id']
)
self.orders.append(order_enriched)
def ingest_shipments(self, shipments_data):
"""
Ingest shipment tracking data
shipments_data: list of shipment dicts with tracking info
"""
for shipment in shipments_data:
shipment_enriched = {
'shipment_id': shipment['shipment_id'],
'order_ids': shipment.get('order_ids', []),
'carrier': shipment.get('carrier'),
'tracking_number': shipment.get('tracking_number'),
'origin': shipment.get('origin'),
'destination': shipment.get('destination'),
'ship_date': shipment.get('ship_date'),
'planned_delivery_date': shipment.get('planned_delivery_date'),
'actual_delivery_date': shipment.get('actual_delivery_date'),
'current_location': shipment.get('current_location'),
'status': shipment.get('status', 'In Transit'),
'milestones': shipment.get('milestones', []),
'ingestion_timestamp': datetime.now()
}
# Calculate delivery performance
if shipment_enriched['actual_delivery_date']:
shipment_enriched['delivery_performance_days'] = self._calculate_days_between(
shipment_enriched['planned_delivery_date'],
shipment_enriched['actual_delivery_date']
)
if shipment_enriched['delivery_performance_days'] < 0:
self._create_exception(
'SHIPMENT_DELAYED',
f"Shipment {shipment['shipment_id']} delivered late",
'Medium',
shipment_id=shipment['shipment_id']
)
# Check for in-transit issues
if shipment_enriched['status'] == 'Delayed':
self._create_exception(
'SHIPMENT_DELAYED',
f"Shipment {shipment['shipment_id']} is experiencing delays",
'High',
shipment_id=shipment['shipment_id']
)
self.shipments.append(shipment_enriched)
def _calculate_days_between(self, date1, date2):
"""Calculate days between two dates (date2 - date1)"""
if isinstance(date1, str):
date1 = datetime.strptime(date1, '%Y-%m-%d')
if isinstance(date2, str):
date2 = datetime.strptime(date2, '%Y-%m-%d')
return (date2 - date1).days
def _create_exception(self, exception_type, description, severity,
order_id=None, shipment_id=None, supplier_id=None):
"""Create exception for exception management"""
exception = {
'exception_id': f"EXC_{len(self.exceptions) + 1:06d}",
'timestamp': datetime.now(),
'type': exception_type,
'description': description,
'severity': severity,
'status': 'Open',
'order_id': order_id,
'shipment_id': shipment_id,
'supplier_id': supplier_id,
'assigned_to': None,
'resolution': None,
'resolution_time': None
}
self.exceptions.append(exception)
# Create alert if high severity
if severity in ['High', 'Critical']:
self._create_alert(exception)
def _create_alert(self, exception):
"""Create alert for critical exceptions"""
alert = {
'alert_id': f"ALT_{len(self.alerts) + 1:06d}",
'timestamp': datetime.now(),
'exception_id': exception['exception_id'],
'type': exception['type'],
'severity': exception['severity'],
'message': exception['description'],
'status': 'New',
'acknowledged_by': None,
'acknowledgment_time': None
}
self.alerts.append(alert)
def calculate_kpis(self):
"""Calculate key performance indicators"""
# Order KPIs
total_orders = len(self.orders)
late_orders = len([o for o in self.orders if o['days_to_requested'] < 0])
on_time_pct = ((total_orders - late_orders) / total_orders * 100) if total_orders > 0 else 100
# Shipment KPIs
total_shipments = len(self.shipments)
delivered_shipments = [s for s in self.shipments if s.get('actual_delivery_date')]
late_deliveries = len([s for s in delivered_shipments if s.get('delivery_performance_days', 0) < 0])
otd_pct = ((len(delivered_shipments) - late_deliveries) / len(delivered_shipments) * 100) if delivered_shipments else 100
# Exception KPIs
open_exceptions = len([e for e in self.exceptions if e['status'] == 'Open'])
high_severity_exceptions = len([e for e in self.exceptions if e['severity'] in ['High', 'Critical']])
self.kpis = {
'orders': {
'total': total_orders,
'late': late_orders,
'on_time_percentage': round(on_time_pct, 1)
},
'shipments': {
'total': total_shipments,
'delivered': len(delivered_shipments),
'late_deliveries': late_deliveries,
'on_time_delivery_percentage': round(otd_pct, 1)
},
'exceptions': {
'total': len(self.exceptions),
'open': open_exceptions,
'high_severity': high_severity_exceptions
},
'alerts': {
'total': len(self.alerts),
'new': len([a for a in self.alerts if a['status'] == 'New'])
}
}
return self.kpis
def get_exceptions_dashboard(self):
"""Generate exceptions dashboard"""
if not self.exceptions:
return pd.DataFrame()
df = pd.DataFrame(self.exceptions)
# Summary by type
by_type = df.groupby(['type', 'severity']).size().reset_index(name='count')
# Summary by status
by_status = df.groupby('status').size().reset_index(name='count')
return {
'all_exceptions': df,
'by_type': by_type,
'by_status': by_status,
'open_high_priority': df[(df['status'] == 'Open') & (df['severity'].isin(['High', 'Critical']))]
}
def get_alerts_requiring_action(self):
"""Get alerts that need immediate attention"""
new_alerts = [a for a in self.alerts if a['status'] == 'New']
return pd.DataFrame(new_alerts) if new_alerts else pd.DataFrame()
def generate_control_tower_dashboard(self):
"""Generate comprehensive control tower dashboard"""
kpis = self.calculate_kpis()
dashboard = {
'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
'kpis': kpis,
'data_sources': {
'connected': len([s for s in self.data_sources.values() if s['status'] == 'Connected']),
'total': len(self.data_sources)
},
'alerts': {
'new_alerts': kpis['alerts']['new'],
'requires_action': len([a for a in self.alerts if a['status'] == 'New'])
},
'exceptions': {
'open_exceptions': kpis['exceptions']['open'],
'high_severity': kpis['exceptions']['high_severity']
},
'performance_summary': {
'order_on_time': f"{kpis['orders']['on_time_percentage']}%",
'delivery_on_time': f"{kpis['shipments']['on_time_delivery_percentage']}%"
}
}
return dashboard
# Example usage
control_tower = ControlTowerPlatform()
# Register data sources
control_tower.register_data_source(
'ERP_System',
'ERP',
update_frequency=15,
connection_params={'host': 'erp.company.com', 'api_key': 'xxx'}
)
control_tower.register_data_source(
'TMS_System',
'TMS',
update_frequency=5,
connection_params={'host': 'tms.company.com', 'api_key': 'yyy'}
)
# Ingest orders
orders = [
{
'order_id': 'ORD001',
'customer_id': 'CUST001',
'order_date': '2025-12-01',
'requested_date': '2025-12-15',
'promised_date': '2025-12-15',
'status': 'Processing',
'value': 50000,
'priority': 'High'
},
{
'order_id': 'ORD002',
'customer_id': 'CUST002',
'order_date': '2025-11-20',
'requested_date': '2025-12-05', # Past date - will trigger exception
'promised_date': '2025-12-10',
'status': 'Delayed',
'value': 30000,
'priority': 'Normal'
}
]
control_tower.ingest_orders(orders)
# Ingest shipments
shipments = [
{
'shipment_id': 'SHP001',
'order_ids': ['ORD001'],
'carrier': 'FedEx',
'tracking_number': 'TRK123456',
'origin': 'Chicago, IL',
'destination': 'New York, NY',
'ship_date': '2025-12-10',
'planned_delivery_date': '2025-12-12',
'actual_delivery_date': '2025-12-14', # Late
'status': 'Delivered'
},
{
'shipment_id': 'SHP002',
'order_ids': ['ORD002'],
'carrier': 'UPS',
'tracking_number': 'TRK789012',
'origin': 'Los Angeles, CA',
'destination': 'Seattle, WA',
'ship_date': '2025-12-08',
'planned_delivery_date': '2025-12-10',
'status': 'Delayed', # Currently delayed
'current_location': 'Portland, OR'
}
]
control_tower.ingest_shipments(shipments)
# Generate dashboard
dashboard = control_tower.generate_control_tower_dashboard()
print("=== CONTROL TOWER DASHBOARD ===")
print(f"Timestamp: {dashboard['timestamp']}")
print(f"\nKPIs:")
print(f" Orders: {dashboard['kpis']['orders']['total']} total, {dashboard['kpis']['orders']['late']} late ({dashboard['performance_summary']['order_on_time']} on-time)")
print(f" Shipments: {dashboard['kpis']['shipments']['delivered']} delivered, {dashboard['kpis']['shipments']['late_deliveries']} late ({dashboard['performance_summary']['delivery_on_time']} on-time)")
print(f" Exceptions: {dashboard['kpis']['exceptions']['open']} open ({dashboard['kpis']['exceptions']['high_severity']} high severity)")
print(f" Alerts: {dashboard['alerts']['new_alerts']} new alerts requiring action")
# Get exceptions
exceptions_dashboard = control_tower.get_exceptions_dashboard()
print(f"\n\nOpen High-Priority Exceptions:")
if not exceptions_dashboard['open_high_priority'].empty:
print(exceptions_dashboard['open_high_priority'][['exception_id', 'type', 'description', 'severity']])
# Get alerts
alerts = control_tower.get_alerts_requiring_action()
print(f"\n\nAlerts Requiring Action:")
if not alerts.empty:
print(alerts[['alert_id', 'type', 'severity', 'message']])
工具与库
Python 库
数据处理：
pandas：数据操作numpy：数值计算sqlalchemy：数据库连接
实时处理：
kafka-python：Apache Kafka 集成paho-mqtt：MQTT 消息传递redis：实时数据存储
可视化：
dash：交互式仪表板plotly：交互式图表streamlit：仪表板应用bokeh：实时可视化
API 与集成：
requests：HTTP 请求fastapi：API 开发flask：Web 框架
商业软件
控制塔平台：
- Blue Yonder (JDA)：Luminate Control Tower
- Kinaxis：RapidResponse
- o9 Solutions：Digital brain platform
- E2open：Control tower platform
- SAP：Integrated Business Planning with Control Tower
- Oracle：Supply Chain Management Cloud
可视化平台：
- FourKites：实时可视化
- project44：运输可视化
- Shippeo：供应链可视化
- ClearMetal：预测性物流
- Descartes MacroPoint：货物追踪
集成与物联网：
- MuleSoft：API 集成
- Boomi：集成平台
- PTC ThingWorx：物联网平台
- AWS IoT：物联网服务
常见挑战与解决方案
挑战：数据集成复杂
问题：
- 存在多个互不关联的系统
- 数据格式和标准各不相同
- 实时数据与批处理数据并存
- 数据质量问题
解决方案：
- 分阶段实施集成（从关键系统开始）
- 主数据管理（MDM）
- API 优先架构
- 数据质量规则与验证
- 中间件/集成平台
- 标准化数据模型
挑战：组织孤岛
问题：
- 抵制集中控制
- 优先事项相互冲突
- 缺乏协作文化
- 权责不明确
解决方案：
- 高管支持与治理
- 跨职能团队结构
- 明确角色与职责（RACI）
- 共享 KPI 和激励措施
- 变更管理计划
- 通过快速见效的成果展现价值
挑战：警报疲劳
问题：
- 警报过多
- 误报
- 遗漏重要信号
- 响应能力下降
解决方案：
- 智能警报阈值
- 使用机器学习检测异常
- 警报优先级排序与路由
- 升级规则
- 警报聚合与抑制
- 持续调整
挑战：投资回报率论证
问题：
- 实施成本高
- 收益难以量化
- 回收周期长
- 投资项目相互竞争
解决方案：
- 从试点开始（范围有限）
- 量化收益（节省成本、保障收入）
- 与人工流程进行基准对比
- 价值案例研究与参考资料
- 分阶段投资
- 跟踪并传达成果
挑战：用户采用
问题：
- 抵制使用新工具
- 工作流受到干扰
- 需要培训
- 需要改变习惯
解决方案：
- 以用户为中心的设计
- 培训与赋能
- 变更倡导者
- 逐步推广
- 根据反馈持续改进
- 游戏化与激励措施
挑战：可扩展性
问题：
- 数据量不断增长
- 用户和用例不断增加
- 性能下降
- 基础设施成本增加
解决方案：
- 基于云的架构
- 微服务设计
- 数据归档与保留策略
- 性能优化
- 横向扩展
- 成本监控与优化
输出格式
控制塔仪表板
执行摘要：
- 供应链整体健康评分
- 需要采取行动的关键警报
- 关键绩效趋势
- 重大中断或风险
KPI 计分板：
| KPI | 当前值 | 目标值 | 趋势 | 状态 |
|---|---|---|---|---|
| 订单满足率 | 94.2% | 96.0% | ↘ -1.2% | ⚠ 未达标 |
| 准时交付率 | 89.5% | 92.0% | ↗ +0.8% | ⚠ 未达标 |
| 完美订单率 | 85.3% | 90.0% | → 0.0% | ⚠ 未达标 |
| 订单周期时间 | 4.2 天 | 4.0 天 | ↗ +0.1 | ⚠ 超标 |
| 供应链成本占收入比例 | 8.2% | 8.0% | ↘ -0.2% | ✓ 按计划进行 |
异常摘要：
| 类别 | 未处理 | 严重 | 高 | 中 | 超过 48 小时 |
|---|---|---|---|---|---|
| 订单 | 23 | 3 | 8 | 12 | 5 |
| 货运 | 18 | 2 | 6 | 10 | 3 |
| 库存 | 12 | 1 | 4 | 7 | 2 |
| 供应商 | 8 | 0 | 3 | 5 | 1 |
| 总计 | 61 | 6 | 21 | 34 | 11 |
关键警报：
| 警报 ID | 时间 | 类型 | 描述 | 负责人 | 状态 |
|---|---|---|---|---|---|
| ALT-0042 | 08:15 | 货运延误 | 集装箱滞留港口 — 影响 50 个订单 | John D. | 处理中 |
| ALT-0041 | 07:30 | 供应商问题 | 供应商 XYZ 停产 — 产能不足 | Mary S. | 新建 |
| ALT-0038 | 昨天 | 库存缺货 | DC3 的 SKU-12345 缺货 | Tom R. | 正在解决 |
货运追踪：
| 货运 ID | 起点 | 目的地 | 状态 | 预计到达时间 | 延误 | 风险 |
|---|---|---|---|---|---|---|
| SHP-10234 | 上海 | LA Port | 运输中 | 1 月 28 日 | 准时 | ✓ 低 |
| SHP-10235 | 孟买 | NY Port | 延误 | 1 月 30 日 | +3 天 | ⚠ 高 |
| SHP-10236 | 汉堡 | 芝加哥 | 海关扣留 | 待定 | +5 天 | ⚠⚠ 严重 |
供应商绩效：
| 供应商 | 准时交付率 | 质量 | 交付周期 | 风险评分 | 趋势 |
|---|---|---|---|---|---|
| Supplier A | 96.2% | 99.8% | 12 天 | 低 | ✓ 改善中 |
| Supplier B | 88.5% | 97.2% | 18 天 | 中 | ⚠ 恶化中 |
| Supplier C | 92.1% | 95.5% | 15 天 | 高 | → 稳定 |
要提出的问题
如果需要更多背景信息：
- 控制塔要解决什么业务问题？
- 要覆盖哪些运营范围？（计划、执行、物流等）
- 需要集成哪些数据源？（ERP、TMS、WMS、供应商等）
- 需要监控哪些关键 KPI？
- 用户和利益相关方有哪些？
- 当前的可视性和监控成熟度如何？
- 现有的技术基础设施是什么？
- 预算和时间安排是什么？
- 这是内部使用还是扩展到外部（供应商、客户）？
- 成功标准是什么？
相关技能
- track-and-trace：用于产品和货运追踪
- supplier-collaboration：用于供应商集成与沟通
- demand-supply-matching：用于供需平衡
- risk-mitigation：用于中断监控与响应
- network-design：用于优化控制塔范围
- inventory-optimization：用于库存可视性与管理
- route-optimization：用于运输监控
- demand-forecasting：用于集成需求可视性
评论
支持 Markdown