🌐 JP | 🇬🇧 EN | Last sync: 2025-11-16
AI寺子屋トップ›プロセス・インフォマティクス›Qa›Chapter 4
この章で学ぶこと
- ISO 9001品質マネジメントシステムの7原則と要求事項
- 文書管理、内部監査、是正処置のプロセス
- リスクベース思考とFMEA（故障モード影響解析）
- Pythonを使った品質マネジメントシステムの実装
- サプライヤー品質管理と継続的改善の実践
4.1 ISO 9001の概要
4.1.1 ISO 9001とは
ISO 9001は、国際標準化機構（ISO）が制定した品質マネジメントシステム（QMS: Quality Management System）に関する国際規格です。組織が顧客満足度を向上させ、継続的に品質を改善するための要求事項を定めています。
📊 ISO 9001の世界的影響
- 認証組織数 : 世界で100万以上の組織が認証取得
- 対象国 : 170カ国以上で採用
- 最新版 : ISO 9001:2015（2015年9月発行）
- 次期改訂 : 2024-2025年頃を予定
4.1.2 品質マネジメントの7原則
ISO 9001:2015は以下の7つの品質マネジメント原則に基づいています：
| 原則 | 説明 | 実践例 |
|---|---|---|
| 1. 顧客重視 | 顧客の要求を理解し、期待を超える | 顧客満足度調査、VOC分析 |
| 2. リーダーシップ | 経営層が方向性と環境を確立 | 品質方針策定、目標設定 |
| 3. 人々の積極的参加 | 全員が能力を発揮し貢献 | 教育訓練、提案制度 |
| 4. プロセスアプローチ | 活動をプロセスとして管理 | フロー図作成、KPI設定 |
| 5. 改善 | 継続的な改善を追求 | PDCA、カイゼン活動 |
| 6. 客観的事実に基づく意思決定 | データと分析に基づく判断 | 統計的分析、ダッシュボード |
| 7. 関係性管理 | 利害関係者との相互利益関係 | サプライヤー評価、パートナーシップ |
4.2 ISO 9001要求事項とPython実装
4.2.1 コンプライアンスチェックリストシステム
ISO 9001の要求事項への適合を管理するチェックリストシステムを実装します。
例1: ISO 9001コンプライアンスチェッカー
import pandas as pd
from datetime import datetime, timedelta
from typing import List, Dict, Optional
import json
class ISO9001ComplianceChecker:
"""ISO 9001要求事項への適合性をチェック・管理するシステム
ISO 9001:2015の10章構成に基づいた要求事項の適合性評価、
エビデンス管理、改善計画の追跡を行う。
"""
def __init__(self):
# ISO 9001:2015の要求事項定義
self.requirements = {
"4": {"title": "組織の状況", "subclauses": [
"4.1 組織及びその状況の理解",
"4.2 利害関係者のニーズ及び期待の理解",
"4.3 品質マネジメントシステムの適用範囲の決定",
"4.4 品質マネジメントシステム及びそのプロセス"
]},
"5": {"title": "リーダーシップ", "subclauses": [
"5.1 リーダーシップ及びコミットメント",
"5.2 方針",
"5.3 組織の役割、責任及び権限"
]},
"6": {"title": "計画", "subclauses": [
"6.1 リスク及び機会への取組み",
"6.2 品質目標及びそれを達成するための計画策定",
"6.3 変更の計画"
]},
"7": {"title": "支援", "subclauses": [
"7.1 資源",
"7.2 力量",
"7.3 認識",
"7.4 コミュニケーション",
"7.5 文書化した情報"
]},
"8": {"title": "運用", "subclauses": [
"8.1 運用の計画及び管理",
"8.2 製品及びサービスに関する要求事項",
"8.3 製品及びサービスの設計・開発",
"8.4 外部から提供されるプロセス、製品及びサービスの管理",
"8.5 製品及びサービスの提供",
"8.6 製品及びサービスのリリース",
"8.7 不適合なアウトプットの管理"
]},
"9": {"title": "パフォーマンス評価", "subclauses": [
"9.1 監視、測定、分析及び評価",
"9.2 内部監査",
"9.3 マネジメントレビュー"
]},
"10": {"title": "改善", "subclauses": [
"10.1 一般",
"10.2 不適合及び是正処置",
"10.3 継続的改善"
]}
}
self.compliance_data = []
def add_compliance_record(self, clause: str, subclause: str,
status: str, evidence: str,
auditor: str, notes: str = "") -> Dict:
"""コンプライアンス記録を追加
Args:
clause: 章番号（例: "4", "5"）
subclause: 詳細要求事項
status: 適合状況（"Compliant", "Partial", "Non-Compliant", "Not Applicable"）
evidence: エビデンス（文書番号、参照先）
auditor: 監査者名
notes: 備考
Returns:
追加された記録
"""
record = {
"date": datetime.now().strftime("%Y-%m-%d"),
"clause": clause,
"subclause": subclause,
"status": status,
"evidence": evidence,
"auditor": auditor,
"notes": notes,
"action_required": status in ["Partial", "Non-Compliant"]
}
self.compliance_data.append(record)
return record
def get_compliance_summary(self) -> pd.DataFrame:
"""適合性サマリーを取得
Returns:
章ごとの適合率を含むDataFrame
"""
if not self.compliance_data:
return pd.DataFrame()
df = pd.DataFrame(self.compliance_data)
# 章ごとの集計
summary = df.groupby('clause').agg({
'status': lambda x: (x == 'Compliant').sum() / len(x) * 100,
'subclause': 'count',
'action_required': 'sum'
}).round(2)
summary.columns = ['適合率(%)', '要求事項数', '対応必要項目']
summary.index.name = '章'
return summary
def get_non_compliant_items(self) -> List[Dict]:
"""非適合項目のリストを取得
Returns:
非適合・部分適合項目のリスト
"""
return [
record for record in self.compliance_data
if record['status'] in ['Partial', 'Non-Compliant']
]
def export_audit_report(self, filename: str):
"""監査レポートをエクスポート
Args:
filename: 出力ファイル名（JSON形式）
"""
report = {
"report_date": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"total_requirements": len(self.compliance_data),
"summary": self.get_compliance_summary().to_dict(),
"non_compliant_items": self.get_non_compliant_items(),
"overall_compliance": (
sum(1 for r in self.compliance_data if r['status'] == 'Compliant')
/ len(self.compliance_data) * 100
if self.compliance_data else 0
)
}
with open(filename, 'w', encoding='utf-8') as f:
json.dump(report, f, indent=2, ensure_ascii=False)
print(f"監査レポートを {filename} に出力しました")
print(f"総合適合率: {report['overall_compliance']:.2f}%")
# 使用例
checker = ISO9001ComplianceChecker()
# コンプライアンス記録の追加
checker.add_compliance_record(
clause="4",
subclause="4.1 組織及びその状況の理解",
status="Compliant",
evidence="DOC-001: SWOT分析書",
auditor="田中太郎"
)
checker.add_compliance_record(
clause="6",
subclause="6.1 リスク及び機会への取組み",
status="Partial",
evidence="DOC-015: リスク登録簿（更新必要）",
auditor="佐藤花子",
notes="リスク評価が6ヶ月以上更新されていない"
)
checker.add_compliance_record(
clause="9",
subclause="9.2 内部監査",
status="Non-Compliant",
evidence="なし",
auditor="鈴木一郎",
notes="年次内部監査が実施されていない"
)
# サマリー表示
print("\n=== ISO 9001適合性サマリー ===")
print(checker.get_compliance_summary())
# 非適合項目の確認
print("\n=== 対応が必要な項目 ===")
for item in checker.get_non_compliant_items():
print(f"[{item['clause']}] {item['subclause']}")
print(f" 状態: {item['status']}")
print(f" 備考: {item['notes']}\n")
# レポート出力
checker.export_audit_report("iso9001_audit_report.json")💡 実装のポイント
- 要求事項の構造化 : ISO 9001の章立てに沿った管理
- エビデンストレーサビリティ : 各要求事項への証拠を明確化
- アクション管理 : 非適合項目への自動フラグ付け
- 適合率の可視化 : 章ごとの進捗状況を数値化
4.2.2 文書管理・バージョン管理システム
ISO 9001要求事項7.5「文書化した情報」に対応する文書管理システムです。
例2: 文書管理システム
import hashlib
from datetime import datetime
from typing import List, Dict, Optional
import os
class DocumentControlSystem:
"""ISO 9001対応の文書管理システム
文書のバージョン管理、承認ワークフロー、改訂履歴、
有効期限管理を行う。
"""
def __init__(self):
self.documents = []
self.revision_history = []
def create_document(self, doc_id: str, title: str, doc_type: str,
content: str, owner: str,
review_period_days: int = 365) -> Dict:
"""新規文書を作成
Args:
doc_id: 文書番号（例: "QP-001"）
title: 文書タイトル
doc_type: 文書種別（"手順書", "様式", "記録"）
content: 文書内容
owner: 文書管理責任者
review_period_days: 見直し周期（日数）
Returns:
作成された文書情報
"""
# 文書のハッシュ値を計算（改ざん検知用）
content_hash = hashlib.sha256(content.encode()).hexdigest()
document = {
"doc_id": doc_id,
"title": title,
"type": doc_type,
"version": "1.0",
"status": "Draft", # Draft, Under Review, Approved, Obsolete
"content_hash": content_hash,
"owner": owner,
"created_date": datetime.now(),
"approved_date": None,
"approved_by": None,
"next_review_date": None,
"review_period_days": review_period_days,
"access_level": "Internal" # Public, Internal, Confidential
}
self.documents.append(document)
# 改訂履歴に記録
self._add_revision_history(
doc_id, "1.0", "Draft", "新規作成", owner
)
return document
def approve_document(self, doc_id: str, approver: str) -> bool:
"""文書を承認
Args:
doc_id: 文書番号
approver: 承認者名
Returns:
承認成功の可否
"""
doc = self._find_document(doc_id)
if not doc:
print(f"文書 {doc_id} が見つかりません")
return False
if doc['status'] != 'Draft' and doc['status'] != 'Under Review':
print(f"文書 {doc_id} は承認できない状態です: {doc['status']}")
return False
doc['status'] = 'Approved'
doc['approved_date'] = datetime.now()
doc['approved_by'] = approver
doc['next_review_date'] = (
datetime.now() +
pd.Timedelta(days=doc['review_period_days'])
)
self._add_revision_history(
doc_id, doc['version'], "Approved",
f"{approver}により承認", approver
)
print(f"文書 {doc_id} v{doc['version']} を承認しました")
return True
def revise_document(self, doc_id: str, new_content: str,
revision_note: str, revised_by: str) -> Dict:
"""文書を改訂（新バージョン作成）
Args:
doc_id: 文書番号
new_content: 新しい文書内容
revision_note: 改訂理由
revised_by: 改訂者
Returns:
改訂後の文書情報
"""
doc = self._find_document(doc_id)
if not doc:
raise ValueError(f"文書 {doc_id} が見つかりません")
# バージョン番号を更新
major, minor = map(int, doc['version'].split('.'))
doc['version'] = f"{major}.{minor + 1}"
# 文書をDraft状態に戻す
doc['status'] = 'Draft'
doc['content_hash'] = hashlib.sha256(new_content.encode()).hexdigest()
doc['approved_date'] = None
doc['approved_by'] = None
# 改訂履歴に記録
self._add_revision_history(
doc_id, doc['version'], "Draft", revision_note, revised_by
)
print(f"文書 {doc_id} を v{doc['version']} に改訂しました")
return doc
def get_documents_for_review(self) -> List[Dict]:
"""見直しが必要な文書のリストを取得
Returns:
見直し期限が近い、または過ぎた文書のリスト
"""
now = datetime.now()
review_needed = []
for doc in self.documents:
if doc['status'] != 'Approved':
continue
if doc['next_review_date'] and doc['next_review_date'] <= now:
days_overdue = (now - doc['next_review_date']).days
review_needed.append({
**doc,
"days_overdue": days_overdue,
"priority": "High" if days_overdue > 30 else "Medium"
})
return sorted(review_needed, key=lambda x: x['days_overdue'], reverse=True)
def get_revision_history(self, doc_id: str) -> pd.DataFrame:
"""文書の改訂履歴を取得
Args:
doc_id: 文書番号
Returns:
改訂履歴のDataFrame
"""
history = [
h for h in self.revision_history if h['doc_id'] == doc_id
]
if not history:
return pd.DataFrame()
return pd.DataFrame(history)
def _find_document(self, doc_id: str) -> Optional[Dict]:
"""文書IDで文書を検索（内部用）"""
for doc in self.documents:
if doc['doc_id'] == doc_id:
return doc
return None
def _add_revision_history(self, doc_id: str, version: str,
status: str, note: str, user: str):
"""改訂履歴を追加（内部用）"""
self.revision_history.append({
"doc_id": doc_id,
"version": version,
"status": status,
"note": note,
"user": user,
"timestamp": datetime.now()
})
# 使用例
import pandas as pd
dcs = DocumentControlSystem()
# 品質手順書を作成
doc = dcs.create_document(
doc_id="QP-001",
title="内部監査手順書",
doc_type="手順書",
content="1. 目的\n2. 適用範囲\n3. 手順...",
owner="品質管理部長",
review_period_days=365
)
print(f"文書作成: {doc['doc_id']} - {doc['title']}")
# 文書を承認
dcs.approve_document("QP-001", "工場長")
# 文書を改訂
dcs.revise_document(
doc_id="QP-001",
new_content="1. 目的\n2. 適用範囲（更新）\n3. 手順（追加）...",
revision_note="適用範囲を拡大、手順を詳細化",
revised_by="品質管理部長"
)
# 改訂後に再承認
dcs.approve_document("QP-001", "工場長")
# 改訂履歴を表示
print("\n=== 改訂履歴 ===")
print(dcs.get_revision_history("QP-001"))4.2.3 内部監査計画・追跡システム
ISO 9001要求事項9.2「内部監査」に対応する監査管理システムです。
例3: 内部監査管理システム
from datetime import datetime, timedelta
import pandas as pd
from typing import List, Dict
class InternalAuditSystem:
"""内部監査の計画、実施、フォローアップを管理
年間監査計画の作成、監査の実施記録、発見事項の追跡、
是正処置の確認を行う。
"""
def __init__(self):
self.audit_plan = []
self.audit_findings = []
self.corrective_actions = []
def create_annual_audit_plan(self, year: int,
departments: List[str],
frequency: str = "Annual") -> List[Dict]:
"""年間監査計画を作成
Args:
year: 対象年
departments: 監査対象部門のリスト
frequency: 監査頻度（"Annual", "Semi-Annual", "Quarterly"）
Returns:
作成された監査計画のリスト
"""
freq_map = {
"Annual": 1,
"Semi-Annual": 2,
"Quarterly": 4
}
audits_per_year = freq_map.get(frequency, 1)
for dept in departments:
for i in range(audits_per_year):
# 監査時期を均等に配分
month = (i * 12 // audits_per_year) + 1
planned_date = datetime(year, month, 15)
audit = {
"audit_id": f"IA-{year}-{dept[:3].upper()}-{i+1:02d}",
"department": dept,
"audit_type": "Internal",
"scope": f"{dept}部門の品質マネジメントシステム",
"planned_date": planned_date,
"status": "Planned", # Planned, In Progress, Completed
"lead_auditor": None,
"audit_team": [],
"completion_date": None
}
self.audit_plan.append(audit)
print(f"{year}年度の監査計画を作成: {len(self.audit_plan)}件")
return self.audit_plan
def assign_auditors(self, audit_id: str, lead_auditor: str,
team_members: List[str]):
"""監査チームをアサイン
Args:
audit_id: 監査ID
lead_auditor: 主任監査員
team_members: 監査チームメンバー
"""
audit = self._find_audit(audit_id)
if audit:
audit['lead_auditor'] = lead_auditor
audit['audit_team'] = team_members
print(f"監査 {audit_id} にチームをアサインしました")
def record_finding(self, audit_id: str, finding_type: str,
clause: str, description: str,
severity: str) -> Dict:
"""監査発見事項を記録
Args:
audit_id: 監査ID
finding_type: "Non-Conformance", "Observation", "Opportunity"
clause: 関連するISO条項
description: 発見事項の詳細
severity: 重大度（"Major", "Minor"）
Returns:
記録された発見事項
"""
finding = {
"finding_id": f"F-{len(self.audit_findings) + 1:04d}",
"audit_id": audit_id,
"type": finding_type,
"clause": clause,
"description": description,
"severity": severity,
"recorded_date": datetime.now(),
"status": "Open", # Open, Under Review, Closed
"ca_required": finding_type == "Non-Conformance"
}
self.audit_findings.append(finding)
return finding
def complete_audit(self, audit_id: str):
"""監査を完了
Args:
audit_id: 監査ID
"""
audit = self._find_audit(audit_id)
if audit:
audit['status'] = 'Completed'
audit['completion_date'] = datetime.now()
# 関連する発見事項の数をカウント
findings = [f for f in self.audit_findings
if f['audit_id'] == audit_id]
print(f"監査 {audit_id} を完了しました")
print(f" 発見事項: {len(findings)}件")
print(f" 不適合: {sum(1 for f in findings if f['type'] == 'Non-Conformance')}件")
def get_audit_dashboard(self) -> pd.DataFrame:
"""監査ダッシュボードデータを取得
Returns:
監査計画の進捗状況を示すDataFrame
"""
if not self.audit_plan:
return pd.DataFrame()
df = pd.DataFrame(self.audit_plan)
# ステータス別集計
status_summary = df['status'].value_counts()
# 発見事項の統計
findings_df = pd.DataFrame(self.audit_findings)
if not findings_df.empty:
nc_count = (findings_df['type'] == 'Non-Conformance').sum()
open_findings = (findings_df['status'] == 'Open').sum()
else:
nc_count = 0
open_findings = 0
summary = pd.DataFrame({
"指標": ["計画済", "完了", "進行中", "不適合件数", "未解決発見事項"],
"値": [
(df['status'] == 'Planned').sum(),
(df['status'] == 'Completed').sum(),
(df['status'] == 'In Progress').sum(),
nc_count,
open_findings
]
})
return summary
def get_overdue_audits(self) -> List[Dict]:
"""期限超過の監査を取得
Returns:
実施予定日を過ぎた監査のリスト
"""
now = datetime.now()
overdue = []
for audit in self.audit_plan:
if (audit['status'] == 'Planned' and
audit['planned_date'] < now):
days_overdue = (now - audit['planned_date']).days
overdue.append({
**audit,
"days_overdue": days_overdue
})
return sorted(overdue, key=lambda x: x['days_overdue'], reverse=True)
def _find_audit(self, audit_id: str) -> Optional[Dict]:
"""監査IDで監査を検索（内部用）"""
for audit in self.audit_plan:
if audit['audit_id'] == audit_id:
return audit
return None
# 使用例
ias = InternalAuditSystem()
# 年間監査計画を作成
departments = ["製造", "品質管理", "購買", "設計", "営業"]
ias.create_annual_audit_plan(2025, departments, frequency="Annual")
# 監査員をアサイン
ias.assign_auditors(
"IA-2025-製造-01",
lead_auditor="田中太郎（主任監査員）",
team_members=["佐藤花子", "鈴木一郎"]
)
# 発見事項を記録
finding1 = ias.record_finding(
audit_id="IA-2025-製造-01",
finding_type="Non-Conformance",
clause="8.5 製品及びサービスの提供",
description="作業手順書が最新版に更新されていない（3年前のまま）",
severity="Major"
)
finding2 = ias.record_finding(
audit_id="IA-2025-製造-01",
finding_type="Observation",
clause="7.2 力量",
description="教育訓練記録が一部不完全",
severity="Minor"
)
print(f"\n発見事項を記録: {finding1['finding_id']}, {finding2['finding_id']}")
# 監査を完了
ias.complete_audit("IA-2025-製造-01")
# ダッシュボード表示
print("\n=== 監査ダッシュボード ===")
print(ias.get_audit_dashboard())4.2.4 不適合管理・CAPA（是正処置・予防処置）システム
ISO 9001要求事項10.2「不適合及び是正処置」に対応するシステムです。
例4: CAPA管理システム
from datetime import datetime, timedelta
from typing import List, Dict, Optional
import pandas as pd
class CAPASystem:
"""是正処置・予防処置（CAPA）管理システム
不適合の記録、根本原因分析、是正処置の計画・実施、
有効性確認までのライフサイクルを管理。
"""
def __init__(self):
self.capas = []
self.root_cause_analyses = []
def create_capa(self, source: str, description: str,
category: str, severity: str,
reported_by: str) -> Dict:
"""CAPA（是正処置・予防処置）を起票
Args:
source: 発生源（"Internal Audit", "Customer Complaint",
"Process Monitoring", "Supplier Issue"）
description: 不適合内容の詳細
category: 分類（"Product", "Process", "Documentation", "Training"）
severity: 重大度（"Critical", "Major", "Minor"）
reported_by: 報告者
Returns:
作成されたCAPA情報
"""
capa_id = f"CAPA-{datetime.now().strftime('%Y%m%d')}-{len(self.capas) + 1:03d}"
# 重大度に応じた対応期限を設定
due_days = {"Critical": 7, "Major": 30, "Minor": 60}
capa = {
"capa_id": capa_id,
"source": source,
"description": description,
"category": category,
"severity": severity,
"reported_by": reported_by,
"reported_date": datetime.now(),
"status": "Open", # Open, Investigation, Action, Verification, Closed
"assigned_to": None,
"due_date": datetime.now() + timedelta(days=due_days[severity]),
"root_cause": None,
"corrective_action": None,
"preventive_action": None,
"effectiveness_verified": False,
"closed_date": None
}
self.capas.append(capa)
print(f"CAPA起票: {capa_id} (期限: {capa['due_date'].strftime('%Y-%m-%d')})")
return capa
def assign_capa(self, capa_id: str, assignee: str):
"""CAPAを担当者にアサイン
Args:
capa_id: CAPA番号
assignee: 担当者名
"""
capa = self._find_capa(capa_id)
if capa:
capa['assigned_to'] = assignee
capa['status'] = 'Investigation'
print(f"CAPA {capa_id} を {assignee} にアサインしました")
def record_root_cause_analysis(self, capa_id: str,
method: str, findings: str,
root_cause: str, analyst: str) -> Dict:
"""根本原因分析を記録
Args:
capa_id: CAPA番号
method: 分析手法（"5 Whys", "Fishbone", "FTA", "FMEA"）
findings: 分析結果
root_cause: 特定された根本原因
analyst: 分析者
Returns:
記録された分析情報
"""
analysis = {
"capa_id": capa_id,
"method": method,
"findings": findings,
"root_cause": root_cause,
"analyst": analyst,
"analysis_date": datetime.now()
}
self.root_cause_analyses.append(analysis)
# CAPAに根本原因を記録
capa = self._find_capa(capa_id)
if capa:
capa['root_cause'] = root_cause
return analysis
def define_corrective_action(self, capa_id: str,
corrective_action: str,
preventive_action: str = ""):
"""是正処置・予防処置を定義
Args:
capa_id: CAPA番号
corrective_action: 是正処置（発生した問題への対応）
preventive_action: 予防処置（再発防止策）
"""
capa = self._find_capa(capa_id)
if not capa:
print(f"CAPA {capa_id} が見つかりません")
return
capa['corrective_action'] = corrective_action
capa['preventive_action'] = preventive_action
capa['status'] = 'Action'
print(f"CAPA {capa_id} の処置を定義しました")
def verify_effectiveness(self, capa_id: str,
verified_by: str,
is_effective: bool,
verification_note: str):
"""是正処置の有効性を検証
Args:
capa_id: CAPA番号
verified_by: 検証者
is_effective: 有効性確認結果
verification_note: 検証メモ
"""
capa = self._find_capa(capa_id)
if not capa:
print(f"CAPA {capa_id} が見つかりません")
return
capa['effectiveness_verified'] = is_effective
capa['status'] = 'Closed' if is_effective else 'Action'
if is_effective:
capa['closed_date'] = datetime.now()
print(f"CAPA {capa_id} を完了しました（有効性確認済み）")
else:
print(f"CAPA {capa_id} の有効性が不十分です。追加処置が必要です")
print(f" 理由: {verification_note}")
def get_open_capas(self) -> pd.DataFrame:
"""未完了CAPAのリストを取得
Returns:
未完了CAPAのDataFrame
"""
open_capas = [c for c in self.capas if c['status'] != 'Closed']
if not open_capas:
return pd.DataFrame()
df = pd.DataFrame(open_capas)
df['days_open'] = df['reported_date'].apply(
lambda x: (datetime.now() - x).days
)
df['overdue'] = df['due_date'] < datetime.now()
return df[['capa_id', 'severity', 'category', 'status',
'assigned_to', 'due_date', 'days_open', 'overdue']]
def get_capa_metrics(self) -> Dict:
"""CAPAメトリクスを取得
Returns:
主要CAPAメトリクス
"""
if not self.capas:
return {}
df = pd.DataFrame(self.capas)
# 平均解決日数（完了済みのみ）
closed_capas = df[df['status'] == 'Closed'].copy()
if not closed_capas.empty:
closed_capas['resolution_days'] = (
closed_capas['closed_date'] - closed_capas['reported_date']
).dt.days
avg_resolution_days = closed_capas['resolution_days'].mean()
else:
avg_resolution_days = 0
# 期限超過率
now = datetime.now()
overdue_count = sum(1 for c in self.capas
if c['status'] != 'Closed' and c['due_date'] < now)
metrics = {
"総CAPA数": len(self.capas),
"未完了": (df['status'] != 'Closed').sum(),
"完了": (df['status'] == 'Closed').sum(),
"期限超過": overdue_count,
"平均解決日数": round(avg_resolution_days, 1),
"完了率(%)": round((df['status'] == 'Closed').sum() / len(self.capas) * 100, 1)
}
return metrics
def _find_capa(self, capa_id: str) -> Optional[Dict]:
"""CAPA IDでCAPAを検索（内部用）"""
for capa in self.capas:
if capa['capa_id'] == capa_id:
return capa
return None
# 使用例
capa_system = CAPASystem()
# CAPAを起票
capa1 = capa_system.create_capa(
source="Internal Audit",
description="製造ラインAで作業手順書の不遵守が発見された。手順3のステップが省略されている。",
category="Process",
severity="Major",
reported_by="監査員 田中"
)
# 担当者にアサイン
capa_system.assign_capa(capa1['capa_id'], "製造部長 佐藤")
# 根本原因分析を実施
capa_system.record_root_cause_analysis(
capa_id=capa1['capa_id'],
method="5 Whys",
findings="""
なぜ手順が省略された? → 時間がかかるから
なぜ時間がかかる? → 設備のセットアップに時間を要する
なぜセットアップに時間がかかる? → 治具の準備が煩雑
なぜ治具の準備が煩雑? → 整理整頓されていない
なぜ整理整頓されていない? → 5S活動が形骸化している
""",
root_cause="5S活動の形骸化により、治具管理が不十分。作業者が手順省略を選択。",
analyst="品質管理部 鈴木"
)
# 是正処置・予防処置を定義
capa_system.define_corrective_action(
capa_id=capa1['capa_id'],
corrective_action="治具の整理整頓を即座に実施。手順書遵守を再教育。",
preventive_action="5S活動を再活性化。月次監査を実施。治具管理システムを導入。"
)
# 有効性を検証
capa_system.verify_effectiveness(
capa_id=capa1['capa_id'],
verified_by="品質管理部長",
is_effective=True,
verification_note="1ヶ月後に再監査を実施。手順遵守率100%を確認。治具管理も改善。"
)
# メトリクス表示
print("\n=== CAPAメトリクス ===")
metrics = capa_system.get_capa_metrics()
for key, value in metrics.items():
print(f"{key}: {value}")4.3 リスクベース思考とFMEA
4.3.1 FMEA（故障モード影響解析）の実装
ISO 9001:2015の6.1「リスク及び機会への取組み」では、リスクベース思考が求められます。FMEA（Failure Mode and Effects Analysis）は、プロセスや製品の潜在的な故障モードを特定し、優先順位付けする手法です。
例5: FMEA実装システム
import pandas as pd
from typing import List, Dict
import numpy as np
class FMEASystem:
"""FMEA（故障モード影響解析）実装システム
プロセスFMEAまたは設計FMEAを実施し、RPN（Risk Priority Number）
に基づいてリスクを優先順位付け。
"""
def __init__(self, fmea_type: str = "Process"):
"""
Args:
fmea_type: FMEAタイプ（"Process" or "Design"）
"""
self.fmea_type = fmea_type
self.failure_modes = []
def add_failure_mode(self, process_step: str,
potential_failure: str,
effects: str, severity: int,
causes: str, occurrence: int,
current_controls: str, detection: int,
responsible_person: str = "") -> Dict:
"""故障モードを追加
Args:
process_step: プロセスステップ/機能
potential_failure: 潜在的故障モード
effects: 故障の影響
severity: 深刻度（1-10, 10が最も深刻）
causes: 故障の原因
occurrence: 発生頻度（1-10, 10が最も頻繁）
current_controls: 現在の管理方法
detection: 検出度（1-10, 10が最も検出困難）
responsible_person: 責任者
Returns:
追加された故障モード情報
"""
# RPN（Risk Priority Number）を計算
rpn = severity * occurrence * detection
failure_mode = {
"process_step": process_step,
"potential_failure": potential_failure,
"effects": effects,
"severity": severity,
"causes": causes,
"occurrence": occurrence,
"current_controls": current_controls,
"detection": detection,
"rpn": rpn,
"responsible_person": responsible_person,
"recommended_actions": "",
"actions_taken": "",
"new_severity": None,
"new_occurrence": None,
"new_detection": None,
"new_rpn": None
}
self.failure_modes.append(failure_mode)
return failure_mode
def prioritize_risks(self, threshold: int = 100) -> pd.DataFrame:
"""リスクを優先順位付け
Args:
threshold: 対応が必要なRPN閾値
Returns:
RPNでソートされたDataFrame
"""
if not self.failure_modes:
return pd.DataFrame()
df = pd.DataFrame(self.failure_modes)
# RPNでソート
df = df.sort_values('rpn', ascending=False)
# リスクレベルを分類
df['risk_level'] = df['rpn'].apply(lambda x:
'High' if x >= 200 else
'Medium' if x >= 100 else
'Low'
)
# 対応が必要な項目をフラグ
df['action_required'] = df['rpn'] >= threshold
return df
def recommend_actions(self, index: int, actions: str):
"""推奨処置を記録
Args:
index: 故障モードのインデックス
actions: 推奨処置の内容
"""
if 0 <= index < len(self.failure_modes):
self.failure_modes[index]['recommended_actions'] = actions
def record_actions_taken(self, index: int, actions: str,
new_severity: int, new_occurrence: int,
new_detection: int):
"""実施した処置と改善後の評価を記録
Args:
index: 故障モードのインデックス
actions: 実施した処置
new_severity: 改善後の深刻度
new_occurrence: 改善後の発生頻度
new_detection: 改善後の検出度
"""
if 0 <= index < len(self.failure_modes):
fm = self.failure_modes[index]
fm['actions_taken'] = actions
fm['new_severity'] = new_severity
fm['new_occurrence'] = new_occurrence
fm['new_detection'] = new_detection
fm['new_rpn'] = new_severity * new_occurrence * new_detection
print(f"RPN改善: {fm['rpn']} → {fm['new_rpn']} "
f"({(1 - fm['new_rpn']/fm['rpn'])*100:.1f}%削減)")
def generate_fmea_report(self, filename: str):
"""FMEAレポートを生成
Args:
filename: 出力ファイル名（Excel形式）
"""
df = self.prioritize_risks()
# Excel出力
with pd.ExcelWriter(filename, engine='openpyxl') as writer:
df.to_excel(writer, sheet_name='FMEA', index=False)
# サマリーシートを作成
summary = pd.DataFrame({
'項目': [
'総故障モード数',
'高リスク項目（RPN≥200）',
'中リスク項目（100≤RPN<200）',
'低リスク項目（RPN<100）',
'平均RPN',
'最大RPN'
],
'値': [
len(df),
(df['rpn'] >= 200).sum(),
((df['rpn'] >= 100) & (df['rpn'] < 200)).sum(),
(df['rpn'] < 100).sum(),
round(df['rpn'].mean(), 1),
df['rpn'].max()
]
})
summary.to_excel(writer, sheet_name='サマリー', index=False)
print(f"FMEAレポートを {filename} に出力しました")
# 使用例
fmea = FMEASystem(fmea_type="Process")
# 故障モードを追加（製造プロセスの例）
fmea.add_failure_mode(
process_step="原材料受入検査",
potential_failure="不適合原材料の受入",
effects="不良品の製造、顧客クレーム、リコール",
severity=9, # 非常に深刻
causes="検査基準の不明確、検査員の力量不足",
occurrence=4, # 時々発生
current_controls="サンプリング検査",
detection=5, # 検出はやや困難
responsible_person="品質管理部"
)
fmea.add_failure_mode(
process_step="成形加工",
potential_failure="寸法不良",
effects="組立不良、性能低下",
severity=7,
causes="金型の摩耗、温度管理不良",
occurrence=6,
current_controls="初品検査、定期測定",
detection=3, # 比較的検出容易
responsible_person="製造部"
)
fmea.add_failure_mode(
process_step="最終検査",
potential_failure="検査漏れ",
effects="不良品の出荷",
severity=10, # 最も深刻
causes="検査項目の多さ、時間的プレッシャー",
occurrence=2, # 稀に発生
current_controls="ダブルチェック",
detection=7, # 検出困難
responsible_person="品質保証部"
)
# リスク優先順位付け
print("=== FMEAリスク分析 ===")
risk_df = fmea.prioritize_risks(threshold=100)
print(risk_df[['process_step', 'potential_failure', 'rpn', 'risk_level']])
# 推奨処置を記録（最高RPNの項目に対して）
fmea.recommend_actions(
index=2, # 最終検査の検査漏れ
actions="自動検査装置の導入、検査手順の簡素化、リストによるトレーサビリティ確保"
)
# 処置実施後の評価
fmea.record_actions_taken(
index=2,
actions="自動外観検査装置を導入。検査項目を重要度で分類。",
new_severity=10, # 深刻度は変わらず
new_occurrence=1, # 発生頻度が大幅減少
new_detection=2 # 検出度が改善（自動化により）
)
# レポート生成
fmea.generate_fmea_report("fmea_report.xlsx")📊 RPNの評価基準
| RPN範囲 | リスクレベル | 対応 |
|---|---|---|
| 200-1000 | 高リスク | 即座に処置が必要 |
| 100-199 | 中リスク | 計画的に処置を実施 |
| 1-99 | 低リスク | 監視を継続 |
4.3.2 サプライヤー品質評価システム
ISO 9001要求事項8.4「外部から提供されるプロセス、製品及びサービスの管理」に対応します。
例6: サプライヤー品質管理システム
import pandas as pd
from datetime import datetime
from typing import List, Dict
import numpy as np
class SupplierQualitySystem:
"""サプライヤー品質管理システム
サプライヤーの評価、品質パフォーマンスの追跡、
改善活動の管理を行う。
"""
def __init__(self):
self.suppliers = []
self.quality_records = []
def register_supplier(self, supplier_id: str, name: str,
product_category: str,
criticality: str = "Medium") -> Dict:
"""サプライヤーを登録
Args:
supplier_id: サプライヤーID
name: サプライヤー名
product_category: 供給品目カテゴリ
criticality: 重要度（"Critical", "High", "Medium", "Low"）
Returns:
登録されたサプライヤー情報
"""
supplier = {
"supplier_id": supplier_id,
"name": name,
"product_category": product_category,
"criticality": criticality,
"registration_date": datetime.now(),
"status": "Approved", # Approved, Conditional, Suspended
"quality_score": None,
"delivery_score": None,
"overall_rating": None
}
self.suppliers.append(supplier)
print(f"サプライヤー登録: {name} ({supplier_id})")
return supplier
def record_quality_data(self, supplier_id: str,
delivery_date: datetime,
lot_number: str,
quantity_received: int,
quantity_accepted: int,
defect_count: int,
defect_types: str = "") -> Dict:
"""品質データを記録
Args:
supplier_id: サプライヤーID
delivery_date: 納品日
lot_number: ロット番号
quantity_received: 受領数量
quantity_accepted: 合格数量
defect_count: 不良数
defect_types: 不良の種類
Returns:
記録された品質データ
"""
# PPM（Parts Per Million）を計算
ppm = (defect_count / quantity_received * 1_000_000) if quantity_received > 0 else 0
record = {
"supplier_id": supplier_id,
"delivery_date": delivery_date,
"lot_number": lot_number,
"quantity_received": quantity_received,
"quantity_accepted": quantity_accepted,
"defect_count": defect_count,
"defect_types": defect_types,
"ppm": ppm,
"acceptance_rate": (quantity_accepted / quantity_received * 100)
if quantity_received > 0 else 0
}
self.quality_records.append(record)
return record
def calculate_supplier_rating(self, supplier_id: str,
months: int = 6) -> Dict:
"""サプライヤーの評価を計算
Args:
supplier_id: サプライヤーID
months: 評価期間（月数）
Returns:
評価結果
"""
# 期間内の品質記録を取得
cutoff_date = datetime.now() - pd.Timedelta(days=months * 30)
records = [
r for r in self.quality_records
if r['supplier_id'] == supplier_id and r['delivery_date'] >= cutoff_date
]
if not records:
return {"error": "評価期間内のデータがありません"}
df = pd.DataFrame(records)
# 品質スコア（0-100）
avg_ppm = df['ppm'].mean()
quality_score = max(0, 100 - (avg_ppm / 100)) # 10,000 PPM = 0点
# 総合評価（A-D）
if quality_score >= 95:
overall_rating = "A"
elif quality_score >= 85:
overall_rating = "B"
elif quality_score >= 70:
overall_rating = "C"
else:
overall_rating = "D"
rating = {
"supplier_id": supplier_id,
"evaluation_period": f"{months}ヶ月",
"total_deliveries": len(records),
"total_quantity": df['quantity_received'].sum(),
"total_defects": df['defect_count'].sum(),
"avg_ppm": round(avg_ppm, 2),
"quality_score": round(quality_score, 2),
"overall_rating": overall_rating
}
# サプライヤー情報を更新
supplier = self._find_supplier(supplier_id)
if supplier:
supplier['quality_score'] = quality_score
supplier['overall_rating'] = overall_rating
return rating
def get_supplier_performance_summary(self) -> pd.DataFrame:
"""全サプライヤーのパフォーマンスサマリーを取得
Returns:
サプライヤー別のパフォーマンスDataFrame
"""
if not self.suppliers:
return pd.DataFrame()
# 各サプライヤーの評価を計算
for supplier in self.suppliers:
self.calculate_supplier_rating(supplier['supplier_id'])
df = pd.DataFrame(self.suppliers)
# 重要度でソート
criticality_order = {"Critical": 0, "High": 1, "Medium": 2, "Low": 3}
df['criticality_rank'] = df['criticality'].map(criticality_order)
df = df.sort_values(['criticality_rank', 'quality_score'], ascending=[True, False])
return df[['supplier_id', 'name', 'criticality',
'quality_score', 'overall_rating', 'status']]
def identify_improvement_targets(self, threshold_rating: str = "C") -> List[Dict]:
"""改善対象サプライヤーを特定
Args:
threshold_rating: 改善対象とする評価閾値
Returns:
改善対象サプライヤーのリスト
"""
rating_map = {"A": 4, "B": 3, "C": 2, "D": 1}
threshold_value = rating_map.get(threshold_rating, 2)
targets = []
for supplier in self.suppliers:
if supplier['overall_rating']:
rating_value = rating_map.get(supplier['overall_rating'], 0)
if rating_value <= threshold_value:
targets.append({
"supplier_id": supplier['supplier_id'],
"name": supplier['name'],
"rating": supplier['overall_rating'],
"quality_score": supplier['quality_score'],
"action": "品質改善計画の策定と実施が必要"
})
return targets
def _find_supplier(self, supplier_id: str) -> Optional[Dict]:
"""サプライヤーIDでサプライヤーを検索（内部用）"""
for supplier in self.suppliers:
if supplier['supplier_id'] == supplier_id:
return supplier
return None
# 使用例
sqs = SupplierQualitySystem()
# サプライヤーを登録
sqs.register_supplier("SUP-001", "ABC部品工業", "電子部品", criticality="Critical")
sqs.register_supplier("SUP-002", "XYZ樹脂", "樹脂材料", criticality="High")
sqs.register_supplier("SUP-003", "123梱包資材", "梱包材", criticality="Low")
# 品質データを記録
sqs.record_quality_data(
supplier_id="SUP-001",
delivery_date=datetime(2025, 1, 15),
lot_number="LOT-20250115-001",
quantity_received=10000,
quantity_accepted=9950,
defect_count=50,
defect_types="寸法不良"
)
sqs.record_quality_data(
supplier_id="SUP-001",
delivery_date=datetime(2025, 2, 20),
lot_number="LOT-20250220-002",
quantity_received=15000,
quantity_accepted=14800,
defect_count=200,
defect_types="外観不良、寸法不良"
)
sqs.record_quality_data(
supplier_id="SUP-002",
delivery_date=datetime(2025, 1, 10),
lot_number="LOT-20250110-001",
quantity_received=5000,
quantity_accepted=4990,
defect_count=10,
defect_types="色ムラ"
)
# サプライヤー評価
print("=== サプライヤー評価 ===")
rating = sqs.calculate_supplier_rating("SUP-001", months=6)
for key, value in rating.items():
print(f"{key}: {value}")
# パフォーマンスサマリー
print("\n=== サプライヤーパフォーマンスサマリー ===")
print(sqs.get_supplier_performance_summary())
# 改善対象の特定
print("\n=== 改善対象サプライヤー ===")
targets = sqs.identify_improvement_targets(threshold_rating="B")
for target in targets:
print(f"{target['name']} - 評価: {target['rating']}, スコア: {target['quality_score']}")
print(f" → {target['action']}")4.3.3 品質目標KPI追跡システム
ISO 9001要求事項6.2「品質目標及びそれを達成するための計画策定」に対応します。
例7: 品質目標KPI管理システム
import pandas as pd
from datetime import datetime
from typing import List, Dict, Optional
import matplotlib.pyplot as plt
class QualityKPISystem:
"""品質目標とKPIの管理システム
品質目標の設定、KPIの追跡、達成状況の可視化を行う。
"""
def __init__(self, fiscal_year: int):
self.fiscal_year = fiscal_year
self.objectives = []
self.kpi_data = []
def define_objective(self, objective_id: str, title: str,
kpi_name: str, target_value: float,
unit: str, measurement_frequency: str,
responsible_dept: str) -> Dict:
"""品質目標を定義
Args:
objective_id: 目標ID
title: 目標タイトル
kpi_name: KPI名称
target_value: 目標値
unit: 単位
measurement_frequency: 測定頻度（"Daily", "Weekly", "Monthly"）
responsible_dept: 責任部門
Returns:
定義された目標情報
"""
objective = {
"objective_id": objective_id,
"title": title,
"kpi_name": kpi_name,
"target_value": target_value,
"unit": unit,
"measurement_frequency": measurement_frequency,
"responsible_dept": responsible_dept,
"fiscal_year": self.fiscal_year,
"status": "Active" # Active, Achieved, At Risk, Not Achieved
}
self.objectives.append(objective)
print(f"品質目標設定: {title} (目標値: {target_value}{unit})")
return objective
def record_kpi_data(self, objective_id: str,
measurement_date: datetime,
actual_value: float,
notes: str = "") -> Dict:
"""KPI実績データを記録
Args:
objective_id: 目標ID
measurement_date: 測定日
actual_value: 実績値
notes: 備考
Returns:
記録されたKPIデータ
"""
objective = self._find_objective(objective_id)
if not objective:
raise ValueError(f"目標 {objective_id} が見つかりません")
# 目標達成率を計算
achievement_rate = (actual_value / objective['target_value'] * 100)
kpi_record = {
"objective_id": objective_id,
"measurement_date": measurement_date,
"actual_value": actual_value,
"target_value": objective['target_value'],
"achievement_rate": achievement_rate,
"notes": notes
}
self.kpi_data.append(kpi_record)
return kpi_record
def get_kpi_trend(self, objective_id: str) -> pd.DataFrame:
"""KPIのトレンドデータを取得
Args:
objective_id: 目標ID
Returns:
トレンドデータのDataFrame
"""
records = [
r for r in self.kpi_data if r['objective_id'] == objective_id
]
if not records:
return pd.DataFrame()
df = pd.DataFrame(records)
df = df.sort_values('measurement_date')
return df
def evaluate_objective_status(self, objective_id: str) -> str:
"""目標の達成状況を評価
Args:
objective_id: 目標ID
Returns:
ステータス（"Achieved", "On Track", "At Risk", "Not Achieved"）
"""
df = self.get_kpi_trend(objective_id)
if df.empty:
return "No Data"
# 直近3回の平均達成率を評価
recent_achievement = df.tail(3)['achievement_rate'].mean()
if recent_achievement >= 100:
status = "Achieved"
elif recent_achievement >= 90:
status = "On Track"
elif recent_achievement >= 70:
status = "At Risk"
else:
status = "Not Achieved"
# 目標情報を更新
objective = self._find_objective(objective_id)
if objective:
objective['status'] = status
return status
def get_objectives_dashboard(self) -> pd.DataFrame:
"""品質目標ダッシュボードを取得
Returns:
全目標の達成状況を示すDataFrame
"""
if not self.objectives:
return pd.DataFrame()
# 各目標の状況を評価
for obj in self.objectives:
self.evaluate_objective_status(obj['objective_id'])
df = pd.DataFrame(self.objectives)
# 最新の実績値を追加
for idx, obj in df.iterrows():
trend_df = self.get_kpi_trend(obj['objective_id'])
if not trend_df.empty:
latest = trend_df.iloc[-1]
df.at[idx, 'latest_value'] = latest['actual_value']
df.at[idx, 'latest_achievement_rate'] = latest['achievement_rate']
else:
df.at[idx, 'latest_value'] = None
df.at[idx, 'latest_achievement_rate'] = None
return df[['objective_id', 'title', 'target_value', 'unit',
'latest_value', 'latest_achievement_rate', 'status',
'responsible_dept']]
def generate_kpi_report(self, objective_id: str, filename: str):
"""KPIレポートを生成（トレンドグラフ付き）
Args:
objective_id: 目標ID
filename: 出力ファイル名（PNG形式）
"""
df = self.get_kpi_trend(objective_id)
objective = self._find_objective(objective_id)
if df.empty or not objective:
print("データが不足しています")
return
# グラフ作成
fig, ax = plt.subplots(figsize=(10, 6))
# 実績値をプロット
ax.plot(df['measurement_date'], df['actual_value'],
marker='o', label='実績値', linewidth=2)
# 目標値のラインを追加
ax.axhline(y=objective['target_value'], color='r',
linestyle='--', label='目標値')
ax.set_xlabel('測定日')
ax.set_ylabel(f"{objective['kpi_name']} ({objective['unit']})")
ax.set_title(f"{objective['title']} - KPIトレンド")
ax.legend()
ax.grid(True, alpha=0.3)
plt.xticks(rotation=45)
plt.tight_layout()
plt.savefig(filename, dpi=300)
plt.close()
print(f"KPIレポートを {filename} に出力しました")
def _find_objective(self, objective_id: str) -> Optional[Dict]:
"""目標IDで目標を検索（内部用）"""
for obj in self.objectives:
if obj['objective_id'] == objective_id:
return obj
return None
# 使用例
kpi_system = QualityKPISystem(fiscal_year=2025)
# 品質目標を定義
kpi_system.define_objective(
objective_id="OBJ-2025-001",
title="顧客クレーム削減",
kpi_name="顧客クレーム件数",
target_value=10,
unit="件/月",
measurement_frequency="Monthly",
responsible_dept="品質保証部"
)
kpi_system.define_objective(
objective_id="OBJ-2025-002",
title="工程内不良率削減",
kpi_name="工程内不良率",
target_value=0.5,
unit="%",
measurement_frequency="Monthly",
responsible_dept="製造部"
)
# KPI実績を記録
for month in range(1, 7):
kpi_system.record_kpi_data(
objective_id="OBJ-2025-001",
measurement_date=datetime(2025, month, 1),
actual_value=15 - month, # 改善傾向
notes=f"{month}月実績"
)
kpi_system.record_kpi_data(
objective_id="OBJ-2025-002",
measurement_date=datetime(2025, month, 1),
actual_value=1.0 - (month * 0.08), # 改善傾向
notes=f"{month}月実績"
)
# ダッシュボード表示
print("=== 品質目標ダッシュボード ===")
print(kpi_system.get_objectives_dashboard())
# ステータス評価
print("\n=== 達成状況評価 ===")
for obj_id in ["OBJ-2025-001", "OBJ-2025-002"]:
status = kpi_system.evaluate_objective_status(obj_id)
print(f"{obj_id}: {status}")4.3.4 マネジメントレビュー報告書生成
ISO 9001要求事項9.3「マネジメントレビュー」に対応する報告書生成システムです。
例8: マネジメントレビュー報告書生成システム
from datetime import datetime
import pandas as pd
from typing import Dict, List
import json
class ManagementReviewSystem:
"""マネジメントレビュー報告書生成システム
ISO 9001要求事項9.3に基づき、QMSのパフォーマンスと
改善機会を経営層にレビューするための報告書を生成。
"""
def __init__(self, review_date: datetime, review_period: str):
"""
Args:
review_date: レビュー実施日
review_period: レビュー期間（例: "2024年度 下半期"）
"""
self.review_date = review_date
self.review_period = review_period
self.review_data = {
"basic_info": {
"review_date": review_date.strftime("%Y-%m-%d"),
"review_period": review_period,
"attendees": []
},
"inputs": {},
"decisions": [],
"action_items": []
}
def add_attendee(self, name: str, position: str):
"""出席者を追加
Args:
name: 氏名
position: 役職
"""
self.review_data["basic_info"]["attendees"].append({
"name": name,
"position": position
})
def add_customer_feedback(self, satisfaction_score: float,
complaint_count: int,
positive_feedback: str,
improvement_areas: str):
"""顧客フィードバックを追加（ISO 9001 9.3.2 a）
Args:
satisfaction_score: 顧客満足度スコア（0-100）
complaint_count: クレーム件数
positive_feedback: 肯定的フィードバック
improvement_areas: 改善が必要な領域
"""
self.review_data["inputs"]["customer_feedback"] = {
"satisfaction_score": satisfaction_score,
"complaint_count": complaint_count,
"positive_feedback": positive_feedback,
"improvement_areas": improvement_areas
}
def add_conformity_status(self, audit_results: str,
nc_count: int,
capa_completion_rate: float):
"""適合性とパフォーマンスを追加（ISO 9001 9.3.2 b, c）
Args:
audit_results: 監査結果サマリー
nc_count: 不適合件数
capa_completion_rate: CAPA完了率（%）
"""
self.review_data["inputs"]["conformity_status"] = {
"audit_results": audit_results,
"nc_count": nc_count,
"capa_completion_rate": capa_completion_rate
}
def add_process_performance(self, kpi_summary: Dict[str, float],
process_efficiency: float):
"""プロセスパフォーマンスを追加（ISO 9001 9.3.2 d）
Args:
kpi_summary: KPIサマリー（KPI名: 達成率）
process_efficiency: プロセス効率（%）
"""
self.review_data["inputs"]["process_performance"] = {
"kpi_summary": kpi_summary,
"process_efficiency": process_efficiency
}
def add_resource_adequacy(self, resource_status: str,
training_completion: float,
infrastructure_issues: str):
"""資源の妥当性を追加（ISO 9001 9.3.2 e）
Args:
resource_status: 資源状況の評価
training_completion: 教育訓練完了率（%）
infrastructure_issues: インフラストラクチャの課題
"""
self.review_data["inputs"]["resource_adequacy"] = {
"resource_status": resource_status,
"training_completion": training_completion,
"infrastructure_issues": infrastructure_issues
}
def add_improvement_actions(self, actions_planned: int,
actions_completed: int,
effectiveness_rate: float):
"""改善処置の有効性を追加（ISO 9001 9.3.2 f）
Args:
actions_planned: 計画された改善処置数
actions_completed: 完了した改善処置数
effectiveness_rate: 有効性確認率（%）
"""
self.review_data["inputs"]["improvement_actions"] = {
"actions_planned": actions_planned,
"actions_completed": actions_completed,
"completion_rate": (actions_completed / actions_planned * 100)
if actions_planned > 0 else 0,
"effectiveness_rate": effectiveness_rate
}
def add_changes_impact(self, internal_changes: str,
external_changes: str,
impact_assessment: str):
"""変更の影響を追加（ISO 9001 9.3.2 g）
Args:
internal_changes: 内部変更
external_changes: 外部変更
impact_assessment: 影響評価
"""
self.review_data["inputs"]["changes_impact"] = {
"internal_changes": internal_changes,
"external_changes": external_changes,
"impact_assessment": impact_assessment
}
def add_decision(self, decision_type: str, decision: str,
responsible: str, due_date: str):
"""経営層の決定事項を追加（ISO 9001 9.3.3）
Args:
decision_type: "QMS Improvement", "Product/Service Improvement",
"Resource Needs"
decision: 決定内容
responsible: 責任者
due_date: 期限
"""
self.review_data["decisions"].append({
"type": decision_type,
"decision": decision,
"responsible": responsible,
"due_date": due_date
})
def add_action_item(self, action: str, responsible: str,
due_date: str, priority: str):
"""アクションアイテムを追加
Args:
action: アクション内容
responsible: 責任者
due_date: 期限
priority: 優先度（"High", "Medium", "Low"）
"""
self.review_data["action_items"].append({
"action": action,
"responsible": responsible,
"due_date": due_date,
"priority": priority,
"status": "Open"
})
def generate_report(self, filename: str):
"""マネジメントレビュー報告書を生成
Args:
filename: 出力ファイル名（JSON形式）
"""
# 総合評価を追加
self.review_data["overall_assessment"] = self._calculate_overall_assessment()
with open(filename, 'w', encoding='utf-8') as f:
json.dump(self.review_data, f, indent=2, ensure_ascii=False)
print(f"マネジメントレビュー報告書を {filename} に出力しました")
# サマリーを表示
self._print_summary()
def _calculate_overall_assessment(self) -> Dict:
"""総合評価を計算（内部用）"""
inputs = self.review_data["inputs"]
# スコア計算
scores = []
if "customer_feedback" in inputs:
scores.append(inputs["customer_feedback"]["satisfaction_score"])
if "conformity_status" in inputs:
scores.append(inputs["conformity_status"]["capa_completion_rate"])
if "process_performance" in inputs:
scores.append(inputs["process_performance"]["process_efficiency"])
if "resource_adequacy" in inputs:
scores.append(inputs["resource_adequacy"]["training_completion"])
if "improvement_actions" in inputs:
scores.append(inputs["improvement_actions"]["completion_rate"])
avg_score = sum(scores) / len(scores) if scores else 0
# 評価ランク
if avg_score >= 90:
rank = "Excellent"
elif avg_score >= 80:
rank = "Good"
elif avg_score >= 70:
rank = "Acceptable"
else:
rank = "Needs Improvement"
return {
"average_score": round(avg_score, 2),
"rank": rank,
"total_decisions": len(self.review_data["decisions"]),
"total_actions": len(self.review_data["action_items"])
}
def _print_summary(self):
"""サマリーを表示（内部用）"""
assessment = self.review_data["overall_assessment"]
print("\n" + "="*60)
print("マネジメントレビュー サマリー")
print("="*60)
print(f"レビュー期間: {self.review_period}")
print(f"実施日: {self.review_date.strftime('%Y年%m月%d日')}")
print(f"出席者: {len(self.review_data['basic_info']['attendees'])}名")
print(f"\n総合評価: {assessment['rank']} ({assessment['average_score']}点)")
print(f"決定事項: {assessment['total_decisions']}件")
print(f"アクションアイテム: {assessment['total_actions']}件")
print("="*60)
# 使用例
mr_system = ManagementReviewSystem(
review_date=datetime(2025, 3, 15),
review_period="2024年度 下半期"
)
# 出席者を追加
mr_system.add_attendee("山田太郎", "代表取締役社長")
mr_system.add_attendee("佐藤花子", "品質管理部長")
mr_system.add_attendee("鈴木一郎", "製造部長")
# インプット情報を追加
mr_system.add_customer_feedback(
satisfaction_score=85.5,
complaint_count=12,
positive_feedback="納期遵守率が改善。製品品質が安定。",
improvement_areas="技術サポート対応の迅速化"
)
mr_system.add_conformity_status(
audit_results="内部監査5回実施。Major NC 0件、Minor NC 3件。",
nc_count=3,
capa_completion_rate=92.0
)
mr_system.add_process_performance(
kpi_summary={
"顧客クレーム削減": 95.0,
"工程内不良率削減": 88.0,
"納期遵守率": 98.0
},
process_efficiency=91.5
)
mr_system.add_resource_adequacy(
resource_status="概ね適切。一部設備の老朽化が課題。",
training_completion=87.0,
infrastructure_issues="測定機器の校正管理システム更新が必要"
)
mr_system.add_improvement_actions(
actions_planned=15,
actions_completed=13,
effectiveness_rate=85.0
)
mr_system.add_changes_impact(
internal_changes="新製造ラインの導入、品質管理システムのデジタル化",
external_changes="ISO 9001:2015改訂版への移行準備",
impact_assessment="ポジティブな影響。追加リソースが必要。"
)
# 決定事項を追加
mr_system.add_decision(
decision_type="QMS Improvement",
decision="品質管理システムの完全デジタル化を2025年度第2四半期までに完了",
responsible="品質管理部長",
due_date="2025-09-30"
)
mr_system.add_decision(
decision_type="Resource Needs",
decision="測定機器の更新予算を確保（500万円）",
responsible="経営企画部長",
due_date="2025-06-30"
)
# アクションアイテムを追加
mr_system.add_action_item(
action="技術サポート対応時間を24時間以内に短縮するプロセスを構築",
responsible="営業部長",
due_date="2025-05-31",
priority="High"
)
mr_system.add_action_item(
action="内部監査員の力量向上研修を実施",
responsible="品質管理部長",
due_date="2025-04-30",
priority="Medium"
)
# 報告書を生成
mr_system.generate_report("management_review_2024H2.json")まとめ
この章では、ISO 9001品質マネジメントシステムの要求事項と、Pythonを使った実践的な実装方法を学びました。
📚 重要ポイント
- ISO 9001の7原則 : 顧客重視、リーダーシップ、人々の参加、プロセスアプローチ、改善、客観的事実、関係性管理
- 文書管理 : バージョン管理、承認ワークフロー、定期見直し
- 内部監査 : 計画的な実施、発見事項の追跡、是正処置
- CAPA : 根本原因分析、是正処置、有効性確認
- リスクベース思考 : FMEAによるリスク評価とRPN計算
- サプライヤー管理 : 品質評価、PPM管理、改善活動
- KPI管理 : 品質目標の設定、追跡、ダッシュボード化
- マネジメントレビュー : QMSパフォーマンスの定期的評価
← 第3章：統計的品質管理（SQC） 第5章：Pythonによる品質管理の自動化 →
免責事項
- 本コンテンツは教育・研究・情報提供のみを目的としており、専門的な助言(法律・会計・技術的保証など)を提供するものではありません。
- 本コンテンツおよび付随するCode examplesは「現状有姿(AS IS)」で提供され、明示または黙示を問わず、商品性、特定目的適合性、権利非侵害、正確性・完全性、動作・安全性等いかなる保証もしません。
- 外部リンク、第三者が提供するデータ・ツール・ライブラリ等の内容・可用性・安全性について、作成者および東北大学は一切の責任を負いません。
- 本コンテンツの利用・実行・解釈により直接的・間接的・付随的・特別・結果的・懲罰的損害が生じた場合でも、適用法で許容される最大限の範囲で、作成者および東北大学は責任を負いません。
- 本コンテンツの内容は、予告なく変更・更新・提供停止されることがあります。
- 本コンテンツの著作権・ライセンスは明記された条件(例: CC BY 4.0)に従います。当該ライセンスは通常、無保証条項を含みます。