LQAgentPlatform/foundation/base/tasks.py

"""
Celery任务定义
只负责任务调度，具体业务逻辑由WorkflowManager处理
"""

from celery import current_task
from .celery_app import app
from core.base.workflow_manager import WorkflowManager
from foundation.logger.loggering import server_logger as logger
from foundation.utils.time_statistics import track_execution_time


@app.task(bind=True)
def submit_task_processing_task(self, file_info: dict):
    """
    提交任务处理到Celery队列
    这个任务只负责调用WorkflowManager，不包含业务逻辑
    """
    import traceback

    # 添加调试信息
    logger.info("=== Celery任务接收调试 ===")
    logger.info(f"队列ID: {self.request.id}")
    logger.info(f"文件ID: {file_info.get('file_id')}")
    logger.info(f"回调任务ID: {file_info.get('callback_task_id')}")
    logger.info("=== 任务接收调用栈 ===")
    for line in traceback.format_stack():
        logger.debug(f"  {line.strip()}")
    logger.info("=== 调用栈结束 ===")

    try:
        # 更新任务状态 - 开始处理
        self.update_state(
            state='PROGRESS',
            meta={
                'current': 0,
                'total': 100,
                'status': '开始处理文档',
                'file_id': file_info.get('file_id')
            }
        )

        logger.info(f"开始执行业务逻辑，文件ID: {file_info.get('file_id')}")

        # 创建独立的WorkflowManager实例执行业务逻辑
        workflow_manager = WorkflowManager(
            max_concurrent_docs=1,  # Celery worker中单任务执行
            max_concurrent_reviews=5
        )

        # 同步执行（Celery worker本身就是独立的进程）

        result = workflow_manager.submit_task_processing_sync(file_info)



        # 更新任务状态 - 完成
        self.update_state(
            state='PROGRESS',
            meta={
                'current': 100,
                'total': 100,
                'status': '处理完成',
                'file_id': file_info.get('file_id')
            }
        )


        return {
            'status': 'success',
            'file_id': file_info.get('file_id'),
            'callback_task_id': file_info.get('callback_task_id'),
            'result': result
        }

    except Exception as e:
        # 记录错误并重试
        logger.error(f"任务处理失败: {str(e)}")
        logger.exception("详细错误信息:")
        # 自动重试，延迟60秒，最多重试2次
        self.retry(countdown=60, max_retries=2, exc=e)
        raise