# ========================================================= # POSCO DX - MRO Composite AI - HUGGING FACE SPACES VERSION # ========================================================= 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 # ✅ 허깅페이스 Secrets에서 API 키 로드 OPENAI_API_KEY = os.environ.get('OPENAI_API_KEY', '') if OPENAI_API_KEY: print("✅ OpenAI API Key loaded from Hugging Face Secrets") os.environ['OPENAI_API_KEY'] = OPENAI_API_KEY else: print("⚠️ DEMO MODE (OpenAI API Key not found)") print("💡 Tip: Add OPENAI_API_KEY to Hugging Face Secrets") print("\n" + "=" * 60) print("✅ 허깅페이스 스페이스 버전 초기화 완료!") print("=" * 60 + "\n") # ========================================================= # 3. 프로세스 가이드 설정 # ========================================================= 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": "AI 시스템 (Knowledge Graph)", "duration": "2분" }, { "id": "4", "name": "전사 재고 현황 확인", "description": "모든 창고의 재고를 실시간으로 조회합니다", "input": "부품 ID 리스트", "output": "창고별 재고수량, 안전재고, 가용재고", "owner": "AI 시스템 (재고 DB)", "duration": "1분" }, { "id": "5", "name": "발주 필요성 판단", "description": "재고 부족 여부를 판단하고 발주 수량을 산출합니다", "input": "가용재고, 수요수량, 안전재고", "output": "발주 필요 여부, 권장 발주량", "owner": "MRO팀 + AI 분석", "duration": "3분" }, { "id": "6", "name": "구매팀 발주 요청", "description": "구매팀에 발주 요청서를 전달합니다", "input": "부품명, 수량, 우선순위, 납기요구", "output": "발주요청번호", "owner": "MRO팀 → 구매팀", "duration": "2분" } ], "success_criteria": [ "설비 정보 정확히 식별", "호환 부품 100% 매칭", "재고 현황 실시간 반영", "발주 수량 최적화" ], "total_duration": "약 15분" }, "procurement": { "title": "💰 구매/조달 프로세스", "steps": [ { "id": "1", "name": "발주 요청 접수", "description": "MRO팀으로부터 발주 요청을 접수합니다", "input": "발주요청번호, 부품정보, 수량", "output": "구매건번호, 담당자 배정", "owner": "구매팀 접수", "duration": "3분" }, { "id": "2", "name": "공급업체 정보 조회", "description": "해당 부품을 공급 가능한 업체를 조회합니다", "input": "부품 ID, 품목 카테고리", "output": "공급업체 리스트, 견적정보, ESG등급", "owner": "AI 시스템 (공급업체 DB)", "duration": "2분" }, { "id": "3", "name": "규정 준수 검증", "description": "Neuro-Symbolic AI로 구매 규정 위반 여부를 자동 검증합니다", "input": "공급업체 정보, 품목 특성, 규정 룰셋", "output": "적합/부적합/경고, 위반 규정 목록", "owner": "AI 시스템 (규정 엔진)", "duration": "1분" }, { "id": "4", "name": "최적 배분 계산", "description": "Linear Programming으로 최저비용 발주 계획을 수립합니다", "input": "공급업체 단가, MOQ, 리드타임, 제약조건", "output": "업체별 발주량, 총 비용, 예상 절감액", "owner": "AI 시스템 (PuLP)", "duration": "2분" }, { "id": "5", "name": "발주 전략 수립", "description": "GPT-4 기반으로 협상 전략 및 리스크를 분석합니다", "input": "시장 동향, 과거 구매 이력, 최적화 결과", "output": "협상 포인트, 리스크 분석, 대안 시나리오", "owner": "AI 시스템 (LLM)", "duration": "5분" }, { "id": "6", "name": "경영진 승인 요청", "description": "발주 계획을 경영진에게 승인 요청합니다", "input": "발주 계획서, KPI 분석, 비용편익 분석", "output": "승인요청번호", "owner": "구매팀 → 경영진", "duration": "3분" }, { "id": "7", "name": "PO 발행", "description": "승인된 발주 계획에 따라 Purchase Order를 발행합니다", "input": "승인번호, 공급업체 정보, 발주 상세", "output": "PO 번호, 공급업체 전송 완료", "owner": "구매팀 (ERP 연동)", "duration": "10분" } ], "success_criteria": [ "규정 100% 준수", "비용 최소화 달성", "리스크 사전 식별", "승인 프로세스 완료" ], "total_duration": "약 25분" }, "executive": { "title": "👔 경영진 의사결정 프로세스", "steps": [ { "id": "1", "name": "승인 요청 알림", "description": "구매팀으로부터 승인 요청을 수신합니다", "input": "승인요청번호, 발주 요약", "output": "알림 확인", "owner": "시스템 → 경영진", "duration": "즉시" }, { "id": "2", "name": "KPI 대시보드 확인", "description": "비용 절감률, 컴플라이언스, ESG 지표 등을 확인합니다", "input": "발주 건번호", "output": "KPI 현황 (비용/규정/ESG/시간)", "owner": "경영진 (대시보드)", "duration": "2분" }, { "id": "3", "name": "Action Items 검토", "description": "우선순위별 조치 사항을 검토합니다", "input": "Action Items 리스트", "output": "검토 완료", "owner": "경영진", "duration": "5분" }, { "id": "4", "name": "발주 상세 분석", "description": "공급업체, 가격, 리스크 등 상세 내용을 분석합니다", "input": "발주 계획서, AI 분석 리포트", "output": "분석 메모", "owner": "경영진", "duration": "10분" }, { "id": "5", "name": "의사결정", "description": "승인/반려/조건부승인 중 하나를 선택합니다", "input": "종합 분석 결과", "output": "승인 여부, 결정 사유", "owner": "경영진", "duration": "3분" }, { "id": "6", "name": "피드백 제공", "description": "향후 개선을 위한 피드백을 입력합니다", "input": "개선 제안 사항", "output": "피드백 저장, AI 학습 반영", "owner": "경영진", "duration": "5분" } ], "success_criteria": [ "KPI 목표 달성 확인", "리스크 검토 완료", "신속한 의사결정", "피드백 루프 구축" ], "total_duration": "약 25분" } } # ========================================================= # 4. 시나리오 프리셋 # ========================================================= SCENARIO_PRESETS = { "긴급 고장 대응": { "description": "🚨 포항제철소 컨베이어 베어링 긴급 고장", "equipment_id": "CONV-PH-001", "item_id": "", "demand_qty": 10, "context": "컨베이어 베어링 고장으로 생산라인 중단. 즉시 교체 필요.", "priority": "긴급", "guide": "리드타임 최소화 우선. 국내 공급업체 우선 고려." }, "정기 발주 계획": { "description": "📋 월간 정기 발주 - 유압펌프 예방정비", "equipment_id": "PUMP-GY-001", "item_id": "", "demand_qty": 50, "context": "월간 예방정비 계획. 최적 가격 및 재고 균형 필요.", "priority": "정상", "guide": "비용 최적화 우선. ESG 등급 고려." }, "규정 준수 검증": { "description": "⚖️ 규제품목(특수화학물질) 구매 검증", "equipment_id": "VALVE-PH-001", "item_id": "", "demand_qty": 20, "context": "특수 실링재 구매. 해외구매 차단 규정 준수 필수.", "priority": "규정준수", "guide": "컴플라이언스 100% 준수. 국내업체만 허용." } } # 실제 부품명 및 카테고리 REAL_PART_NAMES = { "베어링": ["SKF 6205 볼베어링", "NSK 원통베어링", "NTN 테이퍼베어링"], "윤활유": ["쉘 오마라 220", "모빌 DTE 25", "지에스칼텍스 터빈유"], "필터": ["하이드로락 유압필터", "파커 에어필터", "도난드슨 정밀필터"], "벨트": ["게이츠 파워그립 벨트", "반도 V벨트", "옵티벨트 타이밍벨트"], "센서": ["지멘스 근접센서", "오므론 광전센서", "하니웰 압력센서"], "패킹": ["NOK 오링", "파커 유압씰", "발카 그랜드패킹"], "퓨즈": ["LS산전 MCCB", "슈나이더 차단기", "ABB 퓨즈"], "호스": ["파커 유압호스", "만리 고압호스", "브리지스톤 산업호스"], "볼트": ["SUS304 육각볼트", "고장력볼트 F10T", "앵커볼트 M16"], "실링재": ["록타이트 실란트", "쓰리본드 액상패킹", "헨켈 밀봉재"] } # 실제 공급업체 정보 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": "밸브/센서"} ] # ========================================================= # 5. 유틸리티 함수 # ========================================================= 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 # ========================================================= # 6. 데이터 생성 함수 (재고 보장) # ========================================================= def generate_demo_tables(seed: int = 7) -> Dict[str, pd.DataFrame]: """충분한 재고가 보장된 데모 데이터 생성""" 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 = [] eq_configs = [ ("PUMP", "유압펌프", ["PH", "GY"], 6), ("CONV", "컨베이어", ["PH", "GY"], 6), ("VALVE", "제어밸브", ["PH", "GY"], 5), ("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 = [] 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 part_name 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) # 호환성 compat = [] for eq_idx, eq_row in equipment.iterrows(): eq_id = eq_row["equipment_id"] eq_type = eq_row["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 = 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 = [] for _, item in items.iterrows(): num_storages = random.randint(1, 3) selected_storages = storages.sample(num_storages) for _, storage in selected_storages.iterrows(): stock_level = random.randint(50, 150) # 충분한 재고 safety_stock = random.randint(10, min(30, stock_level - 10)) # stock_level보다 작게 available = stock_level - safety_stock reserved = random.randint(0, min(10, available)) if available > 0 else 0 inventory.append({ "storage_id": storage["storage_id"], "item_id": item["item_id"], "on_hand": stock_level, "safety_stock": safety_stock, "reserved": reserved, "last_updated": (datetime.now() - timedelta(days=random.randint(1, 30))).strftime("%Y-%m-%d"), }) inventory = pd.DataFrame(inventory) # 공급업체 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) ]) # 공급업체 오퍼 offers = [] for _, item in items.iterrows(): num_suppliers = random.randint(3, 5) 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 = 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 = [] 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 } # ========================================================= # 7. MCP 도구 레지스트리 # ========================================================= class MCPToolRegistry: def __init__(self): self.log = [] def call_tool(self, tool_name: str, **kwargs): entry = { "timestamp": now_ts(), "tool": tool_name, "input": kwargs, "output": f"Tool {tool_name} executed" } self.log.append(entry) return entry["output"] def get_log(self) -> pd.DataFrame: if not self.log: return pd.DataFrame({"메시지": ["로그 없음"]}) return pd.DataFrame(self.log) # ========================================================= # 8. LLM 오케스트레이터 # ========================================================= class LLMOrchestrator: def __init__(self, api_key: str = ""): self.api_key = api_key self.demo_mode = (not api_key) or (not OPENAI_AVAILABLE) if not self.demo_mode: try: self.client = OpenAI(api_key=api_key) print("✅ LLM: OpenAI 연결") except Exception as e: print(f"⚠️ OpenAI 연결 실패: {e}") self.demo_mode = True else: print("⚠️ LLM: DEMO MODE") def generate(self, prompt: str, temperature: float = 0.1) -> str: if self.demo_mode: return f"[DEMO] {prompt[:100]}에 대한 AI 분석 결과..." try: response = self.client.chat.completions.create( model="gpt-4o-mini", messages=[{"role": "user", "content": prompt}], temperature=temperature, max_tokens=500 ) return response.choices[0].message.content except Exception as e: return f"[ERROR] {str(e)}" # ========================================================= # 9. 핵심 비즈니스 로직 # ========================================================= def match_equipment_items(equipment_id: str, tables: Dict) -> Tuple[Dict, pd.DataFrame]: """설비-부품 매칭""" eq = tables["equipment"] compat = tables["compat"] items = tables["items"] eq_row = eq[eq["equipment_id"] == equipment_id] if len(eq_row) == 0: return {}, pd.DataFrame() eq_info = eq_row.iloc[0].to_dict() compat_items = compat[compat["equipment_id"] == equipment_id] compat_items = compat_items.merge(items, on="item_id", how="left") return eq_info, compat_items def query_inventory(item_id: str, tables: Dict) -> pd.DataFrame: """재고 조회""" inv = tables["inventory"] storages = tables["storages"] item_inv = inv[inv["item_id"] == item_id].copy() if len(item_inv) == 0: return pd.DataFrame() item_inv = item_inv.merge(storages, on="storage_id", how="left") item_inv["available"] = item_inv["on_hand"] - item_inv["reserved"] return item_inv[["storage_name", "on_hand", "safety_stock", "reserved", "available", "last_updated"]] def apply_procurement_rules(offers_df: pd.DataFrame, item_info: Dict, tables: Dict) -> Dict: """규정 검증""" rules_result = { "passed": [], "warnings": [], "blocked": [] } if "supplier_type" not in offers_df.columns: offers_df["supplier_type"] = "국내" if "supplier_name" not in offers_df.columns: offers_df["supplier_name"] = "Unknown" if "esg_level" not in offers_df.columns: offers_df["esg_level"] = "B" # R-001: 규제품목 해외구매 제한 if item_info.get("risk_class") == "규제": blocked_mask = offers_df["supplier_type"] == "해외" blocked_suppliers = offers_df[blocked_mask]["supplier_name"].tolist() if blocked_suppliers: rules_result["blocked"].append({ "rule": "R-001", "suppliers": blocked_suppliers, "reason": "규제품목 해외구매 금지" }) # R-004: ESG C등급 제한 c_grade_mask = offers_df["esg_level"] == "C" c_grade_suppliers = offers_df[c_grade_mask]["supplier_name"].tolist() if c_grade_suppliers: rules_result["warnings"].append({ "rule": "R-004", "suppliers": c_grade_suppliers, "reason": "ESG C등급 패널티" }) return rules_result def optimize_procurement(offers_df: pd.DataFrame, demand_qty: int, rules_eval: Dict) -> Dict: """구매 최적화""" if not PULP_AVAILABLE: return { "status": "UNAVAILABLE", "message": "PuLP not installed", "allocation": {} } blocked_suppliers = [] for block in rules_eval.get("blocked", []): blocked_suppliers.extend(block.get("suppliers", [])) if "supplier_name" in offers_df.columns: valid_offers = offers_df[~offers_df["supplier_name"].isin(blocked_suppliers)].copy() else: valid_offers = offers_df.copy() if len(valid_offers) == 0: return {"status": "NO_VALID_SUPPLIERS"} try: prob = LpProblem("Procurement_Optimization", LpMinimize) vars_dict = {} for idx, row in valid_offers.iterrows(): var = LpVariable(f"qty_{row['supplier_id']}", lowBound=0, cat='Integer') vars_dict[idx] = var prob += lpSum([valid_offers.loc[idx, "unit_price"] * vars_dict[idx] for idx in vars_dict]) prob += lpSum([vars_dict[idx] for idx in vars_dict]) >= demand_qty prob.solve() allocation = {} total_cost = 0 for idx, var in vars_dict.items(): qty = var.varValue if qty and qty > 0: row = valid_offers.loc[idx] allocation[row["supplier_id"]] = { "qty": int(qty), "unit_price": int(row["unit_price"]), "total": int(qty * row["unit_price"]) } total_cost += int(qty * row["unit_price"]) return { "status": LpStatus[prob.status], "total_cost": total_cost, "allocation": allocation, "savings_pct": round(random.uniform(5, 25), 1) } except Exception as e: return { "status": "ERROR", "message": str(e) } # ========================================================= # 10. 에이전트 함수 # ========================================================= def validation_agent(state: Dict) -> Dict: """에이전트 1: 데이터 검증""" print("📋 에이전트 1: 데이터 검증 시작") tables = state["tables"] mcp = state["mcp"] equipment_id = state["equipment_id"] item_id = state.get("item_id", "") mcp.call_tool("match_equipment_items", equipment_id=equipment_id) equipment_info, compat_items = match_equipment_items(equipment_id, tables) if not equipment_info: state["progress"] = "1/4 검증 실패" state["equipment_info"] = {} state["compat_items"] = pd.DataFrame() return state if not item_id and len(compat_items) > 0: mandatory = compat_items[compat_items["is_mandatory"] == True] if len(mandatory) > 0: selected = mandatory.iloc[0] else: selected = compat_items.iloc[0] item_id = selected["item_id"] state["item_id"] = item_id state["selected_item_name"] = selected["item_name"] else: items = tables["items"] item_row = items[items["item_id"] == item_id] if len(item_row) > 0: state["selected_item_name"] = item_row.iloc[0]["item_name"] state["equipment_info"] = equipment_info state["compat_items"] = compat_items state["progress"] = "1/4 검증 완료" print(f"✅ 검증 완료: {equipment_info.get('equipment_name', 'N/A')}") return state def mro_agent(state: Dict) -> Dict: """에이전트 2: MRO 분석""" print("🔧 에이전트 2: MRO 분석 시작") tables = state["tables"] mcp = state["mcp"] llm = state["llm"] item_id = state["item_id"] mcp.call_tool("query_inventory", item_id=item_id) inventory_view = query_inventory(item_id, tables) state["inventory_view"] = inventory_view state["progress"] = "2/4 MRO 완료" if not llm.demo_mode: prompt = f"재고 분석: {state.get('selected_item_name', 'N/A')}, 수요 {state['demand_qty']}개" narrative = llm.generate(prompt, temperature=0.1) state["narrative"] = narrative print(f"✅ MRO 분석 완료: 재고 {len(inventory_view)}개 창고") return state def procurement_agent(state: Dict) -> Dict: """에이전트 3: 구매 분석""" print("💰 에이전트 3: 구매 분석 시작") tables = state["tables"] mcp = state["mcp"] item_id = state["item_id"] demand_qty = state["demand_qty"] offers = tables["supplier_offers"] item_offers = offers[offers["item_id"] == item_id].copy() suppliers = tables["suppliers"] if len(item_offers) > 0: item_offers = item_offers.merge(suppliers, on="supplier_id", how="left") if "supplier_name" not in item_offers.columns: item_offers["supplier_name"] = "Unknown" if "esg_level" not in item_offers.columns: item_offers["esg_level"] = "B" if "supplier_type" not in item_offers.columns: item_offers["supplier_type"] = "국내" items = tables["items"] item_row = items[items["item_id"] == item_id] if len(item_row) > 0: item_info = item_row.iloc[0].to_dict() else: item_info = {"risk_class": "일반"} mcp.call_tool("apply_rules", item_id=item_id) rules_eval = apply_procurement_rules(item_offers, item_info, tables) mcp.call_tool("optimize", demand_qty=demand_qty) optimization = optimize_procurement(item_offers, demand_qty, rules_eval) state["offers_view"] = item_offers state["rules_eval"] = rules_eval state["optimization"] = optimization state["progress"] = "3/4 구매 완료" print(f"✅ 구매 분석 완료: {len(item_offers)}개 공급업체") return state def executive_agent(state: Dict) -> Dict: """에이전트 4: 경영진 의사결정 지원""" print("👔 에이전트 4: 경영진 의사결정 지원 시작") mcp = state["mcp"] audit_log = mcp.get_log() state["audit_log"] = audit_log state["progress"] = "4/4 완료 ✓" print("✅ 경영진 의사결정 지원 완료") return state # ========================================================= # 11. 워크플로우 실행 # ========================================================= def run_composite_ai_workflow( scenario: str, equipment_id: str, item_id: str, demand_qty: int, seed: int = 7 ) -> Dict: """Composite AI 워크플로우 실행""" print(f"\n{'='*60}") print(f"🚀 Composite AI 워크플로우 시작") print(f"{'='*60}\n") tables = generate_demo_tables(seed) state = { "tables": tables, "mcp": MCPToolRegistry(), "llm": LLMOrchestrator(OPENAI_API_KEY), "scenario": scenario, "equipment_id": equipment_id, "item_id": item_id, "demand_qty": demand_qty, "progress": "0/4 시작", "selected_item_name": "N/A" } state = validation_agent(state) state = mro_agent(state) state = procurement_agent(state) state = executive_agent(state) print(f"\n{'='*60}") print(f"✅ Composite AI 워크플로우 완료") print(f"{'='*60}\n") return state # ========================================================= # 12. Plotly 시각화 함수들 # ========================================================= 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="재고 데이터 없음", xref="paper", yref="paper", x=0.5, y=0.5, showarrow=False, font=dict(size=20, color="gray") ) return fig fig = make_subplots( rows=2, cols=2, subplot_titles=("창고별 재고 현황", "총 재고 게이지", "재고 상태 분포", "상세 데이터"), specs=[ [{"type": "bar"}, {"type": "indicator"}], [{"type": "pie"}, {"type": "table"}] ], vertical_spacing=0.12, horizontal_spacing=0.15 ) fig.add_trace( go.Bar( x=inv_df["storage_name"], y=inv_df["on_hand"], name="현재고", marker_color="#2196F3" ), row=1, col=1 ) fig.add_trace( go.Bar( x=inv_df["storage_name"], y=inv_df["safety_stock"], name="안전재고", marker_color="#FFC107" ), row=1, col=1 ) 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, title={"text": f"총 재고량
{item_name}"}, delta={"reference": total_safety, "increasing": {"color": "green"}}, gauge={ "axis": {"range": [0, max(total_stock * 1.2, total_safety * 2)]}, "bar": {"color": "#4CAF50"}, "threshold": { "line": {"color": "red", "width": 4}, "thickness": 0.75, "value": total_safety } } ), row=1, col=2 ) total_available = (inv_df["on_hand"] - inv_df["reserved"]).sum() total_reserved = inv_df["reserved"].sum() fig.add_trace( go.Pie( labels=["가용재고", "예약됨", "안전재고"], values=[total_available, total_reserved, total_safety], hole=0.3, marker=dict(colors=["#4CAF50", "#FF9800", "#F44336"]) ), row=2, col=1 ) fig.add_trace( go.Table( header=dict( values=["창고", "현재고", "안전재고", "예약", "가용"], fill_color="#2196F3", font=dict(color="white", size=12), align="center" ), cells=dict( values=[ inv_df["storage_name"], inv_df["on_hand"], inv_df["safety_stock"], inv_df["reserved"], inv_df["available"] ], fill_color="white", align="center" ) ), row=2, col=2 ) fig.update_layout( title_text=f"📦 MRO 재고 분석 대시보드", height=800, showlegend=True ) return fig def create_mro_workflow_status(equipment_info: Dict, compat_items: pd.DataFrame) -> go.Figure: """MRO 워크플로우 진행 상태""" fig = go.Figure() if not equipment_info: fig.add_annotation(text="설비 정보 없음", showarrow=False) return fig steps = ["고장 접수", "부품 확인", "재고 확인", "발주 요청"] values = [100, 80, 60, 40] fig.add_trace(go.Funnel( y=steps, x=values, textinfo="label+percent initial", marker=dict(color=["#2196F3", "#4CAF50", "#FF9800", "#F44336"]) )) fig.update_layout( title=f"🔄 MRO 워크플로우 진행: {equipment_info.get('equipment_name', 'N/A')}", height=400 ) 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) return fig required_columns = { 'supplier_name': 'Unknown', 'unit_price': 0, 'lead_time_days': 0, 'esg_level': 'B' } for col, default_val in required_columns.items(): if col not in offers_df.columns: offers_df[col] = default_val fig = make_subplots( rows=2, cols=2, subplot_titles=("가격 비교", "가격-납기 분석", "ESG 등급 분포", "종합 평가"), specs=[ [{"type": "bar"}, {"type": "scatter"}], [{"type": "pie"}, {"type": "table"}] ], vertical_spacing=0.12, horizontal_spacing=0.15 ) blocked_suppliers = [] for block in rules_eval.get("blocked", []): blocked_suppliers.extend(block.get("suppliers", [])) if "supplier_name" in offers_df.columns: offers_df["blocked"] = offers_df["supplier_name"].isin(blocked_suppliers) else: offers_df["blocked"] = False colors = ["red" if b else "green" for b in offers_df["blocked"]] fig.add_trace( go.Bar( x=offers_df["supplier_name"], y=offers_df["unit_price"], marker_color=colors, showlegend=False ), row=1, col=1 ) fig.add_trace( go.Scatter( x=offers_df["lead_time_days"], y=offers_df["unit_price"], mode="markers+text", text=offers_df["supplier_name"], textposition="top center", marker=dict( size=15, color=["red" if b else "green" for b in offers_df["blocked"]], line=dict(width=2, color="black") ), showlegend=False ), row=1, col=2 ) try: esg_counts = offers_df["esg_level"].value_counts() fig.add_trace( go.Pie( labels=esg_counts.index, values=esg_counts.values, hole=0.3, marker=dict(colors=["#4CAF50", "#FF9800", "#F44336"]) ), row=2, col=1 ) except: fig.add_trace( go.Pie( labels=["N/A"], values=[1], hole=0.3, marker=dict(colors=["#CCCCCC"]) ), row=2, col=1 ) max_price = offers_df["unit_price"].max() if offers_df["unit_price"].max() > 0 else 1 max_lead = offers_df["lead_time_days"].max() if offers_df["lead_time_days"].max() > 0 else 1 offers_df["price_score"] = 100 - (offers_df["unit_price"] / max_price * 50) offers_df["lead_score"] = 100 - (offers_df["lead_time_days"] / max_lead * 30) esg_map = {"A": 20, "B": 10, "C": 0} offers_df["esg_score"] = offers_df["esg_level"].map(lambda x: esg_map.get(x, 10)) offers_df["total_score"] = offers_df["price_score"] + offers_df["lead_score"] + offers_df["esg_score"] offers_sorted = offers_df.sort_values("total_score", ascending=False) fig.add_trace( go.Table( header=dict( values=["순위", "공급업체", "단가", "납기", "ESG", "점수"], fill_color="#2196F3", font=dict(color="white", size=12), align="center" ), cells=dict( values=[ list(range(1, len(offers_sorted)+1)), offers_sorted["supplier_name"], offers_sorted["unit_price"], offers_sorted["lead_time_days"].astype(str) + "일", offers_sorted["esg_level"], offers_sorted["total_score"].round(1) ], fill_color=[["white" if not b else "#ffcccb" for b in offers_sorted["blocked"]]], align="center" ) ), row=2, col=2 ) fig.update_layout( title_text="📊 공급업체 종합 비교 대시보드", height=800, showlegend=False ) return fig def create_procurement_workflow(opt_result: Dict) -> go.Figure: """구매 워크플로우""" fig = go.Figure() steps = ["요청 접수", "업체 조회", "규정 검증", "최적화", "승인 요청"] values = [100, 90, 75, 60, 40] fig.add_trace(go.Funnel( y=steps, x=values, textinfo="label+percent initial", marker=dict(color=["#2196F3", "#4CAF50", "#FF9800", "#9C27B0", "#F44336"]) )) fig.update_layout( title=f"🔄 구매 워크플로우 (최적화: {opt_result.get('status', 'N/A')})", height=400 ) return fig def create_executive_kpi_dashboard(opt_result: Dict, offers_df: pd.DataFrame, history_df: 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"}] ], vertical_spacing=0.2, horizontal_spacing=0.1 ) savings = opt_result.get("savings_pct", 18.5) fig.add_trace( go.Indicator( mode="gauge+number+delta", value=savings, title={"text": "비용 절감률 (%)"}, delta={"reference": 15, "increasing": {"color": "green"}}, gauge={ "axis": {"range": [0, 30]}, "bar": {"color": "#4CAF50"}, "threshold": {"line": {"color": "red", "width": 4}, "thickness": 0.75, "value": 15} } ), row=1, col=1 ) fig.add_trace( go.Indicator( mode="gauge+number", value=100, title={"text": "규정 준수율 (%)"}, gauge={ "axis": {"range": [80, 100]}, "bar": {"color": "#2196F3"}, "steps": [ {"range": [80, 95], "color": "#FFE082"}, {"range": [95, 100], "color": "#C8E6C9"} ] } ), row=1, col=2 ) if len(offers_df) > 0 and "esg_level" in offers_df.columns: esg_map = {"A": 100, "B": 70, "C": 40} try: avg_esg = offers_df["esg_level"].map(esg_map).mean() if pd.isna(avg_esg): avg_esg = 85 except: avg_esg = 85 else: avg_esg = 85 fig.add_trace( go.Indicator( mode="gauge+number", value=avg_esg, title={"text": "ESG 점수"}, gauge={ "axis": {"range": [0, 100]}, "bar": {"color": "#8BC34A"} } ), row=1, col=3 ) fig.add_trace( go.Indicator( mode="number+delta", value=20, delta={"reference": 30, "decreasing": {"color": "green"}}, title={"text": "처리 시간 (분)"} ), row=2, col=1 ) fig.add_trace( go.Indicator( mode="number+delta", value=92, delta={"reference": 85, "increasing": {"color": "green"}}, title={"text": "KPI 달성률 (%)"} ), row=2, col=2 ) months = ["1월", "2월", "3월", "4월", "5월", "6월"] efficiency = [70, 75, 80, 85, 90, 92] fig.add_trace( go.Scatter( x=months, y=efficiency, mode="lines+markers", line=dict(color="#2196F3", width=3), marker=dict(size=10), name="효율성" ), row=2, col=3 ) fig.update_layout( title_text="📊 경영진 KPI 대시보드", height=600, showlegend=False ) return fig def create_action_items_table(opt_result: Dict, offers_df: pd.DataFrame) -> pd.DataFrame: """Action Items 테이블 생성""" items = [] if opt_result.get("status") == "Optimal": savings = opt_result.get("savings_pct", 0) if savings > 20: items.append({ "우선순위": "🟢 보통", "조치 사항": "비용 절감 목표 초과 달성", "담당": "구매팀", "기한": "1주일" }) elif savings > 10: items.append({ "우선순위": "🟡 중요", "조치 사항": "비용 절감 목표 달성", "담당": "구매팀", "기한": "3일" }) if len(offers_df) > 0: fastest = offers_df.nsmallest(1, "lead_time_days").iloc[0] items.append({ "우선순위": "🔴 긴급", "조치 사항": f"{fastest['supplier_name']} 발주 승인 필요", "담당": "경영진", "기한": "즉시" }) if not items: items.append({ "우선순위": "🟢 보통", "조치 사항": "정상 처리 진행 중", "담당": "MRO팀", "기한": "1주일" }) return pd.DataFrame(items) # ========================================================= # 13. 메인 실행 함수 # ========================================================= def run_analysis( scenario: str, equipment_dropdown: str, item_dropdown: str, demand_qty: int ) -> Tuple: """분석 실행""" try: equipment_id = equipment_dropdown.split(" - ")[0] if equipment_dropdown else "" item_id = item_dropdown.split(" - ")[0] if " - " in item_dropdown else "" if not equipment_id: raise ValueError("설비를 선택해주세요") if demand_qty <= 0: raise ValueError("수량은 1 이상이어야 합니다") out = run_composite_ai_workflow( scenario=scenario, equipment_id=equipment_id, item_id=item_id, demand_qty=int(demand_qty), seed=7 ) tables = out["tables"] equipment_info = out.get("equipment_info", {}) item_name = out.get("selected_item_name", "N/A") mode = "✅ LLM" if (OPENAI_API_KEY and not out["llm"].demo_mode) else "⚠️ DEMO" status_dict = { "mode": mode, "scenario": scenario, "equipment": equipment_info.get("equipment_name", equipment_id), "item_name": item_name, "demand": demand_qty, "priority": SCENARIO_PRESETS.get(scenario, {}).get("priority", "정상"), "tables_ok": True, "progress": out["progress"] } status_text = format_status(status_dict) inv_df = out.get("inventory_view", pd.DataFrame()) offers_df = out.get("offers_view", pd.DataFrame()) audit_df = out.get("audit_log", pd.DataFrame()) 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()) 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) if len(audit_df) == 0: audit_df = pd.DataFrame({"메시지": ["감사로그 없음"]}) print("✅ 대시보드 생성 완료\n") return ( status_text, mro_dashboard, mro_workflow, proc_dashboard, proc_workflow, exec_dashboard, action_items, offers_df, inv_df, opt_result, audit_df, item_name ) 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', '')} """ equipment_value = preset["equipment_id"] item_value = preset["item_id"] if preset["item_id"] else "" return ( equipment_value, item_value, preset["demand_qty"], guide_text ) # ========================================================= # 14. Gradio UI (허깅페이스 최적화) # ========================================================= print("🎨 Gradio UI 구성 중...\n") # 드롭다운 옵션 생성 demo_tables = generate_demo_tables(7) equipment_options = [] for _, row in demo_tables["equipment"].iterrows(): equipment_options.append(f"{row['equipment_id']} - {row['equipment_name']}") item_options = ["(자동 선택)"] for _, row in demo_tables["items"].iterrows(): item_options.append(f"{row['item_id']} - {row['item_name']}") 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 운영 → 구매/조달 → 경영진 승인 ### ✨ 특징 - ✅ Knowledge Graph 기반 설비-부품 매칭 - ✅ Neuro-Symbolic AI 규정 검증 - ✅ Linear Programming 비용 최적화 - ✅ GPT-4 기반 전략 분석 - ✅ 실시간 대시보드 ### 🔑 API 키 설정 OpenAI API 키를 사용하려면 Hugging Face Space의 Settings → Secrets에서 `OPENAI_API_KEY`를 추가하세요. """) with gr.Accordion("📖 전체 업무 프로세스 개요", open=False): gr.Markdown(""" ### 🔄 End-to-End 워크플로우 ``` ┌─────────────────┐ ┌──────────────────┐ ┌─────────────────┐ │ 1️⃣ MRO 운영 │ ───> │ 2️⃣ 구매/조달 │ ───> │ 3️⃣ 경영진 승인 │ │ │ │ │ │ │ │ • 고장 접수 │ │ • 공급업체 조회 │ │ • KPI 확인 │ │ • 부품 확인 │ │ • 규정 검증 │ │ • 의사결정 │ │ • 재고 확인 │ │ • 최적화 분석 │ │ • 피드백 │ │ • 발주 요청 │ │ • 승인 요청 │ │ │ └─────────────────┘ └──────────────────┘ └─────────────────┘ ⏱️ 15분 ⏱️ 25분 ⏱️ 25분 ``` ### 💡 핵심 가치 1. **자동화**: 설비-부품 매칭, 재고 조회, 규정 검증 2. **최적화**: AI 기반 비용 최적화 3. **검증**: 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("### ⚙️ 파라미터") equipment_dropdown = gr.Dropdown( choices=equipment_options, value=equipment_options[0], label="설비 선택 (ID - 명칭)", info="설비를 선택하세요" ) item_dropdown = gr.Dropdown( choices=item_options, value=item_options[0], label="품목 선택 (ID - 명칭)", info="비워두면 자동 선택" ) demand_number = gr.Number(value=10, label="수량", precision=0, minimum=1) 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 담당자"): with gr.Accordion("📋 MRO 운영 프로세스 가이드", open=True): gr.HTML(create_process_guide_html("mro")) gr.Markdown("### 📊 대시보드 및 분석 결과") mro_dashboard_plot = gr.Plot(label="📦 재고 분석 대시보드") mro_workflow_plot = gr.Plot(label="🔄 MRO 워크플로우 진행") mro_inventory_table = gr.Dataframe(label="📋 상세 재고 데이터") with gr.Tab("💰 구매 담당자"): with gr.Accordion("📋 구매/조달 프로세스 가이드", open=True): gr.HTML(create_process_guide_html("procurement")) gr.Markdown("### 📊 대시보드 및 분석 결과") proc_dashboard_plot = gr.Plot(label="📊 공급업체 비교 대시보드") proc_workflow_plot = gr.Plot(label="🔄 구매 워크플로우") proc_offers_table = gr.Dataframe(label="📋 공급업체 상세 정보") with gr.Tab("👔 경영진"): with gr.Accordion("📋 경영진 의사결정 프로세스 가이드", open=True): gr.HTML(create_process_guide_html("executive")) gr.Markdown("### 📊 대시보드 및 분석 결과") exec_dashboard_plot = gr.Plot(label="📊 경영진 KPI 대시보드") exec_action_items_table = gr.Dataframe(label="📋 Action Items") gr.Markdown("### 💬 경영진 피드백") feedback_text = gr.Textbox(label="개선 제안 / 피드백", 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="피드백 처리 결과", interactive=False) with gr.Tab("📝 감사 로그"): gr.Markdown(""" ### 📝 감사 추적 모든 AI 의사결정 과정과 도구 호출 이력을 추적합니다. """) audit_log_table = gr.Dataframe(label="📝 전체 감사 로그") opt_result_display = gr.JSON(label="🔍 최적화 상세 결과") # 이벤트 핸들러 scenario_radio.change( fn=update_scenario, inputs=[scenario_radio], outputs=[equipment_dropdown, item_dropdown, demand_number, scenario_info] ) run_button.click( fn=run_analysis, inputs=[scenario_radio, equipment_dropdown, item_dropdown, demand_number], outputs=[ status_output, mro_dashboard_plot, mro_workflow_plot, proc_dashboard_plot, proc_workflow_plot, exec_dashboard_plot, exec_action_items_table, proc_offers_table, mro_inventory_table, opt_result_display, audit_log_table, selected_item_display ] ) def process_feedback(feedback_text: str, action: str) -> str: timestamp = now_ts() return f"✅ {action} 완료: {timestamp}\n피드백: {feedback_text if feedback_text else '(없음)'}" approve_btn.click( fn=lambda fb: process_feedback(fb, "승인"), inputs=[feedback_text], outputs=[feedback_output] ) reject_btn.click( fn=lambda fb: process_feedback(fb, "반려"), inputs=[feedback_text], outputs=[feedback_output] ) suggest_btn.click( fn=lambda fb: process_feedback(fb, "개선 제안"), inputs=[feedback_text], outputs=[feedback_output] ) # ========================================================= # 15. 허깅페이스 스페이스 실행 # ========================================================= if __name__ == "__main__": print("\n" + "=" * 60) print("🚀 허깅페이스 스페이스에서 Gradio 서버 시작...") print("=" * 60 + "\n") # ✅ 허깅페이스 스페이스 전용 설정 demo.launch( server_name="0.0.0.0", # 모든 IP에서 접근 허용 server_port=7860, # 허깅페이스 기본 포트 share=False, # 허깅페이스는 자체 URL 제공 show_error=True )