feat: Phase 6 보강 도구 5종 (find_tags, query_events, active_alarms, summarize_events, generate_status_report)

이벤트 중심 도구와 LLM 요약/보고서 도구를 추가해 채팅에서 "활성 알람", "교대 보고서", "이벤트 요약" 같은 운전원 요청을 처리. 신규 MCP 도구 (mcp-server/server.py): - find_tags(query, area?, top_k): v_tag_summary 뷰 기반. base_tag 또는 description ILIKE 매칭. PV/SP/OP/설명/area 함께 반환. - query_events(tag_name?, event_type?, area?, since?, until?, limit): event_history_table 필터 조회. since/until 미지정 시 최근 24h. event_type은 ALARM/TRIP/NORMAL/RUN/CHANGE 5종. - active_alarms(area?, limit): DISTINCT ON (tagname)으로 태그별 최신 이벤트 추출 후 ALARM/TRIP만 반환 (NORMAL 들어왔으면 자동 해제). - summarize_events(since?, area?, event_type?, max_events, focus?): query_events 결과를 LLM에 넣어 한국어 6~10줄 구조화 요약 (현황/알람/패턴/권고) + by_type/by_area 통계. - generate_status_report(area?, hours): 활성 알람 + 최근 이벤트 통계/표본을 LLM에 넘겨 교대 보고서 형식(요약/알람/이벤트분석/권고) 마크다운 생성. 윈도우 1~168시간, 기본 24시간. 공통: - prepared statement(파라미터 바인딩)로 SQL 인젝션 방지 - SET statement_timeout = SQL_STATEMENT_TIMEOUT_MS 적용 - _DB_SCHEMA에 event_history_table 정의 추가 (NL2SQL 인지용) 시스템 프롬프트 (OllamaController): - ToolGuideKo에 신규 5종 + search_kb + event_type 5종 명시 - run_sql 자동 가드(LIMIT/timeout) 안내 추가 검증: - dotnet build: 경고 0건, 에러 0건 - python3 -m py_compile: OK - import server 후 5개 도구 attribute 확인 런타임: - mcp-server 재시작 시 신규 도구 자동 인식 - 클라이언트는 ListToolsAsync로 동적 수집 — 추가 작업 불필요 Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-14 05:24:36 +09:00
parent 5a9d60e8a8
commit d09ef95869
3 changed files with 467 additions and 7 deletions
--- a/mcp-server/server.py
+++ b/mcp-server/server.py
@@ -382,6 +382,20 @@ PostgreSQL 시계열 데이터베이스 스키마
  node_id     TEXT         - OPC UA 노드 ID
  loaded_at   TIMESTAMPTZ  - 마지막 로드 시각

+테이블: event_history_table  (디지털 포인트 상태 변경 이벤트)
+  id               BIGSERIAL    - PK
+  tagname          TEXT         - 태그명 (소문자)
+  node_id          TEXT
+  prev_value       TEXT         - 직전 값
+  curr_value       TEXT         - 현재 값
+  event_type       TEXT         - 'ALARM' / 'TRIP' / 'NORMAL' / 'RUN' / 'CHANGE'
+  event_time       TIMESTAMPTZ  - 이벤트 발생 시각(UTC)
+  area             TEXT         - tag_metadata.area 복사본
+  section          TEXT         - 태그명 패턴에서 추출한 차수(예: '6-1차')
+  duration_seconds INT          - 직전 상태에서 머문 시간
+  metadata         JSONB        - 부가 정보 (interlock 등)
+  created_at       TIMESTAMPTZ
+
 뷰: v_tag_summary  (실시간값 + 메타데이터 통합 뷰)
  base_tag          TEXT   - 기본 태그명
  pv                TEXT   - 현재 프로세스 값
@@ -890,6 +904,392 @@ async def query_with_nl(question: str) -> str:
    return json.dumps(result, ensure_ascii=False, default=str)


+# ── Phase 6 보강 도구 (이벤트/태그/보고서) ─────────────────────────────────────
+
+_VALID_EVENT_TYPES = ("ALARM", "TRIP", "NORMAL", "RUN", "CHANGE")
+
+
+@mcp.tool()
+async def find_tags(query: str, area: str | None = None, top_k: int = 20) -> str:
+    """태그 검색 — base_tag/설명(desc)/area 통합 검색 (v_tag_summary 뷰 기반).
+
+    사용 시점: 사용자가 "온도", "Tower 1 압력", "운전 중인 펌프" 같은 자연어로
+    태그를 지칭할 때 실제 base_tag(예: 'ti-6101', 'p-6102')를 역으로 찾기 위해.
+
+    get_tag_metadata와 차이: 단순 tagname LIKE만 보지 않고 description/area에도
+    매칭하며, 현재 PV/SP/OP/description/area를 함께 반환.
+
+    Args:
+        query: 검색어 (base_tag 또는 description 부분 일치, 대소문자 무시)
+        area: (선택) area 필터 (예: 'tower-1', 'utility'). NULL이면 전체
+        top_k: 반환 태그 수 (기본 20, 최대 100)
+
+    Returns:
+        JSON: { success, query, count, tags: [{base_tag, pv, sp, op, description, area}] }
+    """
+    conn = None
+    try:
+        top_k = max(1, min(top_k, 100))
+        q = f"%{query.strip()}%"
+        conn = await _get_db_connection()
+        with conn.cursor() as cur:
+            cur.execute(f"SET statement_timeout = {SQL_STATEMENT_TIMEOUT_MS}")
+            if area:
+                cur.execute(
+                    """SELECT base_tag, pv, sp, op, description, area
+                       FROM v_tag_summary
+                       WHERE (base_tag ILIKE %s OR description ILIKE %s)
+                         AND area = %s
+                       ORDER BY base_tag
+                       LIMIT %s""",
+                    (q, q, area, top_k)
+                )
+            else:
+                cur.execute(
+                    """SELECT base_tag, pv, sp, op, description, area
+                       FROM v_tag_summary
+                       WHERE base_tag ILIKE %s OR description ILIKE %s
+                       ORDER BY base_tag
+                       LIMIT %s""",
+                    (q, q, top_k)
+                )
+            rows = cur.fetchall()
+        tags = [
+            {"base_tag": r[0], "pv": r[1], "sp": r[2], "op": r[3],
+             "description": r[4], "area": r[5]}
+            for r in rows
+        ]
+        return json.dumps({
+            "success": True, "query": query, "area": area,
+            "count": len(tags), "tags": tags
+        }, ensure_ascii=False, indent=2)
+    except Exception as e:
+        return json.dumps({"success": False, "error": f"태그 검색 실패: {e}"}, ensure_ascii=False)
+    finally:
+        if conn:
+            conn.close()
+
+
+@mcp.tool()
+async def query_events(
+    tag_name: str | None = None,
+    event_type: str | None = None,
+    area: str | None = None,
+    since: str | None = None,
+    until: str | None = None,
+    limit: int = 100,
+) -> str:
+    """이벤트 히스토리 조회 (event_history_table — 디지털 포인트 상태 변경).
+
+    Args:
+        tag_name:   (선택) 태그명 LIKE 패턴 (예: 'p-6102', 'xv-%')
+        event_type: (선택) ALARM / TRIP / NORMAL / RUN / CHANGE 중 하나
+        area:       (선택) area 정확 매칭
+        since:      (선택) 시작 시간 ISO 8601. 기본 24시간 전
+        until:      (선택) 종료 시간 ISO 8601. 기본 현재
+        limit:      반환 행 수 (기본 100, 최대 1000)
+
+    Returns:
+        JSON: { success, count, time_range, events: [...] }
+    """
+    if event_type and event_type.upper() not in _VALID_EVENT_TYPES:
+        return json.dumps({
+            "success": False,
+            "error": f"event_type은 {list(_VALID_EVENT_TYPES)} 중 하나여야 합니다."
+        }, ensure_ascii=False)
+
+    conn = None
+    try:
+        limit = max(1, min(limit, 1000))
+        where = ["event_time >= COALESCE(%s::timestamptz, NOW() - INTERVAL '24 hours')",
+                 "event_time <= COALESCE(%s::timestamptz, NOW())"]
+        params: list = [since, until]
+        if tag_name:
+            where.append("tagname ILIKE %s")
+            params.append(f"%{tag_name}%")
+        if event_type:
+            where.append("event_type = %s")
+            params.append(event_type.upper())
+        if area:
+            where.append("area = %s")
+            params.append(area)
+
+        sql = f"""
+            SELECT id, tagname, prev_value, curr_value, event_type, event_time,
+                   area, section, duration_seconds, metadata
+            FROM event_history_table
+            WHERE {' AND '.join(where)}
+            ORDER BY event_time DESC
+            LIMIT %s
+        """
+        params.append(limit)
+
+        conn = await _get_db_connection()
+        with conn.cursor() as cur:
+            cur.execute(f"SET statement_timeout = {SQL_STATEMENT_TIMEOUT_MS}")
+            cur.execute(sql, params)
+            rows = cur.fetchall()
+        events = [
+            {"id": r[0], "tag_name": r[1], "prev_value": r[2], "curr_value": r[3],
+             "event_type": r[4], "event_time": r[5].isoformat() if r[5] else None,
+             "area": r[6], "section": r[7], "duration_seconds": r[8], "metadata": r[9]}
+            for r in rows
+        ]
+        return json.dumps({
+            "success": True,
+            "count": len(events),
+            "time_range": f"{since or '24h ago'} ~ {until or 'now'}",
+            "filters": {"tag_name": tag_name, "event_type": event_type, "area": area},
+            "events": events,
+        }, ensure_ascii=False, indent=2, default=str)
+    except Exception as e:
+        return json.dumps({"success": False, "error": f"이벤트 조회 실패: {e}"}, ensure_ascii=False)
+    finally:
+        if conn:
+            conn.close()
+
+
+@mcp.tool()
+async def active_alarms(area: str | None = None, limit: int = 100) -> str:
+    """현재 활성 알람 — 각 태그의 최신 이벤트가 ALARM 또는 TRIP인 경우만 반환.
+
+    동작: event_history_table에서 태그별 최신 이벤트(DISTINCT ON)를 가져온 뒤,
+    event_type이 ALARM 또는 TRIP인 것만 필터. 이후 NORMAL이 들어왔다면 해제된 것이므로 제외.
+
+    Args:
+        area: (선택) area 필터
+        limit: 최대 반환 수 (기본 100)
+
+    Returns:
+        JSON: { success, count, alarms: [{tag_name, event_type, since, duration_seconds, area, ...}] }
+    """
+    conn = None
+    try:
+        limit = max(1, min(limit, 500))
+        sql = """
+            WITH latest AS (
+              SELECT DISTINCT ON (tagname)
+                     id, tagname, curr_value, event_type, event_time,
+                     area, section, duration_seconds, metadata
+              FROM event_history_table
+              WHERE event_time >= NOW() - INTERVAL '30 days'
+              ORDER BY tagname, event_time DESC
+            )
+            SELECT id, tagname, curr_value, event_type, event_time,
+                   area, section, duration_seconds, metadata,
+                   EXTRACT(EPOCH FROM (NOW() - event_time))::bigint AS age_seconds
+            FROM latest
+            WHERE event_type IN ('ALARM', 'TRIP')
+              AND (%s::text IS NULL OR area = %s)
+            ORDER BY event_time DESC
+            LIMIT %s
+        """
+        conn = await _get_db_connection()
+        with conn.cursor() as cur:
+            cur.execute(f"SET statement_timeout = {SQL_STATEMENT_TIMEOUT_MS}")
+            cur.execute(sql, (area, area, limit))
+            rows = cur.fetchall()
+        alarms = [
+            {"id": r[0], "tag_name": r[1], "curr_value": r[2], "event_type": r[3],
+             "since": r[4].isoformat() if r[4] else None,
+             "area": r[5], "section": r[6], "duration_seconds": r[7], "metadata": r[8],
+             "age_seconds": r[9]}
+            for r in rows
+        ]
+        return json.dumps({
+            "success": True, "area": area,
+            "count": len(alarms), "alarms": alarms,
+        }, ensure_ascii=False, indent=2, default=str)
+    except Exception as e:
+        return json.dumps({"success": False, "error": f"활성 알람 조회 실패: {e}"}, ensure_ascii=False)
+    finally:
+        if conn:
+            conn.close()
+
+
+@mcp.tool()
+async def summarize_events(
+    since: str | None = None,
+    area: str | None = None,
+    event_type: str | None = None,
+    max_events: int = 200,
+    focus: str = "",
+) -> str:
+    """이벤트 히스토리를 LLM으로 한글 요약.
+
+    내부 동작: query_events로 이벤트를 가져와 LLM에 요약 요청. 큰 건수는 잘림.
+
+    Args:
+        since:      (선택) ISO 8601 시작 시간. 기본 24시간 전
+        area:       (선택) area 필터
+        event_type: (선택) event_type 필터
+        max_events: 분석에 포함할 최대 이벤트 (기본 200, 최대 500)
+        focus:      (선택) 요약 시 강조할 관점 (예: "interlock 동작", "교대 시점 이상")
+
+    Returns:
+        JSON: { success, summary, stats: {by_type, by_area, count} }
+    """
+    max_events = max(10, min(max_events, 500))
+    raw = await query_events(event_type=event_type, area=area, since=since, limit=max_events)
+    parsed = json.loads(raw)
+    if not parsed.get("success"):
+        return raw
+
+    events = parsed.get("events", [])
+    if not events:
+        return json.dumps({
+            "success": True, "summary": "지정된 조건 범위에 이벤트가 없습니다.",
+            "stats": {"count": 0}
+        }, ensure_ascii=False, indent=2)
+
+    # 통계
+    by_type: dict[str, int] = {}
+    by_area: dict[str, int] = {}
+    for ev in events:
+        by_type[ev["event_type"]] = by_type.get(ev["event_type"], 0) + 1
+        a = ev.get("area") or "(unknown)"
+        by_area[a] = by_area.get(a, 0) + 1
+
+    # LLM 요약 — 가벼운 토큰 수로 제한
+    sample = events[:max_events]
+    bullet_lines = [
+        f"- [{ev['event_type']}] {ev['tag_name']} @ {ev['event_time']}"
+        f" ({ev.get('prev_value')}→{ev.get('curr_value')})"
+        f"{(' area=' + ev['area']) if ev.get('area') else ''}"
+        for ev in sample
+    ]
+    focus_line = f"\n특히 다음 관점을 우선해 설명하세요: {focus}\n" if focus else ""
+
+    system = (
+        "당신은 IIoT/공장 운전 분석 전문가입니다. 디지털 포인트의 상태 변경 이벤트 로그를 보고 "
+        "한국어로 6~10줄 요약을 만듭니다. 다음 구조를 따릅니다:\n"
+        "1) 핵심 현황 (총 이벤트 수, 주요 area)\n"
+        "2) 알람/트립 (ALARM/TRIP) 핵심 케이스 — 태그·시각·전후값\n"
+        "3) 패턴/특이점 (반복 발생, 동시 발생, area 집중)\n"
+        "4) 다음 점검 권고 (있다면)\n"
+        "구체적인 태그명과 시각을 포함하되 추측은 자제합니다."
+    )
+    user_msg = (
+        f"분석 대상 이벤트 {len(sample)}건 (전체 {parsed.get('count')}건). "
+        f"통계: type={by_type}, area={by_area}.{focus_line}\n\n"
+        + "\n".join(bullet_lines)
+    )
+
+    def _call():
+        return _llm().chat.completions.create(
+            model=VLLM_MODEL,
+            messages=[
+                {"role": "system", "content": system},
+                {"role": "user",   "content": user_msg},
+            ],
+            max_tokens=1200,
+            temperature=0.2,
+        )
+
+    try:
+        resp = await asyncio.to_thread(_call)
+        summary = resp.choices[0].message.content or "(요약 없음)"
+    except Exception as e:
+        summary = f"(LLM 요약 실패: {e})"
+
+    return json.dumps({
+        "success": True,
+        "summary": summary,
+        "stats": {"count": parsed.get("count"), "by_type": by_type, "by_area": by_area},
+        "time_range": parsed.get("time_range"),
+    }, ensure_ascii=False, indent=2)
+
+
+@mcp.tool()
+async def generate_status_report(area: str | None = None, hours: int = 24) -> str:
+    """공장 운전 상태 종합 보고서 — 활성 알람 + 최근 이벤트 + 추세를 LLM이 한 장으로 정리.
+
+    Args:
+        area:  (선택) area 필터 (전체 공장이면 NULL)
+        hours: 이벤트 분석 윈도우 (기본 24시간, 최대 168=7일)
+
+    Returns:
+        JSON: { success, report, sections: {active_alarms, recent_events, by_type}, generated_at }
+    """
+    hours = max(1, min(hours, 168))
+    since_iso = None  # query_events가 24h 기본을 쓰지만, hours로 명시 전달
+    from datetime import datetime, timezone, timedelta
+    since_iso = (datetime.now(timezone.utc) - timedelta(hours=hours)).isoformat()
+
+    # 1) 활성 알람
+    alarms_raw = await active_alarms(area=area, limit=50)
+    alarms = json.loads(alarms_raw).get("alarms", [])
+
+    # 2) 최근 이벤트 통계
+    events_raw = await query_events(area=area, since=since_iso, limit=500)
+    events_parsed = json.loads(events_raw)
+    events = events_parsed.get("events", [])
+    by_type: dict[str, int] = {}
+    for ev in events:
+        by_type[ev["event_type"]] = by_type.get(ev["event_type"], 0) + 1
+
+    # 3) LLM 보고서
+    alarm_lines = [
+        f"- [{a['event_type']}] {a['tag_name']} since {a['since']} "
+        f"({a.get('age_seconds', 0)}s){' area=' + a['area'] if a.get('area') else ''}"
+        for a in alarms[:30]
+    ] or ["- 활성 알람 없음"]
+    recent_lines = [
+        f"- [{ev['event_type']}] {ev['tag_name']} @ {ev['event_time']} "
+        f"({ev.get('prev_value')}→{ev.get('curr_value')})"
+        for ev in events[:40]
+    ] or ["- 최근 이벤트 없음"]
+
+    system = (
+        "당신은 공장 운전 교대 보고서를 작성하는 운전원 보조 AI입니다. "
+        "다음 형식으로 한국어 보고서를 작성하세요. 마크다운 사용 가능.\n\n"
+        "# 운전 상태 종합 보고서\n\n"
+        "## 1. 요약\n(2~3줄로 핵심 상황)\n\n"
+        "## 2. 활성 알람\n(있으면 표 또는 bullet, 없으면 \"없음\")\n\n"
+        "## 3. 최근 N시간 이벤트 분석\n(주요 패턴, 빈번한 태그, 동시 발생)\n\n"
+        "## 4. 권고 조치\n(점검할 태그/area, 우선순위)\n\n"
+        "수치는 통계 데이터를 그대로 인용하고, 추측은 명시적으로 표시하세요."
+    )
+    user_msg = (
+        f"대상 area: {area or '전체'}\n"
+        f"분석 윈도우: 최근 {hours}시간\n"
+        f"이벤트 통계 (type별): {by_type}\n"
+        f"활성 알람 {len(alarms)}건:\n" + "\n".join(alarm_lines) + "\n\n"
+        f"최근 이벤트 표본 (최대 40건):\n" + "\n".join(recent_lines)
+    )
+
+    def _call():
+        return _llm().chat.completions.create(
+            model=VLLM_MODEL,
+            messages=[
+                {"role": "system", "content": system},
+                {"role": "user",   "content": user_msg},
+            ],
+            max_tokens=2048,
+            temperature=0.2,
+        )
+
+    try:
+        resp = await asyncio.to_thread(_call)
+        report = resp.choices[0].message.content or "(보고서 생성 실패)"
+    except Exception as e:
+        report = f"(LLM 보고서 실패: {e})"
+
+    from datetime import datetime as _dt, timezone as _tz
+    return json.dumps({
+        "success": True,
+        "report": report,
+        "sections": {
+            "active_alarms_count": len(alarms),
+            "recent_events_count": len(events),
+            "by_type": by_type,
+            "window_hours": hours,
+            "area": area,
+        },
+        "generated_at": _dt.now(_tz.utc).isoformat(),
+    }, ensure_ascii=False, indent=2)
+
+
 # ── P&ID 추출 도구 ──────────────────────────────────────────────────────────────

@mcp.tool()