diff --git a/src/backend/bisheng/database/models/message.py b/src/backend/bisheng/database/models/message.py index 65c37aed5..bafe1f5c5 100644 --- a/src/backend/bisheng/database/models/message.py +++ b/src/backend/bisheng/database/models/message.py @@ -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 diff --git a/src/backend/bisheng/database/service.py b/src/backend/bisheng/database/service.py index 251c7e5d1..1b4a0c164 100644 --- a/src/backend/bisheng/database/service.py +++ b/src/backend/bisheng/database/service.py @@ -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, diff --git a/src/backend/bisheng/worker/knowledge/rebuild_knowledge_worker.py b/src/backend/bisheng/worker/knowledge/rebuild_knowledge_worker.py index 2a56f23c0..e46f12e33 100644 --- a/src/backend/bisheng/worker/knowledge/rebuild_knowledge_worker.py +++ b/src/backend/bisheng/worker/knowledge/rebuild_knowledge_worker.py @@ -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[