一条 Record 长什么样,最简单的办法,写一个 demo,输出一条记录看看,例如:
import chromadb
client = chromadb.EphemeralClient()
col = client.create_collection("faq")
# 写入几条数据
col.add(
ids=["q1", "q2"],
documents=[
"订单付款后多久发货?一般 48 小时内出库",
"如何申请退款?在订单详情页点击申请退款",
],
metadatas=[
{"category": "物流"},
{"category": "售后"},
],
)
# 查询数据
res = col.query(query_texts=["怎么申请退款"], n_results=1,
# 一次性取出所有核心信息
include=["distances", "documents", "metadatas", "embeddings"])
print(res)注意了,默认 query() 是不输出 embeddings 的,默认为 None。上面通过 include 指定输出所有信息,包含 embeddings。
运行示例输出如下:
{
'ids': [['q2']],
'embeddings': [array([[ 4.05045711e-02, 1.02685414e-01, -8.69980920e-03,
...隐藏...
-9.42492625e-04, 1.17516676e-02, 4.61317971e-02,
8.79870206e-02, 5.30446088e-03, 4.84681455e-04]])],
'documents': [['如何申请退款?在订单详情页点击申请退款']],
'uris': None,
'included': ['distances', 'documents', 'metadatas', 'embeddings'],
'data': None,
'metadatas': [[{'category': '售后'}]],
'distances': [[0.5679782629013062]]
}一条记录有五个字段,分别如下:
| 字段 | 必填 | 说明 |
| id | 是 | 字符串,集合内唯一 |
| embedding | 二选一 | 浮点数数组,即向量 |
| document | 二选一 | 原始文本 |
| metadata | 否 | 字典,用于过滤 |
| uris | 否 | 数据的位置,比如图片路径、S3 地址 |
注意,embedding 和 document 至少要给一个:
如果只给 document,Chroma 用集合的嵌入函数帮你算向量,不推荐这么做。
如果只给 embedding,你自己算好了,Chroma 只存不算,如调用阿里云、火山等提供的嵌入模型,中文支持更好。
如果 document 和 embedding 两个都给了, Chroma 都按你给的存,不会再算一遍。
上面第三条很重要:如果你已经有向量了,一定要连 document 一起传进去,别让 Chroma 再算一次,能省很多时间。反过来,如果你传了 document 但没传 embedding,Chroma 一定会算,这往往就是你 add 慢的原因。
示例,演示如何使用阿里云的 embedding 模型 qwen3.7-text-embedding-flash 实现向量计算。
import os
import numpy as np
import requests
import chromadb
from typing import Any, Dict, cast
from dotenv import load_dotenv
from chromadb import Documents, EmbeddingFunction, Embeddings
# 从 .env 读取环境变量 OPENAI_API_KEY
load_dotenv()
# 自定义阿里云 Embedding(OpenAI兼容HTTP)
class AliQwenEmbeddingFunction(EmbeddingFunction[Documents]):
def __init__(self):
self.api_key = os.getenv("OPENAI_API_KEY", "")
self.model_name = "qwen3.7-text-embedding-flash"
self.endpoint = "https://llm-0sb8vr1k7aav4kn3.cn-beijing.maas.aliyuncs.com/compatible-mode/v1/embeddings"
self.headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
}
def __call__(self, input: Documents) -> Embeddings:
""" 输入字符串列表,返回二维float向量列表"""
payload = {
"model": self.model_name,
"input": input,
"encoding_format": "float",
}
resp = requests.post(self.endpoint, json=payload, headers=self.headers, timeout=60)
resp.raise_for_status()
data = resp.json()
return [
np.array(item["embedding"], dtype=np.float32) for item in data["data"]
]
if __name__ == "__main__":
# 初始化自定义向量化器
ef = AliQwenEmbeddingFunction()
# 内存模式的客户端
client = chromadb.EphemeralClient()
# 创建集合,指定 embedding_function,add/query 会自动调用阿里云接口算向量
col = client.create_collection(name="faq", embedding_function=ef)
# 写入几条数据:只传文本,向量由 ef 自动远程计算并存入 chroma
col.add(
ids=["q1", "q2"],
documents=[
"订单付款后多久发货?一般 48 小时内出库",
"如何申请退款?在订单详情页点击申请退款",
],
metadatas=[
{"category": "物流"},
{"category": "售后"},
],
)
# 查询:query_texts 的文本也会自动调用阿里云接口生成向量检索
res = col.query(query_texts=["质量差,不想要了"], n_results=1,)
print(res)运行示例,输出如下:
{'ids': [['q2']], 'embeddings': None,
'documents': [['如何申请退款?在订单详情页点击申请退款']], 'uris': None,
'included': ['metadatas', 'documents', 'distances'], 'data': None,
'metadatas': [[{'category': '售后'}]],
'distances': [[1.2731304168701172]]}从输出可以看出,即使用户问题“质量差,不想要了”根问题没有任何关键字匹配,也能准确找到对应的 FAQ,这就是前面让你使用自己计算向量的原因,对中文才支持语义。