실습 3 — 색인 갱신 작업 자동화

import hashlib
import json
import os
import shutil
import sys
from pathlib import Path

import numpy as np
import pandas as pd
from sentence_transformers import SentenceTransformer

sys.path.insert(0, ".")                        # 실습 1이 만든 out_flow.py 를 찾게 한다
from out_flow import Stop, run_flow

shutil.rmtree("out_index", ignore_errors=True)   # 다시 돌려도 같은 결과가 나오게 지난 색인을 지운다
ROOT = Path("out_index")
ROOT.mkdir()
emb = SentenceTransformer("intfloat/multilingual-e5-small", device="cpu")


# 실습 2와 같은 함수들
def fingerprint(text):
    return hashlib.sha256(text.encode("utf-8")).hexdigest()[:12]


def snapshot(day):
    return {f.stem: f.read_text(encoding="utf-8") for f in sorted(Path(f"library/day{day}").glob("*.txt"))}


def split_chunks(docs):
    out = []
    for name, text in docs.items():
        paras = [p.strip() for p in text.split("\n\n") if p.strip()]
        out += [f"{name}: {p}" for p in paras[1:]]
    return out


def current():
    """지금 판의 정보(meta.json). 아직 판이 없으면 None"""
    pointer = ROOT / "CURRENT"
    if not pointer.exists():
        return None
    version = pointer.read_text(encoding="utf-8")
    return json.loads((ROOT / version / "meta.json").read_text(encoding="utf-8"))


print("준비 끝 · 지금 판:", current())

CACHE = ROOT / "embed_cache.npz"               # 조각 지문 → 벡터. 판이 바뀌어도 남겨 다시 쓴다


def make_tasks(day):
    """그 밤의 갱신 작업 DAG. 작업마다 ctx 를 받아 결과를 돌려준다"""
    def read(ctx):
        return snapshot(day)

    def check(ctx):
        docs = ctx["읽기"]
        bad = [n for n, t in docs.items() if not t.strip() or t.count("�") / len(t) > 0.01]
        if bad:
            raise Stop(f"망가진 문서 {bad}")
        return len(docs)

    def chunk(ctx):
        return split_chunks(ctx["읽기"])

    def embed(ctx):
        chunks = ctx["조각"]
        cache = {}
        if CACHE.exists():
            with np.load(CACHE) as z:          # with: 다 읽으면 파일을 닫는다
                cache = {k: z[k] for k in z.files}
        keys = [fingerprint(c) for c in chunks]
        todo = [c for c, k in zip(chunks, keys) if k not in cache]
        if todo:                               # 캐시에 없는 조각만 임베딩한다(증분)
            vecs = emb.encode(["passage: " + c for c in todo], normalize_embeddings=True)
            cache.update({fingerprint(c): v for c, v in zip(todo, vecs)})
            np.savez(CACHE, **cache)
        return {"벡터": np.stack([cache[k] for k in keys]), "새로": len(todo), "다시 씀": len(chunks) - len(todo)}

    def save(ctx):
        chunks = ctx["조각"]
        # 판의 지문: 조각 지문을 정렬해 이은 것의 지문. 내용이 같으면 같은 지문이 된다
        content = fingerprint("\n".join(sorted(fingerprint(c) for c in chunks)))
        now = current()
        if now and now["내용 지문"] == content:
            raise Stop("바뀐 것이 없어 새 판을 만들지 않습니다")
        version = f"v{len(list(ROOT.glob('v*'))) + 1:03d}"
        folder = ROOT / version
        folder.mkdir()
        np.save(folder / "vectors.npy", ctx["임베딩"]["벡터"])
        (folder / "chunks.json").write_text(json.dumps(chunks, ensure_ascii=False), encoding="utf-8")
        meta = {"판": version, "밤": day, "조각": len(chunks), "내용 지문": content}
        (folder / "meta.json").write_text(json.dumps(meta, ensure_ascii=False), encoding="utf-8")
        return version

    def swap(ctx):
        tmp = ROOT / "CURRENT.tmp"
        tmp.write_text(ctx["저장"], encoding="utf-8")
        os.replace(tmp, ROOT / "CURRENT")      # 가리킴을 한 번에 바꾼다
        return ctx["저장"]

    return {"읽기": (read, []), "검사": (check, ["읽기"]), "조각": (chunk, ["검사"]),
            "임베딩": (embed, ["조각"]), "저장": (save, ["임베딩"]), "교체": (swap, ["저장"])}


def refresh(day):
    """한 밤의 갱신을 돌리고 한 줄로 요약한다. 작업마다의 로그는 끈다"""
    records, ctx = run_flow(make_tasks(day), log=lambda m: None)
    state = {r["작업"]: r for r in records}
    e = ctx.get("임베딩", {})
    return {"밤": day, "검사": state["검사"]["상태"], "조각": len(ctx.get("조각", [])), "새로 임베딩": e.get("새로", 0),
            "다시 씀": e.get("다시 씀", 0), "저장": state["저장"]["상태"], "메모": state["저장"]["메모"] or state["검사"]["메모"],
            "지금 판": (current() or {}).get("판")}


# 결과: 함수가 만든 DAG 의 작업 이름
print(list(make_tasks(0)))

rows = [refresh(0), refresh(1)]
# 결과: 첫 밤은 모든 조각을, 다음 밤은 바뀐 조각만 임베딩한다
print(pd.DataFrame(rows).to_string(index=False))
print("판 폴더:", sorted(p.name for p in ROOT.glob("v*")))

# 같은 밤(day1)을 한 번 더 돌린다. 서버가 멈춰 다시 돌린 상황이다
print(pd.DataFrame([refresh(1)]).to_string(index=False))
print("판 폴더:", sorted(p.name for p in ROOT.glob("v*")))

# 서비스 쪽: 가리킴이 가리키는 판을 읽어 검색한다
version = (ROOT / "CURRENT").read_text(encoding="utf-8")
vectors = np.load(ROOT / version / "vectors.npy")
chunks = json.loads((ROOT / version / "chunks.json").read_text(encoding="utf-8"))
q = emb.encode(["query: 한 번에 최대 몇 권까지 대출돼요?"], normalize_embeddings=True)
best = int(np.argmax(vectors @ q[0]))          # 정규화한 벡터의 곱 = 코사인 유사도
# 결과: 지금 판과 가장 가까운 조각
print(version, "→", chunks[best])