FLOPI Modules

반도체 FAB 폐쇄망 환경의 통합 AI 플랫폼.
각 모듈의 목적, 필요성, 아키텍처 결정을 기록합니다.

precision_manufacturing FAB AI Platform widgets 20개 모듈 code Python 3.11 calendar_today 2026-03-26
20
Modules
20
UI Pages
170+
API Endpoints
89
Built-in Tools
12
Node Types
4
User Roles
dashboard
Dashboard
실시간 KPI, 이상 현황, 탐지 사이클 요약
sensors
이상탐지 엔진
틱 기반 규칙 평가 + LLM 에이전트 분석
rule
규칙 엔진
DB-first 규칙 관리, 89개 내장 FAB 도구
build
Tool Studio
No-Code SQL 도구 빌더 + MCP 브릿지
account_tree
RCA Studio
AI 대화 → Mermaid 원인분석 플로차트
route
Workflow
비주얼 DAG 파이프라인 빌더
menu_book
Knowledge Base
RAG 파이프라인 — 청킹 + 임베딩 + 시맨틱 검색
chat
AI Chat
SSE 스트리밍 + 스마트 도구 선택 + 코드 인터프리터
play_circle
Chat Scenarios
키워드 매칭 매크로 — 도구 체인 자동 실행
shield
Countermeasures
이상 유형별 대응 절차 플레이북
compare_arrows
Migration
블록별 이관 관리 + 스키마 비교 + DDL 생성

architecture 전체 아키텍처

왜 단일 앱인가

반도체 FAB 폐쇄망에서는 마이크로서비스 배포가 현실적으로 어렵습니다. 네트워크 제약, 인프라 관리 인력 부족, 컨테이너 환경 미지원 등의 이유로 모노리스 단일 앱을 선택했습니다. 대신 내부를 모듈 단위로 명확히 분리하여 코드 결합도를 낮췄습니다.

3-Tier 구조

FastAPI :8600 ─── REST API + SSE + APScheduler │ ├── detection/ 이상탐지 엔진 (스케줄러 + 에이전트) ├── data_studio/ Tool Studio (SQL 빌더 + 캐시 + 실행) ├── rca/ 딥 RCA (AI 분석가 + 구조화) ├── core/workflow/ 워크플로우 DAG 엔진 ├── core/rag/ RAG (청킹 + 임베딩 + 검색) ├── core/llm/ LLM 클라이언트 + 사용량 추적 └── core/db/ Oracle/SQLite 이중 백엔드 NiceGUI :18080 ─── 통합 대시보드 (20개 페이지) └── REST API 호출로 백엔드 연동 MCP Server ─── stdio transport (Claude subprocess) └── Tool Studio 도구를 MCP로 노출

기술 선택 근거

선택이유대안
FastAPIasync/await, 자동 OpenAPI docs, SSE 지원Flask, Django (sync 기반)
NiceGUIPython만으로 풀스택 UI, React 학습 불필요Streamlit (새로고침 모델), Gradio (ML 특화)
APSchedulerin-process 스케줄러, interval/cron 지원Celery (overkill), cron (유연성 부족)
ChromaDB로컬 파일 벡터 저장소, 외부 서버 불필요Milvus (클러스터 필요), Pinecone (클라우드)
OpenAI-compatible API모델 교체 용이 (Gemini/Ollama/사내)각 벤더 SDK 직접 사용
Oracle + SQLite 이중운영(Oracle) + 개발/시뮬(SQLite) 즉시 전환Oracle only (개발 환경 제약)

전체 파일 맵

FLOPI/
├── main.py              # FastAPI 진입점 + APScheduler (655 lines)
├── config.py            # 통합 설정 — 환경변수 기반 (195 lines)
├── init_db.py           # DB 초기화 + admin 계정 + 마이그레이션 (2,202 lines)
├── mcp_server.py        # MCP 서버 — Claude 연동 (132 lines)
├── rules.yaml           # 탐지 규칙 seed (YAML)
├── seed.json            # Tool Studio 도구 seed
│
├── api/                 # REST API 라우터 (22 files, ~5,600 lines, 20 routers)
│   ├── anomalies.py     #   이상 CRUD + 상태 관리
│   ├── dashboard.py     #   대시보드 KPI 집계
│   ├── rules.py         #   규칙 CRUD + 테스트 + 감사
│   ├── system.py        #   시스템 상태 + 수동 감지
│   ├── system_settings.py #  런타임 설정 관리
│   ├── data_studio.py   #   Tool Studio CRUD + 실행
│   ├── deep_rca.py      #   딥 RCA 세션 + 시나리오
│   ├── workflow.py      #   워크플로우 CRUD + 실행
│   ├── knowledge_base.py #  KB 문서 CRUD + 검색
│   ├── ai_chat.py       #   AI 대화 (SSE 스트리밍)
│   ├── llm_usage.py     #   LLM 사용량 통계
│   ├── users.py         #   사용자 CRUD + 인증
│   ├── rca.py           #   이상별 RCA 조회
│   ├── chat_scenarios.py #  챗 매크로 CRUD
│   ├── user_prompts.py  #   사용자 개인 프롬프트
│   ├── countermeasures.py # 대안/조치 플레이북 CRUD
│   ├── migration.py     #   이관 관리 (블록별 상태 + 스키마 비교)
│   ├── activity_log.py  #   활동 로그 조회
│   ├── question_flows.py #  질문 흐름 CRUD
│   ├── tool_requests.py #   도구 요청 CRUD
│   └── deps.py          #   공통 의존성 (인증, 권한 검사)
│
├── core/                # 공통 인프라 (47 files, ~7,300 lines)
│   ├── auth/            #   인증 + 권한 (permissions.py)
│   ├── db/              #   DB 레이어 (oracle.py + queries/ 16개 쿼리 모듈)
│   ├── llm/             #   LLM 클라이언트 + 도구 레지스트리 + 에이전트 루프
│   ├── rag/             #   RAG 파이프라인 (청킹 + 임베딩 + 검색 + 구조화)
│   ├── simulator/       #   SQLite 시뮬레이터 (monkey-patch)
│   ├── workflow/        #   DAG 실행 엔진 + AI 워크플로우 생성기
│   ├── migration/       #   DB 마이그레이션 (checker.py + router.py)
│   ├── audit.py         #   활동 감사 로그 헬퍼
│   ├── code_interpreter.py # Python 코드 실행 엔진 (AI Chat용)
│   └── settings_manager.py # 시스템 설정 인메모리 캐시
│
├── detection/           # 이상탐지 엔진 (14 files, ~1,300 lines)
│   ├── scheduler.py     #   탐지 사이클 오케스트레이터
│   ├── evaluator.py     #   규칙 평가 + 이상 생성
│   ├── lifecycle.py     #   DetectionLifecycleManager
│   ├── agents.py        #   LLM 에이전트 분석
│   ├── resolver.py      #   이상 자동 해소 (auto-resolve)
│   ├── prompts.py       #   이상탐지 시스템 프롬프트
│   ├── rules/           #   규칙 로더 + 엔진 + 모델
│   └── tools/           #   Tool Studio 브릿지
│
├── rca/                 # 딥 RCA (12 files, ~3,000 lines)
│   ├── workflow_converter.py # RCA → 워크플로우 변환
│   └── sample_data/     #   SPC/FDC/알람 테스트 데이터
│
├── data_studio/         # Tool Studio (8 files, ~2,100 lines)
│   └── pages/           #   Data Studio 전용 페이지
│
└── ui/                  # NiceGUI 대시보드 (29 files, ~16,100 lines)
    └── pages/           #   20개 페이지

모듈별 규모 요약

모듈파일 수코드 라인핵심 역할
api/22~5,600170+ REST 엔드포인트 (20 라우터)
core/47~7,300DB, LLM, RAG, Auth, Simulator, Workflow, Migration, Audit, Code Interpreter, Settings
detection/14~1,300규칙 평가 + LLM 에이전트 분석 + 자동 해소
rca/12~3,000AI 대화 → Mermaid 플로차트 + 워크플로우 변환
data_studio/8~2,100No-Code SQL 도구 빌더
ui/29~16,100NiceGUI 대시보드 (20 페이지)
진입점4~3,200main.py, config.py, init_db.py, mcp_server.py
합계140~39,400

dashboard Dashboard

왜 필요한가

FAB 현장 관리자는 한 화면에서 전체 상황을 1초 안에 파악해야 합니다. 이상이 몇 건인지, 심각도는 어떤지, 마지막 탐지 사이클은 정상이었는지를 여러 페이지를 돌아다니며 확인하는 것은 비효율적입니다.

구성 요소

bar_chart
4개 KPI 카드
활성 위험 / 활성 경고 / 총 이상(24h) / 활성 규칙

활성 위험: critical severity + detected/in_progress 상태인 이상 수.
활성 경고: warning severity 이상 수.
총 이상: 24시간 이내 발생한 모든 이상.
활성 규칙: enabled=1인 detection_rules 수.

donut_large
상태 분포 도넛 차트
Plotly.js — 감지됨/처리중/해결

빨강(감지됨) / 노랑(처리중) / 초록(해결) 3색 도넛. 다크/라이트 모드에 따라 배경색을 동적으로 적용합니다.

timer
마지막 탐지 사이클
규칙 평가수 / 이상 감지수 / 소요시간

detection_cycles 테이블의 최신 1건을 조회하여 표시. operator 이상 역할에게만 "수동 감지 실행" 버튼을 노출합니다.

설계 결정

파일 구조

파일역할
api/dashboard.py3개 API 엔드포인트 (overview, timeline, heatmap)
ui/pages/dashboard.pyNiceGUI 대시보드 UI (KPI 카드 + 차트 + 최근 이상)
core/db/queries/dashboard.py집계 쿼리 (severity 분류, 24h 타임라인)

API

MethodPath설명
GET/api/dashboard/overviewKPI 집계 + 마지막 사이클 + severity별 분류
GET/api/dashboard/timeline시간별 이상 발생 타임라인 (?hours=24)
GET/api/dashboard/heatmap카테고리 × 심각도 히트맵

sensors 이상탐지 엔진

왜 필요한가

FAB MES 데이터에서 자동으로 이상을 탐지해야 합니다. 사람이 24시간 모니터링하는 것은 불가능하고, 단순 임계치 비교만으로는 복잡한 패턴(예: 서서히 증가하는 부하율)을 잡아내기 어렵습니다. 임계치 기반 빠른 탐지 + LLM 기반 정밀 분석을 공존시켜야 합니다.

핵심 설계: 틱 기반 평가

lightbulb
핵심 아이디어: 글로벌 인터벌이 아닌 규칙별 인터벌. 매 60초 틱마다 "평가 시점이 된 규칙만" 골라서 실행합니다. 컨베이어 부하는 5분마다, WIP 레벨은 10분마다 — 하나의 스케줄러로 모두 처리.
APScheduler (tick_sec=60초마다) └── run_detection_cycle() ├── get_due_rules() ← eval_interval 경과한 규칙만 선별 ├── asyncio.gather() ← 병렬 평가 (max_concurrent 세마포어) │ └── evaluate_rule() ← SQL 또는 Tool 실행 + 임계치 비교 │ └── evaluate_and_detect() │ ├── llm_enabled=True → analyze_and_save() (LLM 에이전트) │ └── llm_enabled=False → analyze_without_llm() (임계치만) ├── mark_evaluated() ← last_evaluated_at 갱신 └── complete_cycle() ← 사이클 로그 기록

4가지 체크 타입

check_type동작사용 사례
threshold측정값 vs 임계치 비교 (>, <, ≥, ≤)컨베이어 부하율 초과
delta변화량 절댓값 vs 임계치급격한 온도 변화
absence쿼리 결과가 비어있으면 이상데이터 수집 중단 감지
llm데이터만 가져오고 LLM이 최종 판단AGV 가동률 패턴 분석

LLM 에이전트 분석

llm_enabled=true인 규칙은 ReAct 에이전트 루프를 거칩니다. 최대 max_agent_rounds(기본 3)회 LLM ↔ 도구 호출을 반복하며, JSON 형식으로 {is_anomaly, confidence, severity, title, analysis}를 반환합니다.

check_circle
이상 중복 방지: 동일 규칙의 미해결 이상이 이미 존재하면 새로 생성하지 않고 occurrence_count를 증가시킵니다. FAB에서 같은 문제가 매 틱마다 탐지되는 노이즈를 억제합니다.
check_circle
신뢰도 필터: confidence < 0.7이면 is_anomaly=false로 처리하여 오탐을 줄입니다.

데이터 소스 분기

라이프사이클 관리

DetectionLifecycleManager가 APScheduler 잡 등록/해제를 캡슐화합니다. start(), update_interval(), run_once()(수동 트리거)를 제공하여 런타임 중 탐지 주기를 변경하거나 즉시 실행할 수 있습니다.

파일 구조

detection/
├── lifecycle.py      # DetectionLifecycleManager (start/stop/update_interval)
├── scheduler.py      # run_detection_cycle() 오케스트레이터
├── evaluator.py      # 규칙 평가 + 이상 생성
├── agents.py         # LLM 에이전트 분석 (analyze_and_save)
├── resolver.py       # 이상 자동 해소 (auto-resolve)
├── prompts.py        # 이상탐지 시스템 프롬프트
├── rules/
│   ├── models.py     # Pydantic — RuleCreate/RuleUpdate
│   ├── loader.py     # YAML ↔ DB 동기화 (DB-first)
│   └── engine.py     # 4가지 체크 타입 평가
└── tools/
    └── ds_bridge.py  # Tool Studio 브릿지 (sync_ds_tools)

Auto-Resolve (자동 해소) v1.7.0

check_circle
연속 정상 시 자동 해소: detection/resolver.pycheck_and_auto_resolve()가 매 탐지 사이클마다 활성 이상 중 auto_resolve=1인 규칙의 이상을 재평가합니다. 위반이 아니면 normal_streak을 증가시키고, normal_streak ≥ resolve_count(기본 3)이면 자동으로 resolved 상태로 전환합니다. 위반이 발생하면 streak이 0으로 리셋됩니다.
run_detection_cycle() └── check_and_auto_resolve() ├── 활성 이상 중 auto_resolve=1 조회 ├── 각 이상의 규칙을 evaluate_rule()로 재평가 │ ├── 위반 아님 → normal_streak + 1 │ ├── 위반임 → normal_streak = 0 리셋 │ └── streak ≥ resolve_count → auto resolved └── 결과: {checked: N, resolved: N, reset: N}

API 엔드포인트

MethodPath설명
GET/api/anomalies이상 목록 (status, rule_id, limit, offset 필터)
GET/api/anomalies/active활성 이상만 (detected + in_progress)
GET/api/anomalies/{id}이상 상세
PATCH/api/anomalies/{id}/status상태 변경 (전이 규칙 검증)
POST/api/anomalies/{id}/note메모 추가
GET/api/system/status시스템 상태 (lifecycle 포함)
GET/api/system/cycles감지 사이클 이력
POST/api/system/detect수동 감지 실행
PATCH/api/system/interval감지 간격 변경 (최소 10초)
GET/api/system/health헬스 체크 (DB 연결 테스트)

감지 사이클 상세 흐름

run_detection_cycle() ├── get_due_rules() ← eval_interval 경과된 규칙만 선별 ├── evaluate_and_detect() × N (asyncio.gather 병렬) │ ├── engine.evaluate() │ │ ├── threshold: value vs warning/critical 비교 │ │ ├── delta: |현재값 - 이전값| vs 임계치 │ │ ├── absence: 쿼리 결과 비어있으면 이상 │ │ └── llm: 데이터만 가져오고 LLM이 최종 판단 │ ├── llm_enabled? → analyze_and_save() │ │ ├── ReAct 에이전트 루프 (max 3회) │ │ ├── JSON 추출: {is_anomaly, confidence, severity, title, analysis} │ │ ├── confidence < 0.7 → 무시 (오탐 필터) │ │ └── 기존 미해결 이상? → occurrence_count++ (중복 방지) │ └── anomaly → DB INSERT ├── check_and_auto_resolve() ← auto_resolve=1 이상 재평가 (v1.7.0) ├── update last_evaluated_at └── log cycle (cycle_id, duration_ms, anomalies_found, auto_resolved)

rule 규칙 엔진

왜 필요한가 — DB-first 전략

K8s 멀티 Pod 환경에서 YAML을 source of truth로 하면 Pod 재시작 시 DB 변경이 증발합니다. 따라서 DB가 런타임 원본이고, rules.yaml은 초기 seed + 버전관리 백업 역할입니다.

DB-first 동기화 흐름

서버 시작 시 DB 규칙 있음? → 스킵 (DB 유지, YAML 무시) DB 비어있음? → rules.yaml에서 seed (sync_to_db()) UI에서 규칙 변경 시 detection_rules (DB) → sync_db_to_yaml() → rules.yaml (백업) YAML 쓰기 실패해도 에러 무시 (read-only FS 대응)
info
필드 매핑: YAML의 name → DB의 rule_name, queryquery_template, tooltool_name. 네이밍 컨벤션이 다르지만 로더가 자동 변환합니다.

89개 내장 FAB 도구 카탈로그

규칙 생성 UI에서 선택 가능한 사전 정의 도구 목록입니다.

카테고리도구설명
logisticsget_conveyor_load존별 컨베이어 부하율(%)
get_transfer_throughput라인별 반송 처리량
get_bottleneck_zones대기시간 초과 병목존
get_agv_utilizationAGV/OHT 상태별 대수/비율
get_flow_balance공정별 유입/유출 밸런스
wipget_wip_levels공정별 WIP vs 목표 비율
get_queue_length스텝별 대기 LOT 수
get_aging_lots기준시간 초과 장기체류 LOT
get_wip_trend시간별 WIP 변화 트렌드
equipmentget_equipment_status설비 현재 상태(RUN/IDLE/DOWN/PM)
get_equipment_utilization설비 가동률(%)
get_unscheduled_downs비계획정지 이력
get_pm_schedule예방보전 일정
get_equipment_alarms설비 알람 이력

변경 이력 추적

모든 규칙 CRUD 후 sync_db_to_yaml()이 best-effort로 호출되며 (실패 시 무시), 변경 이력은 rule_audit_log 테이블에 diff와 함께 저장됩니다. 누가, 언제, 어떤 규칙을, 어떻게 변경했는지 추적할 수 있습니다. POST /api/rules/sync로 DB → YAML 수동 내보내기도 가능합니다.

API 엔드포인트

MethodPath설명
GET/api/rules규칙 목록 (include_disabled 옵션)
GET/api/rules/{id}규칙 상세
POST/api/rules규칙 생성 (검증 포함)
PUT/api/rules/{id}규칙 수정 (감사 로깅)
DELETE/api/rules/{id}규칙 삭제
GET/api/rules/{id}/history규칙 변경 이력
POST/api/rules/{id}/test샘플 데이터로 규칙 테스트
GET/api/rules/tools/catalog사용 가능한 도구 카탈로그

receipt_long 탐지 로그

왜 필요한가

이상탐지 시스템은 "왜 이 이상이 탐지되었는가?"와 "마지막 탐지 사이클은 정상이었는가?"를 추적할 수 있어야 합니다. 탐지 결과만이 아니라 탐지 과정 자체를 로깅합니다.

UI 구조

view_list
3영역 레이아웃
  • 상단 요약: 마지막 사이클 (규칙 평가수 / 이상 감지수 / 소요시간 / 시작시각) + 수동 감지 버튼
  • 감지 사이클 이력: 최근 20건 테이블 — 시작/완료/소요/상태 컬럼
  • 이상 이력 탭: 감지됨 / 처리중 / 해결 3탭, 각 탭에 건수 표시

상태 전이 매트릭스

현재 상태전이 가능
detectedin_progress, resolved, dismissed
in_progressresolved, dismissed, detected
resolveddetected (재발 시)
dismisseddetected (재발 시)

state_change 권한이 있는 사용자만 상태를 변경할 수 있습니다 (operator 이상). 허용되지 않은 전이는 HTTP 400으로 거부됩니다.

파일 구조

파일역할
api/anomalies.py이상 CRUD + 상태 관리 (5 endpoints)
ui/pages/detection_logs.py로그 뷰어 UI (사이클 이력 + 이상 탭)
core/db/queries/anomalies.py이상 쿼리 (필터, 집계, 상태 전이)

상태 전이 상세

detected ──→ in_progress "조사 시작" │ │ │ ├──→ resolved "해결 완료" │ ├──→ dismissed "오탐 무시" │ └──→ detected "재오픈" │ ├──→ resolved "즉시 해결" └──→ dismissed "즉시 무시" resolved ──→ detected "재발 시 자동 재오픈" dismissed ──→ detected "재발 시 자동 재오픈"

build Tool Studio

왜 필요한가

FAB 엔지니어가 코드를 작성하지 않고 SQL 기반 데이터 조회 도구를 만들 수 있어야 합니다. 만든 도구는 이상탐지 규칙, AI Chat, 워크플로우, Claude(MCP)에서 동일하게 재사용됩니다. 한 번 정의하면 4곳에서 쓸 수 있는 "도구 허브" 개념입니다.

lightbulb
핵심 가치: SQL을 알면 AI 도구를 만들 수 있다. 프로그래밍 지식 없이도 MES 데이터를 조회하는 AI 도구를 생성하고, 이상탐지 규칙에 연결하거나 AI Chat에서 자연어로 호출할 수 있습니다.

아키텍처

ds_connections (DB 연결 정의)ds_tools (SQL + 파라미터 + 컬럼 메타데이터)DataStudioManager.execute_tool() ├── Rate Limit 체크 (tool.rate_limit/분) ├── Cache 확인 (tool.cache_ttl초 TTL) ├── DatabaseEngine.execute(sql, params) │ ├── SQLiteEngine (aiosqlite) │ ├── OracleEngine (oracledb 별도 풀) │ └── FLOPIInternalEngine (core.db.oracle 풀 재사용) └── ds_query_log 기록 (caller 필드로 호출 추적)

3종 엔진

엔진용도연결 관리
SQLiteEngine시뮬레이터 / 로컬 파일 DBaiosqlite 단일 연결
OracleEngine외부 Oracle 서버 연결oracledb 비동기 풀 (min=1, max=4)
FLOPIInternalEngineFLOPI 자체 Oracle 풀 재사용core.db.oracle 풀 (닫지 않음)

Tool Registry 브릿지

서버 시작 시 sync_ds_tools()가 실행됩니다. DB의 활성 Tool Studio 도구를 tool_registry에 동적 등록하여, 이상탐지 규칙의 source_type=tool에서 바로 호출할 수 있습니다. 각 도구마다 OpenAI function calling 스키마도 자동 생성됩니다.

check_circle
파라미터 단순화 (v1.6): Tool Studio UI에서 파라미터 정의가 단순화되었습니다. 모든 파라미터는 기본값을 포함한 단일 테이블로 관리됩니다. Optional 구분 체크박스가 제거되고, 기본값 컬럼에 값을 입력하면 자동으로 해당 파라미터가 LLM에 항상 제공됩니다. 스키마 생성 시 기본값이 있는 파라미터도 모두 required 배열에 포함됩니다.

caller 추적

ds_query_log.caller 필드로 호출 출처를 추적합니다: ai_chat detection workflow mcp api

파일 구조

data_studio/
├── manager.py        # DataStudioManager (도구 실행 + 스키마 빌드 + 카테고리 자동 분류)
├── executor.py       # SQL 실행기 (Rate Limit + 캐싱 + 타임아웃)
├── queries.py        # 연결/도구/로그 CRUD 쿼리
├── codegen.py        # Python 코드 자동 생성
├── cache.py          # 쿼리 결과 캐시 관리
├── rate_limiter.py   # 분당 요청 수 제한
└── pages/            # Data Studio 전용 페이지

API 엔드포인트 (17개)

MethodPath설명
DB 연결 관리
POST/api/tool-studio/connectionsDB 연결 생성
GET/api/tool-studio/connections연결 목록
GET/api/tool-studio/connections/{id}연결 상세
POST/api/tool-studio/connections/{id}/test연결 테스트
DELETE/api/tool-studio/connections/{id}연결 삭제
도구 관리
POST/api/tool-studio/tools도구 생성
GET/api/tool-studio/tools도구 목록 (enabled_only 필터)
GET/api/tool-studio/tools/{id}도구 상세
PUT/api/tool-studio/tools/{id}도구 수정
DELETE/api/tool-studio/tools/{id}도구 삭제
도구 실행
POST/api/tool-studio/tools/{id}/executeID로 실행
POST/api/tool-studio/tools/execute/{name}이름으로 실행
GET/api/tool-studio/tools/{id}/codePython 코드 생성
운영
GET/api/tool-studio/query-log쿼리 실행 이력
GET/api/tool-studio/metrics24h 사용 통계
POST/api/tool-studio/seed/importseed.json 가져오기
GET/api/tool-studio/seed/exportseed.json 내보내기

3종 엔진 비교

엔진용도연결 관리특이사항
SQLiteEngine시뮬레이터 / 로컬 파일 DBaiosqlite 단일 연결파일 경로만 필요
OracleEngine외부 Oracle 서버oracledb 비동기 풀 (min=1, max=4)별도 커넥션 풀
FLOPIInternalEngineFLOPI 자체 Oracle 풀 재사용core.db.oracle 풀 공유풀 닫지 않음 (공유 자원)

account_tree RCA Studio

왜 필요한가

반도체 FAB의 이상 원인분석(Root Cause Analysis) 노하우는 전문가의 머릿속에만 있습니다. SOP 문서는 만들기 번거롭고 비일관적이며, 퇴직 시 지식이 사라집니다. AI 대화를 통해 자연스럽게 분석 과정을 구조화하고, Mermaid 플로차트로 자동 변환하여 지식을 축적합니다.

핵심 플로우

엔지니어: "컨베이어 부하율이 95% 넘었어" │ ▼ AI 분석가 (ANALYST_SYSTEM 프롬프트) ├── "어느 존에서 발생했나요? 시간대는?" ← 프로빙 질문 ├── JSON 구조 추출: {nodes: [...], edges: [...]} └── "빠진 분기는 없나요?" ← 완성도 체크 │ ▼ FlowGraph (노드/엣지 누적 관리) └── to_mermaid() → 실시간 플로차트 시각화

UI: 3패널 스플리터

패널비율내용
좌측20%시나리오 라이브러리 — 진행 중 세션 + 저장된 시나리오
중앙55%AI 채팅 — 대화 + 추출 노드/엣지 알림 카드
우측45%Mermaid 플로차트 — 실시간 렌더링 + 노드 편집

노드 타입

타입Mermaid 형태색상용도
decision{{label}} 다이아몬드파랑판단 분기점
action([label]) 원형주황수행 조치
end((label)) 이중원초록결론/종료
check[label] 직사각형기본확인/측정

그래프 검증 엔진

Analyst.validate_graph()가 STRUCTURER_SYSTEM 프롬프트로 전체 그래프를 검증합니다:

검증 결과는 pass / warning / fail 상태 + 이슈 목록 + 자동 수정 제안(fixes)으로 반환됩니다.

시나리오 재활용

저장된 시나리오는 이후 새 분석 시 참고 시나리오로 AI 컨텍스트에 주입됩니다. AI가 기존 패턴과 비교하며 "이 단계 다음에 보통 X를 확인하는데, 맞나요?" 식으로 질문합니다. 분석이 쌓일수록 AI가 더 정확해지는 선순환 구조입니다.

파일 구조

rca/
├── analyst.py              # AI 대화 + 구조 추출 (Analyst 클래스)
├── structurer.py           # FlowGraph → Mermaid 변환
├── prompts.py              # ANALYST_SYSTEM / STRUCTURER_SYSTEM 프롬프트
├── models.py               # 데이터 모델 (Session, Message, Node, Edge)
├── queries.py              # DB CRUD (세션/메시지/노드/엣지/시나리오)
├── workflow_converter.py   # RCA → 워크플로우 자동 변환
├── pages/                  # RCA 전용 페이지
└── sample_data/
    ├── spc.py              # SPC 통계적 공정 제어 테스트 데이터
    ├── fdc.py              # FDC 결함 분류 테스트 데이터
    └── alarms.py           # 알람 로그 테스트 데이터

API 엔드포인트 (13개)

MethodPath설명
세션 관리
POST/api/deep-rca/sessions분석 세션 생성
GET/api/deep-rca/sessions세션 목록
GET/api/deep-rca/sessions/{id}세션 상세 (메시지 + 노드 + 엣지)
PATCH/api/deep-rca/sessions/{id}/status세션 상태 변경
POST/api/deep-rca/sessions/{id}/chatAI 대화 (SSE 스트리밍)
POST/api/deep-rca/sessions/{id}/validateAI 플로차트 검증
시나리오 관리
POST/api/deep-rca/scenarios시나리오 저장
GET/api/deep-rca/scenarios시나리오 목록 (검색/필터)
GET/api/deep-rca/scenarios/{id}시나리오 상세
DELETE/api/deep-rca/scenarios/{id}시나리오 삭제
POST/api/deep-rca/scenarios/{id}/fork시나리오 포크 (복제 + 수정)
참조 데이터
GET/api/deep-rca/fab-toolsFAB 장비 목록
GET/api/deep-rca/knowledge-gaps지식 갭 분석

route Workflow

왜 필요한가

"데이터 조회 → AI 분석 → 조건 분기 → 알림"의 자동화 파이프라인을 코드 없이 시각적으로 구성하고 실행할 수 있어야 합니다. Dify/n8n 스타일의 비주얼 워크플로우 빌더를 FLOPI 내부에 구현했습니다.

12가지 노드 타입

노드색상실행 방식
trigger파랑시작점 (수동 트리거), 패스스루
event_trigger오렌지이벤트 기반 시작 — 이상탐지 연동 (anomaly_detected)
data_query초록registry.dispatch(tool_name, params)
ai_analysis보라llm_client.simple_chat() — 이전 노드 출력 자동 포함
rag_search시안core.rag.searcher.search() — 시맨틱 검색
condition주황eval(expression) — 샌드박스 실행, true/false 분기
http_request핑크httpx 비동기 (GET/POST/PUT/DELETE/PATCH, 최대 120초)
data_transformeval() 데이터 변환/필터 (sum/min/max/sorted/json 지원)
loop보라리스트 순회 처리 (for_each), 최대 100회
merge슬레이트condition 분기 합류 지점
notification빨강메시지 템플릿 해석 + 알림 출력
end회색워크플로우 종료

이벤트 트리거 — 이상탐지 연동

info
event_trigger 노드: 이상탐지 엔진이 이상을 감지하면 trigger_event_workflows("anomaly_detected", result)를 호출합니다. event_trigger 노드가 있는 워크플로우 중 category/severity/rule_id 조건에 매칭되는 것을 자동 실행합니다. 수동 테스트: POST /api/workflows/trigger-event

AI 워크플로우 생성 (core/workflow/generator.py)

워크플로우 빌더 우측 패널에서 자연어 대화로 워크플로우를 직접 생성합니다.

사용자: "PHOTO 라인 부하율 확인 후 85% 초과면 알림 보내줘" │ ├── WorkflowGenerator.generate(prompt, existing_graph) │ ├── Tool Studio 도구 목록 주입 │ ├── LLM → 워크플로우 JSON 생성 │ └── 후처리: 노드 연결 검증 + 기본 좌표 배치 │ └── 캔버스에 즉시 적용 (멀티턴: 후속 메시지로 수정 가능)

실행 엔진: Kahn's Algorithm

check_circle
DAG 토폴로지 정렬: Kahn's algorithm으로 진입차수 기반 실행 순서를 결정합니다. trigger에서 BFS로 도달 가능한 노드만 추출하고, 순서대로 실행하며 결과를 context[node_id]에 저장합니다. 12종 노드 모두 지원 (loop 최대 100회, http_request 최대 120초 타임아웃).

변수 템플릿 시스템

형식의미
{{prev}}직전 부모 노드의 출력
{{변수명}}노드의 output_var로 정의한 이름
{{nodeId.field}}특정 노드의 출력에서 특정 필드
{{context}}전체 컨텍스트 JSON dump (하위호환)

condition 노드 샌드박스

eval(expression) 실행 시 builtins를 제한합니다: len, str, int, float + ctx, prev, v 변수만 허용. condition 결과가 true/false에 따라 반대 방향 후손 노드를 모두 skipped 처리합니다.

캔버스 구현 (JS)

touch_app
Pointer Events 기반 인터랙션
  • Pointer Events + setPointerCapture — 드래그 신뢰성 확보
  • 좌표 기반 hitTest() — DOM 의존 없이 nodePos 맵으로 노드/포트 판별
  • 5px DRAG_THRESHOLD — 클릭 vs 드래그 구분
  • SVG Bezier 엣지 — 드래그 중 실시간 업데이트
  • emitEvent → ui.on — JS→Python 이벤트 통신 (NiceGUI 패턴)

파일 구조

파일라인 수역할
ui/pages/workflow.py최대 파일3패널 비주얼 빌더 (팔레트 + 캔버스 + 설정)
core/workflow/engine.py~310DAG 실행 엔진 (Kahn's algorithm, 12노드 지원)
core/workflow/generator.py~180AI 워크플로우 생성기 — 자연어 → 워크플로우 JSON
api/workflow.py~240워크플로우 CRUD + 실행 + 이벤트 트리거 API
core/db/queries/workflow.pyworkflows + workflow_runs 쿼리

API 엔드포인트 (11개)

MethodPath설명
GET/api/workflows워크플로우 목록 (검색, 카테고리 필터)
GET/api/workflows/categories카테고리 목록
POST/api/workflows워크플로우 생성
GET/api/workflows/{id}워크플로우 상세 (graph_data 포함)
PUT/api/workflows/{id}워크플로우 수정
DELETE/api/workflows/{id}워크플로우 삭제
POST/api/workflows/{id}/duplicate워크플로우 복제
GET/api/workflows/{id}/exportJSON 내보내기
POST/api/workflows/importJSON 가져오기
POST/api/workflows/{id}/run워크플로우 실행
GET/api/workflows/{id}/runs실행 이력

menu_book Knowledge Base

왜 필요한가

FAB 전문가 지식(SOP, 매뉴얼, 분석 노하우)을 등록하면 자동으로 청킹/임베딩되어 워크플로우의 rag_search 노드와 AI Chat에서 시맨틱 검색으로 활용할 수 있습니다. 키워드 검색이 아닌 의미 기반 검색으로, "컨베이어 병목"이라고 질문하면 "반송 지연 현상"이라고 적힌 문서도 찾아냅니다.

RAG 파이프라인

문서 등록 (제목 + 본문 + 카테고리 + 태그) │ ▼ LLM 구조화 (선택: preserve/reorganize 모드) │ ▼ chunker.py → 문단 기반 청킹 \n\n 문단 구분자로 분리 → chunk_size(500자) 초과 시 overlap(50자) 슬라이딩 │ ▼ GeminiEmbedder.embed_texts() batchEmbedContents API (gemini-embedding-001) │ ▼ ChromaVectorStore.upsert() HNSW cosine space, 메타데이터: doc_id, category, chunk_index, heading

검색 프로세스

search(query, top_k, category): 쿼리 텍스트 임베딩 → ChromaDB cosine 유사도 검색 → score = 1.0 - distance 변환 → {chunk_text, score, doc_title, category} 반환.

교체 가능한 추상화

레이어기본교체 방법
임베딩Gemini (text-embedding-004)환경변수 RAG_EMBED_BACKEND=openai
벡터 저장소ChromaDB (로컬 파일)set_vectorstore(MilvusStore())

always_on 도구

check_circle
knowledge_base_searchalways_on=True로 등록됩니다. ToolSelector의 top_k 제한에서 제외되어 항상 AI Chat에 제공됩니다. 사용자가 질문하면 AI가 자동으로 관련 지식을 검색할 수 있습니다.

파일 구조

core/rag/
├── embedder.py       # 추상 임베더 + Gemini/OpenAI 구현
├── vectorstore.py    # 추상 벡터저장소 + ChromaDB 구현
├── chunker.py        # 문단 기반 텍스트 청킹 (chunk_size=500, overlap=50)
├── structurer.py     # LLM 기반 문서 구조화 (비정형 → 계층 청크)
│                     #   preserve 모드: 원본 구조 유지
│                     #   reorganize 모드: LLM이 계층 재구성
└── searcher.py       # 통합 시맨틱 검색 (임베더 + 벡터저장소)

core/db/queries/knowledge_base.py   # 문서 메타데이터 CRUD
api/knowledge_base.py               # REST API (14개 엔드포인트)
ui/pages/knowledge_base.py          # 관리 페이지 (문서 CRUD + 검색 테스트 + 재임베딩)

검색 결과 반환 형식

{
  "results": [
    {
      "chunk_text": "PHOTO 공정에서 오버레이 불량 발생 시...",
      "score": 0.87,
      "doc_title": "PHOTO 공정 SOP",
      "doc_id": "kb_001",
      "category": "SOP",
      "chunk_index": 3,
      "heading": "오버레이 불량 대응"
    }
  ],
  "total": 5,
  "query": "오버레이 불량 대응 방법"
}

API 엔드포인트 (14개)

MethodPath설명
POST/api/knowledge-base문서 등록 (자동 청킹 + 임베딩)
GET/api/knowledge-base문서 목록 (카테고리 필터)
GET/api/knowledge-base/{id}문서 상세
PUT/api/knowledge-base/{id}문서 수정 (재청킹 + 재임베딩)
DELETE/api/knowledge-base/{id}문서 삭제
POST/api/knowledge-base/search시맨틱 검색 (top_k, category)
GET/api/knowledge-base/categories카테고리 목록
GET/api/knowledge-base/stats통계 (문서 수, 청크 수, 크기)
POST/api/knowledge-base/{id}/structureLLM 기반 문서 구조화 (preserve/reorganize)
POST/api/knowledge-base/{id}/reembed단일 문서 재임베딩
POST/api/knowledge-base/reembed-all전체 재임베딩 시작 (SSE 진행 상황)
GET/api/knowledge-base/reembed-progress재임베딩 진행 SSE 스트리밍
POST/api/knowledge-base/bulk-delete다중 문서 삭제
GET/api/knowledge-base/{id}/chunks청크 목록 + 임베딩 정보

설정 옵션

환경변수기본값설명
RAG_EMBED_BACKENDgemini임베딩 백엔드 (gemini / openai / custom)
RAG_EMBED_BASE_URL임베딩 API base URL
RAG_EMBED_API_KEY임베딩 API 키
RAG_EMBED_MODELtext-embedding-004임베딩 모델명
RAG_CHROMA_DIRFLOPI/chroma_data/ChromaDB 저장 경로

chat AI Chat

왜 필요한가

FAB 운영자가 자연어로 "지금 PHOTO 공정 WIP가 얼마야?"라고 질문하면 AI가 실제 MES 데이터를 조회하고 분석하여 답변해야 합니다. 일반적인 챗봇이 아니라 Tool-augmented AI입니다.

v1.7.0 신규 기능

lightbulb
카테고리 전용 프롬프트: 시스템 설정에서 ai_chat.category_prompt.{카테고리}로 카테고리별 전문 지침을 정의할 수 있습니다. 사용자가 시작 카드로 범주를 선택하면 해당 카테고리의 프롬프트가 시스템 프롬프트에 자동 주입됩니다.
lightbulb
사용자 개인 프롬프트: 각 사용자가 /api/user-prompts/me로 개인 선호사항 프롬프트를 등록할 수 있습니다. 답변 스타일/형식 선호도로 활용되며, 시스템 정책이나 보안 규칙은 오버라이드할 수 없습니다. 버전 이력 관리를 지원합니다.
check_circle
Code Interpreter: core/code_interpreter.py — AI가 데이터 분석, 통계, 시각화가 필요할 때 Python 코드를 생성하여 안전하게 실행합니다. subprocess 기반 샌드박스로 30초 타임아웃, 위험 패턴 차단, 허용된 라이브러리(pandas, numpy, matplotlib 등)만 사용 가능합니다. 실행 결과로 stdout, matplotlib 이미지(base64), DataFrame 테이블을 반환합니다.

SSE 스트리밍 구조

POST /api/ai-chat (SSE StreamingResponse) │ ├── 세션 자동 생성/저장 │ ├── 카테고리 프롬프트 주입 (ai_chat.category_prompt.{cat}) ├── 사용자 개인 프롬프트 주입 (user_prompts 테이블) ├── 챗 매크로 매칭 (chat_scenarios 키워드 매칭) │ ├── SmartToolSelector.select_tools(user_query) │ ← 89개+ 도구 + code_interpreter 중 관련 도구만 선별 │ ├── Tool-calling rounds (최대 max_tool_rounds=5회) │ LLM.chat(messages, tools) │ ├── tool_calls? → registry.dispatch() → 결과 추가 │ │ └── code_interpreter → subprocess 실행 → 이미지/테이블 반환 │ └── no tool_calls? → 최종 답변 단계 │ ├── 최종 답변: LLM.chat_stream() → 토큰 단위 SSE │ └── DB에 세션 + 메시지 저장

SSE 이벤트 타입

이벤트설명
session_id세션 ID 전달 (첫 이벤트)
tool_call도구 호출 시작 (이름 + 인자)
tool_result도구 결과 (최대 2000자 프리뷰)
token최종 답변 토큰 스트리밍
error오류 발생
done응답 완료
info
Open WebUI 스타일: 도구 호출 라운드는 non-streaming, 최종 답변만 streaming. 사용자가 어떤 도구가 호출됐는지 실시간으로 확인할 수 있습니다.

Smart Tool Selector

89개 이상의 도구가 등록될 것을 대비한 시맨틱 도구 선별기입니다.

check_circle
자동 동기화: registry.on_change(selector.mark_dirty) — 도구 등록/해제 시 자동으로 dirty 마킹, 다음 select_tools() 호출 시 재동기화.

API 엔드포인트 (11개)

MethodPath설명
POST/api/ai-chatSSE 스트리밍 대화 (category 스코프 지원)
GET/api/ai-chat/sessions세션 목록 (category 필터 지원)
POST/api/ai-chat/sessions세션 생성 (category 포함)
GET/api/ai-chat/sessions/{id}세션 + 메시지
PATCH/api/ai-chat/sessions/{id}세션 제목/카테고리 변경
DELETE/api/ai-chat/sessions/{id}세션 삭제 (soft delete)
GET/api/ai-chat/sessions/historyAdmin 채팅 이력 (필터)
GET/api/ai-chat/tools사용 가능한 도구 목록
POST/api/ai-chat/tools/select스마트 도구 선택 (scope 지정 가능)
GET/api/ai-chat/suggested-questions카테고리별 웰컴 추천 질문
GET/api/ai-chat/categories채팅 카테고리 목록 (카드 정의)

User Prompts API (사용자 개인 프롬프트)

MethodPath설명
GET/api/user-prompts/me내 프롬프트 조회
PUT/api/user-prompts/me내 프롬프트 생성/수정 (버전 관리)
PATCH/api/user-prompts/me/toggle프롬프트 활성화/비활성화
GET/api/user-prompts/me/versions버전 이력 조회
POST/api/user-prompts/me/restore특정 버전으로 복원

Code Interpreter 상세

code_interpreter 도구 호출 │ ├── _check_code_safety(code) │ ├── 차단 패턴: os.system, subprocess, eval, exec, open(w), socket, requests ... │ └── 위반 시 → 에러 메시지 반환 (실행 안 함) │ ├── subprocess 실행 (격리된 Python 프로세스) │ ├── 허용 라이브러리: pandas, numpy, scipy, matplotlib, sklearn, statistics ... │ ├── 타임아웃: 30초 │ └── 작업 디렉토리: /tmp/flopi_code/ │ └── 결과 반환 ├── stdout 텍스트 ├── matplotlib 이미지 (base64 PNG) └── DataFrame 테이블 (JSON)

Agent Loop 상세 흐름

run_agent_loop(system_prompt, user_message, max_rounds=5) │ ├── 카테고리 프롬프트 + 사용자 프롬프트 + 매크로 주입 │ ├── tool_selector.select_tools(query) │ ├── 도구 수 ≤ top_k(15)? → 전체 반환 (임베딩 없음) │ ├── always_on 도구 → 무조건 포함 (knowledge_base_search, code_interpreter) │ └── ChromaDB 시맨틱 검색 → top_k × 2 후보 → 카테고리 다양성 필터 │ ├── Round 1~5: │ ├── LLM.chat(messages, tools) │ ├── tool_calls 있음? → registry.dispatch() → 결과 append │ │ ├── code_interpreter → subprocess → 이미지/테이블 SSE 전송 │ │ └── SSE: tool_calltool_result 이벤트 │ └── tool_calls 없음? → 최종 응답 단계로 │ ├── 최종 답변: LLM.chat_stream() → 토큰 단위 SSE │ └── SSE: tokentoken → ... → done │ └── DB 저장 (ai_chat_sessions + ai_chat_messages)

compare_arrows Migration (이관 관리)

왜 필요한가

운영 중인 시스템에서 DB 스키마를 변경하고, SQLite에서 Oracle로 블록별 이관을 관리해야 합니다. 블록별 이관 상태 추적, 스키마 비교, DDL 자동 생성을 통해 SQLite 시뮬레이터에서 Oracle 운영 환경으로 안전하게 전환합니다.

파일 구조

core/migration/
├── __init__.py        # 패키지
├── checker.py         # Oracle 스키마 검증 (필수 테이블/컬럼 확인)
└── router.py          # 이관 라우팅 로직

api/migration.py         # 이관 관리 REST API (블록별 상태 + 스키마 비교 + DDL 생성)
ui/pages/migration.py    # 이관 관리 UI 페이지

API 엔드포인트

MethodPath설명
GET/api/migration/status블록별 이관 상태 조회
POST/api/migration/toggle블록 이관 상태 전환 (oracle/sqlite)
POST/api/migration/compareOracle 스키마 비교 (누락 테이블/컬럼)
POST/api/migration/ddlOracle DDL 생성 (누락분 자동 생성)
warning
Oracle 운영 환경에서는 DBA 검토 후 적용을 권장합니다. core/migration/checker.py로 사전 검증 후 DDL을 적용하세요.

analytics LLM Usage

왜 필요한가

FAB 폐쇄망 환경에서 LLM API 비용과 사용량을 모듈별로 추적해야 합니다. 어떤 기능이 토큰을 가장 많이 소비하는지, 일별 추이는 어떤지, 과다 사용이 발생하고 있는지를 관리자가 모니터링할 수 있어야 합니다.

자동 태깅

from core.llm.client import set_llm_module
set_llm_module("detection")   # 이후 모든 LLM 호출이 "detection"으로 태깅
# ... LLM 호출 ...
set_llm_module("ai_chat")     # 모듈 전환

llm_usage_log 테이블에 module, model, prompt_tokens, completion_tokens, total_tokens, duration_ms가 자동 기록됩니다.

파일 구조

core/llm/
├── client.py          # LLMClient (OpenAI-compatible)
│   ├── chat()         # 도구 포함 호출
│   ├── chat_stream()  # 스트리밍 호출
│   └── simple_chat()  # 단순 호출
├── tool_registry.py   # ToolRegistry 싱글턴
│   ├── register()     # 도구 등록 (@registry.tool 데코레이터)
│   ├── dispatch()     # 도구 실행 (단일 진입점)
│   └── get_openai_tools()  # OpenAI function calling 스키마
├── tool_selector.py   # SmartToolSelector
│   ├── select_tools() # 시맨틱 검색으로 관련 도구 선택
│   └── sync()         # 도구 임베딩 동기화 (dirty 플래그)
└── agent_loop.py      # Multi-round ReAct Agent
    └── run_agent_loop()  # LLM ↔ 도구 호출 루프 (max_rounds)

자동 추적 메커니즘

set_llm_module("detection") → ContextVar에 모듈명 저장 │ LLMClient.chat() 호출 │ ├── 응답에서 usage 추출 (prompt_tokens, completion_tokens) │ └── llm_usage_log INSERT ├── module: "detection" ├── model: "gemini-2.0-flash" ├── prompt_tokens: 1234 ├── completion_tokens: 567 ├── total_tokens: 1801 └── duration_ms: 2340

UI 구성

monitoring
4개 KPI + 차트 + 테이블
  • KPI 카드: 총 호출 수 / 총 토큰 / 입력 토큰 / 출력 토큰
  • 일별 추이 차트: ECharts 듀얼 Y축 (호출 수 + 토큰K)
  • 모듈별 테이블: detection, deep_rca, ai_chat, workflow 등 한국어 라벨 매핑
  • 최근 50건: 시간/모듈/모델명/토큰/소요시간, 검색 지원

API 엔드포인트

MethodPath설명
GET/api/llm-usage/summary전체 요약 (총 호출 수, 총 토큰)
GET/api/llm-usage/by-module모듈별 사용량
GET/api/llm-usage/daily일별 추이 (기본 14일)
GET/api/llm-usage/recent최근 호출 (기본 50건)

LLM 설정 환경변수

환경변수기본값설명
LLM_BASE_URLGemini endpointAPI 베이스 URL (OpenAI-compatible)
LLM_API_KEYAPI 키
LLM_MODELgemini-2.0-flash모델명
LLM_TIMEOUT60타임아웃 (초)
LLM_MAX_TOKENS최대 출력 토큰

group Users & RBAC

왜 필요한가

FAB 환경에서 관리자/엔지니어/운영자/열람자의 접근 권한이 명확히 분리되어야 합니다. Tool Studio 설정을 운영자가 변경하거나, 열람자가 이상 상태를 변경하는 것은 심각한 문제를 야기할 수 있습니다.

4단계 RBAC

역할접근 범위제한
admin전체 full없음
engineer대부분 fullTool Studio, Users 접근 불가
operator기본 조회 + 이상 상태변경RCA/Workflow/Tool Studio 불가
viewer읽기 전용모든 쓰기 불가

권한 레벨

커스텀 오버라이드

check_circle
유연한 예외 처리: 사용자별 page_permissions JSON으로 기본 역할 권한을 오버라이드할 수 있습니다. 예: 특정 엔지니어에게만 Tool Studio 접근 허용.
// users.page_permissions 컬럼 (JSON)
{"workflow": "full", "data_studio": "view"}

네비게이션 필터링

get_nav_items(user)가 역할 + 커스텀 권한으로 접근 가능한 페이지만 사이드바에 표시합니다. 그룹("이상감지", "Studio", "관리") 구조로 접기/펼치기를 지원합니다.

파일 구조

파일역할
api/users.py사용자 CRUD + 로그인/로그아웃 API (9 endpoints)
ui/pages/users.py사용자 관리 UI (목록 + 생성 + 권한 편집)
core/auth/permissions.py권한 매트릭스 + get_page_access()
core/db/queries/users.py사용자 쿼리 (해시 비밀번호, 세션 토큰)

모듈별 권한 매트릭스 (상세)

모듈adminengineeroperatorviewer
dashboardfullfullfullfull
anomaliesfullfullstate_changeview
rulesfullfullviewview
detection_logsfullfullfullview
deep_rcafullfull
data_studiofull
workflowfullfull
knowledge_basefullfullviewview
ai_chatfullfullviewview
llm_usagefullview
usersfull
system_settingsfull

API 엔드포인트 (9개)

MethodPath설명
POST/api/users/login로그인 (세션 토큰 + nav 응답)
POST/api/users/register자가 등록 (viewer 역할 기본)
POST/api/users/logout로그아웃
GET/api/users/me현재 사용자 프로필
GET/api/users사용자 목록 (admin only)
POST/api/users사용자 생성 (admin only)
PUT/api/users/{id}사용자 수정
DELETE/api/users/{id}사용자 삭제
PATCH/api/users/{id}/permissions페이지 권한 오버라이드

hub MCP Server

왜 필요한가

Claude(claude.ai 또는 Claude Code)에서 FAB MES 데이터를 직접 조회할 수 있도록 Tool Studio 도구를 MCP(Model Context Protocol)로 노출합니다. 엔지니어가 Claude에서 "지금 PHOTO 공정 WIP 알려줘"라고 하면 실제 데이터를 가져옵니다.

연결 흐름

Claude → MCP (stdio) → mcp_server.py │ ├── list_tools() → DB 활성 도구 로딩 → MCP Tool 스키마 변환 │ └── call_tool() → registry.dispatch(name, args, caller="mcp") → DataStudioManager.execute_tool() → DatabaseEngine (SQLite/Oracle) → 결과 JSON (200행 초과 시 자동 잘라내기)

설정

// .mcp.json
{
  "mcpServers": {
    "flopi": {
      "command": "python",
      "args": ["mcp_server.py", "--sqlite", "simulator.db"]
    }
  }
}
info
시뮬레이터 모드: --sqlite simulator.db 플래그로 실제 Oracle 없이도 동작합니다. FAB 환경에 직접 연결하지 않고도 시뮬레이터 데이터를 Claude에서 탐색할 수 있습니다.

파일 구조

파일역할
mcp_server.pyMCP 서버 진입점 (133 lines, stdio 모드)

별도 디렉토리 없이 단일 파일로 구현. Tool Studio DB에서 활성 도구를 로드하여 MCP Tool 스키마로 자동 변환합니다.

도구 등록 흐름

서버 시작 → list_tools() ├── DB에서 활성 도구 로드 (ds_tools WHERE enabled=1) ├── 각 도구 → MCP Tool 스키마 변환 │ ├── name: 도구명 │ ├── description: SQL + 파라미터 설명 조합 │ └── inputSchema: 파라미터 → JSON Schema └── Claude에 도구 목록 반환 Claude 요청 → call_tool(name, arguments) ├── registry.dispatch(name, arguments, caller="mcp") ├── DataStudioManager.execute_tool() ├── Rate Limit + Cache 체크 ├── DB 엔진 → SQL 실행 └── 결과 JSON (200행 초과 시 잘라내기)

설정 옵션

// .mcp.json — Claude Code / Claude Desktop 연동
{
  "mcpServers": {
    "flopi": {
      "command": "python",
      "args": ["mcp_server.py", "--sqlite", "simulator.db"]
    }
  }
}

// 운영 환경 (Oracle 직접 연결)
{
  "mcpServers": {
    "flopi": {
      "command": "python",
      "args": ["mcp_server.py"]
    }
  }
}

science SQLite 시뮬레이터

왜 필요한가

FAB 폐쇄망의 Oracle DB에 항상 접근할 수 없으므로, 개발/테스트 환경에서 동일한 코드를 SQLite로 실행할 수 있어야 합니다. 코드 분기 없이 --sqlite 플래그 하나로 전환됩니다.

Monkey-Patch 패턴

check_circle
init_sqlite(db_path) 호출 시 core.db.oracle 모듈의 execute, execute_dml, execute_returning 함수를 SQLite 구현으로 monkey-patch합니다. 이후 모든 코드가 동일하게 동작합니다.

파일 구조

core/simulator/
├── sqlite_backend.py  # Monkey-patch + SQLite 연결 관리
├── sql_compat.py      # Oracle → SQLite SQL 변환
├── seeder.py          # MES 더미 데이터 생성 (컨베이어, 차량, WIP, 설비, LOT)
└── scenarios.py       # 사전 정의 장애 시나리오

Monkey-Patch 대상

원본 (core.db.oracle)패치 대상설명
execute(sql, params)_execute()SELECT → SQLite 실행
execute_dml(sql, params)_execute_dml()INSERT/UPDATE/DELETE
execute_returning(sql, params)_execute_returning()RETURNING → lastrowid 변환
init_pool()_noop_async()풀 초기화 → no-op
close_pool()_noop_async()풀 종료 → no-op

SQL 호환 변환

sql_compat.pyoracle_to_sqlite(sql) 함수가 Oracle 문법을 SQLite로 변환:

OracleSQLite비고
SYSTIMESTAMPCURRENT_TIMESTAMP현재 시각
SYSDATEDATE('now')현재 날짜
NUMTODSINTERVAL(n, 'HOUR')n/24.0시간 간격
INTERVAL 'n' HOUR'n hours'시간 리터럴
NVL(a, b)COALESCE(a, b)NULL 대체
:1, :2 (positional)?, ?바인드 변수
RETURNING id INTO :outlastrowid삽입 후 ID 반환

초기화 흐름

python main.py --sqlite simulator.db │ ├── parse_args() → args.sqlite = "simulator.db" │ ├── init_sqlite("simulator.db") │ ├── aiosqlite 연결 생성 │ ├── schema.sql DDL 실행 (20+ 테이블 생성) │ ├── core.db.oracle 함수 5개 monkey-patch │ └── seeder.py → 시뮬레이터 데이터 seed │ ├── 컨베이어 존 (12개) │ ├── AGV/OHT 차량 (30대) │ ├── WIP LOT (200건) │ ├── 설비 상태 (50대) │ └── 알람 이력 (100건) │ └── 이후 모든 코드가 SQLite로 동작 (코드 분기 없음 — monkey-patch가 투명하게 전환)

cable 모듈 간 연결

전체 연결 맵

rules.yaml ←→ detection_rules (DB) │ eval_interval 경과 시 detection/scheduler.py │ detection/engine.py ├── SQL 직접 실행 └── tool_registry.dispatch()ds_bridge.py ← data_studio (DB) ↑ MCP server (Claude) knowledge_base (DB + ChromaDB) ↑ api/knowledge_base.py ↓ core/rag/searcher.py ├── workflow/engine.py (rag_search 노드) ├── tool_registry (knowledge_base_search, always_on) └── api/ai_chat.py → SSE → ui/pages/ai_chat.py rca/analyst.py ← rca/prompts.py │ JSON 구조 추출 ▼ rca/structurer.py (FlowGraph) │ to_mermaid() ▼ ui/pages/rca_studio.py → ui.mermaid() │ save_scenario ▼ rca_scenarios (DB) → analyst.py (참고 시나리오 컨텍스트)

도구 흐름 (Tool Studio 중심)

info
Tool Studio에서 만든 도구 하나가 4곳에서 동시에 사용됩니다: (1) 이상탐지 규칙, (2) AI Chat, (3) 워크플로우 data_query 노드, (4) Claude MCP. tool_registry가 중앙 허브 역할을 합니다.

Tool Registry 중앙 허브

ToolRegistry (싱글턴) │ ├── 내장 도구 │ └── knowledge_base_search (always_on=True) │ ├── Tool Studio 도구 │ └── ds_bridge.sync_ds_tools() → 동적 등록 │ └── 소비자 (caller별 추적) ├── AI Chat → agent_loop.py → registry.dispatch(caller="ai_chat") ├── 이상탐지 → analyzer.py → registry.dispatch(caller="detection") ├── 워크플로우 → engine.py → registry.dispatch(caller="workflow") └── MCP → mcp_server.py → registry.dispatch(caller="mcp")

도구 실행 흐름 비교

소비자도구 선택실행 방식결과 처리
AI ChatSmartToolSelector (시맨틱)LLM이 tool_call 결정SSE 스트리밍 응답
이상탐지규칙의 tool_name 고정스케줄러 자동 실행DB anomaly INSERT
워크플로우data_query 노드 설정DAG 순서대로context[node_id]에 저장
MCPClaude가 list_tools 조회Claude의 tool_useJSON 텍스트 반환

settings System Settings

왜 필요한가

FAB 운영 중 서버 재시작 없이 설정을 변경해야 하는 경우가 많습니다. LLM 모델 변경, 탐지 간격 조정, 알림 설정 등을 UI에서 즉시 반영합니다. 모든 설정은 DB 기반으로 저장되어 재시작 후에도 유지됩니다.

파일 구조

파일역할
api/system_settings.py설정 CRUD API (2 endpoints)
api/system.py시스템 상태 + 감지 제어 API (5 endpoints)
ui/pages/system_settings.py설정 관리 UI
config.py통합 설정 클래스 (환경변수 기반)

API 엔드포인트

MethodPath설명
GET/api/system-settings전체 설정 (카테고리별)
PUT/api/system-settings/{key}설정 변경 (타입/범위 검증)
GET/api/system/health헬스 체크 (DB 연결 테스트)
GET/api/system/status시스템 상태 (lifecycle + 24h 메트릭)
POST/api/system/detect수동 감지 실행
PATCH/api/system/interval감지 간격 변경 (최소 10초)

주요 설정 항목

카테고리설명기본값
LLMllm.timeout타임아웃 (초)60
llm.temperature온도 파라미터0.7
감지detection.interval틱 간격 (초)60
detection.max_concurrent병렬 평가 수5
DBoracle.dsnOracle DSNconfig.py
warning
admin 전용: 시스템 설정 변경은 admin 역할만 가능합니다. engineer/operator/viewer는 접근 자체가 차단됩니다 (사이드바에서 숨김).

quiz Question Flows (질문 흐름)

왜 필요한가

AI Chat 매 응답마다 LLM을 추가 호출해 후속질문을 생성하면 토큰 낭비 + 맥락 없는 질문이 됩니다. FAB에선 질문 패턴이 정해져 있습니다 (설비 상태 → 알람 → OEE → PM). 관리자가 질문 흐름을 DB에 등록하면, LLM이 참고하여 자연스럽고 간결한 후속질문을 생성합니다.

하이브리드 방식

흐름이 매칭되면 → LLM에 흐름 컨텍스트 전달 → 다음 단계 기반 후속질문.
흐름이 없으면 → 기존처럼 자유 생성 (폴백). 키워드 매칭으로 현재 위치를 파악하고 남은 단계를 추출합니다.

파일 구조

파일역할
api/question_flows.pyREST CRUD API (5 endpoints)
core/db/queries/question_flows.pyDB 쿼리 (list/get/create/update/delete)
ui/pages/question_flows.py관리 UI (카드 목록 + 카테고리 필터 + 편집 다이얼로그)
api/ai_chat.py후속질문 생성 시 흐름 매칭 로직 (_match_flows, _build_flow_context)

API 엔드포인트

MethodPath설명
GET/api/question-flows흐름 목록
GET/api/question-flows/{id}흐름 상세
POST/api/question-flows흐름 생성
PUT/api/question-flows/{id}흐름 수정
DELETE/api/question-flows/{id}흐름 삭제

기본 시드 흐름 (5개)

흐름카테고리단계
설비 모니터링equipment장비 상태 → 알람 이력 → OEE 현황 → PM 일정
WIP 관리wipWIP 현황 → 병목 구간 → LOT 추적 → 디스패치 이력
물류 모니터링logistics컨베이어 부하율 → 반송 현황 → 스토커 상태 → 경로 통계
품질 관리quality수율 현황 → 불량 분석 → SPC 데이터
에너지 관리energy에너지 사용량 → 유틸리티 비용 → 에너지 효율

build_circle Tool Requests (도구 요청)

왜 필요한가

AI Chat에서 사용자가 질문했는데 적합한 도구가 없으면, AI가 자동으로 도구 추가 요청을 생성합니다. 관리자가 요청 목록을 확인하고 상태 관리 (대기 → 진행 중 → 완료/반려)하여 도구 확장 사이클을 체계적으로 운영합니다.

파일 구조

파일역할
api/tool_requests.pyREST CRUD API (5 endpoints)
core/db/queries/tool_requests.pyDB 쿼리
ui/pages/tool_requests.py관리 UI (상태 필터 + 상세 다이얼로그)
main.pyAI Chat 함수 등록 (_submit_tool_request)

API 엔드포인트

MethodPath설명
GET/api/tool-requests요청 목록 (상태 필터)
GET/api/tool-requests/{id}요청 상세
POST/api/tool-requests요청 생성 (AI Chat에서 자동)
PATCH/api/tool-requests/{id}상태/메모 변경
DELETE/api/tool-requests/{id}요청 삭제

상태 워크플로우

pending (대기) → in_progress (진행 중) → completed (완료) / rejected (반려)

history Activity Log (활동 로그)

왜 필요한가

누가 언제 무엇을 했는지 추적합니다. 규칙 변경, 이상 상태 변경, 도구 수정 등 주요 액션을 자동 기록하여 감사(audit) 추적과 문제 발생 시 원인 파악에 활용합니다.

파일 구조

파일역할
api/activity_log.py조회 API
core/db/queries/activity_log.pyDB 쿼리 (insert/list)
ui/pages/activity_log.py활동 로그 뷰어 UI

기록 대상

액션대상예시
create / update / deletedetection_rules규칙 생성/수정/삭제
status_changeanomalies이상 상태 변경 (detected → acknowledged)
create / update / deleteds_tools도구 추가/수정/삭제
loginusers사용자 로그인

play_circle Chat Scenarios (챗 매크로) v1.7.0

왜 필요한가

FAB 운영자가 반복적으로 실행하는 도구 조합이 있습니다. "일일 점검"이라고 입력하면 설비가동률 → 재공현황 → 이상이력을 순차 실행하고 종합 분석해야 합니다. 매번 같은 질문을 반복하는 대신, 키워드 트리거 기반 매크로로 도구 체인을 자동 실행합니다.

동작 방식

사용자: "일일 점검 실행해줘" │ ├── 키워드 매칭 (대화 히스토리 전체 검색) │ └── "일일 점검" → chat_scenarios에서 매칭 │ ├── 프롬프트 주입 │ └── 매크로 단계(steps)를 시스템 프롬프트에 추가 │ └── LLM이 매크로 단계에 따라 도구 순차 호출 + 종합 분석

파일 구조

파일역할
api/chat_scenarios.pyREST CRUD API
core/db/queries/chat_scenarios.pyDB 쿼리
ui/pages/chat_scenarios.py매크로 관리 UI

API 엔드포인트

MethodPath설명
GET/api/chat-scenarios매크로 목록
GET/api/chat-scenarios/{id}매크로 상세
POST/api/chat-scenarios매크로 생성
PUT/api/chat-scenarios/{id}매크로 수정
DELETE/api/chat-scenarios/{id}매크로 삭제

forum Chat History (채팅 이력) v1.7.0

왜 필요한가

관리자가 전체 사용자의 AI Chat 세션을 한눈에 조회하고 관리할 수 있어야 합니다. 사용 패턴 분석, 문제 세션 추적, 사용자 지원 등을 위해 별도의 관리자 전용 이력 페이지가 필요합니다.

파일 구조

파일역할
ui/pages/chat_history.py관리자 전체 세션 조회/관리 UI
api/ai_chat.py세션 히스토리 API (GET /api/ai-chat/sessions/history)

shield Countermeasures (대안/조치 플레이북) v1.7.0

왜 필요한가

이상이 탐지되었을 때 "그래서 어떻게 해야 하는가?"에 대한 표준 대응 절차가 필요합니다. 이상 유형(카테고리/심각도)별로 대안/조치 플레이북을 DB에 등록하면, 이상 발생 시 관련 대응 절차를 자동으로 제시할 수 있습니다.

파일 구조

파일역할
api/countermeasures.py대안/조치 CRUD API
core/db/queries/countermeasures.pyDB 쿼리
ui/pages/countermeasures.py플레이북 관리 UI (카테고리 필터 + 편집)

API 엔드포인트

MethodPath설명
GET/api/countermeasures대안 목록 (카테고리/심각도 필터)
GET/api/countermeasures/{id}대안 상세
POST/api/countermeasures대안 생성
PUT/api/countermeasures/{id}대안 수정
DELETE/api/countermeasures/{id}대안 삭제

주요 필드

필드설명
category이상 카테고리 (logistics, equipment, wip 등)
severity대상 심각도 (critical, warning, info)
rule_ids연결 규칙 ID (쉼표 구분)
action_type조치 유형 (manual / auto / workflow)
priority우선순위 (high / medium / low)
verification조치 후 검증 방법

이 문서는 FLOPI 전체 모듈의 설계 과정과 아키텍처 결정을 기록합니다.
System Settings Design Document도 참고하세요.

© 2026 FLOPI — Semiconductor FAB AI Platform