22 KiB
작업지시서 — Phase C LoRA 가속 (C1 데이터 큐레이션 우선)
대상: 코딩 LLM. C1을 우선 완료, C2·C3는 진입 조건 만족 시 별도 트리거. 상위 플랜:
plans/OPUS-v2-실행계획.mdPhase C. Phase B(Verifier) 완료 후의 후속 — Verifier 로그를 학습 데이터의 1차 소스로 활용.
0. 배경 (왜 LoRA, 왜 지금)
Phase B Verifier가 invention을 런타임에서 결정적으로 차단하고 있지만, 이건 증상 차단입니다. 근본 원인 = 모델이 애초에 잘못된 호출을 생성하는 행동 패턴. 이걸 모델 weights에 박는 게 LoRA의 역할:
- "사용자가 명시 안 한 태그는 항상
find_tags먼저 호출" - "area는
P6-1형식만 사용" - "툴이 빈 결과 반환 시 자기 인자를 의심"
핵심 인사이트: Verifier 거부 로그(mcp-server/verifier/logs/*.jsonl)가 이미 production에서 "잘못된 호출 → hint → 자기교정" 행동 데이터를 자동 수집 중. 이걸 가공만 하면 즉시 학습 데이터.
C1 — 데이터 큐레이션 (1주, 본 지시서의 주력)
C1.1 입력 소스
| 소스 | 경로 | 양 | 비고 |
|---|---|---|---|
| Verifier 거부 로그 | mcp-server/verifier/logs/*.jsonl |
현재 3건, 운영 기간 늘수록 자동 증가 | 1차 소스. "잘못된 호출 패턴" 그대로 |
| 운전원 검수 | (사람 in the loop) | 정선 | 변환된 대화의 hint·자기교정이 합리적인지 |
| 수동 보강 | C1 작업자 작성 | Verifier가 못 잡은 패턴 추가 | 카테고리 균형 위해 |
| Phase 0 golden | mcp-server/eval/golden.jsonl |
40 | 학습에 사용 금지(평가셋 누수 차단). 다만 카테고리 분포 참고 |
C1.2 출력 — mcp-server/training/ 신규
| 파일 | 포맷 | 용도 |
|---|---|---|
training/sft_data.jsonl |
OpenAI messages | C2 SFT-LoRA 입력 |
training/sft_data_stats.md |
통계 | 카테고리 분포·토큰 분포·중복 여부 |
training/curate_from_verifier.py |
스크립트 | Verifier 로그 → ShareChat 변환 |
C1.3 변환 규칙 — Verifier 거부 1건 → ShareChat 대화 1건
각 거부 로그를 5턴 대화로:
{"messages":[
{"role":"system","content":"<plant_context 요약 + grounding 규칙>"},
{"role":"user","content":"<자연어 질문 — 거부 시 user input을 운전원이 작성>"},
{"role":"assistant","content":"<잘못된 도구 호출 — params 그대로>"},
{"role":"tool","content":"<verifier_error + hint>"},
{"role":"assistant","content":"<올바른 자기교정 호출 — 운전원이 작성>"}
]}
예 (현재 로그의 RM-6101 거부):
- user: "6-1차 플랜트 원료 투입 경로 알려줘"
- assistant:
[trace_connections] start_tag="RM-6101" direction="downstream"(잘못) - tool:
{"verifier_error":"R1.tag_not_found","hint":"...find_tags 먼저...","suggested":["f-6101a","f-6101b","fcv-6101"]} - assistant:
[find_tags] query="원료 투입" sub_area="P6-1"(자기교정 = 올바른 행동)
학습 후 모델이 그 자기교정 행동을 처음부터 수행하게 됨.
C1.3.1 에러 유형별 conversion mapping
| Verifier Error | 자동 생성 user_input |
자동 생성 corrected_call |
|---|---|---|
R1.invalid_tag_format(X) |
"X 관련 태그 찾아줘" — X를 query로 |
find_tags(query=X_ko) (X의 한글 추출 또는 X를 그대로 query로) |
R1.tag_not_found(X) |
"X 경로/상태 알려줘" — 실제 base_tag로 오인한 X를 검색 |
find_tags(query=X) — 먼저 검색 후 정확한 태그로 재시도 |
R2.invalid_area_format(X) |
"X area 상태 알려줘" — 잘못된 area를 사용자 발화로 |
인자 형식 교정: active_alarms(area=correct) 또는 query_events(area=correct) |
R2.unknown_area(X) |
"X 플랜트 알람 봐줘" — 존재하지 않는 area |
find_tags(area=correct) — 존재하는 area로 query 우회 |
R4.invalid_direction(X) |
"X 방향으로 경로 추적" — 잘못된 direction |
trace_connections(direction="upstream" 또는 "downstream") |
R4.max_depth_out_of_range(X) |
"깊게 추적해줘" — 범위 초과 depth |
trace_connections(max_depth=20) — 기본값으로 교정 |
각 Verifier 로그의 verifier_error.code와 일치하는 행의 template을 적용.
C1.3.2 curate_from_verifier.py 의사코드
"""
curate_from_verifier.py — Verifier 거부 로그 → ShareChat 5-turn 대화 변환.
입력: mcp-server/verifier/logs/*.jsonl
출력: mcp-server/training/sft_data.jsonl (PLACEHOLDER 포함)
mcp-server/training/sft_data_stats.md (통계)
변환 전략:
각 거부 로그에 대해 에러 코드 기반 매핑 테이블(C1.3.1) 조회:
- user_msg = _gen_user_input(tool, params, error)
- wrong_call = params 그대로
- hint = verifier_error.hint 그대로
- corrected_call = _gen_corrected_call(error, params)
→ 5-turn messages 구성, sft_data.jsonl에 append.
"""
import json, hashlib, pathlib, sys
from difflib import SequenceMatcher
SYSTEM_PROMPT = (
"당신은 P6(6차) PGMEA 플랜트 운전 지원 assistant입니다. "
"사용자가 명시하지 않은 태그는 find_tags로 먼저 검색합니다. "
"area는 'P6-1' 형식만 사용합니다. "
"DB에 존재하지 않는 태그·area는 추측하지 않고 검색하거나 거절합니다."
)
CONVERSION_MAP = {
"R1.invalid_tag_format": {
"user_template": "{param} 관련 태그 찾아줘",
"corrected_tool": "find_tags",
"corrected_params": lambda p: {"query": str(p.get("start_tag", p.get("query", "")))},
},
"R1.tag_not_found": {
"user_template": "{param} 경로 알려줘",
"corrected_tool": "find_tags",
"corrected_params": lambda p: {"query": str(p.get("start_tag", p.get("query", "")))},
},
"R2.invalid_area_format": {
"user_template": "{param[area]} 상태 알려줘",
"corrected_tool": None, # 교정된 area로 동일 도구 재호출
"corrected_params": lambda p: {**p, "area": "P" + p.get("area", "").lstrip("P")},
},
"R2.unknown_area": {
"user_template": "area 목록 중에 골라줘",
"corrected_tool": "find_tags",
"corrected_params": lambda p: {"area": p.get("area", "").split("-")[0]},
},
"R4.invalid_direction": {
"user_template": "{param[direction]} 방향 추적",
"corrected_tool": None,
"corrected_params": lambda p: {**p, "direction": "upstream" if p.get("direction") != "upstream" else "downstream"},
},
"R4.max_depth_out_of_range": {
"user_template": "적당한 깊이로 추적해줘",
"corrected_tool": None,
"corrected_params": lambda p: {**p, "max_depth": 20},
},
}
def _tag_as_param(params: dict) -> str:
return str(params.get("start_tag", params.get("query", params.get("tag_name", ""))))
def _gen_user_input(tool: str, params: dict, err: dict) -> str:
code = err.get("verifier_error", "")
entry = CONVERSION_MAP.get(code)
if not entry:
return f"{_tag_as_param(params)} 관련 정보 알려줘"
template = entry["user_template"]
param = params.get("start_tag", params.get("query", params.get("area", params.get("tag_name", ""))))
return template.format(param=param, param_parts=params)
def _gen_corrected_call(tool: str, params: dict, err: dict) -> dict:
code = err.get("verifier_error", "")
entry = CONVERSION_MAP.get(code)
if not entry:
return {"tool": "find_tags", "params": {"query": _tag_as_param(params)}}
new_params = entry["corrected_params"](params)
new_tool = entry["corrected_tool"] or tool
return {"tool": new_tool, "params": new_params}
변환 후 일부 필드는 __PLACEHOLDER__로 남겨 운전원 검수(C1.4 step 2)에서 수동 작성.
C1.4 작업 단계
-
curate_from_verifier.py스크립트 작성:verifier/logs/*.jsonl읽어 거부 trip 추출- 각 trip을 위 5턴 템플릿으로 변환
- user input·자기교정 assistant turn은 placeholder(
<<USER_INPUT>>,<<CORRECTED_CALL>>)로 두고 검수 출력
-
운전원 검수 — 구체 workflow:
a. 검수 파일 생성:
curate_from_verifier.py실행 →training/sft_data.jsonl생성 (user_input 및 corrected_call 중 일부는__PLACEHOLDER__상태)b. 운전원 편집 (30~60분, vim/vscode로 직접 JSONL 편집):
- 각 line의
messages[1].content(__PLACEHOLDER__)를 현실적인 질문으로 대체- 예:
__PLACEHOLDER__→ "6-1차 원료 탱크에서 어디로 가는지 경로 추적해줘"
- 예:
- 각 line의
messages[4].content(__PLACEHOLDER__)를 정확한 도구 호출로 대체- 예:
__PLACEHOLDER__→[find_tags] query="원료 투입" sub_area="P6-1"
- 예:
- hint(turn 3)는 자동 생성된 Verifier hint 그대로 사용 (수정 금지)
c. 검수 기준:
- user input이 실제 운전원이 할 법한 질문인가? (현장 용어 사용)
- 자기교정 응답이 정확한 도구·인자인가? (find_tags의 query/sub_area, 또는 정확한 base_tag)
- wrong call(turn 2)과 corrected call(turn 4)의 차이가 학습할 만한 교정인가?
d. 검증 스크립트 실행 (수정 후):
cd mcp-server .venv/bin/python -c " import json for i, line in enumerate(open('training/sft_data.jsonl'), 1): d = json.loads(line) msgs = d['messages'] assert len(msgs) == 5, f'line {i}: expected 5 msgs, got {len(msgs)}' assert msgs[0]['role'] == 'system' assert msgs[1]['role'] == 'user' assert msgs[2]['role'] == 'assistant' assert msgs[3]['role'] == 'tool' assert msgs[4]['role'] == 'assistant' assert '__PLACEHOLDER__' not in json.dumps(d), f'line {i}: has PLACEHOLDER' print('All OK') "e. 완료 서명: operator가 검수 완료 시
sft_data.jsonl을 그대로 C2 단계로 전달. - 각 line의
-
수동 보강 — abstain 25건 + scaffold 25건 (golden.jsonl의 category 분포 균형):
- Phase 0 골든의 5 카테고리 분포: tool_call(10), nl2sql(10), ground(6), abstain(8), scaffold(6)
- Verifier 거부는 주로 tool_call·grounding 카테고리 → abstain(scaffold 추가 작성 총 50건
- 절대 golden question을 복사하지 말 것 — category 분포만 참고, question은 직접 창작
abstain template 예시 (총 25건 중 첫 5건 예시):
{"id":"abstain-m01","category":"abstain","question":"P3 플랜트 생산량 예측해줘", "context":"P3 area는 펌프 상태만 수집, 생산량 데이터 없음", "expect":{"refuse":true,"must_not":["톤","kg","예상 생산량은"]}} {"id":"abstain-m02","category":"abstain","question":"ficq-6113 고장 원인 진단해줘", "context":"PV=0, SP=50 값만 있음. 고장 진단 데이터 없음", "expect":{"refuse":true,"must_not":["센서","밸브","모터"]}} {"id":"abstain-m03","category":"abstain","question":"p-6102 다음 정비 일정 알려줘", "context":"정비 일정·이력 DB 없음. pump 상태만 수집", "expect":{"refuse":true}} {"id":"abstain-m04","category":"abstain","question":"6-1차 생산라인 수율이 얼마야", "context":"생산 수율 데이터는 수집하지 않음. 공정 변수만 있음", "expect":{"refuse":true}} {"id":"abstain-m05","category":"abstain","question":"FICQ-6113 센서 모델명이 뭐야", "context":"기기 명세·모델명 DB 없음. tag_metadata에 desc만 저장", "expect":{"refuse":true}}scaffold template 예시 (총 25건 중 첫 5건 예시):
{"id":"scaffold-m01","category":"scaffold","question":"리플럭스 밸브 열어도 될까?", "context":"ficq-6113.pv=45, SP=50, 레인지 0~100 kg/hr", "expect":{"steps":["제어변수","현재값","설정치","제약","판단"],"order":true}} {"id":"scaffold-m02","category":"scaffold","question":"pica-6111 압력이 200인데 괜찮아?", "context":"pica-6111.pv=200, SP=760, 레인지 0~760 mmHg. 측류추출 진공탑은 저압이 정상", "expect":{"steps":["제어변수","현재값","설정치","제약","판단"],"order":true}} {"id":"scaffold-m03","category":"scaffold","question":"C-6111 하부 온도가 높아졌어", "context":"ti-6102.pv=175, SP=160, 레인지 0~300°C. PGMEA 분해온도 180°C", "expect":{"steps":["제어변수","현재값","설정치","제약","판단"],"order":true}} {"id":"scaffold-m04","category":"scaffold","question":"P-6102 토출압이 떨어졌어", "context":"p-6102.pv={R-RUN|5|}, pt-6101.pv=3.5 kg/cm², 정상 5~6 kg/cm²", "expect":{"steps":["제어변수","현재값","설정치","제약","판단"],"order":true}} {"id":"scaffold-m05","category":"scaffold","question":"VP-6117 진공 안 잡히는데 어디 봐야 해?", "context":"vp-6117.pv={R-RUN|5|}, pica-6111.pv=680 mmHg (대기압 760 = 진공 약함), 정상 <100 mmHg", "expect":{"steps":["제어변수","현재값","설정치","제약","판단"],"order":true}}scaffold 패턴 골자 — 5단계 추론 유도:
- 제어변수 식별 (어떤 태그·PV인가)
- 현재값 확인 (PV 수치)
- 설정치/레인지 확인 (SP / 레인지 상하한)
- 제약 조건 확인 (분해온도·정상범위·운전한계)
- 판단 (정상·주의·위험 + 조치 방향)
나머지 40건은 위 template을 변형(area/태그번호/수치 변경)하여 작성.
tool_call 보강 (10건, Verifier 에서 부족한 패턴):
- "원료가 어떤 경로로 들어오는지", "P6-2 밸브 상태 확인" →
find_tags먼저 호출 유도 - "온도 트랜드 보고서", "shift 요약" →
generate_status_report또는summarize_events
grounding 보강 (10건, 공정 지식 부족 패턴):
- "측류추출이 뭐야", "진공 증류 왜 해" → 공정 설명 요구
- "P6-1과 P6-2 차이" → 영역 구분 지식
-
누수 차단 — 3단계 중복 제거:
a. 정확 해시 매칭: golden.jsonl의 각
question필드 md5 해시 세트를 구축. sft_data candidate의messages[1].contentmd5가 golden 해시 세트에 존재 → 제거.b. Fuzzy 매칭:
difflib.SequenceMatcher(None, user_content, golden_q).ratio() > 0.9→ 의미는 같고 표현만 다른 question 차단 (예: "알람 봐줘" vs "알람 보여줘")c. Embedding guard (선택):
nomic-embed-text로 golden + candidate 임베딩, cosine > 0.95 제거.mcp-server/의 Ollama 클라이언트 재사용 가능.구현 코드 (curate_from_verifier.py 내):
GOLDEN_PATH = pathlib.Path(__file__).parent.parent / "eval" / "golden.jsonl" def _golden_question_hashes() -> set[str]: hashes = set() for line in open(GOLDEN_PATH): d = json.loads(line) hashes.add(hashlib.md5(d["question"].encode()).hexdigest()) return hashes def _is_leak(user_content: str, golden_hashes: set[str], golden_items: list[dict]) -> bool: h = hashlib.md5(user_content.encode()).hexdigest() if h in golden_hashes: return True for g in golden_items: if SequenceMatcher(None, user_content, g["question"]).ratio() > 0.9: return True return False주의: 수동 보강 작성 시에도 golden question을 참고만 하고 복사 금지. golden category 분포는 참고 가능하나 question 텍스트 직접 재사용 금지.
-
통계 출력 (
sft_data_stats.md):- 총 건수, 카테고리별 분포, 평균 토큰 수, golden 중복 0 확인
C1.5 수용 기준 (C1)
curate_from_verifier.pypy_compile 통과sft_data.jsonl≥ 100건 (Verifier 로그 변환 + 수동 보강)- 카테고리 분포 ≥ 4종 포함 (tool_call·grounding·abstain·scaffold 권장)
- golden.jsonl 누수 0건 (해시 비교 + fuzzy guard)
- 각 라인 JSON 스키마 검증 통과:
json.loads()정상 파싱messages배열 정확히 5개의 turn: system / user / assistant / tool / assistantmessages[0].role == "system",messages[1].role == "user",messages[2].role == "assistant",messages[3].role == "tool",messages[4].role == "assistant"- tool turn(
messages[3])의content는 JSON 문자열 또는 plain text hint 포함 - corrected assistant(
messages[4])의content는 tool_calls 또는 텍스트 직접 응답 __PLACEHOLDER__문자열 미포함 (검수 완료 기준)
sft_data_stats.md자동 생성 — 포함 항목:- 총 건수, Verifier 변환 건수 vs 수동 보강 건수
- 카테고리별 분포 (건수 + 비율)
- 평균/최소/최대 messages 토큰 수 (tokenizer-free: 공백 기준 split)
- golden 누수 0 확인
__PLACEHOLDER__잔여 0 확인
C2 — SFT-LoRA 학습 (1주, C1 완료 후 트리거)
이 섹션은 C1 산출물(
sft_data.jsonl≥ 100건) 확보 시점에 본격화. 지금은 outline·결정 사항만.
C2.1 베이스 모델 선정 (선결 필요)
| 후보 | 장점 | 단점 |
|---|---|---|
| Qwen3-8B bf16 | 이미 production 호환 검증(Phase 0), dense → LoRA 깨끗 | 8B 한계 (L2~L4 부족) — but LoRA로 학습 후 차이 확인 |
| Qwen2.5-7B-Instruct bf16 | obedient, 사고모드 없음 | 한국어 약간↓ |
| Qwen3-14B bf16 (있다면) | 중간 capacity | 미확인 |
→ 1차: Qwen3-8B bf16. 실패 시 7B-Instruct로 fallback. ⚠️ bf16 베이스 필수 (FP8/FP8-dynamic은 LoRA 학습 부적합 — OPUS-v1 §3.1 함정).
C2.2 환경
- Unsloth 컨테이너 (별도 setup. OPUS-v1 §2 참고하되 35B는 보존 — 학습 컨테이너는 별도 GPU 시간 점유)
- 또는 호스트에서 직접
unslothpip 설치 (간단)
C2.3 학습 하이퍼파라미터 (1차 안)
from unsloth import FastLanguageModel
model, tokenizer = FastLanguageModel.from_pretrained(
model_name="Qwen/Qwen3-8B", # bf16
max_seq_length=4096,
dtype=torch.bfloat16,
load_in_4bit=True, # QLoRA
)
model = FastLanguageModel.get_peft_model(
model,
r=64, lora_alpha=128,
target_modules=["q_proj","k_proj","v_proj","o_proj"], # attention-only 1차
lora_dropout=0, bias="none",
use_gradient_checkpointing="unsloth",
use_rslora=True,
)
# TrainingArguments: per_device_train_batch_size=2, grad_accum=4,
# max_steps=200, learning_rate=2e-4, warmup=5, save_steps=50
C2.4 산출
lora-adapters/calibration-v1/— 어댑터 가중치lora-adapters/calibration-v1/training_log.json— loss 곡선lora-adapters/calibration-v1/eval_report.md— Phase 0 골든 + Verifier reject rate (전/후)
C2.5 수용 기준 (C2)
- 학습 손실 안정 수렴 (eval loss 단조 감소 or 안정)
- Phase 0 골든 평가에서 회귀 0건
- 사후 production 시뮬레이션(opencode에서 invention 재현 질문 5건): Verifier reject 비율 ≥ 50% 감소 (모델이 자기교정 행동을 학습)
- 어댑터 hot-swap or merge 검증 (v1 §4.1 절차)
회귀 트리거 — Phase 0 eval에서 회귀 ≥ 2건 또는 invention rate 개선 < 30% 이면:
→ r ∈ {32, 128} sweep + target_modules에 gate_proj/up_proj/down_proj 추가(전결합 LoRA) 1회 시도
→ r=64 / attention-only가 목표 데이터에 적합하지 않을 가능성에 대한 안전판
C3 — DPO 정렬 (1~2주, 진입 조건부)
진입 조건: C2 어댑터로 production 운영 ≥ 2주 + 운전원 👍/👎 데이터 ≥ 200쌍.
C3.1 사전 작업 (별도 트랙)
- opencode 또는 Web UI에 👍/👎 버튼 추가 (응답마다 binary 피드백) — 별도 작업지시서 필요
- 피드백을
feedback/preferences.jsonl에 적재 (응답 텍스트 + tool calls + binary label)
C3.2 DPO 데이터 구성
- chosen = 👍 응답
- rejected = 👎 응답 (또는 Verifier reject된 응답)
- 자동 보상 보조: Verifier 통과/실패를 binary 보상으로 가중
C3.3 학습
trl.DPOTrainer(베이스: C2 SFT 어댑터를 ref_model로)- beta=0.1, learning_rate=5e-6, 1 epoch
C3.4 산출 / 수용 기준
lora-adapters/calibration-v2-dpo/- fabrication_rate (Phase 0 eval) C2 대비 추가 개선
- 운전원 만족도 (👍 비율) 추적 — C2 어댑터 대비 향상
산출물 종합
| Phase | 신규 파일 | 변경 |
|---|---|---|
| C1 | mcp-server/training/curate_from_verifier.pymcp-server/training/sft_data.jsonlmcp-server/training/sft_data_stats.md |
— |
| C2 | lora-adapters/calibration-v1/* |
— |
| C3 | lora-adapters/calibration-v2-dpo/*feedback/preferences.jsonl (수집 시작) |
UI에 👍/👎 추가 |
하지 말 것 (금지)
- ❌ Phase 0 골든셋 학습에 사용 (평가셋 누수 → 측정 무력화)
- ❌ FP8/FP8-dynamic 베이스로 LoRA 학습 시도 (수치 손상 — 우리가 본 RedHatAI 7B 사례)
- ❌ MoE expert 모듈에 LoRA 적용 (v1 §3.2 — vLLM hot-swap 깨짐). attention-only.
- ❌ 자동/야간 자기학습 루프 (검수 없는 학습 = model collapse — v1 §5.2.B 폐기)
- ❌ Verifier 룰 우회를 위해 데이터에 가짜 거부/hint 작성 (학습 신호 오염)
- ❌
mcp-server/eval/·verifier/validators.py수정 (Phase C 범위 밖)
즉시 착수 (C1)
mkdir -p mcp-server/training
# 작업자가 만들 것: curate_from_verifier.py
# 입력: mcp-server/verifier/logs/*.jsonl
# 출력: mcp-server/training/sft_data.jsonl + sft_data_stats.md
# 검증:
cd mcp-server
python3 -m py_compile training/curate_from_verifier.py
.venv/bin/python -c "
import json
items = [json.loads(l) for l in open('training/sft_data.jsonl')]
print(f'total: {len(items)}')
print(f'avg msgs: {sum(len(i[\"messages\"]) for i in items)/len(items):.1f}')
"
# 골든 누수 검사:
.venv/bin/python -c "
import json, hashlib
g = {hashlib.md5(json.loads(l)['question'].encode()).hexdigest() for l in open('eval/golden.jsonl')}
s = [json.loads(l) for l in open('training/sft_data.jsonl')]
leaks = [i for i in s if any(hashlib.md5(m['content'].encode()).hexdigest() in g for m in i['messages'] if m['role']=='user')]
print(f'leaks: {len(leaks)}')
"
완료 보고: total / 카테고리 분포 / golden 누수 (0 기대) / 거부 변환 vs 수동 보강 비율. 보고 받은 뒤 C2 트리거.