Chromadb向量数据库入门指南：从基础概念到实战应用

1次阅读

没有评论

共计 2316 个字符，预计需要花费 6 分钟才能阅读完成。

传统关系型数据库（如 MySQL）在处理文本、图片等非结构化数据时，需要先将其转换为结构化形式。比如用关键词标签描述图片，这种方式会丢失大量原始信息。而向量数据库直接存储数据的向量表示（即一组数字），能够保留语义信息，适合处理：

自然语言处理（NLP）中的词向量
图片 / 视频的特征向量
推荐系统中的用户 / 商品嵌入

传统数据库的局限性在于：

无法高效计算向量间的相似度（如余弦相似度）
缺乏针对高维数据的优化索引结构
扩展性差，海量向量插入 / 查询速度骤降

横向对比主流工具的特点：

工具	开发语言	部署方式	适用场景
Chromadb	Python	轻量级单机	快速原型开发、中小规模
Faiss	C++	需编译	极致性能的超大规模
Milvus	Go	分布式集群	生产级企业应用

Chromadb 的优势在于：

纯 Python 实现，安装即用
内置持久化存储，无需额外配置
类字典风格的简洁 API

确保 Python≥3.8，安装 Chromadb：

pip install chromadb

创建客户端与集合

import chromadb

# 创建内存型客户端（生产环境建议指定持久化路径）client = chromadb.Client()

# 创建集合（类似数据库表）collection = client.create_collection(name="my_vectors")

插入向量数据

需要同时提供：
– 向量数据（list of lists）
– 唯一 ID 列表
– 可选的元数据（字典列表）

# 示例：插入 3 个 128 维向量
vectors = [[0.1*i for i in range(128)] for _ in range(3)]
ids = ["vec1", "vec2", "vec3"]
metadatas = [{"type": "image"}, {"type": "text"}, {"type": "audio"}]

collection.add(
    embeddings=vectors,
    ids=ids,
    metadatas=metadatas
)

相似性搜索

# 查询与目标向量最相似的 2 个结果
target_vector = [0.12] * 128  # 模拟查询向量
results = collection.query(query_embeddings=[target_vector],
    n_results=2
)

print(f"最相似 ID: {results['ids'][0]}")
print(f"相似度分数: {results['distances'][0]}")

实际使用时需捕获常见异常：

try:
    # 尝试查询不存在的集合
    missing = client.get_collection("ghost_collection")
except ValueError as e:
    print(f"错误捕获：{str(e)}")

当数据量增长时，注意：

批量插入

避免循环单条插入，推荐每次批量插入 100-1000 条：

# 高效做法：批量插入 10 万条数据
batch_size = 1000
for i in range(0, 100000, batch_size):
    batch_vectors = generate_vectors(batch_size)  # 你的生成函数
    batch_ids = [f"vec_{j}" for j in range(i, i+batch_size)]
    collection.add(embeddings=batch_vectors, ids=batch_ids)

查询性能
数据量>1 万时，查询延迟会线性增长
考虑对集合分片（sharding）
必要时换用 Faiss 等高性能引擎

维度不一致错误

# 错误示例：混用不同维度向量
collection.add(embeddings=[[1,2,3], [4,5]], ids=["err1", "err2"])  # 报错！

解决：插入前统一检查向量维度

未归一化的影响

相似度计算受向量模长影响，建议插入前做 L2 归一化：

import numpy as np

def normalize(vec):
    arr = np.array(vec)
    return (arr / np.linalg.norm(arr)).tolist()

normalized_vec = normalize([1,2,3])

ID 重复导致数据覆盖

# 重复 ID 会导致旧数据被静默覆盖
collection.add(embeddings=[[1,1]], ids=["dup"])
collection.add(embeddings=[[2,2]], ids=["dup"])  # 只有 [2,2] 被保留

系统集成
将 Chromadb 作为微服务部署
搭配 FastAPI 构建查询接口

from fastapi import FastAPI

app = FastAPI()

@app.get("/search")
def search(vector: list[float]):
    results = collection.query(query_embeddings=[vector], n_results=5)
    return {"matches": results["ids"][0]}