refactor(workflow): improve handling of outputs and inputs

- Use empty dictionaries instead of None for default outputs and inputs.
- Replace manual json loads with dedicated getter methods for inputs and outputs.
- Ensure type hints for converted input and output dictionaries.
- Improve code readability and reduce potential null errors.

api/core/app/task_pipeline/workflow_cycle_manage.py

@@ -135,7 +135,7 @@ class WorkflowCycleManage:
         outputs = WorkflowEntry.handle_special_values(outputs)
 
         workflow_run.status = WorkflowRunStatus.SUCCEEDED.value
-        workflow_run.outputs = json.dumps(outputs) if outputs else None
+        workflow_run.outputs = json.dumps(outputs or {})
         workflow_run.elapsed_time = time.perf_counter() - start_at
         workflow_run.total_tokens = total_tokens
         workflow_run.total_steps = total_steps

api/core/ops/ops_trace_manager.py

@@ -358,8 +358,8 @@ class TraceTask:
         workflow_run_id = workflow_run.id
         workflow_run_elapsed_time = workflow_run.elapsed_time
         workflow_run_status = workflow_run.status
-        workflow_run_inputs = json.loads(workflow_run.inputs) if workflow_run.inputs else {}
-        workflow_run_outputs = json.loads(workflow_run.outputs) if workflow_run.outputs else {}
+        workflow_run_inputs = workflow_run.inputs_dict
+        workflow_run_outputs = workflow_run.outputs_dict
         workflow_run_version = workflow_run.version
         error = workflow_run.error or ""
 

api/models/workflow.py

@@ -391,7 +391,7 @@ class WorkflowRun(db.Model):
     graph = db.Column(db.Text)
     inputs = db.Column(db.Text)
     status = db.Column(db.String(255), nullable=False)
-    outputs = db.Column(db.Text)
+    outputs: Mapped[str] = db.Column(db.Text)
     error = db.Column(db.Text)
     elapsed_time = db.Column(db.Float, nullable=False, server_default=db.text("0"))
     total_tokens = db.Column(db.Integer, nullable=False, server_default=db.text("0"))
@@ -415,15 +415,15 @@ class WorkflowRun(db.Model):
 
     @property
     def graph_dict(self):
-        return json.loads(self.graph) if self.graph else None
+        return json.loads(self.graph) if self.graph else {}
 
     @property
-    def inputs_dict(self):
-        return json.loads(self.inputs) if self.inputs else None
+    def inputs_dict(self) -> Mapping[str, Any]:
+        return json.loads(self.inputs) if self.inputs else {}
 
     @property
-    def outputs_dict(self):
-        return json.loads(self.outputs) if self.outputs else None
+    def outputs_dict(self) -> Mapping[str, Any]:
+        return json.loads(self.outputs) if self.outputs else {}
 
     @property
     def message(self) -> Optional["Message"]: