Embedding API

https://api.aiearth.dev/v1
https://api.aiearth.vip/v1

POST /embeddings

import os
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["AIO_API_KEY"],
    base_url=os.environ.get("AIO_BASE_URL", "https://api.aiearth.dev/v1"),
)

response = client.embeddings.create(
    model="text-embedding-3-large",
    input="人工智能正在改变世界。",
)

embedding = response.data[0].embedding
print("向量维度:", len(embedding))
print("前 5 个值:", embedding[:5])

curl "https://api.aiearth.dev/v1/embeddings" \
  -H "Authorization: Bearer $AIO_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "text-embedding-3-large",
    "input": "人工智能正在改变世界。"
  }'

texts = [
    "机器学习是人工智能的一个分支。",
    "深度学习使用神经网络进行表示学习。",
    "自然语言处理让机器理解和生成语言。",
]

response = client.embeddings.create(
    model="text-embedding-3-large",
    input=texts,
)

embeddings = [item.embedding for item in response.data]
print("处理文本数量:", len(embeddings))

{
  "object": "list",
  "data": [
    {
      "object": "embedding",
      "embedding": [0.0023, -0.0093],
      "index": 0
    }
  ],
  "usage": {
    "prompt_tokens": 8,
    "total_tokens": 8
  }
}

import math

def cosine_similarity(a, b):
    dot = sum(x * y for x, y in zip(a, b))
    norm_a = math.sqrt(sum(x * x for x in a))
    norm_b = math.sqrt(sum(y * y for y in b))
    return dot / (norm_a * norm_b)

response = client.embeddings.create(
    model="text-embedding-3-large",
    input=[
        "我想了解 API 计费方式。",
        "请问接口费用怎么计算？",
    ],
)

vec1 = response.data[0].embedding
vec2 = response.data[1].embedding
print(cosine_similarity(vec1, vec2))

documents = [
    {"id": "doc-1", "text": "AIOAGI 支持 OpenAI 兼容接口。"},
    {"id": "doc-2", "text": "Token 组可用于额度管理。"},
    {"id": "doc-3", "text": "图像生成接口支持多种模型。"},
]

doc_vectors = client.embeddings.create(
    model="text-embedding-3-large",
    input=[doc["text"] for doc in documents],
).data

query_vector = client.embeddings.create(
    model="text-embedding-3-large",
    input="如何管理不同应用的额度？",
).data[0].embedding

ranked = sorted(
    zip(documents, doc_vectors),
    key=lambda item: cosine_similarity(query_vector, item[1].embedding),
    reverse=True,
)

print(ranked[0][0])

labels = {
    "billing": "充值、计费、余额、价格、发票、额度",
    "technical": "接口调用、报错、模型参数、SDK、网络",
    "account": "登录、账号、API Key、权限、控制台",
}

label_response = client.embeddings.create(
    model="text-embedding-3-large",
    input=list(labels.values()),
)
text_response = client.embeddings.create(
    model="text-embedding-3-large",
    input="我的 API Key 返回 401，应该怎么办？",
)

text_vector = text_response.data[0].embedding
label_vectors = [item.embedding for item in label_response.data]
label_names = list(labels.keys())

best_index = max(
    range(len(label_vectors)),
    key=lambda i: cosine_similarity(text_vector, label_vectors[i]),
)

print(label_names[best_index])

{
  "id": "doc-2026-001",
  "text": "AIOAGI 的 OpenAI 兼容接口基础地址为 https://api.aiearth.dev/v1。",
  "metadata": {
    "product": "aioagi",
    "source": "docs",
    "updated_at": "2026-05-15"
  },
  "embedding": [0.0023, -0.0093]
}

def batches(items, size=64):
    for i in range(0, len(items), size):
        yield items[i:i + size]

all_vectors = []
for batch in batches(texts):
    response = client.embeddings.create(
        model="text-embedding-3-large",
        input=batch,
    )
    all_vectors.extend(item.embedding for item in response.data)

import hashlib

def cache_key(text, model):
    digest = hashlib.sha256(text.encode("utf-8")).hexdigest()
    return f"{model}:{digest}"