시놀로지 Drive × n8n 연동: 실시간 증분 임베딩 자동화 (5/5)

pip install watchdog chromadb langchain-community \
            langchain-text-splitters sentence-transformers \
            python-frontmatter hashlib schedule

import os
import hashlib
import json
import time
from threading import Timer
import frontmatter
import chromadb
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
from langchain_text_splitters import RecursiveCharacterTextSplitter
from sentence_transformers import SentenceTransformer

# ── 설정 ──────────────────────────────────────────────
WATCH_DIR   = os.getenv("WATCH_DIR", "/volume1/AI_Knowledge_Base")   # 환경변수 우선 적용
CHROMA_HOST = os.getenv("CHROMA_HOST", "localhost")
CHROMA_PORT = int(os.getenv("CHROMA_PORT", 8000))
COLLECTION  = "synology_notes"
HASH_DB     = "/app/data/file_hashes.json"   # 파일 해시 기록 DB
EMBED_MODEL = "BAAI/bge-m3"                  # 한국어 최적 임베딩 모델

# ── 초기화 ────────────────────────────────────────────
client     = chromadb.HttpClient(host=CHROMA_HOST, port=CHROMA_PORT)
collection = client.get_or_create_collection(
    name=COLLECTION,
    metadata={"hnsw:space": "cosine"}
)
embedder   = SentenceTransformer(EMBED_MODEL)
splitter   = RecursiveCharacterTextSplitter(
    chunk_size=1000, chunk_overlap=150
)

def load_hash_db():
    if os.path.exists(HASH_DB):
        try:
            with open(HASH_DB, "r") as f:
                return json.load(f)
        except Exception:
            return {}
    return {}

def save_hash_db(db):
    os.makedirs(os.path.dirname(HASH_DB), exist_ok=True)
    with open(HASH_DB, "w") as f:
        json.dump(db, f, indent=2)

def file_sha256(path):
    if not os.path.exists(path):
        return ""
    h = hashlib.sha256()
    try:
        with open(path, "rb") as f:
            while chunk := f.read(8192):
                h.update(chunk)
        return h.hexdigest()
    except Exception:
        return ""

def upsert_file(path, hash_db):
    # 마크다운 파일을 청킹 → 임베딩 → ChromaDB 업서트 (안전성 강화)
    if not os.path.exists(path):
        return
    try:
        post = frontmatter.load(path)
        content  = post.content
        metadata = dict(post.metadata)
        
        # 메타데이터 값 타입 방어 (ChromaDB 제약조건에 맞춰 str/int/float/bool로 가공)
        metadata = {k: (str(v) if isinstance(v, (dict, list)) else v) for k, v in metadata.items()}
        metadata["source"] = path

        chunks = splitter.split_text(content)
        if not chunks:
            return

        new_hash = file_sha256(path)

        # 기존 벡터 삭제 (재임베딩 시 중복 인덱스 방지)
        existing = collection.get(where={"source": path})
        if existing["ids"]:
            collection.delete(ids=existing["ids"])

        # 새 벡터 생성 및 추가
        vectors = embedder.encode(chunks).tolist()
        ids     = [f"{hashlib.md5(path.encode()).hexdigest()}::chunk::{i}" for i in range(len(chunks))]
        metas   = [metadata] * len(chunks)

        collection.add(
            ids=ids,
            embeddings=vectors,
            documents=chunks,
            metadatas=metas
        )

        hash_db[path] = new_hash
        save_hash_db(hash_db)
        print(f"✅ 업서트 완료: {os.path.basename(path)} ({len(chunks)} chunks)")

    except Exception as e:
        print(f"❌ 오류 [{path}]: {e}")

def delete_file(path, hash_db):
    # 삭제된 파일의 벡터 제거
    try:
        existing = collection.get(where={"source": path})
        if existing["ids"]:
            collection.delete(ids=existing["ids"])
        hash_db.pop(path, None)
        save_hash_db(hash_db)
        print(f"🗑️  벡터 삭제: {os.path.basename(path)}")
    except Exception as e:
        print(f"❌ 삭제 오류 [{path}]: {e}")

# ── 디바운싱 지원 Watchdog 이벤트 핸들러 ────────────────────────────
class DebouncedNoteHandler(FileSystemEventHandler):
    def __init__(self):
        self.hash_db = load_hash_db()
        self.timers = {}

    def _should_process(self, path):
        return path.endswith(".md") and not os.path.basename(path).startswith(".")

    def on_modified(self, event):
        if event.is_directory or not self._should_process(event.src_path):
            return
        path = event.src_path
        
        # 연속 수정 파일 쓰기 스트림 병목 감쇄 (디바운싱 1.5초)
        if path in self.timers:
            self.timers[path].cancel()
        
        t = Timer(1.5, self._debounced_upsert, [path])
        self.timers[path] = t
        t.start()

    def _debounced_upsert(self, path):
        self.timers.pop(path, None)
        new_hash = file_sha256(path)
        if self.hash_db.get(path) != new_hash:
            print(f"🔄 변경 감지: {os.path.basename(path)}")
            upsert_file(path, self.hash_db)

    def on_created(self, event):
        if not event.is_directory and self._should_process(event.src_path):
            print(f"➕ 신규 파일: {os.path.basename(event.src_path)}")
            upsert_file(event.src_path, self.hash_db)

    def on_deleted(self, event):
        if not event.is_directory and self._should_process(event.src_path):
            print(f"🗑️  파일 삭제: {os.path.basename(event.src_path)}")
            delete_file(event.src_path, self.hash_db)

# ── 메인 실행 ─────────────────────────────────────────
if __name__ == "__main__":
    print(f"🚀 시놀로지 RAG Watcher 시작: {WATCH_DIR}")
    handler  = DebouncedNoteHandler()
    observer = Observer()
    observer.schedule(handler, WATCH_DIR, recursive=True)
    observer.start()
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

FROM python:3.11-slim

WORKDIR /app
COPY synology_rag_watcher.py .

RUN pip install --no-cache-dir watchdog chromadb \
    langchain-text-splitters sentence-transformers \
    python-frontmatter

# bge-m3 모델 사전 다운로드 (이미지 빌드 시 포함)
RUN python -c "from sentence_transformers import SentenceTransformer; SentenceTransformer('BAAI/bge-m3')"

CMD ["python", "synology_rag_watcher.py"]

  rag-watcher:
    build:
      context: .
      dockerfile: Dockerfile.watcher
    container_name: rag_watcher
    volumes:
      - /volume1/AI_Knowledge_Base:/volume1/AI_Knowledge_Base:ro  # 읽기 전용 마운트
      - rag_watcher_data:/app/data                                 # 해시 DB 영속화
    environment:
      - WATCH_DIR=/volume1/AI_Knowledge_Base
      - CHROMA_HOST=chromadb
    depends_on:
      - chromadb
    restart: unless-stopped

volumes:
  rag_watcher_data:

  n8n:
    image: n8nio/n8n:latest
    container_name: n8n_automation
    ports:
      - "5678:5678"
    environment:
      - N8N_BASIC_AUTH_ACTIVE=true
      - N8N_BASIC_AUTH_USER=admin
      - N8N_BASIC_AUTH_PASSWORD=your_secure_password
      - GENERIC_TIMEZONE=Asia/Seoul
      - N8N_HOST=0.0.0.0
    volumes:
      - n8n_data:/home/node/.n8n
      - /volume1/AI_Knowledge_Base:/data/knowledge:ro
    restart: unless-stopped

volumes:
  n8n_data:

# n8n HTTP Request 노드 설정 (컨테이너 매핑 경로 적용)
Method: POST
URL: http://rag-watcher:8080/trigger   # 가상 Watcher 엔드포인트
Body (JSON):
{
  "action": "scan_and_embed",
  "target_folder": "/data/knowledge",    # n8n 마운트 컨테이너 내부 경로 기준 수렴
  "force_reindex": false
}

# 응답 처리 데이터 수집 후 → Slack 알림 노드 매핑
Slack Message:
"✅ **RAG 지식 인덱스 동기화 완료**\n
 - 변경 감지 파일: {{ $json.processed_files_count }} 개\n
 - 새로 생성된 청크: {{ $json.new_chunks_count }} 개\n
 - 파이프라인 소요 시간: {{ $json.elapsed_time }}초"

import chromadb
from sentence_transformers import SentenceTransformer

client     = chromadb.HttpClient(host="localhost", port=8000)
col        = client.get_collection("synology_notes")
embedder   = SentenceTransformer("BAAI/bge-m3")

# 테스트 쿼리 목록
test_queries = [
    "2024년 KOSPI 하락기 대응 전략",
    "Docker 네트워크 트러블슈팅 방법",
    "가족 예방접종 스케줄",
    "기술 블로그 글쓰기 원칙",
]

for query in test_queries:
    vec = embedder.encode([query]).tolist()
    results = col.query(
        query_embeddings=vec,
        n_results=3,
        include=["documents", "metadatas", "distances"]
    )
    print(f"\n🔍 질문: {query}")
    for doc, meta, dist in zip(
        results["documents"][0],
        results["metadatas"][0],
        results["distances"][0]
    ):
        title    = meta.get("title", "제목 없음")
        score    = round((1 - dist) * 100, 1)   # 코사인 유사도 → 점수 변환
        print(f"  [{score}%] {title} — {doc[:60]}...")