Feat/environment variables in workflow (langgenius#6515)

Co-authored-by: JzoNg <[email protected]>

.gitignore

@@ -174,5 +174,6 @@ sdks/python-client/dify_client.egg-info
 .vscode/*
 !.vscode/launch.json
 pyrightconfig.json
+api/.vscode
 
 .idea/

api/app.py

@@ -1,7 +1,5 @@
 import os
 
-from configs import dify_config
-
 if os.environ.get("DEBUG", "false").lower() != 'true':
     from gevent import monkey
 
@@ -23,7 +21,9 @@
 from flask_cors import CORS
 from werkzeug.exceptions import Unauthorized
 
+import contexts
 from commands import register_commands
+from configs import dify_config
 
 # DO NOT REMOVE BELOW
 from events import event_handlers
@@ -181,7 +181,10 @@ def load_user_from_request(request_from_flask_login):
     decoded = PassportService().verify(auth_token)
     user_id = decoded.get('user_id')
 
-    return AccountService.load_logged_in_account(account_id=user_id, token=auth_token)
+    account = AccountService.load_logged_in_account(account_id=user_id, token=auth_token)
+    if account:
+        contexts.tenant_id.set(account.current_tenant_id)
+    return account
 
 
 @login_manager.unauthorized_handler

api/configs/feature/__init__.py

@@ -406,7 +406,6 @@ class DataSetConfig(BaseSettings):
         default=False,
     )
 
-
 class WorkspaceConfig(BaseSettings):
     """
     Workspace configs

api/constants/__init__.py

@@ -0,0 +1,2 @@
+# TODO: Update all string in code to use this constant
+HIDDEN_VALUE = '[__HIDDEN__]'

api/contexts/__init__.py

@@ -0,0 +1,3 @@
+from contextvars import ContextVar
+
+tenant_id: ContextVar[str] = ContextVar('tenant_id')

api/controllers/console/app/app.py

@@ -212,7 +212,7 @@ def post(self, app_model):
         parser.add_argument('icon_background', type=str, location='json')
         args = parser.parse_args()
 
-        data = AppDslService.export_dsl(app_model=app_model)
+        data = AppDslService.export_dsl(app_model=app_model, include_secret=True)
         app = AppDslService.import_and_create_new_app(
             tenant_id=current_user.current_tenant_id,
             data=data,
@@ -234,8 +234,13 @@ def get(self, app_model):
         if not current_user.is_editor:
             raise Forbidden()
 
+        # Add include_secret params
+        parser = reqparse.RequestParser()
+        parser.add_argument('include_secret', type=inputs.boolean, default=False, location='args')
+        args = parser.parse_args()
+
         return {
-            "data": AppDslService.export_dsl(app_model=app_model)
+            "data": AppDslService.export_dsl(app_model=app_model, include_secret=args['include_secret'])
         }
 
 

api/controllers/console/app/workflow.py

@@ -13,6 +13,7 @@
 from controllers.console.wraps import account_initialization_required
 from core.app.apps.base_app_queue_manager import AppQueueManager
 from core.app.entities.app_invoke_entities import InvokeFrom
+from core.app.segments import factory
 from core.errors.error import AppInvokeQuotaExceededError
 from fields.workflow_fields import workflow_fields
 from fields.workflow_run_fields import workflow_run_node_execution_fields
@@ -41,7 +42,7 @@ def get(self, app_model: App):
         # The role of the current user in the ta table must be admin, owner, or editor
         if not current_user.is_editor:
             raise Forbidden()
-        
+
         # fetch draft workflow by app_model
         workflow_service = WorkflowService()
         workflow = workflow_service.get_draft_workflow(app_model=app_model)
@@ -64,13 +65,15 @@ def post(self, app_model: App):
         if not current_user.is_editor:
             raise Forbidden()
 
-        content_type = request.headers.get('Content-Type')
+        content_type = request.headers.get('Content-Type', '')
 
         if 'application/json' in content_type:
             parser = reqparse.RequestParser()
             parser.add_argument('graph', type=dict, required=True, nullable=False, location='json')
             parser.add_argument('features', type=dict, required=True, nullable=False, location='json')
             parser.add_argument('hash', type=str, required=False, location='json')
+            # TODO: set this to required=True after frontend is updated
+            parser.add_argument('environment_variables', type=list, required=False, location='json')
             args = parser.parse_args()
         elif 'text/plain' in content_type:
             try:
@@ -84,7 +87,8 @@ def post(self, app_model: App):
                 args = {
                     'graph': data.get('graph'),
                     'features': data.get('features'),
-                    'hash': data.get('hash')
+                    'hash': data.get('hash'),
+                    'environment_variables': data.get('environment_variables')
                 }
             except json.JSONDecodeError:
                 return {'message': 'Invalid JSON data'}, 400
@@ -94,12 +98,15 @@ def post(self, app_model: App):
         workflow_service = WorkflowService()
 
         try:
+            environment_variables_list = args.get('environment_variables') or []
+            environment_variables = [factory.build_variable_from_mapping(obj) for obj in environment_variables_list]
             workflow = workflow_service.sync_draft_workflow(
                 app_model=app_model,
-                graph=args.get('graph'),
-                features=args.get('features'),
+                graph=args['graph'],
+                features=args['features'],
                 unique_hash=args.get('hash'),
-                account=current_user
+                account=current_user,
+                environment_variables=environment_variables,
             )
         except WorkflowHashNotEqualError:
             raise DraftWorkflowNotSync()

api/core/app/app_config/base_app_config_manager.py

@@ -1,6 +1,7 @@
-from typing import Optional, Union
+from collections.abc import Mapping
+from typing import Any
 
-from core.app.app_config.entities import AppAdditionalFeatures, EasyUIBasedAppModelConfigFrom
+from core.app.app_config.entities import AppAdditionalFeatures
 from core.app.app_config.features.file_upload.manager import FileUploadConfigManager
 from core.app.app_config.features.more_like_this.manager import MoreLikeThisConfigManager
 from core.app.app_config.features.opening_statement.manager import OpeningStatementConfigManager
@@ -10,37 +11,19 @@
     SuggestedQuestionsAfterAnswerConfigManager,
 )
 from core.app.app_config.features.text_to_speech.manager import TextToSpeechConfigManager
-from models.model import AppMode, AppModelConfig
+from models.model import AppMode
 
 
 class BaseAppConfigManager:
-
-    @classmethod
-    def convert_to_config_dict(cls, config_from: EasyUIBasedAppModelConfigFrom,
-                               app_model_config: Union[AppModelConfig, dict],
-                               config_dict: Optional[dict] = None) -> dict:
-        """
-        Convert app model config to config dict
-        :param config_from: app model config from
-        :param app_model_config: app model config
-        :param config_dict: app model config dict
-        :return:
-        """
-        if config_from != EasyUIBasedAppModelConfigFrom.ARGS:
-            app_model_config_dict = app_model_config.to_dict()
-            config_dict = app_model_config_dict.copy()
-
-        return config_dict
-
     @classmethod
-    def convert_features(cls, config_dict: dict, app_mode: AppMode) -> AppAdditionalFeatures:
+    def convert_features(cls, config_dict: Mapping[str, Any], app_mode: AppMode) -> AppAdditionalFeatures:
         """
         Convert app config to app model config
 
         :param config_dict: app config
         :param app_mode: app mode
         """
-        config_dict = config_dict.copy()
+        config_dict = dict(config_dict.items())
 
         additional_features = AppAdditionalFeatures()
         additional_features.show_retrieve_source = RetrievalResourceConfigManager.convert(

api/core/app/app_config/features/file_upload/manager.py

@@ -1,11 +1,12 @@
-from typing import Optional
+from collections.abc import Mapping
+from typing import Any, Optional
 
 from core.app.app_config.entities import FileExtraConfig
 
 
 class FileUploadConfigManager:
     @classmethod
-    def convert(cls, config: dict, is_vision: bool = True) -> Optional[FileExtraConfig]:
+    def convert(cls, config: Mapping[str, Any], is_vision: bool = True) -> Optional[FileExtraConfig]:
         """
         Convert model config to model config
 

api/core/app/app_config/features/text_to_speech/manager.py

@@ -3,13 +3,13 @@
 
 class TextToSpeechConfigManager:
     @classmethod
-    def convert(cls, config: dict) -> bool:
+    def convert(cls, config: dict):
         """
         Convert model config to model config
 
         :param config: model config args
         """
-        text_to_speech = False
+        text_to_speech = None
         text_to_speech_dict = config.get('text_to_speech')
         if text_to_speech_dict:
             if text_to_speech_dict.get('enabled'):

api/core/app/apps/advanced_chat/app_generator.py

@@ -1,3 +1,4 @@
+import contextvars
 import logging
 import os
 import threading
@@ -8,6 +9,7 @@
 from flask import Flask, current_app
 from pydantic import ValidationError
 
+import contexts
 from core.app.app_config.features.file_upload.manager import FileUploadConfigManager
 from core.app.apps.advanced_chat.app_config_manager import AdvancedChatAppConfigManager
 from core.app.apps.advanced_chat.app_runner import AdvancedChatAppRunner
@@ -107,6 +109,7 @@ def generate(
             extras=extras,
             trace_manager=trace_manager
         )
+        contexts.tenant_id.set(application_generate_entity.app_config.tenant_id)
 
         return self._generate(
             app_model=app_model,
@@ -173,6 +176,7 @@ def single_iteration_generate(self, app_model: App,
                 inputs=args['inputs']
             )
         )
+        contexts.tenant_id.set(application_generate_entity.app_config.tenant_id)
 
         return self._generate(
             app_model=app_model,
@@ -225,6 +229,8 @@ def _generate(self, app_model: App,
             'queue_manager': queue_manager,
             'conversation_id': conversation.id,
             'message_id': message.id,
+            'user': user,
+            'context': contextvars.copy_context()
         })
 
         worker_thread.start()
@@ -249,7 +255,9 @@ def _generate_worker(self, flask_app: Flask,
                          application_generate_entity: AdvancedChatAppGenerateEntity,
                          queue_manager: AppQueueManager,
                          conversation_id: str,
-                         message_id: str) -> None:
+                         message_id: str,
+                         user: Account,
+                         context: contextvars.Context) -> None:
         """
         Generate worker in a new thread.
         :param flask_app: Flask app
@@ -259,6 +267,8 @@ def _generate_worker(self, flask_app: Flask,
         :param message_id: message ID
         :return:
         """
+        for var, val in context.items():
+            var.set(val)
         with flask_app.app_context():
             try:
                 runner = AdvancedChatAppRunner()

api/core/app/apps/advanced_chat/app_runner.py

@@ -1,7 +1,8 @@
 import logging
 import os
 import time
-from typing import Optional, cast
+from collections.abc import Mapping
+from typing import Any, Optional, cast
 
 from core.app.apps.advanced_chat.app_config_manager import AdvancedChatAppConfig
 from core.app.apps.advanced_chat.workflow_event_trigger_callback import WorkflowEventTriggerCallback
@@ -14,6 +15,7 @@
 )
 from core.app.entities.queue_entities import QueueAnnotationReplyEvent, QueueStopEvent, QueueTextChunkEvent
 from core.moderation.base import ModerationException
+from core.workflow.callbacks.base_workflow_callback import WorkflowCallback
 from core.workflow.entities.node_entities import SystemVariable
 from core.workflow.nodes.base_node import UserFrom
 from core.workflow.workflow_engine_manager import WorkflowEngineManager
@@ -87,7 +89,7 @@ def run(self, application_generate_entity: AdvancedChatAppGenerateEntity,
 
         db.session.close()
 
-        workflow_callbacks = [WorkflowEventTriggerCallback(
+        workflow_callbacks: list[WorkflowCallback] = [WorkflowEventTriggerCallback(
             queue_manager=queue_manager,
             workflow=workflow
         )]
@@ -161,7 +163,7 @@ def handle_input_moderation(
             self, queue_manager: AppQueueManager,
             app_record: App,
             app_generate_entity: AdvancedChatAppGenerateEntity,
-            inputs: dict,
+            inputs: Mapping[str, Any],
             query: str,
             message_id: str
     ) -> bool:

api/core/app/apps/advanced_chat/generate_response_converter.py

@@ -1,9 +1,11 @@
 import json
 from collections.abc import Generator
-from typing import cast
+from typing import Any, cast
 
 from core.app.apps.base_app_generate_response_converter import AppGenerateResponseConverter
 from core.app.entities.task_entities import (
+    AppBlockingResponse,
+    AppStreamResponse,
     ChatbotAppBlockingResponse,
     ChatbotAppStreamResponse,
     ErrorStreamResponse,
@@ -18,12 +20,13 @@ class AdvancedChatAppGenerateResponseConverter(AppGenerateResponseConverter):
     _blocking_response_type = ChatbotAppBlockingResponse
 
     @classmethod
-    def convert_blocking_full_response(cls, blocking_response: ChatbotAppBlockingResponse) -> dict:
+    def convert_blocking_full_response(cls, blocking_response: AppBlockingResponse) -> dict[str, Any]:
         """
         Convert blocking full response.
         :param blocking_response: blocking response
         :return:
         """
+        blocking_response = cast(ChatbotAppBlockingResponse, blocking_response)
         response = {
             'event': 'message',
             'task_id': blocking_response.task_id,
@@ -39,7 +42,7 @@ def convert_blocking_full_response(cls, blocking_response: ChatbotAppBlockingRes
         return response
 
     @classmethod
-    def convert_blocking_simple_response(cls, blocking_response: ChatbotAppBlockingResponse) -> dict:
+    def convert_blocking_simple_response(cls, blocking_response: AppBlockingResponse) -> dict[str, Any]:
         """
         Convert blocking simple response.
         :param blocking_response: blocking response
@@ -53,8 +56,7 @@ def convert_blocking_simple_response(cls, blocking_response: ChatbotAppBlockingR
         return response
 
     @classmethod
-    def convert_stream_full_response(cls, stream_response: Generator[ChatbotAppStreamResponse, None, None]) \
-            -> Generator[str, None, None]:
+    def convert_stream_full_response(cls, stream_response: Generator[AppStreamResponse, None, None]) -> Generator[str, Any, None]:
         """
         Convert stream full response.
         :param stream_response: stream response
@@ -83,8 +85,7 @@ def convert_stream_full_response(cls, stream_response: Generator[ChatbotAppStrea
             yield json.dumps(response_chunk)
 
     @classmethod
-    def convert_stream_simple_response(cls, stream_response: Generator[ChatbotAppStreamResponse, None, None]) \
-            -> Generator[str, None, None]:
+    def convert_stream_simple_response(cls, stream_response: Generator[AppStreamResponse, None, None]) -> Generator[str, Any, None]:
         """
         Convert stream simple response.
         :param stream_response: stream response