feat: trigger task record

Author: shaohuzhang1

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

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

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(

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',
+        }

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'),
+                }

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'),

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()
 

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,

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