실전 임베딩 튜닝

import time
from openai import OpenAI

client = OpenAI(api_key="sk-your-apiyi-key", base_url="https://api.apiyi.com/v1", timeout=90.0)

def embed_batch(texts, model="bge-m3", retries=3):
    """Batch embedding with backoff. Keep batches between 64 and 128."""
    for attempt in range(retries):
        try:
            resp = client.embeddings.create(model=model, input=texts)
            return [d.embedding for d in sorted(resp.data, key=lambda x: x.index)]
        except Exception:
            if attempt == retries - 1:
                raise
            time.sleep(2 ** attempt)          # 1s, 2s, 4s

from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(
    model="bge-m3",
    api_key="sk-your-apiyi-key",
    base_url="https://api.apiyi.com/v1",
    check_embedding_ctx_length=False,   # required, otherwise tiktoken ids go out instead of text
    chunk_size=64,
)

{ "model": "bge-m3", "input": "...", "dimensions": 512 }

"""Minimal working bge-m3 indexing + retrieval."""
import time
import numpy as np
from openai import OpenAI

client = OpenAI(
    api_key="sk-your-apiyi-key",
    base_url="https://api.apiyi.com/v1",
    timeout=90.0,
)

MODEL = "bge-m3"
BATCH = 96          # keep between 64 and 128
THRESHOLD = 0.50    # starting point; recalibrate on your own samples before launch


def embed(texts, retries=3):
    """Batch embedding with backoff. Returns float16 vectors."""
    texts = [t.strip() for t in texts if t and t.strip()]   # drop blanks, see 7.2
    out = []
    for i in range(0, len(texts), BATCH):
        chunk = texts[i:i + BATCH]
        for attempt in range(retries):
            try:
                resp = client.embeddings.create(model=MODEL, input=chunk)
                out += [d.embedding for d in sorted(resp.data, key=lambda x: x.index)]
                break
            except Exception:
                if attempt == retries - 1:
                    raise
                time.sleep(2 ** attempt)
    # already normalized, so float16 storage is lossless here, see section 6
    return np.array(out, dtype=np.float16)


def search(query, doc_vectors, docs, top_k=50):
    """Vectors are normalized, so the dot product is cosine similarity."""
    qv = embed([query])[0].astype(np.float32)
    scores = doc_vectors.astype(np.float32) @ qv
    order = np.argsort(-scores)[:top_k]
    return [(docs[i], float(scores[i])) for i in order if scores[i] >= THRESHOLD]


if __name__ == "__main__":
    docs = ["...your chunks, 200-500 tokens each..."]
    dv = embed(docs)
    for text, score in search("your question", dv, docs):
        print(f"{score:.4f}  {text[:60]}")
    # In production, hand this Top 50 to bge-reranker-v2-m3, see section 4