前面介绍的 get 方法是根据过滤条件返回数据，不是通过相识度进行搜索，这和数据库查询数据效果一样。而我们使用向量数据库的最终目的是进行相似度查询，而不是根据条件精确查找。
相似度搜索：
不是按完全相等、包含某个字、ID 匹配这种精确条件去查，而是：
把你输入的一段话，转成向量，然后在向量库里，找出向量距离最近的那些文档，代表语义上最相近。
它比关键词匹配聪明：
关键词搜索：必须有一样的字才命中（比如搜“苹果”，文本写“红富士”就搜不到）
相似度搜索：理解语义，搜“水果”，存的“红富士苹果”也能排前面
方法签名如下：
collection.query(
query_embeddings=None, # 直接给向量
query_texts=None, # 给文本，让嵌入函数算
query_images=None, # 给图片（多模态）
query_uris=None, # 给地址（多模态）
ids=None, # 只在这批 id 里搜
n_results=10, # 每个 query 返回几条
where=None, # 元数据过滤
where_document=None, # 文档内容过滤
include=["metadatas", "documents", "distances"],
)参数说明：
query_embeddings：类型为 list[list[float]] | None，直接传入已经计算好的向量数组，不再调用 embedding 函数。适合自己提前算好向量的场景。例如：query_embeddings=[[0.1,0.2,...]]
query_texts：类型为 list[str] | None，传入查询文本列表，Chroma 会自动调用集合绑定的嵌入函数，把文本转为向量再检索。日常 RAG 最常用。例如：query_texts=["怎么排查接口超时"]
query_images：类型为 list | None，多模态专用，传入图片对象，用于以图搜图，依赖 OpenCLIP 等多模态嵌入函数。
query_uris：类型为 list[str] | None，多模态专用，传入图片文件路径，需要集合提前配置 data_loader，否则会报 URI 加载错误。
ids：类型为 list[str] | None，用来限定搜索范围，只在你给出的这些 id 里面做相似度排序，库中其他数据直接跳过。
n_results：类型为 int，每一条查询对应的返回数量。比如 query_texts 给 2 段文字，每一段都会返回 n_results 条相似数据。
where：类型为 dict | None，用于元数据条件过滤（和 get 里面 where 语法一致）。 执行逻辑：先按 where 筛选出符合元数据的数据，再在筛选结果里面做相似度排序。
where_document：类型为 dict | None，文档文本过滤，仅支持 $contains。同样是先过滤、再算相似度。
include：类型为 list[str]，指定返回携带的字段，可选值如下：
"distances"：向量距离（Chroma 默认欧式距离，数值越小代表越相似）
"documents"：原文
"metadatas"：元数据
"embeddings"：匹配到的文档对应的向量（开销大，按需开启）
"uris"：资源地址
注意：query_embeddings / query_texts / query_images / query_uris 四选一，不能同时多个生效。
下面通过简单示例介绍相似度检索的用法：
import chromadb
from common.custom_embedding import AliQwenEmbeddingFunction
# 实例化嵌入函数
ef = AliQwenEmbeddingFunction()
client = chromadb.EphemeralClient()
col = client.get_or_create_collection("demo_col", embedding_function=ef)
col.add(
ids=["r1", "r2", "r3", "r4", "r5"],
documents=[
"退款流程：在订单页点申请退款，3 个工作日内到账",
"发票开具：在订单详情页申请电子发票",
"物流查询：发货后可在物流详情里看到进度",
"退货政策：7 天无理由退货，运费买家承担",
"账号安全：如何修改密码和绑定手机",
],
metadatas=[
{"cat": "售后", "year": 2024, "score": 0.9, "tags": ["退款"]},
{"cat": "财务", "year": 2023, "score": 0.5, "tags": ["发票"]},
{"cat": "物流", "year": 2024, "score": 0.7, "tags": ["物流"]},
{"cat": "售后", "year": 2022, "score": 0.3, "tags": ["退货"]},
{"cat": "账号", "year": 2024, "score": 0.6, "tags": ["安全"]},
],
)
res = col.query(query_texts=["怎么退款"], n_results=2)
print(f"ids={res["ids"]}")
print(f"documents={res["documents"]}")
print(f"distances={res["distances"]}") # 值越小，越相似运行示例，输出如下：
ids=[['r1', 'r4']]
documents=[['退款流程：在订单页点申请退款，3 个工作日内到账', '退货政策：7 天无理由退货，运费买家承担']]
metadatas=[[{'tags': ['退款'], 'cat': '售后', 'year': 2024, 'score': 0.9}, {'tags': ['退货'], 'cat': '售后', 'year': 2022, 'score': 0.3}]]
distances=[[0.5935589075088501, 1.046556830406189]]从输出可以看出，“退款流程：在订单页点申请退款，3 个工作日内到账”问题的相似度为 0.5935589075088501，最为相似。
注意：嵌入函数不要使用默认的嵌入函数，它对中文支持基本为零，需要自己实现嵌入函数，可以选择调用现有大模型平台的嵌入模型，如阿里云的 qwen3.7-text-embedding-flash 等。
使用向量进行查询和使用文本查询类似，需要自己将查询文本转化为向量，传递给 query 方法进行查询。注意，转化向量的模型需要和写入数据使用的向量模型已知，维度大小一致，才能查询出有效数据。
例如：自定义一个 AliEmbedding 类，封装使用 qwen3.7-text-embedding-flash 计算向量的便捷方法，如下：
class AliEmbedding:
def __init__(self):
# 在环境变量添加自己的 API KEY
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 get_embedding(self, content: str) -> List[float]:
"""
对单个文本content调用阿里云qwen3.7-text-embedding-flash向量接口，返回向量数组
:param content: 输入文本字符串
:return: float向量列表
"""
payload = {
"input": content,
"model": self.model_name
}
resp = requests.post(self.endpoint, headers=self.headers, json=payload, timeout=60)
resp.raise_for_status() # http非2xx直接抛异常
data = resp.json()
embedding = data["data"][0]["embedding"]
return embedding
def batch_get_embedding(self, contents: List[str]) -> List[List[float]]:
"""
批量向量化，一次传入多条文本
:param contents: 文本列表
:return: 向量列表
"""
payload = {
"input": contents,
"model": self.model_name
}
resp = requests.post(self.endpoint, headers=self.headers, json=payload, timeout=60)
resp.raise_for_status()
data = resp.json()
# 按index排序保证顺序一致
items = sorted(data["data"], key=lambda x: x["index"])
embeddings = [item["embedding"] for item in items]
return embeddings引入自定义的 AliEmbedding 类，使用提供的方法计算向量：
import chromadb
from common.custom_embedding import AliEmbedding, AliQwenEmbeddingFunction
# 注意：保证写入和查询使用同一个嵌入模型
# 实例化自己封装的嵌入类
ali_embedding = AliEmbedding()
ef = AliQwenEmbeddingFunction()
client = chromadb.EphemeralClient()
col = client.get_or_create_collection("demo_col", embedding_function=None)
# 也自己计算向量
col.add(
ids=["r1", "r2", "r3", "r4", "r5"],
documents=[
"退款流程：在订单页点申请退款，3 个工作日内到账",
"发票开具：在订单详情页申请电子发票",
"物流查询：发货后可在物流详情里看到进度",
"退货政策：7 天无理由退货，运费买家承担",
"账号安全：如何修改密码和绑定手机",
],
# 添加数据时，批量计算向量
embeddings=ali_embedding.batch_get_embedding([
"退款流程：在订单页点申请退款，3 个工作日内到账",
"发票开具：在订单详情页申请电子发票",
"物流查询：发货后可在物流详情里看到进度",
"退货政策：7 天无理由退货，运费买家承担",
"账号安全：如何修改密码和绑定手机",
]),
metadatas=[
{"cat": "售后", "year": 2024, "score": 0.9, "tags": ["退款"]},
{"cat": "财务", "year": 2023, "score": 0.5, "tags": ["发票"]},
{"cat": "物流", "year": 2024, "score": 0.7, "tags": ["物流"]},
{"cat": "售后", "year": 2022, "score": 0.3, "tags": ["退货"]},
{"cat": "账号", "year": 2024, "score": 0.6, "tags": ["安全"]},
],
)
# 查询，计算向量，通过 query_embeddings 将向量传递进去
res = col.query(query_embeddings=[
ali_embedding.get_embedding("怎么退款")
], n_results=2)
print(f"ids={res["ids"]}")
print(f"documents={res["documents"]}")
print(f"distances={res["distances"]}") # 值越小，越相似运行代码，输出结果如下：
ids=[['r1', 'r4']]
documents=[['退款流程：在订单页点申请退款，3 个工作日内到账', '退货政策：7 天无理由退货，运费买家承担']]
distances=[[0.6103367805480957, 1.0393226146697998]]注意，query_embeddings 是“向量的列表”，哪怕只查一个，也要套一层。
这是 Chroma 一个很好用但经常被忽略的能力，一次查多个问题。用法如下：
res = col.query(
# 提供了两组向量
query_embeddings=[
[1, 0, 0], # 问题一的向量
[0, 1, 0] # 问题二的向量
],
n_results=2,
)
print(res["ids"])
# [['a', 'b'], ['b', 'a']] ← 两组结果
print(res["distances"])
# [[0.0, 2.0], [0.0, 2.0]] ← 两组相似度结果批量比循环单查快不少，因为省了网络往返和索引遍历的开销。做评测（比如算 recall@10）的时候特别爽。
如果 query 方法 n_results 参数指定 的值大于集合中数据总数，Chroma 不会报错，而是返回实际有的那么多。例如：
# 假设集合只有 3 条，n_results 设置为 10
res = col.query(query_embeddings=[[1, 0, 0]], n_results=10)
print(len(res["ids"][0])) # 3上面代码不会报错哦，会返回集合全部数据，只是返回的数据的 distances 会有差别，顺序有差别。
如果 query 添加了 ids、where、where_document 等过滤条件，执行逻辑是：先过滤数据，再在过滤后的子集里做向量检索。
例如：先检索 metadatas 中 t 等于 y 的记录，然后再进行相似度检索。
res = col.query(
query_embeddings=[[1, 0, 0]],
n_results=3,
where={"t": "y"}, # 只搜 t='y' 的
)
print(res["ids"][0], res["distances"][0])
# ['c'] [2.0] ← 只有 1 条符合过滤，就只返回 1 条使用 id 进行精确过滤：
res = col.query(query_embeddings=[[1, 0, 0]], n_results=3, ids=["b"])
print(res["ids"][0])
# ['b'] ← 只在 id='b' 里搜重要：这是“过滤后检索”（pre-filter），不是“检索后过滤”。如果过滤条件把候选集砍得很小，返回的条数会少于 n_results。反过来，如果过滤条件很宽松（比如 100 万条里过滤掉 100 条），HNSW 在大量被过滤掉的节点上会退化，召回率和延迟都会变糟。第 6 章详细说这个。
query 的默认 include 是 ["metadatas", "documents", "distances"]，比 get 多一个 distances。distances 表示距离，值越小，表示向量越接近。
如果想省流量可以不带，就像这样：
res = col.query(query_texts=["x"], n_results=5, include=["documents"])
# res["distances"] 是 None
川公网安备51010802032098