diff --git a/api/apps/services/memory_api_service.py b/api/apps/services/memory_api_service.py index 1555ce97dc..7d5955407a 100644 --- a/api/apps/services/memory_api_service.py +++ b/api/apps/services/memory_api_service.py @@ -131,6 +131,20 @@ async def update_memory(memory_id: str, new_memory_setting: dict): """ current_memory = _require_memory_access(memory_id) + def _normalize_memory_type(value): + if value is None: + return [] + if isinstance(value, int): + return sorted(get_memory_type_human(value)) + if isinstance(value, list): + return sorted(str(v).strip().lower() for v in value if str(v).strip()) + return sorted(str(value).strip().lower().split(",")) + + def _normalize_str(value): + if value is None: + return "" + return str(value).strip() + update_dict = {} # check name length if "name" in new_memory_setting: @@ -186,8 +200,21 @@ async def update_memory(memory_id: str, new_memory_setting: dict): memory_dict.update({"memory_type": get_memory_type_human(current_memory.memory_type)}) to_update = {} for k, v in update_dict.items(): - if isinstance(v, list) and set(memory_dict[k]) != set(v): - to_update[k] = v + if k == "memory_type": + current_value = _normalize_memory_type(memory_dict.get(k)) + new_value = _normalize_memory_type(v) + if current_value != new_value: + to_update[k] = new_value + elif k == "embd_id": + current_value = _normalize_str(memory_dict.get(k)) + new_value = _normalize_str(v) + if current_value != new_value: + to_update[k] = new_value + elif isinstance(v, list): + current_value = sorted(str(item).strip() for item in memory_dict.get(k, [])) + new_value = sorted(str(item).strip() for item in v) + if current_value != new_value: + to_update[k] = v elif memory_dict[k] != v: to_update[k] = v diff --git a/api/db/__init__.py b/api/db/__init__.py index 6aa7c5bbf0..ffcd8e7b3d 100644 --- a/api/db/__init__.py +++ b/api/db/__init__.py @@ -17,6 +17,8 @@ from enum import IntEnum from enum import StrEnum +from common.constants import PipelineTaskType + class UserTenantRole(StrEnum): OWNER = 'owner' @@ -59,14 +61,6 @@ class CanvasCategory(StrEnum): DataFlow = "dataflow_canvas" -class PipelineTaskType(StrEnum): - PARSE = "Parse" - DOWNLOAD = "Download" - RAPTOR = "RAPTOR" - GRAPH_RAG = "GraphRAG" - MINDMAP = "Mindmap" - - VALID_PIPELINE_TASK_TYPES = {PipelineTaskType.PARSE, PipelineTaskType.DOWNLOAD, PipelineTaskType.RAPTOR, PipelineTaskType.GRAPH_RAG, PipelineTaskType.MINDMAP} diff --git a/api/db/joint_services/memory_message_service.py b/api/db/joint_services/memory_message_service.py index a25ed86a37..40b830db59 100644 --- a/api/db/joint_services/memory_message_service.py +++ b/api/db/joint_services/memory_message_service.py @@ -14,6 +14,7 @@ # limitations under the License. # import logging +from datetime import datetime from typing import List from common import settings @@ -349,8 +350,9 @@ async def queue_save_to_memory_task(memory_ids: list[str], message_dict: dict): "doc_id": _memory_id, "task_type": "memory", "progress": 0.0, + "begin_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "digest": str(_source_id) - } + } not_found_memory = [] failed_memory = [] @@ -387,6 +389,7 @@ async def queue_save_to_memory_task(memory_ids: list[str], message_dict: dict): "task_id": task["id"], "task_type": task["task_type"], "memory_id": memory_id, + "tenant_id": memory.tenant_id, "source_id": raw_message_id, "message_dict": message_dict } diff --git a/api/db/services/pipeline_operation_log_service.py b/api/db/services/pipeline_operation_log_service.py index ad90acb1f3..6c766f859c 100644 --- a/api/db/services/pipeline_operation_log_service.py +++ b/api/db/services/pipeline_operation_log_service.py @@ -20,14 +20,14 @@ from datetime import datetime, timedelta from peewee import fn -from api.db import VALID_PIPELINE_TASK_TYPES, PipelineTaskType +from api.db import VALID_PIPELINE_TASK_TYPES from api.db.db_models import DB, Document, PipelineOperationLog from api.db.services.canvas_service import UserCanvasService from api.db.services.common_service import CommonService from api.db.services.document_service import DocumentService from api.db.services.knowledgebase_service import KnowledgebaseService from api.db.services.task_service import GRAPH_RAPTOR_FAKE_DOC_ID, TaskService -from common.constants import TaskStatus +from common.constants import PipelineTaskType, TaskStatus from common.misc_utils import get_uuid from common.time_utils import current_timestamp, datetime_format diff --git a/api/db/services/task_service.py b/api/db/services/task_service.py index 1ce0cd4dd6..10ff8c7c29 100644 --- a/api/db/services/task_service.py +++ b/api/db/services/task_service.py @@ -343,8 +343,10 @@ class TaskService(CommonService): ((prog == -1) | (prog > cls.model.progress)))) ).execute() - process_duration = (datetime.now() - task.begin_at).total_seconds() - cls.model.update(process_duration=process_duration).where(cls.model.id == id).execute() + begin_at = task.begin_at + if begin_at is not None: + process_duration = (datetime.now() - begin_at).total_seconds() + cls.model.update(process_duration=process_duration).where(cls.model.id == id).execute() @classmethod @DB.connection_context() diff --git a/rag/svr/task_executor.py b/rag/svr/task_executor.py index a27714d6ef..c3b720811b 100644 --- a/rag/svr/task_executor.py +++ b/rag/svr/task_executor.py @@ -256,6 +256,8 @@ async def collect(): task["kb_id"] = msg.get("kb_id", "") if task_type[:6] == "memory": task["memory_id"] = msg["memory_id"] + if msg.get("tenant_id"): + task["tenant_id"] = msg["tenant_id"] task["source_id"] = msg["source_id"] task["message_dict"] = msg["message_dict"] return redis_msg, task