实战:封装一个知识库管理器

🎉摘要:基于 Chroma 客户端封装了知识库管理器 KBManager,提供幂等集合创建、集合存在判断、统计概况、重命名以及安全删除等功能,简化集合管理逻辑,让持久化知识库操作更高效便捷。

下面使用前面学过的 API 来封装一个简单的知识库管理器。该管理器将持久化知识库信息到临时目录,提供获取集合、删除集合、重命名集合、集合是否存在等功能。

具体代码如下:

import os
import tempfile
import chromadb
from datetime import datetime
from chromadb.api.models.Collection import Collection


# 让持久化示例写到临时目录,工作目录保持干净
os.chdir(tempfile.mkdtemp(prefix="chroma_demo_"))


class KBManager:
    """Chroma 知识库的集合管理封装"""

    def __init__(self, path: str = "./chroma_data"):
        self.client = chromadb.PersistentClient(path=path)

    def ensure_collection(
        self,
        name: str,
        embedding_function=None,
        space: str = "cosine",
        description: str = "",
    ) -> Collection:
        """幂等地拿到一个集合:存在就用,不存在就建"""
        return self.client.get_or_create_collection(
            name=name,
            embedding_function=embedding_function,
            configuration={
                "hnsw": {
                    "space": space
                }
            },
            metadata={
                "description": description,
                "created_at": datetime.now().isoformat(),
            },
        )

    def exists(self, name: str) -> bool:
        """判断集合是否存在"""
        return any(c.name == name for c in self.client.list_collections())

    def stats(self) -> list[dict]:
        """列出所有集合的概况"""
        return [
            {
                "name": c.name,
                "count": c.count(),
                "space": (c.configuration.get("hnsw") or {}).get("space"),
                "description": (c.metadata or {}).get("description", ""),
            }
            for c in self.client.list_collections()
        ]

    def rename(self, old: str, new: str):
        """集合重命名"""
        # 集合不存在
        if not self.exists(old):
            return
        self.client.get_collection(old).modify(name=new)

    def drop(self, name: str, force: bool = False):
        """删除集合"""
        # 集合不存在
        if not self.exists(name):
            return
        # 不强制删除,但集合中还存在数据,则抛出异常进行提示
        tmp_col = self.client.get_collection(name)
        if not force and tmp_col.count() > 0:
            raise RuntimeError(f"{name} 还有 {tmp_col.count()} 条数据,force=True 才允许删")
        self.client.delete_collection(name)


if __name__ == "__main__":
    # 创建实例
    kb = KBManager("./_demo_kb")

    # 获取 product_docs 集合实例
    col = kb.ensure_collection("product_docs", embedding_function=None, description="产品文档")

    # 向集合写入数据
    col.add(ids=["d1"], embeddings=[[1.0, 0.0]], documents=["快速上手"])
    print(kb.stats())
    # [{'name': 'product_docs', 'count': 1, 'space': 'cosine', 'description': '产品文档'}]

    # 删除集合
    kb.drop("product_docs", force=True)

封装成工具类后,在删除、重命名等操作时不用重复写集合是否存在判断等逻辑,用起来更方便。

说说我的看法
全部评论()
没有评论
关于
本网站专注于 Java、数据库(MySQL、Oracle)、Linux、软件架构及大数据等多领域技术知识分享。涵盖丰富的原创与精选技术文章,助力技术传播与交流。无论是技术新手渴望入门,还是资深开发者寻求进阶,这里都能为您提供深度见解与实用经验,让复杂编码变得轻松易懂,携手共赴技术提升新高度。如有侵权,请来信告知:hxstrive@outlook.com
其他应用
公众号