RAG落地利器Weaviate向量数据库部署使用教程

掌握 Weaviate，高效实现向量搜索与 AI 应用落地。
核心内容：
1. Weaviate 向量数据库简介及功能特性
2. Weaviate 的安装步骤与 Docker 部署指南
3. Weaviate 的使用与配置优化建议

Weaviate 部署

1. 简介

简单来说，Weaviate 是一款开源的向量搜索引擎。其优势在于：你可以用类似 JSON 的类属性方式存储文档，同时把机器学习生成的向量附加到这些文档上，让它们在向量空间里各自安家。这样一来，语义搜索、问答提取、分类等任务都能交给它处理。更值得一提的是，通过 GraphQL-API 就能轻松访问数据，完全不用操心底层实现。

官网地址：https://weaviate.io/

2. 安装 Weaviate

先从 Docker Hub 把 Weaviate 的最新镜像拉下来：

docker pull semitechnologies/weaviate:latest

如果拉取速度比较慢，可以尝试更换镜像源——毕竟国内的网络环境大家都有体会。

2.3 运行 Weaviate 容器

接下来看看具体的运行命令——这里已经配好了认证和授权，方便你直接上手：

docker run -d --name weaviate \
    --restart=always \
    -p 8080:8080 \
    -p 50051:50051 \
    -e "AUTHENTICATION_APIKEY_ENABLED=true" \
    -e "AUTHENTICATION_APIKEY_ALLOWED_KEYS=test-secret-key,test2-secret-key" \
    -e "AUTHENTICATION_APIKEY_USERS=test@2024.com,test2@2024.com" \
    -e "AUTHORIZATION_ADMINLIST_ENABLED=true" \
    -e "AUTHORIZATION_ADMINLIST_USERS=test@2024.com" \
    -e "AUTHORIZATION_ADMINLIST_READONLY_USERS=test2@2024.com" \
    -e WEAVIATE_HOSTNAME=0.0.0.0 \
    semitechnologies/weaviate:latest

各个参数的含义需要稍作说明：

• -d：让容器后台运行，不占用你的终端。
• --name weaviate：给容器起个名字，方便管理。
• --restart=always：宿主机重启后容器自动启动，省心省力。
• -p 8080:8080：把容器内的 8080 端口映射到宿主机，方便 HTTP 访问。
• -p 50051:50051：gRPC 端口映射，用于高性能通信。
• -e "AUTHENTICATION_APIKEY_ENABLED=true"：开启 API 密钥认证，安全第一。
• -e "AUTHENTICATION_APIKEY_ALLOWED_KEYS=test-secret-key,test2-secret-key"：设置允许的密钥列表。
• -e "AUTHENTICATION_APIKEY_USERS=test@2024.com,test2@2024.com"：将密钥与用户邮箱关联。
• -e "AUTHORIZATION_ADMINLIST_ENABLED=true"：开启管理员列表授权。
• -e "AUTHORIZATION_ADMINLIST_USERS=test@2024.com"：指定管理员用户。
• -e "AUTHORIZATION_ADMINLIST_READONLY_USERS=test2@2024.com"：指定只读用户。
• -e WEAVIATE_HOSTNAME=0.0.0.0：监听所有网络接口，方便外网访问。
• semitechnologies/weaviate:latest：镜像名称和标签。

3. 测试连接

3.1 安装 Weaviate 客户端

用 pip 安装 Weaviate 的 Python 客户端：

pip install -U weaviate-client

3.2 编写测试脚本

创建 test.py 文件，写下下面这段代码：

import weaviate
from weaviate.auth import AuthApiKey

# 连接到本地部署的 Weaviate
client = weaviate.connect_to_local(
    auth_credentials=AuthApiKey("test-secret-key")
)

# 或者自定义连接
client = weaviate.connect_to_custom(
    skip_init_checks=False,
    http_host="127.0.0.1",
    http_port=8080,
    http_secure=False,
    grpc_host="127.0.0.1",
    grpc_port=50051,
    grpc_secure=False,
    # 对应 AUTHENTICATION_APIKEY_ALLOWED_KEYS 中的密钥
    # 注意：此处只需要密钥即可，不需要用户名称
    auth_credentials=AuthApiKey("test-secret-key")
)

# 检查连接是否成功
print(client.is_ready())

# 关闭连接
print(client.close())

3.3 运行测试脚本

在终端里跑一下：

python test.py

如果输出 True，恭喜，连接成功了！

你也可以通过浏览器访问这个地址来确认：

使用 Python 操作 Weaviate 向量数据库

下面是一套完整的 Python 操作示例，涵盖了连接、检查集合、创建集合、插入数据、查询以及删除集合的整个生命周期。

1. 安装 Weaviate Python 客户端

pip install weaviate-client

2. 连接 Weaviate 数据库

import weaviate
from weaviate.auth import AuthApiKey

# 连接到本地 Weaviate 实例
client = weaviate.connect_to_local(
    auth_credentials=AuthApiKey("test-secret-key")
)

# 或者自定义连接
client = weaviate.connect_to_custom(
    http_host="127.0.0.1",
    http_port=8080,
    http_secure=False,
    grpc_host="127.0.0.1",
    grpc_port=50051,
    grpc_secure=False,
    auth_credentials=AuthApiKey("test-secret-key")
)

# 检查连接是否成功
print(client.is_ready())

3. 检查集合是否存在

def check_collection_exists(client: weaviate.WeaviateClient, collection_name: str) -> bool:
    """
    检查集合是否存在
    :param client: Weaviate 客户端
    :param collection_name: 集合名称
    :return: True 或 False
    """
    try:
        collections = client.collections.list_all()
        return collection_name in collections
    except Exception as e:
        print(f"检查集合异常: {e}")
        return False

4. 创建集合

def create_collection(client: weaviate.WeaviateClient, collection_name: str):
    """
    创建集合
    :param client: Weaviate 客户端
    :param collection_name: 集合名称
    """
    collection_obj = {
        "class": collection_name,
        "description": "A collection for product information",
        "vectorizer": "none",  # 假设你会上传自己的向量
        "vectorIndexType": "hnsw",
        "vectorIndexConfig": {
            "distance": "cosine",
            "efConstruction": 200,
            "maxConnections": 64
        },
        "properties": [
            {
                "name": "text",
                "description": "The text content",
                "dataType": ["text"],
                "tokenization": "word",
                "indexFilterable": True,
                "indexSearchable": True
            }
        ]
    }
    try:
        client.collections.create_from_dict(collection_obj)
        print(f"创建集合 '{collection_name}' 成功.")
    except weaviate.exceptions.UnexpectedStatusCodeException as e:
        print(f"创建集合异常: {e}")

5. 插入数据

def save_documents(client: weaviate.WeaviateClient, collection_name: str, documents: list):
    """
    向集合中插入数据
    :param client: Weaviate 客户端
    :param collection_name: 集合名称
    :param documents: 文档列表
    """
    collection = client.collections.get(collection_name)
    for doc in documents:
        content = doc  # 假设文档是简单的字符串
        vector = [0.1, 0.2, 0.3]  # 假设这是你的向量
        properties = {
            "text": content
        }
        try:
            uuid = collection.data.insert(properties=properties, vector=vector)
            print(f"文档添加内容: {content[:30]}..., uuid: {uuid}")
        except Exception as e:
            print(f"添加文档异常: {e}")

6. 查询数据

def query_vector_collection(client: weaviate.WeaviateClient, collection_name: str, query: str, k: int) -> list:
    """
    从集合中查询数据
    :param client: Weaviate 客户端
    :param collection_name: 集合名称
    :param query: 查询字符串
    :param k: 返回的结果数量
    :return: 查询结果列表
    """
    vector = [0.1, 0.2, 0.3]  # 假设这是你的查询向量
    collection = client.collections.get(collection_name)
    response = collection.query.near_vector(
        near_vector=vector,
        limit=k
    )
    documents = [res.properties['text'] for res in response.objects]
    return documents

7. 删除集合

def delete_collection(client: weaviate.WeaviateClient, collection_name: str):
    """
    删除集合
    :param client: Weaviate 客户端
    :param collection_name: 集合名称
    """
    try:
        client.collections.delete(collection_name)
        print(f"删除集合 '{collection_name}' 成功.")
    except Exception as e:
        print(f"删除集合异常: {e}")

8. 示例使用

if __name__ == "__main__":
    # 连接 Weaviate
    client = weaviate.connect_to_local(auth_credentials=AuthApiKey("test-secret-key"))

    collection_name = "MyCollection"

    # 检查集合是否存在
    if not check_collection_exists(client, collection_name):
        # 创建集合
        create_collection(client, collection_name)

    # 插入数据
    documents = ["This is a test document.", "Another document for testing."]
    save_documents(client, collection_name, documents)

    # 查询数据
    query_results = query_vector_collection(client, collection_name, "test", 2)
    print("查询结果:", query_results)

    # 删除集合
    delete_collection(client, collection_name)

    # 关闭连接
    client.close()

如何实现语义检索

在 TrusRAG 项目中，我们把上面的这些操作封装成了一个更易用的 WeaviateEngine 类，代码在：

WeaviateEngine 的实现如下，关键地方已经加上了注释：

from typing import List, Dict, Any, Optional, Union
import numpy as np
import weaviate
from weaviate import WeaviateClient
from weaviate.collections import Collection
import weaviate.classes.config as wc
from weaviate.classes.config import Property, DataType
from trustrag.modules.retrieval.embedding import EmbeddingGenerator
from weaviate.classes.query import MetadataQuery

class WeaviateEngine:
    def __init__(
            self,
            collection_name: str,
            embedding_generator: EmbeddingGenerator,
            client_params: Dict[str, Any] = {
                "http_host": "localhost",
                "http_port": 8080,
                "http_secure": False,
                "grpc_host": "localhost",
                "grpc_port": 50051,
                "grpc_secure": False,
            },
    ):
        """
        Initialize the Weaviate vector store.
        :param collection_name: Name of the Weaviate collection
        :param embedding_generator: An instance of EmbeddingGenerator to generate embeddings
        :param weaviate_client_params: Dictionary of parameters to pass to Weaviate client
        """
        self.collection_name = collection_name
        self.embedding_generator = embedding_generator

        # Initialize Weaviate client with provided parameters
        self.client = weaviate.connect_to_custom(
            skip_init_checks=False,
            **client_params
        )

        # Create collection if it doesn't exist
        if not self._collection_exists():
            self._create_collection()

    def _collection_exists(self) -> bool:
        """Check if collection exists in Weaviate."""
        try:
            collections = self.client.collections.list_all()
            collection_names = [c.lower() for c in collections]
            return self.collection_name in collection_names
        except Exception as e:
            print(f"Error checking collection existence: {e}")
            return False

    def _create_collection(self):
        """Create a new collection in Weaviate."""
        try:
            self.client.collections.create(
                name=self.collection_name,
                properties=[
                    wc.Property(
                        name="text",
                        data_type=wc.DataType.TEXT
                    ),
                    wc.Property(
                        name="title",
                        data_type=wc.DataType.TEXT,
                        skip_vectorization=True
                    ),
                ]
            )
        except Exception as e:
            print(f"Error creating collection: {e}")
            raise

    def upload_vectors(
            self,
            vectors: Union[np.ndarray, List[List[float]]],
            payload: List[Dict[str, Any]],
            batch_size: int = 100
    ):
        """
        Upload vectors and payload to the Weaviate collection.
        :param vectors: A numpy array or list of vectors to upload
        :param payload: A list of dictionaries containing the payload for each vector
        :param batch_size: Number of vectors to upload in a single batch
        """
        if not isinstance(vectors, np.ndarray):
            vectors = np.array(vectors)
        if len(vectors) != len(payload):
            raise ValueError("Vectors and payload must have the same length.")

        collection = self.client.collections.get(self.collection_name)

        # Process in batches
        for i in range(0, len(vectors), batch_size):
            batch_vectors = vectors[i:i + batch_size]
            batch_payload = payload[i:i + batch_size]

            try:
                with collection.batch.dynamic() as batch:
                    for idx, (properties, vector) in enumerate(zip(batch_payload, batch_vectors)):
                        text_content = properties.get('description', '')
                        metadata = {k: v for k, v in properties.items() if k != 'description'}
                        properties_dict = {
                            "text": text_content,
                            "title": metadata.get('title', f'Object {idx}')
                        }
                        batch.add_object(
                            properties=properties_dict,
                            vector=vector
                        )
            except Exception as e:
                print(f"Error uploading batch: {e}")
                raise

    def search(
            self,
            text: str,
            query_filter: Optional[Dict[str, Any]] = None,
            limit: int = 5
    ) -> List[Dict[str, Any]]:
        """
        Search for the closest vectors in the collection based on the input text.
        :param text: The text query to search for
        :param query_filter: Optional filter to apply to the search
        :param limit: Number of closest results to return
        :return: List of payloads from the closest vectors
        """
        vector = self.embedding_generator.generate_embedding(text)
        print(vector.shape)
        collection = self.client.collections.get(self.collection_name)

        query_args = {
            "near_vector": vector,
            "limit": limit,
            "return_metadata": MetadataQuery(distance=True)
        }

        if query_filter:
            query_args["filter"] = query_filter

        results = collection.query.near_vector(**query_args)

        payloads = []
        for obj in results.objects:
            payload = obj.properties.get('metadata', {})
            payload['text'] = obj.properties.get('text', '')
            payload['_distance'] = obj.metadata.distance
            payloads.append(payload)

        return payloads

    def build_filter(self, conditions: List[Dict[str, Any]]) -> Dict[str, Any]:
        """
        Build a Weaviate filter from a list of conditions.
        :param conditions: A list of conditions, each with 'key' and 'match'
        :return: A Weaviate filter object
        """
        filter_dict = {
            "operator": "And",
            "operands": []
        }

        for condition in conditions:
            key = condition.get("key")
            match_value = condition.get("match")
            if key and match_value is not None:
                filter_dict["operands"].append({
                    "path": [f"metadata.{key}"],
                    "operator": "Equal",
                    "valueString": str(match_value)
                })

        return filter_dict if filter_dict["operands"] else None

下面是一段测试代码，用真实的创业公司数据来演示整个流程：

from trustrag.modules.retrieval.embedding import SentenceTransformerEmbedding
from trustrag.modules.engine.weaviate_cli import WeaviateEngine

if __name__ == '__main__':
    # 初始化 WeaviateEngine
    local_embedding_generator = SentenceTransformerEmbedding(
        model_name_or_path=r"H:pretrained_modelsmteball-MiniLM-L6-v2",
        device="cuda"
    )
    weaviate_engine = WeaviateEngine(
        collection_name="startups",
        embedding_generator=local_embedding_generator,
        client_params={
            "http_host": "localhost",
            "http_port": 8080,
            "http_secure": False,
            "grpc_host": "localhost",
            "grpc_port": 50051,
            "grpc_secure": False,
        }
    )

    documents = [
        {"name": "SaferCodes", "images": "https://safer.codes/img/brand/logo-icon.png",
         "alt": "SaferCodes Logo QR codes generator system forms for COVID-19",
         "description": "QR codes systems for COVID-19.\nSimple tools for bars, restaurants, offices, and other small proximity businesses.",
         "link": "https://safer.codes", "city": "Chicago"},
        # ... 其余文档略（保持原样）
    ]
    vectors = weaviate_engine.embedding_generator.generate_embeddings([doc["description"] for doc in documents])
    print(vectors.shape)
    payload = [doc for doc in documents]

    # Upload vectors and payload
    weaviate_engine.upload_vectors(vectors=vectors, payload=payload)

    # 构建过滤器并搜索
    conditions = [
        {"key": "city", "match": "Berlin"},
    ]
    custom_filter = weaviate_engine.build_filter(conditions)

    # 搜索柏林的度假相关创业公司
    results = weaviate_engine.search(
        text="vacations",
        # query_filter=custom_filter,
        limit=5
    )
    print(results)

输出结果大致如下：

{'text': "Fueling more vacations\nWe enable our customers to travel more, travel better and travel further. 20M+ consumers stock away miles & points to satisfy their wanderlust.\nFlying around or using credit cards are the only good ways to fill the stockpile today. We've built the third way. Customers ...", '_distance': 0.5216099619865417

你是一名 AI 行业编辑，请围绕下面这条热点输出一份资讯解读：
热点：RAG落地利器Weaviate向量数据库部署使用教程要求：
1. 先用一句话解释这条热点在讲什么
2. 再总结它为什么重要
3. 说明会影响哪些 AI 产品或内容方向
4. 最后给出 3 个适合资讯站使用的标题