Files

windpacer 4eccb75193 plans: 새 작업 계획 문서들 추가 (OPUS-v2, P&ID bySonnet, TASK 문서)

2026-05-29 09:49:48 +09:00

12 KiB

Raw Permalink Blame History

작업지시서 — Phase B Verifier MVP (R1·R2·R4 tool-arg validators)

대상: 코딩 LLM. 이 문서만 보고 독립 수행 가능. 상위 플랜: plans/OPUS-v2-실행계획.md Phase B. 예상 분량: 새 모듈 1개(~150줄) + 기존 server.py 진입부에 데코레이터 적용. 로직 변경 없음.

1. 배경 (왜)

production opencode 채팅에서 Qwen3-8B(및 가끔 35B)가 존재하지 않는 태그·잘못된 area 형식으로 MCP 도구를 호출하고, 서버가 silent fallback해 모델이 그 결과를 사용자에게 잘못 포장하는 패턴 관찰:

trace_connections(start_tag="raw_material_input") ← 합성 (영어로 만든 가짜 태그)
trace_connections(start_tag="RM-6101") ← Raw Material + 실재 p-6101 차용 합성
generate_status_report(area="6-1") ← P6-1 형식 무시 → 서버가 무필터 전체 결과 반환 → "6-1차 결과"로 둔갑

Phase 0 eval은 통제 조건이라 이 패턴을 못 잡음. Verifier = 코드 결정적 검증을 MCP 도구 입구에 박아, 잘못된 호출 즉시 reject + 자기교정 hint 반환 → 모델이 find_tags 먼저 호출하는 행동으로 전환.

부수효과: 거부 로그가 Phase C(LoRA) 학습 데이터 자동 수집 역할 (잘못된 호출 → hint → 올바른 호출 trip).

2. 수정 대상

파일	동작
`mcp-server/verifier/__init__.py`	신규 (빈 패키지 마커)
`mcp-server/verifier/validators.py`	신규 — R1·R2 validator 함수 + VerifierError + 로그 적재
`mcp-server/verifier/test_validators.py`	신규 — 단위 테스트 (각 룰별 PASS/FAIL 2건씩 이상)
`mcp-server/verifier/logs/.gitkeep`	신규 (빈 디렉토리 유지)
`mcp-server/verifier/README.md`	신규 — 룰 카탈로그·로그 포맷
`mcp-server/server.py`	수정 — import + 적용 대상 tool 함수 입구에 검증 호출 (return 분기)

⚠️ 수정 금지: mcp-server/worker/*, mcp-server/eval/*, 채점기·골든셋·모델 서빙 스크립트.

3. 룰 명세 (정확히 이대로)

R1 — `tag-existence` (모든 tool 인자의 base_tag)

입력 패턴: 정규식 ^[a-z][a-z0-9]*-\d+[a-z]?(\.[a-z0-9]+)?$ (예: ficq-6113, ficq-6113.pv, p-6102, vp-6117a)
검증: . 앞부분(base_tag)이 tag_metadata.base_tag ∪ pid_equipment.tag_no (소문자 비교)에 존재해야 PASS
캐시: 모듈 전역 set, TTL 300초(5분). 캐시 미스 시 두 테이블 한번에 로드
실패 시 hint에 부분 매칭 top 3 suggested 첨부 (split('-') 토큰 길이 2 초과만)

R2 — `area-format` (`area`, `sub_area` 인자)

정규식: ^P\d+(-\d+)?$ (예: P6, P6-1). 빈 문자열/None은 PASS (선택 필터)
추가: - 앞 부분이 유효 area 코드여야 함: {"P1","P2","P3","P4","P5","P6","P8","P9","P10","UTIL","PACKING"} (P7 없음)
실패 hint: 형식 또는 미존재 모두 valid 코드 목록 포함

R4 — `trace_connections` 보강

start_tag: R1 적용 (위와 동일)
direction: {"upstream", "downstream"} 외 reject
max_depth: int, 1~50 외 reject

⚠️ R3·R5는 이번 MVP 범위 밖 (응답 텍스트 후처리/LLM-judge 필요). validators.py에 stub 함수만 두고 NotImplementedError 또는 NO-OP로 표시. Phase B.2에서 구현.

4. 구현 골자 (정확히 따를 것)

4.1 `mcp-server/verifier/validators.py`

"""Phase B Verifier MVP — tool 인자 결정적 검증.

룰: R1(tag-existence), R2(area-format), R4(trace_connections 보강).
응답 텍스트 검증(R3, R5)은 Phase B.2 — stub만.
"""
from __future__ import annotations
import json, re, time, pathlib
from typing import Optional, Any

TAG_RE  = re.compile(r'^[a-z][a-z0-9]*-\d+[a-z]?(\.[a-z0-9]+)?$')
AREA_RE = re.compile(r'^P\d+(-\d+)?$')
VALID_AREAS = {"P1","P2","P3","P4","P5","P6","P8","P9","P10","UTIL","PACKING"}
VALID_DIRECTIONS = {"upstream","downstream"}

_LOG_DIR = pathlib.Path(__file__).parent / "logs"

class VerifierError(Exception):
    def __init__(self, rule: str, code: str, hint: str, **extra):
        self.rule, self.code, self.hint, self.extra = rule, code, hint, extra
    def to_dict(self) -> dict:
        return {"verifier_error": f"{self.rule}.{self.code}", "hint": self.hint, **self.extra}

# ── 태그 캐시 ──
_tag_cache: set[str] | None = None
_tag_cache_at: float = 0.0

def _load_tag_set(get_conn) -> set[str]:
    global _tag_cache, _tag_cache_at
    if _tag_cache is not None and (time.time() - _tag_cache_at) < 300:
        return _tag_cache
    conn = get_conn()
    try:
        with conn.cursor() as cur:
            cur.execute("SELECT DISTINCT base_tag FROM tag_metadata WHERE base_tag IS NOT NULL")
            s = {r[0].lower() for r in cur.fetchall() if r[0]}
            cur.execute("SELECT DISTINCT tag_no FROM pid_equipment WHERE tag_no IS NOT NULL")
            s |= {r[0].lower() for r in cur.fetchall() if r[0]}
    finally:
        conn.close()
    _tag_cache, _tag_cache_at = s, time.time()
    return s

# ── R1 ──
def validate_tag(tag: str | None, get_conn) -> Optional[VerifierError]:
    if not tag:
        return None
    t = tag.lower()
    if not TAG_RE.match(t):
        return VerifierError("R1","invalid_tag_format",
            hint=f"태그 형식 비정상: '{tag}'. 예시: ficq-6113.pv, p-6102")
    base = t.split('.')[0]
    tags = _load_tag_set(get_conn)
    if base in tags:
        return None
    toks = [p for p in base.split('-') if len(p) > 2]
    suggested = sorted({x for x in tags if any(p in x for p in toks)})[:3]
    return VerifierError("R1","tag_not_found",
        hint=f"태그 '{tag}' 는 DB에 존재하지 않습니다. find_tags(query=..., sub_area=...) 로 먼저 검색하세요.",
        suggested=suggested)

# ── R2 ──
def validate_area(area: str | None, field: str = "area") -> Optional[VerifierError]:
    if not area:
        return None
    if not AREA_RE.match(area):
        return VerifierError("R2","invalid_area_format",
            hint=f"{field}='{area}' 형식 오류. 'P6' 또는 'P6-1' 형식 사용.",
            valid_areas=sorted(VALID_AREAS))
    base = area.split('-')[0]
    if base not in VALID_AREAS:
        return VerifierError("R2","unknown_area",
            hint=f"{field}='{area}' 미존재. valid: {sorted(VALID_AREAS)} (P7 없음)")
    return None

# ── R4 ──
def validate_direction(d: str | None) -> Optional[VerifierError]:
    if d and d not in VALID_DIRECTIONS:
        return VerifierError("R4","invalid_direction",
            hint=f"direction='{d}' 잘못. 'upstream' 또는 'downstream' 만 허용")
    return None

def validate_max_depth(n: Any) -> Optional[VerifierError]:
    if n is None: return None
    try:
        v = int(n)
    except Exception:
        return VerifierError("R4","invalid_max_depth", hint=f"max_depth='{n}' 은 정수여야 함")
    if not (1 <= v <= 50):
        return VerifierError("R4","max_depth_out_of_range", hint=f"max_depth={v} 범위 외 (1~50)")
    return None

# ── R3, R5 stub (Phase B.2) ──
def validate_response_text(text: str) -> Optional[VerifierError]:
    return None  # Phase B.2 구현 예정

# ── 로그 적재 (Phase C LoRA 입력) ──
def log_rejection(tool: str, params: dict, err: VerifierError) -> None:
    _LOG_DIR.mkdir(parents=True, exist_ok=True)
    today = time.strftime("%Y-%m-%d")
    rec = {"ts": time.time(), "tool": tool, "params": params,
           "verifier_error": err.to_dict()}
    with (_LOG_DIR / f"{today}.jsonl").open("a", encoding="utf-8") as f:
        f.write(json.dumps(rec, ensure_ascii=False) + "\n")

4.2 server.py 적용 — 각 tool 입구에 한 줄

DB 커넥션은 server.py에 이미 있는 헬퍼(예: _get_db_connection)를 get_conn 자리에 전달.

from verifier.validators import (
    validate_tag, validate_area, validate_direction, validate_max_depth,
    log_rejection, VerifierError,
)

def _check(tool: str, params: dict, *errs) -> dict | None:
    """첫 번째 비-None 에러를 로그 + 반환(dict). 없으면 None."""
    for e in errs:
        if e:
            log_rejection(tool, params, e)
            return e.to_dict()
    return None

적용 대상 (server.py의 기존 함수 시그너처는 그대로 두고, 본문 첫 줄에 _check 호출):

Tool	검증
`find_tags`	`validate_area(area)`, `validate_area(sub_area, "sub_area")`
`active_alarms`	`validate_area(area)`
`query_events`	`validate_area(area)` + `tag_name`이 인자면 `validate_tag`
`summarize_events`	`validate_area(area)`
`generate_status_report`	`validate_area(area)`
`query_pv_history`	`tag_names` list 각각 `validate_tag` (첫 실패 반환)
`trace_connections`	`validate_tag(start_tag, conn)`, `validate_direction(direction)`, `validate_max_depth(max_depth)`
`upsert_pid_connection`	`validate_tag(tag_no)`

tool이 dict를 반환하는 경우(이미 그렇게 되어있음) 그대로 return _check(...) or 기존_본문() 패턴. tool이 str을 반환하면 dict를 json.dumps로 감싸서 일관 유지.

4.3 `test_validators.py` (필수, 최소 10케이스)

각 룰별 PASS 2 + FAIL 2 이상. 예:

def test_R1_valid_tag(monkeypatch):
    monkeypatch.setattr(validators, "_load_tag_set", lambda gc: {"ficq-6113","p-6102"})
    assert validators.validate_tag("ficq-6113.pv", lambda: None) is None
def test_R1_unknown_tag(monkeypatch):
    monkeypatch.setattr(validators, "_load_tag_set", lambda gc: {"ficq-6113"})
    err = validators.validate_tag("rm-6101", lambda: None)
    assert err and err.code == "tag_not_found"
def test_R2_invalid_area():
    err = validators.validate_area("6-1")
    assert err and err.code == "invalid_area_format"
# ... R2 PASS, R4 direction/max_depth, etc.

4.4 README.md (간단)

룰 카탈로그 표 + 로그 jsonl 포맷 예시 1줄 + Phase B.2(R3/R5) 미구현 명시.

5. 검증 (순서대로)

cd mcp-server

# 1) 컴파일
python3 -m py_compile verifier/validators.py server.py

# 2) 단위 테스트
.venv/bin/python -m pytest verifier/test_validators.py -v

# 3) MCP 서버 재시작
# (사용자 환경: experioncrawler systemd 또는 uv run server.py)

# 4) opencode 재현 시도 — 다음 3개 호출 시 verifier_error 반환되는지
#    a) trace_connections(start_tag="raw_material_input", direction="downstream") → R1 tag_not_found
#    b) trace_connections(start_tag="RM-6101", ...)                                → R1 tag_not_found + suggested 포함
#    c) generate_status_report(area="6-1", hours=24)                                → R2 invalid_area_format
#    각 응답에 verifier_error.{rule, code, hint} 키 확인

# 5) 로그 적재
ls -la verifier/logs/
cat verifier/logs/$(date +%Y-%m-%d).jsonl | head -3  # 거부 3건 보여야 함

# 6) 정상 호출 회귀 0건 확인
#    a) trace_connections(start_tag="p-6102", direction="upstream") → 정상 결과
#    b) generate_status_report(area="P6-1", hours=24)                → 정상 결과
#    c) Phase 0 eval 재실행 (옵션) — 회귀 0
cd eval && ../.venv/bin/python run_eval.py --model Qwen3-8B --no-think \
  --baseline results/Qwen3-8B_20260526_103459.json

6. 수용 기준

py_compile 통과 (validators.py, server.py)
단위 테스트 ≥ 10케이스 전부 PASS
opencode 재현 §5.4 a/b/c 3건 모두 verifier_error 반환 (hint·suggested 포함)
정상 호출 §5.6 a/b 정상 결과 (회귀 없음)
Phase 0 eval --baseline 비교에서 PASS→FAIL 회귀 0건
verifier/logs/YYYY-MM-DD.jsonl 에 거부 3건 자동 적재 확인

7. 하지 말 것 (금지)

❌ 응답 텍스트(LLM 출력) 후처리 — Phase B.2 범위 (R3/R5는 stub만)
❌ LLM-judge 사용 (MVP는 코드 결정적만)
❌ tool 함수 시그너처·반환 타입 변경 (입구에 한 줄 추가만)
❌ 골든셋·평가 러너·채점기 변경
❌ 모델 서빙 설정·chat 템플릿 변경
❌ 캐시 TTL·매직 넘버 튜닝 (300초·top3·1~50 고정)
❌ 새 DB 객체(테이블/뷰) 생성 (read-only 검증)
❌ worker/ 디렉토리 변경

8. 산출물

신규 6개: verifier/{__init__.py, validators.py, test_validators.py, README.md, logs/.gitkeep}
수정 1개: mcp-server/server.py (import + tool 8개 입구 _check 호출)
한 줄 보고: 적용된 tool 개수 / 재현 §5.4 결과 (3/3) / 회귀 (0) / 로그 샘플 1행

12 KiB Raw Permalink Blame History Unescape Escape