Underground-Digital/Workflow-Engine

1import logging2import time3 4from celery import shared_task5from flask import current_app6 7from core.ops.entities.trace_entity import trace_info_info_map8from core.rag.models.document import Document9from models.model import Message10from models.workflow import WorkflowRun11 12 13@shared_task(queue="ops_trace")14def process_trace_tasks(tasks_data):15    """16    Async process trace tasks17    :param tasks_data: List of dictionaries containing task data18 19    Usage: process_trace_tasks.delay(tasks_data)20    """21    from core.ops.ops_trace_manager import OpsTraceManager22 23    trace_info = tasks_data.get("trace_info")24    app_id = tasks_data.get("app_id")25    trace_info_type = tasks_data.get("trace_info_type")26    trace_instance = OpsTraceManager.get_ops_trace_instance(app_id)27 28    if trace_info.get("message_data"):29        trace_info["message_data"] = Message.from_dict(data=trace_info["message_data"])30    if trace_info.get("workflow_data"):31        trace_info["workflow_data"] = WorkflowRun.from_dict(data=trace_info["workflow_data"])32    if trace_info.get("documents"):33        trace_info["documents"] = [Document(**doc) for doc in trace_info["documents"]]34 35    try:36        if trace_instance:37            with current_app.app_context():38                trace_type = trace_info_info_map.get(trace_info_type)39                if trace_type:40                    trace_info = trace_type(**trace_info)41                trace_instance.trace(trace_info)42            end_at = time.perf_counter()43    except Exception:44        logging.exception("Processing trace tasks failed")45