Weaviate 2026: AI 原生向量搜索引擎处理 100 亿+ 对象 — 企业部署指南

docker run -d \
  -p 8080:8080 \
  -p 50051:50051 \
  --name weaviate \
  semitechnologies/weaviate:1.31.0 \
  --host 0.0.0.0 \
  --port 8080 \
  --scheme http \
  --env ENABLE_MODULES='text2vec-openai,generative-openai' \
  --env OPENAI_APIKEY=$OPENAI_API_KEY

curl http://localhost:8080/v1/meta
# 返回: {"hostname":"...","version":"1.31.0","modules":{...}}

# docker-compose.yml
version: '3.8'
services:
  weaviate:
    image: semitechnologies/weaviate:1.31.0
    ports:
      - "8080:8080"
      - "50051:50051"
    environment:
      QUERY_DEFAULTS_LIMIT: 100
      AUTHENTICATION_ANONYMOUS_ACCESS_ENABLED: 'false'
      AUTHENTICATION_APIKEY_ENABLED: 'true'
      AUTHENTICATION_APIKEY_ALLOWED_KEYS: 'your-api-key-here'
      AUTHENTICATION_APIKEY_USERS: 'admin'
      PERSISTENCE_DATA_PATH: '/var/lib/weaviate'
      DEFAULT_VECTORIZER_MODULE: 'none'
      ENABLE_MODULES: ''
      CLUSTER_HOSTNAME: 'node1'
    volumes:
      - weaviate_data:/var/lib/weaviate
    deploy:
      resources:
        limits:
          memory: 16G
volumes:
  weaviate_data:

import weaviate
from weaviate.classes import ConfiguredBatch, Vectorizers

client = weaviate.connect_to_local()

# 定义带向量索引设置的集合
client.collections.create(
    name="Product",
    vectorizer_config=Vectorizers.text2vec_openai(),
    vector_index_config=Configure.VectorIndex.hnsw(
        ef=256,
        ef_construction=128,
        max_connections=64,
        dynamic_ef_enabled=True,
        dynamic_ef_min=100,
        dynamic_ef_max=500
    ),
    properties=[
        Property(name="name", data_type=DataType.TEXT),
        Property(name="description", data_type=DataType.TEXT),
        Property(name="category", data_type=DataType.TEXT),
        Property(name="price", data_type=DataType.NUMBER),
        Property(name="in_stock", data_type=DataType.BOOL)
    ]
)

# 批量导入商品
products = client.collections.get("Product")
with products.batch.dynamic() as batch:
    for item in product_data:
        batch.add_object(properties=item)

print(f"导入了 {len(products)} 个对象")

from langchain_weaviate import WeaviateVectorStore
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain.chains import RetrievalQA
import weaviate

client = weaviate.connect_to_local()

vectorstore = WeaviateVectorStore(
    client=client,
    index_name="Product",
    text_key="description",
    embedding=OpenAIEmbeddings()
)

retriever = vectorstore.as_retriever(search_kwargs={"k": 5})
llm = ChatOpenAI(model="gpt-4o")

qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=retriever
)

result = qa_chain.invoke("有哪些 200 美元以下的无线耳机有库存？")
print(result["result"])

products = client.collections.get("Product")

results = products.query.hybrid(
    query="降噪耳机",
    query_properties=["name", "description"],
    alpha=0.7,  # 0.0 = 纯 BM25, 1.0 = 纯向量
    limit=10,
    filters=Filter.by_property("in_stock").equal(True)
        & Filter.by_property("price").less_than(300)
)

for obj in results.objects:
    print(f"{obj.properties['name']}: ${obj.properties['price']}")

# 添加 Weaviate Helm 仓库
helm repo add weaviate https://weaviate.github.io/weaviate-helm

# 使用生产值安装
helm install weaviate weaviate/weaviate \
  --namespace weaviate \
  --create-namespace \
  --set replicas=3 \
  --set resources.requests.cpu=4 \
  --set resources.requests.memory=16Gi \
  --set resources.limits.cpu=8 \
  --set resources.limits.memory=32Gi \
  --set storage.size=500Gi \
  --set storage.storageClassName=fast-ssd \
  --set env.CLUSTER_DATA_BIND_PORT=7001 \
  --set env.GOMAXPROCS=8 \
  --set service.type=LoadBalancer

from weaviate.classes import ConfiguredBatch, Vectorizers, Multi2VecField

client.collections.create(
    name="MultiModalProduct",
    vectorizer_config=Vectorizers.multi2vec_clip(
        image_fields=[Multi2VecField(name="image", weight=0.9)],
        text_fields=[Multi2VecField(name="description", weight=0.1)]
    ),
    properties=[
        Property(name="description", data_type=DataType.TEXT),
        Property(name="image", data_type=DataType.BLOB),
        Property(name="sku", data_type=DataType.TEXT)
    ]
)

# 通过文本跨图像描述搜索
results = collection.query.near_text(
    query="红色跑鞋",
    limit=5
)

# 通过图像搜索（查找相似商品）
import base64
with open("query_image.jpg", "rb") as f:
    img_b64 = base64.b64encode(f.read()).decode()

results = collection.query.near_image(near_image=img_b64, limit=5)

# 监控的额外环境变量
environment:
  PROMETHEUS_MONITORING_ENABLED: 'true'
  PROMETHEUS_MONITORING_PORT: 2112

# Weaviate 查询延迟
weaviate_queries_durations_ms_bucket

# 对象数量
weaviate_objects_count

# 向量索引队列大小
weaviate_vector_index_queue_size

# 内存使用
weaviate_runtime_mem_sys_bytes

# 请求速率
rate(weaviate_requests_total[5m])

from weaviate.classes.rbac import Permissions, Roles

# 创建只读角色
client.roles.create(
    name="readonly",
    permissions=[
        Permissions.collections.read(),
        Permissions.data.read()
    ]
)

# 将角色分配给用户
client.users.assign_roles("data_scientist_1", ["readonly"])

# 为特定集合创建管理员角色
client.roles.create(
    name="product_admin",
    permissions=[
        Permissions.collections(collection="Product").full(),
        Permissions.data(collection="Product").full()
    ]
)

# 触发手动备份
curl -X POST http://localhost:8080/v1/backups/s3 \
  -H "Content-Type: application/json" \
  -d '{
    "id": "backup-2026-05-19",
    "include": ["Product", "UserEmbedding"],
    "config": {
      "endpoint": "s3.amazonaws.com",
      "bucket": "weaviate-backups",
      "path": "production/"
    }
  }'

# kubernetes/backup-cronjob.yaml
apiVersion: batch/v1
kind: CronJob
metadata:
  name: weaviate-backup
spec:
  schedule: "0 2 * * *"  # 每天凌晨 2 点
  jobTemplate:
    spec:
      template:
        spec:
          containers:
          - name: backup
            image: curlimages/curl:latest
            command:
            - /bin/sh
            - -c
            - |
              curl -X POST http://weaviate:8080/v1/backups/s3 \
                -H "Content-Type: application/json" \
                -d "{\"id\":\"backup-$(date +%Y%m%d)\"}"
          restartPolicy: OnFailure

# 大规模集群的 Helm 值
replicas: 5
env:
  CLUSTER_JOIN: "weaviate-0.weaviate-headless:7001"
  CLUSTER_GOSSIP_BIND_PORT: "7100"
  CLUSTER_DATA_BIND_PORT: "7101"
  RAFT_JOIN: "weaviate-0,weaviate-1,weaviate-2"
  RAFT_BOOTSTRAP_EXPECT: "3"

persistence:
  enabled: true
  size: 1Ti
  storageClass: premium-rwo

resources:
  requests:
    memory: "64Gi"
    cpu: "16"
  limits:
    memory: "128Gi"
    cpu: "32"

import weaviate
from weaviate.classes import DataObject

client = weaviate.connect_to_local(
    grpc_port=50051  # 对批量操作使用 gRPC
)

products = client.collections.get("Product")

# gRPC 批量插入 —— 明显快于 REST
with products.batch.fixed_size(batch_size=1000) as batch:
    for item in large_dataset:  # 1000 万+ 对象
        batch.add_object(properties=item)
        if batch.number_errors > 100:
            print("错误太多，停止")
            break

failed = products.batch.failed_objects
print(f"导入失败: {len(failed)}")

# 跳过向量器 —— 手动提供向量
client.collections.create(
    name="CustomEmbedding",
    vectorizer_config=None,  # 无自动向量化
    vector_index_config=Configure.VectorIndex.hnsw(
        ef=128,
        max_connections=32
    ),
    properties=[...]
)

# 插入预计算向量
collection = client.collections.get("CustomEmbedding")
collection.data.insert(
    properties={"text": "示例文档"},
    vector=[0.01, -0.02, 0.03, ...]  # 您的嵌入
)