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

显示全部楼层

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=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: 给容器命名为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=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: 设置 Weaviate 的主机名，监听所有可用网络接口。
semitechnologies/weaviate:latest: 指定要从 Docker Hub 下载并运行的 Weaviate 镜像的最新版本。

3. 测试连接

3.1 安装 Weaviate 客户端

使用 pip 安装 Weaviate 客户端：

pip install-U weaviate-client

3.2 编写测试脚本

创建一个test.py文件，内容如下：

importweaviate
fromweaviate.authimportAuthApiKey

#连接到本地部署的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 数据库

importweaviate
fromweaviate.authimportAuthApiKey

#连接到本地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. 检查集合是否存在

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

4. 创建集合

defcreate_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}'成功.")
exceptweaviate.exceptions.UnexpectedStatusCodeExceptionase:
print(f"创建集合异常:{e}")

5. 插入数据

defsave_documents(client:weaviate.WeaviateClient,collection_name:str,documents:list):
"""
向集合中插入数据
:param client:Weaviate客户端
:param collection_name:集合名称
:param documents:文档列表
"""
collection=client.collections.get(collection_name)
fordocindocuments:
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}")
exceptExceptionase:
print(f"添加文档异常:{e}")

6. 查询数据

defquery_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']forresinresponse.objects]
returndocuments

7. 删除集合

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

8. 示例使用

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

collection_name="MyCollection"

#检查集合是否存在
ifnotcheck_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
fromweaviate.classes.query import MetadataQuery

class WeaviateEngine:
def __init__(
self,
collection_name:str,
embedding_generator:EmbeddingGenerator,
client_paramsict[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_paramsictionary 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
ifnot 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()forcincollections]
returnself.collection_nameincollection_names
except Exception as e:
print(f"Error checking collection existence:{e}")
returnFalse

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]]],
payloadist[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
"""
ifnot isinstance(vectors,np.ndarray):
vectors=np.array(vectors)
iflen(vectors)!=len(payload):
raise ValueError("Vectors and payload must have the same length.")

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

#Process in batches
foriinrange(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:
foridx,(properties,vector)inenumerate(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:vfork,vinproperties.items()ifk!='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
:returnist 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
ifquery_filter:
query_args["filter"]=query_filter

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

#Convert results to the same format as QdrantEngine
payloads=[]
forobjinresults.objects:
payload=obj.properties.get('metadata',{})
payload['text']=obj.properties.get('text','')
payload['_distance']=obj.metadata.distance
payloads.append(payload)

returnpayloads


def build_filter(self,conditionsist[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":[]
}

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

returnfilter_dictiffilter_dict["operands"]elseNone

测试代码如下：

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":"oint-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($40Bin 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":"ersonalized 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$165Mto 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"]fordocindocuments])
print(vectors.shape)
payload=[docfordocindocuments]

#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