실습 6 — 실패에 강한 호출

import json, urllib.request
# 이 PC 의 Ollama 는 오래 쓰다 보면 가끔 깨진 글자(@@@@)만 내는 일이 있었다. 실습을 시작할 때 모델을 내려 깨끗하게 다시 올린다
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()
import json, time
import pandas as pd
from openai import OpenAI, APIError

client = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama")
MODEL = "qwen2.5:3b"
test = pd.read_csv("inquiries.csv").iloc[20:]
LABELS = ("배송", "교환반품", "결제", "회원", "기타")
ASK = ('고객 문의를 분류해 {"분류": 분류 이름, "이유": 짧은 이유} 꼴의 JSON 으로만 답하세요. '
       '분류 이름은 배송, 교환반품, 결제, 회원, 기타 가운데 하나입니다.')

def check(text):
    """본문 '퓨샷과 출력 꼴 고정'의 검증: (결과 또는 None, 까닭)"""
    try:
        obj = json.loads(text)
    except (json.JSONDecodeError, TypeError):
        return None, "JSON 이 아닙니다"
    if obj.get("분류") not in LABELS:
        return None, f"분류 '{obj.get('분류')}' 는 목록에 없습니다. {', '.join(LABELS)} 가운데 하나만 쓰세요"
    return obj, "통과"

def call_llm(messages, json_mode, tries=3):
    """Ollama 일시 오류에는 0.5초, 1초로 늘리며 다시 보낸다(백오프)"""
    extra = {"response_format": {"type": "json_object"}} if json_mode else {}
    for k in range(tries):
        try:
            return client.chat.completions.create(model=MODEL, messages=messages, temperature=0, seed=42,
                                                  max_tokens=80, **extra).choices[0].message.content
        except APIError:
            if k == tries - 1:
                raise
            time.sleep(0.5 * 2 ** k)

def classify(q, json_mode=False, max_fix=0):
    """max_fix: 검증에 실패하면 까닭을 붙여 다시 묻는 횟수"""
    messages = [{"role": "system", "content": ASK}, {"role": "user", "content": q}]
    for attempt in range(max_fix + 1):
        text = call_llm(messages, json_mode)
        obj, why = check(text)
        if obj:
            return obj, attempt
        messages += [{"role": "assistant", "content": text},
                     {"role": "user", "content": f"형식이 틀렸습니다: {why}. JSON 하나만 다시 쓰세요."}]
    return None, max_fix

print(classify("카드 결제가 두 번 됐어요."))

rows = []
for name, json_mode, max_fix in [("지시만", False, 0), ("지시 + 다시 묻기 2번", False, 2), ("JSON 모드 + 다시 묻기 2번", True, 2)]:
    start, valid, right, fixed = time.time(), 0, 0, 0
    for q, y in zip(test["문의"], test["분류"]):
        obj, attempt = classify(q, json_mode, max_fix)
        valid += obj is not None
        right += bool(obj) and obj["분류"] == y
        fixed += bool(obj) and attempt > 0          # 다시 물어서 통과한 수
    rows.append({"설정": name, "형식 통과(/20)": valid, "정답(/20)": right, "다시 물어 고친 수": fixed,
                 "초": round(time.time() - start)})
print(pd.DataFrame(rows).to_string(index=False))