cyborginc/cyborgdb-py

# CyborgDB Python SDK    **CyborgDB Python SDK** 提供了一个功能全面的客户端库，用于与首个保密向量数据库 [CyborgDB](https://docs.cyborg.co) 进行交互。该 SDK 让你能够执行包括数据摄取、搜索和检索在内的加密向量操作，同时维持向量 embedding 的端到端加密。专为 Python 应用程序打造，可无缝集成到现代的 Python 应用程序和服务中。 此 SDK 提供了与 [`cyborgdb-service`](https://pypi.org/project/cyborgdb-service/) 的接口，你需要单独安装并运行该服务才能使用此 SDK。欲了解更多信息，请查阅我们的[文档](https://docs.cyborg.co)。 ## 主要功能 - **端到端加密**：所有向量操作均使用客户端密钥保持加密状态 - **零信任设计**： novel 架构确保机密推理数据的安全 - **高性能**：支持 CUDA 的 GPU 加速索引和检索 - **熟悉的 API**：轻松集成到现有的 AI 工作流中 - **加密的 DiskIVF 索引**：基于磁盘的倒排文件索引，支持自定义训练参数 ## 快速入门 要在几分钟内快速入门，请查看我们的[快速入门指南](https://docs.cyborg.co/quickstart)。 ### 安装说明 1. 安装 `cyborgdb-service` ``` # Install the CyborgDB Service pip install cyborgdb-service # Or via Docker docker pull cyborginc/cyborgdb-service ``` 2. 安装 `cyborgdb` SDK： ``` # Install the CyborgDB Python SDK pip install cyborgdb ``` ### 用法 ``` from cyborgdb import Client # Initialize the client client = Client('https://localhost:8000', 'your-api-key') # Generate a 32-byte encryption key index_key = client.generate_key() # Create an encrypted index index = client.create_index( index_name='my-index', index_key=index_key ) # Add encrypted vector items items = [ { 'id': 'doc1', 'vector': [0.1] * 128, # Replace with real embeddings 'contents': 'Hello world!', 'metadata': {'category': 'greeting', 'language': 'en'} }, { 'id': 'doc2', 'vector': [0.1] * 128, # Replace with real embeddings 'contents': 'Bonjour le monde!', 'metadata': {'category': 'greeting', 'language': 'fr'} } ] index.upsert(items) # Query the encrypted index query_vector = [0.2] * 128 # 128 dimensions results = index.query(query_vectors=query_vector,top_k=5) # Print the results for result in results: print(f"ID: {result['id']}, Distance: {result['distance']}") ``` ### 高级用法 #### 批量查询 ``` # Search with multiple query vectors simultaneously query_vectors = [ [0.1] * 128, [0.2] * 128 ] batch_results = index.query(query_vectors=query_vectors, top_k=5) # Print the results (batch queries return list of lists) for i, query_results in enumerate(batch_results): print(f"\nResults for query {i}:") for result in query_results: print(f" ID: {result['id']}, Distance: {result['distance']}") ``` #### 元数据过滤 ``` # Search with metadata filters query_vector = [0.1] * 128 results = index.query( query_vectors=query_vector, top_k=10, n_probes=1, greedy=False, filters={'category': 'greeting', 'language': 'en'}, include=['distance', 'metadata', 'contents'] ) # Print the results for result in results: print(f"ID: {result['id']}, Distance: {result['distance']}, Metadata: {result['metadata']}") ``` #### 通过 KMS 自带密钥 (BYOK) 当服务配置了 `kms.registry` 条目时，SDK 可以将密钥管理完全委托给服务器端的 KMS。服务会生成数据加密密钥，在指定的 KMS 插槽下对其进行封装，并持久化存储该 envelope —— SDK 绝不会看到或持有该密钥。 ``` # Create a KMS-backed index — no index_key from the SDK side. # 'vendor-kms-slot' must match an entry in the service's cyborgdb.yaml. index = client.create_index( index_name='kms-backed-index', kms_name='vendor-kms-slot', dimension=128, metric='euclidean', ) # Reopening the index later doesn't require a key either; the service # resolves the data key from the index's stored KMS envelope. loaded = client.load_index(index_name='kms-backed-index') loaded.upsert(items) ``` 或者，SDK 也可以自行提供密钥 —— 传入 `index_key` 并省略 `kms_name`。这是无 KMS 路径，服务会在内部将其记录为 `provider: none`： ``` index = client.create_index( index_name='sdk-keyed-index', index_key=index_key, dimension=128, ) ``` 请提供 `index_key` / `kms_name` 中的**确切一个** —— 同时传入两者会被服务拒绝并返回 400 错误，因为指定的插槽已经决定了密钥来源。 #### 基于角色的访问控制 (RBAC) 当服务在设置了 root admin key (`CYBORGDB_ROOT_API_KEY`) 的情况下运行时，即启用了 RBAC。root 可以生成限定于单个索引的**每用户 API 密钥**，每个密钥都拥有一组 `read` / `write` 权限。权限由服务以*加密方式*强制执行：为用户存在的已封装数据加密密钥**构成了**他们的权限集，因此只读用户根本无法为写操作进行解密，而撤销用户则会清除其密钥。 ``` # Admin (root) client: mint users on an existing index. admin = cyborgdb.Client(base_url, api_key=ROOT_API_KEY) index = admin.load_index(index_name='kms-backed-index') # KMS-backed (see BYOK) reader = index.create_user(permissions=['read']) writer = index.create_user(permissions=['read', 'write']) # Each returns {'user_id': '', 'api_key': 'cdbk_...'} — the api_key is # shown ONCE and never stored by the service. Hand it to the user securely. index.list_users() # [{'user_id': ..., 'permissions': [...]}, ...] index.delete_user(reader['user_id']) # revoke; the key stops working immediately ``` 用户使用其 `cdbk_` 密钥进行身份验证，并且不需要拥有自己的索引密钥 —— 他们只需按名称加载索引，服务就会解析其密钥： ``` user = cyborgdb.Client(base_url, api_key=reader['api_key']) idx = user.load_index(index_name='kms-backed-index') # no index_key idx.query(query_vectors=[...], top_k=5) # allowed for 'read' idx.upsert(items) # raises for read-only users ``` ## 文档 有关 CyborgDB 的更多信息，请参见 [Cyborg 文档](https://docs.cyborg.co)。 ## 许可证 CyborgDB Python SDK 基于 MIT 许可证授权。

标签：JSONLines, Python, Vectored Exception Handling, 向量数据库, 客户端SDK, 数据加密, 无后门, 机密计算, 请求拦截, 逆向工具, 零信任