Chroma 轻量向量数据库

import chromadb
print(chromadb.__version__)  # 查看当前版本

import chromadb

# 创建持久化客户端，数据保存到指定目录
client = chromadb.PersistentClient(path="./chroma_db")

# 如果目录已存在数据，会自动加载
# 同一路径只能有一个客户端实例

import chromadb

# 连接远程Chroma服务
client = chromadb.HttpClient(
    host="localhost",
    port=8000
)

# 带认证的连接
from chromadb.utils.auth_utils import get_credentials
client = chromadb.HttpClient(
    host="localhost",
    port=8000,
    credentials=get_credentials()  # 基本认证
)

# 使用自定义headers（如API Key）
client = chromadb.HttpClient(
    host="localhost",
    port=8000,
    headers={"X-Api-Key": "your-api-key"}
)

import chromadb

# 创建纯内存客户端（临时测试用）
client = chromadb.Client()

# 也可以通过Settings创建
from chromadb.config import Settings
client = chromadb.Client(Settings(
    chroma_db_impl="chromadb.db.duckdb.DuckDB",
    persist_directory=None  # None表示纯内存
))

import chromadb

client = chromadb.PersistentClient(path="./chroma_db")

# 基本创建
collection = client.create_collection(
    name="my_documents",
    metadata={"description": "我的文档集合"}  # 可选的集合元数据
)

# 创建带嵌入函数的集合
from chromadb.utils import embedding_functions

openai_ef = embedding_functions.OpenAIEmbeddingFunction(
    api_key="your-api-key",
    model_name="text-embedding-ada-002"
)

collection = client.create_collection(
    name="openai_docs",
    embedding_function=openai_ef,
    metadata={"hnsw:space": "cosine"}  # 指定距离度量方式
)

# 指定不同的距离度量方式
# "l2" - L2距离（欧几里得距离），默认
# "cosine" - 余弦相似度
# "ip" - 内积
cosine_collection = client.create_collection(
    name="cosine_docs",
    metadata={"hnsw:space": "cosine"}
)

ip_collection = client.create_collection(
    name="ip_docs",
    metadata={"hnsw:space": "ip"}
)

# 获取已存在的集合（不存在则报错）
collection = client.get_collection(
    name="my_documents",
    embedding_function=openai_ef  # 如果集合使用自定义嵌入函数，获取时也要指定
)

# 获取或创建集合（推荐用法）
collection = client.get_or_create_collection(
    name="my_documents",
    metadata={"hnsw:space": "cosine"}
)

# 列出所有集合
collections = client.list_collections()
for col in collections:
    print(f"集合名: {col.name}, ID: {col.id}")

# 删除集合
client.delete_collection(name="my_documents")

# 重置整个客户端数据（慎用！）
client.reset()  # 清除所有数据

collection = client.get_or_create_collection(name="docs")

# 方式1：手动提供向量
collection.add(
    ids=["doc1", "doc2", "doc3"],             # 唯一标识符，必填
    embeddings=[                               # 向量列表，与ids一一对应
        [0.1, 0.2, 0.3, ...],
        [0.4, 0.5, 0.6, ...],
        [0.7, 0.8, 0.9, ...]
    ],
    documents=["文档1的内容", "文档2的内容", "文档3的内容"],  # 可选：原始文本
    metadatas=[                                # 可选：元数据
        {"source": "web", "page": 1},
        {"source": "pdf", "page": 5},
        {"source": "web", "page": 3}
    ]
)

# 方式2：使用嵌入函数自动生成向量（需在创建集合时指定embedding_function）
collection.add(
    ids=["doc4", "doc5"],
    documents=["这是新的文档内容", "另一个文档"],
    metadatas=[
        {"source": "api", "category": "tech"},
        {"source": "api", "category": "science"}
    ]
    # 无需提供embeddings，自动使用集合的嵌入函数生成
)

# 基本查询：根据查询向量找最近的文档
results = collection.query(
    query_embeddings=[[0.1, 0.2, 0.3, ...]],  # 查询向量
    n_results=5                                 # 返回最相似的5个结果
)

# 使用文本查询（需集合有嵌入函数）
results = collection.query(
    query_texts=["什么是机器学习？"],  # 自动转换为向量
    n_results=5
)

# 带元数据过滤的查询
results = collection.query(
    query_texts=["技术相关内容"],
    n_results=5,
    where={"source": "web"}  # 只在source为web的文档中搜索
)

# 复杂过滤条件
results = collection.query(
    query_texts=["技术相关内容"],
    n_results=5,
    where={
        "$and": [
            {"source": "web"},
            {"page": {"$gt": 2}}
        ]
    }
)

# 指定返回内容
results = collection.query(
    query_texts=["查询文本"],
    n_results=5,
    include=["documents", "metadatas", "distances", "embeddings"]
)

# query返回一个字典，结构如下：
{
    'ids': [['doc1', 'doc3', 'doc5']],           # 匹配的文档ID
    'distances': [[0.12, 0.25, 0.38]],           # 距离值（越小越相似）
    'documents': [['文档1内容', '文档3内容', '文档5内容']],  # 原始文本
    'metadatas': [[{...}, {...}, {...}]],         # 元数据
    'embeddings': [[...], [...], [...]]           # 向量（仅include时返回）
}

# 更新已有文档（id必须存在，否则报错）
collection.update(
    ids=["doc1", "doc2"],
    documents=["更新后的文档1内容", "更新后的文档2内容"],
    metadatas=[
        {"source": "updated", "page": 10},
        {"source": "updated", "page": 11}
    ],
    embeddings=[[0.2, 0.3, 0.4, ...], [0.5, 0.6, 0.7, ...]]
)

# 如果id不存在则添加，存在则更新（最安全的写入方式）
collection.upsert(
    ids=["doc1", "doc6"],
    documents=["更新doc1", "新增doc6"],
    metadatas=[
        {"source": "upsert", "version": 2},
        {"source": "upsert", "version": 1}
    ]
)

# 按ID删除
collection.delete(ids=["doc1", "doc2"])

# 按条件删除
collection.delete(
    where={"source": "web"}  # 删除所有source为web的文档
)

# 按文档内容删除
collection.delete(
    where_document={"$contains": "过期内容"}
)

# 删除集合中所有数据
collection.delete(
    where={}  # 空条件匹配所有
)

# 按ID获取
results = collection.get(ids=["doc1", "doc3"])

# 按条件获取
results = collection.get(
    where={"source": "web"},
    include=["documents", "metadatas"]
)

# 获取所有数据
results = collection.get(
    include=["documents", "metadatas", "embeddings"]
)

# 限制返回数量
results = collection.get(
    limit=100,
    offset=0
)

# 按文档内容过滤
results = collection.get(
    where_document={"$contains": "机器学习"}
)

from chromadb.utils import embedding_functions

# 使用默认模型（all-MiniLM-L6-v2）
default_ef = embedding_functions.SentenceTransformerEmbeddingFunction()

# 指定模型
ef = embedding_functions.SentenceTransformerEmbeddingFunction(
    model_name="paraphrase-multilingual-MiniLM-L12-v2",  # 多语言模型
    device="cpu",            # 运行设备：cpu或cuda
    normalize_embeddings=True  # 是否归一化向量
)

# 创建集合时使用
collection = client.create_collection(
    name="multilingual_docs",
    embedding_function=ef
)

from chromadb.utils import embedding_functions

openai_ef = embedding_functions.OpenAIEmbeddingFunction(
    api_key="sk-your-api-key",
    model_name="text-embedding-ada-002",  # 或 text-embedding-3-small/large
    organization_id="org-xxx",  # 可选
    api_base="https://api.openai.com/v1",  # 可选，支持自定义端点
    api_type="open_ai"  # 可选：open_ai, azure, azure_ad
)

# Azure OpenAI
azure_ef = embedding_functions.OpenAIEmbeddingFunction(
    api_key="your-azure-key",
    api_base="https://your-resource.openai.azure.com",
    api_type="azure",
    deployment_id="your-deployment-id"
)

from chromadb.utils import embedding_functions

hf_ef = embedding_functions.HuggingFaceEmbeddingFunction(
    api_key="hf_your-token",
    model_name="sentence-transformers/all-MiniLM-L6-v2"
)

from chromadb.utils import embedding_functions

google_ef = embedding_functions.GoogleGenerativeAIEmbeddingFunction(
    api_key="your-google-api-key",
    model_name="models/embedding-001"
)

from chromadb.utils import embedding_functions

cohere_ef = embedding_functions.CohereEmbeddingFunction(
    api_key="your-cohere-key",
    model_name="embed-english-v3.0",
    input_type="search_document"  # search_document, search_query, classification, clustering
)

# Chroma内置的ONNX格式MiniLM模型，无需额外依赖
from chromadb.utils import embedding_functions

# 这是Chroma的默认嵌入函数
default_ef = embedding_functions.ONNXMiniLM_L6_V2()

import chromadb

# === 方式1：PersistentClient（推荐） ===
# 创建时指定路径，所有操作自动持久化
client = chromadb.PersistentClient(path="./chroma_db")

# 创建集合和添加数据
collection = client.get_or_create_collection(name="persisted_docs")
collection.add(
    ids=["doc1"],
    documents=["持久化存储的文档"],
    metadatas=[{"source": "test"}]
)

# 程序重启后，再次连接同一目录即可恢复数据
client2 = chromadb.PersistentClient(path="./chroma_db")
collection2 = client2.get_collection(name="persisted_docs")
print(collection2.count())  # 输出: 1

# === 方式2：备份与迁移 ===
# 查看集合信息
print(collection.name)
print(collection.count())
print(collection.metadata)

# 导出数据
all_data = collection.get(include=["documents", "metadatas", "embeddings"])
# 可以将all_data序列化保存到文件
import json
with open("backup.json", "w") as f:
    json.dump(all_data, f)

# 等于
results = collection.query(
    query_texts=["查询"],
    where={"category": "tech"}
)

# 不等于
results = collection.query(
    query_texts=["查询"],
    where={"category": {"$ne": "deprecated"}}
)

# 大于/大于等于
results = collection.query(
    query_texts=["查询"],
    where={"year": {"$gt": 2023}}
)
results = collection.query(
    query_texts=["查询"],
    where={"score": {"$gte": 0.8}}
)

# 小于/小于等于
results = collection.query(
    query_texts=["查询"],
    where={"price": {"$lt": 100}}
)

# 不支持数值类型的元数据需在添加时确保类型正确
# Chroma的元数据值支持：str, int, float, bool

# AND逻辑
results = collection.query(
    query_texts=["查询"],
    where={
        "$and": [
            {"category": "tech"},
            {"year": {"$gte": 2023}}
        ]
    }
)

# OR逻辑
results = collection.query(
    query_texts=["查询"],
    where={
        "$or": [
            {"category": "tech"},
            {"category": "science"}
        ]
    }
)

# 嵌套逻辑
results = collection.query(
    query_texts=["查询"],
    where={
        "$and": [
            {"status": "published"},
            {"$or": [
                {"category": "tech"},
                {"category": "science"}
            ]}
        ]
    }
)

# 包含指定文本
results = collection.query(
    query_texts=["查询"],
    where_document={"$contains": "机器学习"}
)

# 不包含指定文本
results = collection.query(
    query_texts=["查询"],
    where_document={"$not_contains": "deprecated"}
)

# 组合使用where和where_document
results = collection.query(
    query_texts=["查询"],
    where={"category": "tech"},
    where_document={"$contains": "深度学习"}
)

import chromadb
from chromadb.utils import embedding_functions

# === 1. 初始化 ===
client = chromadb.PersistentClient(path="./rag_db")

# 使用多语言嵌入模型
ef = embedding_functions.SentenceTransformerEmbeddingFunction(
    model_name="paraphrase-multilingual-MiniLM-L12-v2"
)

collection = client.get_or_create_collection(
    name="knowledge_base",
    embedding_function=ef,
    metadata={"hnsw:space": "cosine"}
)

# === 2. 文档分块与索引 ===
def chunk_text(text, chunk_size=500, overlap=50):
    """简单的文本分块函数"""
    chunks = []
    start = 0
    while start < len(text):
        end = start + chunk_size
        chunks.append(text[start:end])
        start += chunk_size - overlap
    return chunks

# 模拟文档数据
documents = [
    {"id": "doc1", "text": "Python是一种广泛使用的高级编程语言...", "source": "python_intro.pdf"},
    {"id": "doc2", "text": "机器学习是人工智能的一个分支，它使计算机系统能够从数据中学习...", "source": "ml_basics.pdf"},
    {"id": "doc3", "text": "深度学习是机器学习的子集，使用多层神经网络来建模数据中的复杂模式...", "source": "dl_intro.pdf"},
]

# 分块并添加到集合
all_ids = []
all_docs = []
all_metadatas = []

for doc in documents:
    chunks = chunk_text(doc["text"])
    for i, chunk in enumerate(chunks):
        all_ids.append(f"{doc['id']}_chunk{i}")
        all_docs.append(chunk)
        all_metadatas.append({
            "source": doc["source"],
            "chunk_index": i,
            "total_chunks": len(chunks)
        })

collection.upsert(
    ids=all_ids,
    documents=all_docs,
    metadatas=all_metadatas
)

# === 3. 检索相关文档 ===
def retrieve_context(query, n_results=3):
    """检索与查询最相关的文档片段"""
    results = collection.query(
        query_texts=[query],
        n_results=n_results,
        include=["documents", "metadatas", "distances"]
    )

    contexts = []
    for doc, meta, dist in zip(
        results['documents'][0],
        results['metadatas'][0],
        results['distances'][0]
    ):
        contexts.append({
            "text": doc,
            "source": meta["source"],
            "distance": dist
        })
    return contexts

# === 4. 构建RAG问答 ===
def rag_query(query, llm_generate_func):
    """RAG查询流程"""
    # 步骤1：检索相关上下文
    contexts = retrieve_context(query, n_results=3)

    # 步骤2：构建提示词
    context_text = "\n\n".join([c["text"] for c in contexts])
    prompt = f"""基于以下参考资料回答问题。如果资料中没有相关信息，请说明。

参考资料：
{context_text}

问题：{query}
回答："""

    # 步骤3：调用LLM生成回答
    answer = llm_generate_func(prompt)
    return answer, contexts

# 使用示例
query = "什么是深度学习？"
# answer, sources = rag_query(query, your_llm_function)

import chromadb
from chromadb.utils import embedding_functions
from datetime import datetime

client = chromadb.PersistentClient(path="./memory_db")
ef = embedding_functions.SentenceTransformerEmbeddingFunction()

collection = client.get_or_create_collection(
    name="conversation_memory",
    embedding_function=ef,
    metadata={"hnsw:space": "cosine"}
)

def save_message(role, content, session_id, user_id="user1"):
    """保存对话消息到向量存储"""
    timestamp = datetime.now().isoformat()
    message_id = f"{user_id}_{session_id}_{timestamp}"

    collection.upsert(
        ids=[message_id],
        documents=[content],
        metadatas=[{
            "role": role,           # user 或 assistant
            "session_id": session_id,
            "user_id": user_id,
            "timestamp": timestamp
        }]
    )

def recall_relevant(query, n_results=5, user_id="user1"):
    """检索与当前查询相关的历史对话"""
    results = collection.query(
        query_texts=[query],
        n_results=n_results,
        where={"user_id": user_id},
        include=["documents", "metadatas", "distances"]
    )

    memories = []
    for doc, meta in zip(results['documents'][0], results['metadatas'][0]):
        memories.append({
            "content": doc,
            "role": meta["role"],
            "timestamp": meta["timestamp"]
        })
    return memories

# 使用示例
session_id = "session_001"
save_message("user", "我想学习Python编程", session_id)
save_message("assistant", "Python是一门非常适合初学者的语言...", session_id)
save_message("user", "推荐一些Python学习资源", session_id)

# 新对话中回忆相关上下文
relevant = recall_relevant("Python入门教程推荐")

import chromadb
from chromadb.utils import embedding_functions

client = chromadb.PersistentClient(path="./fewshot_db")
ef = embedding_functions.SentenceTransformerEmbeddingFunction()

collection = client.get_or_create_collection(
    name="few_shot_examples",
    embedding_function=ef
)

# 添加示例
examples = [
    {
        "id": "ex1",
        "input": "把这段话翻译成英文",
        "output": "Translate this passage into English.",
        "category": "translation"
    },
    {
        "id": "ex2",
        "input": "总结这篇文章的要点",
        "output": "Summarize the key points of this article.",
        "category": "translation"
    },
    {
        "id": "ex3",
        "input": "写一首关于春天的诗",
        "output": "春风拂面花正开\n桃红柳绿入梦来...",
        "category": "creative"
    }
]

collection.upsert(
    ids=[e["id"] for e in examples],
    documents=[e["input"] for e in examples],
    metadatas=[{"output": e["output"], "category": e["category"]} for e in examples]
)

# 根据新输入检索最相关的示例
def get_few_shot_examples(query, n_results=2, category=None):
    """检索与查询最相关的Few-Shot示例"""
    where_filter = {}
    if category:
        where_filter["category"] = category

    results = collection.query(
        query_texts=[query],
        n_results=n_results,
        where=where_filter if where_filter else None,
        include=["documents", "metadatas"]
    )

    examples = []
    for doc, meta in zip(results['documents'][0], results['metadatas'][0]):
        examples.append({
            "input": doc,
            "output": meta["output"]
        })
    return examples

# 使用示例
examples = get_few_shot_examples("把这句话翻成日语")
# 返回最相关的翻译示例

from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import SentenceTransformerEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter

# 创建嵌入函数
embeddings = SentenceTransformerEmbeddings(
    model_name="all-MiniLM-L6-v2"
)

# 文本分割
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200
)

# 从文档创建向量存储
texts = text_splitter.split_text(your_document_text)
vectorstore = Chroma.from_texts(
    texts=texts,
    embedding=embeddings,
    collection_name="langchain_docs",
    persist_directory="./langchain_chroma_db"
)

# 检索
results = vectorstore.similarity_search(
    query="什么是RAG？",
    k=3
)

# 创建检索器
retriever = vectorstore.as_retriever(
    search_type="mmr",         # 最大边际相关性
    search_kwargs={"k": 3, "fetch_k": 10}
)

# 与RAG链结合
from langchain.chains import RetrievalQA
from langchain_community.llms import OpenAI

qa_chain = RetrievalQA.from_chain_type(
    llm=OpenAI(),
    chain_type="stuff",
    retriever=retriever
)

answer = qa_chain.run("什么是RAG？")

import numpy as np

def cosine_similarity(a, b):
    """计算余弦相似度"""
    dot_product = np.dot(a, b)
    norm_a = np.linalg.norm(a)
    norm_b = np.linalg.norm(b)
    return dot_product / (norm_a * norm_b)

# 示例
a = np.array([1.0, 2.0, 3.0])
b = np.array([2.0, 4.0, 6.0])  # 与a方向相同
c = np.array([1.0, 0.0, 0.0])  # 不同方向

print(cosine_similarity(a, b))  # 1.0（完全相同方向）
print(cosine_similarity(a, c))  # 约0.27（部分相似）

import numpy as np

def l2_distance(a, b):
    """计算L2距离"""
    return np.linalg.norm(a - b)

# 示例
a = np.array([1.0, 2.0, 3.0])
b = np.array([1.0, 2.0, 3.0])
c = np.array([4.0, 5.0, 6.0])

print(l2_distance(a, b))  # 0.0（完全相同）
print(l2_distance(a, c))  # 约5.20（差异较大）

import numpy as np

def inner_product(a, b):
    """计算内积"""
    return np.dot(a, b)

# 归一化后的内积等于余弦相似度
a = np.array([1.0, 2.0, 3.0])
a_normalized = a / np.linalg.norm(a)
b = np.array([2.0, 4.0, 6.0])
b_normalized = b / np.linalg.norm(b)

print(inner_product(a_normalized, b_normalized))  # 1.0（归一化后方向相同）

# 通过metadata设置HNSW参数
collection = client.create_collection(
    name="tuned_collection",
    metadata={
        "hnsw:space": "cosine",        # 距离度量
        "hnsw:M": 16,                   # 每个节点的最大连接数，默认16
        "hnsw:construction_ef": 100,    # 构建时的搜索宽度，默认100
        "hnsw:batch_size": 100          # 构建索引的批大小
    }
)

# Chroma的嵌入函数遵循统一接口
class EmbeddingFunction(Protocol):
    def __call__(self, input: Documents) -> Embeddings:
        """将文档列表转换为向量列表"""
        ...

# 所有内置嵌入函数都实现此接口
# 在add/query时自动调用
# 也可以手动调用
ef = embedding_functions.SentenceTransformerEmbeddingFunction()
vectors = ef(["文本1", "文本2"])  # 返回二维数组

# ❌ 错误：重复ID会报错
collection.add(ids=["doc1"], documents=["内容1"])
collection.add(ids=["doc1"], documents=["内容2"])  # 报错！ID已存在

# ✅ 正确：使用upsert避免重复ID问题
collection.upsert(ids=["doc1"], documents=["内容1"])
collection.upsert(ids=["doc1"], documents=["内容2"])  # 更新而非报错

# ✅ 正确：批量操作提高效率
collection.upsert(
    ids=[f"doc{i}" for i in range(1000)],
    documents=[f"文档内容{i}" for i in range(1000)]
)
# 而非循环调用1000次

# ⚠️ 关键：查询时使用的嵌入函数必须与存储时一致

# ✅ 正确：创建集合时指定嵌入函数，后续操作自动使用
collection = client.create_collection(
    name="consistent_ef",
    embedding_function=same_ef  # 存储和查询使用同一个
)

# ❌ 错误：存储用模型A，查询用模型B
# 向量空间不一致，搜索结果毫无意义

# ✅ 正确：元数据用于过滤，设计时考虑查询需求
collection.add(
    ids=["doc1"],
    documents=["文档内容"],
    metadatas=[{
        "source": "pdf",           # 字符串：用于精确匹配
        "page": 42,                # 整数：用于范围查询
        "timestamp": 1700000000,   # 数字时间戳：用于时间范围
        "is_processed": True,      # 布尔：用于状态过滤
        "tags": "tech,ai,ml"       # 用逗号分隔模拟多标签
    }]
)

# ⚠️ 注意：元数据值类型必须一致
# 如果某个字段有时是字符串有时是数字，查询时可能出问题
# ⚠️ Chroma不支持列表类型的元数据值

# 1. 批量操作
# ✅ 一次添加多条记录
collection.upsert(ids=ids, documents=docs, metadatas=metas)

# ❌ 避免逐条添加
# for id, doc, meta in zip(ids, docs, metas):
#     collection.upsert(ids=[id], documents=[doc], metadatas=[meta])

# 2. 选择合适的距离度量
# 文本语义搜索推荐cosine，默认是l2
collection = client.create_collection(
    name="cosine_collection",
    metadata={"hnsw:space": "cosine"}
)

# 3. 合理设置n_results
# 不要设置过大的n_results，会降低性能
results = collection.query(query_texts=["查询"], n_results=5)  # 足够

# 4. 利用元数据预过滤减少搜索空间
results = collection.query(
    query_texts=["查询"],
    where={"category": "tech"},  # 先过滤再搜索
    n_results=5
)

# 5. 调整HNSW参数
# 更大的M和ef_construct提高搜索精度但增加内存和构建时间
collection = client.create_collection(
    name="high_accuracy",
    metadata={
        "hnsw:M": 32,                 # 增大连接数
        "hnsw:construction_ef": 200,  # 增大构建搜索宽度
    }
)

# ⚠️ 注意：同一持久化路径只能有一个客户端实例
# ❌ 错误
client1 = chromadb.PersistentClient(path="./db")
client2 = chromadb.PersistentClient(path="./db")  # 可能导致数据损坏

# ✅ 正确：复用同一个客户端实例
client = chromadb.PersistentClient(path="./db")

# 如果需要并发访问，使用HttpClient连接独立服务
# 启动Chroma服务端后
client = chromadb.HttpClient(host="localhost", port=8000)

import chromadb
from chromadb.errors import InvalidCollectionException, InvalidArgumentError

# 处理集合不存在的错误
try:
    collection = client.get_collection(name="nonexistent")
except ValueError as e:
    print(f"集合不存在: {e}")
    # 改用get_or_create_collection
    collection = client.get_or_create_collection(name="nonexistent")

# 处理ID重复错误
try:
    collection.add(ids=["existing_id"], documents=["内容"])
except InvalidArgumentError:
    collection.upsert(ids=["existing_id"], documents=["内容"])

# 检查集合是否存在
collections = client.list_collections()
existing_names = [c.name for c in collections]
if "my_collection" in existing_names:
    collection = client.get_collection(name="my_collection")
else:
    collection = client.create_collection(name="my_collection")

# 获取集合中记录数量
count = collection.count()
print(f"集合中有 {count} 条记录")

# 获取集合的元信息
print(collection.name)       # 集合名称
print(collection.id)         # 集合UUID
print(collection.metadata)   # 集合元数据

# 修改集合元数据
collection.modify(metadata={"description": "更新描述"})

# 修改集合名称
collection.modify(name="new_name")