검색 전후 최적화

!pip install -q openai

import json
from pathlib import Path
import numpy as np
from openai import OpenAI
from mindmall_guide import guide

client = OpenAI(
    base_url="https://openrouter.ai/api/v1",
    api_key=Path("openrouter-key.txt").read_text().strip(),
)
model = "openai/gpt-4.1-mini"
embedding_model = "openai/text-embedding-3-small"

def chunk_with_heading(text):
    chunks = []
    for section in text.split("\n## ")[1:]:
        title, body = section.split("\n", 1)
        chunks += [f"[{title}] {p.strip()}" for p in body.split("\n\n") if p.strip()]
    return chunks

def embed(texts):
    response = client.embeddings.create(model=embedding_model, input=texts)
    vectors = np.array([d.embedding for d in response.data])
    return vectors / np.linalg.norm(vectors, axis=1, keepdims=True)

def search(query, k=3):  # 가까운 청크 k개의 번호
    scores = chunk_vectors @ embed([query])[0]
    return [int(i) for i in np.argsort(-scores)[:k]]

chunks = chunk_with_heading(guide)
chunk_vectors = embed(chunks)
len(chunks)

eval_set = [
    ("포인트 언제 들어와요?", 3),
    ("택배 언제 와요?", 6),
    ("키보드를 물에 빠뜨렸는데 공짜로 고쳐 주나요?", 18),
    ("불량품이면 반품 기한이 더 길어요?", 10),
    ("두 개 샀다가 하나만 돌려보내면 돈을 덜 돌려받나요?", 15),
    ("주말에 문의하면 언제 답을 받나요?", 21),
    ("키캡 뺐는데 환불 돼요?", 12),
    ("같은 고장이 계속 나면 새 걸로 바꿔 주나요?", 19),
    ("등급은 언제 바뀌어요?", 0),
    ("결제할 때 포인트를 얼마까지 쓸 수 있어요?", 4),
]

def evaluate(search_fn):
    hits = [(gold, search_fn(question)) for question, gold in eval_set]
    return {
        "hit@1": float(np.mean([gold == found[0] for gold, found in hits])),
        "hit@3": float(np.mean([gold in found[:3] for gold, found in hits])),
        "실패한 질문": [q for (q, _), (gold, found) in zip(eval_set, hits) if gold not in found[:3]],
    }

baseline = evaluate(search)
baseline

def hyde(question):
    prompt = f"""온라인 쇼핑몰 고객 안내서에서 다음 질문에 답하는 문단을 한두 문장으로 지어내세요.
안내서 문체(~한다)로 쓰고 문단만 답하세요.

질문: {question}"""
    response = client.chat.completions.create(model=model, messages=[{"role": "user", "content": prompt}])
    return response.choices[0].message.content.strip()

hyde("택배 언제 와요?")

hyde_result = evaluate(lambda q: search(hyde(q)))
hyde_result

score_schema = {"type": "json_schema", "json_schema": {"name": "relevance", "strict": True, "schema": {
    "type": "object",
    "properties": {"scores": {"type": "array", "items": {"type": "integer"}}},  # 후보 순서대로 0~10점
    "required": ["scores"], "additionalProperties": False}}}

def rerank(question, candidates, k=3):
    listing = "\n".join(f"{n}. {chunks[i]}" for n, i in enumerate(candidates, start=1))
    prompt = f"""질문에 답하는 데 각 자료가 얼마나 필요한지 0~10점으로 매기세요.
자료 순서대로 점수 목록을 답하세요.

[질문]
{question}

[자료]
{listing}"""
    response = client.chat.completions.create(
        model=model, messages=[{"role": "user", "content": prompt}], response_format=score_schema)
    scores = json.loads(response.choices[0].message.content)["scores"]
    order = sorted(range(len(candidates)), key=lambda n: -scores[n])  # 점수 내림차순
    return [candidates[n] for n in order[:k]]

reranked = evaluate(lambda q: rerank(q, search(q, k=10)))
reranked