실습 5 — Ollama 로 같은 일 하기

from openai import OpenAI

# Ollama 의 OpenAI 호환 주소. api_key 는 검사하지 않으므로 아무 글자나 넣는다
client = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama")
# 내려받은 모델 이름 목록. ollama list 와 같다
print(sorted(m.id for m in client.models.list()))

import time

def ask(messages, model="qwen2.5:0.5b"):
    """Ollama 에 메시지를 보내고 답 글을 돌려준다. temperature=0, seed=42 로 다시 물어도 같은 답이 나오게 한다"""
    resp = client.chat.completions.create(model=model, messages=messages, temperature=0, seed=42, max_tokens=120)
    return resp.choices[0].message.content.strip()

# 첫 요청은 모델을 GPU 에 올리느라 오래 걸린다. 한 번 불러 둔 뒤에 시간을 잰다
ask([{"role": "user", "content": "안녕하세요"}])
start = time.time()
print(ask([{"role": "user", "content": "휴대폰으로 결제하면 한도가 있나요?"}]))
print(f"걸린 시간 {time.time() - start:.1f}초")

import json
from collections import Counter
from openai import APIError
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity

# 실습 4와 같은 준비: 검색, 프롬프트, 채점(모델은 불러오지 않는다)
text = open("shop_policy.txt", encoding="utf-8").read()
chunks = [c.strip() for c in text.split("\n\n")[1:]]
qa = [json.loads(line) for line in open("shop_policy_qa.jsonl", encoding="utf-8")]
vec = TfidfVectorizer(analyzer="char_wb", ngram_range=(2, 3))
X = vec.fit_transform(chunks)
def search(question, k=3):
    scores = cosine_similarity(vec.transform([question]), X)[0]
    return [(int(i), float(scores[i])) for i in scores.argsort()[::-1][:k]]
SYSTEM = ("당신은 쇼핑몰 상담원입니다. 아래 [문서]에 적힌 내용만 근거로 짧게 답하세요. "
          "문서에 답이 없으면 '문서에서 찾지 못했습니다.'라고만 답하세요.")
def build_messages(question, k=3):
    found = [chunks[i] for i, _ in search(question, k)]
    return [{"role": "system", "content": SYSTEM},
            {"role": "user", "content": "[문서]\n" + "\n".join(found) + "\n\n[질문]\n" + question}]
REFUSE = ["찾지 못", "찾을 수 없", "나와 있지 않", "제공되지 않"]
def grade(key, answer):
    refused = any(w in answer for w in REFUSE)
    if key == "":
        return "모름 맞음" if refused else "지어냄"
    if any(k in answer for k in key.split("|")):
        return "정답"
    return "거절" if refused else "오답"

def safe_ask(messages, model="qwen2.5:0.5b"):
    """서버가 오류를 돌려주면 멈추지 않고 '[오류]' 로 적어 둔다"""
    try:
        return ask(messages, model)
    except APIError as e:
        return f"[오류] {e.message[:40]}"

start = time.time()
answers = [safe_ask(build_messages(r["질문"])) for r in qa]
results = [grade(r["정답열쇠"], a) for r, a in zip(qa, answers)]
print(f"30문항 {time.time() - start:.0f}초")
print("문서에 있는 질문:", Counter(x for x, r in zip(results, qa) if r["근거조항"]))
print("문서에 없는 질문:", Counter(x for x, r in zip(results, qa) if not r["근거조항"]))
print("오류가 난 답:", sum(a.startswith("[오류]") for a in answers), "개 ·", next((a for a in answers if a.startswith("[오류]")), "없음"))