FLOPI

report_problem Problem: 수동 분석의 한계

핵심 문제 3가지:

• 대응 지연 — 야간/주말 이상은 다음 근무일까지 미분석 상태로 방치
• 반복 수동 작업 — 동일 유형 이상에 대해 매번 같은 분석을 사람이 수행
• 지식 단절 — 이상 발생 시점의 데이터 컨텍스트가 시간이 지나면 소실

visibility Vision: 이상 감지부터 분석까지 자동화

architecture Architecture

기존 FLOPI 아키텍처에 Auto RCA Scheduler를 추가한다. Detection 엔진과 완전히 분리된 별도 스케줄러로 동작하여, 감지 성능에 영향을 주지 않는다.

설계 원칙

psychology RCA Studio — AI Analyst

Analyst 클래스 — 대화형 분석

rca/analyst.py의 Analyst 클래스는 LLM과의 대화를 관리하며, 엔지니어가 설명하는 분석 과정을 실시간으로 플로차트 노드로 추출합니다.

class Analyst:
    """대화를 통해 엔지니어의 암묵지를 추출하는 AI 분석가."""

    def _build_context(self, scenarios_context, tools_context, existing_nodes):
        # 시스템 프롬프트 + 사용 가능한 도구 + 참고 시나리오 + 현재 노드
        context = ANALYST_SYSTEM
        if tools_context:
            context += f"\n\n## 사용 가능한 도구\n{tools_context}"
        if scenarios_context:
            context += f"\n\n## 참고 시나리오\n{scenarios_context}"
        if existing_nodes:
            context += f"\n\n## 현재 추출된 노드\n{json.dumps(existing_nodes)}"
        return context

    async def analyze_stream(self, messages, existing_nodes, ...):
        """스트리밍 응답 — 토큰 단위로 yield."""
        set_llm_module("deep_rca")
        ...

    async def validate_graph(self, nodes, edges):
        """플로우차트 전체 검증 — 빠진 분기, 고립 노드, 불일치 탐지."""

대화 규칙 (ANALYST_SYSTEM 프롬프트)

노드 타입 (RCA 플로차트)

transform RCA → Workflow Converter

rca/workflow_converter.py의 RCAWorkflowConverter 클래스는 RCA Studio에서 생성된 분석 시나리오(check/decision/action/end 노드)를 AI 워크플로우 빌더의 12종 노드로 LLM 기반 지능 변환합니다.

class RCAWorkflowConverter:
    """LLM 기반 RCA → 워크플로우 변환기."""

    async def convert(self, nodes, edges, tools,
                       trigger_type="manual", scenario_title="", process_type=""):
        """RCA 시나리오를 워크플로우로 변환."""
        # 1. LLM에 RCA 노드/엣지 + 사용 가능한 도구 전달
        # 2. LLM이 워크플로우 graph_data (12종 노드) 생성
        # 3. 후처리 검증 + ConversionResult 반환

ConversionResult 구조

{
  "graph_data": {"nodes": [...], "edges": [...]},  // 워크플로우 그래프
  "warnings": [...],          // 변환 경고 (매핑 불확실한 노드)
  "mapping_log": [...],       // RCA → WF 매핑 추적 로그
  "suggested_name": "...",    // 워크플로우 제안 이름
  "suggested_description": "..."  // 워크플로우 제안 설명
}

UI 진입점

• RCA Studio 플로차트 패널 → "워크플로우 변환" 버튼
• 시나리오 카드 → "워크플로우" 버튼
• 변환 프리뷰 확인 후 워크플로우 빌더에 생성

science Sample Data (FDC / SPC / Alarm)

RCA Studio는 시뮬레이터/데모 환경에서 반도체 FAB 공정별 샘플 데이터를 제공합니다. rca/sample_data/ 디렉토리에 3종의 데이터가 구현되어 있습니다.

link Rule-RCA Mapping

핵심 아이디어: 규칙(Rule)마다 "이 이상이 발생하면 어떤 방식으로 RCA를 할 것인가"를 설정한다. 이를 통해 물류 이상에는 물류 전문 분석을, 설비 이상에는 설비 전문 분석을 자동으로 적용할 수 있다.

2계층 설정 구조

schedule RCA Scheduler

Detection 엔진과 독립적으로 동작하는 별도 APScheduler 잡. DB를 폴링하여 rca_status = 'pending'인 이상을 찾아 분석을 실행한다.

스케줄러 의사코드

# detection/auto_rca.py (신규)

async def poll_and_analyze():
    """30초마다 실행되는 Auto RCA 폴링 잡"""

    # 1. pending 이상 조회 (오래된 것부터)
    anomalies = await queries.get_pending_rca_anomalies(limit=5)

    for anomaly in anomalies:
        # 2. 규칙의 RCA 설정 확인
        rule = await queries.get_rule(anomaly["rule_id"])

        if not rule["auto_rca"]:
            continue

        # 3. severity 조건 확인
        if not _severity_matches(anomaly, rule):
            continue

        # 4. 쿨다운 확인 (같은 rule_id 5분 이내 중복 방지)
        if _in_cooldown(rule["rule_id"]):
            continue

        # 5. RCA 에이전트 실행 (비동기)
        await sem.acquire()
        asyncio.create_task(_run_rca(anomaly, rule, sem))


async def _run_rca(anomaly, rule, sem):
    try:
        # rca_status → 'running'
        await queries.update_rca_status(anomaly["anomaly_id"], "running")

        # 프롬프트/도구 결정 (규칙 → 카테고리 기본값 fallback)
        prompt = rule["rca_prompt"] or CATEGORY_DEFAULTS[rule["category"]]["prompt"]
        tools  = rule["rca_tools"]  or CATEGORY_DEFAULTS[rule["category"]]["tools"]

        # ReAct RCA 에이전트 실행
        result = await run_rca_agent(anomaly, prompt, tools,
                                      max_rounds=rule.get("rca_max_rounds", 5))

        # rca_analyses 테이블에 저장
        await queries.insert_rca_analysis(anomaly["anomaly_id"], result)

        # rca_status → 'done'
        await queries.update_rca_status(anomaly["anomaly_id"], "done")

    except Exception:
        await queries.update_rca_status(anomaly["anomaly_id"], "failed")
    finally:
        sem.release()

smart_toy RCA Agent

기존 Detection Agent(detection/agents.py)의 ReAct 루프를 재활용하되, RCA 전용 시스템 프롬프트와 규칙별 도구셋으로 동작한다.

{
  "root_cause":       "컨베이어 C3 → EQ-PHO-02 구간 캐리어 적체",
  "confidence":       0.85,
  "severity":         "critical",
  "evidence": [
    "C3 부하율 97% (정상 대비 +42%)",
    "EQ-PHO-02 DOWN 상태 40분 지속",
    "상류 큐 대기 12건 (평균 3건)"
  ],
  "affected_scope":   "LINE-B / PHOTO 공정",
  "recommendation": [
    "EQ-PHO-02 즉시 PM 점검",
    "C3 → C4 우회 반송 경로 활성화",
    "WIP 유입 속도 일시 조절 (30% 감속)"
  ],
  "tool_calls_used": [
    "get_conveyor_load",
    "get_equipment_status",
    "get_queue_length"
  ]
}

storage DB Schema Changes

1. detection_rules 테이블 (ALTER)

-- 규칙별 Auto RCA 설정 필드 추가
ALTER TABLE detection_rules ADD (
  auto_rca          NUMBER(1)    DEFAULT 0,         -- 자동 RCA 활성화
  rca_severity      VARCHAR2(20) DEFAULT 'critical', -- 트리거 조건
  rca_prompt        CLOB,                               -- RCA 분석 프롬프트
  rca_tools         CLOB,                               -- 도구 목록 (JSON)
  rca_max_rounds    NUMBER       DEFAULT 5           -- ReAct 최대 라운드
);

2. anomalies 테이블 (ALTER)

-- RCA 상태 추적 컬럼 추가
ALTER TABLE anomalies ADD (
  rca_status        VARCHAR2(20) DEFAULT 'none',    -- none/pending/running/done/failed
  rca_started_at    TIMESTAMP,                        -- RCA 시작 시각
  rca_completed_at  TIMESTAMP                         -- RCA 완료 시각
);

3. rca_analyses 테이블 (기존 — 활성화)

-- 이미 schema.sql에 정의됨, 미사용 상태 → 활성화
CREATE TABLE rca_analyses (
  rca_id            NUMBER PRIMARY KEY,
  anomaly_id        NUMBER REFERENCES anomalies,
  root_cause        CLOB,                               -- 근본 원인 요약
  confidence        NUMBER(3,2),                        -- 0.00 ~ 1.00
  evidence          CLOB,                               -- 근거 목록 (JSON)
  recommendation    CLOB,                               -- 조치 제안 (JSON)
  affected_scope    VARCHAR2(200),                     -- 영향 범위
  tool_calls_used   CLOB,                               -- 사용된 도구 (JSON)
  duration_ms       NUMBER,                             -- 분석 소요 시간
  created_at        TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

description Rule YAML Spec

rules.yaml에 Auto RCA 필드를 추가한 전체 예시. 기존 필드는 그대로 유지하고, RCA 관련 필드만 하단에 추가된다.

category Category Defaults

규칙에 rca_prompt/rca_tools가 없으면, 카테고리별 기본 템플릿이 자동 적용된다.

dashboard UI Changes

1. 규칙 관리 페이지 — "자동 RCA" 탭 추가

기존 2탭(조건감시 / AI판단) 구조에 3번째 탭 "자동 RCA"를 추가한다.

┌──────────────────────────────────────────────────┐
│  [조건 감시]  [AI 판단]  [자동 RCA]                │
├──────────────────────────────────────────────────┤
│                                                  │
│  자동 RCA 활성화    [● ON]                       │
│                                                  │
│  트리거 조건         [Critical ▾]                │
│                      Critical / Warning / All    │
│                                                  │
│  RCA 프롬프트       ┌────────────────────────┐   │
│  (비워두면 카테고리  │ 컨베이어 부하율 과부하의│   │
│   기본값 적용)      │ 근본 원인을 분석하세요. │   │
│                     │ 1. 반송 이력 확인      │   │
│                     │ 2. 설비 상태 확인      │   │
│                     └────────────────────────┘   │
│                                                  │
│  RCA 도구            ☑ get_conveyor_load         │
│  (비워두면 카테고리   ☑ get_bottleneck_zones      │
│   기본값 적용)       ☑ get_equipment_status       │
│                      ☐ get_wip_levels            │
│                      ☐ get_aging_lots             │
│                                                  │
│  최대 분석 라운드     [5]                        │
│                                                  │
└──────────────────────────────────────────────────┘

2. 이상 목록 페이지 — RCA 상태 표시

이상 목록 테이블에 RCA 상태 뱃지를 추가한다.

3. 이상 상세 다이얼로그 — RCA 결과 섹션

Auto RCA 완료 시, 이상 상세 다이얼로그에 "AI 자동 분석 결과" 카드가 추가된다. 근본 원인 + 근거 + 조치 제안 + 사용된 도구 + 소요 시간 표시.

api API Endpoints

timeline Implementation Phases

shield Risks & Mitigations