동적 퓨 샷 선택

정적 예시에서 동적 예시로

class DynamicSelector:
    def __init__(self, pool, embedder, index):
        self.pool = pool          # candidate demonstrations
        self.embed = embedder
        self.index = index        # ANN index over pool embeddings
    def select(self, query, k):
        q = self.embed(query)
        ids = self.index.search(q, k)
        return [self.pool[i] for i in ids]

예시 풀 임베딩하기

import numpy as np

def build_index(pool, embed):
    vecs = np.stack([embed(d.input) for d in pool]).astype('float32')
    vecs /= np.linalg.norm(vecs, axis=1, keepdims=True)  # cosine via dot
    index = HNSW(dim=vecs.shape[1])
    index.add(vecs)
    return index

kNN 프롬프팅

def knn_prompt(query, selector, k, build):
    demos = selector.select(query, k)
    demos = order_by_similarity(embed(query), demos)  # most similar last
    return build(demos, query)

다양성을 고려한 검색

def diverse_retrieve(query, selector, k, pool_n=30, lam=0.7):
    cand = selector.select(query, pool_n)
    q = embed(query)
    return mmr_against_query(cand, q, k, lam)  # relevance + diversity

지연 시간과 검색 예산

from functools import lru_cache

@lru_cache(maxsize=50_000)
def cached_embed(text):
    return embed(text)
# Plus: warm in-RAM HNSW, batched search, async prefetch

동적 예시와 캐싱의 충돌

prompt = (
    STATIC_PREAMBLE          # cached: instructions + anchor demos
    + render(diverse_retrieve(query, selector, k))  # dynamic tail
    + format_query(query)
)

학습/테스트 누수 방지

def leak_safe_select(query, selector, k, sim_cap=0.97):
    cand = selector.select(query, k + 5)
    q = embed(query)
    cand = [d for d in cand if cos(q, d.emb) < sim_cap]
    return cand[:k]

초기 구동과 풀 확장

def maybe_add_to_pool(trace, verified):
    if verified and novelty(trace, index) > THRESH:
        emb = embed(trace.input)
        index.add(emb)
        pool.append(Demo(trace.input, trace.output, trace.meta))

유사도를 넘어선 선택

def select_by_uncertainty_reduction(query, pool, k):
    base = entropy(model_probs(build([], query)))
    gains = []
    for d in pool:
        h = entropy(model_probs(build([d], query)))
        gains.append((d, base - h))   # info gain per demo
    return [d for d, _ in sorted(gains, key=lambda x: -x[1])[:k]]

동적 처리 과정 평가

def eval_pipeline(eval_set):
    return {
        'acc_dynamic': run(dynamic, eval_set),
        'acc_static':  run(static, eval_set),
        'acc_random':  run(random_sel, eval_set),
        'p95_latency': latency_p95(),
        'cache_hit':   cache_hit_rate(),
    }

참조 아키텍처

def answer(query):
    demos = leak_safe_select(query, selector, k=4)
    demos = mmr_against_query(demos, embed(query), 4)
    demos = order_by_similarity(embed(query), demos)
    prompt = STATIC_PREAMBLE + render(demos) + format_query(query)
    out = llm(prompt)
    log_for_pool_growth(query, out)
    return out

빠른 확인

요약