lean-manufacturing

精益制造

你是精益制造和卓越运营领域的专家。你的目标是帮助组织消除浪费、优化流动、缩短交付周期、提升质量，并建立持续改进的文化。

初步评估

在实施精益举措之前，先了解以下情况：

1. 当前状态

   • 存在哪些问题或低效之处？
   • 当前的流程流转和周期时间是多少？
   • 已知的浪费或瓶颈有哪些？
   • 存在哪些质量问题或缺陷率？

2. 组织背景

   • 制造类型是什么？（离散制造、流程制造、装配、作业车间）
   • 生产量和产品种类如何？
   • 当前对精益原则的掌握程度如何？
   • 领导层是否支持变革？

3. 改进目标

   • 主要目标是什么？（缩短交付周期、降低成本、提升质量）
   • 目标指标和改进目标是什么？
   • 可用的时间和资源有哪些？
   • 是追求快速见效，还是进行战略转型？

4. 范围

   • 涉及哪些流程或价值流？
   • 是覆盖整个工厂，还是试点区域？
   • 是否涉及供应链？
   • 组织文化是否已做好变革准备？

---

精益制造框架

精益核心原则

1. 价值

• 从客户的角度定义价值
• 客户愿意为哪些内容付费？
• 消除其他一切

2. 价值流

• 绘制所有步骤（增值和非增值）
• 识别并消除浪费
• 优化端到端流动

3. 流动

• 让增值步骤顺畅流动
• 消除中断、延迟和批量处理
• 尽可能采用单件流

4. 拉动

• 只生产客户需要的产品
• 根据实际消耗触发生产
• 避免过度生产

5. 尽善尽美

• 持续改进（Kaizen）
• 追求零缺陷、零浪费
• 让所有员工参与改进

7 大浪费（Muda）

1. 搬运

• 不必要的物料移动
• 多次搬运、距离过长
• 示例：物料在工厂内跨区域搬运、来回移动

2. 库存

• 过多的原材料、在制品和成品
• 掩盖问题（质量、可靠性、计划）
• 占用资金和空间

3. 动作

• 人员不必要的移动
• 伸手、行走、寻找
• 示例：布局不合理、杂乱无序、缺少工具

4. 等待

• 人员或机器的闲置时间
• 等待物料、信息或审批
• 工作负荷不平衡、瓶颈

5. 过度生产

• 生产超出需求或提前于需求
• 最严重的浪费（会导致其他浪费）
• 示例：大批量生产、按预测而非需求生产

6. 过度加工

• 完成超出客户要求的工作
• 过度追求质量、不必要的步骤
• 示例：额外审批、重复检验、过度设计

7. 缺陷

• 错误、返工、报废、退货
• 检验和纠正所需的工作
• 客户不满

其他浪费（第 8 种浪费）：

• 未发挥的人力潜能：未能发挥员工的想法和技能

---

价值流图（VSM）

当前状态 VSM

**目的：**记录当前流程流转，识别浪费

步骤：

1. 选择产品族或价值流
2. 走查流程（前往现场）
3. 绘制每个流程步骤
4. 收集数据（周期时间、换型时间、开机率、缺陷、库存）
5. 绘制物料流和信息流
6. 计算增值时间与非增值时间
7. 识别改进机会

VSM 分析代码：

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import matplotlib.patches as mpatches

class ValueStreamMap:
    """
    Value Stream Mapping analysis and visualization
    Calculate lead time, value-added ratio, and identify waste
    """

    def __init__(self, process_steps):
        """
        process_steps: list of dicts with process details

        Example:
        {
            'name': 'Cutting',
            'cycle_time': 45,  # seconds
            'changeover_time': 30,  # minutes
            'uptime': 0.85,
            'batch_size': 500,
            'operators': 2,
            'inventory_days': 2.5,
            'distance_to_next': 50  # meters
        }
        """
        self.process_steps = process_steps
        self.df = pd.DataFrame(process_steps)

    def calculate_metrics(self):
        """Calculate key VSM metrics"""

        # Total cycle time (value-added time)
        total_cycle_time = self.df['cycle_time'].sum()

        # Total lead time (includes waiting/inventory)
        total_lead_time_days = self.df['inventory_days'].sum()
        total_lead_time_seconds = total_lead_time_days * 24 * 3600

        # Value-added ratio
        va_ratio = (total_cycle_time / total_lead_time_seconds) * 100

        # Total distance traveled
        total_distance = self.df['distance_to_next'].sum()

        # Total operators
        total_operators = self.df['operators'].sum()

        # OEE (Overall Equipment Effectiveness)
        # Simplified: uptime * performance * quality (assume 100% quality for now)
        avg_uptime = self.df['uptime'].mean()

        metrics = {
            'total_cycle_time_sec': total_cycle_time,
            'total_cycle_time_min': total_cycle_time / 60,
            'total_lead_time_days': total_lead_time_days,
            'total_lead_time_hours': total_lead_time_days * 24,
            'value_added_ratio_pct': va_ratio,
            'total_distance_m': total_distance,
            'total_operators': total_operators,
            'avg_uptime_pct': avg_uptime * 100
        }

        return metrics

    def identify_waste(self):
        """Identify and quantify waste in the value stream"""

        waste_analysis = []

        # Transportation waste
        total_transport = self.df['distance_to_next'].sum()
        waste_analysis.append({
            'waste_type': 'Transportation',
            'quantity': total_transport,
            'unit': 'meters',
            'impact': 'High' if total_transport > 200 else 'Medium' if total_transport > 100 else 'Low'
        })

        # Inventory waste
        total_inventory_days = self.df['inventory_days'].sum()
        waste_analysis.append({
            'waste_type': 'Inventory',
            'quantity': total_inventory_days,
            'unit': 'days',
            'impact': 'High' if total_inventory_days > 10 else 'Medium' if total_inventory_days > 5 else 'Low'
        })

        # Waiting waste (from low uptime)
        avg_uptime = self.df['uptime'].mean()
        downtime_pct = (1 - avg_uptime) * 100
        waste_analysis.append({
            'waste_type': 'Waiting (Downtime)',
            'quantity': downtime_pct,
            'unit': 'percent',
            'impact': 'High' if downtime_pct > 20 else 'Medium' if downtime_pct > 10 else 'Low'
        })

        # Overproduction (large batches)
        avg_batch = self.df['batch_size'].mean()
        waste_analysis.append({
            'waste_type': 'Overproduction',
            'quantity': avg_batch,
            'unit': 'units',
            'impact': 'High' if avg_batch > 1000 else 'Medium' if avg_batch > 500 else 'Low'
        })

        # Value-added ratio (overall waste indicator)
        metrics = self.calculate_metrics()
        va_ratio = metrics['value_added_ratio_pct']
        waste_analysis.append({
            'waste_type': 'Overall VA Ratio',
            'quantity': va_ratio,
            'unit': 'percent',
            'impact': 'Excellent' if va_ratio > 10 else 'Good' if va_ratio > 5 else 'Poor'
        })

        return pd.DataFrame(waste_analysis)

    def calculate_takt_time(self, customer_demand_per_day, available_work_time_minutes):
        """
        Calculate takt time = available time / customer demand

        Takt time is the pace of customer demand
        Process should be designed to match takt time
        """

        takt_time_minutes = available_work_time_minutes / customer_demand_per_day
        takt_time_seconds = takt_time_minutes * 60

        # Compare to current cycle time
        total_cycle_time = self.df['cycle_time'].sum()

        return {
            'takt_time_seconds': takt_time_seconds,
            'takt_time_minutes': takt_time_minutes,
            'current_cycle_time_seconds': total_cycle_time,
            'cycle_time_vs_takt': 'OK' if total_cycle_time <= takt_time_seconds else 'TOO SLOW',
            'margin_seconds': takt_time_seconds - total_cycle_time
        }

    def generate_future_state_recommendations(self):
        """Generate improvement recommendations for future state VSM"""

        recommendations = []

        metrics = self.calculate_metrics()

        # Inventory reduction
        if metrics['total_lead_time_days'] > 5:
            recommendations.append({
                'area': 'Inventory Reduction',
                'current': f"{metrics['total_lead_time_days']:.1f} days",
                'target': f"{metrics['total_lead_time_days'] * 0.5:.1f} days",
                'actions': [
                    'Implement pull/kanban system',
                    'Reduce batch sizes',
                    'Improve supplier delivery frequency',
                    'Eliminate safety stock where possible'
                ]
            })

        # Layout optimization
        if metrics['total_distance_m'] > 100:
            recommendations.append({
                'area': 'Layout Optimization',
                'current': f"{metrics['total_distance_m']:.0f} meters",
                'target': f"{metrics['total_distance_m'] * 0.3:.0f} meters",
                'actions': [
                    'Cellular manufacturing layout',
                    'Collocate sequential processes',
                    'Point-of-use material storage',
                    'Eliminate backtracking'
                ]
            })

        # Flow improvement
        if metrics['value_added_ratio_pct'] < 5:
            recommendations.append({
                'area': 'Flow Improvement',
                'current': f"{metrics['value_added_ratio_pct']:.2f}% VA ratio",
                'target': '>10% VA ratio',
                'actions': [
                    'Implement continuous flow (one-piece flow)',
                    'Eliminate batching where possible',
                    'Balance workloads across processes',
                    'Create FIFO lanes between processes'
                ]
            })

        # Equipment reliability
        if metrics['avg_uptime_pct'] < 85:
            recommendations.append({
                'area': 'Equipment Reliability',
                'current': f"{metrics['avg_uptime_pct']:.1f}% uptime",
                'target': '>95% uptime',
                'actions': [
                    'Implement TPM (Total Productive Maintenance)',
                    'Autonomous maintenance by operators',
                    'Preventive maintenance schedule',
                    'Root cause analysis on breakdowns'
                ]
            })

        return recommendations

# Example usage
process_steps = [
    {
        'name': 'Raw Material Storage',
        'cycle_time': 0,
        'changeover_time': 0,
        'uptime': 1.0,
        'batch_size': 5000,
        'operators': 0,
        'inventory_days': 5.0,
        'distance_to_next': 80
    },
    {
        'name': 'Cutting',
        'cycle_time': 45,
        'changeover_time': 30,
        'uptime': 0.85,
        'batch_size': 500,
        'operators': 2,
        'inventory_days': 2.5,
        'distance_to_next': 50
    },
    {
        'name': 'Welding',
        'cycle_time': 120,
        'changeover_time': 45,
        'uptime': 0.78,
        'batch_size': 500,
        'operators': 3,
        'inventory_days': 3.0,
        'distance_to_next': 30
    },
    {
        'name': 'Assembly',
        'cycle_time': 180,
        'changeover_time': 20,
        'uptime': 0.92,
        'batch_size': 100,
        'operators': 4,
        'inventory_days': 1.5,
        'distance_to_next': 25
    },
    {
        'name': 'Inspection',
        'cycle_time': 60,
        'changeover_time': 0,
        'uptime': 0.95,
        'batch_size': 100,
        'operators': 1,
        'inventory_days': 0.5,
        'distance_to_next': 15
    },
    {
        'name': 'Finished Goods',
        'cycle_time': 0,
        'changeover_time': 0,
        'uptime': 1.0,
        'batch_size': 100,
        'operators': 0,
        'inventory_days': 4.0,
        'distance_to_next': 0
    }
]

vsm = ValueStreamMap(process_steps)

# Calculate metrics
metrics = vsm.calculate_metrics()
print("Value Stream Metrics:")
print(f"  Total Cycle Time: {metrics['total_cycle_time_min']:.1f} minutes")
print(f"  Total Lead Time: {metrics['total_lead_time_days']:.1f} days")
print(f"  Value-Added Ratio: {metrics['value_added_ratio_pct']:.2f}%")
print(f"  Total Distance: {metrics['total_distance_m']:.0f} meters")

# Identify waste
print("\nWaste Analysis:")
waste_df = vsm.identify_waste()
print(waste_df)

# Takt time
takt = vsm.calculate_takt_time(customer_demand_per_day=400, available_work_time_minutes=450)
print(f"\nTakt Time: {takt['takt_time_seconds']:.1f} seconds")
print(f"Current Cycle Time: {takt['current_cycle_time_seconds']:.0f} seconds")
print(f"Status: {takt['cycle_time_vs_takt']}")

# Recommendations
print("\nFuture State Recommendations:")
recommendations = vsm.generate_future_state_recommendations()
for rec in recommendations:
    print(f"\n{rec['area']}:")
    print(f"  Current: {rec['current']}")
    print(f"  Target: {rec['target']}")
    print(f"  Actions:")
    for action in rec['actions']:
        print(f"    - {action}")

---

5S 工作场所组织

5S 系统

1. 整理（Seiri）

• 区分必需品与非必需品
• 清除不必要的物品
• 开展红牌作战

2. 整顿（Seiton）

• 整理留下的物品
• 物有所归，各归其位
• 目视化管理、标签和影子板

3. 清扫（Seiso）

• 清洁并检查
• 将清洁作为检查的一种形式
• 识别异常

4. 清洁（Seiketsu）

• 制定标准和流程
• 通过目视化管控维持 5S
• 检查清单和审核

5. 素养（Shitsuke）

• 纪律与承诺
• 定期审核
• 持续改进标准

5S 评估工具

class FiveSAssessment:
    """
    5S workplace assessment and scoring
    """

    def __init__(self):
        self.criteria = {
            'Sort': [
                'Only needed items present in work area',
                'No unnecessary materials, tools, or equipment',
                'Clear distinction between needed and unneeded items',
                'Regular red tag events conducted'
            ],
            'Set in Order': [
                'Everything has a designated location',
                'Visual controls and labels in place',
                'Items stored near point of use',
                'Easy to find and return items'
            ],
            'Shine': [
                'Work area is clean and free of debris',
                'Equipment is clean and well-maintained',
                'Cleaning standards are defined',
                'Cleaning is part of daily routine'
            ],
            'Standardize': [
                'Standard procedures documented',
                'Visual management tools in place',
                'Responsibilities are clearly assigned',
                'Standards are easy to follow'
            ],
            'Sustain': [
                'Regular 5S audits conducted',
                'All team members follow 5S practices',
                'Continuous improvement of standards',
                'Management support visible'
            ]
        }

    def conduct_assessment(self, scores):
        """
        Conduct 5S assessment

        Parameters:
        - scores: dict {S: [ratings]} where rating is 1-5 for each criterion

        Example:
        scores = {
            'Sort': [4, 5, 3, 4],
            'Set in Order': [3, 4, 4, 3],
            ...
        }

        Returns assessment results with scores and recommendations
        """

        results = {}

        for s, ratings in scores.items():
            avg_score = np.mean(ratings)
            max_score = 5.0

            results[s] = {
                'score': avg_score,
                'percentage': (avg_score / max_score) * 100,
                'level': self._get_level(avg_score),
                'criteria_scores': list(zip(self.criteria[s], ratings))
            }

        # Overall score
        overall = np.mean([r['score'] for r in results.values()])
        results['Overall'] = {
            'score': overall,
            'percentage': (overall / 5.0) * 100,
            'level': self._get_level(overall)
        }

        return results

    def _get_level(self, score):
        """Determine 5S maturity level"""
        if score >= 4.5:
            return 'World Class'
        elif score >= 4.0:
            return 'Excellent'
        elif score >= 3.5:
            return 'Good'
        elif score >= 3.0:
            return 'Fair'
        elif score >= 2.0:
            return 'Needs Improvement'
        else:
            return 'Poor'

    def generate_action_plan(self, results):
        """Generate improvement action plan based on assessment"""

        actions = []

        for s, result in results.items():
            if s == 'Overall':
                continue

            if result['score'] < 3.5:
                # Priority improvement area
                actions.append({
                    'area': s,
                    'priority': 'High' if result['score'] < 2.5 else 'Medium',
                    'current_score': result['score'],
                    'target_score': 4.0,
                    'gap': 4.0 - result['score'],
                    'focus_areas': [
                        criterion for criterion, score in result['criteria_scores']
                        if score < 3
                    ]
                })

        return sorted(actions, key=lambda x: x['gap'], reverse=True)

# Example usage
assessment = FiveSAssessment()

scores = {
    'Sort': [4, 4, 3, 4],
    'Set in Order': [3, 3, 4, 3],
    'Shine': [5, 4, 5, 4],
    'Standardize': [2, 3, 2, 3],
    'Sustain': [2, 2, 3, 2]
}

results = assessment.conduct_assessment(scores)

print("5S Assessment Results:\n")
for s, result in results.items():
    print(f"{s}: {result['score']:.2f} / 5.00 ({result['percentage']:.1f}%) - {result['level']}")

print("\n\nAction Plan:")
action_plan = assessment.generate_action_plan(results)
for action in action_plan:
    print(f"\n{action['area']} - Priority: {action['priority']}")
    print(f"  Current Score: {action['current_score']:.2f}")
    print(f"  Target Score: {action['target_score']:.2f}")
    print(f"  Focus Areas:")
    for area in action['focus_areas']:
        print(f"    - {area}")

---

看板拉动系统

看板计算

class KanbanSystem:
    """
    Kanban system design and calculation
    Determine number of kanban cards needed
    """

    def __init__(self, demand_rate, lead_time, container_size, safety_factor=1.1):
        """
        Parameters:
        - demand_rate: average daily demand (units/day)
        - lead_time: replenishment lead time (days)
        - container_size: units per kanban container
        - safety_factor: buffer factor (typically 1.0 - 1.2)
        """
        self.demand_rate = demand_rate
        self.lead_time = lead_time
        self.container_size = container_size
        self.safety_factor = safety_factor

    def calculate_number_of_kanbans(self):
        """
        Calculate number of kanban cards required

        Formula: N = (D × L × (1 + S)) / C
        Where:
        - N = number of kanbans
        - D = demand rate
        - L = lead time
        - S = safety factor (0.1 = 10% buffer)
        - C = container size
        """

        n_kanbans = (self.demand_rate * self.lead_time * self.safety_factor) / self.container_size

        # Round up
        n_kanbans = int(np.ceil(n_kanbans))

        # Calculate inventory levels
        max_inventory = n_kanbans * self.container_size
        avg_inventory = max_inventory / 2
        safety_stock = max_inventory - (self.demand_rate * self.lead_time)

        return {
            'number_of_kanbans': n_kanbans,
            'container_size': self.container_size,
            'max_inventory': max_inventory,
            'average_inventory': avg_inventory,
            'safety_stock': safety_stock,
            'inventory_days': max_inventory / self.demand_rate,
            'turnover_rate': self.demand_rate * 365 / avg_inventory
        }

    def optimize_container_size(self, min_size=10, max_size=200, step=10):
        """
        Find optimal container size to minimize inventory
        while meeting service level
        """

        results = []

        for container_size in range(min_size, max_size + 1, step):
            self.container_size = container_size
            kanban_calc = self.calculate_number_of_kanbans()

            results.append({
                'container_size': container_size,
                'num_kanbans': kanban_calc['number_of_kanbans'],
                'avg_inventory': kanban_calc['average_inventory'],
                'inventory_days': kanban_calc['inventory_days'],
                'turnover_rate': kanban_calc['turnover_rate']
            })

        df = pd.DataFrame(results)

        # Find optimal (minimum average inventory while ensuring coverage)
        optimal = df.loc[df['avg_inventory'].idxmin()]

        return {
            'optimal_container_size': optimal['container_size'],
            'optimal_num_kanbans': optimal['num_kanbans'],
            'analysis': df
        }

    def simulate_kanban_system(self, days=30, demand_variation=0.1):
        """
        Simulate kanban system performance

        Parameters:
        - days: simulation period
        - demand_variation: coefficient of variation for demand
        """

        n_kanbans = self.calculate_number_of_kanbans()['number_of_kanbans']

        # Initialize
        inventory = n_kanbans * self.container_size
        stockouts = 0
        daily_inventory = []

        for day in range(days):
            # Random demand (normal distribution)
            daily_demand = max(0, np.random.normal(
                self.demand_rate,
                self.demand_rate * demand_variation
            ))

            # Consume inventory
            inventory -= daily_demand

            # Replenishment (if kanban triggered and lead time elapsed)
            # Simplified: replenish when below reorder point
            reorder_point = self.demand_rate * self.lead_time
            if inventory < reorder_point:
                # Replenish (assuming lead time already passed)
                inventory += self.container_size

            # Check for stockout
            if inventory < 0:
                stockouts += 1
                inventory = 0

            daily_inventory.append(inventory)

        return {
            'avg_inventory': np.mean(daily_inventory),
            'min_inventory': np.min(daily_inventory),
            'max_inventory': np.max(daily_inventory),
            'stockout_days': stockouts,
            'service_level': (days - stockouts) / days * 100,
            'daily_inventory': daily_inventory
        }

# Example usage
kanban = KanbanSystem(
    demand_rate=100,      # 100 units per day
    lead_time=3,          # 3 days replenishment time
    container_size=50,    # 50 units per container
    safety_factor=1.1     # 10% buffer
)

result = kanban.calculate_number_of_kanbans()
print("Kanban System Design:")
print(f"  Number of Kanbans: {result['number_of_kanbans']}")
print(f"  Container Size: {result['container_size']} units")
print(f"  Maximum Inventory: {result['max_inventory']} units")
print(f"  Average Inventory: {result['average_inventory']:.0f} units")
print(f"  Inventory Days: {result['inventory_days']:.1f} days")
print(f"  Turnover Rate: {result['turnover_rate']:.1f}x per year")

# Optimize container size
print("\nOptimizing Container Size...")
optimization = kanban.optimize_container_size(min_size=20, max_size=100, step=10)
print(f"  Optimal Container Size: {optimization['optimal_container_size']} units")
print(f"  Optimal Number of Kanbans: {optimization['optimal_num_kanbans']}")

# Simulate
print("\nSimulating Kanban System (30 days)...")
simulation = kanban.simulate_kanban_system(days=30)
print(f"  Average Inventory: {simulation['avg_inventory']:.1f} units")
print(f"  Service Level: {simulation['service_level']:.1f}%")
print(f"  Stockout Days: {simulation['stockout_days']}")

---

SMED（单分钟换模）

换型缩短分析

class SMEDAnalysis:
    """
    SMED (Single Minute Exchange of Die) changeover reduction
    Analyze and reduce setup/changeover times
    """

    def __init__(self, changeover_steps):
        """
        changeover_steps: list of dicts with step details

        Example:
        {
            'step': 'Remove old die',
            'type': 'Internal',  # Internal or External
            'duration_minutes': 15,
            'category': 'Removal'
        }
        """
        self.steps = pd.DataFrame(changeover_steps)

    def analyze_current_state(self):
        """Analyze current changeover process"""

        # Total time by internal/external
        time_by_type = self.steps.groupby('type')['duration_minutes'].sum()

        # Total changeover time
        total_internal = time_by_type.get('Internal', 0)
        total_external = time_by_type.get('External', 0)
        total_time = total_internal + total_external

        # Internal time is the bottleneck (machine downtime)
        # External can be done while machine is running

        analysis = {
            'total_changeover_time': total_time,
            'internal_time': total_internal,
            'external_time': total_external,
            'internal_pct': (total_internal / total_time * 100) if total_time > 0 else 0,
            'steps_breakdown': self.steps.groupby(['type', 'category'])['duration_minutes'].sum()
        }

        return analysis

    def apply_smed_principles(self):
        """
        Apply SMED principles to reduce changeover time

        SMED Stages:
        1. Separate internal from external activities
        2. Convert internal to external
        3. Streamline internal activities
        4. Streamline external activities
        """

        recommendations = []

        # Stage 1: Identify internal that should be external
        internal_steps = self.steps[self.steps['type'] == 'Internal']

        for idx, step in internal_steps.iterrows():
            if 'Fetch' in step['step'] or 'Prepare' in step['step'] or 'Get' in step['step']:
                recommendations.append({
                    'step': step['step'],
                    'current_time': step['duration_minutes'],
                    'stage': 'Stage 2: Convert to External',
                    'action': 'Perform this step before stopping machine',
                    'time_saved': step['duration_minutes'],
                    'priority': 'High'
                })

        # Stage 3: Streamline internal
        for idx, step in internal_steps.iterrows():
            if step['duration_minutes'] > 5:
                recommendations.append({
                    'step': step['step'],
                    'current_time': step['duration_minutes'],
                    'stage': 'Stage 3: Streamline Internal',
                    'action': 'Use quick-release fasteners, standardize tools, use guides',
                    'time_saved': step['duration_minutes'] * 0.5,  # Assume 50% reduction
                    'priority': 'High' if step['duration_minutes'] > 10 else 'Medium'
                })

        # Stage 4: Standardize and parallel operations
        recommendations.append({
            'step': 'General',
            'current_time': 0,
            'stage': 'Stage 4: Standardize',
            'action': 'Create changeover checklist, use multiple operators in parallel',
            'time_saved': self.steps['duration_minutes'].sum() * 0.2,  # Assume 20% improvement
            'priority': 'Medium'
        })

        return pd.DataFrame(recommendations)

    def calculate_improvement_impact(self, recommendations_df, annual_changeovers):
        """Calculate business impact of SMED improvements"""

        total_time_saved = recommendations_df['time_saved'].sum()

        current_analysis = self.analyze_current_state()
        current_time = current_analysis['total_changeover_time']
        improved_time = current_time - total_time_saved

        # Calculate productivity impact
        annual_time_saved_hours = (total_time_saved / 60) * annual_changeovers
        annual_time_saved_days = annual_time_saved_hours / 8

        # Assuming $100/hour downtime cost
        cost_per_hour = 100
        annual_cost_savings = annual_time_saved_hours * cost_per_hour

        return {
            'current_changeover_time_min': current_time,
            'improved_changeover_time_min': improved_time,
            'time_reduction_min': total_time_saved,
            'improvement_pct': (total_time_saved / current_time * 100) if current_time > 0 else 0,
            'annual_changeovers': annual_changeovers,
            'annual_time_saved_hours': annual_time_saved_hours,
            'annual_time_saved_days': annual_time_saved_days,
            'annual_cost_savings': annual_cost_savings
        }

# Example usage
changeover_steps = [
    {'step': 'Fetch new die from storage', 'type': 'Internal', 'duration_minutes': 10, 'category': 'Preparation'},
    {'step': 'Get tools and equipment', 'type': 'Internal', 'duration_minutes': 5, 'category': 'Preparation'},
    {'step': 'Loosen bolts on old die', 'type': 'Internal', 'duration_minutes': 15, 'category': 'Removal'},
    {'step': 'Remove old die', 'type': 'Internal', 'duration_minutes': 20, 'category': 'Removal'},
    {'step': 'Clean machine surface', 'type': 'Internal', 'duration_minutes': 8, 'category': 'Cleaning'},
    {'step': 'Position new die', 'type': 'Internal', 'duration_minutes': 12, 'category': 'Installation'},
    {'step': 'Align and center die', 'type': 'Internal', 'duration_minutes': 18, 'category': 'Installation'},
    {'step': 'Tighten bolts', 'type': 'Internal', 'duration_minutes': 15, 'category': 'Installation'},
    {'step': 'Run test pieces', 'type': 'Internal', 'duration_minutes': 20, 'category': 'Adjustment'},
    {'step': 'Make adjustments', 'type': 'Internal', 'duration_minutes': 12, 'category': 'Adjustment'},
]

smed = SMEDAnalysis(changeover_steps)

# Current state
current = smed.analyze_current_state()
print("Current State Analysis:")
print(f"  Total Changeover Time: {current['total_changeover_time']:.0f} minutes")
print(f"  Internal Time (downtime): {current['internal_time']:.0f} minutes")
print(f"  External Time: {current['external_time']:.0f} minutes")

# Recommendations
print("\nSMED Recommendations:")
recommendations = smed.apply_smed_principles()
print(recommendations[['step', 'stage', 'action', 'time_saved', 'priority']])

# Impact
impact = smed.calculate_improvement_impact(recommendations, annual_changeovers=250)
print("\nImprovement Impact:")
print(f"  Current Changeover: {impact['current_changeover_time_min']:.0f} minutes")
print(f"  Improved Changeover: {impact['improved_changeover_time_min']:.0f} minutes")
print(f"  Improvement: {impact['improvement_pct']:.1f}%")
print(f"  Annual Time Saved: {impact['annual_time_saved_days']:.0f} days")
print(f"  Annual Cost Savings: ${impact['annual_cost_savings']:,.0f}")

---

Kaizen（持续改进）

Kaizen 活动框架

class KaizenEvent:
    """
    Kaizen event planning and tracking
    Rapid improvement workshop (typically 3-5 days)
    """

    def __init__(self, focus_area, current_metrics, target_metrics):
        """
        Parameters:
        - focus_area: description of improvement area
        - current_metrics: dict of baseline metrics
        - target_metrics: dict of target metrics
        """
        self.focus_area = focus_area
        self.current_metrics = current_metrics
        self.target_metrics = target_metrics
        self.improvements = []

    def calculate_improvement_potential(self):
        """Calculate improvement gap and potential"""

        improvements = {}

        for metric, current_value in self.current_metrics.items():
            target_value = self.target_metrics.get(metric)

            if target_value is not None:
                # For metrics where lower is better (time, cost, defects)
                if 'time' in metric.lower() or 'cost' in metric.lower() or 'defect' in metric.lower():
                    improvement_pct = ((current_value - target_value) / current_value) * 100
                else:
                    # For metrics where higher is better (OEE, quality, output)
                    improvement_pct = ((target_value - current_value) / current_value) * 100

                improvements[metric] = {
                    'current': current_value,
                    'target': target_value,
                    'gap': target_value - current_value if 'time' not in metric.lower() else current_value - target_value,
                    'improvement_pct': improvement_pct
                }

        return improvements

    def track_action_items(self, action_items):
        """
        Track kaizen action items

        action_items: list of dicts with action details
        {
            'action': 'Install shadow board for tools',
            'responsible': 'John Smith',
            'due_date': '2025-02-15',
            'status': 'In Progress',
            'impact': 'Reduce tool search time by 5 min/day'
        }
        """

        self.action_items = pd.DataFrame(action_items)

        # Calculate completion rate
        if not self.action_items.empty:
            completed = (self.action_items['status'] == 'Completed').sum()
            total = len(self.action_items)
            completion_rate = (completed / total) * 100
        else:
            completion_rate = 0

        return {
            'total_actions': len(self.action_items),
            'completed': (self.action_items['status'] == 'Completed').sum(),
            'in_progress': (self.action_items['status'] == 'In Progress').sum(),
            'not_started': (self.action_items['status'] == 'Not Started').sum(),
            'completion_rate': completion_rate
        }

    def measure_results(self, actual_metrics):
        """
        Measure actual results after kaizen implementation

        actual_metrics: dict of post-kaizen metrics
        """

        results = {}

        for metric, current_value in self.current_metrics.items():
            actual_value = actual_metrics.get(metric)
            target_value = self.target_metrics.get(metric)

            if actual_value is not None and target_value is not None:
                # Calculate actual improvement
                if 'time' in metric.lower() or 'cost' in metric.lower() or 'defect' in metric.lower():
                    actual_improvement = ((current_value - actual_value) / current_value) * 100
                    target_achieved = (current_value - actual_value) >= (current_value - target_value)
                else:
                    actual_improvement = ((actual_value - current_value) / current_value) * 100
                    target_achieved = (actual_value - current_value) >= (target_value - current_value)

                results[metric] = {
                    'baseline': current_value,
                    'target': target_value,
                    'actual': actual_value,
                    'improvement_pct': actual_improvement,
                    'target_achieved': target_achieved
                }

        return results

# Example usage
kaizen = KaizenEvent(
    focus_area="Assembly line efficiency improvement",
    current_metrics={
        'cycle_time_seconds': 180,
        'defect_rate_ppm': 5000,
        'changeover_time_minutes': 45,
        'oee_pct': 72
    },
    target_metrics={
        'cycle_time_seconds': 150,
        'defect_rate_ppm': 2000,
        'changeover_time_minutes': 20,
        'oee_pct': 85
    }
)

# Calculate potential
potential = kaizen.calculate_improvement_potential()
print("Improvement Potential:")
for metric, data in potential.items():
    print(f"  {metric}:")
    print(f"    Current: {data['current']}")
    print(f"    Target: {data['target']}")
    print(f"    Improvement: {data['improvement_pct']:.1f}%")

# Track actions
action_items = [
    {'action': 'Implement 5S in assembly area', 'responsible': 'Team Lead', 'due_date': '2025-02-15', 'status': 'Completed', 'impact': 'Reduce search time'},
    {'action': 'Install quick-release fixtures', 'responsible': 'Maintenance', 'due_date': '2025-02-20', 'status': 'In Progress', 'impact': 'Reduce changeover time'},
    {'action': 'Create standard work instructions', 'responsible': 'Quality', 'due_date': '2025-02-18', 'status': 'Completed', 'impact': 'Reduce defects'},
    {'action': 'Train operators on new process', 'responsible': 'Supervisor', 'due_date': '2025-02-22', 'status': 'Not Started', 'impact': 'Improve cycle time'},
]

tracking = kaizen.track_action_items(action_items)
print(f"\nAction Item Tracking:")
print(f"  Completion Rate: {tracking['completion_rate']:.1f}%")
print(f"  Completed: {tracking['completed']}")
print(f"  In Progress: {tracking['in_progress']}")
print(f"  Not Started: {tracking['not_started']}")

# Measure results (after implementation)
actual_metrics = {
    'cycle_time_seconds': 155,
    'defect_rate_ppm': 2500,
    'changeover_time_minutes': 25,
    'oee_pct': 80
}

results = kaizen.measure_results(actual_metrics)
print("\nKaizen Results:")
for metric, data in results.items():
    print(f"  {metric}:")
    print(f"    Baseline: {data['baseline']}")
    print(f"    Target: {data['target']}")
    print(f"    Actual: {data['actual']}")
    print(f"    Improvement: {data['improvement_pct']:.1f}%")
    print(f"    Target Achieved: {'✓' if data['target_achieved'] else '✗'}")

---

工具与库

Python 库

分析与优化：

• pandas：数据分析与处理
• numpy：数值计算
• matplotlib、seaborn、plotly：可视化和 VSM 图表
• scipy：用于流程改进的统计分析
• simpy：用于流程仿真的离散事件模拟

精益专用：

• 自定义 VSM 工具（如上方示例所示）
• 用于改进时间线的甘特图

商业精益软件

价值流图：

• VSM 软件：LeanKit、Lucidchart、Microsoft Visio
• eVSM：电子价值流图
• SIMUL8：结合 VSM 的流程仿真

持续改进平台：

• KaiNexus：改进管理软件
• Ingage：精益日常管理系统
• Weever：数字化日常管理和审核
• Lean Focus：目视化管理软件

制造执行系统（MES）：

• SAP ME：具备精益功能的制造执行系统
• Apriso：精益制造执行系统
• Plex：配备精益工具的云端 MES

---

常见挑战与解决方案

挑战：抵制变革

问题：

• 员工不愿采用精益方法
• “我们一直都是这么做的”
• 担心失去工作

解决方案：

• 从教育和培训入手
• 让员工参与改进（自下而上）
• 通过快速见效的成果展示益处
• 管理层承诺并以身作则
• 透明地回应对工作保障的担忧

挑战：改进难以持续

问题：

• 初期改进逐渐退回原状
• 5S 区域再次变得杂乱
• Kaizen 成果未能维持

解决方案：

• 将维持阶段纳入每项改进
• 定期审核并实施目视化管理
• 让正确的做法更容易执行（防错）
• 将绩效与责任落实挂钩
• 持续强化并予以认可

挑战：指标相互冲突

问题：

• 精益目标（减少库存）与其他目标（提高设备利用率）相冲突
• 局部优化损害整体系统

解决方案：

• 让指标与精益原则保持一致
• 采用系统层面的思维（约束理论）
• 关注流动，而非利用率
• 说明库存和过量生产的真实成本
• 调整激励机制

挑战：产品组合复杂

问题：

• 产品种类繁多，难以实现标准化
• 难以确定节拍时间
• 频繁换型

解决方案：

• 按产品族分组
• 为相似产品采用单元化制造
• 通过 SMED 降低换型损失
• 采用混流生产排程
• 模块化设计并延迟差异化

挑战：供应链不稳定

问题：

• JIT 需要可靠的供应商
• 供应波动会扰乱流程
• 供应商交货周期长

解决方案：

• 培养供应商并建立合作伙伴关系
• 采用带有战略缓冲的拉动系统
• 为关键物料建立双重供应渠道
• 供应商管理库存（VMI）
• 提高供应商信息透明度并加强沟通

---

输出格式

精益评估报告

执行摘要：

• 当前成熟度水平
• 识别出的关键浪费
• 改进优先事项
• 预期收益

价值流分析：

指标	当前状态	未来状态	改进幅度
交付周期	16.5 天	5.0 天	缩短 70%
增值率	2.4%	8.5%	提高 6.1 个百分点
行进距离	200 m	60 m	缩短 70%
库存（天）	12 天	3 天	减少 75%
缺陷率	5,000 ppm	1,000 ppm	降低 80%

浪费分析：

浪费类型	影响	根本原因	建议措施
过量生产	高	批量过大	实施看板，将批量减至 1/5
等待	高	生产线不平衡	平衡生产线，开展交叉培训
搬运	中	布局不佳	采用单元式布局，将相关工序集中布置
库存	高	安全库存	实施拉动系统，提高可靠性

实施路线图：

阶段 1（第 1-3 个月）：奠定基础

• 在试点区域实施 5S
• 为所有员工开展精益培训
• 选择用于 VSM 的价值流

阶段 2（第 4-6 个月）：实现流动

• 在试点区域实施连续流
• 为组件建立看板系统
• 设置可视化管理看板

阶段 3（第 7-9 个月）：拉动

• 将看板扩展至所有物料
• 均衡生产计划
• 建立供应商拉动系统

阶段 4（第 10-12 个月）：追求完美

• 培育 Kaizen 文化并开展活动
• 扩展至其他价值流
• 标准化并持续改进

预期收益：

• 交付周期缩短：60-70%
• 库存减少：40-60%
• 生产率提高：20-30%
• 质量提升：缺陷减少 50%
• 占用空间减少：30-40%
• 每年节省成本：$2-5M

---

需要提出的问题

如果需要更多背景信息：

1. 主要问题或机会是什么？（质量、交付周期、成本、流动）
2. 属于哪种制造环境？（单件小批生产、流水生产、装配、流程制造）
3. 当前的精益成熟度如何？（刚开始接触精益、已开展一些改进、成熟）
4. 涉及哪个价值流或流程区域？
5. 目前跟踪哪些指标？
6. 有哪些领导层支持和可用资源？
7. 改进的时间安排是什么？
8. 组织文化是否已准备好迎接变革？

---

相关技能

• production-scheduling：用于生产计划和排序
• quality-management：用于质量改进和六西格玛
• maintenance-planning：用于 TPM 和设备可靠性
• process-optimization：用于流程分析和改进
• assembly-line-balancing：用于生产线平衡和节拍时间
• inventory-optimization：用于库存削减策略
• supply-chain-analytics：用于绩效衡量
• value-analysis：用于价值工程和成本削减