kaizen-event-facilitator

name	kaizen-event-facilitator
description	Kaizen event facilitation skill for rapid improvement workshops and action planning.
allowed-tools	Bash(*) Read Write Edit Glob Grep WebFetch
metadata	{"author":"babysitter-sdk","version":"1.0.0","category":"continuous-improvement","backlog-id":"SK-IE-039"}

kaizen-event-facilitator

You are kaizen-event-facilitator - a specialized skill for facilitating Kaizen events and rapid improvement workshops.

Overview

This skill enables AI-powered Kaizen facilitation including:

• Event planning and preparation
• Current state documentation
• Waste identification (8 wastes)
• Future state design
• Action plan development
• Standard work creation
• Results tracking
• Sustainability planning

Capabilities

1. Kaizen Event Planning

import pandas as pd
from datetime import datetime, timedelta
from dataclasses import dataclass
from typing import List, Dict

@dataclass
class KaizenEvent:
    title: str
    scope: str
    objectives: List[str]
    metrics: List[Dict]
    start_date: datetime
    duration_days: int
    team_members: List[str]
    sponsor: str

def plan_kaizen_event(event: KaizenEvent):
    """
    Generate comprehensive Kaizen event plan
    """
    # Pre-event preparation (2-4 weeks before)
    prep_activities = [
        {"week": -4, "activity": "Define scope and objectives with sponsor", "owner": "Facilitator"},
        {"week": -4, "activity": "Identify team members and get commitment", "owner": "Sponsor"},
        {"week": -3, "activity": "Collect baseline data for metrics", "owner": "Team Lead"},
        {"week": -3, "activity": "Schedule room and equipment", "owner": "Facilitator"},
        {"week": -2, "activity": "Prepare training materials", "owner": "Facilitator"},
        {"week": -2, "activity": "Communicate to affected employees", "owner": "Sponsor"},
        {"week": -1, "activity": "Confirm all logistics", "owner": "Facilitator"},
        {"week": -1, "activity": "Pre-brief team on event objectives", "owner": "Facilitator"}
    ]

    # Event agenda by day
    agenda = generate_event_agenda(event.duration_days)

    # Post-event follow-up
    followup = [
        {"week": 1, "activity": "Complete 30-day action items", "owner": "Team"},
        {"week": 2, "activity": "First results check", "owner": "Facilitator"},
        {"week": 4, "activity": "30-day review meeting", "owner": "Sponsor"},
        {"week": 12, "activity": "90-day sustainability audit", "owner": "Facilitator"}
    ]

    return {
        "event_summary": {
            "title": event.title,
            "scope": event.scope,
            "objectives": event.objectives,
            "duration": f"{event.duration_days} days",
            "team_size": len(event.team_members)
        },
        "preparation": prep_activities,
        "event_agenda": agenda,
        "followup": followup,
        "success_metrics": event.metrics
    }

def generate_event_agenda(duration_days: int):
    """Generate standard Kaizen event agenda"""
    if duration_days == 5:
        return {
            "day_1": {
                "theme": "Training and Current State",
                "activities": [
                    {"time": "8:00-8:30", "activity": "Welcome and introductions"},
                    {"time": "8:30-10:00", "activity": "Lean fundamentals training"},
                    {"time": "10:00-12:00", "activity": "Go to gemba - observe current state"},
                    {"time": "1:00-3:00", "activity": "Document current state process map"},
                    {"time": "3:00-5:00", "activity": "Collect time observations"}
                ]
            },
            "day_2": {
                "theme": "Waste Identification and Analysis",
                "activities": [
                    {"time": "8:00-10:00", "activity": "Complete current state map"},
                    {"time": "10:00-12:00", "activity": "Identify 8 wastes"},
                    {"time": "1:00-3:00", "activity": "Root cause analysis"},
                    {"time": "3:00-5:00", "activity": "Prioritize opportunities"}
                ]
            },
            "day_3": {
                "theme": "Future State Design",
                "activities": [
                    {"time": "8:00-10:00", "activity": "Brainstorm improvements"},
                    {"time": "10:00-12:00", "activity": "Design future state"},
                    {"time": "1:00-3:00", "activity": "Develop action plans"},
                    {"time": "3:00-5:00", "activity": "Begin implementation"}
                ]
            },
            "day_4": {
                "theme": "Implementation",
                "activities": [
                    {"time": "8:00-12:00", "activity": "Implement changes"},
                    {"time": "1:00-3:00", "activity": "Test and adjust"},
                    {"time": "3:00-5:00", "activity": "Document standard work"}
                ]
            },
            "day_5": {
                "theme": "Standardize and Report",
                "activities": [
                    {"time": "8:00-10:00", "activity": "Finalize standard work"},
                    {"time": "10:00-12:00", "activity": "Train affected employees"},
                    {"time": "1:00-3:00", "activity": "Prepare report-out"},
                    {"time": "3:00-4:00", "activity": "Management report-out"},
                    {"time": "4:00-5:00", "activity": "Celebrate and close"}
                ]
            }
        }
    elif duration_days == 3:
        return {
            "day_1": {"theme": "Analyze", "activities": ["Current state", "Waste identification"]},
            "day_2": {"theme": "Improve", "activities": ["Future state", "Implementation"]},
            "day_3": {"theme": "Standardize", "activities": ["Standard work", "Report-out"]}
        }
    return {}

2. Waste Identification (8 Wastes)

def identify_wastes(observations: List[Dict]):
    """
    Categorize observations into 8 wastes (TIMWOODS)

    observations: list of {'description': str, 'location': str, 'frequency': str, 'impact': str}
    """
    waste_categories = {
        'T': {'name': 'Transport', 'description': 'Unnecessary movement of materials', 'examples': []},
        'I': {'name': 'Inventory', 'description': 'Excess inventory beyond immediate need', 'examples': []},
        'M': {'name': 'Motion', 'description': 'Unnecessary movement of people', 'examples': []},
        'W': {'name': 'Waiting', 'description': 'Idle time waiting for next step', 'examples': []},
        'O': {'name': 'Overproduction', 'description': 'Producing more than needed', 'examples': []},
        'O2': {'name': 'Overprocessing', 'description': 'More processing than required', 'examples': []},
        'D': {'name': 'Defects', 'description': 'Rework, scrap, errors', 'examples': []},
        : {: , : , : []}
    }

    categorized = []

     obs  observations:
        
        category = categorize_waste(obs[])
        obs[] = category
        waste_categories[category][].append(obs)
        categorized.append(obs)

    
    summary = []
     code, waste  waste_categories.items():
         waste[]:
            summary.append({
                : code,
                : waste[],
                : (waste[]),
                : (  e  waste[]  e.get() == )
            })

    summary.sort(key= x: x[], reverse=)

     {
        : waste_categories,
        : categorized,
        : summary,
        : (observations),
        : summary[]  summary  
    }

 ():
    
    desc_lower = description.lower()

     (w  desc_lower  w  [, , , ]):
             desc_lower  
     (w  desc_lower  w  [, , , ]):
         
     (w  desc_lower  w  [, , , ]):
         
     (w  desc_lower  w  [, , , , ]):
         
     (w  desc_lower  w  [, , ]):
         
     (w  desc_lower  w  [, , ]):
         
     (w  desc_lower  w  [, , , ]):
         
    :
           

3. Action Plan Development

def create_action_plan(improvements: List[Dict], event_end_date: datetime):
    """
    Create structured action plan from improvements

    improvements: list of {'description': str, 'owner': str, 'priority': str, 'effort': str}
    """
    actions = []

    for i, imp in enumerate(improvements):
        # Determine timeline based on effort
        if imp['effort'] == 'just_do_it':
            due_date = event_end_date
            category = 'Do During Event'
        elif imp['effort'] == 'short_term':
            due_date = event_end_date + timedelta(days=30)
            category = '30-Day Action'
        elif imp['effort'] == 'medium_term':
            due_date = event_end_date + timedelta(days=90)
            category = '90-Day Action'
        else:
            due_date = event_end_date + timedelta(days=180)
            category = 'Long-Term Initiative'

        actions.append({
            'id': f'A{i+1:03d}',
            'description': imp['description'],
            'owner': imp['owner'],
            'priority': imp['priority'],
            'category': category,
            'due_date': due_date.strftime('%Y-%m-%d'),
            'status': 'Not Started',
            : ,
            : imp.get(, ),
            : imp.get(, )
        })

    
    by_category = {}
     action  actions:
        cat = action[]
         cat   by_category:
            by_category[cat] = []
        by_category[cat].append(action)

     {
        : actions,
        : by_category,
        : {
            : (actions),
            : (by_category.get(, [])),
            : (by_category.get(, [])),
            : (by_category.get(, [])),
            : (by_category.get(, []))
        }
    }

4. Results Tracking

def track_kaizen_results(baseline: Dict, current: Dict, targets: Dict):
    """
    Track Kaizen event results against baseline and targets
    """
    results = []

    for metric, baseline_value in baseline.items():
        target_value = targets.get(metric, baseline_value)
        current_value = current.get(metric, baseline_value)

        # Calculate improvements
        if baseline_value != 0:
            improvement_pct = (baseline_value - current_value) / baseline_value * 100
            target_improvement_pct = (baseline_value - target_value) / baseline_value * 100
        else:
            improvement_pct = 0
            target_improvement_pct = 0

        # Check if target met
        better_is_lower = target_value < baseline_value
        if better_is_lower:
            target_met = current_value <= target_value
        else:
            target_met = current_value >= target_value

        results.append({
            'metric': metric,
            'baseline': baseline_value,
            'target': target_value,
            'current': current_value,
            'improvement_percent': round(improvement_pct, 1),
            'target_improvement_percent': round(target_improvement_pct, 1),
            'target_met': target_met,
            'status': 'green' if target_met else 'yellow' if abs(improvement_pct) > 0  
        })

    
    targets_met = (  r  results  r[])

     {
        : results,
        : {
            : (results),
            : targets_met,
            : (targets_met / (results) * , )  results  
        }
    }

5. Standard Work Documentation

def create_standard_work(process_steps: List[Dict], takt_time: float,
                        cycle_time: float, work_in_process: int = 1):
    """
    Create standard work documentation

    process_steps: list of {'step': int, 'description': str, 'time': float, 'key_points': list}
    """
    # Calculate totals
    total_time = sum(s['time'] for s in process_steps)
    manual_time = sum(s['time'] for s in process_steps if not s.get('machine_time', False))
    walk_time = sum(s.get('walk_time', 0) for s in process_steps)

    # Standard work elements
    standard_work = {
        'header': {
            'process_name': '',  # To be filled
            'takt_time': takt_time,
            'cycle_time': cycle_time,
            'operators': 1,
            'standard_wip': work_in_process
        },
        'time_summary': {
            'total_cycle_time': round(total_time, 1),
            'manual_time': round(manual_time, 1),
            'walk_time': round(walk_time, ),
            : (takt_time - cycle_time, )
        },
        : [],
        : [],
        : []
    }

    
    cumulative = 
     step  process_steps:
        cumulative += step[]
        standard_work[].append({
            : step[],
            : step[],
            : step[],
            : cumulative,
            : step.get(, []),
            : step.get(, )  
        })

        
         kp  step.get(, []):
               kp.lower()    kp.lower():
                standard_work[].append({: step[], : kp})
               kp.lower()    kp.lower():
                standard_work[].append({: step[], : kp})

     standard_work

 ():
    
    combination = {
        : takt_time,
        : []
    }

     op  operators:
        op_data = {
            : op[],
            : op[],
            : (t.get(, )  t  op[]),
            : (t.get(, )  t  op[]),
            : (t.get(, )  t  op[]),
            : (t.get(, ) + t.get(, ) + t.get(, )  t  op[])
        }
        op_data[] = (op_data[] / takt_time * , )
        combination[].append(op_data)

     combination

6. Sustainability Assessment

def assess_sustainability(event_id: str, days_since_event: int,
                         audit_findings: List[Dict]):
    """
    Assess sustainability of Kaizen improvements
    """
    categories = {
        'standard_work_adherence': [],
        'metrics_maintained': [],
        'visual_management': [],
        'employee_engagement': [],
        'system_support': []
    }

    for finding in audit_findings:
        cat = finding.get('category', 'standard_work_adherence')
        if cat in categories:
            categories[cat].append(finding)

    # Score each category
    scores = {}
    for cat, findings in categories.items():
        if findings:
            positive = sum(1 for f in findings if f.get('status') == 'maintained')
            scores[cat] = round(positive / len(findings) * 100, 1)
        else:
            scores[cat] = None

    # Overall sustainability score
    valid_scores = [s for s in scores.values() if s is not None]
    overall = round(sum(valid_scores) / len(valid_scores), )  valid_scores  

     {
        : event_id,
        : days_since_event,
        : scores,
        : overall,
        :   overall >=     overall >=   ,
        : categories,
        : generate_sustainability_recommendations(scores, overall)
    }

 ():
    recommendations = []

     scores.get(, ) < :
        recommendations.append()

     scores.get(, ) < :
        recommendations.append()

     scores.get(, ) < :
        recommendations.append()

     overall < :
        recommendations.append()

     recommendations

Process Integration

This skill integrates with the following processes:

• kaizen-event-execution.js
• continuous-improvement-program.js
• standard-work-development.js

Output Format

{
  "event_plan": {
    "title": "Assembly Cell Improvement",
    "duration": "5 days",
    "objectives": ["Reduce cycle time 20%", "Eliminate 3 wastes"]
  },
  "waste_identification": {
    "total_identified": 15,
    "top_category": "Motion",
    "high_impact": 5
  },
  "action_plan": {
    "total_actions": 22,
    "during_event": 8,
    "30_day": 10,
    "90_day": 4
  },
  "results" 
     
     
     
  

Best Practices

1. Prepare thoroughly - 80% of success is in preparation
2. Go to gemba - Observe reality, not reports
3. Involve operators - They know the problems
4. Do, don't discuss - Implement during the event
5. Measure before/after - Data proves improvement
6. Follow up - Sustainability requires attention

Constraints

• Requires dedicated team time
• Management support essential
• Scope must be achievable in timeframe
• Not all problems suit Kaizen events