feat: knowledge workflow

Author: shaohuzhang1

apps/application/flow/i_step_node.py

@@ -21,6 +21,7 @@
 from application.models import ApplicationChatUserStats
 from application.models import ChatRecord, ChatUserType
 from common.field.common import InstanceField
+from knowledge.models.knowledge_action import KnowledgeAction, State
 
 chat_cache = cache
 
@@ -97,11 +98,13 @@ def handler(self, workflow):
 
 
 class KnowledgeWorkflowPostHandler(WorkFlowPostHandler):
-    def __init__(self, chat_info):
+    def __init__(self, chat_info, knowledge_action_id):
         super().__init__(chat_info)
+        self.knowledge_action_id = knowledge_action_id
 
     def handler(self, workflow):
-        pass
+        QuerySet(KnowledgeAction).filter(id=self.knowledge_action_id).update(
+            state=State.SUCCESS)
 
 
 class NodeResult:
@@ -161,7 +164,8 @@ class FlowParamsSerializer(serializers.Serializer):
 
 
 class KnowledgeFlowParamsSerializer(serializers.Serializer):
-    knowledge_id = serializers.CharField(required=True, label="知识库id")
+    knowledge_id = serializers.UUIDField(required=True, label="知识库id")
+    knowledge_action_id = serializers.UUIDField(required=True, label="知识库任务执行器id")
     data_source = serializers.DictField(required=True, label="数据源")
     knowledge_base = serializers.DictField(required=False, label="知识库设置")
 
@@ -241,7 +245,8 @@ def get_write_error_context(self, e):
         self.status = 500
         self.answer_text = str(e)
         self.err_message = str(e)
-        self.context['run_time'] = time.time() - self.context['start_time']
+        current_time = time.time()
+        self.context['run_time'] = current_time - (self.context.get('start_time') or current_time)
 
         def write_error_context(answer, status=200):
             pass

apps/application/flow/knowledge_workflow_manage.py

@@ -6,12 +6,20 @@
     @date：2025/11/13 19:02
     @desc:
 """
+import traceback
+from concurrent.futures import ThreadPoolExecutor
+
+from django.db.models import QuerySet
+from django.utils.translation import get_language
 
 from application.flow.common import Workflow
 from application.flow.i_step_node import WorkFlowPostHandler, KnowledgeFlowParamsSerializer
 from application.flow.workflow_manage import WorkflowManage
 from common.handle.base_to_response import BaseToResponse
 from common.handle.impl.response.system_to_response import SystemToResponse
+from knowledge.models.knowledge_action import KnowledgeAction, State
+
+executor = ThreadPoolExecutor(max_workers=200)
 
 
 class KnowledgeWorkflowManage(WorkflowManage):
@@ -33,3 +41,46 @@ def get_start_node(self):
         start_node_list = [node for node in self.flow.nodes if
                            self.params.get('data_source', {}).get('node_id') == node.id]
         return start_node_list[0]
+
+    def run(self):
+        executor.submit(self._run)
+
+    def _run(self):
+        QuerySet(KnowledgeAction).filter(id=self.params.get('knowledge_action_id')).update(
+            state=State.STARTED)
+        language = get_language()
+        self.run_chain_async(self.start_node, None, language)
+        while self.is_run():
+            pass
+        self.work_flow_post_handler.handler(self)
+
+    def run_chain(self, current_node, node_result_future=None):
+        if node_result_future is None:
+            node_result_future = self.run_node_future(current_node)
+        try:
+            result = self.hand_node_result(current_node, node_result_future)
+            return result
+        except Exception as e:
+            traceback.print_exc()
+        return None
+
+    def hand_node_result(self, current_node, node_result_future):
+        try:
+            current_result = node_result_future.result()
+            result = current_result.write_context(current_node, self)
+            if result is not None:
+                # 阻塞获取结果
+                list(result)
+            return current_result
+        except Exception as e:
+            traceback.print_exc()
+            self.status = 500
+            current_node.get_write_error_context(e)
+            self.answer += str(e)
+            QuerySet(KnowledgeAction).filter(id=self.params.get('knowledge_action_id')).update(
+                details=self.get_runtime_details(),
+                state=State.FAILURE)
+        finally:
+            current_node.node_chunk.end()
+            QuerySet(KnowledgeAction).filter(id=self.params.get('knowledge_action_id')).update(
+                details=self.get_runtime_details())

apps/application/flow/step_node/data_source_local_node/impl/base_data_source_local_node.py

@@ -37,3 +37,15 @@ def get_form_list(node):
     def execute(self, file_type_list, file_size_limit, file_count_limit, **kwargs) -> NodeResult:
         return NodeResult({'file_list': self.workflow_manage.params.get('data_source', {}).get('file_list')},
                           self.workflow_manage.params.get('knowledge_base') or {})
+
+    def get_details(self, index: int, **kwargs):
+        return {
+            'name': self.node.properties.get('stepName'),
+            "index": index,
+            'run_time': self.context.get('run_time'),
+            'type': self.node.type,
+            'file_list': self.context.get('file_list'),
+            'knowledge_base': self.workflow_params.get('knowledge_base'),
+            'status': self.status,
+            'err_message': self.err_message
+        }

apps/knowledge/migrations/0005_knowledgeaction.py

@@ -0,0 +1,32 @@
+# Generated by Django 5.2.8 on 2025-11-19 06:06
+
+import common.encoder.encoder
+import django.db.models.deletion
+import uuid_utils.compat
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('knowledge', '0004_alter_document_type_alter_knowledge_type_and_more'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='KnowledgeAction',
+            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')),
+                ('state', models.CharField(choices=[('PENDING', 'Pending'), ('STARTED', 'Started'), ('SUCCESS', 'Success'), ('FAILURE', 'Failure'), ('REVOKE', 'Revoke'), ('REVOKED', 'Revoked')], default='STARTED', max_length=20, verbose_name='状态')),
+                ('details', models.JSONField(default=dict, encoder=common.encoder.encoder.SystemEncoder, verbose_name='执行详情')),
+                ('run_time', models.FloatField(default=0, verbose_name='运行时长')),
+                ('meta', models.JSONField(default=dict, verbose_name='元数据')),
+                ('knowledge', models.ForeignKey(db_constraint=False, on_delete=django.db.models.deletion.DO_NOTHING, to='knowledge.knowledge', verbose_name='知识库')),
+            ],
+            options={
+                'db_table': 'knowledge_action',
+            },
+        ),
+    ]

apps/knowledge/models/knowledge_action.py

@@ -0,0 +1,49 @@
+# coding=utf-8
+"""
+    @project: MaxKB
+    @Author：虎虎
+    @file： knowledge_action.py
+    @date：2025/11/18 17:59
+    @desc:
+"""
+import uuid_utils.compat as uuid
+
+from django.db import models
+
+from common.encoder.encoder import SystemEncoder
+from common.mixins.app_model_mixin import AppModelMixin
+from knowledge.models import Knowledge
+
+
+class State(models.TextChoices):
+    # 等待
+    PENDING = 'PENDING'
+    # 执行中
+    STARTED = 'STARTED'
+    # 成功
+    SUCCESS = 'SUCCESS'
+    # 失败
+    FAILURE = 'FAILURE'
+    # 取消任务
+    REVOKE = 'REVOKE'
+    # 取消成功
+    REVOKED = 'REVOKED'
+
+
+class KnowledgeAction(AppModelMixin):
+    id = models.UUIDField(primary_key=True, max_length=128, default=uuid.uuid7, editable=False, verbose_name="主键id")
+
+    knowledge = models.ForeignKey(Knowledge, on_delete=models.DO_NOTHING, verbose_name="知识库", db_constraint=False)
+
+    state = models.CharField(verbose_name='状态', max_length=20,
+                             choices=State.choices,
+                             default=State.STARTED)
+
+    details = models.JSONField(verbose_name="执行详情", default=dict, encoder=SystemEncoder)
+
+    run_time = models.FloatField(verbose_name="运行时长", default=0)
+
+    meta = models.JSONField(verbose_name="元数据", default=dict)
+
+    class Meta:
+        db_table = "knowledge_action"

apps/knowledge/serializers/knowledge_workflow.py

@@ -14,6 +14,7 @@
 from application.flow.step_node import get_node
 from common.exception.app_exception import AppApiException
 from knowledge.models import KnowledgeScope, Knowledge, KnowledgeType, KnowledgeWorkflow
+from knowledge.models.knowledge_action import KnowledgeAction, State
 from knowledge.serializers.knowledge import KnowledgeModelSerializer
 from system_manage.models import AuthTargetType
 from system_manage.serializers.user_resource_permission import UserResourcePermissionSerializer
@@ -34,13 +35,30 @@ def action(self, instance: Dict, with_valid=True):
         if with_valid:
             self.is_valid(raise_exception=True)
         knowledge_workflow = QuerySet(KnowledgeWorkflow).filter(knowledge_id=self.data.get("knowledge_id")).first()
+        knowledge_action_id = uuid.uuid7()
+        KnowledgeAction(id=knowledge_action_id, knowledge_id=self.data.get("knowledge_id"), state=State.STARTED).save()
         work_flow_manage = KnowledgeWorkflowManage(
             Workflow.new_instance(knowledge_workflow.work_flow, WorkflowMode.KNOWLEDGE),
-            {'knowledge_id': self.data.get("knowledge_id"), 'stream': True,
+            {'knowledge_id': self.data.get("knowledge_id"), 'knowledge_action_id': knowledge_action_id, 'stream': True,
              **instance},
-            KnowledgeWorkflowPostHandler(None))
-        r = work_flow_manage.run()
-        return r
+            KnowledgeWorkflowPostHandler(None, knowledge_action_id))
+        work_flow_manage.run()
+        return {'id': knowledge_action_id, 'knowledge_id': self.data.get("knowledge_id"), 'state': State.STARTED,
+                'details': {}}
+
+    class Operate(serializers.Serializer):
+        workspace_id = serializers.CharField(required=True, label=_('workspace id'))
+        knowledge_id = serializers.UUIDField(required=True, label=_('knowledge id'))
+        id = serializers.UUIDField(required=True, label=_('knowledge action id'))
+
+        def one(self, is_valid=True):
+            if is_valid:
+                self.is_valid(raise_exception=True)
+            knowledge_action_id = self.data.get("id")
+            knowledge_action = QuerySet(KnowledgeAction).filter(id=knowledge_action_id).first()
+            return {'id': knowledge_action_id, 'knowledge_id': knowledge_action.knowledge_id,
+                    'state': knowledge_action.state,
+                    'details': knowledge_action.details}
 
 
 class KnowledgeWorkflowSerializer(serializers.Serializer):

apps/knowledge/urls.py

@@ -70,6 +70,7 @@
     path('workspace/<str:workspace_id>/knowledge/<str:knowledge_id>/document/<int:current_page>/<int:page_size>', views.DocumentView.Page.as_view()),
     path('workspace/<str:workspace_id>/knowledge/<int:current_page>/<int:page_size>', views.KnowledgeView.Page.as_view()),
     path('workspace/<str:workspace_id>/knowledge/<str:knowledge_id>/form_list/<str:type>/<str:id>', views.KnowledgeWorkflowFormView.as_view()),
-    path('workspace/<str:workspace_id>/knowledge/<str:knowledge_id>/action', views.KnowledgeWorkflowActionView.as_view())
+    path('workspace/<str:workspace_id>/knowledge/<str:knowledge_id>/action', views.KnowledgeWorkflowActionView.as_view()),
+    path('workspace/<str:workspace_id>/knowledge/<str:knowledge_id>/action/<str:knowledge_action_id>', views.KnowledgeWorkflowActionView.Operate.as_view())
 
 ]

apps/knowledge/views/knowledge_workflow.py

@@ -27,8 +27,16 @@ class KnowledgeWorkflowActionView(APIView):
     authentication_classes = [TokenAuth]
 
     def post(self, request: Request, workspace_id: str, knowledge_id: str):
-        return KnowledgeWorkflowActionSerializer(
-            data={'workspace_id': workspace_id, 'knowledge_id': knowledge_id}).action(request.data, True)
+        return result.success(KnowledgeWorkflowActionSerializer(
+            data={'workspace_id': workspace_id, 'knowledge_id': knowledge_id}).action(request.data, True))
+
+    class Operate(APIView):
+        authentication_classes = [TokenAuth]
+
+        def get(self, request, workspace_id: str, knowledge_id: str, knowledge_action_id: str):
+            return result.success(KnowledgeWorkflowActionSerializer.Operate(
+                data={'workspace_id': workspace_id, 'knowledge_id': knowledge_id, 'id': knowledge_action_id})
+                                  .one())
 
 
 class KnowledgeWorkflowView(APIView):

ui/src/api/knowledge/knowledge.ts

@@ -337,7 +337,13 @@ const workflowAction: (
 ) => Promise<Result<any>> = (knowledge_id: string, instance, loading) => {
   return post(`${prefix.value}/${knowledge_id}/action`, instance, {}, loading)
 }
-
+const getWorkflowAction: (
+  knowledge_id: string,
+  knowledge_action_id: string,
+  loading?: Ref<boolean>,
+) => Promise<Result<any>> = (knowledge_id: string, knowledge_action_id, loading) => {
+  return get(`${prefix.value}/${knowledge_id}/action/${knowledge_action_id}`, {}, loading)
+}
 export default {
   getKnowledgeList,
   getKnowledgeListPage,
@@ -364,4 +370,5 @@ export default {
   createWorkflowKnowledge,
   getKnowledgeWorkflowFormList,
   workflowAction,
+  getWorkflowAction,
 }

ui/src/views/knowledge-workflow/component/action/Result.vue

@@ -0,0 +1,38 @@
+<template>
+  <div>
+    <ExecutionDetailContent :detail="detail" app-type="WORK_FLOW"></ExecutionDetailContent>
+  </div>
+</template>
+<script setup lang="ts">
+import { onUnmounted, ref, computed } from 'vue'
+import knowledgeApi from '@/api/knowledge/knowledge'
+const props = defineProps<{ id: string; knowledge_id: string }>()
+import ExecutionDetailContent from '@/components/ai-chat/component/knowledge-source-component/ExecutionDetailContent.vue'
+const detail = computed(() => {
+  if (knowledge_action.value) {
+    return Object.values(knowledge_action.value.details)
+  }
+  return []
+})
+const state = computed(() => {
+  if (knowledge_action.value) {
+    return knowledge_action.value.state
+  }
+  return 'PADDING'
+})
+const knowledge_action = ref<any>()
+const getKnowledgeWorkflowAction = () => {
+  knowledgeApi.getWorkflowAction(props.knowledge_id, props.id).then((ok) => {
+    knowledge_action.value = ok.data
+    if (['SUCCESS', 'FAILURE', 'REVOKED'].includes(state.value)) {
+      clearInterval(polling)
+    }
+  })
+}
+const polling = setInterval(getKnowledgeWorkflowAction, 2000)
+
+onUnmounted(() => {
+  clearInterval(polling)
+})
+</script>
+<style lang="scss"></style>