要让 Nemotron 3 Embed 真正接进 RAG 管线，先别急着改检索逻辑，而是先确认推理服务有没有把模型加载好，以及返回的向量结构是否稳定。向量化这一环最容易出的问题不是模型效果，而是请求路径不对、向量维度不匹配、向量没归一化导致后面索引距离失真。下面这套流程按“服务确认→请求构造→批量转向量→维度和归一化校验→接入索引”的顺序走，每个步骤都给出能直接改的代码骨架。

Nemotron 3 Embed 能否用，重点不在模型名称，而在向量服务返回的维度是否稳定、向量是否归一化。适用场景：已有 RAG 框架，希望替换文本向量化组件。操作动作：先健康检查，再发单条请求，随后批量调用，最后校验长度和模长。验证方式：打印每个响应中的向量长度，计算模长并对比 1.0。风险边界：不同部署方式的请求路径不同，维度需以模型配置为准，索引维度必须与返回维度一致。

先确认推理服务的模型加载状态

嵌入服务如果没起来，后面所有请求都会 4xx 或连接失败。常见部署方式是 Triton、vLLM 或者自定义 FastAPI 网关。先看服务健康检查接口和日志输出，确认模型已经 ready，再进入下一步。

curl http://localhost:8000/health
# 期望输出里包含 ready 或 ok 状态，具体字段看服务实现
# 也可以直接看服务日志，搜“model loaded”或“ready”

如果健康检查通过但调用仍然失败，检查日志里有没有报显存不足或 CUDA 错误。确认服务本身是好的，才继续做嵌入请求。

构造嵌入请求并检查响应结构

嵌入服务通常接收文本列表，返回一个带向量字段的 JSON。这里给一个最小请求示例，假设服务地址是 http://localhost:8000/v1/embeddings，这是 OpenAI 兼容接口的常见路径，不是固定标准，需要按实际部署调整。

import requests
import json

payload = {
    "input": ["RAG 检索测试文本"],
    "model": "nemotron-3-embed"
}
resp = requests.post("http://localhost:8000/v1/embeddings", json=payload, timeout=10)
resp.raise_for_status()
data = resp.json()
print(json.dumps(data, indent=2, ensure_ascii=False)[:500])

响应中应出现的字段通常包括：object、data、model，其中 data 是一个列表，每个元素里有 embedding 和 index。检查时先打印前 500 字符，确认 embedding 存在且是数组。如果响应里没有 embedding，说明请求路径或参数名不对，需要回翻服务文档。

编写循环脚本对文档列表批量向量化

手动一条条调用不现实。写一个批量脚本，用 ThreadPoolExecutor 控制并发，同时加入失败重试。注意重试时先判断错误类型，网络超时和 HTTP 5xx 才重试，4xx 一般是请求本身问题，不要盲目重试。

import concurrent.futures
import time
import requests

def get_embedding(text, retries=3):
    url = "http://localhost:8000/v1/embeddings"
    payload = {
        "input": [text],
        "model": "nemotron-3-embed"
    }
    for attempt in range(retries):
        try:
            r = requests.post(url, json=payload, timeout=15)
            r.raise_for_status()
            return r.json()["data"][0]["embedding"]
        except Exception as e:
            if attempt == retries - 1:
                raise
            time.sleep(2 ** attempt)  # 指数退避

def batch_embed(texts, max_workers=4):
    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as ex:
        return list(ex.map(get_embedding, texts))

docs = ["文档1内容", "文档2内容", "文档3内容"]
vectors = batch_embed(docs)
print(len(vectors), len(vectors[0]))

并发数建议先设 4，确认服务能稳定处理后再增加。如果服务端有速率限制，需要捕获 429 状态码并增加重试等待。

验证向量维度与归一化状态

拿到一批向量后，先检查所有向量长度是否一致，再计算模长。很多向量化模型在服务端已经归一化，但也有不归一化的情况。如果后续用 faiss 等索引做内积或余弦检索，需要确保输入都是单位向量，否则可以先做归一化处理。

import numpy as np

v = vectors[0]
print("维度:", len(v))
assert all(len(vec) == len(v) for vec in vectors), "向量维度不一致"

arr = np.array(vectors, dtype=np.float32)
norms = np.linalg.norm(arr, axis=1, keepdims=True)
print("最大模长:", norms.max())
print("最小模长:", norms.min())

# 若模长明显不等于 1，做归一化
if not np.allclose(norms.flatten(), 1.0, atol=1e-3):
    arr = arr / norms
    print("已归一化，归一化后模长:", np.linalg.norm(arr, axis=1).max())

维度结果要以服务返回为准，不要从模型名推测。如果维度不满足后续索引要求，比如索引建好就不能改维度，那就需要重新调整服务配置或换索引类型。

将向量接入现有检索索引

以 faiss 为例，建索引时要明确使用内积还是余弦。如果向量已经归一化，内积和余弦等价。下面给出插入向量和查询的骨架。

import faiss
import numpy as np

# 假设 arr 是归一化后的 float32 矩阵，shape (N, D)
dim = arr.shape[1]
index = faiss.IndexFlatIP(dim)  # 内积索引，等价于余弦（向量已归一化）
index.add(arr)  # 插入全部向量

# 查询：query 是一个归一化后的向量，shape (1, D)
query = get_embedding("检索问题")
query = np.array(query, dtype=np.float32).reshape(1, -1)
# 若没有归一化，需要先做相同处理
query_norm = np.linalg.norm(query)
if not np.allclose(query_norm, 1.0, atol=1e-3):
    query = query / query_norm

D, I = index.search(query, k=3)  # 返回相似度和索引号
print("相似度:", D)
print("文档索引:", I)

接入时注意索引维度必须与向量维度一致，建议在插入前用 assert dim == index.d 做校验。如果后续要删除或更新向量，需要选择支持动态更新的索引类型（如 faiss.MultiIndexQuantizer），但这类索引往往要求更严格的预处理，建议先用 Flat 索引验证整体链路，再考虑量化索引。