Weaviate 部署
1. 简介
Weaviate 是一种开源的向量搜索引擎数据库，允许以类属性的方式存储 JSON 文档，并将机器学习向量附加到这些文档上，以在向量空间中表示它们。Weaviate 支持语义搜索、问答提取、分类等功能，并且可以通过 GraphQL-API 轻松访问数据。
官网地址：https://weaviate.io/
2. 安装 Weaviate
从 Docker Hub 下载 Weaviate 的最新镜像：
docker pull semitechnologies/weaviate:latest
如果拉取镜像速度较慢，可以尝试替换镜像源。
2.3 运行 Weaviate 容器
使用以下命令运行 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=," \
-e "AUTHORIZATION_ADMINLIST_ENABLED=true" \
-e "AUTHORIZATION_ADMINLIST_USERS=" \
-e "AUTHORIZATION_ADMINLIST_READONLY_USERS=" \
-e WEAVIATE_HOSTNAME=0.0.0.0 \
semitechnologies/weaviate:latest
参数说明
-d: 让容器在后台运行。--name weaviate: 给容器命名为weaviate。--restart=always: 配置容器在宿主机重启后自动启动。-p 8080:8080: 将容器内的 8080 端口映射到宿主机的 8080 端口。-p 50051:50051: 将容器内的 50051 端口映射到宿主机的 50051 端口。-e "AUTHENTICATION_APIKEY_ENABLED=true": 启用 API 密钥认证功能。-e "AUTHENTICATION_APIKEY_ALLOWED_KEYS=test-secret-key,test2-secret-key": 指定允许使用的 API 密钥列表。-e "AUTHENTICATION_APIKEY_USERS=,": 关联密钥与用户邮箱。-e "AUTHORIZATION_ADMINLIST_ENABLED=true": 开启管理员列表授权。-e "AUTHORIZATION_ADMINLIST_USERS=": 指定管理员列表中的用户。-e "AUTHORIZATION_ADMINLIST_READONLY_USERS=": 指定只读权限的用户列表。-e WEAVIATE_HOSTNAME=0.0.0.0: 设置 Weaviate 的主机名，监听所有可用网络接口。semitechnologies/weaviate:latest: 指定要从 Docker Hub 下载并运行的 Weaviate 镜像的最新版本。
3. 测试连接
3.1 安装 Weaviate 客户端
使用 pip 安装 Weaviate 客户端：
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，则表示连接成功。
可以通过浏览器访问地址：
http://localhost:8080/v1/docs
使用python操作Weaviate向量数据库
以下是使用 Python 操作 Weaviate 向量数据库的完整示例，涵盖连接数据库、检查集合是否存在、创建集合、插入数据、查询数据以及删除集合等操作。1. 安装 Weaviate Python 客户端
首先，确保你已经安装了 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项目中，对上面教程进行了封装，具体链接如下：https://github.com/gomate-community/TrustRAG/blob/pipeline/trustrag/modules/engine/weaviate_cli.py
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,
# Define properties of metadata
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)):
# Separate text content and other metadata
text_content = properties.get('description',
'') # Assuming 'description' is the main text field
metadata = {k: v for k, v in properties.items() if k != 'description'}
# Prepare the properties dictionary
properties_dict = {
"text": text_content,
"title": metadata.get('title', f'Object {idx}') # Using title from metadata or default
}
# Add the object with properties and vector
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
"""
# Generate embedding for the query text
vector = self.embedding_generator.generate_embedding(text)
print(vector.shape)
collection = self.client.collections.get(self.collection_name)
# Prepare query arguments
query_args = {
"near_vector": vector,
"limit": limit,
"return_metadata": MetadataQuery(distance=True)
}
# Add filter if provided
if query_filter:
query_args["filter"] = query_filter
results = collection.query.near_vector(**query_args)
# Convert results to the same format as QdrantEngine
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, where each condition is a dictionary with:
- key: The field name to filter on
- match: The value to 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__':
# 初始化 MilvusEngine
local_embedding_generator = SentenceTransformerEmbedding(model_name_or_path=r"H:\pretrained_models\mteb\all-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"},
{"name": "Human Practice",
"images": "https://d1qb2nb5cznatu.cloudfront.net/startups/i/373036-94d1e190f12f2c919c3566ecaecbda68-thumb_jpg.jpg?buster=1396498835",
"alt": "Human Practice - health care information technology",
"description": "Point-of-care word of mouth\nPreferral is a mobile platform that channels physicians\u2019 interest in networking with their peers to build referrals within a hospital system.\nHospitals are in a race to employ physicians, even though they lose billions each year ($40B in 2014) on employment. Why ...",
"link": "http://humanpractice.com", "city": "Chicago"},
{"name": "StyleSeek",
"images": "https://d1qb2nb5cznatu.cloudfront.net/startups/i/3747-bb0338d641617b54f5234a1d3bfc6fd0-thumb_jpg.jpg?buster=1329158692",
"alt": "StyleSeek - e-commerce fashion mass customization online shopping",
"description": "Personalized e-commerce for lifestyle products\nStyleSeek is a personalized e-commerce site for lifestyle products.\nIt works across the style spectrum by enabling users (both men and women) to create and refine their unique StyleDNA.\nStyleSeek also promotes new products via its email newsletter, 100% personalized ...",
"link": "http://styleseek.com", "city": "Chicago"},
{"name": "Scout",
"images": "https://d1qb2nb5cznatu.cloudfront.net/startups/i/190790-dbe27fe8cda0614d644431f853b64e8f-thumb_jpg.jpg?buster=1389652078",
"alt": "Scout - security consumer electronics internet of things",
"description": "Hassle-free Home Security\nScout is a self-installed, wireless home security system. We've created a more open, affordable and modern system than what is available on the market today. With month-to-month contracts and portable devices, Scout is a renter-friendly solution for the other ...",
"link": "http://www.scoutalarm.com", "city": "Chicago"},
{"name": "Invitation codes", "images": "https://invitation.codes/img/inv-brand-fb3.png",
"alt": "Invitation App - Share referral codes community ",
"description": "The referral community\nInvitation App is a social network where people post their referral codes and collect rewards on autopilot.",
"link": "https://invitation.codes", "city": "Chicago"},
{"name": "Hyde Park Angels",
"images": "https://d1qb2nb5cznatu.cloudfront.net/startups/i/61114-35cd9d9689b70b4dc1d0b3c5f11c26e7-thumb_jpg.jpg?buster=1427395222",
"alt": "Hyde Park Angels - ",
"description": "Hyde Park Angels is the largest and most active angel group in the Midwest. With a membership of over 100 successful entrepreneurs, executives, and venture capitalists, the organization prides itself on providing critical strategic expertise to entrepreneurs and ...",
"link": "http://hydeparkangels.com", "city": "Chicago"},
{"name": "GiveForward",
"images": "https://d1qb2nb5cznatu.cloudfront.net/startups/i/1374-e472ccec267bef9432a459784455c133-thumb_jpg.jpg?buster=1397666635",
"alt": "GiveForward - health care startups crowdfunding",
"description": "Crowdfunding for medical and life events\nGiveForward lets anyone to create a free fundraising page for a friend or loved one's uncovered medical bills, memorial fund, adoptions or any other life events in five minutes or less. Millions of families have used GiveForward to raise more than $165M to let ...",
"link": "http://giveforward.com", "city": "Chicago"},
{"name": "MentorMob",
"images": "https://d1qb2nb5cznatu.cloudfront.net/startups/i/19374-3b63fcf38efde624dd79c5cbd96161db-thumb_jpg.jpg?buster=1315734490",
"alt": "MentorMob - digital media education ventures for good crowdsourcing",
"description": "Google of Learning, indexed by experts\nProblem: Google doesn't index for learning. Nearly 1 billion Google searches are done for \"how to\" learn various topics every month, from photography to entrepreneurship, forcing learners to waste their time sifting through the millions of results.\nMentorMob is ...",
"link": "http://www.mentormob.com", "city": "Chicago"},
{"name": "The Boeing Company",
"images": "https://d1qb2nb5cznatu.cloudfront.net/startups/i/49394-df6be7a1eca80e8e73cc6699fee4f772-thumb_jpg.jpg?buster=1406172049",
"alt": "The Boeing Company - manufacturing transportation", "description": "",
"link": "http://www.boeing.com", "city": "Berlin"},
{"name": "NowBoarding \u2708\ufe0f",
"images": "https://static.above.flights/img/lowcost/envelope_blue.png",
"alt": "Lowcost Email cheap flights alerts",
"description": "Invite-only mailing list.\n\nWe search the best weekend and long-haul flight deals\nso you can book before everyone else.",
"link": "https://nowboarding.club/", "city": "Berlin"},
{"name": "Rocketmiles",
"images": "https://d1qb2nb5cznatu.cloudfront.net/startups/i/158571-e53ddffe9fb3ed5e57080db7134117d0-thumb_jpg.jpg?buster=1361371304",
"alt": "Rocketmiles - e-commerce online travel loyalty programs hotels",
"description": "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 ...",
"link": "http://www.Rocketmiles.com", "city": "Berlin"}
]
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
