r2 transform

api/core/app/apps/pipeline/pipeline_runner.py

@@ -16,6 +16,7 @@ from core.workflow.entities.variable_pool import VariablePool
 from core.workflow.enums import SystemVariableKey
 from core.workflow.graph_engine.entities.event import GraphEngineEvent, GraphRunFailedEvent
 from core.workflow.graph_engine.entities.graph import Graph
+from core.workflow.system_variable import SystemVariable
 from core.workflow.workflow_entry import WorkflowEntry
 from extensions.ext_database import db
 from models.dataset import Document, Pipeline
@@ -127,7 +128,7 @@ class PipelineRunner(WorkflowBasedAppRunner):
                         )
 
             variable_pool = VariablePool(
-                system_variables=system_inputs,
+                system_variables=SystemVariable(**system_inputs),
                 user_inputs=inputs,
                 environment_variables=workflow.environment_variables,
                 conversation_variables=[],

api/core/workflow/system_variable.py

@@ -1,5 +1,5 @@
 from collections.abc import Sequence
-from typing import Any
+from typing import Any, Mapping
 
 from pydantic import AliasChoices, BaseModel, ConfigDict, Field, model_validator
 
@@ -43,6 +43,12 @@ class SystemVariable(BaseModel):
     query: str | None = None
     conversation_id: str | None = None
     dialogue_count: int | None = None
+    document_id: str | None = None
+    dataset_id: str | None = None
+    batch: str | None = None
+    datasource_type: str | None = None
+    datasource_info: Mapping[str, Any] | None = None
+    invoke_from: str | None = None
 
     @model_validator(mode="before")
     @classmethod
@@ -86,4 +92,16 @@ class SystemVariable(BaseModel):
             d[SystemVariableKey.CONVERSATION_ID] = self.conversation_id
         if self.dialogue_count is not None:
             d[SystemVariableKey.DIALOGUE_COUNT] = self.dialogue_count
+        if self.document_id is not None:
+            d[SystemVariableKey.DOCUMENT_ID] = self.document_id
+        if self.dataset_id is not None:
+            d[SystemVariableKey.DATASET_ID] = self.dataset_id
+        if self.batch is not None:
+            d[SystemVariableKey.BATCH] = self.batch
+        if self.datasource_type is not None:
+            d[SystemVariableKey.DATASOURCE_TYPE] = self.datasource_type
+        if self.datasource_info is not None:
+            d[SystemVariableKey.DATASOURCE_INFO] = self.datasource_info
+        if self.invoke_from is not None:
+            d[SystemVariableKey.INVOKE_FROM] = self.invoke_from
         return d