cd ../返回博客
$Guide//2026年6月4日//8 min read

用 OpenAI SDK 实现嵌入与语义检索(RAG 实战指南)

用 /v1/embeddings 生成文本嵌入,以余弦相似度打分,在 pgvector 中存储与查询向量,再接入 RAG —— 可运行,附成本提示。

关键词检索匹配的是字符串。语义检索匹配的是 语义 ——「我忘了我的登录信息」能找到标题为 「重置你的密码」的文章,哪怕两者一个词都不重合。背后的 机制是嵌入:模型把每一段文本变成一个由几千个数字组成的向量, 语义相近的文本,在那个空间里也彼此靠得很近。把这些向量一次性 生成好、存起来,你就能按相关度给任何东西排序 —— 这正是每一个 RAG 系统里负责检索的那一半。

Brievio 把 /v1/embeddings 作为一个可直接替换、 与 OpenAI 兼容的接口暴露出来,所以 SDK 调用和你已经写惯的 一模一样 —— 只需改一个 base_url。嵌入很便宜,你 付的是诚实的 token 数,失败的 4xx/5xx 调用则不收费。这篇文章 走的是实战路线:生成向量、用余弦相似度给它们打分、用 pgvector 存储和查询,再把结果接进一条检索流水线 —— 连同那些真正会咬人的 坑一起讲清楚。

生成嵌入向量

一次调用。传入一个字符串列表,按相同顺序拿回一个向量列表。 永远要批量 —— 每个请求几百条输入,单 token 成本和一条一条发 完全相同,却帮你省下成百上千次往返:

embed.py
# 用 OpenAI SDK 生成嵌入向量 —— 同样的调用,只是换了 base_url。
from openai import OpenAI

client = OpenAI(
    api_key="sk-brievio-...",
    base_url="https://api.brievio.com/v1",
)

EMBED_MODEL = "text-embedding-3-large"   # 选定一个模型,从一而终

def embed(texts: list[str]) -> list[list[float]]:
    # 每次调用批量处理几百条输入 —— 往返次数大幅减少,
    # 单 token 价格不变。响应会保持输入的原始顺序。
    resp = client.embeddings.create(model=EMBED_MODEL, input=texts)
    return [row.embedding for row in resp.data]

vecs = embed(["怎么重置我的密码?", "我的发票在哪里?"])
print(len(vecs), "x", len(vecs[0]))     # 2 x 3072  (维度取决于模型)

# usage 如实上报 —— 嵌入和其他调用一样按 token 计量
print(resp.usage.prompt_tokens, "tokens billed")

向量长度(它的维度)由模型固定 —— 常见的是 768、1536 或 3072。维度越高,能捕捉的细微差别略多 一些,但存储和比较的成本也越高。最重要的一条铁律:选定一个嵌入模型,绝不混用。来自不同模型的 向量活在彼此不兼容的空间里 —— 拿它们互相比较 只会得到噪声。可用的嵌入模型及其维度,见线上的 模型目录;每个模型的定价则在 定价页上。

用余弦相似度打分

要找出最接近的匹配,你需要把查询向量和你已存的向量逐一比较。 标准的度量是余弦相似度:它衡量两个向量之间的 夹角,忽略它们的长度,因此只在乎方向(语义)而非大小。1.0 表示 方向完全一致,0 表示毫不相关:

search.py
# 语义检索 = 把查询嵌入向量,与每一个已存向量打分,
# 返回最接近的那些。打分函数就是余弦相似度。
import numpy as np

def cosine(a: np.ndarray, b: np.ndarray) -> float:
    # 1.0 = 方向完全一致,0 = 毫不相关,-1 = 完全相反。
    return float(a @ b / (np.linalg.norm(a) * np.linalg.norm(b)))

# 很多嵌入模型本身就返回单位长度的向量。一旦如此,
# 余弦相似度就退化成一个普通的点积 —— 大规模下更省算力:
def cosine_unit(a: np.ndarray, b: np.ndarray) -> float:
    return float(a @ b)

query = np.array(embed(["我忘了我的登录信息"])[0])
corpus = np.array(embed(documents))                 # (N, dim)

scores = corpus @ query / (
    np.linalg.norm(corpus, axis=1) * np.linalg.norm(query)
)
top = scores.argsort()[::-1][:5]                    # 最匹配的 5 条的下标
for i in top:
    print(round(float(scores[i]), 3), documents[i][:60])

两点实用提醒。第一,很多嵌入模型本身就返回单位长度的向量 —— 一旦如此,余弦相似度就只是一个点积,这也正是为什么对几万个 向量做暴力检索,也能快到完全用不上数据库。第二,在大约几万个 向量以内,暴力循环都没问题。一旦超过,每次查询都扫一遍所有 向量就会变慢,这时你就需要一个带索引的真正向量库。

用 pgvector 存储和查询

如果你的数据本来就在 Postgres 里,你不需要另一个独立的 向量数据库 —— pgvector 扩展会加上一个 vector 列类型和最近邻运算符。你把每个片段嵌入 一次,把向量存在它的文本旁边,然后让 Postgres 来做检索:

store.sql
-- pgvector 把 Postgres 变成一个向量库。只需启用一次,随后就能
-- 把每个片段的嵌入向量,和它的文本与元数据存在一起。
CREATE EXTENSION IF NOT EXISTS vector;

CREATE TABLE chunks (
    id        bigserial PRIMARY KEY,
    doc_id    text NOT NULL,
    content   text NOT NULL,
    embedding vector(3072)            -- 必须与你的模型维度一致
);

-- 近似最近邻索引。在 pgvector 里 <=> 是余弦距离
-- (0 = 最近)。索引要在批量灌数据之后再建,不要在之前。
CREATE INDEX ON chunks
    USING hnsw (embedding vector_cosine_ops);

-- 检索:把查询向量作为参数($1)传入,取出最近的几行。
-- 按距离升序排列 = 最相似的排在最前面。
SELECT id, doc_id, content, 1 - (embedding <=> $1) AS similarity
FROM   chunks
WHERE  doc_id = ANY($2)               -- 可选的元数据预过滤
ORDER  BY embedding <=> $1
LIMIT  5;

<=> 运算符算的是余弦距离(0 = 完全相同),所以 1 - (embedding <=> $1) 又把 相似度分数还给了你。hnsw 索引让检索变成近似的, 但很快 —— 对大多数检索负载来说,那一点点召回率的折损根本看不 出来,而且你要在批量灌数据之后再建它,绝不要在之前。 那条可选的 WHERE 子句是个低调的杀手锏:在向量 检索之前就按租户、文档、语言或日期过滤,这样你永远 不会检索到一个用户无权看到的邻居。

把它接进 RAG

检索增强生成就是两步拼接在一起:先用语义检索找到相关上下文, 再用一个生成模型基于这些上下文作答。检索那一侧就是上面讲的 全部。生成那一侧是一次普通的对话补全 —— 而这里正是你把一个 便宜的嵌入模型,和一个强大的推理模型配在一起的地方:

  • 先切片,再嵌入。把文档拆成大约几百个 token 的段落,并留一点重叠,逐段嵌入后存起来。切片是大多数团队 投入最不足的那根杠杆:片段太大,答案会淹没在噪声里;太小, 又会丢掉让它有意义的那点上下文。请在你自己的数据上调它。
  • 在查询时检索。同一个模型嵌入用户 的问题,拉出排在最前的 3~8 个片段,把它们的文本作为上下文 贴进提示里。
  • 生成答案。把这些上下文连同问题一起,发给一个 有能力的模型 —— 通过同一个 Brievio 接口调用 Claude 或 Gemini —— 并指示它只依据所给上下文作答,并标明它用了哪个 片段。

这套经济账之所以成立,是因为这两半的价格档次天差地别。把你 整个语料库嵌入是一次性的、低成本的批处理作业;只有内容变动 时才需要重新嵌入。每次查询的成本由生成那一步主导,而不是检索 —— 这也正是为什么我们那篇 AI API 成本优化手册 里那些控本技巧(对静态指令做提示缓存、限制输出上限、为每个 任务挑对模型)所能撬动的,远比嵌入这一侧的任何招数都多。

那些真正会咬人的坑

  • 向量在不同模型之间不可互换。一旦更换嵌入 模型,你就必须把整个语料库重新嵌入一遍 —— 查询向量和已存 向量必须来自同一个模型,否则分数毫无意义。把模型选择当成 一个 schema 层面的决定来对待。
  • 维度是一桩存储与速度的权衡。一个 3072 维的 向量,字节数是 768 维的四倍,比较起来也更慢。对你的任务而言, 越大并不自动等于越好 —— 在为最大的模型买单之前,先在你自己 的查询上测一测召回率。
  • 切片质量胜过模型质量。大多数糟糕的 RAG 答案, 根子都在糟糕的切片上,而不是嵌入模型不够强。请尊重文档结构 —— 按标题和段落来切,而不是盲目地按字符数。
  • 语义检索可能漏掉精确词项。产品 SKU、错误码 和名称,有时用关键词检索反而匹配得更准。实践中最强的系统都是混合式的:把向量相似度和经典的全文检索或 BM25 分数 结合起来。
  • 嵌入也有上下文上限。超出模型输入上限的文本 会被悄悄截断 —— 你的「嵌入」于是只代表了一个过长片段的开头 那部分。请让片段稳稳地留在上限以内。

可落地的要点

语义检索由四个活动部件组成:一个嵌入模型、一个相似度度量、 一个存储,和一套切片策略。选定一个嵌入模型并坚持下去;用余弦 相似度;从暴力的 NumPy 起步,等语料库大到一次扫描扛不住时,再 升级到 pgvector 的 HNSW 索引;把你的调优预算花在切片上,因为 检索质量的成败正是在那里决出。通过 Brievio,这次嵌入调用就是你 早已熟悉的 OpenAI SDK,只改一行,按诚实的 token 数计量,而且它 就紧挨着你在生成那一步会用到的 Claude 和 Gemini 模型 —— 一把 密钥、一个 base_url,整条 RAG 流水线。接口参考和一个 可直接运行的快速上手,都在 文档里。