프로젝트 1 — 아파트 생활 안내 챗봇

import json, re, time, urllib.request
import numpy as np
from sentence_transformers import SentenceTransformer
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
from openai import OpenAI, APIError

docs, kinds = [], []
chapter = ""
for line in open("apt_rules.txt", encoding="utf-8").read().split("\n"):
    if re.match(r"^제\d+장 ", line):
        chapter = line
    elif line.startswith("제") and "조(" in line:
        docs.append(f"[규약] {chapter} {line}")          # 꼬리표 + 장 제목 + 조항
        kinds.append("규약")
for line in open("apt_notices.txt", encoding="utf-8").read().strip().split("\n"):
    docs.append(f"[공지 {line[1:11]}] {line[13:]}")      # [공지 2026-08-20] 제목: 내용
    kinds.append("공지")
tests = [json.loads(l) for l in open("apt_chat_tests.jsonl", encoding="utf-8")]
print("조각:", len(docs), "개 (규약", kinds.count("규약"), "· 공지", kinds.count("공지"), ")")
print(docs[16])
print(docs[27])
print("대화 시험:", len(tests), "개 · 첫 시험:", tests[2])

# 임베딩은 CPU 에, GPU 는 Ollama 에(7차시)
emb = SentenceTransformer("intfloat/multilingual-e5-small", device="cpu")

class Search:
    """본문 "서비스의 짜임"의 하이브리드 검색"""
    def __init__(self, docs):
        self.E = emb.encode(["passage: " + d for d in docs], normalize_embeddings=True)
        self.vec = TfidfVectorizer(analyzer="char_wb", ngram_range=(2, 3))
        self.X = self.vec.fit_transform(docs)
    def top(self, question, k=4, alpha=0.7):
        sem = self.E @ emb.encode("query: " + question, normalize_embeddings=True)
        kw = cosine_similarity(self.vec.transform([question]), self.X)[0]
        mm = lambda a: (a - a.min()) / (a.max() - a.min() + 1e-9)
        return np.argsort(-(alpha * mm(sem) + (1 - alpha) * mm(kw)))[:k].tolist()

client = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama")
MODEL = "qwen2.5:3b"

def unload(model):
    """keep_alive=0 으로 모델을 내린다. 다음 요청 때 새로 올라온다"""
    body = json.dumps({"model": model, "keep_alive": 0}).encode()
    urllib.request.urlopen(urllib.request.Request("http://localhost:11434/api/generate", data=body,
                                                  headers={"Content-Type": "application/json"})).read()

def ask(messages, max_tokens=150):
    """7차시 프로젝트의 ask: 오류가 나면 모델을 내렸다 다시 보낸다"""
    for _ in range(3):
        try:
            resp = client.chat.completions.create(model=MODEL, messages=messages, temperature=0, seed=42, max_tokens=max_tokens)
            return resp.choices[0].message.content.strip()
        except APIError as e:
            error = e.message[:40]
            unload(MODEL)
            time.sleep(2)
    return f"[오류] {error}"

search = Search(docs)
REFUSE = ["찾지 못", "찾을 수 없", "나와 있지 않", "제공되지 않"]
SYSTEM = ("당신은 가람숲 아파트 관리사무소의 안내 직원입니다. 오늘은 2026년 9월 30일입니다. "
          "아래 [자료]의 규약과 공지를 읽고 한두 문장으로 답하세요. "
          "공지가 규약과 다르면 날짜가 늦은 공지를 따르세요. 답 끝에 근거를 (규약 제N조) 또는 (공지 날짜) 꼴로 적으세요. "
          "관련된 자료가 하나도 없을 때만 '자료에서 찾지 못했습니다.'라고 답하세요.")

def answer(turns, query):
    """turns: 사용자의 말 목록. query: 검색에 쓸 글. 검색한 조각 4개를 붙여 마지막 질문에 답한다"""
    found = search.top(query)
    context = "\n".join(docs[i] for i in found)
    msgs = [{"role": "system", "content": SYSTEM}]
    for t in turns[:-1]:                                  # 앞의 대화를 메시지로 넣는다
        msgs += [{"role": "user", "content": t}, {"role": "assistant", "content": "(앞의 답)"}]
    msgs.append({"role": "user", "content": f"[자료]\n{context}\n\n[질문]\n{turns[-1]}"})
    return ask(msgs), [kinds[i] for i in found]

for q in ["재활용품은 무슨 요일에 버려요?", "10월 관리비는 언제까지 내야 해요?"]:
    a, found_kinds = answer([q], q)
    print(q, "→", a.replace("\n", " ")[:90], "| 찾은 조각:", found_kinds)

def rewrite(turns):
    """앞의 대화를 보고 마지막 질문을 혼자서도 통하는 질문으로 바꿔 쓴다"""
    if len(turns) == 1:
        return turns[0]
    history = "\n".join(f"사용자: {t}" for t in turns[:-1])
    msgs = [{"role": "system", "content": "대화 기록을 보고, 마지막 질문을 앞의 대화 없이도 뜻이 통하는 질문 한 문장으로 바꿔 쓰세요. 질문만 쓰세요."},
            {"role": "user", "content": f"[대화 기록]\n{history}\n\n[마지막 질문]\n{turns[-1]}"}]
    return ask(msgs, max_tokens=60)

WAYS = {"마지막 질문만": lambda turns: turns[-1],
        "앞 질문과 이어 붙이기": lambda turns: " ".join(turns),
        "LLM 으로 바꿔 쓰기": rewrite}
results = {}
for name, make_query in WAYS.items():
    start = time.time()
    rows = []
    for t in tests:
        query = make_query(t["대화"])
        a, found_kinds = answer(t["대화"], query)
        if t["정답열쇠"]:
            ok = any(k in a for k in t["정답열쇠"].split("|"))
        else:
            ok = any(w in a for w in REFUSE)                 # 어디에도 없는 질문은 모른다고 해야 맞다
        found_ok = (t["근거"] in found_kinds) if t["근거"] else None
        rows.append({"query": query, "answer": a, "ok": ok, "found": found_ok})
    results[name] = rows
    print(f"{name:12s} 정답 {sum(r['ok'] for r in rows)}/12 · 근거 종류를 찾음 "
          f"{sum(r['found'] is True for r in rows)}/11 · {time.time() - start:.0f}초")

for name, rows in results.items():
    print("■", name)
    for t, r in zip(tests, rows):
        if not r["ok"]:
            print("  질문:", " / ".join(t["대화"]))
            print("  검색:", r["query"][:60], "| 근거 종류를 찾음:", r["found"])
            print("  답  :", r["answer"].replace("\n", " ")[:90])