프로젝트 1 — 문서가 바뀌어도 맞게 답하는 사내 문서 RAG

import csv, hashlib, io, json, re, shutil, time, urllib.request, zipfile
import xml.etree.ElementTree as ET
from html.parser import HTMLParser
from pathlib import Path
import pandas as pd
import sys
sys.path.insert(0, ".")                        # 작업 폴더(data)에 만든 out_offline.py 를 찾게 한다
import out_offline
out_offline.guard()                                       # 모든 작업을 바깥 연결을 막은 채 한다

import chromadb
from chromadb.config import Settings
from openai import OpenAI
from sentence_transformers import SentenceTransformer

# Ollama 모델을 내려 깨끗하게 다시 올린다(실습 6과 같은 까닭)
def unload():
    urllib.request.urlopen(urllib.request.Request("http://localhost:11434/api/generate", headers={"Content-Type": "application/json"},
                                                  data=json.dumps({"model": "qwen2.5:3b", "keep_alive": 0}).encode())).read()
unload()


# ----- 읽기(실습 2와 같은 방법) -----
class HtmlText(HTMLParser):
    SKIP_TAGS, SKIP_CLASSES, VOID = {"script", "style", "title"}, {"gnb", "footer", "info"}, {"meta", "br", "img", "link", "input", "hr"}
    def __init__(self):
        super().__init__(); self.parts, self.stack = [], []
    def handle_starttag(self, tag, attrs):
        if tag in self.VOID: return
        skip = tag in self.SKIP_TAGS or dict(attrs).get("class") in self.SKIP_CLASSES or bool(self.stack and self.stack[-1])
        self.stack.append(skip)
        if not skip and tag in ("h1", "h2"): self.parts.append("\n# " if tag == "h1" else "\n## ")
    def handle_endtag(self, tag):
        if tag in self.VOID: return
        if self.stack: self.stack.pop()
        if tag in ("p", "h1", "h2"): self.parts.append("\n")
    def handle_data(self, data):
        if not (self.stack and self.stack[-1]): self.parts.append(data)


W = "{http://schemas.openxmlformats.org/wordprocessingml/2006/main}"


def read_file(path):
    raw = path.read_bytes()
    if path.suffix == ".docx":                              # 워드: 문단과 표
        root = ET.fromstring(zipfile.ZipFile(io.BytesIO(raw)).read("word/document.xml"))
        text_of = lambda node: "".join(t.text or "" for t in node.iter(W + "t"))
        lines = []
        for el in root.find(W + "body"):
            if el.tag == W + "p":
                st = el.find(f"{W}pPr/{W}pStyle")
                lines.append({"Heading1": "# ", "Heading2": "## "}.get(st.get(W + "val") if st is not None else "", "") + text_of(el))
            elif el.tag == W + "tbl":
                rows = [[text_of(c) for c in r.iter(W + "tc")] for r in el.iter(W + "tr")]
                lines += [", ".join([r[0]] + [f"{h} {v}" for h, v in zip(rows[0][1:], r[1:])]) for r in rows[1:]]
        return "\n".join(lines)
    for enc in ("utf-8-sig", "cp949"):                      # 글: utf-8 먼저, 안 되면 cp949
        try:
            text = raw.decode(enc); break
        except UnicodeDecodeError:
            pass
    if path.suffix == ".html":
        p = HtmlText(); p.feed(text); return "".join(p.parts)
    if path.suffix == ".csv":
        return "\n".join(", ".join(f"{k} {v}" for k, v in r.items()) for r in csv.DictReader(text.splitlines()))
    return text


# ----- 정제(실습 3과 같은 방법) -----
def clean_text(text):
    lines = text.split("\n")
    if any(re.fullmatch(r"- \d+ -", l.strip()) for l in lines):     # PDF 에서 뽑은 글
        header, out = lines[0], []
        for l in lines:
            if l == header or re.fullmatch(r"- \d+ -", l.strip()): continue
            if out and out[-1] and l and not out[-1].endswith("."): out[-1] += l
            else: out.append(l)
        lines = out
    text = "\n".join(lines)
    text = re.sub(r"01[016789]-\d{3,4}-\d{4}", "[전화번호]", text)
    text = re.sub(r"[\w.-]+@[\w-]+\.[\w.]+", "[메일]", text)
    return re.sub(r"\n{3,}", "\n\n", re.sub(r"[ \t]+", " ", text)).strip()


def split_doc(d):
    pieces, head, body = [], "", []
    for line in d["글"].split("\n"):
        if line.startswith("# ") or line.startswith("["): continue
        is_head = line.startswith("## ") or re.match(r"\d+\. ", line)
        is_item = re.match(r"제\d+조", line) or (d["경로"].endswith(".csv") and line.strip())
        if is_head or is_item:
            if "".join(body).strip(): pieces.append((head, body))
            head, body = (line.lstrip("# "), []) if is_head else ("", [line])
        else:
            body.append(line)
    if "".join(body).strip(): pieces.append((head, body))
    return [{"id": f"{d['경로']}#{n}", "글": f"[{d['제목']}] {h}".strip() + "\n" + "\n".join(l for l in b if l.strip()),
             "경로": d["경로"], "제목": d["제목"], "등급": d["등급"], "버전": d["버전"]} for n, (h, b) in enumerate(pieces)]


def load_docs(root):
    """폴더의 문서를 읽어 정제하고, 사본과 옛 판을 거른 문서 목록을 돌려준다"""
    root = Path(root)
    meta = pd.read_csv(root / "_문서목록.csv").set_index("경로")
    docs = []
    for p in sorted(root.rglob("*"), key=lambda p: len(str(p))):      # 이름이 짧은(원래 이름) 것부터
        rel = p.relative_to(root).as_posix()
        if p.is_file() and not p.name.startswith("_") and rel in meta.index:
            m = meta.loc[rel]
            docs.append({"경로": rel, "제목": m["제목"], "등급": int(m["등급"]), "버전": int(m["버전"]),
                         "해시": hashlib.sha256(p.read_bytes()).hexdigest(), "글": clean_text(read_file(p))})
    seen, unique = set(), []
    for d in docs:                                           # 똑같은 파일은 하나만
        if d["해시"] not in seen:
            seen.add(d["해시"]); unique.append(d)
    latest = {}
    for d in unique:                                         # 같은 제목은 최신 판만
        if d["제목"] not in latest or d["버전"] > latest[d["제목"]]["버전"]:
            latest[d["제목"]] = d
    return sorted(latest.values(), key=lambda d: d["경로"])


print([d["경로"] for d in load_docs("intranet")])

LIVE, STATE = Path("out_live"), Path("out_index_state.json")
shutil.rmtree(LIVE, ignore_errors=True)
shutil.copytree("intranet", LIVE)                           # 운영 폴더 = 지금의 사내 문서

emb = SentenceTransformer("intfloat/multilingual-e5-small", device="cpu")
client = chromadb.PersistentClient(path="out_chroma_live", settings=Settings(anonymized_telemetry=False))
if "live" in [c.name for c in client.list_collections()]:
    client.delete_collection("live")
col = client.create_collection("live", metadata={"hnsw:space": "cosine"})
STATE.unlink(missing_ok=True)       # 색인을 새로 만들면 해시 기록도 함께 지운다. 기록만 남으면 "이미 넣었다"고 착각한다


def add_doc(d):
    """문서 하나의 조각을 임베딩해 색인에 넣는다. 돌려주는 것: 조각 수"""
    chunks = split_doc(d)
    vecs = emb.encode(["passage: " + c["글"] for c in chunks], normalize_embeddings=True)
    col.add(ids=[c["id"] for c in chunks], embeddings=vecs.tolist(), documents=[c["글"] for c in chunks],
            metadatas=[{k: c[k] for k in ("경로", "제목", "등급", "버전")} for c in chunks])
    return len(chunks)


def sync():
    """운영 폴더와 해시 기록을 견주어 바뀐 문서만 색인에 반영한다"""
    before = json.loads(STATE.read_text(encoding="utf-8")) if STATE.exists() else {}
    docs = {d["경로"]: d for d in load_docs(LIVE)}
    after = {p: d["해시"] for p, d in docs.items()}
    added = sorted(after.keys() - before.keys())
    deleted = sorted(before.keys() - after.keys())
    changed = sorted(p for p in after.keys() & before.keys() if after[p] != before[p])
    for p in deleted + changed:                              # 옛 조각을 문서 단위로 지운다
        col.delete(where={"경로": p})
    n = sum(add_doc(docs[p]) for p in added + changed)       # 새 조각을 넣는다
    STATE.write_text(json.dumps(after, ensure_ascii=False, indent=1), encoding="utf-8")
    print(f"추가 {added} · 수정 {changed} · 삭제 {deleted} · 넣은 조각 {n}개 · 색인 {col.count()}개")


t = time.time()
sync()                                                       # 기록이 없으니 모든 문서가 "추가"다
print(f"{time.time() - t:.1f}초")

llm = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama")
SYSTEM = ("당신은 누리솔정밀(가상 회사)의 사내 문서 도우미입니다. [자료]에 있는 내용만으로 한두 문장으로 답하고, "
          "끝에 근거 문서의 제목을 (출처: 제목) 꼴로 적으세요. [자료]에 답이 없으면 '찾을 수 없습니다'라고만 답하세요.")


def answer(question, level):
    """등급으로 거른 조각 3개로 답한다. 오류나 깨진 글자면 모델을 내렸다 두 번까지 다시 묻는다"""
    q = emb.encode("query: " + question, normalize_embeddings=True)
    r = col.query(query_embeddings=[q.tolist()], n_results=3, where={"등급": {"$lte": level}})
    messages = [{"role": "system", "content": SYSTEM},
                {"role": "user", "content": "[자료]\n" + "\n\n".join(r["documents"][0]) + f"\n\n[질문]\n{question}"}]
    for _ in range(3):
        try:
            text = llm.chat.completions.create(model="qwen2.5:3b", messages=messages, temperature=0, seed=42).choices[0].message.content.strip()
            if "@" not in text:                 # 깨진 글자(@)가 섞이면 다시 묻는다. 메일 주소는 정제에서 가렸다
                return text
        except Exception as e:
            text = f"(오류: {e})"
        unload()
    return text


tests = [json.loads(line) for line in open("intranet_update_qa.jsonl", encoding="utf-8")]
norm = lambda s: s.replace(" ", "").replace(",", "")
before_answers = {}
for t in tests:
    before_answers[t["질문"]] = answer(t["질문"], 2)        # 정직원(등급 2)으로 묻는다
    print("Q:", t["질문"], "\n   ", before_answers[t["질문"]])

changes = pd.read_csv("intranet_update/_변경목록.csv")
listing = pd.read_csv(LIVE / "_문서목록.csv")
for _, ch in changes.iterrows():
    target = LIVE / ch["경로"]
    if ch["변경"] == "삭제":
        target.unlink()                                      # 파일을 지우고
        listing = listing[listing["경로"] != ch["경로"]]     # 목록에서도 뺀다
    else:                                                    # 추가와 수정: 새 파일을 복사하고 목록 행을 바꾼다
        target.parent.mkdir(parents=True, exist_ok=True)
        shutil.copy2(Path("intranet_update") / ch["경로"], target)
        listing = listing[listing["경로"] != ch["경로"]]
        listing = pd.concat([listing, ch.drop("변경").to_frame().T], ignore_index=True)
listing.to_csv(LIVE / "_문서목록.csv", index=False, encoding="utf-8-sig")
print(changes[["경로", "변경", "버전"]].to_string(index=False))

t = time.time()
sync()                                                       # 해시가 다른 문서만 고친다
print(f"{time.time() - t:.1f}초")

right_before = right_after = 0
for t in tests:
    after_answer = answer(t["질문"], 2)
    ok_b, ok_a = norm(t["열쇠"]) in norm(before_answers[t["질문"]]), norm(t["열쇠"]) in norm(after_answer)
    right_before, right_after = right_before + ok_b, right_after + ok_a
    print("Q:", t["질문"], f"(열쇠: {t['열쇠']})")
    print("   전:", before_answers[t["질문"]], "→", "맞음" if ok_b else "틀림")
    print("   뒤:", after_answer, "→", "맞음" if ok_a else "틀림")
print(f"새 문서 기준 정답: 갱신 전 {right_before}/{len(tests)} → 갱신 뒤 {right_after}/{len(tests)}")
print("막은 주소:", out_offline.BLOCKED)