실습 6 — 생성 데이터 거르고 나누기

import json, re
import pandas as pd

raw = pd.DataFrame([json.loads(line) for line in open("out_generated_raw.jsonl", encoding="utf-8")])
text = open("care_guide.md", encoding="utf-8").read()
source = {m.group(1): m.group(2).strip() for m in re.finditer(r"## (.+)\n\n(.+)", text)}     # 절 제목 → 본문
count = {"만든 쌍": len(raw)}

# 꼴: 질문은 5글자 이상, 답은 비지 않게
df = raw[(raw["질문"].str.len() >= 5) & (raw["답"].str.len() > 0)]
count["꼴 통과"] = len(df)
# 글자: 한자, 일본 가나, 러시아 문자(키릴)가 섞이면 뺀다
FOREIGN = re.compile(r"[一-鿿぀-ヿЀ-ӿ]")
bad_chars = df[df["질문"].str.contains(FOREIGN) | df["답"].str.contains(FOREIGN)]
df = df.drop(bad_chars.index)
count["글자 통과"] = len(df)
print(count)
for _, r in bad_chars.iterrows():
    print("  글자로 걸림:", r["질문"], "→", r["답"])

def grounded(answer, src):
    """답의 숫자가 모두 원래 절에 있고, 답의 두 글자 묶음이 절반 넘게 원래 절에 있으면 True"""
    nums = re.findall(r"\d[\d,]*", answer)
    if any(n.replace(",", "") not in src.replace(",", "") for n in nums):
        return False
    a, s = answer.replace(" ", ""), src.replace(" ", "")
    pairs = [a[i:i + 2] for i in range(len(a) - 1)]
    return sum(p in s for p in pairs) / max(len(pairs), 1) > 0.5


ok = [grounded(a, source[t]) for a, t in zip(df["답"], df["출처"])]
not_grounded = df[[not x for x in ok]]
df = df[ok]
count["근거 통과"] = len(df)
print(count)
for _, r in not_grounded.head(6).iterrows():
    print(f"  근거로 걸림: [{r['출처']}] {r['질문']} → {r['답']}")

import numpy as np
from sentence_transformers import SentenceTransformer

emb = SentenceTransformer("intfloat/multilingual-e5-small", device="cpu")
Q = emb.encode(["query: " + q for q in df["질문"]], normalize_embeddings=True)
keep = []
for i in range(len(df)):                                   # 앞에서부터 보며, 이미 남긴 질문과 0.92 이상 닮았으면 뺀다
    if not keep or (Q[keep] @ Q[i]).max() < 0.92:
        keep.append(i)
dups = df.iloc[[i for i in range(len(df)) if i not in keep]]
df = df.iloc[keep]
count["중복 통과"] = len(df)
print(count)
for _, r in dups.head(4).iterrows():
    print("  중복으로 걸림:", r["질문"])

test = [json.loads(line) for line in open("care_test.jsonl", encoding="utf-8")]
norm = lambda s: s.replace(" ", "").replace(",", "")
# 시험 질문의 열쇠(정답 낱말)가 학습 답 가운데 하나에라도 있으면 "그 사실을 배울 수 있다"
covered = [t for t in test if any(norm(t["열쇠"]) in norm(a) for a in df["답"])]
print(f"시험 사실이 학습 데이터에 있음: {len(covered)} / {len(test)}")
print("  빠진 사실:", [t["열쇠"] for t in test if t not in covered])
# 누출: 띄어쓰기와 물음표만 다른 같은 질문이 학습 데이터에 있나
train_set = {norm(q).rstrip("?") for q in df["질문"]}
print("시험과 겹친 학습 질문:", [t["질문"] for t in test if norm(t["질문"]).rstrip("?") in train_set])

SYSTEM = "당신은 솔빛생활(가상 쇼핑몰)의 상담원입니다."
with open("out_sft_train.jsonl", "w", encoding="utf-8") as fp:
    for _, r in df.iterrows():
        fp.write(json.dumps({"messages": [{"role": "system", "content": SYSTEM}, {"role": "user", "content": r["질문"]},
                                          {"role": "assistant", "content": r["답"]}], "출처": r["출처"]}, ensure_ascii=False) + "\n")
card = {"원래 문서": "care_guide.md (절 12개)", "생성 모델": "qwen2.5:3b (Ollama)", "온도": 0.3, "씨앗": [42, 7],
        "거르기": count, "시험 묶음": "care_test.jsonl (사람이 쓴 12개)", "시험 사실이 든 수": len(covered)}
json.dump(card, open("out_dataset_card.json", "w", encoding="utf-8"), ensure_ascii=False, indent=1)
print(json.dumps(card, ensure_ascii=False))