"""Tier 1~4 순차 등장 렌더러 (개발 가이드 8·9·10장) st.empty() placeholder 4개 + time.sleep으로 스켈레톤 → 콘텐츠 순차 교체 이미 완료된 상태(stage>=5)거나 animation_pending이 아닐 때는 즉시 정적 렌더 ★ Tier 데이터는 반드시 core.pipeline.get_tier_data() 한 함수로만 받음 data.demo.TIER_DATA를 직접 import 하지 말 것 (실제 에이전트 교체가 막힘) """ import time from datetime import datetime import streamlit as st from components.skeleton import skeleton_html from components.tiers_body import ( tier_1_body_html, tier_2_body_html, tier_3_body_html, tier_4_body_html, ) from core.pipeline import get_tier_data SPEED_PRESETS = { "fast": {"skel": 0.6, "hold": 0.4}, "normal": {"skel": 1.4, "hold": 0.7}, "real": {"skel": 3.0, "hold": 1.5}, } TIER_NAMES = {1: "이상 탐지", 2: "원인 분석", 3: "공정 간 영향 평가", 4: "대응 권고"} # 각 Tier의 에이전트 페르소나 + 활동 요약 (loading=False일 때만 활성) AGENT_PERSONA = { 1: {"name": "Detection Agent", "tool": "IsolationForest · SECOM/PHM"}, 2: {"name": "Cause Agent", "tool": "GPT-5-mini · RAG 검색"}, 3: {"name": "Impact Agent", "tool": "GPT-5-mini · RAG + WIP 결합"}, 4: {"name": "Response Agent", "tool": "GPT-5-mini · RAG + 근거 추출"}, } def _agent_summary(tier_num: int, data) -> str: """Tier별 에이전트 활동 1줄 요약, data가 None이면 분석 중 메시지""" if data is None: return "분석 중..." if tier_num == 1: return f"{len(data['features'])}개 기여 센서 검출 (이상 점수 {data['score']})" if tier_num == 2: cites = sum(len(c["citations"]) for c in data["causes"]) return f"사내 문서 {cites}건 인용해 원인 {len(data['causes'])}개 추정" if tier_num == 3: return f"의존성 + WIP 결합해 예상 수율 손실 {data['yield_loss']} %p 추정" if tier_num == 4: return f"근거 {len(data['refs'])}건 인용해 즉시 {len(data['immediate'])}건 · 중장기 {len(data['longterm'])}건 권고" return "" BODY_BUILDERS = { 1: tier_1_body_html, 2: tier_2_body_html, 3: tier_3_body_html, 4: tier_4_body_html, } LOADING_HTML = """

4-Tier 멀티 에이전트 분석 진행 중

알람을 분석하기 위해 4단계 에이전트가 순차 실행됩니다
Tier 1 이상 탐지 Tier 2 원인 분석 Tier 3 영향 평가 Tier 4 대응 권고
첫 호출은 LLM 3회 직렬로 약 60초 소요됩니다 · 이후 동일 알람은 캐시로 즉시 응답
""" def render_tier_cascade(): ss = st.session_state # 첫 호출 시 LLM 3회 직렬로 약 60초, 사용자 안내를 위해 로딩 카드 먼저 렌더 loading_slot = st.empty() loading_slot.html(LOADING_HTML) try: data = get_tier_data(ss.selected_alarm_id) finally: loading_slot.empty() if data is None: st.markdown( """

분석 데이터가 없습니다

선택한 알람에 매핑된 분석 결과가 없습니다

""", unsafe_allow_html=True, ) return if ss.animation_pending: ss.animation_pending = False _run_sequence(data) return # 정적 렌더 (재실행/새로고침 대응) for t in range(1, 5): render_tier(t, data[f"tier{t}"], loading=False, with_actions=(t == 4)) def _run_sequence(data): ss = st.session_state timing = SPEED_PRESETS[ss.speed] slots = [st.empty() for _ in range(4)] for tier_idx in range(1, 5): ss.stage = tier_idx with slots[tier_idx - 1].container(): render_tier(tier_idx, None, loading=True) time.sleep(timing["skel"]) with slots[tier_idx - 1].container(): render_tier( tier_idx, data[f"tier{tier_idx}"], loading=False, with_actions=(tier_idx == 4), ) ss.completed_tiers.add(tier_idx) if tier_idx < 4: time.sleep(timing["hold"]) ss.stage = 5 def render_tier(tier_num: int, data, loading: bool, with_actions: bool = False): """Tier 카드 한 덩어리를 단일 markdown으로 렌더, Tier 4 액션 바만 별도 위젯""" status_html = ( '
실행 중...
' if loading else '
완료
' ) body_html = skeleton_html(tier_num) if loading else BODY_BUILDERS[tier_num](data) p = AGENT_PERSONA[tier_num] summary = _agent_summary(tier_num, None if loading else data) st.html( f'
' f'
' f'
' f'
' f'
' f'{p["name"]}' f'Tier {tier_num} · {TIER_NAMES[tier_num]}' f'
' f'
{summary} · {p["tool"]}
' f'
' f'
' f'{status_html}' f'
' f'
{body_html}
' f'
' ) if tier_num == 4 and with_actions and not loading: _render_action_bar() def _render_action_bar(): ss = st.session_state st.markdown( '
' '운영자 결정' '권고서를 검토하고 액션을 선택하세요' '
', unsafe_allow_html=True, ) # 양쪽 spacer로 가운데 정렬 _, c1, c2, c3, _ = st.columns([2.5, 1, 1, 2.2, 2.5]) with c1: if st.button("거절", key="btn-reject", disabled=ss.approved, use_container_width=True): _on_reject() with c2: if st.button("보류", key="btn-hold", disabled=ss.approved, use_container_width=True): _on_hold() with c3: label = "작업지시서 생성됨" if ss.approved else "승인 및 작업지시서 생성" if st.button(label, key="btn-approve", disabled=ss.approved, type="primary", use_container_width=True): _on_approve() _render_action_result() def _render_action_result(): ss = st.session_state act = ss.get("last_action") if not act: return kind = act["type"] icon = {"approved": "✓", "held": "⏸", "rejected": "✗"}[kind] title = act["title"] meta = act["meta"] st.html( f'
' f'
{icon}
' f'
' f'
{title}
' f'
{meta}
' f'
' f'
' ) def _on_reject(): st.session_state.last_action = { "type": "rejected", "title": "권고 거절", "meta": "사유 수집 모달은 MVP 범위 외, 후속 분석에 반영하려면 사유 입력 후 인시던트 DB에 기록 필요", } st.toast("권고 거절됨", icon="❌") st.rerun() def _on_hold(): st.session_state.last_action = { "type": "held", "title": "권고 보류", "meta": "5분 후 동일 알람이 재발생 예정, 추가 데이터 확보 후 재검토 권장", } st.toast("권고 보류됨", icon="⏸️") st.rerun() def _on_approve(): ss = st.session_state ss.approved = True # 자가 학습 - 현재 알람의 분석 결과를 knowledge로 자동 기록 alarm = next((a for a in ss.alarms if a["id"] == ss.selected_alarm_id), None) work_order = f"W-{datetime.now():%Y%m%d}-{ss.selected_alarm_id[1:].zfill(3)}" incident_doc = "" if alarm: from agents.rag.learn import record_incident tier_data = get_tier_data(ss.selected_alarm_id) if tier_data: path = record_incident(alarm, tier_data, work_order) incident_doc = path.name # 알람 상태 업데이트 for a in ss.alarms: if a["id"] == ss.selected_alarm_id: a["status"] = "done" a["time"] = "방금 전" break ss.last_action = { "type": "approved", "title": "승인 완료, 작업지시서 발행", "meta": ( f"작업지시서 {work_order} 생성 · " f"인시던트 DB {incident_doc} 자동 기록 · " "사이드바 알람 상태가 [완료]로 전환되었습니다" ), } st.toast( f"✓ 작업 지시서 {work_order} 생성 · 인시던트 DB 자동 기록", icon="✅", ) st.rerun()