# -*- coding: utf-8 -*-
# dzw_knowledge_server.py — 도진웍스 공정지식 서버 [핵심기술 2026-07-06]
# ARIA 코파일럿의 뇌: E:\VLA시스템 전체 공정서를 검색·제공 (127.0.0.1 전용, 외부 비노출)
# /api/search?q=12500 → 공정서 후보 목록 (점수순)
# /api/read?p=<경로> → 공정서 원문 (VLA 폴더 내 .md만 허용)
# /api/ask POST {q, ctx} → Ollama(llama3.1:8b) 로컬 LLM 답변 (없으면 501)
import http.server, socketserver, os, re, json, urllib.parse, urllib.request, subprocess, tempfile, threading, time
DISTILL_DIR = r'E:\VLA시스템\FINETUNE\distill_pairs' # [CEO 2026-07-18] 페이블 증류 데이터 (질문+근거+답 jsonl)
_CLAUDE_LOCK = threading.Lock() # claude 동시 실행 방지 (겹치면 빈 응답·충돌)
TTS_VOICE = 'ko-KR-SunHiNeural' # 뉴럴 한국어 (SunHi 여성 / InJoonNeural 남성 / HyunsuMultilingualNeural)
# [CEO 2026-07-22 시연대비 결함수정] 음성 임시파일을 요청마다 고유 이름으로 만든다.
# 원인: 여태 'dzw_tts_out.mp3' 한 개 이름을 모두가 같이 썼다. 대표가 연속으로 질문해서 음성 요청이
# 겹치면, 뒤 요청이 앞 요청 파일을 덮어써서 소리가 깨지거나 엉뚱한 문장이 재생될 수 있었다.
# 조치: uuid로 요청마다 다른 파일을 쓰고, 다 읽으면 즉시 지운다(임시폴더에 쌓이지 않게).
def _tts_tmp(ext):
import uuid
return os.path.join(tempfile.gettempdir(), 'dzw_tts_%s.%s' % (uuid.uuid4().hex, ext))
def _rm(*paths):
for p in paths:
try: os.remove(p)
except Exception: pass
def tts_edge(text, voice=None):
"""edge-tts(GitHub rany2/edge-tts) 뉴럴 음성 → MP3. 사람에 가까운 품질(무료, 인터넷 필요)."""
import asyncio, edge_tts
out = _tts_tmp('mp3')
v = voice or TTS_VOICE
async def go():
await edge_tts.Communicate((text or '')[:600], v).save(out)
try:
asyncio.run(go())
with open(out, 'rb') as f: return f.read(), 'audio/mpeg'
finally:
_rm(out)
def tts_wav(text, voice=None):
"""Windows 내장 한국어 음성(Heami)으로 WAV 생성 — 오프라인 폴백(삼성인터넷 포함 재생)."""
txt = _tts_tmp('txt')
wav = _tts_tmp('wav')
with open(txt, 'w', encoding='utf-8') as f: f.write((text or '')[:600])
ps = ("Add-Type -AssemblyName System.Speech;"
"$s=New-Object System.Speech.Synthesis.SpeechSynthesizer;"
"try{$s.SelectVoice('Microsoft Heami Desktop')}catch{};"
"$s.SetOutputToWaveFile('" + wav.replace('\\', '\\\\') + "');"
"$t=[IO.File]::ReadAllText('" + txt.replace('\\', '\\\\') + "',[Text.Encoding]::UTF8);"
"$s.Speak($t);$s.Dispose()")
try:
subprocess.run(['powershell', '-NoProfile', '-NonInteractive', '-Command', ps],
capture_output=True, timeout=40)
with open(wav, 'rb') as f: return f.read(), 'audio/wav'
finally:
_rm(txt, wav)
def tts(text, voice=None):
"""뉴럴(edge-tts) 우선, 실패 시 Heami WAV 폴백. voice는 요청별로 받아 동시요청이 서로 안 섞이게 한다."""
try:
return tts_edge(text, voice)
except Exception:
return tts_wav(text, voice)
CLAUDE_EXE = os.path.expanduser(r'~\.local\bin\claude.exe') # 클로드 코드 CLI (소넷 — 비용규칙)
PROJECT_DIR = os.path.dirname(os.path.abspath(__file__)) # 폰 개발모드가 작업할 폴더 (뷰어 프로젝트)
DEV_VOICE_FRAME = (
'[대화 규칙] 너는 도진웍스 대표(비개발자·용접전문가)와 폰에서 음성으로 대화하는 코파일럿 SHIN이다. '
'네가 쓰는 답은 폰이 소리내어 읽어준다 — 그러니 "나는 소리를 못 낸다/터미널 프로그램이다" 같은 말은 절대 하지 마라. '
'잡담·질문이면 짧고 자연스러운 한국어로 대화하고, 코드나 파일 작업 지시면 실제로 수행한 뒤 무엇을 했는지 1~3문장으로 보고하라. '
'음성으로 읽기 좋게 간결히.\n'
'[너의 두뇌 — E: 드라이브 지식. 필요하면 이 경로를 직접 Read/Grep 해서 근거로 답하라]\n'
'· 용접 공정서/불량대책: E:\\VLA시스템 (제품별 폴더, !_XXXX_용접공정서.md — 조건표·속도·공정순서)\n'
'· 사전생성 즉답 QA뱅크: E:\\VLA시스템\\_QA뱅크\n'
'· 숙련 지식 씨앗(GOLD 등): E:\\지식정제시스템\\05_KNOWLEDGE\\seeds\n'
'· 옵시디언 노트(도진웍스): E:\\옵시디언\\도진웍스\n'
'· 뷰어/티칭 시스템 코드: E:\\도진팩토리\\3D스캔및티칭시스템 (현재 작업 폴더)\n'
'숫자·사실은 반드시 위 문서에서 확인하고 출처(파일명)를 밝혀라. 지어내지 마라.\n\n'
'[대표의 말]\n')
def run_claude_dev(prompt, model='opus'):
"""폰 개발모드 — claude 헤드리스 에이전트(파일수정·명령 실제수행 + 음성대화). tailnet 전용."""
if model not in ('sonnet', 'opus'): model = 'opus'
with _CLAUDE_LOCK: # 한 번에 하나씩만 실행 (동시 호출 시 빈 응답 방지)
r = subprocess.run(
[CLAUDE_EXE, '-p', DEV_VOICE_FRAME + prompt, '--model', model, '--dangerously-skip-permissions'],
cwd=PROJECT_DIR, capture_output=True, timeout=600)
out = r.stdout.decode('utf-8', errors='ignore').strip()
err = r.stderr.decode('utf-8', errors='ignore').strip()
if out: return out
# 재시도 1회 (드물게 빈 출력)
with _CLAUDE_LOCK:
r = subprocess.run([CLAUDE_EXE, '-p', DEV_VOICE_FRAME + prompt, '--model', model, '--dangerously-skip-permissions'],
cwd=PROJECT_DIR, capture_output=True, timeout=600)
out = r.stdout.decode('utf-8', errors='ignore').strip()
return out or ('죄송합니다, 응답 생성에 실패했어요. 다시 한 번 말씀해 주세요. ' + err[:150])
DEV_PAGE = '''
SHIN 개발모드 (클로드 코드)⚒ SHIN 개발모드 클로드 코드 · 폰🔊
이 PC의 클로드 코드가 실제로 파일을 고치고 명령을 실행합니다. 작업을 한국어로 지시하세요. 예: "dzw_process.js에서 SHIN 패널 폰 폭을 100vw로 바꿔줘"
'''
def ask_claude(q, ctx, model='sonnet'):
"""클로드 코드 CLI 경유 답변 — 웹/폰 SHIN과 연동 (CEO 지시 2026-07-06). model: sonnet|opus|claude-fable-5"""
if model not in ('sonnet', 'opus', 'claude-fable-5'): model = 'sonnet' # [CEO 2026-07-18] 페이블 허용 — 이 줄이 페이블을 몰래 소넷으로 바꾸고 있었음
# [CEO 2026-07-19 확정 원칙] 응축 금지·인용 우선 — 대표가 꼼꼼히 써둔 기록을 AI 요약으로 뭉개면
# 마지막에 듣는 사람(작업자)에게 대표의 고생이 전달되지 않는다. SHIN은 옆에 선 선배처럼 말한다.
frame = ('너는 도진웍스 용접공정 코파일럿 SHIN이다. 듣는 사람은 현장 작업자(사람)다. '
'옆에서 일 가르치는 선배처럼, 완결된 사람 말로 설명하라. '
'핵심 원칙: 근거 기록에 있는 대표(장인)의 표현과 설명은 요약해 뭉개지 말고 그대로 살려 인용하라. '
'"왜 그런지"까지 풀어서 말하라. 목록 나열로 때우지 마라. '
'발췌에 없는 내용은 일반 용접공학 지식임을 밝혀라. 숫자를 지어내지 마라. '
# [CEO 2026-07-22] 답변 스타일 — 현장 작업자 상대. 결론 먼저, 짧게, 끝은 되묻기.
'스타일 규칙: ①결론을 맨 앞에 한 줄로 먼저 말하고 그 다음에 이유를 풀어라. '
'②현장 작업자가 알아듣는 쉬운 말로. 전체 10~15줄을 넘기지 마라(길게 늘어놓지 마라). '
'③모르면 지어내지 말고 "확인 필요"라고 말하라. '
'④답의 마지막은 반드시 대표에게 다음 행동을 고르게 하는 되묻는 질문 한 줄로 끝내라. '
'근거 기록에 되물을 문구가 지정돼 있으면 그 문구를 그대로 쓴다. ')
if model == 'claude-fable-5':
# [CEO 2026-07-19 확정] 장인식 진단 포맷 — 증상→의심→확인→판단→조치. 원인 나열 금지.
frame += ('불량·문제 진단 질문이면 장인이 옆에서 진단하듯 답하라: '
'①가장 의심되는 가설 하나를 먼저 말한다("저는 ~부터 의심합니다") + 그렇게 보는 이유. '
'②확신의 정도를 밝힌다(사진/정보만으로 확정 못 하면 못 한다고). '
'③지금 바로 확인할 것 2~3가지를 가능성 높은 순서로 — 하나씩 제거하는 순서다. 원인을 몽땅 나열하지 마라. '
'④근거 기록·사례를 연결하되, 조건이 다른 제품의 수치는 "그대로 적용하지 마세요" 안전장치를 붙인다. '
'근거에 판단트리(TREE-)가 있으면 그 트리 순서를 따라라. 단순 조회면 짧게. ')
prompt = frame + '\n\n[근거]\n' + (ctx or '')[:6000] + '\n\n[질문] ' + q
r = subprocess.run([CLAUDE_EXE, '-p', '--model', model], input=prompt.encode('utf-8'),
capture_output=True, timeout=180)
out = r.stdout.decode('utf-8', errors='ignore').strip()
if r.returncode != 0 or not out:
raise RuntimeError(r.stderr.decode('utf-8', errors='ignore')[:200] or 'claude 빈 응답')
return out
def ask_ollama(q, ctx, model):
# [CEO 2026-07-18] 로컬에도 페이블 추론틀 적용 — 판단 질문이면 현상→원인→결론 순, 불확실은 "확인 필요"
prompt = ('너는 도진웍스 용접공정 코파일럿 SHIN이다. 아래 공정서 발췌를 최우선 근거로 한국어로 답하라. '
'판단이 필요한 질문이면 ①현상 ②원인·근거(출처) ③결론 순으로 답하고, 단순 조회면 2~3문장. '
'근거가 없으면 "공정서에 없음", 불확실하면 "확인 필요"라고 말하라. 숫자를 지어내지 마라.\n\n[공정서]\n' + (ctx or '')[:6000] + '\n\n[질문] ' + q)
body = json.dumps({'model': model, 'prompt': prompt, 'stream': False,
'options': {'temperature': 0.2, 'num_predict': 220}}).encode()
r = urllib.request.urlopen(urllib.request.Request(
'http://localhost:11434/api/generate', data=body,
headers={'Content-Type': 'application/json'}), timeout=90)
return json.loads(r.read()).get('response', '').strip()
# ─────────────────────────────────────────────────────────────────────────────
# [CEO 2026-07-22 자비스화 1단계] 기본 대화(스몰토크) 층
# 목적: "안녕", "내 말 들려?", "지금 몇 시야" 같은 인사·상태 질문에 엉뚱한 용접 씨앗을
# 물어오지 않고 즉답하게 한다. (시연 중 사고 방지)
# 안전장치 3중 게이트 — 하나라도 어긋나면 무조건 기존 경로(씨앗검색/LLM)로 넘긴다:
# ① 짧아야 함(공백 제거 24자 이하) ② 인사/상태 허용목록에 정확히 걸려야 함
# ③ 제관·용접 도메인 낱말이 하나라도 있으면 스몰토크 아님 (클라이언트 wantDoc 낱말 재사용)
# → 시연 질문 "24톤 하도 준설차 작업 준비 뭐부터…"는 ①②③ 모두에서 걸러져 기존 경로 유지.
_DOMAIN_RE = re.compile(r'공정|용접|어떻게|들어왔|시작|셋팅|세팅|조건|불량|가접|준비|제작|자재|입고|부터|'
r'도면|치수|씨앗|로봇|티칭|판A|판B|준설|크레인|붐|각도|토치|전류|전압|사진|핀|'
r'경로|지그|절단|취부|조립|검사|도장|\d{3,}')
def pick_engine(eng):
"""[CEO 2026-07-22 시연 안정화] 답변 엔진 기본값 = Claude 소넷.
· 브라우저에 저장된 값(dzw_shin_engine)이 비어 있거나 알 수 없으면 Claude 로 떨어진다.
· deepseek-r1 계열은 추론용( 출력·14B)이라 공정서 근거 12,000자를 물면
Ollama 90초 응답한도를 넘겨 답이 비고, 대신 엉뚱한 문서 조각이 화면에 나갔다(실측 사고).
→ 서버에서 Claude 소넷으로 승격한다. llama3.1·qwen 등 다른 로컬 모델 선택지는 그대로 둔다.
"""
e = (eng or '').strip()
if not e or e.lower() in ('undefined', 'null', 'auto', 'default'):
return ('claude' if os.path.exists(CLAUDE_EXE) else 'ollama:llama3.1:8b')
if 'deepseek-r1' in e.lower() and os.path.exists(CLAUDE_EXE):
return 'claude'
if not (e == 'claude' or e.startswith('claude:') or e.startswith('ollama:')):
return ('claude' if os.path.exists(CLAUDE_EXE) else 'ollama:llama3.1:8b')
return e
def smalltalk(q):
"""인사·상태 질문이면 즉답 문자열, 아니면 None(=기존 경로 그대로)."""
t = (q or '').strip()
if not t: return None
s = re.sub(r'[\s,.!?~…·]+', '', t)
if len(s) > 24: return None # ① 길면 스몰토크 아님
if _DOMAIN_RE.search(t): return None # ③ 도메인 낱말 있으면 절대 스몰토크 아님
now = time.localtime()
# ② 허용목록 (앞쪽이 우선)
if re.search(r'몇시|지금몇|시간알려|시각', s):
h, ap = now.tm_hour, '오전'
if h >= 12: ap, h = '오후', (h - 12 if h > 12 else 12)
if h == 0: h = 12
return '지금 %s %d시 %d분입니다.' % (ap, h, now.tm_min)
if re.search(r'며칠|몇일|오늘날짜|날짜|무슨요일|요일', s):
yo = '월화수목금토일'[now.tm_wday]
return '오늘은 %d년 %d월 %d일 %s요일입니다.' % (now.tm_year, now.tm_mon, now.tm_mday, yo)
if re.search(r'뭘할수|뭐할수|무엇을할수|뭐가능|기능|도와줄수|할수있는', s):
return ('저는 도진웍스 현장 코파일럿 SHIN입니다. 이런 걸 도와드립니다.\n'
'· 제관 공정 안내 — 어떤 순서로 작업할지 공정서 근거로 알려드립니다.\n'
'· 씨앗 지식 검색 — 대표님이 쌓아두신 현장 노하우에서 답을 찾아드립니다.\n'
'· 사진 공정서 브리핑 — 3D 화면의 핀 사진을 짚어가며 설명드립니다.\n'
'· 로봇 티칭 연계 — 말씀하시면 뷰어에서 자세·경로를 함께 확인합니다.\n'
'궁금한 작업이 있으시면 편하게 물어보세요.')
if re.search(r'누구야|누구니|누구세요|넌누구|네이름|이름이뭐|정체', s):
return '저는 도진웍스의 현장 코파일럿 SHIN입니다. 제관·용접 공정 지식을 찾아 대표님을 돕습니다.'
if re.search(r'들려|들리니|들리세요|잘들|응답해|거기있|살아있|테스트|말들려', s):
return '네, 대표님. 잘 들립니다. 말씀하세요.'
if re.search(r'고마워|고맙|감사|수고|잘했', s):
return '감사합니다, 대표님. 더 필요하시면 말씀해 주세요.'
if re.search(r'미안|죄송', s):
return '괜찮습니다, 대표님. 편하게 말씀하세요.'
if re.search(r'^(안녕|하이|헬로|여보세요|신아|샤인|굿모닝|좋은아침)', s):
return '안녕하세요, 대표님. SHIN입니다. 무엇을 도와드릴까요?'
if re.search(r'^(잘가|끝|종료|그만하자|수고했어)$', s):
return '네, 대표님. 필요하실 때 다시 불러주세요.'
if re.search(r'잘지내|기분어때|괜찮아|어때$', s):
return '저는 준비돼 있습니다, 대표님. 무엇부터 볼까요?'
return None
# ─────────────────────────────────────────────────────────────────────────────
PORT = 8093
ROOT = r'E:\VLA시스템'
CACHE = [] # [{path,name,txt_lower}]
QA = [] # QA뱅크 [{q,a,low,src}] — 사전생성 즉답 (자비스 지연 병목 제거)
def load_qa():
global QA
QA = []
qdir = os.path.join(ROOT, '_QA뱅크')
if not os.path.isdir(qdir): return
for fn in os.listdir(qdir):
if not fn.endswith('.qa.json'): continue
try:
d = json.load(open(os.path.join(qdir, fn), encoding='utf-8'))
for it in d.get('qa', []):
QA.append({'q': it['q'], 'a': it['a'], 'low': it['q'].lower(), 'src': d.get('src', fn)})
except Exception:
pass
print('[지식서버] QA뱅크:', len(QA), '문항')
# 동의어 정규화 — 현장 표현 흔들림 흡수 (같은 뜻 → 같은 대표어)
SYN = [
(r'어떻게\s*(시작|들어가|해|하면|가야|진행)\w*|들어가야?\s*[돼되해]\w*|뭐부터|어디서부터|착수|절차|순서|시작\w*', '순서'),
(r'해야\s*[돼되해할]\w*|하면\s*[돼되]\w*|할까요?|인가요?|입니까', ''),
(r'스피드|빠르기|속도\w*', '속도'),
(r'암페어|전류\w*|몇\s*[aA]\b', '전류'),
(r'볼트|전압\w*', '전압'),
(r'얼마야|얼마|몇이야|몇\b|알려줘|알려주|말해줘|뭐야|뭔데|어떻게\s*돼', ''),
(r'보스|boss', 'boss'),
(r'너클붐|크레인', '12500'),
(r'트랙\s*서포트', '트랙서포트'),
(r'용접\s*조건|셋팅\s*값|설정\s*값', '조건'),
(r'왜\s*생[겨기]\w*|생기는\s*이유|원인이?\s*뭐\w*', '이유'),
(r'줄이려면|줄이는|줄일려면|잡으려면|안\s*나게|방지\w*', '줄이'),
(r'방법|하는\s*법', ''),
]
def norm(s):
s = s.lower()
for pat, rep in SYN: s = re.sub(pat, rep, s)
s = re.sub(r'[,?？!.~]+', ' ', s)
# 조사·어미는 단어 끝에서만 제거 (단어 내부 글자 파괴 금지 — "순서/속도" 보존)
toks = []
for t in s.split():
prev = None
while t != prev:
prev = t
t = re.sub(r'(?<=[가-힣])(은|는|이|가|을|를|의|로|에서|부터|도|좀|요|이야|야|만)$', '', t)
if t: toks.append(t)
return ' '.join(toks)
def bigrams(s):
s = s.replace(' ', '')
return {s[i:i+2] for i in range(len(s)-1)} if len(s) > 1 else {s}
def qa_match(q):
"""정규화+제품앵커+바이그램 유사도. 애매하면 None → 상위 파이프라인이 검색/LLM 폴백"""
nq = norm(q)
if len(nq) < 2: return (None, 0)
# 제품번호 앵커: 질문에 도번(4자리+)이 있으면 그 공정서 QA만 후보
anchors = re.findall(r'\d{4,}|[a-z]-?\d{6,}', nq)
qb = bigrams(nq)
scored = []
for d in QA:
if anchors and not any(an in (d['src'] + d['q'] + d['a']).lower() for an in anchors):
continue
nd = d.get('nq') or d.setdefault('nq', norm(d['q']))
db = bigrams(nd)
sim = len(qb & db) / max(1, len(qb | db)) # 자카드 유사도
kw = sum(1 for t in nq.split() if len(t) >= 2 and t in nd) # 정규화 키워드 겹침
scored.append((sim + kw * 0.15, d, sim))
if not scored: return (None, 0)
scored.sort(key=lambda x: -x[0])
top = scored[0]
# [2026-07-22] 유사도 하한 — kw 보너스(0.15×개수)만으로 게이트를 넘는 오답 차단.
# '작업/부분' 같은 흔한 낱말 3개만 겹쳐도 0.45가 채워져, 질문과 무관한 즉답이 나가고 있었다.
# (실측: 무관한 준설차 질문 sim=0.045인데 kw=3으로 total 0.495 → 오답 즉답.
# 반면 정상 즉답들은 sim 0.417~1.000. 그 사이 빈 구간에 하한을 둔다.)
if top[2] < 0.25: return (None, 0)
# 산업용 게이트: 1위가 약하거나(오답 위험) 2위와 박빙(모호)이면 즉답 포기
if top[0] < 0.45: return (None, 0)
if len(scored) > 1 and scored[1][1]['a'] != top[1]['a'] and scored[1][0] > top[0] * 0.92:
return (None, 0)
return (top[1], round(top[0], 2))
CONSTITUTION_PATH = r'E:\지식정제시스템\05_KNOWLEDGE\헌법서\AI_헌법서_v3.0.json'
def _load_constitution():
"""헌법서 각 조항을 검색 문서로 편입. 원본 무수정. 이름은 '헌법-'로 최상위 가중."""
if not os.path.exists(CONSTITUTION_PATH):
print('[지식서버] 헌법서 파일 없음:', CONSTITUTION_PATH); return
c = json.load(open(CONSTITUTION_PATH, encoding='utf-8'))
n = 0
def add(item):
nonlocal n
if not isinstance(item, dict): return
lid = item.get('law_id') or item.get('id') or item.get('rule_id') or '?'
name = item.get('name_ko') or item.get('name') or item.get('title') or ''
title = '헌법-%s %s' % (lid, name)
body = json.dumps(item, ensure_ascii=False)
CACHE.append({'path': CONSTITUTION_PATH + '#' + str(lid), 'name': title,
'txt': title + '\n' + body, 'low': (title + '\n' + body).lower(),
'head': title.lower()})
n += 1
# 절대금지 규칙
for rc in (c.get('hard_constraints', {}) or {}).get('rules', []) or []:
add(rc)
# 상황 정의 엔진
for ce in (c.get('context_engine', {}) or {}).get('laws', []) or []:
add(ce)
# 축별 법칙 (HUMAN/MNT/DES/MOT/SNS/NRG/HRM) → 각 축의 items
for axis, ax in (c.get('laws', {}) or {}).items():
for it in (ax or {}).get('items', []) or []:
add(it)
# AI 의사결정 원칙
for ap in (c.get('ai_decision_principles', {}) or {}).get('principles', []) or []:
add(ap)
print('[지식서버] 헌법서 조항 편입:', n, '개')
def build_index():
# [CEO 2026-07-18] 읽기범위 확장 — 기존엔 VLA시스템의 '공정서/불량/목록' md만 읽어
# 씨앗 214개(GOLD 59 포함)를 하나도 못 보고 있었음. 씨앗 폴더 추가 + 파일명 제한 완화(모든 md).
global CACHE
CACHE = []
# [CEO 2026-07-19] 헌법서 로딩 — SHIN의 최상위 기준. 각 조항을 검색 대상 문서로 편입(원본 JSON 무수정, 읽기만).
try:
_load_constitution()
except Exception as e:
print('[지식서버] 헌법서 로딩 실패:', str(e)[:120])
roots = [ROOT, r'E:\지식정제시스템\05_KNOWLEDGE\seeds',
r'E:\옵시디언'] # [CEO 2026-07-18] 옵시디언 볼트 연결 — CEO가 옵시디언에 쓰면 즉시 SHIN 지식이 됨
# [CEO 2026-07-19] 넣을 것: .md(지식) + .txt(현장 음성기록=구술 암묵지) + .csv(용접조건표).
# 넣지 않을 것: .pdf/.dxf/.dwg 도면 원본(CEO IP규칙 — AI·임베딩 반입 금지). 로그·자막광고·빈 파일도 제외.
stt = 0
for root in roots:
for dp, _, fns in os.walk(root):
if any(x in dp for x in ('EMBEDDINGS','MODELS','OUTPUT','FINETUNE')): continue
for fn in fns:
fl = fn.lower()
if not (fl.endswith('.md') or fl.endswith('.txt') or fl.endswith('.csv')): continue
p = os.path.join(dp, fn)
try:
txt = open(p, encoding='utf-8', errors='ignore').read()
except Exception:
continue
if fl.endswith('.txt') or fl.endswith('.csv'):
clean = re.sub(r'자막제작.*|https?://\S+|by UpTitle', '', txt)
if len(re.findall(r'[가-힣]', clean)) < 40: continue # 빈 STT·자막광고·잡파일 차단
if '_stt' in fl or 'stt_results' in dp.lower(): stt += 1
# [2026-07-22] 경로도 검색 대상에 포함 — 발주처·제품 폴더명(예: 동우인더스트리\하수도준설차\24톤)이
# 파일명·본문에 없어서 "동우인더스트리 준설차"로 물으면 그 폴더 문서가 아예 안 잡혔다.
rel = p.replace('\\', ' ').replace('/', ' ')
CACHE.append({'path': p, 'name': fn, 'txt': txt,
'low': (rel + '\n' + fn + '\n' + txt).lower(),
'head': (rel + ' ' + fn).lower(),
# 기계판독 가능한 공정서인가 = [공정N] 단계 또는 조건표(전류)를 가졌는가.
# 뷰어는 이름에 '공정서'가 들어간 1위 문서를 자동으로 해석해 화면에 적용하는데,
# 단계·조건표가 없는 서술형 문서를 해석하면 "속도: - / - / -" 같은 빈 껍데기가 나간다.
'spec': bool(re.search(r'\[공정\s*\d+\]', txt) or re.search(r'\|\s*\*?\*?전류', txt))})
print('[지식서버] 공정 문서 인덱스:', len(CACHE), '건 (VLA+씨앗, 현장구술 %d건 포함)' % stt)
# [CEO 2026-07-22 시연대비] 검색어 별칭표 — 같은 물건을 부르는 다른 말·음성인식 오인식을 표준어로 바꾼다.
# 근거: 이 검색은 '글자 그대로 들어있는지'만 본다. 그래서 CEO가 "두설차"(준설차 오인식)나 "동우"라고만
# 말하면 문서에 그 글자가 없어 0점이 된다. 가중치를 손대는 게 아니라 '부르는 이름'을 맞춰주는 것이므로
# 해당 낱말이 없는 질문(예: 12500 용접)에는 아무 영향이 없다.
QUERY_ALIAS = [
(r'두설차|준설처|준설챠|즌설차|준썰차|준설\s*차|오수차|흡입차|하수도\s*차', '준설차'),
(r'하도\s*준설차|하수도\s*준설차', '준설차'),
(r'동우\s*인더스트리|동우\s*인더스|동우인더스|동우', '동우인더스트리'),
(r'24\s*t\b|24\s*톤|24\s*돈|이십사\s*톤|이십사톤', '24톤'),
(r'작업\s*공정서|공정\s*서', '공정서'),
]
def _alias(q):
s = q.lower()
for pat, rep in QUERY_ALIAS:
s = re.sub(pat, rep, s)
return s
def search(q):
# 조사 떼기 — "24톤을/준설차를/공정서를"이 붙은 채로는 문서의 "24톤·준설차·공정서"와 글자가 안 맞는다.
terms = []
for t in _alias(q).replace(',', ' ').replace('.', ' ').replace('?', ' ').split():
t = re.sub(r'(?<=[가-힣0-9])(을|를|은|는|이|가|의|에|에서|으로|로|과|와|도|만|랑|이랑)$', '', t)
if len(t) >= 2: terms.append(t)
out = []
for d in CACHE:
s = 0
for t in terms:
if t in d.get('head', d['name'].lower()): s += 5 # 파일명 또는 폴더경로(발주처·제품) 일치
s += min(5, d['low'].count(t))
# [CEO 2026-07-19 설계] 지식 층위 = 헌법(최상위) → 법칙 → 판단트리 → 패턴노트 → 사례. 검색 가중치로 그 순서 재현.
# [2026-07-22 원인규명 후 수정] 층위 가중을 '덧셈(+22 등)'에서 '곱셈'으로 바꿈.
# 원인: 덧셈이라 '작업' 같은 흔한 낱말 하나만 걸려도 헌법 조항이 +22를 받아
# 제품 공정서·씨앗을 제치고 1위가 됐다(실측: 동우인더스트리 준설차 질문 1위 = 헌법-HUMAN-01).
# 층위 가중은 "관련 있을 때 순위를 올리는 것"이지 "무관한 문서를 1위로 만드는 것"이 아니다.
if s > 0 and d['name'].startswith('헌법-'): s = int(s * 1.6) # 헌법서 — 최상위 기준(관련 있을 때만)
if s > 0 and d['name'].startswith('LAW-'): s = int(s * 1.5) # 법칙(원리)
if s > 0 and d['name'].startswith('TREE-'): s = int(s * 1.4)
if s > 0 and d['name'].startswith('PATTERN-'): s = int(s * 1.3)
if s > 0:
i = max(0, d['low'].find(terms[0]) - 40) if terms else 0
# 단계·조건표가 없는 서술형 문서는 이름의 '공정서'를 '안내서'로 표기해 뷰어 자동해석을 피한다
# (원본 파일명·경로는 그대로. 표시 이름만 바꿔 빈 껍데기 적용 사고를 막는다.)
nm = d['name'] if d.get('spec', True) else d['name'].replace('공정서', '안내서')
out.append({'path': d['path'], 'name': nm, 'score': s,
'snippet': d['txt'][i:i+180].replace('\n', ' ')})
out.sort(key=lambda x: -x['score'])
# [2026-07-22] 같은 파일이 두 뿌리(VLA·지식정제시스템)에 중복 색인돼 상위 3칸을 같은 문서가 차지했다.
# → 근거로 붙는 문서가 실제로는 1종뿐이 되는 문제. 파일명 기준 중복 제거(점수 높은 쪽만 남김).
seen, uniq = set(), []
for h in out:
if h['name'] in seen: continue
seen.add(h['name']); uniq.append(h)
# [2026-07-22] 헌법 조항은 '판단 기준'이지 '질문의 답'이 아니다. 1위로 올라오면 화면 답변의 근거가
# 헌법 한 줄로 바뀌어 엉뚱한 답이 나간다(실측: "동우인더스트리 준비할 거?" → 헌법-HUMAN-01 1위).
# → 다른 문서가 하나라도 있으면 헌법은 1위 자리를 양보한다(순위만 조정, 근거에서 빼지는 않는다).
if uniq and uniq[0]['name'].startswith('헌법-'):
for i, h in enumerate(uniq):
if not h['name'].startswith('헌법-'):
uniq.insert(0, uniq.pop(i)); break
return uniq[:8]
class H(http.server.BaseHTTPRequestHandler):
def _send(self, code, obj):
body = json.dumps(obj, ensure_ascii=False).encode('utf-8')
self.send_response(code)
self.send_header('Content-Type', 'application/json; charset=utf-8')
self.send_header('Access-Control-Allow-Origin', '*')
self.send_header('Content-Length', str(len(body)))
self.end_headers()
self.wfile.write(body)
def do_GET(self):
u = urllib.parse.urlparse(self.path)
qs = urllib.parse.parse_qs(u.query)
if u.path == '/api/search':
self._send(200, {'hits': search(qs.get('q', [''])[0])})
elif u.path == '/api/read':
p = qs.get('p', [''])[0]
rp = os.path.abspath(p)
if not rp.startswith(os.path.abspath(ROOT)) or not rp.endswith('.md'):
self._send(403, {'error': 'VLA 폴더 내 md만 허용'}); return
try:
self._send(200, {'name': os.path.basename(rp),
'text': open(rp, encoding='utf-8', errors='ignore').read()})
except Exception as e:
self._send(404, {'error': str(e)})
elif u.path == '/api/qa':
hit, sc = qa_match(qs.get('q', [''])[0])
self._send(200, {'hit': hit and {'q': hit['q'], 'a': hit['a'], 'src': hit['src'], 'score': sc}})
elif u.path == '/api/models':
models = []
if os.path.exists(CLAUDE_EXE):
models.append({'id': 'claude', 'label': 'Claude 소넷'})
models.append({'id': 'claude:opus', 'label': 'Claude 오푸스'})
models.append({'id': 'claude:claude-fable-5', 'label': 'Claude 페이블'}) # [CEO 2026-07-18] 최상위 모델
try:
r = urllib.request.urlopen('http://localhost:11434/api/tags', timeout=5)
for m in json.loads(r.read()).get('models', []):
if any(k in m['name'] for k in ('llama3.1', 'qwen2.5vl', 'deepseek-r1', 'gemma3')):
models.append({'id': 'ollama:' + m['name'], 'label': m['name'] + ' (로컬)'})
except Exception: pass
self._send(200, {'models': models})
elif u.path == '/api/reload':
build_index(); load_qa()
self._send(200, {'ok': True, 'docs': len(CACHE), 'qa': len(QA)})
elif u.path == '/api/tts':
try:
global TTS_VOICE
v = qs.get('voice', [''])[0]
if v in ('ko-KR-SunHiNeural', 'ko-KR-InJoonNeural', 'ko-KR-HyunsuMultilingualNeural'):
TTS_VOICE = v # 다음 요청부터의 기본 목소리(기존 동작 유지)
# 이번 요청에 쓸 목소리는 지역 변수로 넘긴다 — 동시 요청이 서로의 목소리를 바꾸지 않게
data, ctype = tts(qs.get('text', [''])[0], v if v else TTS_VOICE)
self.send_response(200)
self.send_header('Content-Type', ctype)
self.send_header('Access-Control-Allow-Origin', '*')
self.send_header('Content-Length', str(len(data)))
self.end_headers(); self.wfile.write(data)
except Exception as e:
self._send(501, {'error': str(e)})
elif u.path == '/dev' or u.path == '/dev/':
body = DEV_PAGE.encode('utf-8')
self.send_response(200)
self.send_header('Content-Type', 'text/html; charset=utf-8')
self.send_header('Content-Length', str(len(body)))
self.end_headers(); self.wfile.write(body)
elif u.path == '/api/ping':
self._send(200, {'ok': True, 'docs': len(CACHE), 'qa': len(QA)})
else:
self._send(404, {'error': 'unknown'})
def do_POST(self):
# [CEO 혁신①→대화형 2026-07-19] 음성 인터뷰 — SHIN이 묻고 CEO가 말로 답하면 판정·되묻기·저장
if self.path == '/api/interview/next':
# 다음 미답 구멍 질문 1개 반환 (질문큐.json)
try:
qp = r'E:\VLA시스템\KNOWLEDGE\_역질문\질문큐.json'
data = json.load(open(qp, encoding='utf-8')) if os.path.exists(qp) else {'items': []}
nxt = next((it for it in data.get('items', []) if not it.get('answered')), None)
self._send(200, {'q': nxt, 'remaining': sum(1 for it in data.get('items', []) if not it.get('answered'))})
except Exception as e:
self._send(200, {'q': None, 'error': str(e)[:150]})
return
if self.path == '/api/interview/answer':
# {id, q, a(CEO 음성답)} → 페이블이 판정: 충분하면 저장, 애매하면 되물음 한 문장
try:
n = int(self.headers.get('Content-Length', 0))
req = json.loads(self.rfile.read(n))
q, a, qid = req.get('q', ''), req.get('a', ''), req.get('id', '')
judge = ('너는 장인의 암묵지를 캐내는 인터뷰어 SHIN이다. 아래 질문에 대한 대표(장인)의 구술 답을 보고 '
'JSON 한 줄로만 답하라: {"enough":true/false,"followup":"되물을 한 문장(enough=false일 때만)",'
'"clean":"지식으로 저장할 완결된 한두 문장(enough=true일 때, 대표 표현 살려서)"}. '
'수치가 빠졌거나 애매하면 enough=false로 딱 한 가지만 더 물어라. 지어내지 마라.\n'
'[질문] %s\n[대표 구술답] %s' % (q, a))
try:
verdict = ask_claude(judge, '', 'claude-fable-5')
except Exception:
verdict = ask_ollama(judge, '', 'llama3.1:8b')
m = re.search(r'\{.*\}', verdict, re.S)
v = json.loads(m.group(0)) if m else {'enough': True, 'clean': a}
if v.get('enough'):
# 저장: 증류쌍 + 흡수본 md + 질문큐 answered 표시
dd = r'E:\VLA시스템\FINETUNE\distill_pairs'; os.makedirs(dd, exist_ok=True)
with open(os.path.join(dd, 'ceo_interview.jsonl'), 'a', encoding='utf-8') as f:
f.write(json.dumps({'ts': time.strftime('%Y-%m-%d %H:%M:%S'), 'via': 'voice_interview',
'q': q, 'a': a, 'clean': v.get('clean', a)}, ensure_ascii=False) + '\n')
ad = r'E:\VLA시스템\KNOWLEDGE\_역질문'; os.makedirs(ad, exist_ok=True)
with open(os.path.join(ad, '_CEO답변_흡수본.md'), 'a', encoding='utf-8') as f:
f.write('\n**Q: %s**\n- A(구술): %s\n- 정리: %s\n' % (q, a, v.get('clean', a)))
try:
qp = os.path.join(ad, '질문큐.json'); d = json.load(open(qp, encoding='utf-8'))
for it in d.get('items', []):
if it.get('id') == qid: it['answered'] = True
json.dump(d, open(qp, 'w', encoding='utf-8'), ensure_ascii=False, indent=1)
except Exception: pass
self._send(200, {'enough': bool(v.get('enough')), 'followup': v.get('followup', ''),
'clean': v.get('clean', ''), 'say': ('알겠습니다, 기록했습니다. ' + v.get('clean', ''))
if v.get('enough') else v.get('followup', '조금 더 자세히 말씀해 주세요.')})
except Exception as e:
self._send(200, {'enough': False, 'say': '다시 한 번 말씀해 주세요.', 'error': str(e)[:150]})
return
if self.path == '/api/see':
# VLM 눈: {q, img(base64 jpeg)} → qwen2.5vl:7b (로컬 — 영상 외부 비유출)
try:
n = int(self.headers.get('Content-Length', 0))
req = json.loads(self.rfile.read(n))
prompt = ('너는 용접공장 코파일럿 "SHIN"의 눈이다. 사진은 용접 작업장/제품이다. '
'한국어 2~4문장으로 관찰 사실만 답하라(제품 형상, 판재 구성, 용접선 위치, 가접 상태, 이상 여부). '
'확실하지 않으면 추정임을 밝혀라.\n질문: ' + req.get('q', '무엇이 보이나'))
body = json.dumps({'model': 'qwen2.5vl:7b', 'prompt': prompt,
'images': [req.get('img', '')], 'stream': False,
'options': {'temperature': 0.2, 'num_predict': 300}}).encode()
r = urllib.request.urlopen(urllib.request.Request(
'http://localhost:11434/api/generate', data=body,
headers={'Content-Type': 'application/json'}), timeout=180)
self._send(200, {'answer': json.loads(r.read()).get('response', '').strip()})
except Exception as e:
self._send(501, {'error': 'VLM 불가: ' + str(e)})
return
if self.path == '/api/log':
# SHIN 세션 로그 = 암묵지 자산 (대표 말·SHIN 작업·콘솔 제어 전부 기록)
try:
n = int(self.headers.get('Content-Length', 0))
rec = json.loads(self.rfile.read(n).decode('utf-8', errors='replace'))
logdir = os.path.join(PROJECT_DIR, 'SHIN세션로그')
os.makedirs(logdir, exist_ok=True)
day = (rec.get('t', '') or '0000-00-00')[:10] or 'undated'
with open(os.path.join(logdir, day + '.jsonl'), 'a', encoding='utf-8') as f:
f.write(json.dumps(rec, ensure_ascii=False) + '\n')
self._send(200, {'ok': True})
except Exception as e:
self._send(501, {'error': str(e)})
return
if self.path == '/api/dev':
try:
n = int(self.headers.get('Content-Length', 0))
req = json.loads(self.rfile.read(n).decode('utf-8', errors='replace'))
# [2026-07-22] 인사·상태 질문은 개발모드(클로드 CLI)로 새지 않게 여기서 즉답
# 클라이언트는 '[화면상태…]\n[대표의 말] 안녕' 형태로 보낸다 → 대표의 말만 떼어내 판별
_p = req.get('prompt', '')
_said = _p.split('[대표의 말]')[-1] if '[대표의 말]' in _p else _p
st = smalltalk(_said)
if st:
self._send(200, {'answer': st, 'engine': 'smalltalk'}); return
ans = run_claude_dev(req.get('prompt', ''), req.get('model', 'opus'))
self._send(200, {'answer': ans})
except Exception as e:
self._send(501, {'error': str(e)})
return
if self.path == '/api/act':
# [CEO 자비스 3단계 2026-07-20] 자연어→행동. LLM이 JSON 행동을 내놓고, 순수 질문이면 actions:[]+say.
# 기존 티어(Ollama→Claude CLI) 재사용. 로봇 수학 없음 — 클라이언트가 기존 jog/goto 함수로만 실행.
try:
n = int(self.headers.get('Content-Length', 0))
req = json.loads(self.rfile.read(n))
q, ctx = req.get('q', ''), req.get('ctx', '')
# [2026-07-22] 인사·상태 질문이면 로봇 행동 해석 없이 즉답 (actions 비움)
st = smalltalk(q)
if st:
self._send(200, {'actions': [], 'say': st}); return
hist = req.get('history', []) or []
if hist:
hl = '\n'.join(('%s: %s' % (h.get('role', ''), str(h.get('text', ''))[:300])) for h in hist[-6:])
ctx = ('[직전 대화]\n' + hl + '\n\n' + (ctx or ''))
# 순수 질문 대비 공정서 근거 동봉 (/api/ask 와 동일 grounding)
try:
hits = search(q)
if hits:
frag = []
for h in hits[:3]:
d = next((x for x in CACHE if x['path'] == h['path']), None)
if d: frag.append('### ' + h['name'] + '\n' + d['txt'][:1200])
if frag: ctx = (ctx + '\n\n[공정서 근거]\n' + '\n\n'.join(frag))[:9000]
except Exception: pass
schema = (
'너는 3D 로봇 티칭 뷰어를 직접 조작하는 코파일럿 "SHIN"이다. 대표의 말을 로봇 행동으로 번역한다.\n'
'반드시 JSON 한 줄로만 답하라(설명 문장 금지). 형식:\n'
'{"actions":[{"type":"adjust_joint","step":"current","axis":"B","delta_deg":-5,"reason":"토치 각도 과대"}],'
'"say":"현재 스텝 B축을 5도 안쪽으로 접었습니다"}\n'
'지원 type: adjust_joint(관절 미세조정; axis는 S,L,U,R,B,T 중 하나, delta_deg는 상대각 -45~45, '
'step은 "current" 또는 스텝번호), goto_step(step=번호), play, pause, stop, '
'run_collision(충돌검사), show_weld_path(용접경로), add_point(포인트추가), '
'go_mech_zero(기계적원점), go_work_home(홈자세), analyze_u(U축분석), undo(직전조정 되돌리기).\n'
'규칙: axis는 반드시 S,L,U,R,B,T 중 하나. delta_deg 범위 -45~45. '
'"안쪽으로 접어/각도 줄여/숙여"=음수, "펴/키워/세워"=양수 (문맥으로 판단). '
'조작이면 actions를 채우고 say엔 무엇을 얼마나 바꿨는지 한 문장. '
'행동이 아닌 순수 질문이면 actions는 빈 배열 []로 두고 say에 한국어 답만 넣어라. 지어내지 마라.\n'
'[현재 상태] ' + (ctx or '') + '\n[대표 말] ' + q
)
eng = pick_engine(req.get('engine')) # [2026-07-22] 기본값 Claude 강제
try:
if eng == 'claude' or eng.startswith('claude:'):
cm = eng.split(':', 1)[1] if ':' in eng else 'sonnet'
raw = ask_claude(schema, '', cm)
else:
raw = ask_ollama(schema, '', eng.split(':', 1)[1])
except Exception:
raw = ask_ollama(schema, '', 'llama3.1:8b')
m = re.search(r'\{.*\}', raw, re.S)
try:
obj = json.loads(m.group(0)) if m else {'actions': [], 'say': raw.strip()[:600]}
except Exception:
obj = {'actions': [], 'say': raw.strip()[:600]}
# 서버측 안전 정제: 화이트리스트 + 축검증 + 각도 클램프
AXES = set('SLURBT')
OK = {'adjust_joint', 'jog', 'goto_step', 'play', 'pause', 'stop', 'run_collision',
'show_weld_path', 'add_point', 'go_mech_zero', 'go_work_home', 'analyze_u', 'undo'}
clean = []
for a in (obj.get('actions', []) or []):
if not isinstance(a, dict): continue
t = a.get('type') or a.get('action')
if t not in OK: continue
if t in ('adjust_joint', 'jog'):
ax = str(a.get('axis', '')).upper()
if ax not in AXES: continue
try: dl = float(a.get('delta_deg', a.get('delta', 0)))
except Exception: dl = 0.0
dl = max(-45.0, min(45.0, dl))
a = {'type': 'adjust_joint', 'axis': ax, 'delta_deg': dl,
'step': a.get('step', 'current'), 'reason': str(a.get('reason', ''))[:120]}
clean.append(a)
self._send(200, {'actions': clean, 'say': str(obj.get('say', ''))[:600]})
except Exception as e:
self._send(200, {'actions': [], 'say': '', 'error': str(e)[:200]})
return
if self.path != '/api/ask':
self._send(404, {'error': 'unknown'}); return
try:
n = int(self.headers.get('Content-Length', 0))
# [2026-07-22] 윈도우 curl/PowerShell 은 한글 본문을 cp949 로 보낸다 → utf-8 먼저, 실패 시 cp949.
# errors='replace' 금지(한글이 ??? 로 뭉개져 LLM 이 엉뚱한 질문에 답하게 됨)
raw = self.rfile.read(n)
try:
req = json.loads(raw.decode('utf-8'))
except UnicodeDecodeError:
req = json.loads(raw.decode('cp949', errors='replace'))
# 'question' 도 질문 키로 허용 (curl 로 점검할 때 흔히 쓰는 이름)
q, ctx = (req.get('q') or req.get('question') or ''), req.get('ctx', '')
# [2026-07-22] 기본 대화 층 — 인사·상태 질문이면 씨앗검색·LLM 타지 않고 즉답
st = smalltalk(q)
if st:
self._send(200, {'answer': st, 'engine': 'smalltalk'}); return
# [CEO 2026-07-19 기억 이식] 직전 대화를 근거 앞에 붙임 → "적용해줘" 같은 지시대명사가 앞말을 가리킬 수 있게
hist = req.get('history', []) or []
if hist:
hl = '\n'.join(('%s: %s' % (h.get('role', ''), str(h.get('text', ''))[:400])) for h in hist[-6:])
ctx = ('[직전 대화 — 이걸 기억하고 답하라. "적용해줘/그거/방금 그것"은 이 대화에서 찾아라]\n' + hl + '\n\n' + (ctx or ''))
# [CEO 2026-07-18] 공정서 근거 자동 동봉 — 여태 ctx가 비어 있어 AI가 공정서 근거 없이 맨몸으로 답하고 있었음
try:
hits = search(q)
if hits:
frag = []
for h in hits[:3]:
d = next((x for x in CACHE if x['path'] == h['path']), None)
if d: frag.append('### ' + h['name'] + ' (출처: ' + h['path'] + ')\n' + d['txt'][:1800])
if frag: ctx = (ctx + '\n\n[공정서 근거 — 자동 검색]\n' + '\n\n'.join(frag))[:12000]
except Exception: pass
eng = pick_engine(req.get('engine')) # [2026-07-22] 기본값 Claude 강제 (위 pick_engine 주석 참조)
no_fallback = bool(req.get('no_fallback', False)) # true면 지정 엔진 실패 시 폴백 금지(수동 모드용)
try:
if eng == 'claude' or eng.startswith('claude:'):
cm = eng.split(':', 1)[1] if ':' in eng else 'sonnet'
ans = ask_claude(q, ctx, cm); used = 'claude:' + cm
else:
ans = ask_ollama(q, ctx, eng.split(':', 1)[1]); used = eng
except Exception as e1:
if no_fallback:
# 폴백 없이 실패 원인을 그대로 반환 (클라이언트가 다음 티어 선택) — 200이되 answer 비움+error
self._send(200, {'answer': '', 'engine': eng, 'error': str(e1)[:200]}); return
# 선택 엔진 실패 → 로컬 폴백 (기본 동작 유지)
ans = ask_ollama(q, ctx, 'llama3.1:8b'); used = 'ollama:llama3.1:8b(폴백 — %s)' % str(e1)[:60]
# [CEO 2026-07-18] 페이블 증류 로그 — 상위모델 답변을 질문+근거+답 쌍으로 저장. 로컬 모델 파인튜닝 교재.
try:
if used.startswith('claude:'):
os.makedirs(DISTILL_DIR, exist_ok=True)
with open(os.path.join(DISTILL_DIR, 'distill_%s.jsonl' % time.strftime('%Y%m')), 'a', encoding='utf-8') as f:
f.write(json.dumps({'ts': time.strftime('%Y-%m-%d %H:%M:%S'), 'model': used,
'q': q, 'ctx': (ctx or '')[:6000], 'a': ans}, ensure_ascii=False) + '\n')
except Exception: pass
self._send(200, {'answer': ans, 'engine': used})
except Exception as e:
self._send(501, {'error': 'LLM 불가: ' + str(e)})
def do_OPTIONS(self):
self.send_response(204)
self.send_header('Access-Control-Allow-Origin', '*')
self.send_header('Access-Control-Allow-Headers', 'Content-Type')
self.end_headers()
def log_message(self, *a): pass
if __name__ == '__main__':
build_index()
load_qa()
with socketserver.ThreadingTCPServer(('127.0.0.1', PORT), H) as srv:
print('[지식서버] http://127.0.0.1:%d (로컬 전용)' % PORT)
srv.serve_forever()