""" GUARDiA Chatbot Engine - Per-user session & conversation history - Ollama(sLLM) integration with smart fallback - Intent classification + domain knowledge base """ import asyncio import random import re from typing import Optional try: import httpx _HTTPX = True except ImportError: _HTTPX = False # ── Ollama 설정 ──────────────────────────────────────────────── OLLAMA_URL = "http://localhost:11434/api/chat" OLLAMA_MODEL = "llama3:8b-instruct-q4_K_M" SYSTEM_PROMPT = """당신은 GUARDiA 인프라 자동화 시스템의 AI 어시스턴트입니다. 1,000개 이상의 관공서 레거시 서버(WEB/WAS/DB) 운영, 배포 자동화, 장애 대응을 전문으로 합니다. - 응답은 한국어로, 간결하고 실용적으로 작성하세요. - 배포·운영 요청은 SR(Service Request) 절차를 안내하세요. - 서버 자격증명(IP, 비밀번호) 등 민감 정보는 절대 노출하지 마세요. - 모르는 내용은 솔직하게 모른다고 하세요.""" # ── 대화 세션 저장소 ─────────────────────────────────────────── # { user_id: [{"role": "user"|"assistant", "content": str}, ...] } chat_sessions: dict[str, list] = {} # ── Intent 패턴 ──────────────────────────────────────────────── _INTENTS = [ ("greeting", r"안녕|반가|하이|헬로|hello|hi\b|처음"), ("deploy", r"배포|deploy|올려|업로드|파일.*전송|전송.*파일"), ("restart", r"재기동|restart|리스타트|재시작|서비스.*시작|기동"), ("logs", r"로그|log|에러|error|오류|warn|exception|스택"), ("status", r"상태|status|모니터|현황|서버.*확인|확인.*서버|cpu|mem|메모리"), ("ssl", r"ssl|인증서|https|certificate|만료|expire"), ("disk", r"디스크|disk|용량|space|df\b|파티션"), ("db", r"db|데이터베이스|database|쿼리|query|락|lock|oracle|postgres"), ("cron", r"크론|cron|스케줄|schedule|배치|batch|정기"), ("security", r"보안|security|권한|permission|계정|접근|차단"), ("help", r"도움|help|명령|command|뭐.*할|어떻게|기능|사용법"), ("thanks", r"감사|고마|thank|ㄱㅅ|수고"), ("clear", r"초기화|대화.*지워|새로.*시작|clear|reset"), ] # ── 지식베이스 ───────────────────────────────────────────────── _KB: dict[str, list[str]] = { "greeting": [ "안녕하세요! GUARDiA 인프라 봇입니다.\n배포, 서버 운영, 장애 대응을 도와드립니다. 무엇을 도와드릴까요?", "반갑습니다! 오늘 어떤 작업을 도와드릴까요?", ], "deploy": [ ( "배포 작업을 안내해드리겠습니다. 다음 정보를 알려주세요:\n" "1. **대상 기관명** (예: 기재부, 국토부)\n" "2. **시스템명** (예: 예산시스템, 민원포털)\n" "3. **배포 레이어** — WEB / WAS\n" "4. **파일 종류** — class(동적) / html·js·css(정적)\n\n" "동적 파일(class)은 무중단 롤링 재기동이 자동 수행됩니다." ), ], "restart": [ ( "WAS 재기동 요청 절차:\n" "1. SR 자동 생성 → CMDB 서버 매핑\n" "2. PM 승인 대기\n" "3. 롤링 방식: WAS#1 재기동 → 헬스체크 → WAS#2 재기동\n\n" "대상 서버(기관명/시스템명)를 알려주시면 SR을 즉시 생성합니다." ), ], "logs": [ ( "로그 분석을 시작합니다. 어떤 서버의 로그인가요?\n" "• **WEB** (nginx/apache access·error log)\n" "• **WAS** (tomcat catalina.out / jboss server.log)\n" "• **DB** (alert log / slow query log)\n\n" "기관명과 서버 유형을 말씀해주세요." ), ], "status": [ ( "서버 현황 조회 결과 (모의 데이터):\n" "```\n" "WEB-01 ✅ 정상 CPU 12% MEM 45% DISK 38%\n" "WAS-01 ✅ 정상 CPU 34% MEM 62% DISK 55%\n" "WAS-02 ⚠️ 경고 CPU 87% MEM 78% DISK 55%\n" "DB-01 ✅ 정상 CPU 9% MEM 71% CONN 42/200\n" "DB-02 ✅ 정상 CPU 5% MEM 68% CONN 11/200\n" "```\n" "WAS-02 CPU가 임계치(80%)를 초과했습니다. 재기동을 권장합니다." ), ], "ssl": [ ( "SSL 인증서 만료일 현황 (모의 데이터):\n" "```\n" "portal.mof.go.kr D-14 ⚠️ 갱신 필요\n" "api.molit.go.kr D-87 ✅\n" "www.nts.go.kr D-203 ✅\n" "eis.mosf.go.kr D-7 🚨 긴급 갱신 필요\n" "```\n" "D-30 이하 인증서에 대해 갱신 SR을 생성해드릴까요?" ), ], "disk": [ ( "디스크 사용량 현황 (모의 데이터):\n" "```\n" "WAS-01 /app/logs 92% 🚨 즉시 정리 필요\n" "WAS-02 /app/logs 78% ⚠️\n" "DB-01 /data 61% ✅\n" "WEB-01 /var/log 43% ✅\n" "```\n" "WAS-01 로그 파티션이 임계치를 초과했습니다.\n" "30일 이상 된 로그를 자동 압축·이동할까요?" ), ], "db": [ ( "DB 상태 점검 항목:\n" "• **대기 쿼리 조회** — 현재 실행 중인 슬로우 쿼리\n" "• **락 분석** — 락 홀더·웨이터 확인\n" "• **연결 수** — 최대 연결 수 대비 현황\n" "• **테이블스페이스** — 사용량 조회\n\n" "어떤 항목을 확인하시겠습니까?" ), ], "cron": [ ( "크론 작업 관리 기능:\n" "• crontab 목록 조회\n" "• 특정 배치 실행 이력 확인\n" "• 배치 강제 실행 / 일시 중지\n\n" "대상 서버와 작업명을 알려주세요." ), ], "security": [ ( "GUARDiA 보안 정책:\n" "• SSH 접근: opsagent 전용 계정 사용, root 직접 접속 금지\n" "• 자격증명: AES-256 암호화 DB 저장\n" "• 명령 필터: 파괴적 명령(rm -rf /, drop table) 자동 차단\n" "• 감사 로그: SHA-256 해시 체이닝으로 위변조 방지\n\n" "특정 보안 정책에 대해 더 자세히 알려드릴까요?" ), ], "help": [ ( "**GUARDiA Bot 전체 기능**\n\n" "**배포**\n" "• 파일 배포 (class/html/js/img)\n" "• 무중단 롤링 재기동\n\n" "**운영**\n" "• 실시간 로그 분석\n" "• 서버 상태 모니터링\n" "• 디스크·DB·SSL 점검\n" "• 크론 작업 관리\n\n" "**이 채팅방에서는 자연어로 바로 말씀하세요.**\n" "예: `기재부 예산시스템 WAS 재기동해줘`" ), ], "thanks": [ "천만에요! 다른 도움이 필요하시면 언제든지 말씀해주세요.", "도움이 됐다니 다행입니다! 추가 작업이 있으시면 알려주세요.", ], "clear": [ "대화 내용을 초기화했습니다. 새로운 대화를 시작해주세요!", ], } # ── 공개 API ──────────────────────────────────────────────────── def clear_session(user_id: str) -> None: chat_sessions.pop(user_id, None) async def get_chatbot_response(user_id: str, text: str) -> str: """ 챗봇 응답 생성. 1순위: Ollama sLLM → 2순위: 의도 분류 + 지식베이스 """ if user_id not in chat_sessions: chat_sessions[user_id] = [] history = chat_sessions[user_id] history.append({"role": "user", "content": text}) # 세션 초기화 명령 intent = _classify(text) if intent == "clear": clear_session(user_id) return random.choice(_KB["clear"]) # Ollama 시도 (타임아웃 5초) if _HTTPX: reply = await _try_ollama(history) if reply: _append(user_id, "assistant", reply) return reply # 폴백: 지식베이스 + 컨텍스트 추론 reply = _fallback_response(intent, text, history) _append(user_id, "assistant", reply) return reply # ── 내부 헬퍼 ────────────────────────────────────────────────── def _classify(text: str) -> str: t = text.lower() for name, pattern in _INTENTS: if re.search(pattern, t): return name return "unknown" def _append(user_id: str, role: str, content: str) -> None: h = chat_sessions.setdefault(user_id, []) h.append({"role": role, "content": content}) if len(h) > 20: chat_sessions[user_id] = h[-20:] def _prev_intent(history: list) -> Optional[str]: for msg in reversed(history[:-1]): if msg["role"] == "user": return _classify(msg["content"]) return None def _fallback_response(intent: str, text: str, history: list) -> str: if intent in _KB: return random.choice(_KB[intent]) prev = _prev_intent(history) # 이전 대화 맥락 기반 추론 if prev in ("deploy", "restart", "logs"): # 기관명/시스템명 입력으로 간주 institutions = ["기재부", "국토부", "국세청", "복지부", "행안부", "교육부"] if any(inst in text for inst in institutions) or len(text) < 20: return ( f"'{text}' 정보를 확인했습니다.\n" "SR을 생성하겠습니다. 추가 정보가 있으시면 계속 말씀해주세요." ) if any(k in text for k in ["어떻게", "방법", "절차", "가이드"]): return ( "GUARDiA 작업 절차:\n" "1. 이 채팅에서 자연어로 요청\n" "2. Bot이 SR(Service Request) 자동 생성\n" "3. CMDB에서 대상 서버 자동 매핑\n" "4. PM 승인 (민감 작업)\n" "5. SSH/SFTP 에이전트리스 실행\n" "6. 결과 실시간 알림" ) return ( f"'{text[:30]}' 요청을 접수했습니다.\n" "좀 더 구체적으로 말씀해주시면 바로 처리해드리겠습니다.\n" "예: `기재부 예산시스템 WAS 재기동해줘`\n" "또는 `도움`을 입력하면 전체 기능을 안내해드립니다." ) async def _try_ollama(history: list) -> Optional[str]: """Ollama API 호출 — 실패 시 None.""" try: messages = [{"role": "system", "content": SYSTEM_PROMPT}] + history[-10:] async with httpx.AsyncClient(timeout=5.0) as client: resp = await client.post(OLLAMA_URL, json={ "model": OLLAMA_MODEL, "messages": messages, "stream": False, "options": {"temperature": 0.7}, }) if resp.status_code == 200: return resp.json()["message"]["content"] except Exception: pass return None