FLOPI Modules
반도체 FAB 폐쇄망 환경의 통합 AI 플랫폼.
각 모듈의 목적, 필요성, 아키텍처 결정을 기록합니다.
precision_manufacturing FAB AI Platform
widgets 20개 모듈
code Python 3.11
calendar_today 2026-03-26
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로 노출
기술 선택 근거
| 선택 | 이유 | 대안 |
| FastAPI | async/await, 자동 OpenAPI docs, SSE 지원 | Flask, Django (sync 기반) |
| NiceGUI | Python만으로 풀스택 UI, React 학습 불필요 | Streamlit (새로고침 모델), Gradio (ML 특화) |
| APScheduler | in-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,600 | 170+ REST 엔드포인트 (20 라우터) |
core/ | 47 | ~7,300 | DB, LLM, RAG, Auth, Simulator, Workflow, Migration, Audit, Code Interpreter, Settings |
detection/ | 14 | ~1,300 | 규칙 평가 + LLM 에이전트 분석 + 자동 해소 |
rca/ | 12 | ~3,000 | AI 대화 → Mermaid 플로차트 + 워크플로우 변환 |
data_studio/ | 8 | ~2,100 | No-Code SQL 도구 빌더 |
ui/ | 29 | ~16,100 | NiceGUI 대시보드 (20 페이지) |
| 진입점 | 4 | ~3,200 | main.py, config.py, init_db.py, mcp_server.py |
| 합계 | 140 | ~39,400 | |
dashboard Dashboard
왜 필요한가
FAB 현장 관리자는 한 화면에서 전체 상황을 1초 안에 파악해야 합니다.
이상이 몇 건인지, 심각도는 어떤지, 마지막 탐지 사이클은 정상이었는지를
여러 페이지를 돌아다니며 확인하는 것은 비효율적입니다.
구성 요소
활성 위험: critical severity + detected/in_progress 상태인 이상 수.
활성 경고: warning severity 이상 수.
총 이상: 24시간 이내 발생한 모든 이상.
활성 규칙: enabled=1인 detection_rules 수.
빨강(감지됨) / 노랑(처리중) / 초록(해결) 3색 도넛.
다크/라이트 모드에 따라 배경색을 동적으로 적용합니다.
detection_cycles 테이블의 최신 1건을 조회하여 표시.
operator 이상 역할에게만 "수동 감지 실행" 버튼을 노출합니다.
설계 결정
- 30초 자동 갱신:
ui.timer(30.0, refresh)로 API 연결 상태 표시를 자동 갱신합니다. 데이터 자체는 페이지 진입 시 1회 + 수동 감지 후 갱신.
@ui.refreshable: NiceGUI 패턴으로, 수동 감지 실행 후 대시보드 전체가 자동 재렌더링됩니다.
- 최근 이상 5건: severity_badge + status_badge + 감지시각으로 즉시 식별 가능한 카드 형태.
파일 구조
| 파일 | 역할 |
api/dashboard.py | 3개 API 엔드포인트 (overview, timeline, heatmap) |
ui/pages/dashboard.py | NiceGUI 대시보드 UI (KPI 카드 + 차트 + 최근 이상) |
core/db/queries/dashboard.py | 집계 쿼리 (severity 분류, 24h 타임라인) |
API
| Method | Path | 설명 |
GET | /api/dashboard/overview | KPI 집계 + 마지막 사이클 + 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로 처리하여
오탐을 줄입니다.
데이터 소스 분기
source_type=sql: query_template SQL을 Oracle/SQLite에 직접 실행
source_type=tool: tool_registry.dispatch()로 Tool Studio 도구 또는 내장 도구 호출
라이프사이클 관리
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.py의 check_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 엔드포인트
| Method | Path | 설명 |
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,
query → query_template, tool → tool_name.
네이밍 컨벤션이 다르지만 로더가 자동 변환합니다.
89개 내장 FAB 도구 카탈로그
규칙 생성 UI에서 선택 가능한 사전 정의 도구 목록입니다.
| 카테고리 | 도구 | 설명 |
| logistics | get_conveyor_load | 존별 컨베이어 부하율(%) |
get_transfer_throughput | 라인별 반송 처리량 |
get_bottleneck_zones | 대기시간 초과 병목존 |
get_agv_utilization | AGV/OHT 상태별 대수/비율 |
get_flow_balance | 공정별 유입/유출 밸런스 |
| wip | get_wip_levels | 공정별 WIP vs 목표 비율 |
get_queue_length | 스텝별 대기 LOT 수 |
get_aging_lots | 기준시간 초과 장기체류 LOT |
get_wip_trend | 시간별 WIP 변화 트렌드 |
| equipment | get_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 엔드포인트
| Method | Path | 설명 |
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 구조
- 상단 요약: 마지막 사이클 (규칙 평가수 / 이상 감지수 / 소요시간 / 시작시각) + 수동 감지 버튼
- 감지 사이클 이력: 최근 20건 테이블 — 시작/완료/소요/상태 컬럼
- 이상 이력 탭: 감지됨 / 처리중 / 해결 3탭, 각 탭에 건수 표시
상태 전이 매트릭스
| 현재 상태 | 전이 가능 |
detected | in_progress, resolved, dismissed |
in_progress | resolved, dismissed, detected |
resolved | detected (재발 시) |
dismissed | detected (재발 시) |
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 "재발 시 자동 재오픈"
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 프롬프트로 전체 그래프를 검증합니다:
- 빠진 분기: decision 노드에 정상/비정상 두 분기 모두 있는지
- 고립 노드: 들어오는/나가는 엣지 모두 없는 노드
- 끊긴 경로: start에서 도달 불가능한 노드
- 판단 기준 누락: decision에 임계값/조건이 없음
검증 결과는 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개)
| Method | Path | 설명 |
| 세션 관리 |
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}/chat | AI 대화 (SSE 스트리밍) |
POST | /api/deep-rca/sessions/{id}/validate | AI 플로차트 검증 |
| 시나리오 관리 |
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-tools | FAB 장비 목록 |
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_transform | 틸 | eval() 데이터 변환/필터 (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)
- 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 | ~310 | DAG 실행 엔진 (Kahn's algorithm, 12노드 지원) |
core/workflow/generator.py | ~180 | AI 워크플로우 생성기 — 자연어 → 워크플로우 JSON |
api/workflow.py | ~240 | 워크플로우 CRUD + 실행 + 이벤트 트리거 API |
core/db/queries/workflow.py | — | workflows + workflow_runs 쿼리 |
API 엔드포인트 (11개)
| Method | Path | 설명 |
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}/export | JSON 내보내기 |
POST | /api/workflows/import | JSON 가져오기 |
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_search는 always_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개)
| Method | Path | 설명 |
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}/structure | LLM 기반 문서 구조화 (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_BACKEND | gemini | 임베딩 백엔드 (gemini / openai / custom) |
RAG_EMBED_BASE_URL | — | 임베딩 API base URL |
RAG_EMBED_API_KEY | — | 임베딩 API 키 |
RAG_EMBED_MODEL | text-embedding-004 | 임베딩 모델명 |
RAG_CHROMA_DIR | FLOPI/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개 이상의 도구가 등록될 것을 대비한 시맨틱 도구 선별기입니다.
- Short-circuit: 도구 수 ≤ top_k(15)이면 전체 반환 (임베딩 호출 없음)
- always_on 도구: knowledge_base_search 등은 무조건 포함
- 쿼리 임베딩: 사용자 질문을 벡터화
- ChromaDB 검색: tool_embeddings 컬렉션에서 top_k×2 후보 검색
- 카테고리 다양성: 카테고리당 max_per_category(5)개까지만 허용
check_circle
자동 동기화: registry.on_change(selector.mark_dirty) —
도구 등록/해제 시 자동으로 dirty 마킹, 다음 select_tools() 호출 시 재동기화.
API 엔드포인트 (11개)
| Method | Path | 설명 |
POST | /api/ai-chat | SSE 스트리밍 대화 (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/history | Admin 채팅 이력 (필터) |
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 (사용자 개인 프롬프트)
| Method | Path | 설명 |
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_call → tool_result 이벤트
│ └── tool_calls 없음? → 최종 응답 단계로
│
├── 최종 답변: LLM.chat_stream() → 토큰 단위 SSE
│ └── SSE: token → token → ... → 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 엔드포인트
| Method | Path | 설명 |
GET | /api/migration/status | 블록별 이관 상태 조회 |
POST | /api/migration/toggle | 블록 이관 상태 전환 (oracle/sqlite) |
POST | /api/migration/compare | Oracle 스키마 비교 (누락 테이블/컬럼) |
POST | /api/migration/ddl | Oracle 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 구성
- KPI 카드: 총 호출 수 / 총 토큰 / 입력 토큰 / 출력 토큰
- 일별 추이 차트: ECharts 듀얼 Y축 (호출 수 + 토큰K)
- 모듈별 테이블: detection, deep_rca, ai_chat, workflow 등 한국어 라벨 매핑
- 최근 50건: 시간/모듈/모델명/토큰/소요시간, 검색 지원
API 엔드포인트
| Method | Path | 설명 |
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_URL | Gemini endpoint | API 베이스 URL (OpenAI-compatible) |
LLM_API_KEY | — | API 키 |
LLM_MODEL | gemini-2.0-flash | 모델명 |
LLM_TIMEOUT | 60 | 타임아웃 (초) |
LLM_MAX_TOKENS | — | 최대 출력 토큰 |
group Users & RBAC
왜 필요한가
FAB 환경에서 관리자/엔지니어/운영자/열람자의 접근 권한이 명확히 분리되어야 합니다.
Tool Studio 설정을 운영자가 변경하거나, 열람자가 이상 상태를 변경하는 것은
심각한 문제를 야기할 수 있습니다.
4단계 RBAC
| 역할 | 접근 범위 | 제한 |
| admin | 전체 full | 없음 |
| engineer | 대부분 full | Tool Studio, Users 접근 불가 |
| operator | 기본 조회 + 이상 상태변경 | RCA/Workflow/Tool Studio 불가 |
| viewer | 읽기 전용 | 모든 쓰기 불가 |
권한 레벨
full — CRUD 전체
view — 읽기만
state_change — 이상 상태 변경만 (operator 전용)
None — 접근 불가 (사이드바에서도 숨김)
커스텀 오버라이드
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 | 사용자 쿼리 (해시 비밀번호, 세션 토큰) |
모듈별 권한 매트릭스 (상세)
| 모듈 | admin | engineer | operator | viewer |
| dashboard | full | full | full | full |
| anomalies | full | full | state_change | view |
| rules | full | full | view | view |
| detection_logs | full | full | full | view |
| deep_rca | full | full | — | — |
| data_studio | full | — | — | — |
| workflow | full | full | — | — |
| knowledge_base | full | full | view | view |
| ai_chat | full | full | view | view |
| llm_usage | full | view | — | — |
| users | full | — | — | — |
| system_settings | full | — | — | — |
API 엔드포인트 (9개)
| Method | Path | 설명 |
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.py | MCP 서버 진입점 (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"]
}
}
}
--sqlite <path> — 시뮬레이터 모드 (Oracle 불필요)
- 플래그 없음 — 운영 모드 (Oracle 연결 필수)
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.py의 oracle_to_sqlite(sql) 함수가 Oracle 문법을 SQLite로 변환:
| Oracle | SQLite | 비고 |
SYSTIMESTAMP | CURRENT_TIMESTAMP | 현재 시각 |
SYSDATE | DATE('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 :out | lastrowid | 삽입 후 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 Chat | SmartToolSelector (시맨틱) | LLM이 tool_call 결정 | SSE 스트리밍 응답 |
| 이상탐지 | 규칙의 tool_name 고정 | 스케줄러 자동 실행 | DB anomaly INSERT |
| 워크플로우 | data_query 노드 설정 | DAG 순서대로 | context[node_id]에 저장 |
| MCP | Claude가 list_tools 조회 | Claude의 tool_use | JSON 텍스트 반환 |
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 엔드포인트
| Method | Path | 설명 |
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초) |
주요 설정 항목
| 카테고리 | 키 | 설명 | 기본값 |
| LLM | llm.timeout | 타임아웃 (초) | 60 |
llm.temperature | 온도 파라미터 | 0.7 |
| 감지 | detection.interval | 틱 간격 (초) | 60 |
detection.max_concurrent | 병렬 평가 수 | 5 |
| DB | oracle.dsn | Oracle DSN | config.py |
warning
admin 전용: 시스템 설정 변경은 admin 역할만 가능합니다.
engineer/operator/viewer는 접근 자체가 차단됩니다 (사이드바에서 숨김).
quiz Question Flows (질문 흐름)
왜 필요한가
AI Chat 매 응답마다 LLM을 추가 호출해 후속질문을 생성하면 토큰 낭비 + 맥락 없는 질문이 됩니다.
FAB에선 질문 패턴이 정해져 있습니다 (설비 상태 → 알람 → OEE → PM).
관리자가 질문 흐름을 DB에 등록하면, LLM이 참고하여 자연스럽고 간결한 후속질문을 생성합니다.
하이브리드 방식
흐름이 매칭되면 → LLM에 흐름 컨텍스트 전달 → 다음 단계 기반 후속질문.
흐름이 없으면 → 기존처럼 자유 생성 (폴백).
키워드 매칭으로 현재 위치를 파악하고 남은 단계를 추출합니다.
파일 구조
| 파일 | 역할 |
api/question_flows.py | REST CRUD API (5 endpoints) |
core/db/queries/question_flows.py | DB 쿼리 (list/get/create/update/delete) |
ui/pages/question_flows.py | 관리 UI (카드 목록 + 카테고리 필터 + 편집 다이얼로그) |
api/ai_chat.py | 후속질문 생성 시 흐름 매칭 로직 (_match_flows, _build_flow_context) |
API 엔드포인트
| Method | Path | 설명 |
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 관리 | wip | WIP 현황 → 병목 구간 → LOT 추적 → 디스패치 이력 |
| 물류 모니터링 | logistics | 컨베이어 부하율 → 반송 현황 → 스토커 상태 → 경로 통계 |
| 품질 관리 | quality | 수율 현황 → 불량 분석 → SPC 데이터 |
| 에너지 관리 | energy | 에너지 사용량 → 유틸리티 비용 → 에너지 효율 |
history Activity Log (활동 로그)
왜 필요한가
누가 언제 무엇을 했는지 추적합니다. 규칙 변경, 이상 상태 변경, 도구 수정 등
주요 액션을 자동 기록하여 감사(audit) 추적과 문제 발생 시 원인 파악에 활용합니다.
파일 구조
| 파일 | 역할 |
api/activity_log.py | 조회 API |
core/db/queries/activity_log.py | DB 쿼리 (insert/list) |
ui/pages/activity_log.py | 활동 로그 뷰어 UI |
기록 대상
| 액션 | 대상 | 예시 |
| create / update / delete | detection_rules | 규칙 생성/수정/삭제 |
| status_change | anomalies | 이상 상태 변경 (detected → acknowledged) |
| create / update / delete | ds_tools | 도구 추가/수정/삭제 |
| login | users | 사용자 로그인 |
play_circle Chat Scenarios (챗 매크로) v1.7.0
왜 필요한가
FAB 운영자가 반복적으로 실행하는 도구 조합이 있습니다. "일일 점검"이라고 입력하면
설비가동률 → 재공현황 → 이상이력을 순차 실행하고 종합 분석해야 합니다.
매번 같은 질문을 반복하는 대신, 키워드 트리거 기반 매크로로 도구 체인을 자동 실행합니다.
동작 방식
사용자: "일일 점검 실행해줘"
│
├── 키워드 매칭 (대화 히스토리 전체 검색)
│ └── "일일 점검" → chat_scenarios에서 매칭
│
├── 프롬프트 주입
│ └── 매크로 단계(steps)를 시스템 프롬프트에 추가
│
└── LLM이 매크로 단계에 따라 도구 순차 호출 + 종합 분석
파일 구조
| 파일 | 역할 |
api/chat_scenarios.py | REST CRUD API |
core/db/queries/chat_scenarios.py | DB 쿼리 |
ui/pages/chat_scenarios.py | 매크로 관리 UI |
API 엔드포인트
| Method | Path | 설명 |
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.py | DB 쿼리 |
ui/pages/countermeasures.py | 플레이북 관리 UI (카테고리 필터 + 편집) |
API 엔드포인트
| Method | Path | 설명 |
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 | 조치 후 검증 방법 |