class AutomationMaturityAssessor:
    """AI自動化成熟度の診断と移行計画策定"""
 
    LEVEL_CRITERIA = {
        0: {
            "name": "手動作業",
            "characteristics": [
                "業務プロセスが属人化している",
                "同じ作業を複数人が重複して行っている",
                "データがExcelやメールに散在している"
            ],
            "upgrade_trigger": "同一作業の週5回以上の繰り返し",
            "upgrade_cost": 0,
            "upgrade_time": "1-2週間"
        },
        1: {
            "name": "ルールベース自動化",
            "characteristics": [
                "定型業務のマクロ/スクリプト化が完了",
                "IFTTT/cron等でスケジュール実行している",
                "トリガーとアクションが明確に定義されている"
            ],
            "upgrade_trigger": "ルールでは対応できない例外処理が30%以上",
            "upgrade_cost": 50000,
            "upgrade_time": "2-4週間"
        },
        2: {
            "name": "ノーコードAI",
            "characteristics": [
                "Zapier/Make等でAI APIと連携している",
                "自然言語処理による分類・要約が稼働している",
                "非エンジニアがワークフローを管理できる"
            ],
            "upgrade_trigger": "ノーコードツールの制約に月5回以上ぶつかる",
            "upgrade_cost": 200000,
            "upgrade_time": "1-2ヶ月"
        },
        3: {
            "name": "カスタムAIパイプライン",
            "characteristics": [
                "LangChain/独自APIでカスタムパイプラインが稼働",
                "複数AIモデルの使い分けが実装されている",
                "品質監視・コスト管理ダッシュボードが稼働"
            ],
            "upgrade_trigger": "人間の介入なしで対応すべきタスクが50%以上",
            "upgrade_cost": 1000000,
            "upgrade_time": "3-6ヶ月"
        },
        4: {
            "name": "自律エージェント",
            "characteristics": [
                "マルチエージェントシステムが自律的に動作",
                "自己改善ループが組み込まれている",
                "人間は例外処理と戦略的判断のみ"
            ],
            "upgrade_trigger": "N/A（最上位レベル）",
            "upgrade_cost": 5000000,
            "upgrade_time": "6-12ヶ月"
        }
    }
 
    def assess_current_level(self, answers: dict) -> dict:
        """現在の成熟度レベルを診断"""
        score = 0
 
        # 自動化の範囲
        if answers.get("has_scheduled_tasks"):
            score += 1
        if answers.get("uses_ai_api"):
            score += 1
        if answers.get("has_custom_pipeline"):
            score += 1
        if answers.get("has_autonomous_agents"):
            score += 1
 
        current_level = min(score, 4)
        next_level = min(current_level + 1, 4)
 
        return {
            "current_level": current_level,
            "level_name": self.LEVEL_CRITERIA[current_level]["name"],
            "characteristics": self.LEVEL_CRITERIA[current_level]["characteristics"],
            "next_level": next_level,
            "upgrade_trigger": self.LEVEL_CRITERIA[current_level]["upgrade_trigger"],
            "upgrade_cost": self.LEVEL_CRITERIA[next_level]["upgrade_cost"],
            "upgrade_time": self.LEVEL_CRITERIA[next_level]["upgrade_time"],
            "recommendation": self._generate_recommendation(current_level)
        }
 
    def _generate_recommendation(self, level: int) -> str:
        """レベル別推奨アクション"""
        recommendations = {
            0: "まず繰り返し作業を洗い出し、Google Apps Scriptやcronで自動化を開始してください",
            1: "Zapier無料プランでAI連携を試し、効果を測定してください",
            2: "月額コストが$200を超えたら、n8nセルフホストかカスタム開発への移行を検討",
            3: "人間の介入率を計測し、50%未満ならエージェント化の投資回収が見込めます",
            4: "自己改善ループの精度モニタリングとガバナンス強化に注力してください"
        }
        return recommendations.get(level, "")
 
    def create_migration_plan(self, current: int, target: int) -> list[dict]:
        """移行計画の策定"""
        plan = []
        for level in range(current + 1, target + 1):
            criteria = self.LEVEL_CRITERIA[level]
            plan.append({
                "target_level": level,
                "name": criteria["name"],
                "estimated_cost": criteria["upgrade_cost"],
                "estimated_time": criteria["upgrade_time"],
                "prerequisites": criteria["characteristics"],
                "success_criteria": [
                    f"Level {level}の全特性を満たす",
                    "ROI 100%以上を達成",
                    "運用チームのトレーニング完了"
                ]
            })
        return plan

# Zapier + OpenAI の概念的フロー（Pythonで表現）
automation_flow = {
    "trigger": "新規メール受信（Gmail）",
    "steps": [
        {"action": "OpenAI GPT-4で要約生成", "model": "gpt-4"},
        {"action": "感情分析で優先度判定", "threshold": 0.7},
        {"action": "Slackに通知送信", "channel": "#urgent"},
        {"action": "Notionにタスク作成", "database": "inbox"}
    ],
    "estimated_time_saved": "1日あたり2時間"
}

# n8n ワークフロー定義（YAML形式）
name: "AI顧客対応自動化"
nodes:
  - type: webhook
    name: "問い合わせ受信"
    config:
      method: POST
      path: /customer-inquiry
 
  - type: openai
    name: "意図分類"
    config:
      model: gpt-4
      prompt: |
        以下の問い合わせを分類してください:
        - billing: 請求関連
        - technical: 技術サポート
        - sales: 営業関連
 
  - type: switch
    name: "ルーティング"
    rules:
      - value: "billing"
        output: "経理チーム"
      - value: "technical"
        output: "技術チーム"
      - value: "sales"
        output: "営業チーム"

# 実践例1: 問い合わせメールの自動分類と通知
workflow_1 = {
    "name": "カスタマーサポート自動トリアージ",
    "platform": "Zapier",
    "setup_time": "30分",
    "monthly_cost": "$20",
    "flow": [
        "Gmail新着メール → OpenAI分類 → Slack通知 + Notion登録",
    ],
    "roi": "対応時間50%削減（1日2時間→1時間）"
}
 
# 実践例2: 議事録の自動要約と共有
workflow_2 = {
    "name": "会議議事録の自動要約",
    "platform": "Make",
    "setup_time": "1時間",
    "monthly_cost": "$15",
    "flow": [
        "Google Meet録画 → Whisper文字起こし → Claude要約 → Slackに共有",
    ],
    "roi": "議事録作成時間90%削減（30分→3分）"
}
 
# 実践例3: SNS投稿の自動生成
workflow_3 = {
    "name": "ブログ記事→SNS投稿自動生成",
    "platform": "Zapier",
    "setup_time": "45分",
    "monthly_cost": "$20",
    "flow": [
        "WordPress新記事 → GPT-4で3プラットフォーム向け投稿生成 → Buffer予約投稿",
    ],
    "roi": "SNS運用時間75%削減（月20時間→5時間）"
}
 
# 実践例4: 請求書データの自動抽出
workflow_4 = {
    "name": "請求書OCR自動処理",
    "platform": "n8n",
    "setup_time": "2時間",
    "monthly_cost": "$0（セルフホスト）",
    "flow": [
        "メール添付PDF → Cloud Vision OCR → GPT-4でデータ抽出 → スプレッドシート記録",
    ],
    "roi": "経理作業60%削減（月10時間→4時間）"
}
 
# 実践例5: 競合モニタリング
workflow_5 = {
    "name": "競合サイト変更検知と分析",
    "platform": "Make",
    "setup_time": "1.5時間",
    "monthly_cost": "$30",
    "flow": [
        "競合サイト定期チェック → 変更検知 → Claude分析 → レポート生成 → Slack通知",
    ],
    "roi": "競合分析の手動工数95%削減"
}

from typing import Any
import openai
import json
 
class AIAutomationPipeline:
    """AI自動化パイプラインの基本実装"""
 
    def __init__(self, api_key: str):
        self.client = openai.OpenAI(api_key=api_key)
        self.steps: list[dict] = []
 
    def add_step(self, name: str, prompt_template: str,
                 model: str = "gpt-4") -> "AIAutomationPipeline":
        """パイプラインにステップを追加"""
        self.steps.append({
            "name": name,
            "prompt_template": prompt_template,
            "model": model
        })
        return self  # メソッドチェーン対応
 
    def execute(self, input_data: dict[str, Any]) -> list[dict]:
        """パイプラインを実行"""
        results = []
        context = input_data.copy()
 
        for step in self.steps:
            prompt = step["prompt_template"].format(**context)
            response = self.client.chat.completions.create(
                model=step["model"],
                messages=[{"role": "user", "content": prompt}]
            )
            output = response.choices[0].message.content
            context[step["name"]] = output
            results.append({"step": step["name"], "output": output})
 
        return results
 
# 使用例
pipeline = AIAutomationPipeline(api_key="sk-...")
pipeline.add_step(
    name="summary",
    prompt_template="以下の文書を3行で要約:\n{document}"
).add_step(
    name="action_items",
    prompt_template="要約: {summary}\n\nアクションアイテムを抽出:"
).add_step(
    name="priority",
    prompt_template="アクション: {action_items}\n\n優先度を判定(高/中/低):"
)
 
results = pipeline.execute({"document": "長い会議議事録..."})

import time
from functools import wraps
 
def with_retry(max_retries: int = 3, backoff_factor: float = 2.0):
    """AI API呼び出しのリトライデコレータ"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            last_exception = None
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except openai.RateLimitError:
                    wait = backoff_factor ** attempt
                    print(f"レート制限。{wait}秒後にリトライ...")
                    time.sleep(wait)
                except openai.APIError as e:
                    last_exception = e
                    if attempt == max_retries - 1:
                        raise
                    time.sleep(1)
            raise last_exception
        return wrapper
    return decorator
 
@with_retry(max_retries=3)
def call_ai(prompt: str) -> str:
    """リトライ付きAI呼び出し"""
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content

import time
from dataclasses import dataclass
from typing import Optional
import openai
import anthropic
 
@dataclass
class AIProvider:
    name: str
    priority: int
    is_healthy: bool = True
    last_error_time: float = 0
    error_count: int = 0
    circuit_breaker_threshold: int = 5
    circuit_breaker_timeout: float = 60.0
 
class MultiProviderAI:
    """マルチプロバイダーAI呼び出しエンジン"""
 
    def __init__(self):
        self.providers = {
            "openai": AIProvider(name="openai", priority=1),
            "anthropic": AIProvider(name="anthropic", priority=2),
            "openai_fallback": AIProvider(name="openai_fallback", priority=3),
        }
        self.openai_client = openai.OpenAI()
        self.anthropic_client = anthropic.Anthropic()
 
    def call(self, prompt: str, max_tokens: int = 1024) -> dict:
        """フォールバック付きAI呼び出し"""
        sorted_providers = sorted(
            self.providers.values(),
            key=lambda p: p.priority
        )
 
        for provider in sorted_providers:
            if not self._is_available(provider):
                continue
 
            try:
                result = self._call_provider(
                    provider.name, prompt, max_tokens
                )
                # 成功時: エラーカウントリセット
                provider.error_count = 0
                provider.is_healthy = True
                return {
                    "provider": provider.name,
                    "content": result,
                    "status": "success"
                }
            except Exception as e:
                provider.error_count += 1
                provider.last_error_time = time.time()
                if provider.error_count >= provider.circuit_breaker_threshold:
                    provider.is_healthy = False
                print(f"[{provider.name}] エラー: {e}、次のプロバイダーへ")
 
        raise Exception("全プロバイダーが応答不可")
 
    def _is_available(self, provider: AIProvider) -> bool:
        """サーキットブレーカーチェック"""
        if provider.is_healthy:
            return True
        # タイムアウト後に復帰を試みる
        elapsed = time.time() - provider.last_error_time
        if elapsed > provider.circuit_breaker_timeout:
            provider.is_healthy = True
            provider.error_count = 0
            return True
        return False
 
    def _call_provider(self, name: str, prompt: str,
                       max_tokens: int) -> str:
        """プロバイダー別のAPI呼び出し"""
        if name in ("openai", "openai_fallback"):
            model = "gpt-4o" if name == "openai" else "gpt-4o-mini"
            response = self.openai_client.chat.completions.create(
                model=model,
                messages=[{"role": "user", "content": prompt}],
                max_tokens=max_tokens
            )
            return response.choices[0].message.content
        elif name == "anthropic":
            response = self.anthropic_client.messages.create(
                model="claude-sonnet-4-20250514",
                max_tokens=max_tokens,
                messages=[{"role": "user", "content": prompt}]
            )
            return response.content[0].text
        raise ValueError(f"Unknown provider: {name}")
 
# 使用例
ai = MultiProviderAI()
result = ai.call("売上データを3行で要約してください")
print(f"使用プロバイダー: {result['provider']}")
print(f"結果: {result['content']}")

from abc import ABC, abstractmethod
from typing import Any
 
class BaseAgent(ABC):
    """エージェントの基底クラス"""
 
    def __init__(self, name: str, model: str = "gpt-4o"):
        self.name = name
        self.model = model
        self.memory: list[dict] = []
 
    @abstractmethod
    def execute(self, task: dict) -> dict:
        """タスクを実行（サブクラスで実装）"""
        pass
 
    def add_to_memory(self, item: dict):
        """メモリに追加"""
        self.memory.append(item)
 
class PlannerAgent(BaseAgent):
    """計画エージェント: タスクを分解して実行計画を立てる"""
 
    def execute(self, task: dict) -> dict:
        prompt = f"""
タスク: {task['description']}
 
このタスクを実行するために必要なステップを分解してください。
各ステップには以下を含めてください:
- step_name: ステップ名
- agent_type: 実行するエージェント（searcher/analyzer/executor）
- input: 必要な入力
- expected_output: 期待する出力
 
JSON配列で返してください。
"""
        response = call_ai(prompt)
        plan = json.loads(response)
        self.add_to_memory({"task": task, "plan": plan})
        return {"plan": plan, "status": "planned"}
 
class SearcherAgent(BaseAgent):
    """検索エージェント: 情報を検索・収集する"""
 
    def execute(self, task: dict) -> dict:
        prompt = f"""
検索タスク: {task['description']}
検索対象: {task.get('source', '一般知識')}
 
以下の情報を収集してください:
{task.get('query', '')}
 
構造化されたJSON形式で結果を返してください。
"""
        response = call_ai(prompt)
        self.add_to_memory({"task": task, "result": response})
        return {"findings": response, "status": "completed"}
 
class AnalyzerAgent(BaseAgent):
    """分析エージェント: データを分析して洞察を生成する"""
 
    def execute(self, task: dict) -> dict:
        prompt = f"""
分析タスク: {task['description']}
入力データ: {task.get('data', '')}
 
以下の観点で分析してください:
1. 主要な発見
2. リスクと機会
3. 推奨アクション
 
JSON形式で返してください。
"""
        response = call_ai(prompt)
        self.add_to_memory({"task": task, "analysis": response})
        return {"analysis": response, "status": "completed"}
 
class MultiAgentOrchestrator:
    """マルチエージェントのオーケストレーター"""
 
    def __init__(self):
        self.agents = {
            "planner": PlannerAgent("Planner"),
            "searcher": SearcherAgent("Searcher"),
            "analyzer": AnalyzerAgent("Analyzer"),
        }
        self.execution_log: list[dict] = []
 
    def execute_task(self, description: str) -> dict:
        """タスクの計画→実行→結果統合"""
        # Step 1: 計画
        plan_result = self.agents["planner"].execute(
            {"description": description}
        )
 
        # Step 2: 計画に従って各エージェントを実行
        results = []
        for step in plan_result["plan"]:
            agent_type = step["agent_type"]
            if agent_type in self.agents:
                agent = self.agents[agent_type]
                result = agent.execute(step)
                results.append({
                    "step": step["step_name"],
                    "agent": agent_type,
                    "result": result
                })
                self.execution_log.append({
                    "step": step["step_name"],
                    "status": result["status"]
                })
 
        return {
            "task": description,
            "plan": plan_result["plan"],
            "results": results,
            "execution_log": self.execution_log
        }
 
# 使用例
orchestrator = MultiAgentOrchestrator()
result = orchestrator.execute_task(
    "競合3社のAI SaaS製品を調査し、自社プロダクトの差別化ポイントを分析"
)

# 導入フェーズの定義
roadmap = {
    "Phase 1 (1-2週間)": {
        "目標": "クイックウィン獲得",
        "対象": "メール自動分類、定型回答生成",
        "ツール": "Zapier + OpenAI",
        "KPI": "対応時間 50% 削減"
    },
    "Phase 2 (1-2ヶ月)": {
        "目標": "部門横断自動化",
        "対象": "契約書レビュー、レポート生成",
        "ツール": "n8n + Claude API",
        "KPI": "月間 100時間削減"
    },
    "Phase 3 (3-6ヶ月)": {
        "目標": "戦略的AI統合",
        "対象": "顧客対応エージェント、予測分析",
        "ツール": "カスタムパイプライン",
        "KPI": "売上 20% 向上"
    }
}

class AutomationROICalculator:
    """AI自動化のROI自動計算エンジン"""
 
    def __init__(self):
        self.cost_items: list[dict] = []
        self.benefit_items: list[dict] = []
        self.implementation_cost: float = 0
 
    def add_cost(self, name: str, monthly_amount: float,
                 category: str = "operational"):
        """月次コスト項目を追加"""
        self.cost_items.append({
            "name": name,
            "amount": monthly_amount,
            "category": category
        })
 
    def add_benefit(self, name: str, monthly_amount: float,
                    category: str = "cost_reduction",
                    confidence: float = 0.8):
        """月次効果項目を追加（信頼度付き）"""
        self.benefit_items.append({
            "name": name,
            "amount": monthly_amount,
            "category": category,
            "confidence": confidence
        })
 
    def set_implementation_cost(self, amount: float):
        """初期導入コストを設定"""
        self.implementation_cost = amount
 
    def calculate(self) -> dict:
        """ROI計算"""
        monthly_cost = sum(item["amount"] for item in self.cost_items)
        monthly_benefit = sum(
            item["amount"] * item["confidence"]
            for item in self.benefit_items
        )
        monthly_net = monthly_benefit - monthly_cost
 
        # 投資回収期間
        payback_months = (
            self.implementation_cost / monthly_net
            if monthly_net > 0 else float('inf')
        )
 
        # 1年間のROI
        year1_total_benefit = monthly_benefit * 12
        year1_total_cost = monthly_cost * 12 + self.implementation_cost
        year1_roi = (
            (year1_total_benefit - year1_total_cost) / year1_total_cost * 100
            if year1_total_cost > 0 else 0
        )
 
        # 3年間NPV（割引率8%）
        discount_rate = 0.08
        npv = -self.implementation_cost
        for month in range(1, 37):
            npv += monthly_net / (1 + discount_rate / 12) ** month
 
        return {
            "monthly_cost": monthly_cost,
            "monthly_benefit": monthly_benefit,
            "monthly_net": monthly_net,
            "implementation_cost": self.implementation_cost,
            "payback_months": round(payback_months, 1),
            "year1_roi_percent": round(year1_roi, 1),
            "npv_3years": round(npv, 0),
            "recommendation": self._get_recommendation(year1_roi, payback_months),
            "risk_adjusted_benefit": monthly_benefit,
            "cost_breakdown": self.cost_items,
            "benefit_breakdown": self.benefit_items
        }
 
    def _get_recommendation(self, roi: float, payback: float) -> str:
        """投資判断の推奨"""
        if roi > 200 and payback < 3:
            return "強く推奨: ROI非常に高く、投資回収も早い"
        elif roi > 100 and payback < 6:
            return "推奨: 健全なROIと合理的な投資回収期間"
        elif roi > 50 and payback < 12:
            return "条件付き推奨: ROIは正だが慎重に進めるべき"
        elif roi > 0:
            return "要検討: ROIは正だがリスク要因を精査すべき"
        else:
            return "非推奨: 現時点ではコストが効果を上回る"
 
    def generate_report(self) -> str:
        """レポート生成"""
        result = self.calculate()
        report = f"""
=== AI自動化 ROI分析レポート ===
 
■ 月次コスト: ¥{result['monthly_cost']:,.0f}
"""
        for item in self.cost_items:
            report += f"  - {item['name']}: ¥{item['amount']:,.0f}\n"
 
        report += f"""
■ 月次効果（リスク調整済み）: ¥{result['monthly_benefit']:,.0f}
"""
        for item in self.benefit_items:
            adjusted = item['amount'] * item['confidence']
            report += f"  - {item['name']}: ¥{adjusted:,.0f} (信頼度{item['confidence']*100:.0f}%)\n"
 
        report += f"""
■ 月次純効果: ¥{result['monthly_net']:,.0f}
■ 初期投資: ¥{result['implementation_cost']:,.0f}
■ 投資回収期間: {result['payback_months']}ヶ月
■ 初年度ROI: {result['year1_roi_percent']}%
■ 3年NPV: ¥{result['npv_3years']:,.0f}
 
■ 判定: {result['recommendation']}
"""
        return report
 
# 使用例
calc = AutomationROICalculator()
calc.set_implementation_cost(500000)  # 初期導入50万円
calc.add_cost("Zapier Pro", 5000)
calc.add_cost("OpenAI API", 30000)
calc.add_cost("保守工数", 50000)
calc.add_benefit("メール対応工数削減", 120000, confidence=0.9)
calc.add_benefit("レポート作成自動化", 80000, confidence=0.85)
calc.add_benefit("エラー削減", 50000, confidence=0.7)
 
print(calc.generate_report())

import re
from typing import Optional
 
class PIIMasker:
    """個人情報の検出とマスキング"""
 
    PATTERNS = {
        "email": r'[\w.-]+@[\w.-]+\.\w+',
        "phone_jp": r'0\d{1,4}-?\d{1,4}-?\d{3,4}',
        "credit_card": r'\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}',
        "my_number": r'\d{4}\s?\d{4}\s?\d{4}',
        "name_jp": r'[一-龥]{1,4}[　\s][一-龥]{1,4}',
        "postal_code": r'〒?\d{3}-?\d{4}',
    }
 
    def mask(self, text: str) -> tuple[str, dict]:
        """テキスト中のPIIをマスク"""
        masked = text
        found_pii = {}
 
        for pii_type, pattern in self.PATTERNS.items():
            matches = re.findall(pattern, masked)
            if matches:
                found_pii[pii_type] = len(matches)
                for i, match in enumerate(matches):
                    placeholder = f"[{pii_type.upper()}_{i+1}]"
                    masked = masked.replace(match, placeholder, 1)
 
        return masked, found_pii
 
    def is_safe_for_api(self, text: str,
                         threshold: int = 0) -> tuple[bool, dict]:
        """APIに送信して安全か判定"""
        _, found_pii = self.mask(text)
        total_pii = sum(found_pii.values())
        return total_pii <= threshold, found_pii
 
# 使用例
masker = PIIMasker()
text = "田中太郎さん（tanaka@example.com）の電話番号は03-1234-5678です"
masked, pii_found = masker.mask(text)
print(masked)
# → [NAME_JP_1]さん（[EMAIL_1]）の電話番号は[PHONE_JP_1]です
print(pii_found)
# → {'name_jp': 1, 'email': 1, 'phone_jp': 1}

class ChangeManagementPlan:
    """AI自動化の組織変革管理計画"""
 
    PHASES = {
        "awareness": {
            "name": "認知フェーズ",
            "duration": "2-4週間",
            "activities": [
                "経営層へのAI自動化ブリーフィング",
                "部門長向けワークショップ",
                "全社向けAI活用事例の共有",
                "FAQ作成と社内公開"
            ],
            "key_message": "AIは仕事を奪うのではなく、退屈な作業を引き受けてくれるパートナー",
            "success_criteria": "80%の社員がAI自動化の目的を理解"
        },
        "involvement": {
            "name": "巻き込みフェーズ",
            "duration": "2-4週間",
            "activities": [
                "各部門からチャンピオンを選出",
                "自動化候補業務のワークショップ",
                "小規模パイロットの実施",
                "成功体験の社内共有"
            ],
            "key_message": "あなたの業務を最もよく知るのはあなた。改善提案を歓迎します",
            "success_criteria": "各部門から1名以上のチャンピオン選出"
        },
        "execution": {
            "name": "実行フェーズ",
            "duration": "1-3ヶ月",
            "activities": [
                "本格的な自動化ワークフロー構築",
                "トレーニングプログラムの実施",
                "週次レビューと改善",
                "成功事例のドキュメント化"
            ],
            "key_message": "段階的に進め、問題があればすぐに調整します",
            "success_criteria": "パイロット部門で目標KPIの80%達成"
        },
        "optimization": {
            "name": "最適化フェーズ",
            "duration": "継続",
            "activities": [
                "全社展開の計画策定",
                "ROIレポートの定期公開",
                "新しい自動化候補の継続的発掘",
                "ベストプラクティスの標準化"
            ],
            "key_message": "AI自動化は導入で終わりではなく、継続的な改善プロセス",
            "success_criteria": "年間でROI 200%以上を維持"
        }
    }
 
    def generate_plan(self, company_size: str) -> dict:
        """会社規模に応じた変革管理計画を生成"""
        timeline_multiplier = {
            "startup": 0.5,     # スタートアップは速い
            "smb": 1.0,         # 中小企業は標準
            "enterprise": 2.0   # 大企業は時間がかかる
        }
 
        multiplier = timeline_multiplier.get(company_size, 1.0)
        plan = {}
 
        for phase_key, phase in self.PHASES.items():
            plan[phase_key] = {
                **phase,
                "adjusted_duration": f"{phase['duration']}×{multiplier}"
            }
 
        return plan

# BAD: 判断が必要な業務まで無理に自動化
def auto_approve_all_contracts(contract):
    """全契約を自動承認 — 危険！"""
    ai_review = call_ai(f"この契約を承認すべきか: {contract}")
    if "承認" in ai_review:
        approve(contract)  # AIの判断だけで承認してしまう
 
# GOOD: AIは補助、最終判断は人間
def ai_assisted_contract_review(contract):
    """AI支援付き契約レビュー"""
    risk_analysis = call_ai(f"リスク分析: {contract}")
    recommendation = call_ai(f"推奨アクション: {contract}")
 
    return {
        "risk_analysis": risk_analysis,
        "recommendation": recommendation,
        "status": "要人間レビュー",  # 必ず人間が確認
        "reviewer": assign_reviewer(contract)
    }

# BAD: 特定プラットフォームに密結合
class ZapierOnlyWorkflow:
    def run(self):
        zapier.trigger("hook_abc123")  # Zapier固有のAPI
 
# GOOD: 抽象化レイヤーで切り替え可能に
class AutomationPlatform:
    """プラットフォーム抽象化"""
    def trigger_workflow(self, event: dict): ...
    def get_status(self, workflow_id: str): ...
 
class ZapierAdapter(AutomationPlatform):
    def trigger_workflow(self, event): ...
 
class N8nAdapter(AutomationPlatform):
    def trigger_workflow(self, event): ...
 
# プラットフォーム切り替えが容易
platform = N8nAdapter()  # Zapier → n8n への移行が1行
platform.trigger_workflow({"type": "new_email"})

# BAD: 効果測定なしに自動化を推進
def automate_blindly():
    """効果を測らずに自動化を続ける"""
    for process in all_processes:
        automate(process)
    # → 実は手動の方が速かったプロセスまで自動化
    # → コストが増えたことに気づかない
 
# GOOD: 導入前後の定量比較を必須にする
def automate_with_measurement(process):
    """効果測定付き自動化"""
    # Before: 現状計測
    baseline = measure_process(process)
    # 処理時間、エラー率、コスト、満足度を記録
 
    # 自動化実装
    automated = implement_automation(process)
 
    # After: 2週間後に効果測定
    result = measure_process(automated)
 
    comparison = {
        "time_reduction": (baseline.time - result.time) / baseline.time,
        "error_reduction": (baseline.errors - result.errors) / baseline.errors,
        "cost_change": result.cost - baseline.cost,
        "satisfaction_change": result.satisfaction - baseline.satisfaction,
    }
 
    # 効果がなければロールバック
    if comparison["time_reduction"] < 0.2:
        rollback(process)
        return {"status": "rolled_back", "reason": "効果不十分"}
 
    return {"status": "success", "improvement": comparison}

# BAD: 全部門に同時導入
def big_bang_rollout():
    for department in all_departments:
        deploy_ai_automation(department)
    # → 全部門で同時に問題発生、サポートが追いつかない
 
# GOOD: 段階的ロールアウト
def phased_rollout():
    # Phase 1: 最もモチベーションが高い部門
    pilot = deploy_ai_automation(departments["marketing"])
    evaluate(pilot)  # 2週間の評価期間
 
    if pilot.roi > 100:
        # Phase 2: 類似業務の部門
        wave2 = [departments["sales"], departments["cs"]]
        for dept in wave2:
            deploy_with_lessons_learned(dept, pilot.learnings)
 
        # Phase 3: 全社展開
        for dept in remaining_departments:
            deploy_with_best_practices(dept)

# 設計判断の記録テンプレート
class ArchitectureDecisionRecord:
    """ADR (Architecture Decision Record) の作成"""
 
    def __init__(self, title: str):
        self.title = title
        self.context = ""
        self.decision = ""
        self.consequences = []
        self.alternatives = []
 
    def set_context(self, context: str):
        """背景と課題の記述"""
        self.context = context
        return self
 
    def set_decision(self, decision: str):
        """決定内容の記述"""
        self.decision = decision
        return self
 
    def add_consequence(self, consequence: str, positive: bool = True):
        """結果の追加"""
        self.consequences.append({
            'description': consequence,
            'type': 'positive' if positive else 'negative'
        })
        return self
 
    def add_alternative(self, name: str, reason_rejected: str):
        """却下した代替案の追加"""
        self.alternatives.append({
            'name': name,
            'reason_rejected': reason_rejected
        })
        return self
 
    def to_markdown(self) -> str:
        """Markdown形式で出力"""
        md = f"# ADR: {self.title}\n\n"
        md += f"## 背景\n{self.context}\n\n"
        md += f"## 決定\n{self.decision}\n\n"
        md += "## 結果\n"
        for c in self.consequences:
            icon = "✅" if c['type'] == 'positive' else "⚠️"
            md += f"- {icon} {c['description']}\n"
        md += "\n## 却下した代替案\n"
        for a in self.alternatives:
            md += f"- **{a['name']}**: {a['reason_rejected']}\n"
        return md