add celery prefetch setting

api/core/workflow/nodes/knowledge_index/knowledge_index_node.py

@@ -136,6 +136,11 @@ class KnowledgeIndexNode(Node):
         document = db.session.query(Document).filter_by(id=document_id.value).first()
         if not document:
             raise KnowledgeIndexNodeError(f"Document {document_id.value} not found.")
+        doc_id_value = document.id
+        ds_id_value = dataset.id
+        dataset_name_value = dataset.name
+        document_name_value = document.name
+        created_at_value = document.created_at
         # chunk nodes by chunk size
         indexing_start_at = time.perf_counter()
         index_processor = IndexProcessorFactory(dataset.chunk_structure).init_index_processor()
@@ -161,16 +166,16 @@ class KnowledgeIndexNode(Node):
         document.word_count = (
             db.session.query(func.sum(DocumentSegment.word_count))
             .where(
-                DocumentSegment.document_id == document.id,
-                DocumentSegment.dataset_id == dataset.id,
+                DocumentSegment.document_id == doc_id_value,
+                DocumentSegment.dataset_id == ds_id_value,
             )
             .scalar()
         )
         db.session.add(document)
         # update document segment status
         db.session.query(DocumentSegment).where(
-            DocumentSegment.document_id == document.id,
-            DocumentSegment.dataset_id == dataset.id,
+            DocumentSegment.document_id == doc_id_value,
+            DocumentSegment.dataset_id == ds_id_value,
         ).update(
             {
                 DocumentSegment.status: "completed",
@@ -182,13 +187,13 @@ class KnowledgeIndexNode(Node):
         db.session.commit()
 
         return {
-            "dataset_id": dataset.id,
-            "dataset_name": dataset.name,
+            "dataset_id": ds_id_value,
+            "dataset_name": dataset_name_value,
             "batch": batch.value,
-            "document_id": document.id,
-            "document_name": document.name,
-            "created_at": document.created_at.timestamp(),
-            "display_status": document.indexing_status,
+            "document_id": doc_id_value,
+            "document_name": document_name_value,
+            "created_at": created_at_value.timestamp(),
+            "display_status": "completed",
         }
 
     def _get_preview_output(self, chunk_structure: str, chunks: Any) -> Mapping[str, Any]:

api/core/workflow/workflow_entry.py

@@ -416,4 +416,8 @@ class WorkflowEntry:
 
             # append variable and value to variable pool
             if variable_node_id != ENVIRONMENT_VARIABLE_NODE_ID:
+                # In single run, the input_value is set as the LLM's structured output value within the variable_pool.
+                if len(variable_key_list) == 2 and variable_key_list[0] == "structured_output":
+                    input_value = {variable_key_list[1]: input_value}
+                    variable_key_list = variable_key_list[0:1]
                 variable_pool.add([variable_node_id] + variable_key_list, input_value)

api/docker/entrypoint.sh

@@ -32,7 +32,8 @@ if [[ "${MODE}" == "worker" ]]; then
 
   exec celery -A celery_entrypoint.celery worker -P ${CELERY_WORKER_CLASS:-gevent} $CONCURRENCY_OPTION \
     --max-tasks-per-child ${MAX_TASKS_PER_CHILD:-50} --loglevel ${LOG_LEVEL:-INFO} \
-    -Q ${CELERY_QUEUES:-dataset,pipeline,mail,ops_trace,app_deletion,plugin,workflow_storage,conversation}
+    -Q ${CELERY_QUEUES:-dataset,pipeline,mail,ops_trace,app_deletion,plugin,workflow_storage,conversation} \
+    --prefetch-multiplier=1 --without-mingle --without-gossip
 
 elif [[ "${MODE}" == "beat" ]]; then
   exec celery -A app.celery beat --loglevel ${LOG_LEVEL:-INFO}

api/services/dataset_service.py

@@ -93,7 +93,7 @@ logger = logging.getLogger(__name__)
 class DatasetService:
     @staticmethod
     def get_datasets(page, per_page, tenant_id=None, user=None, search=None, tag_ids=None, include_all=False):
-        query = select(Dataset).where(Dataset.tenant_id == tenant_id).order_by(Dataset.created_at.desc())
+        query = select(Dataset).where(Dataset.tenant_id == tenant_id).order_by(Dataset.created_at.desc(), Dataset.id)
 
         if user:
             # get permitted dataset ids

web/app/components/base/file-uploader/store.tsx

@@ -1,7 +1,6 @@
 import {
   createContext,
   useContext,
-  useEffect,
   useRef,
 } from 'react'
 import {
@@ -19,11 +18,13 @@ type Shape = {
 
 export const createFileStore = (
   value: FileEntity[] = [],
+  onChange?: (files: FileEntity[]) => void,
 ) => {
   return create<Shape>(set => ({
     files: value ? [...value] : [],
     setFiles: (files) => {
       set({ files })
+      onChange?.(files)
     },
   }))
 }
@@ -54,35 +55,9 @@ export const FileContextProvider = ({
   onChange,
 }: FileProviderProps) => {
   const storeRef = useRef<FileStore | undefined>(undefined)
-  const onChangeRef = useRef<FileProviderProps['onChange']>(onChange)
-  const isSyncingRef = useRef(false)
 
   if (!storeRef.current)
-    storeRef.current = createFileStore(value)
-
-  // keep latest onChange
-  useEffect(() => {
-    onChangeRef.current = onChange
-  }, [onChange])
-
-  // subscribe to store changes and call latest onChange
-  useEffect(() => {
-    const store = storeRef.current!
-    const unsubscribe = store.subscribe((state: Shape) => {
-      if (isSyncingRef.current) return
-      onChangeRef.current?.(state.files)
-    })
-    return unsubscribe
-  }, [])
-
-  // sync external value into internal store when value changes
-  useEffect(() => {
-    const store = storeRef.current!
-    const nextFiles = value ? [...value] : []
-    isSyncingRef.current = true
-    store.setState({ files: nextFiles })
-    isSyncingRef.current = false
-  }, [value])
+    storeRef.current = createFileStore(value, onChange)
 
   return (
     <FileContext.Provider value={storeRef.current}>

web/app/components/explore/app-list/index.tsx

@@ -152,23 +152,20 @@ const Apps = ({
       <div className={cn(
         'mt-6 flex items-center justify-between px-12',
       )}>
-        <>
-          <Category
-            list={categories}
-            value={currCategory}
-            onChange={setCurrCategory}
-            allCategoriesEn={allCategoriesEn}
-          />
-        </>
+        <Category
+          list={categories}
+          value={currCategory}
+          onChange={setCurrCategory}
+          allCategoriesEn={allCategoriesEn}
+        />
         <Input
           showLeftIcon
           showClearIcon
-          wrapperClassName='w-[200px]'
+          wrapperClassName='w-[200px] self-start'
           value={keywords}
           onChange={e => handleKeywordsChange(e.target.value)}
           onClear={() => handleKeywordsChange('')}
         />
-
       </div>
 
       <div className={cn(