274 lines
12 KiB
Markdown
274 lines
12 KiB
Markdown
# 작업지시서 — 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`
|
||
|
||
```python
|
||
"""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` 자리에 전달.
|
||
|
||
```python
|
||
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 이상. 예:
|
||
```python
|
||
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. 검증 (순서대로)
|
||
|
||
```bash
|
||
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행
|