Dify如何解决Weaviate数据库 1.19.0 到 1.27.0+ 版本之间架构不兼容问题（附迁移脚本）

importos
importweaviate
fromweaviate.classes.configimportConfigure, VectorDistances
importsys
importtime
fromtypingimportList,Dict,Any

# Configuration
WEAVIATE_ENDPOINT = os.getenv("WEAVIATE_ENDPOINT","http://weaviate:8080")
WEAVIATE_GRPC_ENDPOINT = os.getenv("WEAVIATE_GRPC_ENDPOINT","grpc://weaviate:50051")
WEAVIATE_API_KEY = os.getenv("WEAVIATE_API_KEY","WVF5YThaHlkYwhGUSmCRgsX3tD5ngdN8pkih")
BATCH_SIZE =1000
WEAVIATE_HOST = WEAVIATE_ENDPOINT.split("//")[-1].split(":")[0]
WEAVIATE_PORT =int(WEAVIATE_ENDPOINT.split(":")[-1])
WEAVIATE_GRPC_PORT =int(WEAVIATE_GRPC_ENDPOINT.split(":")[-1])

defidentify_old_collections(client: weaviate.WeaviateClient) ->List[str]:
"""Identify collections that need migration (those without vectorConfig)"""
collections_to_migrate = []

all_collections = client.collections.list_all()
print(f"Found{len(all_collections)}total collections")

forcollection_nameinall_collections.keys():
# Only check Vector_index collections (Dify knowledge bases)
ifnotcollection_name.startswith("Vector_index_"):
continue

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

# Check if this collection has the old schema
ifconfig.vector_configisNone:
collections_to_migrate.append(collection_name)
print(f" -{collection_name}: OLD SCHEMA (needs migration)")
else:
print(f" -{collection_name}: NEW SCHEMA (skip)")

returncollections_to_migrate

defget_collection_schema(
client: weaviate.WeaviateClient, collection_name:str
) ->Dict[str,Any]:
"""Get the full schema of a collection via REST API"""
importrequests

response = requests.get(
f"http://{WEAVIATE_HOST}:{WEAVIATE_PORT}/v1/schema/{collection_name}",
headers={"Authorization":f"Bearer{WEAVIATE_API_KEY}"},
)

ifresponse.status_code ==200:
returnresponse.json()
else:
raiseException(f"Failed to get schema:{response.text}")

defcreate_new_collection(
client: weaviate.WeaviateClient, old_name:str, schema:Dict[str,Any]
) ->str:
"""Create a new collection with updated schema using REST API"""
importrequests

# Generate new collection name
new_name =f"{old_name}_migrated"

print(f"Creating new collection:{new_name}")

# Build new schema with proper vectorConfig
# Note: When using vectorConfig (named vectors), we don't set class-level vectorizer
new_schema = {
"class": new_name,
# This is the key: define vectorConfig with "default" named vector
# Do NOT set class-level vectorizer when using vectorConfig
"vectorConfig": {
"default": {
"vectorizer": {"none": {}},
"vectorIndexType":"hnsw",
"vectorIndexConfig": {
"distance":"cosine",
"ef": -1,
"efConstruction":128,
"maxConnections":32,
},
}
},
"properties": [],
}

# Copy properties from old schema
if"properties"inschema:
new_schema["properties"] = schema["properties"]

# Create collection via REST API
response = requests.post(
f"{WEAVIATE_ENDPOINT}/v1/schema",
json=new_schema,
headers={"Authorization":f"Bearer{WEAVIATE_API_KEY}"},
)

ifresponse.status_codenotin[200,201]:
raiseException(f"Failed to create collection:{response.text}")

print(f" Created new collection:{new_name}")
returnnew_name

defmigrate_collection_data(
client: weaviate.WeaviateClient, old_collection_name:str, new_collection_name:str
) ->int:
"""Migrate data from old collection to new collection using cursor-based pagination"""

old_collection = client.collections.get(old_collection_name)
new_collection = client.collections.get(new_collection_name)

total_migrated =0
cursor =None

print(f"Migrating data from{old_collection_name}to{new_collection_name}")

whileTrue:
# Fetch batch of objects using cursor-based pagination
ifcursorisNone:
# First batch
response = old_collection.query.fetch_objects(
limit=BATCH_SIZE, include_vector=True
)
else:
# Subsequent batches using cursor
response = old_collection.query.fetch_objects(
limit=BATCH_SIZE, include_vector=True, after=cursor
)

objects = response.objects

ifnotobjects:
break

# Use batch insert for efficiency
withnew_collection.batch.dynamic()asbatch:
forobjinobjects:
# Prepare properties
properties = obj.properties

# Add object with vector
batch.add_object(
properties=properties,
vector=(
obj.vector["default"]
ifisinstance(obj.vector,dict)
elseobj.vector
),
uuid=obj.uuid,
)

total_migrated +=len(objects)
print(f" Migrated{total_migrated}objects...")

# Update cursor for next iteration
iflen(objects) < BATCH_SIZE:
# Last batch
break
else:
# Get the last object's UUID for cursor
cursor = objects[-1].uuid

print(f" Total migrated:{total_migrated}objects")
returntotal_migrated

defverify_migration(
client: weaviate.WeaviateClient, old_collection_name:str, new_collection_name:str
):
"""Verify that the migration was successful"""

old_collection = client.collections.get(old_collection_name)
new_collection = client.collections.get(new_collection_name)

# Count objects in both collections
old_count_response = old_collection.query.fetch_objects(limit=1)
new_count_response = new_collection.query.fetch_objects(limit=1)

# Get aggregation for accurate counts
old_agg = old_collection.aggregate.over_all(total_count=True)
new_agg = new_collection.aggregate.over_all(total_count=True)

old_count = old_agg.total_count
new_count = new_agg.total_count

print(f"
Verification:")
print(f" Old collection ({old_collection_name}):{old_count}objects")
print(f" New collection ({new_collection_name}):{new_count}objects")

ifold_count == new_count:
print(f" Status: SUCCESS - Counts match!")
returnTrue
else:
print(f" Status: WARNING - Counts don't match!")
returnFalse

defreplace_old_collection(
client: weaviate.WeaviateClient, old_collection_name:str, new_collection_name:str
):
"""Replace old collection with migrated one by recreating with original name"""
importrequests

print(f"
Replacing old collection with migrated data...")

# Step 1: Delete old collection
print(f" Step 1: Deleting old collection...")
response = requests.delete(
f"{WEAVIATE_ENDPOINT}/v1/schema/{old_collection_name}",
headers={"Authorization":f"Bearer{WEAVIATE_API_KEY}"},
)
ifresponse.status_code !=200:
print(f" Warning: Could not delete old collection:{response.text}")
else:
print(f" Deleted")

# Step 2: Get schema from migrated collection
print(f" Step 2: Getting schema from migrated collection...")
schema_response = requests.get(
f"{WEAVIATE_ENDPOINT}/v1/schema/{new_collection_name}",
headers={"Authorization":f"Bearer{WEAVIATE_API_KEY}"},
)
schema = schema_response.json()
schema["class"] = old_collection_name

# Step 3: Create collection with original name and new schema
print(f" Step 3: Creating collection with original name...")
create_response = requests.post(
f"{WEAVIATE_ENDPOINT}/v1/schema",
json=schema,
headers={"Authorization":f"Bearer{WEAVIATE_API_KEY}"},
)
ifcreate_response.status_codenotin[200,201]:
raiseException(f"Failed to create collection:{create_response.text}")
print(f" Created")

# Step 4: Copy data to collection with original name using cursor-based pagination
print(f" Step 4: Copying data to original collection name...")
migrated_collection = client.collections.get(new_collection_name)
new_collection = client.collections.get(old_collection_name)

total_copied =0
cursor =None

whileTrue:
# Fetch batch of objects using cursor-based pagination
ifcursorisNone:
# First batch
response = migrated_collection.query.fetch_objects(
include_vector=True, limit=BATCH_SIZE
)
else:
# Subsequent batches using cursor
response = migrated_collection.query.fetch_objects(
include_vector=True, limit=BATCH_SIZE, after=cursor
)

objects = response.objects

ifnotobjects:
break

# Use batch insert for efficiency
withnew_collection.batch.dynamic()asbatch:
forobjinobjects:
batch.add_object(
properties=obj.properties, vector=obj.vector, uuid=obj.uuid
)

total_copied +=len(objects)
print(f" Copied{total_copied}objects...")

# Update cursor for next iteration
iflen(objects) < BATCH_SIZE:
break
else:
cursor = objects[-1].uuid

print(f" Total copied:{total_copied}objects")

# Step 5: Delete the temporary migrated collection
print(f" Step 5: Cleaning up temporary migrated collection...")
response = requests.delete(
f"{WEAVIATE_ENDPOINT}/v1/schema/{new_collection_name}",
headers={"Authorization":f"Bearer{WEAVIATE_API_KEY}"},
)
ifresponse.status_code ==200:
print(f" Cleaned up")

print(
f"
SUCCESS!{old_collection_name}now has the new schema with{total_copied}objects"
)
returnTrue

defmigrate_all_collections():
"""Main migration function"""

print("="*80)
print("Weaviate Collection Migration Script")
print("Migrating from Weaviate 1.19.0 schema to 1.27.0+ schema")
print("="*80)
print()

client = weaviate.connect_to_local(
host=WEAVIATE_HOST,
port=WEAVIATE_PORT,
grpc_port=WEAVIATE_GRPC_PORT,
auth_credentials=weaviate.auth.AuthApiKey(WEAVIATE_API_KEY),
)

try:
# Step 1: Identify collections that need migration
print("Step 1: Identifying collections that need migration...")
collections_to_migrate = identify_old_collections(client)

ifnotcollections_to_migrate:
print("
No collections need migration. All collections are up to date!")
return

print(f"
Found{len(collections_to_migrate)}collections to migrate:")
forcolincollections_to_migrate:
print(f" -{col}")

# Confirm before proceeding
print("
This script will:")
print("1. Create new collections with updated schema")
print("2. Copy all data using efficient batch operations")
print("3. Verify the migration")
print("4. Optionally rename collections to activate the new ones")
print()

# Step 2: Migrate each collection
forcollection_nameincollections_to_migrate:
print("
"+"="*80)
print(f"Migrating:{collection_name}")
print("="*80)

try:
# Get old schema
schema = get_collection_schema(client, collection_name)

# Create new collection
new_collection_name = create_new_collection(
client, collection_name, schema
)

# Migrate data
migrated_count = migrate_collection_data(
client, collection_name, new_collection_name
)

# Verify migration
success = verify_migration(client, collection_name, new_collection_name)

ifsuccess:
print(f"
Migration successful for{collection_name}!")
print(f"New collection:{new_collection_name}")

# Automatically replace old collection with migrated one
try:
replace_old_collection(
client, collection_name, new_collection_name
)
exceptExceptionase:
print(
f"
Warning: Could not automatically replace collection:{e}"
)
print(f"
To activate manually:")
print(f"1. Delete the old collection:{collection_name}")
print(f"2. Rename{new_collection_name}to{collection_name}")

exceptExceptionase:
print(f"
Error migrating{collection_name}:{e}")
print(f"Skipping this collection and continuing...")
continue

print("
"+"="*80)
print("Migration Complete!")
print("="*80)
print("
Summary:")
print(f" Collections migrated:{len(collections_to_migrate)}")

finally:
client.close()

if__name__ =="__main__":
try:
migrate_all_collections()
exceptKeyboardInterrupt:
print("

Migration interrupted by user.")
sys.exit(1)
exceptExceptionase:
print(f"

Fatal error:{e}")
importtraceback