"""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()