数据库字符集修改为utf8mb4,重新知识库向量失败则删除es对应的数据

This commit is contained in:
Wenruli
2025-09-12 17:58:40 +08:00
parent 65942e7acf
commit dd685acc7f
3 changed files with 33 additions and 13 deletions
@@ -50,6 +50,12 @@ class ChatMessage(MessageBase, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
receiver: Optional[Dict] = Field(default=None, sa_column=Column(JSON))
# 关键:设置表级别字符集为 utf8mb4
__table_args__ = {
"mysql_charset": "utf8mb4",
"mysql_collate": "utf8mb4_unicode_ci"
}
class ChatMessageRead(MessageBase):
id: Optional[int] = None
+3 -1
View File
@@ -29,17 +29,19 @@ class DatabaseService(Service):
def _create_engine(self):
connect_args = {'check_same_thread': False} if self.database_url.startswith("sqlite") else {}
connect_args['charset'] = "utf8mb4" # Ensure utf8mb4 charset for MySQL
return create_engine(
self.database_url,
connect_args=connect_args,
pool_size=100,
max_overflow=20,
pool_timeout=3,
pool_pre_ping=True
pool_pre_ping=True,
)
def _create_async_engine(self):
connect_args = {'check_same_thread': False} if self.async_database_url.startswith("sqlite") else {}
connect_args['charset'] = "utf8mb4" # Ensure utf8mb4 charset for MySQL
return create_async_engine(
self.async_database_url,
connect_args=connect_args,
@@ -86,6 +86,10 @@ def rebuild_knowledge_celery(knowledge_id: int, new_model_id: str) -> str:
# 5. 更新knowledge状态
if failed_files:
# 删除es索引和milvus集合,避免数据不一致
_delete_es_files(knowledge, failed_files)
knowledge.state = KnowledgeState.FAILED.value
logger.error(f"knowledge_id={knowledge_id} rebuild failed, failed_files={failed_files}")
else:
@@ -110,24 +114,32 @@ def rebuild_knowledge_celery(knowledge_id: int, new_model_id: str) -> str:
raise e
def _delete_milvus_collection(knowledge: Knowledge):
"""删除Milvus集合"""
def _delete_es_files(knowledge: Knowledge, file_ids: List[int]):
"""删除ES中的文件数据"""
try:
index_name = knowledge.index_name or knowledge.collection_name
embeddings = FakeEmbedding()
vector_client = decide_vectorstores(knowledge.collection_name, "Milvus", embeddings)
es_client = decide_vectorstores(index_name, "ElasticKeywordsSearch", embeddings)
if isinstance(vector_client.col, Collection):
logger.info(f"Deleting milvus collection: {knowledge.collection_name}")
vector_client.col.drop(timeout=10)
logger.info(f"Successfully deleted milvus collection: {knowledge.collection_name}")
else:
logger.warning(f"Milvus collection not found: {knowledge.collection_name}")
if not es_client.client.indices.exists(index=index_name):
logger.warning(f"ES index {index_name} does not exist, skipping deletion")
return
vector_client.close_connection(vector_client.alias)
for file_id in file_ids:
delete_query = {
"query": {
"match": {
"metadata.file_id": file_id
}
}
}
response = es_client.client.delete_by_query(index=index_name, body=delete_query)
deleted = response.get("deleted", 0)
logger.info(f"Deleted {deleted} documents from ES for file_id={file_id}")
except Exception as e:
logger.warning(f"Failed to delete milvus collection {knowledge.collection_name}: {str(e)}")
# 即使删除失败也继续执行,因为可能集合本来就不存在
logger.exception(f"Failed to delete ES files for knowledge_id={knowledge.id}: {str(e)}")
def _rebuild_embeddings(knowledge: Knowledge, files: List[KnowledgeFile], new_model_id: str) -> tuple[