Embeddings 向量检索

import json
import math
import os
from openai import OpenAI, APIConnectionError, APIStatusError

query = "How can I reset my password?"
documents = [
    "Invoices are available in the billing dashboard.",
    "Use the password reset link on the sign-in page.",
]
inputs = [query, *documents]

with OpenAI(base_url=os.environ["API_BASE_URL"].rstrip("/"),
            api_key=os.environ["API_KEY"], timeout=60.0, max_retries=0) as client:
    try:
        result = client.embeddings.create(
            model=os.environ["API_MODEL"], input=inputs, encoding_format="float",
        )
        vectors = {}
        dimensions = None
        if not isinstance(result.data, list) or len(result.data) != len(inputs):
            raise ValueError("Expected one vector per input.")
        for item in result.data:
            index = getattr(item, "index", None)
            vector = getattr(item, "embedding", None)
            if type(index) is not int or not 0 <= index < len(inputs) or index in vectors:
                raise ValueError("Invalid or duplicate input index.")
            if not isinstance(vector, list) or not vector:
                raise ValueError("Expected a nonempty float vector.")
            if any(type(value) not in (int, float) or not math.isfinite(value) for value in vector):
                raise ValueError("Invalid vector value.")
            if dimensions is not None and len(vector) != dimensions:
                raise ValueError("Inconsistent vector dimensions.")
            dimensions = len(vector)
            norm = math.hypot(*vector)
            if not math.isfinite(norm) or norm == 0:
                raise ValueError("Cannot normalize this vector.")
            vectors[index] = [value / norm for value in vector]
        matches = []
        for index, document in enumerate(documents):
            score = math.fsum(a * b for a, b in zip(vectors[0], vectors[index + 1]))
            matches.append({"index": index, "score": score, "text": document})
        matches.sort(key=lambda item: item["score"], reverse=True)
        print(json.dumps(matches, ensure_ascii=False, allow_nan=False))
    except APIStatusError as error:
        raise SystemExit(f"Request failed (HTTP {error.status_code}). Check model access and request limits.")
    except APIConnectionError:
        raise SystemExit("Connection failed. Check the API address and network.")
    except (ValueError, TypeError, OverflowError):
        raise SystemExit("Invalid embedding response; check indices, dimensions, and numeric values.")