feat(canonical): restore GPU STT owner and extend KGS watch paths

D9 Track B revised (2026-05-08):

1) STT owner GPU 정식 복귀:
   - docker-compose.yml: stt-service profiles:[legacy] 제거 → 상시 활성
   - fastapi STT_ENDPOINT = http://stt-service:3300 (compose 내부 DNS)
   - 정책: Mac mini = Gemma 26B 전용 우선이므로 STT/Whisper 는 호출량 무관
     GPU 서버 소유. 이전 "Mac mini 이전본" 주석은 trace 오인 기반.

2) KGS Code 등 외부 학습 자료 추가 스캔 경로:
   - ADDITIONAL_WATCH_TARGETS env (쉼표 구분, PKM 상대경로)
   - app/core/config.py: additional_watch_targets list 설정 추가
   - app/workers/file_watcher.py: 추가 watch path 처리
   - app/workers/classify_worker.py: KGS Code 분류 분기 (가스기사 학습 자료)
   - 모두 expected_category=library 처리 (md/pdf/docx 만)

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Hyungi Ahn
2026-05-10 05:18:41 +00:00
parent c1b22d8833
commit aca2f0d62c
4 changed files with 134 additions and 13 deletions
+96 -4
View File
@@ -21,9 +21,12 @@ PR-B B-1 tier triage (신규, 4B gemma Ollama):
from __future__ import annotations
import json
import re
import time
from datetime import datetime, timezone
import yaml
from pydantic import BaseModel, Field, ValidationError
from sqlalchemy import text as sql_text
from sqlalchemy.ext.asyncio import AsyncSession
@@ -301,8 +304,42 @@ def _distill(triage_out: TriageOutput, limit: int = 2000) -> str:
return "\n".join(parts)[:limit]
# ───────────────────── frontmatter 파싱 (옵션 C) ──────────────────────
# YAML frontmatter (--- ... ---) + body 분리. body 가 없거나 frontmatter 가 형식 오류여도 안전하게 fallback.
_FM_PATTERN = re.compile("^---\\s*\\n(.*?)\\n---\\s*\\n?(.*)$", re.DOTALL)
def _parse_frontmatter(extracted_text: str) -> tuple[dict, str]:
"""extracted_text 시작에 YAML frontmatter 가 있으면 (frontmatter_dict, body) 반환.
없으면 ({}, extracted_text). YAML 파싱 실패 시도 ({}, extracted_text) 로 안전 fallback.
"""
if not extracted_text or not extracted_text.startswith("---"):
return {}, extracted_text
m = _FM_PATTERN.match(extracted_text)
if not m:
return {}, extracted_text
fm_text, body = m.group(1), m.group(2)
try:
fm = yaml.safe_load(fm_text)
if not isinstance(fm, dict):
return {}, extracted_text
return fm, body
except yaml.YAMLError:
return {}, extracted_text
# frontmatter 우선 인식: code/section/source_pdf/source_pages/source_basis/verified_level/verification_pending
# 등 원문 추적 메타데이터는 LLM 이 절대 덮어쓰지 못하게 차단.
_FRONTMATTER_PRESERVED_KEYS = {
"code", "section", "source_pdf", "source_pages", "source_basis",
"verified_level", "verification_pending", "source_type", "kgs_code",
}
# ───────────────────────── main process ────────────────────────────────
async def process(document_id: int, session: AsyncSession) -> None:
"""문서 분류 + 요약 + tier triage.
@@ -334,6 +371,59 @@ async def process(document_id: int, session: AsyncSession) -> None:
if not doc.extracted_text:
raise ValueError(f"문서 ID {document_id}: extracted_text가 비어있음")
# ─── 옵션 C: markdown frontmatter 우선 인식 ───────────────────────────
# KGS Code 등 외부 작성 마크다운은 frontmatter 에 정확한 메타가 있다.
# title / tags / ai_summary / ai_domain 은 frontmatter 에 있으면 그대로 사용,
# 없는 필드만 LLM 호출. code/section/source_pages/verified_level 등 원문
# 추적 메타는 documents.md_frontmatter JSONB 에 보존하고 LLM 이 덮어쓰지 못하게 한다.
fm, body = _parse_frontmatter(doc.extracted_text)
if fm:
# frontmatter 전체를 md_frontmatter JSONB 에 저장 (원문 추적용 single source)
doc.md_frontmatter = fm
# 우선 반영 (LLM 보다 신뢰도 높음, frontmatter 가 authoritative)
if fm.get("title"):
doc.title = str(fm["title"])
fm_tags = fm.get("tags")
if isinstance(fm_tags, list) and fm_tags:
# ai_tags 에 frontmatter 태그 우선 적재 (LLM 이 추가만 가능)
doc.ai_tags = [str(t) for t in fm_tags]
if fm.get("ai_domain"):
doc.ai_domain = str(fm["ai_domain"])
parts = doc.ai_domain.split("/")
if len(parts) > 1 and not doc.ai_sub_group:
doc.ai_sub_group = parts[1]
if fm.get("ai_sub_group"):
doc.ai_sub_group = str(fm["ai_sub_group"])
if fm.get("document_type"):
doc.document_type = str(fm["document_type"])
if fm.get("ai_summary"):
doc.ai_summary = str(fm["ai_summary"])
if fm.get("importance") in ("high", "medium", "low"):
doc.importance = fm["importance"]
# 핵심 메타 (title + ai_domain + ai_summary) 가 모두 frontmatter 로 채워졌으면
# LLM classify/summarize 스킵. tier triage 도 스킵 (frontmatter 가 더 정확).
# frontmatter 미커버 필드는 그대로 두어 향후 필요 시 manual UI 채움.
if doc.title and doc.ai_domain and doc.ai_summary:
if not doc.ai_confidence:
doc.ai_confidence = 1.0 # frontmatter 는 사람이 작성한 단정값
doc.ai_processed_at = datetime.now(timezone.utc)
doc.ai_model_version = "frontmatter@manual"
await session.commit()
logger.info(f"doc {document_id}: frontmatter 옵션 C → classify/summarize/triage 전부 skip")
return
# 일부만 frontmatter 에 있을 때는 LLM 으로 미설정 필드 보완. 단 _FRONTMATTER_PRESERVED_KEYS
# 는 이미 md_frontmatter 에 있으므로 LLM 이 ai_domain/document_type 등에 영향 못 준다.
logger.info(f"doc {document_id}: frontmatter 부분 인식 → LLM 으로 미설정 필드 보완")
client = AIClient()
try:
# ─── 1. Legacy classify (primary 26B) ───
@@ -344,17 +434,19 @@ async def process(document_id: int, session: AsyncSession) -> None:
if not parsed:
raise ValueError(f"AI 응답에서 JSON 추출 실패: {raw_response[:200]}")
# domain 검증
# domain 검증 (frontmatter 가 이미 채웠으면 LLM 결과 무시)
domain = _validate_domain(parsed.get("domain", ""))
doc.ai_domain = domain
if not doc.ai_domain:
doc.ai_domain = domain
# sub_group은 domain 경로에서 추출 (호환성)
parts = domain.split("/")
doc.ai_sub_group = parts[1] if len(parts) > 1 else ""
# document_type 검증
# document_type 검증 (frontmatter 가 이미 채웠으면 LLM 결과 무시)
doc_type = parsed.get("document_type", "")
doc.document_type = doc_type if doc_type in DOCUMENT_TYPES else "Note"
if not doc.document_type:
doc.document_type = doc_type if doc_type in DOCUMENT_TYPES else "Note"
# confidence
confidence = parsed.get("confidence", 0.5)