feat: trigger task record

shaohuzhang1 · shaohuzhang1 · commit b2f06edabaa5 · 2026-01-23T16:49:08.000+08:00
diff --git a/apps/application/chat_pipeline/I_base_chat_pipeline.py b/apps/application/chat_pipeline/I_base_chat_pipeline.py
@@ -9,7 +9,7 @@
 import time
 from abc import abstractmethod
 from typing import Type
-
+import uuid_utils.compat as uuid
 from rest_framework import serializers
 
 from knowledge.models import Paragraph
@@ -128,6 +128,8 @@ class IBaseChatPipelineStep:
     def __init__(self):
         # 当前步骤上下文,用于存储当前步骤信息
         self.context = {}
+        self.status = 200
+        self.err_message = ''
 
     @abstractmethod
     def get_step_serializer(self, manage) -> Type[serializers.Serializer]:
@@ -145,12 +147,29 @@ def run(self, manage):
         :param manage:      步骤管理器
         :return: 执行结果
         """
-        start_time = time.time()
-        self.context['start_time'] = start_time
-        # 校验参数,
-        self.valid_args(manage)
-        self._run(manage)
-        self.context['run_time'] = time.time() - start_time
+        try:
+            start_time = time.time()
+            self.context['start_time'] = start_time
+            # 校验参数,
+            self.valid_args(manage)
+            self._run(manage)
+            self.context['run_time'] = time.time() - start_time
+        except Exception as e:
+            self.err_message = str(e)
+            self.status = 500
+            chat_record_id = manage.context.get('chat_record_id') or str(uuid.uuid7())
+            manage.context['message_tokens'] = 0
+            manage.context['answer_tokens'] = 0
+            end_time = time.time()
+            manage.context['run_time'] = end_time - (manage.context.get('start_time') or end_time)
+            post_response_handler = manage.context.get('post_response_handler')
+            post_response_handler.handler(manage.context.get('chat_id'), chat_record_id,
+                                          manage.context.get('paragraph_list') or [],
+                                          manage.context.get('problem_text'),
+                                          str(e), manage, self, manage.context.get('padding_problem_text'),
+                                          reasoning_content='')
+
+            raise e
 
     def _run(self, manage):
         pass
diff --git a/apps/application/chat_pipeline/pipeline_manage.py b/apps/application/chat_pipeline/pipeline_manage.py
@@ -21,6 +21,7 @@ def __init__(self, step_list: List[Type[IBaseChatPipelineStep]],
                  debug=False):
         # 步骤执行器
         self.step_list = [step() for step in step_list]
+        self.run_step_list = []
         # 上下文
         self.context = {'message_tokens': 0, 'answer_tokens': 0}
         self.base_to_response = base_to_response
@@ -32,12 +33,13 @@ def run(self, context: Dict = None):
             for key, value in context.items():
                 self.context[key] = value
         for step in self.step_list:
+            self.run_step_list.append(step)
             step.run(self)
 
     def get_details(self):
         return reduce(lambda x, y: {**x, **y}, [{item.get('step_type'): item} for item in
                                                 filter(lambda r: r is not None,
-                                                       [row.get_details(self) for row in self.step_list])], {})
+                                                       [row.get_details(self) for row in self.run_step_list])], {})
 
     def get_base_to_response(self):
         return self.base_to_response
diff --git a/apps/application/chat_pipeline/step/chat_step/impl/base_chat_step.py b/apps/application/chat_pipeline/step/chat_step/impl/base_chat_step.py
@@ -204,8 +204,10 @@ def execute(self, message_list: List[BaseMessage],
 
     def get_details(self, manage, **kwargs):
         return {
+            'status': self.status,
+            'err_message': self.err_message,
             'step_type': 'chat_step',
-            'run_time': self.context['run_time'],
+            'run_time': self.context.get('run_time') or 0,
             'model_id': str(manage.context['model_id']),
             'message_list': self.reset_message_list(self.context['step_args'].get('message_list'),
                                                     self.context['answer_text']),
@@ -273,7 +275,8 @@ def _handle_mcp_request(self, mcp_source, mcp_servers, mcp_tool_ids, tool_ids,
                     if application_access_token is not None and application_access_token.authentication:
                         raise AppApiException(
                             500,
-                            _('Agent 【{name}】 access token authentication is not supported for agent tool').format(name=app.name)
+                            _('Agent 【{name}】 access token authentication is not supported for agent tool').format(
+                                name=app.name)
                         )
                 else:
                     raise AppApiException(
diff --git a/apps/application/chat_pipeline/step/generate_human_message_step/impl/base_generate_human_message_step.py b/apps/application/chat_pipeline/step/generate_human_message_step/impl/base_generate_human_message_step.py
@@ -71,3 +71,10 @@ def to_human_message(prompt: str,
                 data_list.append(f"<data>{content}</data>")
         data = "\n".join(data_list)
         return HumanMessage(content=prompt.replace('{data}', data).replace('{question}', problem))
+
+    def get_details(self, manage, **kwargs):
+        return {
+            'status': self.status,
+            'err_message': self.err_message,
+            'step_type': 'generate_human_message',
+        }
diff --git a/apps/application/chat_pipeline/step/reset_problem_step/impl/base_reset_problem_step.py b/apps/application/chat_pipeline/step/reset_problem_step/impl/base_reset_problem_step.py
@@ -56,13 +56,14 @@ def execute(self, problem_text: str, history_chat_record: List[ChatRecord] = Non
         return padding_problem
 
     def get_details(self, manage, **kwargs):
-        return {
-            'step_type': 'problem_padding',
-            'run_time': self.context['run_time'],
-            'model_id': str(manage.context['model_id']) if 'model_id' in manage.context else None,
-            'message_tokens': self.context.get('message_tokens', 0),
-            'answer_tokens': self.context.get('answer_tokens', 0),
-            'cost': 0,
-            'padding_problem_text': self.context.get('padding_problem_text'),
-            'problem_text': self.context.get("step_args").get('problem_text'),
-        }
+        return {'status': self.status,
+                'err_message': self.err_message,
+                'step_type': 'problem_padding',
+                'run_time': self.context['run_time'],
+                'model_id': str(manage.context['model_id']) if 'model_id' in manage.context else None,
+                'message_tokens': self.context.get('message_tokens', 0),
+                'answer_tokens': self.context.get('answer_tokens', 0),
+                'cost': 0,
+                'padding_problem_text': self.context.get('padding_problem_text'),
+                'problem_text': self.context.get("step_args").get('problem_text'),
+                }
diff --git a/apps/application/chat_pipeline/step/search_dataset_step/impl/base_search_dataset_step.py b/apps/application/chat_pipeline/step/search_dataset_step/impl/base_search_dataset_step.py
@@ -68,7 +68,8 @@ def execute(self, problem_text: str, knowledge_id_list: list[str], exclude_docum
         embedding_model = ModelManage.get_model(model_id, lambda _id: get_model(model, **{**default_params}))
         embedding_value = embedding_model.embed_query(exec_problem_text)
         vector = VectorStore.get_embedding_vector()
-        embedding_list = vector.query(exec_problem_text, embedding_value, knowledge_id_list, None, exclude_document_id_list,
+        embedding_list = vector.query(exec_problem_text, embedding_value, knowledge_id_list, None,
+                                      exclude_document_id_list,
                                       exclude_paragraph_id_list, True, top_n, similarity, SearchMode(search_mode))
         if embedding_list is None:
             return []
@@ -132,12 +133,14 @@ def list_paragraph(embedding_list: List, vector):
         return paragraph_list
 
     def get_details(self, manage, **kwargs):
-        step_args = self.context['step_args']
+        step_args = self.context.get('step_args') or {}
 
         return {
+            'status': self.status,
+            'err_message': self.err_message,
             'step_type': 'search_step',
-            'paragraph_list': [row.to_dict() for row in self.context['paragraph_list']],
-            'run_time': self.context['run_time'],
+            'paragraph_list': [row.to_dict() for row in (self.context.get('paragraph_list') or [])],
+            'run_time': self.context.get('run_time') or 0,
             'problem_text': step_args.get(
                 'padding_problem_text') if 'padding_problem_text' in step_args else step_args.get('problem_text'),
             'model_name': self.context.get('model_name'),
diff --git a/apps/application/serializers/common.py b/apps/application/serializers/common.py
@@ -263,7 +263,7 @@ def append_chat_record(self, chat_record: ChatRecord):
                                                                    'improve_paragraph_id_list': chat_record.improve_paragraph_id_list,
                                                                    'run_time': chat_record.run_time,
                                                                    'source': chat_record.source,
-                                                                   'ip_address': chat_record.ip_address,
+                                                                   'ip_address': chat_record.ip_address or '',
                                                                    'index': chat_record.index},
                                                   defaults={
                                                       "vote_status": chat_record.vote_status,
@@ -278,7 +278,7 @@ def append_chat_record(self, chat_record: ChatRecord):
                                                       'run_time': chat_record.run_time,
                                                       'index': chat_record.index,
                                                       'source': chat_record.source,
-                                                      'ip_address': chat_record.ip_address,
+                                                      'ip_address': chat_record.ip_address or '',
                                                   })
             ChatCountSerializer(data={'chat_id': self.chat_id}).update_chat()
 
diff --git a/apps/chat/serializers/chat.py b/apps/chat/serializers/chat.py
@@ -36,7 +36,7 @@
 from common.handle.base_to_response import BaseToResponse
 from common.handle.impl.response.openai_to_response import OpenaiToResponse
 from common.handle.impl.response.system_to_response import SystemToResponse
-from common.utils.common import flat_map, get_file_content
+from common.utils.common import flat_map, get_file_content, is_valid_uuid
 from knowledge.models import Document, Paragraph
 from maxkb.conf import PROJECT_DIR
 from models_provider.models import Model, Status
@@ -381,7 +381,8 @@ def get_chat_record(chat_info, chat_record_id):
                 return chat_record_list[-1]
         chat_record = QuerySet(ChatRecord).filter(id=chat_record_id, chat_id=chat_info.chat_id).first()
         if chat_record is None:
-            raise ChatException(500, _("Conversation record does not exist"))
+            if not is_valid_uuid(chat_record_id):
+                raise ChatException(500, _("Conversation record does not exist"))
         chat_record = QuerySet(ChatRecord).filter(id=chat_record_id).first()
         return chat_record
 
@@ -406,12 +407,13 @@ def chat_work_flow(self, chat_info: ChatInfo, instance: dict, base_to_response):
         history_chat_record = chat_info.chat_record_list
         if chat_record_id is not None:
             chat_record = self.get_chat_record(chat_info, chat_record_id)
-            history_chat_record = [r for r in chat_info.chat_record_list if str(r.id) != chat_record_id]
+            if chat_record:
+                history_chat_record = [r for r in chat_info.chat_record_list if str(r.id) != chat_record_id]
         work_flow = chat_info.application.work_flow
         work_flow_manage = WorkflowManage(Workflow.new_instance(work_flow),
                                           {'history_chat_record': history_chat_record, 'question': message,
                                            'chat_id': chat_info.chat_id, 'chat_record_id': str(
-                                              uuid.uuid7()) if chat_record is None else str(chat_record.id),
+                                              uuid.uuid7()) if chat_record_id is None else str(chat_record_id),
                                            'stream': stream,
                                            're_chat': re_chat,
                                            'chat_user_id': chat_user_id,
diff --git a/apps/common/utils/common.py b/apps/common/utils/common.py
@@ -351,3 +351,12 @@ def filter_special_character(_str):
     for t in s_list:
         _str = _str.replace(t, '')
     return _str
+
+
+def is_valid_uuid(uuid_string):
+    """判断字符串是否为有效的UUID"""
+    try:
+        uuid_obj = uuid.UUID(uuid_string)
+        return str(uuid_obj) == uuid_string
+    except ValueError:
+        return False
diff --git a/apps/trigger/handler/impl/task/application_task.py b/apps/trigger/handler/impl/task/application_task.py
@@ -6,11 +6,15 @@
     @date：2026/1/14 19:14
     @desc:
 """
-import uuid
 
-from application.models import ChatUserType, Chat
+import uuid_utils.compat as uuid
+from django.db.models import QuerySet
+
+from application.models import ChatUserType, Chat, ChatRecord, ChatSourceChoices
 from chat.serializers.chat import ChatSerializers
+from knowledge.models import State
 from trigger.handler.base_task import BaseTriggerTask
+from trigger.models import TaskRecord
 
 
 def get_reference(fields, obj):
@@ -43,15 +47,34 @@ def get_application_execute_parameters(parameter_setting, kwargs):
             parameters[field] = get_field_value(field_setting, kwargs)
     api_input_field_list = parameter_setting.get('api_input_field_list')
     if api_input_field_list:
-        for key, value in api_input_field_list.item():
+        for key, value in api_input_field_list.items():
             parameters['form_data'][key] = get_field_value(value, kwargs)
     user_input_field_list = parameter_setting.get('user_input_field_list')
     if user_input_field_list:
-        for key, value in user_input_field_list.item():
+        for key, value in user_input_field_list.items():
             parameters['form_data'][key] = get_field_value(value, kwargs)
     return parameters
 
 
+def get_loop_workflow_node(node_list):
+    result = []
+    for item in node_list:
+        if item.get('type') == 'loop-node':
+            for loop_item in item.get('loop_node_data') or []:
+                for inner_item in loop_item.values():
+                    result.append(inner_item)
+    return result
+
+
+def get_workflow_state(details):
+    node_list = details.values()
+    all_node = [*node_list, *get_loop_workflow_node(node_list)]
+    err = any([True for value in all_node if value.get('status') == 500 and not value.get('enableException')])
+    if err:
+        return State.FAILURE
+    return State.SUCCESS
+
+
 class ApplicationTask(BaseTriggerTask):
     def support(self, trigger_task, **kwargs):
         return trigger_task.get('source_type') == 'APPLICATION'
@@ -61,22 +84,39 @@ def execute(self, trigger_task, **kwargs):
         parameters = get_application_execute_parameters(parameter_setting, kwargs)
         parameters['re_chat'] = False
         parameters['stream'] = True
-        chat_id = uuid.uuid1()
-        chat_user_id = str(uuid.uuid1())
+        chat_id = uuid.uuid7()
+        chat_user_id = str(uuid.uuid7())
+        chat_record_id = str(uuid.uuid7())
+        parameters['chat_record_id'] = chat_record_id
         application_id = trigger_task.get('source_id')
         message = parameters.get('message')
         Chat.objects.get_or_create(id=chat_id, defaults={
             'application_id': application_id,
             'abstract': message,
             'chat_user_id': chat_user_id,
             'chat_user_type': ChatUserType.ANONYMOUS_USER.value,
-            'asker': {'username': "游客"}
+            'asker': {'username': "游客"},
+            'source': {
+                'type': ChatSourceChoices.TRIGGER.value
+            },
         })
-
-        list(ChatSerializers(data={
-            "chat_id": chat_id,
-            "chat_user_id": chat_user_id,
-            'chat_user_type': ChatUserType.ANONYMOUS_USER.value,
-            'application_id': application_id,
-            'debug': False
-        }).chat(instance=parameters))
+        task_record_id = uuid.uuid7()
+        TaskRecord(id=task_record_id, source_type="APPLICATION", source_id=application_id,
+                   task_record_id=chat_record_id,
+                   meta={'chat_id': chat_id},
+                   state=State.STARTED).save()
+        try:
+            list(ChatSerializers(data={
+                "chat_id": chat_id,
+                "chat_user_id": chat_user_id,
+                'chat_user_type': ChatUserType.ANONYMOUS_USER.value,
+                'application_id': application_id,
+                'source': {
+                    'type': ChatSourceChoices.TRIGGER.value
+                },
+                'debug': False
+            }).chat(instance=parameters))
+        finally:
+            chat_record = QuerySet(ChatRecord).filter(id=chat_record_id).first()
+            state = get_workflow_state(chat_record.details)
+            QuerySet(TaskRecord).filter(id=task_record_id).update(state=state, run_time=chat_record.run_time)
diff --git a/apps/trigger/handler/simple_tools.py b/apps/trigger/handler/simple_tools.py
@@ -6,6 +6,8 @@
     @date：2026/1/14 19:18
     @desc:
 """
+from threading import Thread
+
 from trigger.handler.impl.task.application_task import ApplicationTask
 from trigger.handler.impl.trigger.scheduled_trigger import ScheduledTrigger
 
@@ -23,7 +25,8 @@ def execute(trigger_task, **kwargs):
     """
     for simple_task_handler in simple_task_handlers:
         if simple_task_handler.support(trigger_task, **kwargs):
-            return simple_task_handler.execute(trigger_task, **kwargs)
+            Thread(target=simple_task_handler.execute, args=(trigger_task,), kwargs=kwargs).start()
+            return
     raise Exception("不支持的处理器类型")
 
 
diff --git a/apps/trigger/migrations/0002_taskrecord.py b/apps/trigger/migrations/0002_taskrecord.py
@@ -0,0 +1,32 @@
+# Generated by Django 5.2.9 on 2026-01-22 09:32
+
+import common.encoder.encoder
+import uuid_utils.compat
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('trigger', '0001_initial'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='TaskRecord',
+            fields=[
+                ('create_time', models.DateTimeField(auto_now_add=True, db_index=True, verbose_name='创建时间')),
+                ('update_time', models.DateTimeField(auto_now=True, db_index=True, verbose_name='修改时间')),
+                ('id', models.UUIDField(default=uuid_utils.compat.uuid7, editable=False, primary_key=True, serialize=False, verbose_name='主键id')),
+                ('source_type', models.CharField(choices=[('APPLICATION', 'Application'), ('TOOL', 'Tool')], default='APPLICATION', max_length=256, verbose_name='触发器任务类型')),
+                ('source_id', models.UUIDField(blank=True, null=True, verbose_name='资源id')),
+                ('task_record_id', models.UUIDField(blank=True, verbose_name='任务记录id')),
+                ('meta', models.JSONField(default=dict, encoder=common.encoder.encoder.SystemEncoder)),
+                ('state', models.CharField(choices=[('PENDING', 'Pending'), ('STARTED', 'Started'), ('SUCCESS', 'Success'), ('FAILURE', 'Failure'), ('REVOKE', 'Revoke'), ('REVOKED', 'Revoked')], default='STARTED', max_length=20, verbose_name='状态')),
+                ('run_time', models.FloatField(default=0, verbose_name='运行时长')),
+            ],
+            options={
+                'db_table': 'event_trigger_task_record',
+            },
+        ),
+    ]
diff --git a/apps/trigger/migrations/0003_alter_taskrecord_task_record_id.py b/apps/trigger/migrations/0003_alter_taskrecord_task_record_id.py
diff --git a/apps/trigger/migrations/0004_alter_triggertask_unique_together.py b/apps/trigger/migrations/0004_alter_triggertask_unique_together.py
diff --git a/apps/trigger/models/trigger.py b/apps/trigger/models/trigger.py
