실습 1 — 답 만들고 규칙으로 채점하기

from pathlib import Path

# 실습 3, 4와 프로젝트가 불러 쓸 챗봇 모듈. r''' 로 감싸 \n 이 그대로 파일에 들어가게 한다
Path("out_libbot.py").write_text(r'''"""하늘숲 도서관 챗봇(20장). 19장의 day4 문서로 검색하고 Ollama qwen2.5:3b 로 답한다."""
import json
import time
import urllib.request
from pathlib import Path

import numpy as np
from sentence_transformers import SentenceTransformer

MODEL = "qwen2.5:3b"
PROMPTS = {
    "v2": "도서관 안내 도우미입니다. 아래 자료로만 답하세요.",
    "v3": ("도서관 안내 도우미입니다. 아래 자료로만 한두 문장으로 답하세요. 자료에 답이 없으면 지어내지 말고 "
           "'안내 자료에 없습니다. 1층 안내 데스크에 문의해 주세요.'라고 답하세요."),
}
retried = {"수": 0}                              # Ollama 오류로 다시 물은 횟수

emb = SentenceTransformer("intfloat/multilingual-e5-small", device="cpu")
CHUNKS = []
for f in sorted(Path("library/day4").glob("*.txt")):
    paras = [p.strip() for p in f.read_text(encoding="utf-8").split("\n\n") if p.strip()]
    CHUNKS += [f"{f.stem}: {p}" for p in paras[1:]]          # 첫 문단은 제목
VECS = emb.encode(["passage: " + c for c in CHUNKS], normalize_embeddings=True)


def search(question, k=3):
    """(상위 k 조각, 1등 조각의 점수)"""
    q = emb.encode(["query: " + question], normalize_embeddings=True)[0]
    scores = VECS @ q
    top = np.argsort(-scores)[:k]
    return [CHUNKS[i] for i in top], float(scores[top[0]])


def _post(path, body, timeout=120):
    req = urllib.request.Request("http://127.0.0.1:11434" + path, data=json.dumps(body).encode(),
                                 headers={"Content-Type": "application/json"})
    with urllib.request.urlopen(req, timeout=timeout) as r:
        return json.loads(r.read())


def chat(system, user, fmt=None, max_tokens=200):
    """Ollama /api/chat. fmt 에 JSON 스키마를 주면 그 꼴로 답한다. 오류면 모델을 내렸다 네 번까지 다시 묻는다"""
    body = {"model": MODEL, "stream": False, "keep_alive": "30m",
            "messages": [{"role": "system", "content": system}, {"role": "user", "content": user}],
            "options": {"temperature": 0, "seed": 42, "num_predict": max_tokens}}
    if fmt:
        body["format"] = fmt
    for _ in range(4):
        try:
            return _post("/api/chat", body)["message"]["content"]
        except OSError:
            retried["수"] += 1
            try:
                _post("/api/generate", {"model": MODEL, "keep_alive": 0})
            except OSError:
                pass
            time.sleep(3)
    raise RuntimeError("Ollama 가 계속 오류를 냅니다")


def chat_json(system, user, schema, max_tokens=250):
    """JSON 스키마로 답을 받아 사전으로 돌려준다. 끊기거나 깨진 JSON 이 오면 모델을 내렸다 다시 묻는다"""
    for _ in range(4):
        text = chat(system, user, fmt=schema, max_tokens=max_tokens)
        try:
            return json.loads(text)
        except ValueError:
            retried["수"] += 1
            try:
                _post("/api/generate", {"model": MODEL, "keep_alive": 0})
            except OSError:
                pass
            time.sleep(3)
    raise RuntimeError("Ollama 가 계속 깨진 JSON 을 돌려줍니다")


def answer(question, version="v3"):
    """(답, 검색한 조각들)"""
    chunks, _ = search(question)
    text = chat(PROMPTS[version], "자료:\n" + "\n".join(chunks) + "\n\n질문: " + question)
    return text.strip(), chunks
''', encoding="utf-8")

import sys
sys.path.insert(0, ".")                        # 작업 폴더(data)에 만든 out_libbot.py 를 찾게 한다
import out_libbot as bot
# 결과: 조각 수, 그리고 질문 하나의 검색 결과
chunks, top = bot.search("대출 기간은 며칠이에요?")
print(len(bot.CHUNKS), "조각 ·", f"1등 점수 {top:.3f} ·", chunks[0])

import json
import time

import pandas as pd

evalset = [json.loads(line) for line in Path("library_eval.jsonl").read_text(encoding="utf-8").splitlines()]
bot.chat("짧게 답하세요.", "안녕하세요")      # 몸풀기: 모델을 GPU 에 올려 둔다
rows = []
t = time.perf_counter()
for e in evalset:
    for version in ["v2", "v3"]:
        text, chunks = bot.answer(e["질문"], version)
        rows.append({"번호": e["번호"], "종류": e["종류"], "판": version, "질문": e["질문"], "답": text,
                     "자료": "\n".join(chunks)})
answers = pd.DataFrame(rows)
answers.to_csv("out_lib_answers.csv", index=False, encoding="utf-8-sig")
# 결과: 답의 수, 걸린 시간, Ollama 오류로 다시 물은 횟수
print(len(answers), "개 답 ·", f"{time.perf_counter() - t:.0f}초 · 다시 물음 {bot.retried['수']}")
print(answers[answers["번호"].isin([2, 20])][["번호", "판", "답"]].to_string(index=False))

import re

keys = {e["번호"]: e["열쇠"] for e in evalset}


def rule_check(answer, key):
    return {"열쇠": any(k in answer for k in key.split("|")),
            "형식": re.search(r"[A-Za-z]{3,}", answer) is None,     # 영문 낱말(3글자 이상)이 섞였나
            "길이": 0 < len(answer) <= 120}


checks = pd.DataFrame([rule_check(a, keys[n]) for a, n in zip(answers["답"], answers["번호"])])
scored = pd.concat([answers[["번호", "종류", "판"]], checks], axis=1)
scored["규칙"] = scored[["열쇠", "형식", "길이"]].all(axis=1)
# 결과: 판과 종류마다 규칙을 모두 통과한 답의 수
print(scored.groupby(["판", "종류"])[["열쇠", "형식", "길이", "규칙"]].sum())

view = answers.merge(scored[["번호", "판", "열쇠", "형식", "규칙"]], on=["번호", "판"])
# 모름 질문의 답: 열쇠(안내 데스크｜문의)로 매긴 판정과 실제 답
print(view[view["종류"] == "모름"][["번호", "판", "규칙", "답"]].to_string(index=False))
# 형식 규칙에 걸린 답: 엉뚱한 영어가 섞였다(없으면 빈 표)
print(view[~view["형식"]][["번호", "판", "답"]].to_string(index=False))