프로젝트 1 — 매일 밤 도는 RAG 갱신

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

import matplotlib.pyplot as plt
import mlflow
import numpy as np
import pandas as pd
from mlflow import MlflowClient
from sentence_transformers import SentenceTransformer

sys.path.insert(0, ".")
from out_flow import Stop, run_flow

plt.rcParams["font.family"] = "Malgun Gothic"        # 한글 글꼴(5단계의 그래프)
plt.rcParams["axes.unicode_minus"] = False

shutil.rmtree("out_rag_index", ignore_errors=True)   # 다시 돌려도 같게: 지난 색인을 지운다
ROOT = Path("out_rag_index")
ROOT.mkdir()
CACHE = ROOT / "embed_cache.npz"
emb = SentenceTransformer("intfloat/multilingual-e5-small", device="cpu")
QA = [json.loads(line) for line in Path("library_qa.jsonl").read_text(encoding="utf-8").splitlines()]

# 밤마다 실행 하나를 남길 실험. 다시 돌릴 때를 생각해 지난 실행은 지운다
mlflow.set_tracking_uri("sqlite:///out_mlflow.db")
if mlflow.get_experiment_by_name("도서관-색인") is None:
    mlflow.create_experiment("도서관-색인", artifact_location=Path("out_mlruns").resolve().as_uri())
mlflow.set_experiment("도서관-색인")
client = MlflowClient()
for r in mlflow.search_runs(experiment_names=["도서관-색인"]).itertuples():
    client.delete_run(r.run_id)


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():
    """지금 판의 (정보, 조각, 벡터). 판이 없으면 None"""
    if not (ROOT / "CURRENT").exists():
        return None
    folder = ROOT / (ROOT / "CURRENT").read_text(encoding="utf-8")
    return (json.loads((folder / "meta.json").read_text(encoding="utf-8")),
            json.loads((folder / "chunks.json").read_text(encoding="utf-8")), np.load(folder / "vectors.npy"))


def hit_at_3(questions, q_vecs, chunks, vectors):
    """근거가 든 조각이 상위 3개 안에 든 질문의 비율, 그리고 놓친 질문"""
    top = np.argsort(-(q_vecs @ vectors.T), axis=1)[:, :3]
    hits = [any(q["근거"] in chunks[j] for j in top[i]) for i, q in enumerate(questions)]
    return float(np.mean(hits)), [q["질문"] for q, h in zip(questions, hits) if not h]


print("질문", len(QA), "줄 · 지금 판:", current())

def nightly(night):
    """밤 하나의 갱신 작업. (작업 기록, ctx)"""
    def read(ctx):
        return snapshot(night)

    def check(ctx):
        docs, now = ctx["읽기"], current()
        problems = []
        if now and len(docs) < now[0]["문서"] * 0.8:
            problems.append(f"문서 수 {now[0]['문서']}→{len(docs)}")
        empty = [n for n, t in docs.items() if not t.strip()]
        broken = [n for n, t in docs.items() if t and t.count("�") / len(t) > 0.01]
        if empty:
            problems.append(f"빈 파일 {empty}")
        if broken:
            problems.append(f"깨진 글자 {broken}")
        if problems:
            raise Stop("; ".join(problems))
        return len(docs)

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

    def embed(ctx):
        chunks, cache = ctx["조각"], {}
        if CACHE.exists():
            with np.load(CACHE) as z:
                cache = {k: z[k] for k in z.files}
        todo = [c for c in chunks if fingerprint(c) 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[fingerprint(c)] for c in chunks]), "새로": len(todo)}

    def evaluate(ctx):
        qs = [q for q in QA if q["부터"] <= night <= q["까지"]]       # 그 밤에 맞는 질문만
        q_vecs = emb.encode(["query: " + q["질문"] for q in qs], normalize_embeddings=True)
        cand, missed = hit_at_3(qs, q_vecs, ctx["조각"], ctx["임베딩"]["벡터"])
        now = current()
        live = hit_at_3(qs, q_vecs, now[1], now[2])[0] if now else None   # 지금 판을 같은 질문으로
        return {"질문": len(qs), "후보": cand, "지금 판": live, "놓친 질문": missed}

    def gate(ctx):
        r = ctx["평가"]
        if r["후보"] < 0.9:
            raise Stop(f"hit@3 {r['후보']:.3f} < 0.9")
        if r["지금 판"] is not None and r["후보"] < r["지금 판"] - 0.05:
            raise Stop(f"지금 판 {r['지금 판']:.3f} 보다 {r['지금 판'] - r['후보']:.3f} 떨어짐")
        return "통과"

    def save(ctx):
        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(ctx["조각"], ensure_ascii=False), encoding="utf-8")
        meta = {"판": version, "밤": night, "문서": ctx["검사"], "조각": len(ctx["조각"])}
        (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["저장"]

    tasks = {"읽기": (read, []), "검사": (check, ["읽기"]), "조각": (chunk, ["검사"]),
             "임베딩": (embed, ["조각"]), "평가": (evaluate, ["임베딩"]), "게이트": (gate, ["평가"]),
             "저장": (save, ["게이트"]), "교체": (swap, ["저장"])}
    return run_flow(tasks, retries=1, log=lambda m: None)


print("밤 작업 준비 끝: 읽기 → 검사 → 조각 → 임베딩 → 평가 → 게이트 → 저장 → 교체")

log = []
for night in range(5):
    records, ctx = nightly(night)
    state = {r["작업"]: r for r in records}
    ev = ctx.get("평가", {})
    stopped = next((r for r in records if r["상태"] == "멈춤"), None)
    row = {"밤": night, "검사": state["검사"]["상태"], "새로 임베딩": ctx.get("임베딩", {}).get("새로"),
           "질문": ev.get("질문"), "후보 hit@3": ev.get("후보"), "지금 판 hit@3": ev.get("지금 판"),
           "게이트": state["게이트"]["상태"], "지금 판": current()[0]["판"] if current() else None,
           "막은 까닭": stopped["메모"] if stopped else ""}
    log.append(dict(row, 놓친질문=ev.get("놓친 질문", [])))
    # MLflow: 밤 하나 = 실행 하나. 작업마다의 상태는 파라미터, 점수는 지표로 남긴다
    with mlflow.start_run(run_name=f"night-{night}"):
        mlflow.log_params({f"task_{r['작업']}": r["상태"] for r in records})
        mlflow.log_params({"night": night, "live_version": row["지금 판"]})
        mlflow.log_metrics({k: v for k, v in {"candidate_hit3": ev.get("후보"), "live_hit3": ev.get("지금 판"),
                                              "new_chunks": row["새로 임베딩"]}.items() if v is not None})
table = pd.DataFrame(log).drop(columns="놓친질문")
table[["새로 임베딩", "질문"]] = table[["새로 임베딩", "질문"]].astype("Int64")   # 빈칸(검사에서 멈춘 밤)이 있어도 정수로 보인다
# 결과: 다섯 밤의 운영 기록
print(table.round(3).to_string(index=False))

for row in log:
    if row["게이트"] != "성공":
        print(f"밤 {row['밤']}: 검사 {row['검사']}, 게이트 {row['게이트']} · {row['막은 까닭']}")
        for q in row["놓친질문"]:
            print("   놓친 질문:", q)

# 밤 2에 덮인 문서를 밤 1과 문단 단위로 견준다
d1, d2 = snapshot(1), snapshot(2)
for name in d1:
    if d1[name] != d2.get(name):
        p1 = {x.strip() for x in d1[name].split("\n\n")}
        p2 = {x.strip() for x in d2[name].split("\n\n")}
        print(f"{name}: 사라진 문단 {sorted(p1 - p2)} · 되살아난 문단 {sorted(p2 - p1)}")
print("판 폴더:", sorted(p.name for p in ROOT.glob("v*")), "· CURRENT →", (ROOT / "CURRENT").read_text(encoding="utf-8"))

# MLflow 에 남은 밤마다의 실행을 표로 뽑는다(웹 화면 없이 코드로)
runs = mlflow.search_runs(experiment_names=["도서관-색인"], order_by=["params.night ASC"])
print(runs[["tags.mlflow.runName", "params.task_검사", "params.task_게이트", "params.live_version",
            "metrics.candidate_hit3"]].round(3).to_string(index=False))

# 밤마다 후보의 hit@3 과 게이트의 바닥선
fig, ax = plt.subplots(figsize=(7, 3.5))
cand = table["후보 hit@3"].astype(float)
ax.plot(table["밤"], cand, marker="o", label="후보 hit@3")
ax.axhline(0.9, linestyle="--", color="gray", label="바닥 0.9")
for _, r in table.iterrows():
    ax.annotate(f"{r['게이트']}\n{r['지금 판']}", (r["밤"], 0.55), ha="center", fontsize=9)
ax.set_ylim(0.5, 1.05)
ax.set_xticks(table["밤"])
ax.set_xlabel("밤")
ax.set_ylabel("hit@3")
ax.set_title("밤마다의 색인 후보 점수와 게이트")
ax.legend(loc="center left")
plt.tight_layout()
plt.show()