# -*- coding: utf-8 -*- r""" question_forge.py — 역질문 캡처 엔진 [CEO 혁신①: 자율주행 그림자모드 이식, 2026-07-19] 사상: 암묵지를 기다리지 말고 캐낸다. 지식 전체에서 '구멍'([확인요청]·미확인·기록없음)을 찾아 '가치 높은 순'으로 CEO에게 하루 몇 개씩만 묻는다. CEO 답 → ①구멍 메움 ②증류쌍(파인튜닝 교재). 평범한 문서 100개보다 정확히 겨눈 질문 1개가 값지다. 동작: scan : 지식(md) 전체에서 구멍 채집 → 가치순 랭크 → 질문큐 md 생성 (CEO가 그 파일에 답만 적음) intake: CEO가 답을 적은 질문큐를 읽어 → 답변 지식 md + 증류쌍 jsonl로 흡수, 답한 항목은 큐에서 제거 실행: python question_forge.py scan [개수] (기본 20개) python question_forge.py intake """ import os, io, re, sys, glob, json, time, subprocess CLAUDE_EXE = os.path.expanduser(r'~\.local\bin\claude.exe') def sharp_question(g): """페이블로 구멍을 정확한 한 문장 질문으로. 실패하면 템플릿 폴백.""" try: prompt = ('너는 용접공정 코파일럿 SHIN이다. 아래는 도진웍스 지식에서 발견한 "구멍"(미확정 항목)이다. ' '대표(장인)에게 물어 이 구멍을 메울, 딱 한 문장짜리 구체적 질문을 한국어로 만들어라. ' '맥락에 있는 제품·부위·수치 종류를 반영해서 대표가 바로 답할 수 있게. 질문 문장만 출력.\n\n' '[제품/맥락] %s\n[구멍이 있는 문장] %s\n[주변 문맥] %s' % (g['product'], g['gap_line'], g['ctx'])) r = subprocess.run([CLAUDE_EXE, '-p', '--model', 'claude-fable-5'], input=prompt.encode('utf-8'), capture_output=True, timeout=90) out = r.stdout.decode('utf-8', errors='ignore').strip().split('\n')[0].strip() if out and len(out) > 6: return out[:200] except Exception: pass return phrase(g) ROOTS = [r'E:\VLA시스템', r'E:\지식정제시스템\05_KNOWLEDGE\seeds', r'E:\옵시디언'] SKIP = ('EMBEDDINGS', 'MODELS', 'OUTPUT', 'FINETUNE', '.obsidian', '.trash', '_QA뱅크', '_forge') OUT_DIR = r'E:\VLA시스템\KNOWLEDGE\_역질문' QUEUE = os.path.join(OUT_DIR, '질문큐.md') # CEO가 여기에 답을 적는다 ANSWERS_MD = os.path.join(OUT_DIR, '_CEO답변_흡수본.md') # 흡수된 답 누적 DISTILL = r'E:\VLA시스템\FINETUNE\distill_pairs\ceo_answers.jsonl' STATE = os.path.join(OUT_DIR, '_asked_state.json') # 이미 물어본 구멍(중복질문 방지) GAP_PAT = re.compile(r'\[?\s*(확인\s*요청|확인\s*필요|CEO\s*확인|미확인|미언급|기록에?\s*(없|남아있지 않)|지어내지)\s*\]?') # 안전·핵심 키워드 = 가치 가중 (현장 사고·품질 직결) HOT = re.compile(r'용락|크랙|균열|전류|전압|갭|각장|용입|변형|예열|토치각|불량|안전|방향|치수|리비전') def product_of(path): parts = path.replace('/', '\\').split('\\') for anchor in ('seeds', 'KNOWLEDGE', 'VLA시스템'): if anchor in parts: i = parts.index(anchor) return ' / '.join(p for p in parts[i + 1:i + 4] if not p.endswith('.md')) return os.path.basename(os.path.dirname(path)) def load_json(p, d): try: return json.load(io.open(p, encoding='utf-8')) except Exception: return d def scan(n): asked = set(load_json(STATE, {'asked': []}).get('asked', [])) gaps = [] for root in ROOTS: for dp, _, fns in os.walk(root): if any(s in dp for s in SKIP): continue for fn in fns: if not fn.endswith('.md'): continue p = os.path.join(dp, fn) try: lines = io.open(p, encoding='utf-8', errors='ignore').read().split('\n') except Exception: continue for i, ln in enumerate(lines): if not GAP_PAT.search(ln): continue ctx = ' '.join(x.strip() for x in lines[max(0, i - 2):i + 1] if x.strip())[:300] key = (os.path.basename(p) + '|' + re.sub(r'\s+', '', ln)[:60]) if key in asked: continue score = 1 + len(HOT.findall(ctx)) * 2 if 'GOLD' in fn: score += 4 if '공정서' in fn or '불량' in fn: score += 3 if '브리핑' in fn: score += 2 gaps.append({'key': key, 'score': score, 'product': product_of(p), 'src': p, 'gap_line': ln.strip()[:200], 'ctx': ctx}) # 중복 문맥 제거 + 가치순 seen, uniq = set(), [] for g in sorted(gaps, key=lambda x: -x['score']): h = re.sub(r'\s+', '', g['ctx'])[:80] if h in seen: continue seen.add(h); uniq.append(g) top = uniq[:n] os.makedirs(OUT_DIR, exist_ok=True) out = ['# SHIN 역질문 큐 — 대표님께 묻습니다', '> SHIN이 지식에서 발견한 "구멍"을 가치 높은 순으로 %d개 뽑았습니다.' % len(top), '> 각 질문 아래 **답:** 줄에 한 줄만 적어주세요. 모르면 비워두면 다음에 다시 안 묻습니다(스킵하려면 "스킵").', '> 다 적으신 뒤 저에게 "역질문 흡수해"라고 하시면 지식에 넣고 이 큐에서 지웁니다.', '', '_생성: %s | 미해결 구멍 총 %d개 중 상위 %d개_' % (time.strftime('%Y-%m-%d'), len(uniq), len(top)), '', '---', ''] use_sharp = os.path.exists(CLAUDE_EXE) and '--fast' not in sys.argv _q_cache = {} for idx, g in enumerate(top, 1): q = sharp_question(g) if use_sharp else phrase(g) _q_cache[g['key']] = q out += ['## Q%d. %s' % (idx, q), '- 제품/맥락: %s' % (g['product'] or '—'), '- 근거 문장: "%s"' % g['gap_line'], '- 출처: `%s`' % g['src'], '', '**답:** ', '', '---', ''] io.open(QUEUE, 'w', encoding='utf-8').write('\n'.join(out)) # [혁신①→대화형] 기계가 읽는 JSON 큐도 생성 — 서버 인터뷰 엔드포인트가 사용 qjson = [] for idx, g in enumerate(top, 1): qjson.append({'id': 'Q%d_%d' % (int(time.time()) % 100000, idx), 'q': _q_cache.get(g['key']) or phrase(g), 'product': g['product'], 'src': g['src'], 'gap_line': g['gap_line'], 'answered': False}) io.open(os.path.join(OUT_DIR, '질문큐.json'), 'w', encoding='utf-8').write( json.dumps({'ts': time.strftime('%Y-%m-%d %H:%M'), 'items': qjson}, ensure_ascii=False, indent=1)) # asked 갱신(이번에 낸 질문은 다시 안 냄) st = {'asked': list(asked | {g['key'] for g in top}), 'ts': time.strftime('%Y-%m-%d %H:%M')} io.open(STATE, 'w', encoding='utf-8').write(json.dumps(st, ensure_ascii=False)) print('[역질문] 큐 생성: %s (%d개, 총 구멍 %d개)' % (QUEUE, len(top), len(uniq))) def phrase(g): """구멍 문맥을 사람이 답하기 쉬운 질문으로. 템플릿 기반(결정적).""" ctx = g['ctx'] prod = (g['product'].split('/')[-1] if g['product'] else '이 부품').strip() if re.search(r'전류|전압|조건', ctx): return '%s 용접조건(전류·전압)은 실제로 얼마였나요?' % prod if re.search(r'갭', ctx): return '%s의 갭은 몇 mm로 잡나요?' % prod if re.search(r'각장', ctx): return '%s의 각장 기준은 얼마인가요?' % prod if re.search(r'방향|좌우|리비전', ctx): return '%s의 방향/리비전 확정 값은 무엇인가요?' % prod if re.search(r'치수|높이|기준값', ctx): return '%s의 기준 치수는 얼마인가요?' % prod if re.search(r'예열|열처리', ctx): return '%s는 예열이 필요한가요? 필요하면 몇 도인가요?' % prod # 일반형 frag = re.sub(r'\[?\s*(확인요청|확인 필요|CEO확인|미확인)\s*\]?', '', g['gap_line']).strip(' .-:')[:80] return '%s 관련 — "%s", 이 부분 실제 값/사실은 무엇인가요?' % (prod, frag or '이 항목') def intake(): if not os.path.exists(QUEUE): print('[역질문] 질문큐 없음 — 먼저 scan'); return txt = io.open(QUEUE, encoding='utf-8').read() blocks = re.split(r'\n## Q\d+\.', txt) picked = 0 os.makedirs(os.path.dirname(DISTILL), exist_ok=True) ans_lines = ['', '## 흡수 %s' % time.strftime('%Y-%m-%d %H:%M'), ''] for b in blocks[1:]: qline = b.split('\n', 1)[0].strip() m = re.search(r'\*\*답:\*\*\s*(.+)', b) src = re.search(r'출처: `([^`]+)`', b) prod = re.search(r'제품/맥락: (.+)', b) if not m: continue ans = m.group(1).strip() if not ans or ans in ('스킵', 'skip', '-'): continue picked += 1 ans_lines += ['**Q: %s**' % qline, '- A(CEO): %s' % ans, '- 출처: %s' % (src.group(1) if src else '—'), ''] with io.open(DISTILL, 'a', encoding='utf-8') as f: f.write(json.dumps({'ts': time.strftime('%Y-%m-%d %H:%M:%S'), 'via': 'question_forge', 'product': (prod.group(1).strip() if prod else ''), 'q': qline, 'a': ans, 'src': (src.group(1) if src else '')}, ensure_ascii=False) + '\n') if picked: with io.open(ANSWERS_MD, 'a', encoding='utf-8') as f: f.write('\n'.join(ans_lines)) # 답한 항목 비운 새 큐로 교체(빈 답만 남김) io.open(QUEUE, 'w', encoding='utf-8').write( '# 역질문 큐 — %d개 흡수 완료(%s). 남은 미답 항목은 다음 scan에서 새로 채웁니다.\n' % (picked, time.strftime('%Y-%m-%d %H:%M'))) print('[역질문] 흡수 완료: %d개 → %s + 증류쌍 %s' % (picked, ANSWERS_MD, DISTILL)) else: print('[역질문] 답이 적힌 항목이 없습니다.') if __name__ == '__main__': cmd = sys.argv[1] if len(sys.argv) > 1 else 'scan' if cmd == 'intake': intake() else: scan(int(sys.argv[2]) if len(sys.argv) > 2 else 20)