Chroma 四层结构

Chroma 的数据组织是四层：

Tenant（租户）
 └── Database（数据库）
      └── Collection（集合）
           └── Record（记录）

其中：

• Tenant：最外层的隔离单位。单机版几乎不用管，默认是 default_tenant。多租户 SaaS 才会用到。

• Database：一个租户下可以有多个库，默认 default_database。

• Collection：真正干活的单位。一个集合 = 一批同构的向量 + 一套索引配置 + 一个嵌入函数。类比关系库里的一张“表”。

• Record：集合里的一条数据，类比表里的一"行"。

注意：本地持久化 PersistentClient 默认只有 default_tenant、default_database；只有 HTTP 服务端模式（chroma run）才支持自定义多 Tenant / 多 Database。

客户端初始化时可以指定用哪个租户哪个库，例如：

import chromadb
import os
import tempfile

# 持久化示例写到临时目录，保证工作目录保持干净
# tempfile.mkdtemp(prefix="chroma_demo_") 用于创建一个临时目录，返回路径
# os.chdir() 用于切换
os.chdir(tempfile.mkdtemp(prefix="chroma_demo_"))

client = chromadb.PersistentClient(
    path="./chroma_data",
    tenant="default_tenant", # 指定租户
    database="default_database", # 指定数据库
)

demo = client.get_or_create_collection("demo")
demo.add(ids=["q1", "q2"],
    documents=[
        "订单付款后多久发货？一般 48 小时内出库",
        "如何申请退款？在订单详情页点击申请退款",
    ],
    metadatas=[
        {"category": "物流"},
        {"category": "售后"},
    ],)
print(demo.query(query_texts=["申请退款"], n_results=1))
# 输出：
# {'ids': [['q2']], 'embeddings': None, 'documents': [['如何申请退款？在订单详情页点击申请退款']], 'uris': None,
# 'included': ['metadatas', 'documents', 'distances'], 'data': None, 'metadatas': [[{'category': '售后'}]],
# 'distances': [[0.7056398391723633]]}

绝大多数情况下这两个参数你都不用动。需要多租户隔离时，直接用 AdminClient 创建。

AdminClient 分两种形态：

（1）不带 host/port 参数，使用嵌入式（in‑process）本地系统客户端，直接调用进程内 Rust 后端，不走 TCP 网络，不需要外部服务。例如：

import chromadb

admin = chromadb.AdminClient()
admin.create_tenant("tenant_a") # 创建租户
admin.create_database("kb_prod", tenant="tenant_a") # 创建数据库
print(admin.list_databases(tenant="tenant_a"))
# 输出：
# [{'id': UUID('4226f4ea-7ef3-469a-ba74-ecf64d6708fb'), 'name': 'kb_prod', 'tenant': 'tenant_a'}]

（2）传了 host="localhost", port=8000 参数，使用 HTTP 远程管理员客户端，走网络，必须 chroma run 启动服务。例如：

import chromadb
from chromadb.config import Settings

settings = Settings(
    chroma_api_impl="chromadb.api.fastapi.FastAPI",
    chroma_server_host="localhost",
    chroma_server_http_port=8000,
)

# 远程管理员客户端，走HTTP调用 chroma run 服务端
admin = chromadb.AdminClient(settings=settings)

# 下面API和嵌入式完全一样
admin.create_tenant("tenant_a")
admin.create_database("kb_prod", tenant="tenant_a")
print(admin.list_databases(tenant="tenant_a"))

运行过程：

（1） 先启动 chroma 服务，例如：

root@localhost:~# docker start 3b777707ec23
3b777707ec23
root@localhost:~# docker ps
CONTAINER ID   IMAGE             COMMAND                  CREATED      STATUS          PORTS                                         NAMES
3b777707ec23   chromadb/chroma   "dumb-init -- chroma…"   3 days ago   Up 51 seconds   0.0.0.0:8000->8000/tcp, [::]:8000->8000/tcp   reverent_lamport

（2）再运行上面代码，输出如下：

[{'id': '21963bf5-3572-4a54-9aff-e659684ede0f', 'name': 'kb_prod', 'tenant': 'tenant_a'}]

如果租户已经存在，也会出现错误，如下：

chromadb.errors.ChromaError: Tenant [tenant_a] already exists

如果没有启动 chroma 服务，运行上面代码，将输出如下错误：

httpx.ConnectError: [WinError 10061] 由于目标计算机积极拒绝，无法连接。

通常，创建租户和数据的标准写法是：先尝试 get，捕获 NotFound 异常，不存在就创建。例如：

import chromadb
from chromadb.config import Settings
from chromadb.errors import NotFoundError

# 初始化AdminClient
settings = Settings(
    chroma_api_impl="chromadb.api.fastapi.FastAPI",
    chroma_server_host="localhost",
    chroma_server_http_port=8000,
)
admin = chromadb.AdminClient(settings=settings)


def get_or_create_tenant(admin_client, tenant_name: str):
    """获取租户，如果租户不存在，则创建租户"""
    try:
        t = admin_client.get_tenant(tenant_name)
        print(f"租户已存在: {tenant_name}")
        return t
    except NotFoundError:
        print(f"创建租户: {tenant_name}")
        admin_client.create_tenant(tenant_name)
        return admin_client.get_tenant(tenant_name)


def get_or_create_database(admin_client, db_name: str, tenant_name: str):
    """获取数据库，如果不存在，则创建数据库"""
    try:
        database = admin_client.get_database(db_name, tenant=tenant_name)
        print(f"数据库已存在: tenant={tenant_name}, db={db_name}")
        return database
    except NotFoundError:
        print(f"创建数据库: tenant={tenant_name}, db={db_name}")
        admin_client.create_database(db_name, tenant=tenant_name)
        return admin_client.get_database(db_name, tenant=tenant_name)


if __name__ == "__main__":
    tenant = get_or_create_tenant(admin, "tenant_a")
    db = get_or_create_database(admin, "kb_prod", "tenant_a")

    print("\ntenant信息：", tenant)
    print("database信息：", db)

运行示例，输出如下：

租户已存在: tenant_a
数据库已存在: tenant=tenant_a, db=kb_prod

tenant信息： {'name': 'tenant_a'}
database信息： {'id': '762477b3-2def-4d64-901e-2a2395b1a006', 'name': 'kb_prod', 'tenant': 'tenant_a'}

踩坑：集合之间是完全隔离的，不能跨集合检索，也没有事务能跨集合。你想要“一次搜多个知识库”，只能自己查多次再合并，或者干脆建一个大集合用 metadata 里的 kb_id 区分（我更推荐后者，第 6 章会讲）。