# ========================================================= # POSCO DX - MRO Composite AI - PROCESS GUIDE ENHANCED # 업무 프로세스 가이드 통합 버전 - Hugging Face Spaces 배포용 # ========================================================= import os import json import time import random import traceback from dataclasses import dataclass from typing import Dict, Any, List, Optional, Tuple, TypedDict from datetime import datetime, timedelta import numpy as np import pandas as pd import networkx as nx # ✅ Plotly imports import plotly import plotly.graph_objects as go import plotly.express as px from plotly.subplots import make_subplots print(f"✅ NumPy: {np.__version__}") print(f"✅ Pandas: {pd.__version__}") print(f"✅ Plotly: {plotly.__version__}") try: from pulp import LpProblem, LpMinimize, LpVariable, lpSum, LpStatus PULP_AVAILABLE = True print("✅ PuLP available") except ImportError: print("⚠️ PuLP not available") PULP_AVAILABLE = False import gradio as gr print(f"✅ Gradio: {gr.__version__}") try: from langgraph.graph import StateGraph, END LANGGRAPH_AVAILABLE = True print("✅ LangGraph available") except ImportError: print("⚠️ LangGraph not available") LANGGRAPH_AVAILABLE = False try: from openai import OpenAI OPENAI_AVAILABLE = True print("✅ OpenAI available") except ImportError: print("⚠️ OpenAI not available") OPENAI_AVAILABLE = False # ========================================================= # API Key Configuration for Hugging Face Spaces # ========================================================= # Hugging Face Spaces에서 환경 변수로 API 키 로드 OPENAI_API_KEY = os.environ.get('OPENAI_API_KEY', '').strip() if OPENAI_API_KEY: os.environ['OPENAI_API_KEY'] = OPENAI_API_KEY print("✅ OpenAI API Key loaded from environment") else: print("⚠️ DEMO MODE - No API Key found") print("💡 To use OpenAI features, add OPENAI_API_KEY to your Hugging Face Space Secrets") print("\n" + "=" * 60) print("✅ 프로세스 가이드 통합 버전 초기화 완료!") print("=" * 60 + "\n") # ========================================================= # Process Guide Configuration # ========================================================= PROCESS_WORKFLOWS = { "mro": { "title": "🔧 MRO 운영 프로세스", "steps": [ { "id": "1", "name": "고장/정비 요청 접수", "description": "설비 고장 또는 예방정비 요청을 접수합니다", "input": "설비 ID, 고장 유형, 우선순위", "output": "요청 번호, 설비 상세정보", "owner": "현장 담당자 → MRO팀", "duration": "5분" }, { "id": "2", "name": "설비 정보 조회", "description": "Knowledge Graph에서 설비 상세 정보를 조회합니다", "input": "설비 ID", "output": "설비명, 위치, 중요도, 정비이력", "owner": "MRO팀 (AI 자동)", "duration": "1분" }, { "id": "3", "name": "호환 부품 자동 매칭", "description": "설비와 호환되는 모든 부품을 자동으로 조회합니다", "input": "설비 ID, 설비 타입", "output": "호환 부품 리스트, 필수/선택 구분", "owner": "MRO팀 (AI 자동)", "duration": "2분" }, { "id": "4", "name": "전사 재고 현황 확인", "description": "본사 및 각 제철소의 재고 현황을 실시간 확인합니다", "input": "품목 ID", "output": "창고별 재고량, 안전재고, 예약수량", "owner": "MRO팀 (AI 자동)", "duration": "1분" }, { "id": "5", "name": "발주 필요성 판단", "description": "재고 부족 시 발주 요청을 생성합니다", "input": "현재고, 안전재고, 수요량", "output": "발주 필요 여부, 발주 수량", "owner": "MRO팀", "duration": "3분" }, { "id": "6", "name": "구매팀 발주 요청", "description": "구매팀에 발주 요청서를 전달합니다", "input": "품목 정보, 수량, 납기 요구사항", "output": "발주 요청 번호", "owner": "MRO팀 → 구매팀", "duration": "2분" } ], "total_duration": "약 15분", "success_criteria": [ "✓ 설비 정보 정확히 식별", "✓ 호환 부품 100% 매칭", "✓ 재고 현황 실시간 반영", "✓ 발주 수량 최적화" ] }, "procurement": { "title": "💰 구매/조달 프로세스", "steps": [ { "id": "1", "name": "발주 요청 접수", "description": "MRO팀으로부터 발주 요청을 접수합니다", "input": "발주 요청서, 품목, 수량, 납기", "output": "구매 작업 번호", "owner": "구매팀", "duration": "3분" }, { "id": "2", "name": "공급업체 정보 조회", "description": "품목별 등록된 모든 공급업체를 조회합니다", "input": "품목 ID", "output": "공급업체 리스트, 단가, 납기, ESG등급", "owner": "구매팀 (AI 자동)", "duration": "2분" }, { "id": "3", "name": "규정 준수 검증", "description": "Neuro-Symbolic AI로 구매 규정을 자동 검증합니다", "input": "품목 속성, 공급업체 정보", "output": "규정 위반 여부, 차단/경고 리스트", "owner": "구매팀 (AI 자동)", "duration": "1분" }, { "id": "4", "name": "최적 배분 계산", "description": "Linear Programming으로 최적 발주 계획을 수립합니다", "input": "공급업체 오퍼, 수요량, 제약조건", "output": "업체별 발주량, 총 비용, 예상 납기", "owner": "구매팀 (AI 자동)", "duration": "2분" }, { "id": "5", "name": "발주 전략 수립", "description": "LLM이 최적화 결과를 바탕으로 구매 전략을 제안합니다", "input": "최적화 결과, 시장 상황", "output": "발주 전략, 리스크 분석, 대안", "owner": "구매팀 (AI 지원)", "duration": "5분" }, { "id": "6", "name": "경영진 승인 요청", "description": "발주 계획을 경영진에게 승인 요청합니다", "input": "발주 계획서, 비용 분석", "output": "승인 요청 번호", "owner": "구매팀 → 경영진", "duration": "3분" }, { "id": "7", "name": "PO 발행 (승인 후)", "description": "승인 후 공급업체에 정식 발주서를 발행합니다", "input": "승인된 발주 계획", "output": "PO 번호, 계약서", "owner": "구매팀", "duration": "10분" } ], "total_duration": "약 25분 (승인 대기 제외)", "success_criteria": [ "✓ 규정 100% 준수", "✓ 비용 최적화 달성", "✓ 납기 요구사항 충족", "✓ ESG 등급 기준 만족" ] }, "executive": { "title": "👔 경영진 의사결정 프로세스", "steps": [ { "id": "1", "name": "승인 요청 알림", "description": "발주 승인 요청 알림을 수신합니다", "input": "승인 요청 번호, 요약 정보", "output": "알림 확인", "owner": "시스템 → 경영진", "duration": "즉시" }, { "id": "2", "name": "KPI 대시보드 확인", "description": "실시간 KPI 대시보드를 통해 전반적 현황을 파악합니다", "input": "없음", "output": "비용절감률, 컴플라이언스, ESG점수 등", "owner": "경영진", "duration": "2분" }, { "id": "3", "name": "Action Items 검토", "description": "우선순위별 조치 항목을 검토합니다", "input": "Action Items 리스트", "output": "검토 의견", "owner": "경영진", "duration": "5분" }, { "id": "4", "name": "발주 상세 분석", "description": "발주 계획의 타당성을 면밀히 검토합니다", "input": "발주 계획서, 최적화 결과, 규정 검증", "output": "분석 의견", "owner": "경영진", "duration": "10분" }, { "id": "5", "name": "의사결정", "description": "승인/반려/조건부승인을 결정합니다", "input": "검토 결과", "output": "승인 결정, 피드백", "owner": "경영진", "duration": "3분" }, { "id": "6", "name": "피드백 제공", "description": "개선 제안 또는 지시사항을 전달합니다", "input": "의사결정 근거", "output": "피드백 메시지, 개선 방향", "owner": "경영진 → 구매팀", "duration": "5분" } ], "total_duration": "약 25분", "success_criteria": [ "✓ 전략적 타당성 검증", "✓ 리스크 수용 가능 수준", "✓ 예산 범위 내 집행", "✓ 장기 목표 부합" ] } } # ========================================================= # Enhanced Configuration with Real Part Names # ========================================================= SCENARIO_PRESETS = { "긴급 고장 대응": { "description": "🚨 포항제철소 컨베이어 베어링 긴급 고장", "equipment_id": "CONV-PH-007", "item_id": "", "demand_qty": 10, "context": "컨베이어 베어링 고장으로 생산라인 중단. 즉시 교체 필요.", "priority": "긴급", "guide": "리드타임 최소화 우선. 국내 공급업체 우선 고려." }, "정기 발주 계획": { "description": "📋 월간 정기 발주 - 유압펌프 예방정비", "equipment_id": "PUMP-GY-003", "item_id": "SEAL-A45", "demand_qty": 50, "context": "월간 예방정비 계획. 최적 가격 및 재고 균형 필요.", "priority": "정상", "guide": "비용 최적화 우선. ESG 등급 고려." }, "규정 준수 검증": { "description": "⚖️ 규제품목(특수화학물질) 구매 검증", "equipment_id": "VALVE-PH-005", "item_id": "", "demand_qty": 20, "context": "특수 실링재 구매. 해외구매 차단 규정 준수 필수.", "priority": "규정준수", "guide": "컴플라이언스 100% 준수. 국내업체만 허용." } } # Real part names and categories REAL_PART_NAMES = { "베어링": ["SKF 6205 볼베어링", "NSK 원통베어링", "NTN 테이퍼베어링"], "윤활유": ["쉘 오마라 220", "모빌 DTE 25", "지에스칼텍스 터빈유"], "필터": ["하이드로락 유압필터", "파커 에어필터", "도난드슨 정밀필터"], "벨트": ["게이츠 파워그립 벨트", "반도 V벨트", "옵티벨트 타이밍벨트"], "센서": ["지멘스 근접센서", "오므론 광전센서", "하니웰 압력센서"], "패킹": ["NOK 오링", "파커 유압씰", "발카 그랜드패킹"], "퓨즈": ["LS산전 MCCB", "슈나이더 차단기", "ABB 퓨즈"], "호스": ["파커 유압호스", "만리 고압호스", "브리지스톤 산업호스"], "볼트": ["SUS304 육각볼트", "고장력볼트 F10T", "앵커볼트 M16"], "실링재": ["록타이트 실란트", "쓰리본드 액상패킹", "헨켈 밀봉재"] } # Enhanced supplier info REAL_SUPPLIERS = [ {"name": "포스코케미칼", "type": "국내", "esg": "A", "specialty": "화학/윤활유"}, {"name": "효성중공업", "type": "국내", "esg": "A", "specialty": "베어링/기계"}, {"name": "LS산전", "type": "국내", "esg": "B", "specialty": "전기/센서"}, {"name": "삼화콘덴서", "type": "국내", "esg": "B", "specialty": "전기부품"}, {"name": "태광산업", "type": "국내", "esg": "C", "specialty": "호스/패킹"}, {"name": "한국파커", "type": "국내", "esg": "A", "specialty": "유압부품"}, {"name": "그라코(Graco)", "type": "해외", "esg": "B", "specialty": "유압장비"}, {"name": "에머슨(Emerson)", "type": "해외", "esg": "C", "specialty": "밸브/센서"} ] # ========================================================= # Utility Functions # ========================================================= def now_ts() -> str: return time.strftime("%Y-%m-%d %H:%M:%S") def safe_json(obj: Any) -> str: try: return json.dumps(obj, ensure_ascii=False, indent=2) except Exception: return str(obj) def format_status(status_dict: Dict[str, Any]) -> str: lines = [ "=" * 60, "📊 시스템 실행 상태", "=" * 60, "", f"🔌 연결: {status_dict.get('mode', 'Unknown')}", f"🎯 시나리오: {status_dict.get('scenario', 'N/A')}", f"⚙️ 설비: {status_dict.get('equipment', 'N/A')}", f"📦 품목: {status_dict.get('item_name', 'N/A')}", f"📊 수요: {status_dict.get('demand', 'N/A')}개", f"🚨 우선순위: {status_dict.get('priority', 'N/A')}", f"\n✅ 데이터 검증: {'통과' if status_dict.get('tables_ok') else '실패'}", f"⏱️ 진행: {status_dict.get('progress', 'N/A')}", "\n" + "=" * 60 ] return "\n".join(lines) def create_process_guide_html(process_key: str) -> str: """업무 프로세스 가이드를 HTML로 생성""" workflow = PROCESS_WORKFLOWS.get(process_key, {}) if not workflow: return "

프로세스 정보가 없습니다.

" html = f"""

{workflow['title']}

총 소요시간: {workflow['total_duration']}

""" for step in workflow['steps']: html += f"""
{step['id']}

{step['name']}

⏱️ {step['duration']}

{step['description']}

📥 입력:
{step['input']}
📤 출력:
{step['output']}
👤 담당: {step['owner']}
""" html += """

✅ 성공 기준

""" return html # ========================================================= # Enhanced Data Generator with Real Names # ========================================================= def generate_demo_tables(seed: int = 7) -> Dict[str, pd.DataFrame]: """Generate realistic demo data""" random.seed(seed) np.random.seed(seed) plants = pd.DataFrame([ {"plant_id": "PH", "plant_name": "포항제철소", "region": "경북", "capacity": 1000}, {"plant_id": "GY", "plant_name": "광양제철소", "region": "전남", "capacity": 1200}, {"plant_id": "HQ", "plant_name": "본사", "region": "서울", "capacity": 0}, ]) # Equipment with real names equipment = [] eq_configs = [ ("PUMP", "유압펌프", ["PH", "GY"], 6), ("CONV", "컨베이어", ["PH", "GY"], 4), ("VALVE", "제어밸브", ["PH", "GY"], 3), ("MOTOR", "구동모터", ["PH", "GY"], 5), ] eq_id = 1 for eq_type, eq_name_kr, plants_list, count in eq_configs: for plant in plants_list: for i in range(1, count + 1): equipment.append({ "equipment_id": f"{eq_type}-{plant}-{eq_id:03d}", "equipment_name": f"{eq_name_kr}-{plant}-{i}호기", "plant_id": plant, "equipment_type": eq_name_kr, "criticality": random.choice(["긴급", "긴급", "중요", "보통"]), "status": "가동중", "last_maintenance": (datetime.now() - timedelta(days=random.randint(30, 180))).strftime("%Y-%m-%d"), }) eq_id += 1 equipment = pd.DataFrame(equipment) # Items with real part names items = [] item_id = 1 for category, part_list in REAL_PART_NAMES.items(): for part_name in part_list: items.append({ "item_id": f"{category[:3].upper()}-{chr(65 + (item_id % 3))}{item_id:02d}", "item_name": part_name, "category": category, "uom": "EA", "risk_class": "규제" if "화학" in part_name or "특수" in category else "일반", "unit_weight": round(0.5 + random.random() * 5, 1), "shelf_life_days": random.choice([365, 730, 1095, None]), }) item_id += 1 items = pd.DataFrame(items) # Compatibility compat = [] for eq_idx, eq_row in equipment.iterrows(): eq_id = eq_row["equipment_id"] eq_type = eq_row["equipment_type"] # Match parts to equipment type if "펌프" in eq_type: relevant_cats = ["베어링", "윤활유", "패킹"] elif "컨베이어" in eq_type: relevant_cats = ["베어링", "벨트", "센서"] elif "밸브" in eq_type: relevant_cats = ["실링재", "패킹", "윤활유"] else: relevant_cats = ["베어링", "센서", "필터"] for cat in relevant_cats: cat_items = items[items["category"] == cat] if len(cat_items) > 0: selected = cat_items.sample(min(2, len(cat_items))) for _, item in selected.iterrows(): compat.append({ "equipment_id": eq_id, "item_id": item["item_id"], "is_mandatory": (cat == relevant_cats[0]), "annual_consumption_est": random.randint(20, 200), "failure_rate": round(random.random() * 0.05, 3), }) compat = pd.DataFrame(compat).drop_duplicates(["equipment_id", "item_id"]) # Storages storages = pd.DataFrame([ {"storage_id": "WH-HQ", "plant_id": "HQ", "storage_name": "본사 중앙창고", "capacity": 10000}, {"storage_id": "WH-PH", "plant_id": "PH", "storage_name": "포항 MRO창고", "capacity": 5000}, {"storage_id": "WH-GY", "plant_id": "GY", "storage_name": "광양 MRO창고", "capacity": 5000}, ]) # Inventory with realistic levels inventory = [] for st_idx, st_row in storages.iterrows(): sampled_items = items.sample(min(25, len(items))) for _, item in sampled_items.iterrows(): stock_level = random.randint(10, 100) safety_stock = int(stock_level * 0.2) inventory.append({ "storage_id": st_row["storage_id"], "item_id": item["item_id"], "on_hand": stock_level, "safety_stock": safety_stock, "reserved": random.randint(0, min(5, stock_level)), "last_updated": (datetime.now() - timedelta(days=random.randint(1, 30))).strftime("%Y-%m-%d"), }) inventory = pd.DataFrame(inventory) # Suppliers with real names suppliers = pd.DataFrame([ { "supplier_id": f"SUP-{i:03d}", "supplier_name": sup["name"], "supplier_type": sup["type"], "rating": round(3.5 + random.random() * 1.5, 1), "esg_level": sup["esg"], "specialty": sup["specialty"], "region": sup["type"], "payment_terms": random.choice(["NET30", "NET45", "NET60"]), "established_year": random.randint(1990, 2020), } for i, sup in enumerate(REAL_SUPPLIERS, 1) ]) # Supplier offers with realistic pricing offers = [] for _, item in items.iterrows(): num_suppliers = random.randint(3, 4) selected_sups = suppliers.sample(min(num_suppliers, len(suppliers))) base_price = 10000 + random.randint(0, 90000) for rank, (_, sup) in enumerate(selected_sups.iterrows()): price_multiplier = 1.0 + (rank * 0.05) + random.uniform(-0.1, 0.1) offers.append({ "item_id": item["item_id"], "supplier_id": sup["supplier_id"], "unit_price": int(base_price * price_multiplier), "lead_time_days": 3 + rank * 2 + random.randint(0, 5), "moq": [10, 20, 50, 100][rank % 4], "contract_type": random.choice(["단가계약", "장기계약", "스팟"]), "discount_rate": round(random.random() * 0.1, 2) if rank == 0 else 0, "quality_grade": random.choice(["A", "A", "B", "C"]), }) supplier_offers = pd.DataFrame(offers) # Policies policies = pd.DataFrame([ { "policy_id": "R-001", "rule_name": "규제품목 해외구매 제한", "rule_logic": "IF item.risk_class == '규제' AND supplier.region == '해외' THEN block", "severity": "차단", "department": "법무팀" }, { "policy_id": "R-002", "rule_name": "안전재고 미만 긴급발주", "rule_logic": "IF (on_hand - reserved) < safety_stock THEN expedite", "severity": "경고", "department": "MRO팀" }, { "policy_id": "R-003", "rule_name": "긴급설비 우선배분", "rule_logic": "IF equipment.criticality == '긴급' THEN priority", "severity": "우선순위", "department": "생산팀" }, { "policy_id": "R-004", "rule_name": "ESG C등급 제한", "rule_logic": "IF supplier.esg_level == 'C' THEN penalize", "severity": "패널티", "department": "구매팀" }, ]) # Purchase history purchase_history = [] for i in range(200): item = items.sample(1).iloc[0] supplier = suppliers.sample(1).iloc[0] qty = random.randint(10, 100) price = random.randint(10000, 100000) purchase_history.append({ "po_id": f"PO-2024-{10000 + i}", "date": (datetime.now() - timedelta(days=random.randint(1, 365))).strftime("%Y-%m-%d"), "item_id": item["item_id"], "supplier_id": supplier["supplier_id"], "qty": qty, "unit_price": price, "total_amount": qty * price, "delivery_status": random.choice(["완료", "완료", "완료", "지연", "진행중"]), }) purchase_history = pd.DataFrame(purchase_history) return { "plants": plants, "equipment": equipment, "items": items, "compat": compat, "storages": storages, "inventory": inventory, "suppliers": suppliers, "supplier_offers": supplier_offers, "policies": policies, "purchase_history": purchase_history } def validate_tables(tables: Dict[str, pd.DataFrame]) -> Tuple[bool, List[str]]: """Validate tables""" required = ["plants", "equipment", "items", "compat", "storages", "inventory", "suppliers", "supplier_offers", "policies", "purchase_history"] issues = [] for k in required: if k not in tables: issues.append(f"Missing: {k}") elif not isinstance(tables[k], pd.DataFrame): issues.append(f"Invalid type: {k}") elif len(tables[k]) == 0: issues.append(f"Empty: {k}") return len(issues) == 0, issues # ========================================================= # Plotly Dashboard Functions # ========================================================= def create_mro_inventory_dashboard(inv_df: pd.DataFrame, item_name: str) -> go.Figure: """MRO 재고 현황 대시보드""" if len(inv_df) == 0: fig = go.Figure() fig.add_annotation(text="재고 데이터 없음", showarrow=False, font_size=20) fig.update_layout(height=700, title_text="재고 정보 없음") return fig # 서브플롯 생성 fig = make_subplots( rows=2, cols=2, subplot_titles=('창고별 재고 현황', '안전재고 대비 현재고', '재고 상태', '창고별 점유율'), specs=[[{"type": "bar"}, {"type": "indicator"}], [{"type": "pie"}, {"type": "table"}]] ) # 1. 창고별 재고 바 차트 fig.add_trace( go.Bar( x=inv_df['storage_name'], y=inv_df['on_hand'], name='현재고', marker_color='lightblue', text=inv_df['on_hand'], textposition='auto', ), row=1, col=1 ) fig.add_trace( go.Bar( x=inv_df['storage_name'], y=inv_df['safety_stock'], name='안전재고', marker_color='orange', text=inv_df['safety_stock'], textposition='auto', ), row=1, col=1 ) # 2. 총 재고 게이지 total_stock = inv_df['on_hand'].sum() total_safety = inv_df['safety_stock'].sum() fig.add_trace( go.Indicator( mode="gauge+number+delta", value=total_stock, delta={'reference': total_safety, 'increasing': {'color': "green"}}, title={'text': f"총 재고량
{item_name}"}, gauge={ 'axis': {'range': [0, total_safety * 2]}, 'bar': {'color': "darkblue"}, 'steps': [ {'range': [0, total_safety], 'color': "lightgray"}, {'range': [total_safety, total_safety * 1.5], 'color': "lightgreen"} ], 'threshold': { 'line': {'color': "red", 'width': 4}, 'thickness': 0.75, 'value': total_safety } } ), row=1, col=2 ) # 3. 재고 상태 파이 차트 inv_df['available'] = inv_df['on_hand'] - inv_df['reserved'] fig.add_trace( go.Pie( labels=['가용재고', '예약됨', '안전재고'], values=[ inv_df['available'].sum(), inv_df['reserved'].sum(), max(0, total_safety - inv_df['available'].sum()) ], marker_colors=['green', 'orange', 'red'], hole=0.3, ), row=2, col=1 ) # 4. 상세 테이블 fig.add_trace( go.Table( header=dict( values=['창고', '현재고', '안전재고', '예약', '가용'], fill_color='paleturquoise', align='left' ), cells=dict( values=[ inv_df['storage_name'], inv_df['on_hand'], inv_df['safety_stock'], inv_df['reserved'], inv_df['available'] ], fill_color='lavender', align='left' ) ), row=2, col=2 ) fig.update_layout( height=700, showlegend=True, title_text=f"📦 MRO 재고 분석 대시보드 - {item_name}", title_font_size=20 ) return fig def create_mro_workflow_status(equipment_info: Dict, compat_items: pd.DataFrame) -> go.Figure: """MRO 워크플로우 상태 시각화""" fig = go.Figure() # 워크플로우 단계 steps = [ "설비 확인", "호환부품 조회", "재고 확인", "수요 검증", "발주 요청" ] statuses = ["완료", "완료", "진행중", "대기", "대기"] colors = ["green", "green", "orange", "gray", "gray"] # Funnel 차트로 워크플로우 표현 fig.add_trace(go.Funnel( y=steps, x=[100, 80, 60, 40, 20], textposition="inside", textinfo="label+percent initial", marker={"color": colors}, connector={"line": {"color": "royalblue", "width": 3}} )) equipment_name = equipment_info.get('equipment_name', 'N/A') if equipment_info else 'N/A' fig.update_layout( title_text=f"🔄 MRO 워크플로우 진행 상태
설비: {equipment_name}", height=400, showlegend=False ) return fig def create_procurement_comparison_dashboard(offers_df: pd.DataFrame, rules_eval: Dict) -> go.Figure: """구매 담당자 - 공급업체 비교 대시보드""" if len(offers_df) == 0: fig = go.Figure() fig.add_annotation(text="공급업체 데이터 없음", showarrow=False, font_size=20) fig.update_layout(height=700, title_text="공급업체 정보 없음") return fig # 서브플롯 fig = make_subplots( rows=2, cols=2, subplot_titles=( '💰 가격 비교', '⏱️ 납기 비교', '📊 ESG 등급 분포', '🎯 종합 평가' ), specs=[ [{"type": "bar"}, {"type": "scatter"}], [{"type": "pie"}, {"type": "table"}] ] ) # 규칙 평가 결과 추가 offers_df['blocked'] = offers_df['supplier_id'].apply( lambda x: rules_eval.get(x, {}).get('block', False) ) offers_df['color'] = offers_df['blocked'].apply(lambda x: 'red' if x else 'green') # 1. 가격 비교 바 차트 fig.add_trace( go.Bar( x=offers_df['supplier_name'], y=offers_df['unit_price'], marker_color=offers_df['color'], text=[f"{p:,}원" for p in offers_df['unit_price']], textposition='auto', name='단가', ), row=1, col=1 ) # 2. 가격-납기 스캐터 fig.add_trace( go.Scatter( x=offers_df['lead_time_days'], y=offers_df['unit_price'], mode='markers+text', marker=dict( size=15, color=offers_df['color'], line=dict(width=2, color='white') ), text=offers_df['supplier_name'], textposition="top center", name='공급업체', ), row=1, col=2 ) # 3. ESG 등급 파이 esg_counts = offers_df['esg_level'].value_counts() fig.add_trace( go.Pie( labels=esg_counts.index, values=esg_counts.values, marker_colors=['lightgreen', 'lightyellow', 'lightcoral'], hole=0.3, ), row=2, col=1 ) # 4. 종합 평가 테이블 evaluation = offers_df.copy() evaluation['종합점수'] = ( (100 - (evaluation['unit_price'] / evaluation['unit_price'].max() * 50)) + (100 - (evaluation['lead_time_days'] / evaluation['lead_time_days'].max() * 30)) + evaluation['esg_level'].map({'A': 20, 'B': 10, 'C': 0}) ).round(1) evaluation['순위'] = evaluation['종합점수'].rank(ascending=False).astype(int) fig.add_trace( go.Table( header=dict( values=['순위', '공급업체', '단가', '납기', 'ESG', '점수'], fill_color='paleturquoise', align='center' ), cells=dict( values=[ evaluation['순위'], evaluation['supplier_name'], [f"{p:,}" for p in evaluation['unit_price']], [f"{d}일" for d in evaluation['lead_time_days']], evaluation['esg_level'], evaluation['종합점수'] ], fill_color=[['white' if not b else 'lightcoral' for b in evaluation['blocked']]], align='center' ) ), row=2, col=2 ) fig.update_layout( height=700, showlegend=False, title_text="📊 공급업체 종합 비교 대시보드", title_font_size=20 ) fig.update_xaxes(title_text="납기 (일)", row=1, col=2) fig.update_yaxes(title_text="단가 (원)", row=1, col=2) return fig def create_procurement_workflow(opt_result: Dict) -> go.Figure: """구매 워크플로우 진행 상태""" fig = go.Figure() # 워크플로우 단계와 상태 workflow_steps = [ {"step": "1. 수요 접수", "status": "완료", "time": "10분"}, {"step": "2. 공급업체 조회", "status": "완료", "time": "5분"}, {"step": "3. 규정 검증", "status": "완료", "time": "2분"}, {"step": "4. 최적화 분석", "status": "완료", "time": "3분"}, {"step": "5. 발주 승인", "status": "대기중", "time": "-"}, {"step": "6. PO 발행", "status": "대기중", "time": "-"}, ] # Progress Bar 스타일 y_pos = list(range(len(workflow_steps))) colors = [] for step_info in workflow_steps: if step_info["status"] == "완료": colors.append("lightgreen") elif step_info["status"] == "진행중": colors.append("lightyellow") else: colors.append("lightgray") fig.add_trace(go.Bar( y=[s["step"] for s in workflow_steps], x=[100 if s["status"] == "완료" else 50 if s["status"] == "진행중" else 0 for s in workflow_steps], orientation='h', marker=dict(color=colors), text=[f"{s['status']} ({s['time']})" for s in workflow_steps], textposition='auto', )) fig.update_layout( title_text="🔄 구매 워크플로우 진행 현황", xaxis_title="진행률 (%)", height=400, showlegend=False ) return fig def create_executive_kpi_dashboard( opt_result: Dict, offers_df: pd.DataFrame, purchase_history: pd.DataFrame ) -> go.Figure: """경영진 KPI 대시보드""" fig = make_subplots( rows=2, cols=3, subplot_titles=( '💰 비용 절감', '⚖️ 컴플라이언스', '📈 ESG 점수', '⏱️ 처리 시간', '🎯 목표 달성률', '📊 월간 트렌드' ), specs=[ [{"type": "indicator"}, {"type": "indicator"}, {"type": "indicator"}], [{"type": "indicator"}, {"type": "indicator"}, {"type": "scatter"}] ] ) # 1. 비용 절감 if len(offers_df) > 0: min_price = offers_df['unit_price'].min() max_price = offers_df['unit_price'].max() savings = ((max_price - min_price) / max_price * 100) if max_price > 0 else 0 else: savings = 0 fig.add_trace( go.Indicator( mode="gauge+number+delta", value=savings, title={'text': "비용 절감률 (%)"}, delta={'reference': 10}, gauge={ 'axis': {'range': [0, 50]}, 'bar': {'color': "darkblue"}, 'steps': [ {'range': [0, 10], 'color': "lightgray"}, {'range': [10, 25], 'color': "lightgreen"}, {'range': [25, 50], 'color': "green"} ], 'threshold': { 'line': {'color': "red", 'width': 4}, 'thickness': 0.75, 'value': 15 } } ), row=1, col=1 ) # 2. 컴플라이언스 준수율 fig.add_trace( go.Indicator( mode="gauge+number", value=100, title={'text': "규정 준수율 (%)"}, gauge={ 'axis': {'range': [0, 100]}, 'bar': {'color': "green"}, 'steps': [ {'range': [0, 80], 'color': "lightcoral"}, {'range': [80, 95], 'color': "lightyellow"}, {'range': [95, 100], 'color': "lightgreen"} ] } ), row=1, col=2 ) # 3. ESG 평균 점수 if len(offers_df) > 0: esg_score = offers_df['esg_level'].map({'A': 100, 'B': 70, 'C': 40}).mean() else: esg_score = 0 fig.add_trace( go.Indicator( mode="gauge+number", value=esg_score, title={'text': "ESG 평균 점수"}, gauge={ 'axis': {'range': [0, 100]}, 'bar': {'color': "darkgreen"}, 'steps': [ {'range': [0, 50], 'color': "lightcoral"}, {'range': [50, 80], 'color': "lightyellow"}, {'range': [80, 100], 'color': "lightgreen"} ] } ), row=1, col=3 ) # 4. 평균 처리 시간 fig.add_trace( go.Indicator( mode="number+delta", value=20, title={'text': "처리 시간 (분)"}, delta={'reference': 30, 'increasing': {'color': "red"}, 'decreasing': {'color': "green"}}, number={'suffix': "분"} ), row=2, col=1 ) # 5. 목표 달성률 fig.add_trace( go.Indicator( mode="gauge+number", value=85, title={'text': "월간 목표 달성률 (%)"}, gauge={ 'axis': {'range': [0, 100]}, 'bar': {'color': "royalblue"}, 'threshold': { 'line': {'color': "red", 'width': 4}, 'thickness': 0.75, 'value': 80 } } ), row=2, col=2 ) # 6. 월간 트렌드 months = ['1월', '2월', '3월', '4월', '5월', '6월'] values = [75, 78, 82, 85, 88, 90] fig.add_trace( go.Scatter( x=months, y=values, mode='lines+markers', name='발주 효율', line=dict(color='royalblue', width=3), marker=dict(size=10) ), row=2, col=3 ) fig.update_layout( height=700, showlegend=False, title_text="📊 경영진 KPI 대시보드", title_font_size=22 ) return fig def create_action_items_table(opt_result: Dict, offers_df: pd.DataFrame) -> pd.DataFrame: """경영진 Action Items 생성""" action_items = [] # 1. 즉시 발주 승인 항목 alloc = opt_result.get('allocation', {}) if alloc: for supplier_id, details in alloc.items(): if isinstance(details, dict): action_items.append({ "우선순위": "🔴 긴급", "Action Item": f"{details.get('supplier_name')} 발주 승인", "수량": f"{details.get('qty')}개", "예상 비용": f"{details.get('qty', 0) * details.get('unit_price', 0):,}원", "담당": "구매팀", "기한": "즉시", "상태": "승인 대기" }) # 2. 재고 보충 권고 action_items.append({ "우선순위": "🟡 중요", "Action Item": "안전재고 미달 품목 보충", "수량": "3개 품목", "예상 비용": "검토 필요", "담당": "MRO팀", "기한": "1주일 내", "상태": "검토 중" }) # 3. ESG 개선 if len(offers_df) > 0: c_grade_count = len(offers_df[offers_df['esg_level'] == 'C']) if c_grade_count > 0: action_items.append({ "우선순위": "🟢 보통", "Action Item": "ESG C등급 공급업체 대체 검토", "수량": f"{c_grade_count}개사", "예상 비용": "영향도 분석 필요", "담당": "구매팀", "기한": "1개월 내", "상태": "계획 단계" }) # 4. 장기 계약 협상 action_items.append({ "우선순위": "🟢 보통", "Action Item": "주요 공급업체 장기계약 협상", "수량": "2-3개사", "예상 비용": "5-10% 절감 예상", "담당": "구매팀", "기한": "분기 내", "상태": "계획 단계" }) return pd.DataFrame(action_items) # ========================================================= # Core Components # ========================================================= @dataclass class ToolCallLog: ts: str actor: str tool: str input: Dict[str, Any] output_preview: str class MCPToolRegistry: def __init__(self, tables: Dict[str, pd.DataFrame]): self.tables = tables self.logs: List[ToolCallLog] = [] def _log(self, actor: str, tool: str, inp: Dict[str, Any], out: Any): self.logs.append(ToolCallLog( ts=now_ts(), actor=actor, tool=tool, input=inp, output_preview=str(out)[:500] )) def query_inventory(self, actor: str, item_id: str) -> pd.DataFrame: inv = self.tables["inventory"] stor = self.tables["storages"] df = inv[inv["item_id"] == item_id].copy() if len(df) > 0: df = df.merge(stor, on="storage_id", how="left") self._log(actor, "query_inventory", {"item_id": item_id}, f"{len(df)} rows") return df def query_offers(self, actor: str, item_id: str) -> pd.DataFrame: offers = self.tables["supplier_offers"] suppliers = self.tables["suppliers"] df = offers[offers["item_id"] == item_id].copy() if len(df) > 0: df = df.merge(suppliers, on="supplier_id", how="left") self._log(actor, "query_offers", {"item_id": item_id}, f"{len(df)} rows") return df def query_compat_items(self, actor: str, equipment_id: str) -> pd.DataFrame: compat = self.tables["compat"] items = self.tables["items"] df = compat[compat["equipment_id"] == equipment_id].copy() if len(df) > 0: df = df.merge(items, on="item_id", how="left") self._log(actor, "query_compat_items", {"equipment_id": equipment_id}, f"{len(df)} rows") return df def get_equipment_info(self, actor: str, equipment_id: str) -> Dict[str, Any]: eq = self.tables["equipment"] match = eq[eq["equipment_id"] == equipment_id] if len(match) == 0: return {} info = match.iloc[0].to_dict() self._log(actor, "get_equipment_info", {"equipment_id": equipment_id}, safe_json(info)) return info def audit_log_df(self) -> pd.DataFrame: if not self.logs: return pd.DataFrame({"메시지": ["로그 없음"]}) return pd.DataFrame([{ "시간": l.ts[:19], "에이전트": l.actor, "도구": l.tool, "입력": str(l.input)[:50], } for l in self.logs]) def apply_rules(tables: Dict[str, pd.DataFrame], item_id: str, supplier_row: Dict[str, Any]) -> Dict[str, Any]: """Apply rules""" items = tables["items"] item_match = items[items["item_id"] == item_id] if len(item_match) == 0: return {"block": False, "alerts": [], "explanations": [], "rules_fired": []} item = item_match.iloc[0].to_dict() decision = { "block": False, "alerts": [], "explanations": [], "rules_fired": [] } if item.get("risk_class") == "규제" and supplier_row.get("region") == "해외": decision["block"] = True decision["rules_fired"].append("R-001") decision["explanations"].append( f"🚫 R-001: 규제품목({item.get('item_name')}) 해외업체({supplier_row.get('supplier_name')}) 구매 차단" ) if supplier_row.get("esg_level") == "C": decision["rules_fired"].append("R-004") decision["explanations"].append( f"📊 R-004: ESG C등급({supplier_row.get('supplier_name')}) 패널티" ) return decision def optimize_order_allocation(demand_qty: int, offers_df: pd.DataFrame, rules_eval: Dict[str, Dict[str, Any]]) -> Dict[str, Any]: """Optimize allocation""" if not PULP_AVAILABLE: return { "status": "UNAVAILABLE", "reason": "PuLP not installed", "allocation": {}, "demand": demand_qty } feasible = [] blocked = [] for _, r in offers_df.iterrows(): sid = r["supplier_id"] if rules_eval.get(sid, {}).get("block"): blocked.append({ "supplier_id": sid, "supplier_name": r.get("supplier_name", sid), "reason": "규칙 위반" }) else: feasible.append(r) if len(feasible) == 0: return { "status": "INFEASIBLE", "reason": "모든 공급업체 차단", "allocation": {}, "blocked_suppliers": blocked, "demand": demand_qty } fdf = pd.DataFrame(feasible) prob = LpProblem("MRO_Allocation", LpMinimize) x = {} for _, r in fdf.iterrows(): sid = r["supplier_id"] x[sid] = LpVariable(f"x_{sid}", lowBound=0, cat="Integer") prob += lpSum(list(x.values())) >= demand_qty, "DemandConstraint" obj_terms = [] for _, r in fdf.iterrows(): sid = r["supplier_id"] price = float(r["unit_price"]) obj_terms.append(x[sid] * price) prob += lpSum(obj_terms), "TotalCost" prob.solve() alloc = {} total_cost = 0.0 for _, r in fdf.iterrows(): sid = r["supplier_id"] val = x[sid].value() if val is not None and val > 0: qty = int(val) alloc[sid] = { "qty": qty, "unit_price": float(r["unit_price"]), "supplier_name": r.get("supplier_name", sid), "lead_time": int(r.get("lead_time_days", 0)) } total_cost += qty * float(r["unit_price"]) return { "status": LpStatus.get(prob.status, "Unknown"), "allocation": alloc, "demand": demand_qty, "blocked_suppliers": blocked, "total_cost": round(total_cost, 2) } class LLMOrchestrator: def __init__(self): self.api_key = os.environ.get("OPENAI_API_KEY", "").strip() self.demo_mode = (not self.api_key or not OPENAI_AVAILABLE) if not self.demo_mode: try: self.client = OpenAI(api_key=self.api_key) test_resp = self.client.chat.completions.create( model="gpt-4o-mini", messages=[{"role": "user", "content": "test"}], max_tokens=5 ) print("✅ OpenAI API 연결 성공!") except Exception: self.demo_mode = True self.client = None else: self.client = None def chat(self, role: str, system: str, user: str) -> str: if self.demo_mode: return self._demo_response(role) try: resp = self.client.chat.completions.create( model="gpt-4o-mini", temperature=0.2, messages=[ {"role": "system", "content": system}, {"role": "user", "content": user} ] ) return resp.choices[0].message.content except Exception as e: return f"[ERROR: {e}]\n" + self._demo_response(role) def _demo_response(self, role: str) -> str: return f"[DEMO MODE - {role}] AI 분석 완료" # ========================================================= # LangGraph Workflow # ========================================================= class DemoState(TypedDict, total=False): tables: Dict[str, pd.DataFrame] mcp: MCPToolRegistry llm: LLMOrchestrator scenario: str equipment_id: str item_id: str demand_qty: int priority: str tables_ok: bool validation_issues: List[str] progress: str inventory_view: pd.DataFrame offers_view: pd.DataFrame rules_eval: Dict[str, Any] optimization: Dict[str, Any] narrative: Dict[str, str] audit_log: pd.DataFrame selected_item_name: str equipment_info: Dict[str, Any] compat_items: pd.DataFrame def node_validate(state: DemoState) -> DemoState: ok, issues = validate_tables(state["tables"]) state["tables_ok"] = ok state["validation_issues"] = issues state["progress"] = "1/4 검증 완료" return state def node_mro_agent(state: DemoState) -> DemoState: mcp: MCPToolRegistry = state["mcp"] equipment_id = state.get("equipment_id", "") item_id = state.get("item_id", "") # Get equipment info equipment_info = mcp.get_equipment_info("MRO_AGENT", equipment_id) state["equipment_info"] = equipment_info # Get compatible items compat_df = pd.DataFrame() if equipment_id: compat_df = mcp.query_compat_items("MRO_AGENT", equipment_id) state["compat_items"] = compat_df if not item_id and len(compat_df) > 0: mandatory = compat_df[compat_df["is_mandatory"] == True] if len(mandatory) > 0: selected = mandatory.iloc[0] else: selected = compat_df.iloc[0] item_id = selected["item_id"] state["item_id"] = item_id state["selected_item_name"] = selected.get("item_name", item_id) inv_df = pd.DataFrame() if item_id: inv_df = mcp.query_inventory("MRO_AGENT", item_id) state["inventory_view"] = inv_df llm: LLMOrchestrator = state["llm"] if "narrative" not in state: state["narrative"] = {} state["narrative"]["mro"] = llm.chat("MRO", "MRO 분석", "설비/재고") state["progress"] = "2/4 MRO 완료" return state def node_procurement_agent(state: DemoState) -> DemoState: mcp: MCPToolRegistry = state["mcp"] item_id = state.get("item_id", "") demand_qty = int(state.get("demand_qty", 10)) offers_df = pd.DataFrame() if item_id: offers_df = mcp.query_offers("PROC_AGENT", item_id) state["offers_view"] = offers_df rules_eval = {} if len(offers_df) > 0: for _, r in offers_df.iterrows(): sid = r["supplier_id"] supplier_row = { "supplier_id": sid, "supplier_name": r.get("supplier_name", sid), "region": r.get("region", ""), "esg_level": r.get("esg_level", ""), } rules_eval[sid] = apply_rules(state["tables"], item_id, supplier_row) state["rules_eval"] = rules_eval opt_result = {} if len(offers_df) > 0: opt_result = optimize_order_allocation(demand_qty, offers_df, rules_eval) else: opt_result = { "status": "NO_DATA", "reason": "공급업체 정보 없음", "allocation": {}, "demand": demand_qty } state["optimization"] = opt_result llm: LLMOrchestrator = state["llm"] state["narrative"]["proc"] = llm.chat("PROC", "구매 전략", "최적화") state["narrative"]["exec"] = llm.chat("EXEC", "임원 요약", "종합") state["progress"] = "3/4 구매 완료" return state def node_collect_audit(state: DemoState) -> DemoState: mcp: MCPToolRegistry = state["mcp"] state["audit_log"] = mcp.audit_log_df() state["progress"] = "4/4 완료 ✓" return state def build_workflow(): if not LANGGRAPH_AVAILABLE: return None try: graph = StateGraph(DemoState) graph.add_node("validate", node_validate) graph.add_node("mro_agent", node_mro_agent) graph.add_node("procurement_agent", node_procurement_agent) graph.add_node("collect_audit", node_collect_audit) graph.set_entry_point("validate") graph.add_edge("validate", "mro_agent") graph.add_edge("mro_agent", "procurement_agent") graph.add_edge("procurement_agent", "collect_audit") graph.add_edge("collect_audit", END) return graph.compile() except Exception as e: print(f"⚠️ LangGraph failed: {e}") return None APP = build_workflow() # ========================================================= # Main Execution - Enhanced with Dashboards # ========================================================= def run_demo(scenario: str, seed: int, equipment_id: str, item_id: str, demand_qty: int) -> Tuple: """Main execution - returns 12 outputs (enhanced)""" try: seed_int = int(seed) demand_int = int(demand_qty) tables = generate_demo_tables(seed=seed_int) mcp = MCPToolRegistry(tables) llm = LLMOrchestrator() preset = SCENARIO_PRESETS.get(scenario, SCENARIO_PRESETS["긴급 고장 대응"]) state: DemoState = { "tables": tables, "mcp": mcp, "llm": llm, "scenario": scenario, "equipment_id": equipment_id.strip(), "item_id": item_id.strip(), "demand_qty": demand_int, "priority": preset.get("priority", "정상"), } if APP is not None: out = APP.invoke(state) else: out = node_validate(state) out = node_mro_agent(out) out = node_procurement_agent(out) out = node_collect_audit(out) status = { "mode": "⚠️ DEMO" if llm.demo_mode else "✅ LLM", "scenario": scenario, "tables_ok": out.get("tables_ok", False), "equipment": out.get("equipment_id", ""), "item_name": out.get("selected_item_name", ""), "demand": out.get("demand_qty", 0), "priority": out.get("priority", "정상"), "progress": out.get("progress", "완료"), } status_text = format_status(status) # Data extraction inv_df = out.get("inventory_view", pd.DataFrame()) offers_df = out.get("offers_view", pd.DataFrame()) audit_df = out.get("audit_log", pd.DataFrame()) equipment_info = out.get("equipment_info", {}) compat_items = out.get("compat_items", pd.DataFrame()) rules_eval = out.get("rules_eval", {}) opt_result = out.get("optimization", {}) purchase_history = tables.get("purchase_history", pd.DataFrame()) item_name = out.get("selected_item_name", "부품") # Create dashboards mro_dashboard = create_mro_inventory_dashboard(inv_df, item_name) mro_workflow = create_mro_workflow_status(equipment_info, compat_items) proc_dashboard = create_procurement_comparison_dashboard(offers_df, rules_eval) proc_workflow = create_procurement_workflow(opt_result) exec_dashboard = create_executive_kpi_dashboard(opt_result, offers_df, purchase_history) action_items = create_action_items_table(opt_result, offers_df) # Fallback for empty dataframes if len(audit_df) == 0: audit_df = pd.DataFrame({"메시지": ["감사로그 없음"]}) print("✅ 대시보드 생성 완료\n") # Return 12 outputs return ( status_text, # 1 mro_dashboard, # 2 - MRO 재고 대시보드 mro_workflow, # 3 - MRO 워크플로우 proc_dashboard, # 4 - 구매 비교 대시보드 proc_workflow, # 5 - 구매 워크플로우 exec_dashboard, # 6 - 경영진 KPI action_items, # 7 - Action Items offers_df, # 8 - 공급업체 원본 데이터 inv_df, # 9 - 재고 원본 데이터 opt_result, # 10 - 최적화 결과 (dict를 text로) audit_df, # 11 - 감사 로그 out.get("selected_item_name", "N/A") # 12 - 품목명 ) except Exception as e: print(f"❌ 오류: {e}\n{traceback.format_exc()}") error_msg = f"❌ 오류\n\n{str(e)}" empty_fig = go.Figure() empty_fig.add_annotation(text="오류 발생", showarrow=False) empty_df = pd.DataFrame({"오류": [str(e)[:100]]}) return ( error_msg, empty_fig, empty_fig, empty_fig, empty_fig, empty_fig, empty_df, empty_df, empty_df, {}, empty_df, "N/A" ) def update_scenario(scenario: str) -> Tuple[str, str, int, str]: preset = SCENARIO_PRESETS.get(scenario, SCENARIO_PRESETS["긴급 고장 대응"]) guide_text = f"""**📌 {preset['description']}** **배경**: {preset['context']} **우선순위**: {preset.get('priority', '정상')} **가이드**: {preset.get('guide', '')} """ return ( preset["equipment_id"], preset["item_id"], preset["demand_qty"], guide_text ) # ========================================================= # Enhanced Gradio UI with Process Guides # ========================================================= print("🎨 프로세스 가이드 통합 UI 구성 중...\n") with gr.Blocks(title="POSCO DX MRO Composite AI", theme=gr.themes.Soft()) as demo: gr.Markdown(""" # 🏭 POSCO DX - MRO Composite AI 프로세스 가이드 통합 버전 ## 🎯 업무 프로세스 자동화 + AI 의사결정 지원 시스템 **3-Agent Collaboration**: MRO 운영 → 구매/조달 → 경영진 승인 """) # Process Overview Section with gr.Accordion("📖 전체 업무 프로세스 개요", open=False): gr.Markdown(""" ### 🔄 End-to-End 워크플로우 ``` ┌─────────────────┐ ┌──────────────────┐ ┌─────────────────┐ │ 1️⃣ MRO 운영 │ ───> │ 2️⃣ 구매/조달 │ ───> │ 3️⃣ 경영진 승인 │ │ │ │ │ │ │ │ • 고장 접수 │ │ • 공급업체 조회 │ │ • KPI 확인 │ │ • 부품 확인 │ │ • 규정 검증 │ │ • 의사결정 │ │ • 재고 확인 │ │ • 최적화 분석 │ │ • 피드백 │ │ • 발주 요청 │ │ • 승인 요청 │ │ │ └─────────────────┘ └──────────────────┘ └─────────────────┘ ⏱️ 15분 ⏱️ 25분 ⏱️ 25분 ``` ### 💡 핵심 가치 1. **자동화**: 설비-부품 매칭, 재고 조회, 규정 검증 등 반복 업무 자동화 2. **최적화**: AI 기반 비용 최적화 및 공급업체 선정 3. **검증**: Neuro-Symbolic AI로 100% 규정 준수 보장 4. **가시성**: 실시간 대시보드로 전 과정 모니터링 5. **추적성**: 모든 의사결정 과정 자동 기록 ### 📊 기대 효과 - ⏱️ **처리 시간**: 기존 3-5일 → **1시간 이내** - 💰 **비용 절감**: 평균 **15-25%** 구매 비용 절감 - ⚖️ **컴플라이언스**: **100%** 규정 준수 - 📈 **효율성**: 담당자 업무 시간 **60%** 단축 """) with gr.Row(): with gr.Column(): gr.Markdown("### 🎯 시나리오") scenario_radio = gr.Radio( choices=list(SCENARIO_PRESETS.keys()), value="긴급 고장 대응", label="분석 시나리오" ) scenario_info = gr.Markdown( value=f"""**📌 {SCENARIO_PRESETS['긴급 고장 대응']['description']}** **배경**: {SCENARIO_PRESETS['긴급 고장 대응']['context']} **가이드**: {SCENARIO_PRESETS['긴급 고장 대응']['guide']} """ ) with gr.Column(): gr.Markdown("### ⚙️ 파라미터") seed_number = gr.Number(value=7, label="데이터 시드", precision=0) equipment_text = gr.Textbox(value="CONV-PH-007", label="설비 ID") item_text = gr.Textbox(value="", label="품목 ID (선택)") demand_number = gr.Number(value=10, label="수량", precision=0) run_button = gr.Button("🚀 Composite AI 분석 실행", variant="primary", size="lg") gr.Markdown("---") status_output = gr.Textbox(label="📊 실행 상태", lines=10) selected_item_display = gr.Textbox(label="📦 선택된 품목", interactive=False) with gr.Tabs(): with gr.Tab("🔧 MRO 담당자"): # Process Guide for MRO with gr.Accordion("📋 MRO 운영 프로세스 가이드", open=True): mro_process_html = gr.HTML(create_process_guide_html("mro")) gr.Markdown("---") gr.Markdown("### 📊 대시보드 및 분석 결과") mro_inventory_plot = gr.Plot(label="📦 재고 분석 대시보드") mro_workflow_plot = gr.Plot(label="🔄 MRO 워크플로우 진행") mro_inventory_table = gr.Dataframe(label="📋 상세 재고 데이터") with gr.Tab("💰 구매 담당자"): # Process Guide for Procurement with gr.Accordion("📋 구매/조달 프로세스 가이드", open=True): proc_process_html = gr.HTML(create_process_guide_html("procurement")) gr.Markdown("---") gr.Markdown("### 📊 대시보드 및 분석 결과") proc_comparison_plot = gr.Plot(label="📊 공급업체 비교 대시보드") proc_workflow_plot = gr.Plot(label="🔄 구매 워크플로우") proc_offers_table = gr.Dataframe(label="📋 공급업체 상세 정보") with gr.Tab("👔 경영진"): # Process Guide for Executive with gr.Accordion("📋 경영진 의사결정 프로세스 가이드", open=True): exec_process_html = gr.HTML(create_process_guide_html("executive")) gr.Markdown("---") gr.Markdown("### 📊 대시보드 및 분석 결과") exec_kpi_plot = gr.Plot(label="📊 경영진 KPI 대시보드") exec_action_table = gr.Dataframe(label="📋 Action Items") gr.Markdown("### 💬 경영진 피드백") with gr.Row(): feedback_text = gr.Textbox( label="개선 제안 / 피드백", placeholder="예: ESG C등급 업체 비중을 20% 이하로 제한하시기 바랍니다.", lines=3 ) with gr.Row(): approve_btn = gr.Button("✅ 승인", variant="primary") reject_btn = gr.Button("❌ 반려", variant="stop") suggest_btn = gr.Button("💡 개선 제안", variant="secondary") feedback_output = gr.Textbox(label="피드백 처리 결과", lines=2) with gr.Tab("📝 감사 로그"): gr.Markdown(""" ### 감사 추적 (Audit Trail) **목적**: 모든 의사결정 과정 추적 및 컴플라이언스 확보 **기록 항목**: - 🕐 시간: 작업 수행 시각 - 👤 에이전트: MRO/구매/경영진 - 🔧 도구: 사용한 기능 - 📥 입력: 파라미터 - 📤 출력: 결과 요약 **활용**: - 규정 준수 감사 - 프로세스 개선 - 책임 추적성 """) audit_table = gr.Dataframe(label="📝 전체 감사 로그") # Hidden outputs for optimization result opt_result_json = gr.JSON(label="최적화 상세 결과", visible=False) gr.Markdown(""" --- ## 💡 시스템 사용 가이드 ### 📋 단계별 사용법 #### 1️⃣ 시나리오 선택 - **긴급 고장 대응**: 설비 고장으로 즉시 교체가 필요한 경우 - **정기 발주 계획**: 월간/분기 정기 발주 최적화 - **규정 준수 검증**: 특수 품목 구매 시 컴플라이언스 확인 #### 2️⃣ 파라미터 입력 - **설비 ID**: 고장/정비 대상 설비 (자동 입력) - **품목 ID**: 특정 품목 지정 (선택사항, 비우면 자동 선택) - **수량**: 발주 필요 수량 #### 3️⃣ 분석 실행 - "🚀 Composite AI 분석 실행" 버튼 클릭 - 약 5-10초 내 결과 확인 #### 4️⃣ 결과 검토 - **MRO 탭**: 재고 현황 및 발주 필요성 확인 - **구매 탭**: 공급업체 비교 및 최적 선택 - **경영진 탭**: KPI 확인 및 의사결정 #### 5️⃣ 승인/피드백 - Action Items 검토 후 승인/반려 결정 - 개선 제안 입력 시 자동으로 담당 부서에 전달 ### 🎓 프로세스 이해 각 탭의 "📋 프로세스 가이드"를 펼치면: - 단계별 상세 절차 - 입력/출력 명세 - 담당자 및 소요 시간 - 성공 기준 을 확인할 수 있습니다. ### 🔍 주요 기능 1. **자동 부품 매칭**: 설비 ID만으로 호환 부품 자동 검색 2. **전사 재고 통합**: 본사, 포항, 광양 전체 창고 실시간 조회 3. **AI 규정 검증**: 규제품목, ESG 등급 등 자동 검증 4. **최적화 엔진**: Linear Programming으로 비용 최소화 5. **인터랙티브 대시보드**: Plotly 차트로 드릴다운 분석 가능 ### 🔑 API Key 설정 (Hugging Face Spaces) OpenAI 기능을 사용하려면: 1. Space Settings → Secrets으로 이동 2. 새 Secret 추가: - Name: `OPENAI_API_KEY` - Value: `your-openai-api-key` 3. Space 재시작 API 키 없이도 데모 모드로 기본 기능 사용 가능합니다. """) # Event Handlers scenario_radio.change( fn=update_scenario, inputs=[scenario_radio], outputs=[equipment_text, item_text, demand_number, scenario_info] ) run_button.click( fn=run_demo, inputs=[scenario_radio, seed_number, equipment_text, item_text, demand_number], outputs=[ status_output, # 1 mro_inventory_plot, # 2 mro_workflow_plot, # 3 proc_comparison_plot, # 4 proc_workflow_plot, # 5 exec_kpi_plot, # 6 exec_action_table, # 7 proc_offers_table, # 8 mro_inventory_table, # 9 opt_result_json, # 10 audit_table, # 11 selected_item_display # 12 ] ) # Feedback handlers def handle_approve(feedback): return f"✅ 승인 완료: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n피드백: {feedback}" def handle_reject(feedback): return f"❌ 반려됨: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n사유: {feedback}" def handle_suggest(feedback): return f"💡 개선 제안 접수: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n내용: {feedback}" approve_btn.click(fn=handle_approve, inputs=[feedback_text], outputs=[feedback_output]) reject_btn.click(fn=handle_reject, inputs=[feedback_text], outputs=[feedback_output]) suggest_btn.click(fn=handle_suggest, inputs=[feedback_text], outputs=[feedback_output]) print("=" * 60) print("✅ 프로세스 가이드 통합 UI 완료!") print("=" * 60) if __name__ == "__main__": demo.launch( server_name="0.0.0.0", server_port=7860, show_error=True ) print("\n🎉 프로세스 가이드 통합 버전 실행 중!\n")