向量檢索實戰調優

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):
    """帶退避重試的批次向量化。批次控制在 64-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 as e:
            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,   # 必須關掉，否則發出去的是 tiktoken 的 token id
    chunk_size=64,                      # 每批 64 條
)

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

"""bge-m3 灌庫 + 檢索的最小可用實現。"""
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          # 64-128 之間
THRESHOLD = 0.50    # 起步值，上線前用自己的樣本重標


def embed(texts, retries=3):
    """批次向量化，帶退避重試。返回 float32 陣列。"""
    texts = [t.strip() for t in texts if t and t.strip()]   # 過濾空塊，見 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)
    # 向量已歸一化，直接存 float16 無精度損失，見第六節
    return np.array(out, dtype=np.float16)


def search(query, doc_vectors, docs, top_k=50):
    """向量已歸一化，點積即餘弦相似度。"""
    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 = ["……你的切塊結果，每塊 200-500 token……"]
    dv = embed(docs)
    for text, score in search("使用者的問題", dv, docs):
        print(f"{score:.4f}  {text[:60]}")
    # 生產環境請把這裡的 Top 50 交給 bge-reranker-v2-m3 精排，見第四節