源码级别解析 · 源码深度解析 · 1.5.9版本
2026-05-18 | 每日技术深度解读
采用分层设计，关注点分离，便于扩展和维护
import chromadb
from chromadb.config import Settings
# 配置客户端
client = chromadb.Client()
# 或者使用配置
settings = Settings(anonymized_telemetry=False)
client = chromadb.Client(settings=settings)
# 创建集合
collection = client.create_collection("my_collection")
# 添加文档
collection.add(
documents=["AI is transforming industries", "Machine learning powers modern AI"],
metadatas=[{"category": "technology"}, {"category": "ai"}],
ids=["doc1", "doc2"]
)
# 查询
collection.query(
query_texts=["What is AI?"],
n_results=1
)核心 API 只有 4 个主要函数：create_collection, add, query, get
ChromaDB 采用分层架构，各层职责明确，支持多种部署模式
清晰的模块化设计，便于维护和扩展
class Client(SharedSystemClient, ClientAPI):
def __init__(
self,
tenant: Optional[str] = DEFAULT_TENANT,
database: Optional[str] = DEFAULT_DATABASE,
settings: Settings = Settings(),
) -> None:
super().__init__(settings=settings)
# 获取服务器 API 实例
self._server = self._system.instance(ServerAPI)
# 管理客户端用于验证租户和数据库
self._admin_client = AdminClient.from_system(self._system)
# 设置租户和数据库
user_identity = self.get_user_identity()
maybe_set_tenant_and_database(
user_identity,
overwrite_singleton_tenant_database_access_from_auth=
settings.chroma_overwrite_singleton_tenant_database_access_from_auth,
user_provided_tenant=tenant,
user_provided_database=database,
)
self._validate_tenant_database(tenant=self.tenant, database=self.database)Client 初始化涉及系统组件设置、认证和租户验证
企业级多租户支持，确保数据安全性和隔离性
class Tenant:
"""租户配置"""
class Database:
"""数据库配置"""
class UserIdentity:
"""用户身份验证"""
def maybe_set_tenant_and_database(
user_identity: UserIdentity,
overwrite_singleton_tenant_database_access_from_auth: bool,
user_provided_tenant: Optional[str],
user_provided_database: Optional[str]
):
"""根据用户身份和提供的参数设置租户和数据库"""
# 处理默认值、认证信息和用户覆盖
# 确保租户和数据库存在且可访问
pass
# 使用示例
client = chromadb.Client(
tenant="my-tenant",
database="my-db"
)支持动态租户设置，适应不同部署需求
SQLite 作为默认选择，轻量级且可靠
def is_in_colab() -> bool:
"""检测是否在 Colab 环境中"""
try:
import google.colab
return True
except ImportError:
return False
IN_COLAB = is_in_colab()
if not is_client:
import sqlite3
if sqlite3.sqlite_version_info < (3, 35, 0):
if IN_COLAB:
# 在 Colab 中使用 pysqlite-binary
import subprocess
import sys
subprocess.check_call([
sys.executable, "-m", "pip", "install", "pysqlite3-binary"
])
__import__("pysqlite3")
sys.modules["sqlite3"] = sys.modules.pop("pysqlite3")
else:
raise RuntimeError(
"您的系统 sqlite3 版本过低，需要 >= 3.35.0"
)智能处理不同环境下的 SQLite 版本兼容性问题
使用先进的索引算法，平衡查询速度和内存使用
HNSW 参数影响索引质量和查询性能
from chromadb.types import (
HnswIndexConfig,
VectorIndexConfig
)
# HNSW 索引配置
hnsw_config = HnswIndexConfig(
ef=128, # 搜索时的邻居数量
m=16, # 每个节点的最大连接数
ef_construction=64, # 构建时的 ef 值
distance_function="l2" # 欧几里得距离
)
# 向量索引配置
vector_config = VectorIndexConfig(
index_type="hnsw",
params=hnsw_config,
"metadata": {"dimension": 1536} # 向量维度
)灵活的索引配置，适应不同应用场景
强大的查询引擎，支持复杂的搜索场景
from chromadb.execution.expression.plan import Search
from chromadb.execution.expression.operator import (
Key, K, Knn, Rrf
)
# 构建 KNN 查询
knn_search = Search(
Knn(
vector=query_vector,
k=top_k,
space="cosine",
filter=Key("metadata.category").eq("technology")
)
)
# 混合查询
hybrid_search = Search(
Rrf([
Knn(vector=query_vector, k=10),
Key("metadata.importance").gt(5)
]),
weights=[0.7, 0.3]
)基于表达式的查询系统，支持复杂的搜索逻辑
高效的数据摄入管道，支持实时处理
class Collection:
def add(
self,
documents: Optional[Documents] = None,
embeddings: Optional[Embeddings] = None,
metadatas: Optional[Metadatas] = None,
ids: Optional[IDs] = None,
**kwargs
) -> None:
"""添加文档到集合"""
# 数据验证
self._validate_add_input(documents, embeddings, metadatas, ids)
# 向量化处理（如果未提供 embeddings）
if embeddings is None and documents is not None:
embeddings = self._embedding_function(documents)
# 批量处理
batch_size = self.settings.batch_size
total = len(ids or [])
for i in range(0, total, batch_size):
batch_end = min(i + batch_size, total)
# 构建数据段
segment = self._create_data_segment(
ids[i:batch_end],
embeddings[i:batch_end] if embeddings else None,
metadatas[i:batch_end] if metadatas else None
)
# 写入存储
self._write_to_storage(segment)支持批量处理，优化性能和内存使用
企业级安全认证，确保数据访问安全
from chromadb.auth.token_authn import TokenTransportHeader
class TokenTransportHeader:
"""基于 HTTP Header 的 Token 认证"""
def __init__(self, token: str):
self.token = token
def authenticate(self, request: httpx.Request) -> httpx.Request:
"""在请求头中添加认证信息"""
request.headers["Authorization"] = f"Bearer {self.token}"
return request
def validate(self, response: httpx.Response) -> bool:
"""验证响应认证状态"""
return response.status_code != 401
# 使用示例
token_auth = TokenTransportHeader("your-api-key")
client = chromadb.Client(auth=token_auth)简洁的 Token 认证机制，易于集成
多层缓存策略，提升系统性能
| 优化项 | 参数 | 默认值 | 说明 |
|---|---|---|---|
| 批量大小 | batch_size | 100 | 批量处理文档数量 |
| 缓存大小 | cache_size | 1000 | 缓存查询结果数量 |
| 连接超时 | timeout | 30s | 网络连接超时 |
| 索引预加载 | preload_indexes | false | 是否预加载索引 |
| 线程池 | thread_pool_size | 10 | 并发处理线程数 |
支持多种部署场景，适应不同需求
# 内存模式（开发环境）
pip install chromadb
python -c "import chromadb; client = chromadb.Client()"
# 文件存储模式（生产环境）
python -m chromadb.db.server --path /path/to/db
# Docker 部署
docker run -p 8000:8000 chromadb/chroma
# 客户端连接
python -c "
import chromadb
client = chromadb.Client(
chromadb.config.Settings(
chroma_db_impl="duckdb+parquet",
persist_directory="/path/to/db"
)
)"
"灵活的部署选项，适应不同环境和规模
完善的版本控制和迁移机制
class MigrationHandler:
"""数据库迁移处理器"""
def __init__(self, current_version: str):
self.current_version = current_version
self.migrations = self._load_migrations()
def migrate(self, target_version: str) -> None:
"""执行迁移到目标版本"""
current = self.current_version
target = target_version
while current != target:
# 找到下一个迁移
next_migration = self._find_next_migration(current, target)
if next_migration:
# 执行迁移
next_migration.migrate()
current = next_migration.to_version
else:
raise MigrationError(f"无法从 {current} 迁移到 {target}")
def rollback(self, target_version: str) -> None:
"""回滚到指定版本"""
# 执行回滚操作
pass支持数据库版本迁移，确保数据兼容性
完善的监控和日志系统，便于运维
持续创新的实验性功能，引领行业发展
# 向量搜索 + 元数据过滤
results = collection.query(
query_embeddings=[embedding],
n_results=5,
where={"category": "technology", "year": {"$gt": 2023}},
where_document={"$contains": "AI"}
)
# 多查询融合
results = collection.query(
query_texts=["What is AI?", "Machine learning basics"],
n_results=3,
rrf_weights=[0.6, 0.4] # RRf 权重
)
# 分页查询
results = collection.query(
query_texts=["AI applications"],
n_results=10,
include=["documents","metadatas","distances"],
offset=0,
limit=5 # 分页大小
)强大的混合搜索能力，满足复杂搜索需求
多语言客户端支持，便于不同技术栈集成
// 安装
npm install chromadb
// 使用
import { ChromaClient } from 'chromadb';
// 创建客户端
const client = new ChromaClient({
path: 'http://localhost:8000'
});
// 创建集合
const collection = await client.createCollection({
name: 'my_collection',
metadata: { 'description': 'My test collection' }
});
// 添加文档
await collection.add({
documents: ['AI is transformative', 'ML powers AI'],
metadatas: [{ 'category': 'tech' }, { 'category': 'ai' }],
ids: ['doc1', 'doc2']
});
// 查询
const results = await collection.query({
query_texts: ['What is AI?'],
n_results: 1
});完整的 JavaScript SDK 支持，支持浏览器和 Node.js
灵活的配置系统，适应不同部署环境
# 通过环境变量配置
export CHROMA_SERVER_HOST=0.0.0.0
export CHROMA_SERVER_PORT=8000
export CHROMA_PERSIST_DIRECTORY=/data/chroma
export CHROMA_ANNOTATIONS=True
# 通过代码配置
from chromadb.config import Settings
settings = Settings(
chroma_db_impl="duckdb+parquet",
persist_directory="/data/chroma",
anonymized_telemetry=False,
allow_reset=True,
server_host="0.0.0.0",
server_http_port=8000,
server_grpc_port=8001,
log_level="INFO"
)
client = chromadb.Client(settings=settings)多种配置方式，适应不同使用场景
支持多种数据类型，满足不同应用需求
| 数据类型 | 支持格式 | 示例 | 说明 |
|---|---|---|---|
| 稠密向量 | float32[], float64[] | [0.1, 0.2, 0.3] | 默认向量格式 |
| 稀疏向量 | CSR 格式 | {"indices": [0, 2], "values": [1.0, 2.0]} | 高维稀疏数据 |
| 文本 | string[] | ["AI technology", "Machine learning"] | 自动分词和向量化 |
| 元数据 | JSON | {"category": "tech", "year": 2024} | 过滤和排序 |
| ID | UUID, string | 550e8400-e29b-41d4-a716-446655440000 | 唯一标识符 |
完善的错误处理机制，便于调试和恢复
from chromadb.errors import (
ChromaError,
ChromaAuthError,
ChromaConnectionError,
ChromaValueError,
ChromaTimeoutError
)
def handle_chroma_errors():
"""处理 Chroma 相关错误"""
try:
collection.add(documents=["test"])
except ChromaAuthError as e:
print(f"认证错误: {e}")
except ChromaConnectionError as e:
print(f"连接错误: {e}")
except ChromaValueError as e:
print(f"参数错误: {e}")
except ChromaTimeoutError as e:
print(f"超时错误: {e}")
except ChromaError as e:
print(f"Chroma 错误: {e}")
except Exception as e:
print(f"其他错误: {e}")细致的错误分类，便于针对性处理
全面的测试框架，确保代码质量
丰富的开发工具，提升开发效率
活跃的生态系统，便于集成各种 AI 工具
# LangChain Chroma 向量存储
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings
from langchain.text_splitter import CharacterTextSplitter
from langchain.document_loaders import TextLoader
# 加载文档
loader = TextLoader("document.txt")
documents = loader.load()
# 文档分词
text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=0)
chunks = text_splitter.split_documents(documents)
# 创建向量存储
embeddings = OpenAIEmbeddings()
vectordb = Chroma.from_documents(
documents=chunks,
embedding=embeddings,
persist_directory="./chroma_db"
)
# 查询
def search(query: str) -> str:
docs = vectordb.similarity_search(query, k=3)
return "\n".join([doc.page_content for doc in docs])与主流 AI 框架深度集成
出色的性能表现，满足大规模应用需求
基于实际使用经验总结的最佳实践
持续创新的路线图，引领行业发展
ChromaDB 成为 AI 应用数据基础设施的理想选择
感谢阅读！
访问 https://atcfu.com/ai-articles/chromadb-vector-database/ 回顾本文