external knowledge api

# Conflicts:
#	api/core/rag/datasource/retrieval_service.py
#	api/models/dataset.py
#	api/services/dataset_service.py

.gitignore

@@ -153,6 +153,9 @@ docker-legacy/volumes/etcd/*
 docker-legacy/volumes/minio/*
 docker-legacy/volumes/milvus/*
 docker-legacy/volumes/chroma/*
+docker-legacy/volumes/opensearch/data/*
+docker-legacy/volumes/pgvectors/data/*
+docker-legacy/volumes/pgvector/data/*
 
 docker/volumes/app/storage/*
 docker/volumes/certbot/*
@@ -164,6 +167,12 @@ docker/volumes/etcd/*
 docker/volumes/minio/*
 docker/volumes/milvus/*
 docker/volumes/chroma/*
+docker/volumes/opensearch/data/*
+docker/volumes/myscale/data/*
+docker/volumes/myscale/log/*
+docker/volumes/unstructured/*
+docker/volumes/pgvector/data/*
+docker/volumes/pgvecto_rs/data/*
 
 docker/nginx/conf.d/default.conf
 docker/middleware.env

api/app.py

@@ -164,7 +164,7 @@ def initialize_extensions(app):
 @login_manager.request_loader
 def load_user_from_request(request_from_flask_login):
     """Load user based on the request."""
-    if request.blueprint not in ["console", "inner_api"]:
+    if request.blueprint not in {"console", "inner_api"}:
         return None
     # Check if the user_id contains a dot, indicating the old format
     auth_header = request.headers.get("Authorization", "")

api/commands.py

@@ -104,7 +104,7 @@ def reset_email(email, new_email, email_confirm):
 )
 @click.confirmation_option(
     prompt=click.style(
-        "Are you sure you want to reset encrypt key pair?" " this operation cannot be rolled back!", fg="red"
+        "Are you sure you want to reset encrypt key pair? this operation cannot be rolled back!", fg="red"
     )
 )
 def reset_encrypt_key_pair():
@@ -131,7 +131,7 @@ def reset_encrypt_key_pair():
 
         click.echo(
             click.style(
-                "Congratulations! " "the asymmetric key pair of workspace {} has been reset.".format(tenant.id),
+                "Congratulations! The asymmetric key pair of workspace {} has been reset.".format(tenant.id),
                 fg="green",
             )
         )
@@ -140,9 +140,9 @@ def reset_encrypt_key_pair():
 @click.command("vdb-migrate", help="migrate vector db.")
 @click.option("--scope", default="all", prompt=False, help="The scope of vector database to migrate, Default is All.")
 def vdb_migrate(scope: str):
-    if scope in ["knowledge", "all"]:
+    if scope in {"knowledge", "all"}:
         migrate_knowledge_vector_database()
-    if scope in ["annotation", "all"]:
+    if scope in {"annotation", "all"}:
         migrate_annotation_vector_database()
 
 
@@ -275,8 +275,7 @@ def migrate_knowledge_vector_database():
         for dataset in datasets:
             total_count = total_count + 1
             click.echo(
-                f"Processing the {total_count} dataset {dataset.id}. "
-                + f"{create_count} created, {skipped_count} skipped."
+                f"Processing the {total_count} dataset {dataset.id}. {create_count} created, {skipped_count} skipped."
             )
             try:
                 click.echo("Create dataset vdb index: {}".format(dataset.id))
@@ -411,7 +410,8 @@ def migrate_knowledge_vector_database():
                     try:
                         click.echo(
                             click.style(
-                                f"Start to created vector index with {len(documents)} documents of {segments_count} segments for dataset {dataset.id}.",
+                                f"Start to created vector index with {len(documents)} documents of {segments_count}"
+                                f" segments for dataset {dataset.id}.",
                                 fg="green",
                             )
                         )
@@ -593,7 +593,7 @@ def create_tenant(email: str, language: Optional[str] = None, name: Optional[str
 
     click.echo(
         click.style(
-            "Congratulations! Account and tenant created.\n" "Account: {}\nPassword: {}".format(email, new_password),
+            "Congratulations! Account and tenant created.\nAccount: {}\nPassword: {}".format(email, new_password),
             fg="green",
         )
     )

api/configs/feature/__init__.py

@@ -129,12 +129,12 @@ class EndpointConfig(BaseSettings):
     )
 
     SERVICE_API_URL: str = Field(
-        description="Service API Url prefix." "used to display Service API Base Url to the front-end.",
+        description="Service API Url prefix. used to display Service API Base Url to the front-end.",
         default="",
     )
 
     APP_WEB_URL: str = Field(
-        description="WebApp Url prefix." "used to display WebAPP API Base Url to the front-end.",
+        description="WebApp Url prefix. used to display WebAPP API Base Url to the front-end.",
         default="",
     )
 
@@ -272,7 +272,7 @@ class LoggingConfig(BaseSettings):
     """
 
     LOG_LEVEL: str = Field(
-        description="Log output level, default to INFO." "It is recommended to set it to ERROR for production.",
+        description="Log output level, default to INFO. It is recommended to set it to ERROR for production.",
         default="INFO",
     )
 

api/configs/packaging/__init__.py

@@ -9,7 +9,7 @@ class PackagingInfo(BaseSettings):
 
     CURRENT_VERSION: str = Field(
         description="Dify version",
-        default="0.8.0",
+        default="0.8.2",
     )
 
     COMMIT_SHA: str = Field(

api/controllers/console/__init__.py

@@ -37,7 +37,7 @@ from .auth import activate, data_source_bearer_auth, data_source_oauth, forgot_p
 from .billing import billing
 
 # Import datasets controllers
-from .datasets import data_source, datasets, datasets_document, datasets_segments, file, hit_testing, website
+from .datasets import data_source, datasets, datasets_document, datasets_segments, external, file, hit_testing, website
 
 # Import explore controllers
 from .explore import (

api/controllers/console/admin.py

@@ -60,23 +60,15 @@ class InsertExploreAppListApi(Resource):
 
         site = app.site
         if not site:
-            desc = args["desc"] if args["desc"] else ""
-            copy_right = args["copyright"] if args["copyright"] else ""
-            privacy_policy = args["privacy_policy"] if args["privacy_policy"] else ""
-            custom_disclaimer = args["custom_disclaimer"] if args["custom_disclaimer"] else ""
+            desc = args["desc"] or ""
+            copy_right = args["copyright"] or ""
+            privacy_policy = args["privacy_policy"] or ""
+            custom_disclaimer = args["custom_disclaimer"] or ""
         else:
-            desc = site.description if site.description else args["desc"] if args["desc"] else ""
-            copy_right = site.copyright if site.copyright else args["copyright"] if args["copyright"] else ""
-            privacy_policy = (
-                site.privacy_policy if site.privacy_policy else args["privacy_policy"] if args["privacy_policy"] else ""
-            )
-            custom_disclaimer = (
-                site.custom_disclaimer
-                if site.custom_disclaimer
-                else args["custom_disclaimer"]
-                if args["custom_disclaimer"]
-                else ""
-            )
+            desc = site.description or args["desc"] or ""
+            copy_right = site.copyright or args["copyright"] or ""
+            privacy_policy = site.privacy_policy or args["privacy_policy"] or ""
+            custom_disclaimer = site.custom_disclaimer or args["custom_disclaimer"] or ""
 
         recommended_app = RecommendedApp.query.filter(RecommendedApp.app_id == args["app_id"]).first()
 

api/controllers/console/apikey.py

@@ -57,7 +57,7 @@ class BaseApiKeyListResource(Resource):
     def post(self, resource_id):
         resource_id = str(resource_id)
         _get_resource(resource_id, current_user.current_tenant_id, self.resource_model)
-        if not current_user.is_admin_or_owner:
+        if not current_user.is_editor:
             raise Forbidden()
 
         current_key_count = (

api/controllers/console/app/audio.py

@@ -94,19 +94,15 @@ class ChatMessageTextApi(Resource):
             message_id = args.get("message_id", None)
             text = args.get("text", None)
             if (
-                app_model.mode in [AppMode.ADVANCED_CHAT.value, AppMode.WORKFLOW.value]
+                app_model.mode in {AppMode.ADVANCED_CHAT.value, AppMode.WORKFLOW.value}
                 and app_model.workflow
                 and app_model.workflow.features_dict
             ):
                 text_to_speech = app_model.workflow.features_dict.get("text_to_speech")
-                voice = args.get("voice") if args.get("voice") else text_to_speech.get("voice")
+                voice = args.get("voice") or text_to_speech.get("voice")
             else:
                 try:
-                    voice = (
-                        args.get("voice")
-                        if args.get("voice")
-                        else app_model.app_model_config.text_to_speech_dict.get("voice")
-                    )
+                    voice = args.get("voice") or app_model.app_model_config.text_to_speech_dict.get("voice")
                 except Exception:
                     voice = None
             response = AudioService.transcript_tts(app_model=app_model, text=text, message_id=message_id, voice=voice)

api/controllers/console/app/conversation.py

@@ -20,7 +20,7 @@ from fields.conversation_fields import (
     conversation_pagination_fields,
     conversation_with_summary_pagination_fields,
 )
-from libs.helper import datetime_string
+from libs.helper import DatetimeString
 from libs.login import login_required
 from models.model import AppMode, Conversation, EndUser, Message, MessageAnnotation
 
@@ -36,8 +36,8 @@ class CompletionConversationApi(Resource):
             raise Forbidden()
         parser = reqparse.RequestParser()
         parser.add_argument("keyword", type=str, location="args")
-        parser.add_argument("start", type=datetime_string("%Y-%m-%d %H:%M"), location="args")
-        parser.add_argument("end", type=datetime_string("%Y-%m-%d %H:%M"), location="args")
+        parser.add_argument("start", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
+        parser.add_argument("end", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
         parser.add_argument(
             "annotation_status", type=str, choices=["annotated", "not_annotated", "all"], default="all", location="args"
         )
@@ -143,8 +143,8 @@ class ChatConversationApi(Resource):
             raise Forbidden()
         parser = reqparse.RequestParser()
         parser.add_argument("keyword", type=str, location="args")
-        parser.add_argument("start", type=datetime_string("%Y-%m-%d %H:%M"), location="args")
-        parser.add_argument("end", type=datetime_string("%Y-%m-%d %H:%M"), location="args")
+        parser.add_argument("start", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
+        parser.add_argument("end", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
         parser.add_argument(
             "annotation_status", type=str, choices=["annotated", "not_annotated", "all"], default="all", location="args"
         )

api/controllers/console/app/statistic.py

@@ -11,7 +11,7 @@ from controllers.console.app.wraps import get_app_model
 from controllers.console.setup import setup_required
 from controllers.console.wraps import account_initialization_required
 from extensions.ext_database import db
-from libs.helper import datetime_string
+from libs.helper import DatetimeString
 from libs.login import login_required
 from models.model import AppMode
 
@@ -25,14 +25,17 @@ class DailyMessageStatistic(Resource):
         account = current_user
 
         parser = reqparse.RequestParser()
-        parser.add_argument("start", type=datetime_string("%Y-%m-%d %H:%M"), location="args")
-        parser.add_argument("end", type=datetime_string("%Y-%m-%d %H:%M"), location="args")
+        parser.add_argument("start", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
+        parser.add_argument("end", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
         args = parser.parse_args()
 
-        sql_query = """
-        SELECT date(DATE_TRUNC('day', created_at AT TIME ZONE 'UTC' AT TIME ZONE :tz )) AS date, count(*) AS message_count
-            FROM messages where app_id = :app_id 
-        """
+        sql_query = """SELECT
+    DATE(DATE_TRUNC('day', created_at AT TIME ZONE 'UTC' AT TIME ZONE :tz )) AS date,
+    COUNT(*) AS message_count
+FROM
+    messages
+WHERE
+    app_id = :app_id"""
         arg_dict = {"tz": account.timezone, "app_id": app_model.id}
 
         timezone = pytz.timezone(account.timezone)
@@ -45,7 +48,7 @@ class DailyMessageStatistic(Resource):
             start_datetime_timezone = timezone.localize(start_datetime)
             start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)
 
-            sql_query += " and created_at >= :start"
+            sql_query += " AND created_at >= :start"
             arg_dict["start"] = start_datetime_utc
 
         if args["end"]:
@@ -55,10 +58,10 @@ class DailyMessageStatistic(Resource):
             end_datetime_timezone = timezone.localize(end_datetime)
             end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
 
-            sql_query += " and created_at < :end"
+            sql_query += " AND created_at < :end"
             arg_dict["end"] = end_datetime_utc
 
-        sql_query += " GROUP BY date order by date"
+        sql_query += " GROUP BY date ORDER BY date"
 
         response_data = []
 
@@ -79,14 +82,17 @@ class DailyConversationStatistic(Resource):
         account = current_user
 
         parser = reqparse.RequestParser()
-        parser.add_argument("start", type=datetime_string("%Y-%m-%d %H:%M"), location="args")
-        parser.add_argument("end", type=datetime_string("%Y-%m-%d %H:%M"), location="args")
+        parser.add_argument("start", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
+        parser.add_argument("end", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
         args = parser.parse_args()
 
-        sql_query = """
-        SELECT date(DATE_TRUNC('day', created_at AT TIME ZONE 'UTC' AT TIME ZONE :tz )) AS date, count(distinct messages.conversation_id) AS conversation_count
-            FROM messages where app_id = :app_id 
-        """
+        sql_query = """SELECT
+    DATE(DATE_TRUNC('day', created_at AT TIME ZONE 'UTC' AT TIME ZONE :tz )) AS date,
+    COUNT(DISTINCT messages.conversation_id) AS conversation_count
+FROM
+    messages
+WHERE
+    app_id = :app_id"""
         arg_dict = {"tz": account.timezone, "app_id": app_model.id}
 
         timezone = pytz.timezone(account.timezone)
@@ -99,7 +105,7 @@ class DailyConversationStatistic(Resource):
             start_datetime_timezone = timezone.localize(start_datetime)
             start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)
 
-            sql_query += " and created_at >= :start"
+            sql_query += " AND created_at >= :start"
             arg_dict["start"] = start_datetime_utc
 
         if args["end"]:
@@ -109,10 +115,10 @@ class DailyConversationStatistic(Resource):
             end_datetime_timezone = timezone.localize(end_datetime)
             end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
 
-            sql_query += " and created_at < :end"
+            sql_query += " AND created_at < :end"
             arg_dict["end"] = end_datetime_utc
 
-        sql_query += " GROUP BY date order by date"
+        sql_query += " GROUP BY date ORDER BY date"
 
         response_data = []
 
@@ -133,14 +139,17 @@ class DailyTerminalsStatistic(Resource):
         account = current_user
 
         parser = reqparse.RequestParser()
-        parser.add_argument("start", type=datetime_string("%Y-%m-%d %H:%M"), location="args")
-        parser.add_argument("end", type=datetime_string("%Y-%m-%d %H:%M"), location="args")
+        parser.add_argument("start", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
+        parser.add_argument("end", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
         args = parser.parse_args()
 
-        sql_query = """
-                SELECT date(DATE_TRUNC('day', created_at AT TIME ZONE 'UTC' AT TIME ZONE :tz )) AS date, count(distinct messages.from_end_user_id) AS terminal_count
-                    FROM messages where app_id = :app_id 
-                """
+        sql_query = """SELECT
+    DATE(DATE_TRUNC('day', created_at AT TIME ZONE 'UTC' AT TIME ZONE :tz )) AS date,
+    COUNT(DISTINCT messages.from_end_user_id) AS terminal_count
+FROM
+    messages
+WHERE
+    app_id = :app_id"""
         arg_dict = {"tz": account.timezone, "app_id": app_model.id}
 
         timezone = pytz.timezone(account.timezone)
@@ -153,7 +162,7 @@ class DailyTerminalsStatistic(Resource):
             start_datetime_timezone = timezone.localize(start_datetime)
             start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)
 
-            sql_query += " and created_at >= :start"
+            sql_query += " AND created_at >= :start"
             arg_dict["start"] = start_datetime_utc
 
         if args["end"]:
@@ -163,10 +172,10 @@ class DailyTerminalsStatistic(Resource):
             end_datetime_timezone = timezone.localize(end_datetime)
             end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
 
-            sql_query += " and created_at < :end"
+            sql_query += " AND created_at < :end"
             arg_dict["end"] = end_datetime_utc
 
-        sql_query += " GROUP BY date order by date"
+        sql_query += " GROUP BY date ORDER BY date"
 
         response_data = []
 
@@ -187,16 +196,18 @@ class DailyTokenCostStatistic(Resource):
         account = current_user
 
         parser = reqparse.RequestParser()
-        parser.add_argument("start", type=datetime_string("%Y-%m-%d %H:%M"), location="args")
-        parser.add_argument("end", type=datetime_string("%Y-%m-%d %H:%M"), location="args")
+        parser.add_argument("start", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
+        parser.add_argument("end", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
         args = parser.parse_args()
 
-        sql_query = """
-                SELECT date(DATE_TRUNC('day', created_at AT TIME ZONE 'UTC' AT TIME ZONE :tz )) AS date, 
-                    (sum(messages.message_tokens) + sum(messages.answer_tokens)) as token_count,
-                    sum(total_price) as total_price
-                    FROM messages where app_id = :app_id 
-                """
+        sql_query = """SELECT
+    DATE(DATE_TRUNC('day', created_at AT TIME ZONE 'UTC' AT TIME ZONE :tz )) AS date,
+    (SUM(messages.message_tokens) + SUM(messages.answer_tokens)) AS token_count,
+    SUM(total_price) AS total_price
+FROM
+    messages
+WHERE
+    app_id = :app_id"""
         arg_dict = {"tz": account.timezone, "app_id": app_model.id}
 
         timezone = pytz.timezone(account.timezone)
@@ -209,7 +220,7 @@ class DailyTokenCostStatistic(Resource):
             start_datetime_timezone = timezone.localize(start_datetime)
             start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)
 
-            sql_query += " and created_at >= :start"
+            sql_query += " AND created_at >= :start"
             arg_dict["start"] = start_datetime_utc
 
         if args["end"]:
@@ -219,10 +230,10 @@ class DailyTokenCostStatistic(Resource):
             end_datetime_timezone = timezone.localize(end_datetime)
             end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
 
-            sql_query += " and created_at < :end"
+            sql_query += " AND created_at < :end"
             arg_dict["end"] = end_datetime_utc
 
-        sql_query += " GROUP BY date order by date"
+        sql_query += " GROUP BY date ORDER BY date"
 
         response_data = []
 
@@ -245,16 +256,26 @@ class AverageSessionInteractionStatistic(Resource):
         account = current_user
 
         parser = reqparse.RequestParser()
-        parser.add_argument("start", type=datetime_string("%Y-%m-%d %H:%M"), location="args")
-        parser.add_argument("end", type=datetime_string("%Y-%m-%d %H:%M"), location="args")
+        parser.add_argument("start", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
+        parser.add_argument("end", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
         args = parser.parse_args()
 
-        sql_query = """SELECT date(DATE_TRUNC('day', c.created_at AT TIME ZONE 'UTC' AT TIME ZONE :tz )) AS date, 
-AVG(subquery.message_count) AS interactions
-FROM (SELECT m.conversation_id, COUNT(m.id) AS message_count
-    FROM conversations c
-    JOIN messages m ON c.id = m.conversation_id
-    WHERE c.override_model_configs IS NULL AND c.app_id = :app_id"""
+        sql_query = """SELECT
+    DATE(DATE_TRUNC('day', c.created_at AT TIME ZONE 'UTC' AT TIME ZONE :tz )) AS date,
+    AVG(subquery.message_count) AS interactions
+FROM
+    (
+        SELECT
+            m.conversation_id,
+            COUNT(m.id) AS message_count
+        FROM
+            conversations c
+        JOIN
+            messages m
+            ON c.id = m.conversation_id
+        WHERE
+            c.override_model_configs IS NULL
+            AND c.app_id = :app_id"""
         arg_dict = {"tz": account.timezone, "app_id": app_model.id}
 
         timezone = pytz.timezone(account.timezone)
@@ -267,7 +288,7 @@ FROM (SELECT m.conversation_id, COUNT(m.id) AS message_count
             start_datetime_timezone = timezone.localize(start_datetime)
             start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)
 
-            sql_query += " and c.created_at >= :start"
+            sql_query += " AND c.created_at >= :start"
             arg_dict["start"] = start_datetime_utc
 
         if args["end"]:
@@ -277,14 +298,19 @@ FROM (SELECT m.conversation_id, COUNT(m.id) AS message_count
             end_datetime_timezone = timezone.localize(end_datetime)
             end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
 
-            sql_query += " and c.created_at < :end"
+            sql_query += " AND c.created_at < :end"
             arg_dict["end"] = end_datetime_utc
 
         sql_query += """
-        GROUP BY m.conversation_id) subquery
-LEFT JOIN conversations c on c.id=subquery.conversation_id
-GROUP BY date
-ORDER BY date"""
+        GROUP BY m.conversation_id
+    ) subquery
+LEFT JOIN
+    conversations c
+    ON c.id = subquery.conversation_id
+GROUP BY
+    date
+ORDER BY
+    date"""
 
         response_data = []
 
@@ -307,17 +333,21 @@ class UserSatisfactionRateStatistic(Resource):
         account = current_user
 
         parser = reqparse.RequestParser()
-        parser.add_argument("start", type=datetime_string("%Y-%m-%d %H:%M"), location="args")
-        parser.add_argument("end", type=datetime_string("%Y-%m-%d %H:%M"), location="args")
+        parser.add_argument("start", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
+        parser.add_argument("end", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
         args = parser.parse_args()
 
-        sql_query = """
-                        SELECT date(DATE_TRUNC('day', m.created_at AT TIME ZONE 'UTC' AT TIME ZONE :tz )) AS date, 
-                            COUNT(m.id) as message_count, COUNT(mf.id) as feedback_count 
-                            FROM messages m
-                            LEFT JOIN message_feedbacks mf on mf.message_id=m.id and mf.rating='like'
-                            WHERE m.app_id = :app_id 
-                        """
+        sql_query = """SELECT
+    DATE(DATE_TRUNC('day', m.created_at AT TIME ZONE 'UTC' AT TIME ZONE :tz )) AS date,
+    COUNT(m.id) AS message_count,
+    COUNT(mf.id) AS feedback_count
+FROM
+    messages m
+LEFT JOIN
+    message_feedbacks mf
+    ON mf.message_id=m.id AND mf.rating='like'
+WHERE
+    m.app_id = :app_id"""
         arg_dict = {"tz": account.timezone, "app_id": app_model.id}
 
         timezone = pytz.timezone(account.timezone)
@@ -330,7 +360,7 @@ class UserSatisfactionRateStatistic(Resource):
             start_datetime_timezone = timezone.localize(start_datetime)
             start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)
 
-            sql_query += " and m.created_at >= :start"
+            sql_query += " AND m.created_at >= :start"
             arg_dict["start"] = start_datetime_utc
 
         if args["end"]:
@@ -340,10 +370,10 @@ class UserSatisfactionRateStatistic(Resource):
             end_datetime_timezone = timezone.localize(end_datetime)
             end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
 
-            sql_query += " and m.created_at < :end"
+            sql_query += " AND m.created_at < :end"
             arg_dict["end"] = end_datetime_utc
 
-        sql_query += " GROUP BY date order by date"
+        sql_query += " GROUP BY date ORDER BY date"
 
         response_data = []
 
@@ -369,16 +399,17 @@ class AverageResponseTimeStatistic(Resource):
         account = current_user
 
         parser = reqparse.RequestParser()
-        parser.add_argument("start", type=datetime_string("%Y-%m-%d %H:%M"), location="args")
-        parser.add_argument("end", type=datetime_string("%Y-%m-%d %H:%M"), location="args")
+        parser.add_argument("start", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
+        parser.add_argument("end", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
         args = parser.parse_args()
 
-        sql_query = """
-                SELECT date(DATE_TRUNC('day', created_at AT TIME ZONE 'UTC' AT TIME ZONE :tz )) AS date, 
-                    AVG(provider_response_latency) as latency
-                    FROM messages
-                    WHERE app_id = :app_id
-                """
+        sql_query = """SELECT
+    DATE(DATE_TRUNC('day', created_at AT TIME ZONE 'UTC' AT TIME ZONE :tz )) AS date,
+    AVG(provider_response_latency) AS latency
+FROM
+    messages
+WHERE
+    app_id = :app_id"""
         arg_dict = {"tz": account.timezone, "app_id": app_model.id}
 
         timezone = pytz.timezone(account.timezone)
@@ -391,7 +422,7 @@ class AverageResponseTimeStatistic(Resource):
             start_datetime_timezone = timezone.localize(start_datetime)
             start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)
 
-            sql_query += " and created_at >= :start"
+            sql_query += " AND created_at >= :start"
             arg_dict["start"] = start_datetime_utc
 
         if args["end"]:
@@ -401,10 +432,10 @@ class AverageResponseTimeStatistic(Resource):
             end_datetime_timezone = timezone.localize(end_datetime)
             end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
 
-            sql_query += " and created_at < :end"
+            sql_query += " AND created_at < :end"
             arg_dict["end"] = end_datetime_utc
 
-        sql_query += " GROUP BY date order by date"
+        sql_query += " GROUP BY date ORDER BY date"
 
         response_data = []
 
@@ -425,17 +456,20 @@ class TokensPerSecondStatistic(Resource):
         account = current_user
 
         parser = reqparse.RequestParser()
-        parser.add_argument("start", type=datetime_string("%Y-%m-%d %H:%M"), location="args")
-        parser.add_argument("end", type=datetime_string("%Y-%m-%d %H:%M"), location="args")
+        parser.add_argument("start", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
+        parser.add_argument("end", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
         args = parser.parse_args()
 
-        sql_query = """SELECT date(DATE_TRUNC('day', created_at AT TIME ZONE 'UTC' AT TIME ZONE :tz )) AS date, 
-    CASE 
+        sql_query = """SELECT
+    DATE(DATE_TRUNC('day', created_at AT TIME ZONE 'UTC' AT TIME ZONE :tz )) AS date,
+    CASE
         WHEN SUM(provider_response_latency) = 0 THEN 0
         ELSE (SUM(answer_tokens) / SUM(provider_response_latency))
     END as tokens_per_second
-FROM messages
-WHERE app_id = :app_id"""
+FROM
+    messages
+WHERE
+    app_id = :app_id"""
         arg_dict = {"tz": account.timezone, "app_id": app_model.id}
 
         timezone = pytz.timezone(account.timezone)
@@ -448,7 +482,7 @@ WHERE app_id = :app_id"""
             start_datetime_timezone = timezone.localize(start_datetime)
             start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)
 
-            sql_query += " and created_at >= :start"
+            sql_query += " AND created_at >= :start"
             arg_dict["start"] = start_datetime_utc
 
         if args["end"]:
@@ -458,10 +492,10 @@ WHERE app_id = :app_id"""
             end_datetime_timezone = timezone.localize(end_datetime)
             end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
 
-            sql_query += " and created_at < :end"
+            sql_query += " AND created_at < :end"
             arg_dict["end"] = end_datetime_utc
 
-        sql_query += " GROUP BY date order by date"
+        sql_query += " GROUP BY date ORDER BY date"
 
         response_data = []
 

api/controllers/console/app/workflow.py

@@ -465,6 +465,6 @@ api.add_resource(
 api.add_resource(PublishedWorkflowApi, "/apps/<uuid:app_id>/workflows/publish")
 api.add_resource(DefaultBlockConfigsApi, "/apps/<uuid:app_id>/workflows/default-workflow-block-configs")
 api.add_resource(
-    DefaultBlockConfigApi, "/apps/<uuid:app_id>/workflows/default-workflow-block-configs" "/<string:block_type>"
+    DefaultBlockConfigApi, "/apps/<uuid:app_id>/workflows/default-workflow-block-configs/<string:block_type>"
 )
 api.add_resource(ConvertToWorkflowApi, "/apps/<uuid:app_id>/convert-to-workflow")

api/controllers/console/app/workflow_statistic.py

@@ -11,7 +11,7 @@ from controllers.console.app.wraps import get_app_model
 from controllers.console.setup import setup_required
 from controllers.console.wraps import account_initialization_required
 from extensions.ext_database import db
-from libs.helper import datetime_string
+from libs.helper import DatetimeString
 from libs.login import login_required
 from models.model import AppMode
 from models.workflow import WorkflowRunTriggeredFrom
@@ -26,16 +26,18 @@ class WorkflowDailyRunsStatistic(Resource):
         account = current_user
 
         parser = reqparse.RequestParser()
-        parser.add_argument("start", type=datetime_string("%Y-%m-%d %H:%M"), location="args")
-        parser.add_argument("end", type=datetime_string("%Y-%m-%d %H:%M"), location="args")
+        parser.add_argument("start", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
+        parser.add_argument("end", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
         args = parser.parse_args()
 
-        sql_query = """
-        SELECT date(DATE_TRUNC('day', created_at AT TIME ZONE 'UTC' AT TIME ZONE :tz )) AS date, count(id) AS runs
-            FROM workflow_runs 
-            WHERE app_id = :app_id 
-                AND triggered_from = :triggered_from
-        """
+        sql_query = """SELECT
+    DATE(DATE_TRUNC('day', created_at AT TIME ZONE 'UTC' AT TIME ZONE :tz )) AS date,
+    COUNT(id) AS runs
+FROM
+    workflow_runs
+WHERE
+    app_id = :app_id
+    AND triggered_from = :triggered_from"""
         arg_dict = {
             "tz": account.timezone,
             "app_id": app_model.id,
@@ -52,7 +54,7 @@ class WorkflowDailyRunsStatistic(Resource):
             start_datetime_timezone = timezone.localize(start_datetime)
             start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)
 
-            sql_query += " and created_at >= :start"
+            sql_query += " AND created_at >= :start"
             arg_dict["start"] = start_datetime_utc
 
         if args["end"]:
@@ -62,10 +64,10 @@ class WorkflowDailyRunsStatistic(Resource):
             end_datetime_timezone = timezone.localize(end_datetime)
             end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
 
-            sql_query += " and created_at < :end"
+            sql_query += " AND created_at < :end"
             arg_dict["end"] = end_datetime_utc
 
-        sql_query += " GROUP BY date order by date"
+        sql_query += " GROUP BY date ORDER BY date"
 
         response_data = []
 
@@ -86,16 +88,18 @@ class WorkflowDailyTerminalsStatistic(Resource):
         account = current_user
 
         parser = reqparse.RequestParser()
-        parser.add_argument("start", type=datetime_string("%Y-%m-%d %H:%M"), location="args")
-        parser.add_argument("end", type=datetime_string("%Y-%m-%d %H:%M"), location="args")
+        parser.add_argument("start", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
+        parser.add_argument("end", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
         args = parser.parse_args()
 
-        sql_query = """
-                SELECT date(DATE_TRUNC('day', created_at AT TIME ZONE 'UTC' AT TIME ZONE :tz )) AS date, count(distinct workflow_runs.created_by) AS terminal_count
-                    FROM workflow_runs 
-                    WHERE app_id = :app_id 
-                        AND triggered_from = :triggered_from
-                """
+        sql_query = """SELECT
+    DATE(DATE_TRUNC('day', created_at AT TIME ZONE 'UTC' AT TIME ZONE :tz )) AS date,
+    COUNT(DISTINCT workflow_runs.created_by) AS terminal_count
+FROM
+    workflow_runs
+WHERE
+    app_id = :app_id
+    AND triggered_from = :triggered_from"""
         arg_dict = {
             "tz": account.timezone,
             "app_id": app_model.id,
@@ -112,7 +116,7 @@ class WorkflowDailyTerminalsStatistic(Resource):
             start_datetime_timezone = timezone.localize(start_datetime)
             start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)
 
-            sql_query += " and created_at >= :start"
+            sql_query += " AND created_at >= :start"
             arg_dict["start"] = start_datetime_utc
 
         if args["end"]:
@@ -122,10 +126,10 @@ class WorkflowDailyTerminalsStatistic(Resource):
             end_datetime_timezone = timezone.localize(end_datetime)
             end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
 
-            sql_query += " and created_at < :end"
+            sql_query += " AND created_at < :end"
             arg_dict["end"] = end_datetime_utc
 
-        sql_query += " GROUP BY date order by date"
+        sql_query += " GROUP BY date ORDER BY date"
 
         response_data = []
 
@@ -146,18 +150,18 @@ class WorkflowDailyTokenCostStatistic(Resource):
         account = current_user
 
         parser = reqparse.RequestParser()
-        parser.add_argument("start", type=datetime_string("%Y-%m-%d %H:%M"), location="args")
-        parser.add_argument("end", type=datetime_string("%Y-%m-%d %H:%M"), location="args")
+        parser.add_argument("start", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
+        parser.add_argument("end", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
         args = parser.parse_args()
 
-        sql_query = """
-                SELECT 
-                    date(DATE_TRUNC('day', created_at AT TIME ZONE 'UTC' AT TIME ZONE :tz )) AS date, 
-                    SUM(workflow_runs.total_tokens) as token_count
-                FROM workflow_runs 
-                WHERE app_id = :app_id 
-                    AND triggered_from = :triggered_from
-                """
+        sql_query = """SELECT
+    DATE(DATE_TRUNC('day', created_at AT TIME ZONE 'UTC' AT TIME ZONE :tz )) AS date,
+    SUM(workflow_runs.total_tokens) AS token_count
+FROM
+    workflow_runs
+WHERE
+    app_id = :app_id
+    AND triggered_from = :triggered_from"""
         arg_dict = {
             "tz": account.timezone,
             "app_id": app_model.id,
@@ -174,7 +178,7 @@ class WorkflowDailyTokenCostStatistic(Resource):
             start_datetime_timezone = timezone.localize(start_datetime)
             start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)
 
-            sql_query += " and created_at >= :start"
+            sql_query += " AND created_at >= :start"
             arg_dict["start"] = start_datetime_utc
 
         if args["end"]:
@@ -184,10 +188,10 @@ class WorkflowDailyTokenCostStatistic(Resource):
             end_datetime_timezone = timezone.localize(end_datetime)
             end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
 
-            sql_query += " and created_at < :end"
+            sql_query += " AND created_at < :end"
             arg_dict["end"] = end_datetime_utc
 
-        sql_query += " GROUP BY date order by date"
+        sql_query += " GROUP BY date ORDER BY date"
 
         response_data = []
 
@@ -213,27 +217,31 @@ class WorkflowAverageAppInteractionStatistic(Resource):
         account = current_user
 
         parser = reqparse.RequestParser()
-        parser.add_argument("start", type=datetime_string("%Y-%m-%d %H:%M"), location="args")
-        parser.add_argument("end", type=datetime_string("%Y-%m-%d %H:%M"), location="args")
+        parser.add_argument("start", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
+        parser.add_argument("end", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
         args = parser.parse_args()
 
-        sql_query = """
-            SELECT 
-                AVG(sub.interactions) as interactions,
-                sub.date
-            FROM
-                (SELECT 
-                    date(DATE_TRUNC('day', c.created_at AT TIME ZONE 'UTC' AT TIME ZONE :tz )) AS date, 
-                    c.created_by,
-                    COUNT(c.id) AS interactions
-                FROM workflow_runs c
-                WHERE c.app_id = :app_id
-                    AND c.triggered_from = :triggered_from
-                    {{start}}
-                    {{end}}
-                GROUP BY date, c.created_by) sub
-            GROUP BY sub.date
-            """
+        sql_query = """SELECT
+    AVG(sub.interactions) AS interactions,
+    sub.date
+FROM
+    (
+        SELECT
+            DATE(DATE_TRUNC('day', c.created_at AT TIME ZONE 'UTC' AT TIME ZONE :tz )) AS date,
+            c.created_by,
+            COUNT(c.id) AS interactions
+        FROM
+            workflow_runs c
+        WHERE
+            c.app_id = :app_id
+            AND c.triggered_from = :triggered_from
+            {{start}}
+            {{end}}
+        GROUP BY
+            date, c.created_by
+    ) sub
+GROUP BY
+    sub.date"""
         arg_dict = {
             "tz": account.timezone,
             "app_id": app_model.id,
@@ -262,7 +270,7 @@ class WorkflowAverageAppInteractionStatistic(Resource):
             end_datetime_timezone = timezone.localize(end_datetime)
             end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
 
-            sql_query = sql_query.replace("{{end}}", " and c.created_at < :end")
+            sql_query = sql_query.replace("{{end}}", " AND c.created_at < :end")
             arg_dict["end"] = end_datetime_utc
         else:
             sql_query = sql_query.replace("{{end}}", "")

api/controllers/console/auth/activate.py

@@ -8,7 +8,7 @@ from constants.languages import supported_language
 from controllers.console import api
 from controllers.console.error import AlreadyActivateError
 from extensions.ext_database import db
-from libs.helper import email, str_len, timezone
+from libs.helper import StrLen, email, timezone
 from libs.password import hash_password, valid_password
 from models.account import AccountStatus
 from services.account_service import RegisterService
@@ -37,7 +37,7 @@ class ActivateApi(Resource):
         parser.add_argument("workspace_id", type=str, required=False, nullable=True, location="json")
         parser.add_argument("email", type=email, required=False, nullable=True, location="json")
         parser.add_argument("token", type=str, required=True, nullable=False, location="json")
-        parser.add_argument("name", type=str_len(30), required=True, nullable=False, location="json")
+        parser.add_argument("name", type=StrLen(30), required=True, nullable=False, location="json")
         parser.add_argument("password", type=valid_password, required=True, nullable=False, location="json")
         parser.add_argument(
             "interface_language", type=supported_language, required=True, nullable=False, location="json"

api/controllers/console/auth/oauth.py

@@ -71,7 +71,7 @@ class OAuthCallback(Resource):
 
         account = _generate_account(provider, user_info)
         # Check account status
-        if account.status == AccountStatus.BANNED.value or account.status == AccountStatus.CLOSED.value:
+        if account.status in {AccountStatus.BANNED.value, AccountStatus.CLOSED.value}:
             return {"error": "Account is banned or closed."}, 403
 
         if account.status == AccountStatus.PENDING.value:
@@ -101,7 +101,7 @@ def _generate_account(provider: str, user_info: OAuthUserInfo):
 
     if not account:
         # Create account
-        account_name = user_info.name if user_info.name else "Dify"
+        account_name = user_info.name or "Dify"
         account = RegisterService.register(
             email=user_info.email, name=account_name, password=None, open_id=user_info.id, provider=provider
         )

api/controllers/console/datasets/datasets.py

@@ -110,6 +110,26 @@ class DatasetListApi(Resource):
             nullable=True,
             help="Invalid indexing technique.",
         )
+        parser.add_argument(
+            "external_api_template_id",
+            type=str,
+            nullable=True,
+            required=False,
+        )
+        parser.add_argument(
+            "provider",
+            type=str,
+            nullable=True,
+            choices=Dataset.PROVIDER_LIST,
+            required=False,
+            default="vendor",
+        )
+        parser.add_argument(
+            "external_knowledge_id",
+            type=str,
+            nullable=True,
+            required=False,
+        )
         args = parser.parse_args()
 
         # The role of the current user in the ta table must be admin, owner, or editor, or dataset_operator
@@ -123,6 +143,9 @@ class DatasetListApi(Resource):
                 indexing_technique=args["indexing_technique"],
                 account=current_user,
                 permission=DatasetPermissionEnum.ONLY_ME,
+                provider=args["provider"],
+                external_api_template_id=args["external_api_template_id"],
+                external_knowledge_id=args["external_knowledge_id"],
             )
         except services.errors.dataset.DatasetNameDuplicateError:
             raise DatasetNameDuplicateError()
@@ -399,7 +422,7 @@ class DatasetIndexingEstimateApi(Resource):
             )
         except LLMBadRequestError:
             raise ProviderNotInitializeError(
-                "No Embedding Model available. Please configure a valid provider " "in the Settings -> Model Provider."
+                "No Embedding Model available. Please configure a valid provider in the Settings -> Model Provider."
             )
         except ProviderTokenNotInitError as ex:
             raise ProviderNotInitializeError(ex.description)
@@ -550,12 +573,7 @@ class DatasetApiBaseUrlApi(Resource):
     @login_required
     @account_initialization_required
     def get(self):
-        return {
-            "api_base_url": (
-                dify_config.SERVICE_API_URL if dify_config.SERVICE_API_URL else request.host_url.rstrip("/")
-            )
-            + "/v1"
-        }
+        return {"api_base_url": (dify_config.SERVICE_API_URL or request.host_url.rstrip("/")) + "/v1"}
 
 
 class DatasetRetrievalSettingApi(Resource):

api/controllers/console/datasets/datasets_document.py

@@ -354,7 +354,7 @@ class DocumentIndexingEstimateApi(DocumentResource):
         document_id = str(document_id)
         document = self.get_document(dataset_id, document_id)
 
-        if document.indexing_status in ["completed", "error"]:
+        if document.indexing_status in {"completed", "error"}:
             raise DocumentAlreadyFinishedError()
 
         data_process_rule = document.dataset_process_rule
@@ -421,7 +421,7 @@ class DocumentBatchIndexingEstimateApi(DocumentResource):
         info_list = []
         extract_settings = []
         for document in documents:
-            if document.indexing_status in ["completed", "error"]:
+            if document.indexing_status in {"completed", "error"}:
                 raise DocumentAlreadyFinishedError()
             data_source_info = document.data_source_info_dict
             # format document files info
@@ -665,7 +665,7 @@ class DocumentProcessingApi(DocumentResource):
             db.session.commit()
 
         elif action == "resume":
-            if document.indexing_status not in ["paused", "error"]:
+            if document.indexing_status not in {"paused", "error"}:
                 raise InvalidActionError("Document not in paused or error state.")
 
             document.paused_by = None

api/controllers/console/datasets/external.py

@@ -0,0 +1,254 @@
+from flask import request
+from flask_login import current_user
+from flask_restful import Resource, marshal, reqparse
+from werkzeug.exceptions import Forbidden, NotFound
+
+import services
+from controllers.console import api
+from controllers.console.app.error import ProviderNotInitializeError
+from controllers.console.datasets.error import DatasetNameDuplicateError
+from controllers.console.setup import setup_required
+from controllers.console.wraps import account_initialization_required
+from fields.dataset_fields import dataset_detail_fields
+from libs.login import login_required
+from services.external_knowledge_service import ExternalDatasetService
+
+
+def _validate_name(name):
+    if not name or len(name) < 1 or len(name) > 100:
+        raise ValueError("Name must be between 1 to 100 characters.")
+    return name
+
+
+def _validate_description_length(description):
+    if len(description) > 400:
+        raise ValueError("Description cannot exceed 400 characters.")
+    return description
+
+
+class ExternalApiTemplateListApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self):
+        page = request.args.get("page", default=1, type=int)
+        limit = request.args.get("limit", default=20, type=int)
+        search = request.args.get("keyword", default=None, type=str)
+
+        api_templates, total = ExternalDatasetService.get_external_api_templates(
+            page, limit, current_user.current_tenant_id, search
+        )
+        response = {
+            "data": [item.to_dict() for item in api_templates],
+            "has_more": len(api_templates) == limit,
+            "limit": limit,
+            "total": total,
+            "page": page,
+        }
+        return response, 200
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self):
+        parser = reqparse.RequestParser()
+        parser.add_argument(
+            "name",
+            nullable=False,
+            required=True,
+            help="Name is required. Name must be between 1 to 100 characters.",
+            type=_validate_name,
+        )
+        parser.add_argument(
+            "description",
+            nullable=False,
+            required=True,
+            help="Description is required. Description must be between 1 to 400 characters.",
+            type=_validate_description_length,
+        )
+        parser.add_argument(
+            "settings",
+            type=dict,
+            location="json",
+            nullable=False,
+            required=True,
+        )
+        args = parser.parse_args()
+
+        ExternalDatasetService.validate_api_list(args["settings"])
+
+        # The role of the current user in the ta table must be admin, owner, or editor, or dataset_operator
+        if not current_user.is_dataset_editor:
+            raise Forbidden()
+
+        try:
+            api_template = ExternalDatasetService.create_api_template(
+                tenant_id=current_user.current_tenant_id, user_id=current_user.id, args=args
+            )
+        except services.errors.dataset.DatasetNameDuplicateError:
+            raise DatasetNameDuplicateError()
+
+        return api_template.to_dict(), 201
+
+
+class ExternalApiTemplateApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, api_template_id):
+        api_template_id = str(api_template_id)
+        api_template = ExternalDatasetService.get_api_template(api_template_id)
+        if api_template is None:
+            raise NotFound("API template not found.")
+
+        return api_template.to_dict(), 200
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def patch(self, api_template_id):
+        api_template_id = str(api_template_id)
+
+        parser = reqparse.RequestParser()
+        parser.add_argument(
+            "name",
+            nullable=False,
+            required=True,
+            help="type is required. Name must be between 1 to 100 characters.",
+            type=_validate_name,
+        )
+        parser.add_argument(
+            "description",
+            nullable=False,
+            required=True,
+            help="description is required. Description must be between 1 to 400 characters.",
+            type=_validate_description_length,
+        )
+        parser.add_argument(
+            "settings",
+            type=dict,
+            location="json",
+            nullable=False,
+            required=True,
+        )
+        args = parser.parse_args()
+        ExternalDatasetService.validate_api_list(args["settings"])
+
+        api_template = ExternalDatasetService.update_api_template(
+            tenant_id=current_user.current_tenant_id,
+            user_id=current_user.id,
+            api_template_id=api_template_id,
+            args=args,
+        )
+
+        return api_template.to_dict(), 200
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def delete(self, api_template_id):
+        api_template_id = str(api_template_id)
+
+        # The role of the current user in the ta table must be admin, owner, or editor
+        if not current_user.is_editor or current_user.is_dataset_operator:
+            raise Forbidden()
+
+        ExternalDatasetService.delete_api_template(current_user.current_tenant_id, api_template_id)
+        return {"result": "success"}, 204
+
+
+class ExternalApiUseCheckApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, api_template_id):
+        api_template_id = str(api_template_id)
+
+        external_api_template_is_using = ExternalDatasetService.external_api_template_use_check(api_template_id)
+        return {"is_using": external_api_template_is_using}, 200
+
+
+class ExternalDatasetInitApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self):
+        # The role of the current user in the ta table must be admin, owner, or editor
+        if not current_user.is_editor:
+            raise Forbidden()
+
+        parser = reqparse.RequestParser()
+        parser.add_argument("api_template_id", type=str, required=True, nullable=True, location="json")
+        # parser.add_argument('name', nullable=False, required=True,
+        #                     help='name is required. Name must be between 1 to 100 characters.',
+        #                     type=_validate_name)
+        # parser.add_argument('description', type=str, required=True, nullable=True, location='json')
+        parser.add_argument("data_source", type=dict, required=True, nullable=True, location="json")
+        parser.add_argument("process_parameter", type=dict, required=True, nullable=True, location="json")
+
+        args = parser.parse_args()
+
+        # The role of the current user in the ta table must be admin, owner, or editor, or dataset_operator
+        if not current_user.is_dataset_editor:
+            raise Forbidden()
+
+        # validate args
+        ExternalDatasetService.document_create_args_validate(
+            current_user.current_tenant_id, args["api_template_id"], args["process_parameter"]
+        )
+
+        try:
+            dataset, documents, batch = ExternalDatasetService.init_external_dataset(
+                tenant_id=current_user.current_tenant_id,
+                user_id=current_user.id,
+                args=args,
+            )
+        except Exception as ex:
+            raise ProviderNotInitializeError(ex.description)
+        response = {"dataset": dataset, "documents": documents, "batch": batch}
+
+        return response
+
+
+class ExternalDatasetCreateApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self):
+        # The role of the current user in the ta table must be admin, owner, or editor
+        if not current_user.is_editor:
+            raise Forbidden()
+
+        parser = reqparse.RequestParser()
+        parser.add_argument("external_api_template_id", type=str, required=True, nullable=False, location="json")
+        parser.add_argument("external_knowledge_id", type=str, required=True, nullable=False, location="json")
+        parser.add_argument(
+            "name",
+            nullable=False,
+            required=True,
+            help="name is required. Name must be between 1 to 100 characters.",
+            type=_validate_name,
+        )
+        parser.add_argument("description", type=str, required=True, nullable=True, location="json")
+
+        args = parser.parse_args()
+
+        # The role of the current user in the ta table must be admin, owner, or editor, or dataset_operator
+        if not current_user.is_dataset_editor:
+            raise Forbidden()
+
+        try:
+            dataset = ExternalDatasetService.create_external_dataset(
+                tenant_id=current_user.current_tenant_id,
+                user_id=current_user.id,
+                args=args,
+            )
+        except services.errors.dataset.DatasetNameDuplicateError:
+            raise DatasetNameDuplicateError()
+
+        return marshal(dataset, dataset_detail_fields), 201
+
+
+api.add_resource(ExternalApiTemplateListApi, "/datasets/external-api-template")
+api.add_resource(ExternalApiTemplateApi, "/datasets/external-api-template/<uuid:api_template_id>")
+api.add_resource(ExternalApiUseCheckApi, "/datasets/external-api-template/<uuid:api_template_id>/use-check")

api/controllers/console/datasets/hit_testing.py

@@ -47,6 +47,7 @@ class HitTestingApi(Resource):
         parser = reqparse.RequestParser()
         parser.add_argument("query", type=str, location="json")
         parser.add_argument("retrieval_model", type=dict, required=False, location="json")
+        parser.add_argument("external_retrival_model", type=dict, required=False, location="json")
         args = parser.parse_args()
 
         HitTestingService.hit_testing_args_check(args)
@@ -57,6 +58,7 @@ class HitTestingApi(Resource):
                 query=args["query"],
                 account=current_user,
                 retrieval_model=args["retrieval_model"],
+                external_retrieval_model=args["external_retrival_model"],
                 limit=10,
             )
 

api/controllers/console/datasets/test_external.py

@@ -0,0 +1,49 @@
+from flask import request
+from flask_login import current_user
+from flask_restful import Resource, marshal, reqparse
+from werkzeug.exceptions import Forbidden, NotFound
+
+import services
+from controllers.console import api
+from controllers.console.app.error import ProviderNotInitializeError
+from controllers.console.datasets.error import DatasetNameDuplicateError
+from controllers.console.setup import setup_required
+from controllers.console.wraps import account_initialization_required
+from fields.dataset_fields import dataset_detail_fields
+from libs.login import login_required
+from services.external_knowledge_service import ExternalDatasetService
+
+class TestExternalApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self):
+        parser = reqparse.RequestParser()
+        parser.add_argument(
+            "top_k",
+            nullable=False,
+            required=True,
+            type=int,
+        )
+        parser.add_argument(
+            "score_threshold",
+            nullable=False,
+            required=True,
+            type=float,
+        )
+        args = parser.parse_args()
+        result = ExternalDatasetService.test_external_knowledge_retrival(
+            args["top_k"], args["score_threshold"]
+        )
+        response = {
+            "data": [item.to_dict() for item in api_templates],
+            "has_more": len(api_templates) == limit,
+            "limit": limit,
+            "total": total,
+            "page": page,
+        }
+        return response, 200
+
+
+
+api.add_resource(TestExternalApi, "/dify/external-knowledge/retrival-documents")

api/controllers/console/error.py

@@ -18,9 +18,7 @@ class NotSetupError(BaseHTTPException):
 
 class NotInitValidateError(BaseHTTPException):
     error_code = "not_init_validated"
-    description = (
-        "Init validation has not been completed yet. " "Please proceed with the init validation process first."
-    )
+    description = "Init validation has not been completed yet. Please proceed with the init validation process first."
     code = 401
 
 

api/controllers/console/explore/audio.py

@@ -81,19 +81,15 @@ class ChatTextApi(InstalledAppResource):
             message_id = args.get("message_id", None)
             text = args.get("text", None)
             if (
-                app_model.mode in [AppMode.ADVANCED_CHAT.value, AppMode.WORKFLOW.value]
+                app_model.mode in {AppMode.ADVANCED_CHAT.value, AppMode.WORKFLOW.value}
                 and app_model.workflow
                 and app_model.workflow.features_dict
             ):
                 text_to_speech = app_model.workflow.features_dict.get("text_to_speech")
-                voice = args.get("voice") if args.get("voice") else text_to_speech.get("voice")
+                voice = args.get("voice") or text_to_speech.get("voice")
             else:
                 try:
-                    voice = (
-                        args.get("voice")
-                        if args.get("voice")
-                        else app_model.app_model_config.text_to_speech_dict.get("voice")
-                    )
+                    voice = args.get("voice") or app_model.app_model_config.text_to_speech_dict.get("voice")
                 except Exception:
                     voice = None
             response = AudioService.transcript_tts(app_model=app_model, message_id=message_id, voice=voice, text=text)

api/controllers/console/explore/completion.py

@@ -92,7 +92,7 @@ class ChatApi(InstalledAppResource):
     def post(self, installed_app):
         app_model = installed_app.app
         app_mode = AppMode.value_of(app_model.mode)
-        if app_mode not in [AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT]:
+        if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
             raise NotChatAppError()
 
         parser = reqparse.RequestParser()
@@ -140,7 +140,7 @@ class ChatStopApi(InstalledAppResource):
     def post(self, installed_app, task_id):
         app_model = installed_app.app
         app_mode = AppMode.value_of(app_model.mode)
-        if app_mode not in [AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT]:
+        if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
             raise NotChatAppError()
 
         AppQueueManager.set_stop_flag(task_id, InvokeFrom.EXPLORE, current_user.id)

api/controllers/console/explore/conversation.py

@@ -20,7 +20,7 @@ class ConversationListApi(InstalledAppResource):
     def get(self, installed_app):
         app_model = installed_app.app
         app_mode = AppMode.value_of(app_model.mode)
-        if app_mode not in [AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT]:
+        if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
             raise NotChatAppError()
 
         parser = reqparse.RequestParser()
@@ -50,7 +50,7 @@ class ConversationApi(InstalledAppResource):
     def delete(self, installed_app, c_id):
         app_model = installed_app.app
         app_mode = AppMode.value_of(app_model.mode)
-        if app_mode not in [AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT]:
+        if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
             raise NotChatAppError()
 
         conversation_id = str(c_id)
@@ -68,7 +68,7 @@ class ConversationRenameApi(InstalledAppResource):
     def post(self, installed_app, c_id):
         app_model = installed_app.app
         app_mode = AppMode.value_of(app_model.mode)
-        if app_mode not in [AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT]:
+        if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
             raise NotChatAppError()
 
         conversation_id = str(c_id)
@@ -90,7 +90,7 @@ class ConversationPinApi(InstalledAppResource):
     def patch(self, installed_app, c_id):
         app_model = installed_app.app
         app_mode = AppMode.value_of(app_model.mode)
-        if app_mode not in [AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT]:
+        if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
             raise NotChatAppError()
 
         conversation_id = str(c_id)
@@ -107,7 +107,7 @@ class ConversationUnPinApi(InstalledAppResource):
     def patch(self, installed_app, c_id):
         app_model = installed_app.app
         app_mode = AppMode.value_of(app_model.mode)
-        if app_mode not in [AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT]:
+        if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
             raise NotChatAppError()
 
         conversation_id = str(c_id)

api/controllers/console/explore/installed_app.py

@@ -31,7 +31,7 @@ class InstalledAppsListApi(Resource):
                 "app_owner_tenant_id": installed_app.app_owner_tenant_id,
                 "is_pinned": installed_app.is_pinned,
                 "last_used_at": installed_app.last_used_at,
-                "editable": current_user.role in ["owner", "admin"],
+                "editable": current_user.role in {"owner", "admin"},
                 "uninstallable": current_tenant_id == installed_app.app_owner_tenant_id,
             }
             for installed_app in installed_apps

api/controllers/console/explore/message.py

@@ -40,7 +40,7 @@ class MessageListApi(InstalledAppResource):
         app_model = installed_app.app
 
         app_mode = AppMode.value_of(app_model.mode)
-        if app_mode not in [AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT]:
+        if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
             raise NotChatAppError()
 
         parser = reqparse.RequestParser()
@@ -125,7 +125,7 @@ class MessageSuggestedQuestionApi(InstalledAppResource):
     def get(self, installed_app, message_id):
         app_model = installed_app.app
         app_mode = AppMode.value_of(app_model.mode)
-        if app_mode not in [AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT]:
+        if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
             raise NotChatAppError()
 
         message_id = str(message_id)

api/controllers/console/explore/parameter.py

@@ -43,7 +43,7 @@ class AppParameterApi(InstalledAppResource):
         """Retrieve app parameters."""
         app_model = installed_app.app
 
-        if app_model.mode in [AppMode.ADVANCED_CHAT.value, AppMode.WORKFLOW.value]:
+        if app_model.mode in {AppMode.ADVANCED_CHAT.value, AppMode.WORKFLOW.value}:
             workflow = app_model.workflow
             if workflow is None:
                 raise AppUnavailableError()

api/controllers/console/init_validate.py

@@ -4,7 +4,7 @@ from flask import session
 from flask_restful import Resource, reqparse
 
 from configs import dify_config
-from libs.helper import str_len
+from libs.helper import StrLen
 from models.model import DifySetup
 from services.account_service import TenantService
 
@@ -28,7 +28,7 @@ class InitValidateAPI(Resource):
             raise AlreadySetupError()
 
         parser = reqparse.RequestParser()
-        parser.add_argument("password", type=str_len(30), required=True, location="json")
+        parser.add_argument("password", type=StrLen(30), required=True, location="json")
         input_password = parser.parse_args()["password"]
 
         if input_password != os.environ.get("INIT_PASSWORD"):

api/controllers/console/setup.py

@@ -4,7 +4,7 @@ from flask import request
 from flask_restful import Resource, reqparse
 
 from configs import dify_config
-from libs.helper import email, get_remote_ip, str_len
+from libs.helper import StrLen, email, get_remote_ip
 from libs.password import valid_password
 from models.model import DifySetup
 from services.account_service import RegisterService, TenantService
@@ -40,7 +40,7 @@ class SetupApi(Resource):
 
         parser = reqparse.RequestParser()
         parser.add_argument("email", type=email, required=True, location="json")
-        parser.add_argument("name", type=str_len(30), required=True, location="json")
+        parser.add_argument("name", type=StrLen(30), required=True, location="json")
         parser.add_argument("password", type=valid_password, required=True, location="json")
         args = parser.parse_args()
 

api/controllers/console/workspace/model_providers.py

@@ -218,7 +218,7 @@ api.add_resource(ModelProviderCredentialApi, "/workspaces/current/model-provider
 api.add_resource(ModelProviderValidateApi, "/workspaces/current/model-providers/<string:provider>/credentials/validate")
 api.add_resource(ModelProviderApi, "/workspaces/current/model-providers/<string:provider>")
 api.add_resource(
-    ModelProviderIconApi, "/workspaces/current/model-providers/<string:provider>/" "<string:icon_type>/<string:lang>"
+    ModelProviderIconApi, "/workspaces/current/model-providers/<string:provider>/<string:icon_type>/<string:lang>"
 )
 
 api.add_resource(

api/controllers/console/workspace/tool_providers.py

@@ -327,7 +327,7 @@ class ToolApiProviderPreviousTestApi(Resource):
 
         return ApiToolManageService.test_api_tool_preview(
             current_user.current_tenant_id,
-            args["provider_name"] if args["provider_name"] else "",
+            args["provider_name"] or "",
             args["tool_name"],
             args["credentials"],
             args["parameters"],

api/controllers/console/workspace/workspace.py

@@ -194,7 +194,7 @@ class WebappLogoWorkspaceApi(Resource):
             raise TooManyFilesError()
 
         extension = file.filename.split(".")[-1]
-        if extension.lower() not in ["svg", "png"]:
+        if extension.lower() not in {"svg", "png"}:
             raise UnsupportedFileTypeError()
 
         try:

api/controllers/console/wraps.py

@@ -64,7 +64,8 @@ def cloud_edition_billing_resource_check(resource: str):
                 elif resource == "vector_space" and 0 < vector_space.limit <= vector_space.size:
                     abort(403, "The capacity of the vector space has reached the limit of your subscription.")
                 elif resource == "documents" and 0 < documents_upload_quota.limit <= documents_upload_quota.size:
-                    # The api of file upload is used in the multiple places, so we need to check the source of the request from datasets
+                    # The api of file upload is used in the multiple places,
+                    # so we need to check the source of the request from datasets
                     source = request.args.get("source")
                     if source == "datasets":
                         abort(403, "The number of documents has reached the limit of your subscription.")

api/controllers/service_api/app/app.py

@@ -42,7 +42,7 @@ class AppParameterApi(Resource):
     @marshal_with(parameters_fields)
     def get(self, app_model: App):
         """Retrieve app parameters."""
-        if app_model.mode in [AppMode.ADVANCED_CHAT.value, AppMode.WORKFLOW.value]:
+        if app_model.mode in {AppMode.ADVANCED_CHAT.value, AppMode.WORKFLOW.value}:
             workflow = app_model.workflow
             if workflow is None:
                 raise AppUnavailableError()

api/controllers/service_api/app/audio.py

@@ -79,19 +79,15 @@ class TextApi(Resource):
             message_id = args.get("message_id", None)
             text = args.get("text", None)
             if (
-                app_model.mode in [AppMode.ADVANCED_CHAT.value, AppMode.WORKFLOW.value]
+                app_model.mode in {AppMode.ADVANCED_CHAT.value, AppMode.WORKFLOW.value}
                 and app_model.workflow
                 and app_model.workflow.features_dict
             ):
                 text_to_speech = app_model.workflow.features_dict.get("text_to_speech")
-                voice = args.get("voice") if args.get("voice") else text_to_speech.get("voice")
+                voice = args.get("voice") or text_to_speech.get("voice")
             else:
                 try:
-                    voice = (
-                        args.get("voice")
-                        if args.get("voice")
-                        else app_model.app_model_config.text_to_speech_dict.get("voice")
-                    )
+                    voice = args.get("voice") or app_model.app_model_config.text_to_speech_dict.get("voice")
                 except Exception:
                     voice = None
             response = AudioService.transcript_tts(

api/controllers/service_api/app/completion.py

@@ -96,7 +96,7 @@ class ChatApi(Resource):
     @validate_app_token(fetch_user_arg=FetchUserArg(fetch_from=WhereisUserArg.JSON, required=True))
     def post(self, app_model: App, end_user: EndUser):
         app_mode = AppMode.value_of(app_model.mode)
-        if app_mode not in [AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT]:
+        if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
             raise NotChatAppError()
 
         parser = reqparse.RequestParser()
@@ -144,7 +144,7 @@ class ChatStopApi(Resource):
     @validate_app_token(fetch_user_arg=FetchUserArg(fetch_from=WhereisUserArg.JSON, required=True))
     def post(self, app_model: App, end_user: EndUser, task_id):
         app_mode = AppMode.value_of(app_model.mode)
-        if app_mode not in [AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT]:
+        if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
             raise NotChatAppError()
 
         AppQueueManager.set_stop_flag(task_id, InvokeFrom.SERVICE_API, end_user.id)

api/controllers/service_api/app/conversation.py

@@ -18,7 +18,7 @@ class ConversationApi(Resource):
     @marshal_with(conversation_infinite_scroll_pagination_fields)
     def get(self, app_model: App, end_user: EndUser):
         app_mode = AppMode.value_of(app_model.mode)
-        if app_mode not in [AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT]:
+        if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
             raise NotChatAppError()
 
         parser = reqparse.RequestParser()
@@ -52,7 +52,7 @@ class ConversationDetailApi(Resource):
     @marshal_with(simple_conversation_fields)
     def delete(self, app_model: App, end_user: EndUser, c_id):
         app_mode = AppMode.value_of(app_model.mode)
-        if app_mode not in [AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT]:
+        if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
             raise NotChatAppError()
 
         conversation_id = str(c_id)
@@ -69,7 +69,7 @@ class ConversationRenameApi(Resource):
     @marshal_with(simple_conversation_fields)
     def post(self, app_model: App, end_user: EndUser, c_id):
         app_mode = AppMode.value_of(app_model.mode)
-        if app_mode not in [AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT]:
+        if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
             raise NotChatAppError()
 
         conversation_id = str(c_id)

api/controllers/service_api/app/message.py

@@ -76,7 +76,7 @@ class MessageListApi(Resource):
     @marshal_with(message_infinite_scroll_pagination_fields)
     def get(self, app_model: App, end_user: EndUser):
         app_mode = AppMode.value_of(app_model.mode)
-        if app_mode not in [AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT]:
+        if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
             raise NotChatAppError()
 
         parser = reqparse.RequestParser()
@@ -117,7 +117,7 @@ class MessageSuggestedApi(Resource):
     def get(self, app_model: App, end_user: EndUser, message_id):
         message_id = str(message_id)
         app_mode = AppMode.value_of(app_model.mode)
-        if app_mode not in [AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT]:
+        if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
             raise NotChatAppError()
 
         try:

api/controllers/service_api/app/workflow.py

@@ -1,6 +1,7 @@
 import logging
 
 from flask_restful import Resource, fields, marshal_with, reqparse
+from flask_restful.inputs import int_range
 from werkzeug.exceptions import InternalServerError
 
 from controllers.service_api import api
@@ -22,10 +23,12 @@ from core.errors.error import (
 )
 from core.model_runtime.errors.invoke import InvokeError
 from extensions.ext_database import db
+from fields.workflow_app_log_fields import workflow_app_log_pagination_fields
 from libs import helper
 from models.model import App, AppMode, EndUser
 from models.workflow import WorkflowRun
 from services.app_generate_service import AppGenerateService
+from services.workflow_app_service import WorkflowAppService
 
 logger = logging.getLogger(__name__)
 
@@ -113,6 +116,30 @@ class WorkflowTaskStopApi(Resource):
         return {"result": "success"}
 
 
+class WorkflowAppLogApi(Resource):
+    @validate_app_token
+    @marshal_with(workflow_app_log_pagination_fields)
+    def get(self, app_model: App):
+        """
+        Get workflow app logs
+        """
+        parser = reqparse.RequestParser()
+        parser.add_argument("keyword", type=str, location="args")
+        parser.add_argument("status", type=str, choices=["succeeded", "failed", "stopped"], location="args")
+        parser.add_argument("page", type=int_range(1, 99999), default=1, location="args")
+        parser.add_argument("limit", type=int_range(1, 100), default=20, location="args")
+        args = parser.parse_args()
+
+        # get paginate workflow app logs
+        workflow_app_service = WorkflowAppService()
+        workflow_app_log_pagination = workflow_app_service.get_paginate_workflow_app_logs(
+            app_model=app_model, args=args
+        )
+
+        return workflow_app_log_pagination
+
+
 api.add_resource(WorkflowRunApi, "/workflows/run")
 api.add_resource(WorkflowRunDetailApi, "/workflows/run/<string:workflow_id>")
 api.add_resource(WorkflowTaskStopApi, "/workflows/tasks/<string:task_id>/stop")
+api.add_resource(WorkflowAppLogApi, "/workflows/logs")

api/controllers/service_api/dataset/dataset.py

@@ -82,6 +82,26 @@ class DatasetListApi(DatasetApiResource):
             required=False,
             nullable=False,
         )
+        parser.add_argument(
+            "external_api_template_id",
+            type=str,
+            nullable=True,
+            required=False,
+            default="_validate_name",
+        )
+        parser.add_argument(
+            "provider",
+            type=str,
+            nullable=True,
+            required=False,
+            default="vendor",
+        )
+        parser.add_argument(
+            "external_knowledge_id",
+            type=str,
+            nullable=True,
+            required=False,
+        )
         args = parser.parse_args()
 
         try:
@@ -91,6 +111,9 @@ class DatasetListApi(DatasetApiResource):
                 indexing_technique=args["indexing_technique"],
                 account=current_user,
                 permission=args["permission"],
+                provider=args["provider"],
+                external_api_template_id=args["external_api_template_id"],
+                external_knowledge_id=args["external_knowledge_id"],
             )
         except services.errors.dataset.DatasetNameDuplicateError:
             raise DatasetNameDuplicateError()

api/controllers/web/app.py

@@ -41,7 +41,7 @@ class AppParameterApi(WebApiResource):
     @marshal_with(parameters_fields)
     def get(self, app_model: App, end_user):
         """Retrieve app parameters."""
-        if app_model.mode in [AppMode.ADVANCED_CHAT.value, AppMode.WORKFLOW.value]:
+        if app_model.mode in {AppMode.ADVANCED_CHAT.value, AppMode.WORKFLOW.value}:
             workflow = app_model.workflow
             if workflow is None:
                 raise AppUnavailableError()

api/controllers/web/audio.py

@@ -78,19 +78,15 @@ class TextApi(WebApiResource):
             message_id = args.get("message_id", None)
             text = args.get("text", None)
             if (
-                app_model.mode in [AppMode.ADVANCED_CHAT.value, AppMode.WORKFLOW.value]
+                app_model.mode in {AppMode.ADVANCED_CHAT.value, AppMode.WORKFLOW.value}
                 and app_model.workflow
                 and app_model.workflow.features_dict
             ):
                 text_to_speech = app_model.workflow.features_dict.get("text_to_speech")
-                voice = args.get("voice") if args.get("voice") else text_to_speech.get("voice")
+                voice = args.get("voice") or text_to_speech.get("voice")
             else:
                 try:
-                    voice = (
-                        args.get("voice")
-                        if args.get("voice")
-                        else app_model.app_model_config.text_to_speech_dict.get("voice")
-                    )
+                    voice = args.get("voice") or app_model.app_model_config.text_to_speech_dict.get("voice")
                 except Exception:
                     voice = None
 

api/controllers/web/completion.py

@@ -87,7 +87,7 @@ class CompletionStopApi(WebApiResource):
 class ChatApi(WebApiResource):
     def post(self, app_model, end_user):
         app_mode = AppMode.value_of(app_model.mode)
-        if app_mode not in [AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT]:
+        if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
             raise NotChatAppError()
 
         parser = reqparse.RequestParser()
@@ -136,7 +136,7 @@ class ChatApi(WebApiResource):
 class ChatStopApi(WebApiResource):
     def post(self, app_model, end_user, task_id):
         app_mode = AppMode.value_of(app_model.mode)
-        if app_mode not in [AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT]:
+        if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
             raise NotChatAppError()
 
         AppQueueManager.set_stop_flag(task_id, InvokeFrom.WEB_APP, end_user.id)

api/controllers/web/conversation.py

@@ -18,7 +18,7 @@ class ConversationListApi(WebApiResource):
     @marshal_with(conversation_infinite_scroll_pagination_fields)
     def get(self, app_model, end_user):
         app_mode = AppMode.value_of(app_model.mode)
-        if app_mode not in [AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT]:
+        if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
             raise NotChatAppError()
 
         parser = reqparse.RequestParser()
@@ -56,7 +56,7 @@ class ConversationListApi(WebApiResource):
 class ConversationApi(WebApiResource):
     def delete(self, app_model, end_user, c_id):
         app_mode = AppMode.value_of(app_model.mode)
-        if app_mode not in [AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT]:
+        if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
             raise NotChatAppError()
 
         conversation_id = str(c_id)
@@ -73,7 +73,7 @@ class ConversationRenameApi(WebApiResource):
     @marshal_with(simple_conversation_fields)
     def post(self, app_model, end_user, c_id):
         app_mode = AppMode.value_of(app_model.mode)
-        if app_mode not in [AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT]:
+        if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
             raise NotChatAppError()
 
         conversation_id = str(c_id)
@@ -92,7 +92,7 @@ class ConversationRenameApi(WebApiResource):
 class ConversationPinApi(WebApiResource):
     def patch(self, app_model, end_user, c_id):
         app_mode = AppMode.value_of(app_model.mode)
-        if app_mode not in [AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT]:
+        if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
             raise NotChatAppError()
 
         conversation_id = str(c_id)
@@ -108,7 +108,7 @@ class ConversationPinApi(WebApiResource):
 class ConversationUnPinApi(WebApiResource):
     def patch(self, app_model, end_user, c_id):
         app_mode = AppMode.value_of(app_model.mode)
-        if app_mode not in [AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT]:
+        if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
             raise NotChatAppError()
 
         conversation_id = str(c_id)

api/controllers/web/message.py

@@ -78,7 +78,7 @@ class MessageListApi(WebApiResource):
     @marshal_with(message_infinite_scroll_pagination_fields)
     def get(self, app_model, end_user):
         app_mode = AppMode.value_of(app_model.mode)
-        if app_mode not in [AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT]:
+        if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
             raise NotChatAppError()
 
         parser = reqparse.RequestParser()
@@ -160,7 +160,7 @@ class MessageMoreLikeThisApi(WebApiResource):
 class MessageSuggestedQuestionApi(WebApiResource):
     def get(self, app_model, end_user, message_id):
         app_mode = AppMode.value_of(app_model.mode)
-        if app_mode not in [AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT]:
+        if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
             raise NotCompletionAppError()
 
         message_id = str(message_id)

api/controllers/web/wraps.py

@@ -80,7 +80,8 @@ def _validate_web_sso_token(decoded, system_features, app_code):
             if not source or source != "sso":
                 raise WebSSOAuthRequiredError()
 
-    # Check if SSO is not enforced for web, and if the token source is SSO, raise an error and redirect to normal passport login
+    # Check if SSO is not enforced for web, and if the token source is SSO,
+    # raise an error and redirect to normal passport login
     if not system_features.sso_enforced_for_web or not app_web_sso_enabled:
         source = decoded.get("token_source")
         if source and source == "sso":

api/core/agent/cot_agent_runner.py

@@ -256,7 +256,7 @@ class CotAgentRunner(BaseAgentRunner, ABC):
                     model=model_instance.model,
                     prompt_messages=prompt_messages,
                     message=AssistantPromptMessage(content=final_answer),
-                    usage=llm_usage["usage"] if llm_usage["usage"] else LLMUsage.empty_usage(),
+                    usage=llm_usage["usage"] or LLMUsage.empty_usage(),
                     system_fingerprint="",
                 )
             ),

api/core/agent/fc_agent_runner.py

@@ -298,7 +298,7 @@ class FunctionCallAgentRunner(BaseAgentRunner):
                     model=model_instance.model,
                     prompt_messages=prompt_messages,
                     message=AssistantPromptMessage(content=final_answer),
-                    usage=llm_usage["usage"] if llm_usage["usage"] else LLMUsage.empty_usage(),
+                    usage=llm_usage["usage"] or LLMUsage.empty_usage(),
                     system_fingerprint="",
                 )
             ),

api/core/agent/output_parser/cot_output_parser.py

@@ -90,7 +90,7 @@ class CotAgentOutputParser:
 
                 if not in_code_block and not in_json:
                     if delta.lower() == action_str[action_idx] and action_idx == 0:
-                        if last_character not in ["\n", " ", ""]:
+                        if last_character not in {"\n", " ", ""}:
                             index += steps
                             yield delta
                             continue
@@ -117,7 +117,7 @@ class CotAgentOutputParser:
                             action_idx = 0
 
                     if delta.lower() == thought_str[thought_idx] and thought_idx == 0:
-                        if last_character not in ["\n", " ", ""]:
+                        if last_character not in {"\n", " ", ""}:
                             index += steps
                             yield delta
                             continue

api/core/agent/prompt/template.py

@@ -41,7 +41,8 @@ Begin! Reminder to ALWAYS respond with a valid json blob of a single action. Use
 {{historic_messages}}
 Question: {{query}}
 {{agent_scratchpad}}
-Thought:"""
+Thought:"""  # noqa: E501
+
 
 ENGLISH_REACT_COMPLETION_AGENT_SCRATCHPAD_TEMPLATES = """Observation: {{observation}}
 Thought:"""
@@ -86,7 +87,8 @@ Action:
 ```
 
 Begin! Reminder to ALWAYS respond with a valid json blob of a single action. Use tools if necessary. Respond directly if appropriate. Format is Action:```$JSON_BLOB```then Observation:.
-"""
+"""  # noqa: E501
+
 
 ENGLISH_REACT_CHAT_AGENT_SCRATCHPAD_TEMPLATES = ""
 

api/core/app/app_config/base_app_config_manager.py

@@ -29,7 +29,7 @@ class BaseAppConfigManager:
         additional_features.show_retrieve_source = RetrievalResourceConfigManager.convert(config=config_dict)
 
         additional_features.file_upload = FileUploadConfigManager.convert(
-            config=config_dict, is_vision=app_mode in [AppMode.CHAT, AppMode.COMPLETION, AppMode.AGENT_CHAT]
+            config=config_dict, is_vision=app_mode in {AppMode.CHAT, AppMode.COMPLETION, AppMode.AGENT_CHAT}
         )
 
         additional_features.opening_statement, additional_features.suggested_questions = (

api/core/app/app_config/easy_ui_based_app/agent/manager.py

@@ -18,7 +18,7 @@ class AgentConfigManager:
 
             if agent_strategy == "function_call":
                 strategy = AgentEntity.Strategy.FUNCTION_CALLING
-            elif agent_strategy == "cot" or agent_strategy == "react":
+            elif agent_strategy in {"cot", "react"}:
                 strategy = AgentEntity.Strategy.CHAIN_OF_THOUGHT
             else:
                 # old configs, try to detect default strategy
@@ -43,10 +43,10 @@ class AgentConfigManager:
 
                     agent_tools.append(AgentToolEntity(**agent_tool_properties))
 
-            if "strategy" in config["agent_mode"] and config["agent_mode"]["strategy"] not in [
+            if "strategy" in config["agent_mode"] and config["agent_mode"]["strategy"] not in {
                 "react_router",
                 "router",
-            ]:
+            }:
                 agent_prompt = agent_dict.get("prompt", None) or {}
                 # check model mode
                 model_mode = config.get("model", {}).get("mode", "completion")

api/core/app/app_config/easy_ui_based_app/dataset/manager.py

@@ -167,7 +167,7 @@ class DatasetConfigManager:
             config["agent_mode"]["strategy"] = PlanningStrategy.ROUTER.value
 
         has_datasets = False
-        if config["agent_mode"]["strategy"] in [PlanningStrategy.ROUTER.value, PlanningStrategy.REACT_ROUTER.value]:
+        if config["agent_mode"]["strategy"] in {PlanningStrategy.ROUTER.value, PlanningStrategy.REACT_ROUTER.value}:
             for tool in config["agent_mode"]["tools"]:
                 key = list(tool.keys())[0]
                 if key == "dataset":

api/core/app/app_config/easy_ui_based_app/prompt_template/manager.py

@@ -86,7 +86,7 @@ class PromptTemplateConfigManager:
         if config["prompt_type"] == PromptTemplateEntity.PromptType.ADVANCED.value:
             if not config["chat_prompt_config"] and not config["completion_prompt_config"]:
                 raise ValueError(
-                    "chat_prompt_config or completion_prompt_config is required " "when prompt_type is advanced"
+                    "chat_prompt_config or completion_prompt_config is required when prompt_type is advanced"
                 )
 
             model_mode_vals = [mode.value for mode in ModelMode]

api/core/app/app_config/easy_ui_based_app/variables/manager.py

@@ -42,12 +42,12 @@ class BasicVariablesConfigManager:
                         variable=variable["variable"], type=variable["type"], config=variable["config"]
                     )
                 )
-            elif variable_type in [
+            elif variable_type in {
                 VariableEntityType.TEXT_INPUT,
                 VariableEntityType.PARAGRAPH,
                 VariableEntityType.NUMBER,
                 VariableEntityType.SELECT,
-            ]:
+            }:
                 variable = variables[variable_type]
                 variable_entities.append(
                     VariableEntity(
@@ -97,7 +97,7 @@ class BasicVariablesConfigManager:
         variables = []
         for item in config["user_input_form"]:
             key = list(item.keys())[0]
-            if key not in ["text-input", "select", "paragraph", "number", "external_data_tool"]:
+            if key not in {"text-input", "select", "paragraph", "number", "external_data_tool"}:
                 raise ValueError("Keys in user_input_form list can only be 'text-input', 'paragraph'  or 'select'")
 
             form_item = item[key]
@@ -115,7 +115,7 @@ class BasicVariablesConfigManager:
 
             pattern = re.compile(r"^(?!\d)[\u4e00-\u9fa5A-Za-z0-9_\U0001F300-\U0001F64F\U0001F680-\U0001F6FF]{1,100}$")
             if pattern.match(form_item["variable"]) is None:
-                raise ValueError("variable in user_input_form must be a string, " "and cannot start with a number")
+                raise ValueError("variable in user_input_form must be a string, and cannot start with a number")
 
             variables.append(form_item["variable"])
 

api/core/app/app_config/entities.py

@@ -92,7 +92,7 @@ class VariableEntityType(str, Enum):
     SELECT = "select"
     PARAGRAPH = "paragraph"
     NUMBER = "number"
-    EXTERNAL_DATA_TOOL = "external-data-tool"
+    EXTERNAL_DATA_TOOL = "external_data_tool"
 
 
 class VariableEntity(BaseModel):

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

@@ -54,14 +54,14 @@ class FileUploadConfigManager:
 
             if is_vision:
                 detail = config["file_upload"]["image"]["detail"]
-                if detail not in ["high", "low"]:
+                if detail not in {"high", "low"}:
                     raise ValueError("detail must be in ['high', 'low']")
 
             transfer_methods = config["file_upload"]["image"]["transfer_methods"]
             if not isinstance(transfer_methods, list):
                 raise ValueError("transfer_methods must be of list type")
             for method in transfer_methods:
-                if method not in ["remote_url", "local_file"]:
+                if method not in {"remote_url", "local_file"}:
                     raise ValueError("transfer_methods must be in ['remote_url', 'local_file']")
 
         return config, ["file_upload"]

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

@@ -15,7 +15,7 @@ from core.app.apps.advanced_chat.app_config_manager import AdvancedChatAppConfig
 from core.app.apps.advanced_chat.app_runner import AdvancedChatAppRunner
 from core.app.apps.advanced_chat.generate_response_converter import AdvancedChatAppGenerateResponseConverter
 from core.app.apps.advanced_chat.generate_task_pipeline import AdvancedChatAppGenerateTaskPipeline
-from core.app.apps.base_app_queue_manager import AppQueueManager, GenerateTaskStoppedException, PublishFrom
+from core.app.apps.base_app_queue_manager import AppQueueManager, GenerateTaskStoppedError, PublishFrom
 from core.app.apps.message_based_app_generator import MessageBasedAppGenerator
 from core.app.apps.message_based_app_queue_manager import MessageBasedAppQueueManager
 from core.app.entities.app_invoke_entities import AdvancedChatAppGenerateEntity, InvokeFrom
@@ -293,7 +293,7 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
                 )
 
                 runner.run()
-            except GenerateTaskStoppedException:
+            except GenerateTaskStoppedError:
                 pass
             except InvokeAuthorizationError:
                 queue_manager.publish_error(
@@ -349,7 +349,7 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
             return generate_task_pipeline.process()
         except ValueError as e:
             if e.args[0] == "I/O operation on closed file.":  # ignore this error
-                raise GenerateTaskStoppedException()
+                raise GenerateTaskStoppedError()
             else:
                 logger.exception(e)
                 raise e

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

@@ -21,7 +21,7 @@ class AudioTrunk:
         self.status = status
 
 
-def _invoiceTTS(text_content: str, model_instance, tenant_id: str, voice: str):
+def _invoice_tts(text_content: str, model_instance, tenant_id: str, voice: str):
     if not text_content or text_content.isspace():
         return
     return model_instance.invoke_tts(
@@ -81,7 +81,7 @@ class AppGeneratorTTSPublisher:
                 if message is None:
                     if self.msg_text and len(self.msg_text.strip()) > 0:
                         futures_result = self.executor.submit(
-                            _invoiceTTS, self.msg_text, self.model_instance, self.tenant_id, self.voice
+                            _invoice_tts, self.msg_text, self.model_instance, self.tenant_id, self.voice
                         )
                         future_queue.put(futures_result)
                     break
@@ -97,7 +97,7 @@ class AppGeneratorTTSPublisher:
                     self.MAX_SENTENCE += 1
                     text_content = "".join(sentence_arr)
                     futures_result = self.executor.submit(
-                        _invoiceTTS, text_content, self.model_instance, self.tenant_id, self.voice
+                        _invoice_tts, text_content, self.model_instance, self.tenant_id, self.voice
                     )
                     future_queue.put(futures_result)
                     if text_tmp:
@@ -110,7 +110,7 @@ class AppGeneratorTTSPublisher:
                 break
         future_queue.put(None)
 
-    def checkAndGetAudio(self) -> AudioTrunk | None:
+    def check_and_get_audio(self) -> AudioTrunk | None:
         try:
             if self._last_audio_event and self._last_audio_event.status == "finish":
                 if self.executor:

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

@@ -19,7 +19,7 @@ from core.app.entities.queue_entities import (
     QueueStopEvent,
     QueueTextChunkEvent,
 )
-from core.moderation.base import ModerationException
+from core.moderation.base import ModerationError
 from core.workflow.callbacks.base_workflow_callback import WorkflowCallback
 from core.workflow.entities.node_entities import UserFrom
 from core.workflow.entities.variable_pool import VariablePool
@@ -73,7 +73,7 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
             raise ValueError("Workflow not initialized")
 
         user_id = None
-        if self.application_generate_entity.invoke_from in [InvokeFrom.WEB_APP, InvokeFrom.SERVICE_API]:
+        if self.application_generate_entity.invoke_from in {InvokeFrom.WEB_APP, InvokeFrom.SERVICE_API}:
             end_user = db.session.query(EndUser).filter(EndUser.id == self.application_generate_entity.user_id).first()
             if end_user:
                 user_id = end_user.session_id
@@ -175,7 +175,7 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
             user_id=self.application_generate_entity.user_id,
             user_from=(
                 UserFrom.ACCOUNT
-                if self.application_generate_entity.invoke_from in [InvokeFrom.EXPLORE, InvokeFrom.DEBUGGER]
+                if self.application_generate_entity.invoke_from in {InvokeFrom.EXPLORE, InvokeFrom.DEBUGGER}
                 else UserFrom.END_USER
             ),
             invoke_from=self.application_generate_entity.invoke_from,
@@ -217,7 +217,7 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
                 query=query,
                 message_id=message_id,
             )
-        except ModerationException as e:
+        except ModerationError as e:
             self._complete_with_stream_output(text=str(e), stopped_by=QueueStopEvent.StopBy.INPUT_MODERATION)
             return True
 

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

@@ -179,10 +179,10 @@ class AdvancedChatAppGenerateTaskPipeline(BasedGenerateTaskPipeline, WorkflowCyc
                 stream_response=stream_response,
             )
 
-    def _listenAudioMsg(self, publisher, task_id: str):
+    def _listen_audio_msg(self, publisher, task_id: str):
         if not publisher:
             return None
-        audio_msg: AudioTrunk = publisher.checkAndGetAudio()
+        audio_msg: AudioTrunk = publisher.check_and_get_audio()
         if audio_msg and audio_msg.status != "finish":
             return MessageAudioStreamResponse(audio=audio_msg.audio, task_id=task_id)
         return None
@@ -204,7 +204,7 @@ class AdvancedChatAppGenerateTaskPipeline(BasedGenerateTaskPipeline, WorkflowCyc
 
         for response in self._process_stream_response(tts_publisher=tts_publisher, trace_manager=trace_manager):
             while True:
-                audio_response = self._listenAudioMsg(tts_publisher, task_id=task_id)
+                audio_response = self._listen_audio_msg(tts_publisher, task_id=task_id)
                 if audio_response:
                     yield audio_response
                 else:
@@ -217,7 +217,7 @@ class AdvancedChatAppGenerateTaskPipeline(BasedGenerateTaskPipeline, WorkflowCyc
             try:
                 if not tts_publisher:
                     break
-                audio_trunk = tts_publisher.checkAndGetAudio()
+                audio_trunk = tts_publisher.check_and_get_audio()
                 if audio_trunk is None:
                     # release cpu
                     # sleep 20 ms ( 40ms => 1280 byte audio file,20ms => 640 byte audio file)
@@ -451,7 +451,9 @@ class AdvancedChatAppGenerateTaskPipeline(BasedGenerateTaskPipeline, WorkflowCyc
                     tts_publisher.publish(message=queue_message)
 
                 self._task_state.answer += delta_text
-                yield self._message_to_stream_response(delta_text, self._message.id)
+                yield self._message_to_stream_response(
+                    answer=delta_text, message_id=self._message.id, from_variable_selector=event.from_variable_selector
+                )
             elif isinstance(event, QueueMessageReplaceEvent):
                 # published by moderation
                 yield self._message_replace_to_stream_response(answer=event.text)

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

@@ -13,7 +13,7 @@ from core.app.app_config.features.file_upload.manager import FileUploadConfigMan
 from core.app.apps.agent_chat.app_config_manager import AgentChatAppConfigManager
 from core.app.apps.agent_chat.app_runner import AgentChatAppRunner
 from core.app.apps.agent_chat.generate_response_converter import AgentChatAppGenerateResponseConverter
-from core.app.apps.base_app_queue_manager import AppQueueManager, GenerateTaskStoppedException, PublishFrom
+from core.app.apps.base_app_queue_manager import AppQueueManager, GenerateTaskStoppedError, PublishFrom
 from core.app.apps.message_based_app_generator import MessageBasedAppGenerator
 from core.app.apps.message_based_app_queue_manager import MessageBasedAppQueueManager
 from core.app.entities.app_invoke_entities import AgentChatAppGenerateEntity, InvokeFrom
@@ -205,7 +205,7 @@ class AgentChatAppGenerator(MessageBasedAppGenerator):
                     conversation=conversation,
                     message=message,
                 )
-            except GenerateTaskStoppedException:
+            except GenerateTaskStoppedError:
                 pass
             except InvokeAuthorizationError:
                 queue_manager.publish_error(

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

@@ -15,7 +15,7 @@ from core.model_manager import ModelInstance
 from core.model_runtime.entities.llm_entities import LLMMode, LLMUsage
 from core.model_runtime.entities.model_entities import ModelFeature, ModelPropertyKey
 from core.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
-from core.moderation.base import ModerationException
+from core.moderation.base import ModerationError
 from core.tools.entities.tool_entities import ToolRuntimeVariablePool
 from extensions.ext_database import db
 from models.model import App, Conversation, Message, MessageAgentThought
@@ -103,7 +103,7 @@ class AgentChatAppRunner(AppRunner):
                 query=query,
                 message_id=message.id,
             )
-        except ModerationException as e:
+        except ModerationError as e:
             self.direct_output(
                 queue_manager=queue_manager,
                 app_generate_entity=application_generate_entity,

api/core/app/apps/base_app_generate_response_converter.py

@@ -16,7 +16,7 @@ class AppGenerateResponseConverter(ABC):
     def convert(
         cls, response: Union[AppBlockingResponse, Generator[AppStreamResponse, Any, None]], invoke_from: InvokeFrom
     ) -> dict[str, Any] | Generator[str, Any, None]:
-        if invoke_from in [InvokeFrom.DEBUGGER, InvokeFrom.SERVICE_API]:
+        if invoke_from in {InvokeFrom.DEBUGGER, InvokeFrom.SERVICE_API}:
             if isinstance(response, AppBlockingResponse):
                 return cls.convert_blocking_full_response(response)
             else:

api/core/app/apps/base_app_generator.py

@@ -22,11 +22,11 @@ class BaseAppGenerator:
             return var.default or ""
         if (
             var.type
-            in (
+            in {
                 VariableEntityType.TEXT_INPUT,
                 VariableEntityType.SELECT,
                 VariableEntityType.PARAGRAPH,
-            )
+            }
             and user_input_value
             and not isinstance(user_input_value, str)
         ):
@@ -44,7 +44,7 @@ class BaseAppGenerator:
             options = var.options or []
             if user_input_value not in options:
                 raise ValueError(f"{var.variable} in input form must be one of the following: {options}")
-        elif var.type in (VariableEntityType.TEXT_INPUT, VariableEntityType.PARAGRAPH):
+        elif var.type in {VariableEntityType.TEXT_INPUT, VariableEntityType.PARAGRAPH}:
             if var.max_length and user_input_value and len(user_input_value) > var.max_length:
                 raise ValueError(f"{var.variable} in input form must be less than {var.max_length} characters")
 

api/core/app/apps/base_app_queue_manager.py

@@ -32,7 +32,7 @@ class AppQueueManager:
         self._user_id = user_id
         self._invoke_from = invoke_from
 
-        user_prefix = "account" if self._invoke_from in [InvokeFrom.EXPLORE, InvokeFrom.DEBUGGER] else "end-user"
+        user_prefix = "account" if self._invoke_from in {InvokeFrom.EXPLORE, InvokeFrom.DEBUGGER} else "end-user"
         redis_client.setex(
             AppQueueManager._generate_task_belong_cache_key(self._task_id), 1800, f"{user_prefix}-{self._user_id}"
         )
@@ -118,7 +118,7 @@ class AppQueueManager:
         if result is None:
             return
 
-        user_prefix = "account" if invoke_from in [InvokeFrom.EXPLORE, InvokeFrom.DEBUGGER] else "end-user"
+        user_prefix = "account" if invoke_from in {InvokeFrom.EXPLORE, InvokeFrom.DEBUGGER} else "end-user"
         if result.decode("utf-8") != f"{user_prefix}-{user_id}":
             return
 
@@ -171,5 +171,5 @@ class AppQueueManager:
                 )
 
 
-class GenerateTaskStoppedException(Exception):
+class GenerateTaskStoppedError(Exception):
     pass

api/core/app/apps/base_app_runner.py

@@ -161,7 +161,7 @@ class AppRunner:
                 app_mode=AppMode.value_of(app_record.mode),
                 prompt_template_entity=prompt_template_entity,
                 inputs=inputs,
-                query=query if query else "",
+                query=query or "",
                 files=files,
                 context=context,
                 memory=memory,
@@ -189,7 +189,7 @@ class AppRunner:
             prompt_messages = prompt_transform.get_prompt(
                 prompt_template=prompt_template,
                 inputs=inputs,
-                query=query if query else "",
+                query=query or "",
                 files=files,
                 context=context,
                 memory_config=memory_config,
@@ -238,7 +238,7 @@ class AppRunner:
                     model=app_generate_entity.model_conf.model,
                     prompt_messages=prompt_messages,
                     message=AssistantPromptMessage(content=text),
-                    usage=usage if usage else LLMUsage.empty_usage(),
+                    usage=usage or LLMUsage.empty_usage(),
                 ),
             ),
             PublishFrom.APPLICATION_MANAGER,
@@ -351,7 +351,7 @@ class AppRunner:
             tenant_id=tenant_id,
             app_config=app_generate_entity.app_config,
             inputs=inputs,
-            query=query if query else "",
+            query=query or "",
             message_id=message_id,
             trace_manager=app_generate_entity.trace_manager,
         )
@@ -379,7 +379,7 @@ class AppRunner:
                 queue_manager=queue_manager,
                 app_generate_entity=application_generate_entity,
                 prompt_messages=prompt_messages,
-                text="I apologize for any confusion, " "but I'm an AI assistant to be helpful, harmless, and honest.",
+                text="I apologize for any confusion, but I'm an AI assistant to be helpful, harmless, and honest.",
                 stream=application_generate_entity.stream,
             )
 

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

@@ -10,7 +10,7 @@ from pydantic import ValidationError
 
 from core.app.app_config.easy_ui_based_app.model_config.converter import ModelConfigConverter
 from core.app.app_config.features.file_upload.manager import FileUploadConfigManager
-from core.app.apps.base_app_queue_manager import AppQueueManager, GenerateTaskStoppedException, PublishFrom
+from core.app.apps.base_app_queue_manager import AppQueueManager, GenerateTaskStoppedError, PublishFrom
 from core.app.apps.chat.app_config_manager import ChatAppConfigManager
 from core.app.apps.chat.app_runner import ChatAppRunner
 from core.app.apps.chat.generate_response_converter import ChatAppGenerateResponseConverter
@@ -205,7 +205,7 @@ class ChatAppGenerator(MessageBasedAppGenerator):
                     conversation=conversation,
                     message=message,
                 )
-            except GenerateTaskStoppedException:
+            except GenerateTaskStoppedError:
                 pass
             except InvokeAuthorizationError:
                 queue_manager.publish_error(

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

@@ -11,7 +11,7 @@ from core.app.entities.queue_entities import QueueAnnotationReplyEvent
 from core.callback_handler.index_tool_callback_handler import DatasetIndexToolCallbackHandler
 from core.memory.token_buffer_memory import TokenBufferMemory
 from core.model_manager import ModelInstance
-from core.moderation.base import ModerationException
+from core.moderation.base import ModerationError
 from core.rag.retrieval.dataset_retrieval import DatasetRetrieval
 from extensions.ext_database import db
 from models.model import App, Conversation, Message
@@ -98,7 +98,7 @@ class ChatAppRunner(AppRunner):
                 query=query,
                 message_id=message.id,
             )
-        except ModerationException as e:
+        except ModerationError as e:
             self.direct_output(
                 queue_manager=queue_manager,
                 app_generate_entity=application_generate_entity,

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

@@ -10,7 +10,7 @@ from pydantic import ValidationError
 
 from core.app.app_config.easy_ui_based_app.model_config.converter import ModelConfigConverter
 from core.app.app_config.features.file_upload.manager import FileUploadConfigManager
-from core.app.apps.base_app_queue_manager import AppQueueManager, GenerateTaskStoppedException, PublishFrom
+from core.app.apps.base_app_queue_manager import AppQueueManager, GenerateTaskStoppedError, PublishFrom
 from core.app.apps.completion.app_config_manager import CompletionAppConfigManager
 from core.app.apps.completion.app_runner import CompletionAppRunner
 from core.app.apps.completion.generate_response_converter import CompletionAppGenerateResponseConverter
@@ -185,7 +185,7 @@ class CompletionAppGenerator(MessageBasedAppGenerator):
                     queue_manager=queue_manager,
                     message=message,
                 )
-            except GenerateTaskStoppedException:
+            except GenerateTaskStoppedError:
                 pass
             except InvokeAuthorizationError:
                 queue_manager.publish_error(

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

@@ -9,7 +9,7 @@ from core.app.entities.app_invoke_entities import (
 )
 from core.callback_handler.index_tool_callback_handler import DatasetIndexToolCallbackHandler
 from core.model_manager import ModelInstance
-from core.moderation.base import ModerationException
+from core.moderation.base import ModerationError
 from core.rag.retrieval.dataset_retrieval import DatasetRetrieval
 from extensions.ext_database import db
 from models.model import App, Message
@@ -79,7 +79,7 @@ class CompletionAppRunner(AppRunner):
                 query=query,
                 message_id=message.id,
             )
-        except ModerationException as e:
+        except ModerationError as e:
             self.direct_output(
                 queue_manager=queue_manager,
                 app_generate_entity=application_generate_entity,

api/core/app/apps/message_based_app_generator.py

@@ -8,7 +8,7 @@ from sqlalchemy import and_
 
 from core.app.app_config.entities import EasyUIBasedAppModelConfigFrom
 from core.app.apps.base_app_generator import BaseAppGenerator
-from core.app.apps.base_app_queue_manager import AppQueueManager, GenerateTaskStoppedException
+from core.app.apps.base_app_queue_manager import AppQueueManager, GenerateTaskStoppedError
 from core.app.entities.app_invoke_entities import (
     AdvancedChatAppGenerateEntity,
     AgentChatAppGenerateEntity,
@@ -77,7 +77,7 @@ class MessageBasedAppGenerator(BaseAppGenerator):
             return generate_task_pipeline.process()
         except ValueError as e:
             if e.args[0] == "I/O operation on closed file.":  # ignore this error
-                raise GenerateTaskStoppedException()
+                raise GenerateTaskStoppedError()
             else:
                 logger.exception(e)
                 raise e
@@ -148,7 +148,7 @@ class MessageBasedAppGenerator(BaseAppGenerator):
         # get from source
         end_user_id = None
         account_id = None
-        if application_generate_entity.invoke_from in [InvokeFrom.WEB_APP, InvokeFrom.SERVICE_API]:
+        if application_generate_entity.invoke_from in {InvokeFrom.WEB_APP, InvokeFrom.SERVICE_API}:
             from_source = "api"
             end_user_id = application_generate_entity.user_id
         else:
@@ -165,11 +165,11 @@ class MessageBasedAppGenerator(BaseAppGenerator):
             model_provider = application_generate_entity.model_conf.provider
             model_id = application_generate_entity.model_conf.model
             override_model_configs = None
-            if app_config.app_model_config_from == EasyUIBasedAppModelConfigFrom.ARGS and app_config.app_mode in [
+            if app_config.app_model_config_from == EasyUIBasedAppModelConfigFrom.ARGS and app_config.app_mode in {
                 AppMode.AGENT_CHAT,
                 AppMode.CHAT,
                 AppMode.COMPLETION,
-            ]:
+            }:
                 override_model_configs = app_config.app_model_config_dict
 
         # get conversation introduction

api/core/app/apps/message_based_app_queue_manager.py

@@ -1,4 +1,4 @@
-from core.app.apps.base_app_queue_manager import AppQueueManager, GenerateTaskStoppedException, PublishFrom
+from core.app.apps.base_app_queue_manager import AppQueueManager, GenerateTaskStoppedError, PublishFrom
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.app.entities.queue_entities import (
     AppQueueEvent,
@@ -53,4 +53,4 @@ class MessageBasedAppQueueManager(AppQueueManager):
             self.stop_listen()
 
         if pub_from == PublishFrom.APPLICATION_MANAGER and self._is_stopped():
-            raise GenerateTaskStoppedException()
+            raise GenerateTaskStoppedError()

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

@@ -12,7 +12,7 @@ from pydantic import ValidationError
 import contexts
 from core.app.app_config.features.file_upload.manager import FileUploadConfigManager
 from core.app.apps.base_app_generator import BaseAppGenerator
-from core.app.apps.base_app_queue_manager import AppQueueManager, GenerateTaskStoppedException, PublishFrom
+from core.app.apps.base_app_queue_manager import AppQueueManager, GenerateTaskStoppedError, PublishFrom
 from core.app.apps.workflow.app_config_manager import WorkflowAppConfigManager
 from core.app.apps.workflow.app_queue_manager import WorkflowAppQueueManager
 from core.app.apps.workflow.app_runner import WorkflowAppRunner
@@ -253,7 +253,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
                 )
 
                 runner.run()
-            except GenerateTaskStoppedException:
+            except GenerateTaskStoppedError:
                 pass
             except InvokeAuthorizationError:
                 queue_manager.publish_error(
@@ -302,7 +302,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
             return generate_task_pipeline.process()
         except ValueError as e:
             if e.args[0] == "I/O operation on closed file.":  # ignore this error
-                raise GenerateTaskStoppedException()
+                raise GenerateTaskStoppedError()
             else:
                 logger.exception(e)
                 raise e

api/core/app/apps/workflow/app_queue_manager.py

@@ -1,4 +1,4 @@
-from core.app.apps.base_app_queue_manager import AppQueueManager, GenerateTaskStoppedException, PublishFrom
+from core.app.apps.base_app_queue_manager import AppQueueManager, GenerateTaskStoppedError, PublishFrom
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.app.entities.queue_entities import (
     AppQueueEvent,
@@ -39,4 +39,4 @@ class WorkflowAppQueueManager(AppQueueManager):
             self.stop_listen()
 
         if pub_from == PublishFrom.APPLICATION_MANAGER and self._is_stopped():
-            raise GenerateTaskStoppedException()
+            raise GenerateTaskStoppedError()

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

@@ -53,7 +53,7 @@ class WorkflowAppRunner(WorkflowBasedAppRunner):
         app_config = cast(WorkflowAppConfig, app_config)
 
         user_id = None
-        if self.application_generate_entity.invoke_from in [InvokeFrom.WEB_APP, InvokeFrom.SERVICE_API]:
+        if self.application_generate_entity.invoke_from in {InvokeFrom.WEB_APP, InvokeFrom.SERVICE_API}:
             end_user = db.session.query(EndUser).filter(EndUser.id == self.application_generate_entity.user_id).first()
             if end_user:
                 user_id = end_user.session_id
@@ -113,7 +113,7 @@ class WorkflowAppRunner(WorkflowBasedAppRunner):
             user_id=self.application_generate_entity.user_id,
             user_from=(
                 UserFrom.ACCOUNT
-                if self.application_generate_entity.invoke_from in [InvokeFrom.EXPLORE, InvokeFrom.DEBUGGER]
+                if self.application_generate_entity.invoke_from in {InvokeFrom.EXPLORE, InvokeFrom.DEBUGGER}
                 else UserFrom.END_USER
             ),
             invoke_from=self.application_generate_entity.invoke_from,

api/core/app/apps/workflow/generate_task_pipeline.py

@@ -162,10 +162,10 @@ class WorkflowAppGenerateTaskPipeline(BasedGenerateTaskPipeline, WorkflowCycleMa
 
             yield WorkflowAppStreamResponse(workflow_run_id=workflow_run_id, stream_response=stream_response)
 
-    def _listenAudioMsg(self, publisher, task_id: str):
+    def _listen_audio_msg(self, publisher, task_id: str):
         if not publisher:
             return None
-        audio_msg: AudioTrunk = publisher.checkAndGetAudio()
+        audio_msg: AudioTrunk = publisher.check_and_get_audio()
         if audio_msg and audio_msg.status != "finish":
             return MessageAudioStreamResponse(audio=audio_msg.audio, task_id=task_id)
         return None
@@ -187,7 +187,7 @@ class WorkflowAppGenerateTaskPipeline(BasedGenerateTaskPipeline, WorkflowCycleMa
 
         for response in self._process_stream_response(tts_publisher=tts_publisher, trace_manager=trace_manager):
             while True:
-                audio_response = self._listenAudioMsg(tts_publisher, task_id=task_id)
+                audio_response = self._listen_audio_msg(tts_publisher, task_id=task_id)
                 if audio_response:
                     yield audio_response
                 else:
@@ -199,7 +199,7 @@ class WorkflowAppGenerateTaskPipeline(BasedGenerateTaskPipeline, WorkflowCycleMa
             try:
                 if not tts_publisher:
                     break
-                audio_trunk = tts_publisher.checkAndGetAudio()
+                audio_trunk = tts_publisher.check_and_get_audio()
                 if audio_trunk is None:
                     # release cpu
                     # sleep 20 ms ( 40ms => 1280 byte audio file,20ms => 640 byte audio file)
@@ -376,7 +376,9 @@ class WorkflowAppGenerateTaskPipeline(BasedGenerateTaskPipeline, WorkflowCycleMa
                     tts_publisher.publish(message=queue_message)
 
                 self._task_state.answer += delta_text
-                yield self._text_chunk_to_stream_response(delta_text)
+                yield self._text_chunk_to_stream_response(
+                    delta_text, from_variable_selector=event.from_variable_selector
+                )
             else:
                 continue
 
@@ -412,14 +414,17 @@ class WorkflowAppGenerateTaskPipeline(BasedGenerateTaskPipeline, WorkflowCycleMa
         db.session.commit()
         db.session.close()
 
-    def _text_chunk_to_stream_response(self, text: str) -> TextChunkStreamResponse:
+    def _text_chunk_to_stream_response(
+        self, text: str, from_variable_selector: Optional[list[str]] = None
+    ) -> TextChunkStreamResponse:
         """
         Handle completed event.
         :param text: text
         :return:
         """
         response = TextChunkStreamResponse(
-            task_id=self._application_generate_entity.task_id, data=TextChunkStreamResponse.Data(text=text)
+            task_id=self._application_generate_entity.task_id,
+            data=TextChunkStreamResponse.Data(text=text, from_variable_selector=from_variable_selector),
         )
 
         return response

api/core/app/apps/workflow_logging_callback.py

@@ -84,10 +84,12 @@ class WorkflowLoggingCallback(WorkflowCallback):
         if route_node_state.node_run_result:
             node_run_result = route_node_state.node_run_result
             self.print_text(
-                f"Inputs: {jsonable_encoder(node_run_result.inputs) if node_run_result.inputs else ''}", color="green"
+                f"Inputs: {jsonable_encoder(node_run_result.inputs) if node_run_result.inputs else ''}",
+                color="green",
             )
             self.print_text(
-                f"Process Data: {jsonable_encoder(node_run_result.process_data) if node_run_result.process_data else ''}",
+                f"Process Data: "
+                f"{jsonable_encoder(node_run_result.process_data) if node_run_result.process_data else ''}",
                 color="green",
             )
             self.print_text(
@@ -114,14 +116,17 @@ class WorkflowLoggingCallback(WorkflowCallback):
             node_run_result = route_node_state.node_run_result
             self.print_text(f"Error: {node_run_result.error}", color="red")
             self.print_text(
-                f"Inputs: {jsonable_encoder(node_run_result.inputs) if node_run_result.inputs else ''}", color="red"
-            )
-            self.print_text(
-                f"Process Data: {jsonable_encoder(node_run_result.process_data) if node_run_result.process_data else ''}",
+                f"Inputs: {jsonable_encoder(node_run_result.inputs) if node_run_result.inputs else ''}",
                 color="red",
             )
             self.print_text(
-                f"Outputs: {jsonable_encoder(node_run_result.outputs) if node_run_result.outputs else ''}", color="red"
+                f"Process Data: "
+                f"{jsonable_encoder(node_run_result.process_data) if node_run_result.process_data else ''}",
+                color="red",
+            )
+            self.print_text(
+                f"Outputs: {jsonable_encoder(node_run_result.outputs) if node_run_result.outputs else ''}",
+                color="red",
             )
 
     def on_node_text_chunk(self, event: NodeRunStreamChunkEvent) -> None:

api/core/app/entities/task_entities.py

@@ -90,6 +90,7 @@ class MessageStreamResponse(StreamResponse):
     event: StreamEvent = StreamEvent.MESSAGE
     id: str
     answer: str
+    from_variable_selector: Optional[list[str]] = None
 
 
 class MessageAudioStreamResponse(StreamResponse):
@@ -479,6 +480,7 @@ class TextChunkStreamResponse(StreamResponse):
         """
 
         text: str
+        from_variable_selector: Optional[list[str]] = None
 
     event: StreamEvent = StreamEvent.TEXT_CHUNK
     data: Data

api/core/app/features/annotation_reply/annotation_reply.py

@@ -63,7 +63,7 @@ class AnnotationReplyFeature:
                 score = documents[0].metadata["score"]
                 annotation = AppAnnotationService.get_annotation_by_id(annotation_id)
                 if annotation:
-                    if invoke_from in [InvokeFrom.SERVICE_API, InvokeFrom.WEB_APP]:
+                    if invoke_from in {InvokeFrom.SERVICE_API, InvokeFrom.WEB_APP}:
                         from_source = "api"
                     else:
                         from_source = "console"

api/core/app/segments/segments.py

@@ -15,6 +15,7 @@ class Segment(BaseModel):
     value: Any
 
     @field_validator("value_type")
+    @classmethod
     def validate_value_type(cls, value):
         """
         This validator checks if the provided value is equal to the default value of the 'value_type' field.

api/core/app/task_pipeline/based_generate_task_pipeline.py

@@ -65,7 +65,7 @@ class BasedGenerateTaskPipeline:
 
         if isinstance(e, InvokeAuthorizationError):
             err = InvokeAuthorizationError("Incorrect API key provided")
-        elif isinstance(e, InvokeError) or isinstance(e, ValueError):
+        elif isinstance(e, InvokeError | ValueError):
             err = e
         else:
             err = Exception(e.description if getattr(e, "description", None) is not None else str(e))

api/core/app/task_pipeline/easy_ui_based_generate_task_pipeline.py

@@ -201,10 +201,10 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline, MessageCycleMan
                     stream_response=stream_response,
                 )
 
-    def _listenAudioMsg(self, publisher, task_id: str):
+    def _listen_audio_msg(self, publisher, task_id: str):
         if publisher is None:
             return None
-        audio_msg: AudioTrunk = publisher.checkAndGetAudio()
+        audio_msg: AudioTrunk = publisher.check_and_get_audio()
         if audio_msg and audio_msg.status != "finish":
             # audio_str = audio_msg.audio.decode('utf-8', errors='ignore')
             return MessageAudioStreamResponse(audio=audio_msg.audio, task_id=task_id)
@@ -225,7 +225,7 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline, MessageCycleMan
             publisher = AppGeneratorTTSPublisher(tenant_id, text_to_speech_dict.get("voice", None))
         for response in self._process_stream_response(publisher=publisher, trace_manager=trace_manager):
             while True:
-                audio_response = self._listenAudioMsg(publisher, task_id)
+                audio_response = self._listen_audio_msg(publisher, task_id)
                 if audio_response:
                     yield audio_response
                 else:
@@ -237,7 +237,7 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline, MessageCycleMan
         while (time.time() - start_listener_time) < TTS_AUTO_PLAY_TIMEOUT:
             if publisher is None:
                 break
-            audio = publisher.checkAndGetAudio()
+            audio = publisher.check_and_get_audio()
             if audio is None:
                 # release cpu
                 # sleep 20 ms ( 40ms => 1280 byte audio file,20ms => 640 byte audio file)
@@ -372,7 +372,7 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline, MessageCycleMan
             self._message,
             application_generate_entity=self._application_generate_entity,
             conversation=self._conversation,
-            is_first_message=self._application_generate_entity.app_config.app_mode in [AppMode.AGENT_CHAT, AppMode.CHAT]
+            is_first_message=self._application_generate_entity.app_config.app_mode in {AppMode.AGENT_CHAT, AppMode.CHAT}
             and self._application_generate_entity.conversation_id is None,
             extras=self._application_generate_entity.extras,
         )

api/core/app/task_pipeline/message_cycle_manage.py

@@ -153,14 +153,21 @@ class MessageCycleManage:
 
         return None
 
-    def _message_to_stream_response(self, answer: str, message_id: str) -> MessageStreamResponse:
+    def _message_to_stream_response(
+        self, answer: str, message_id: str, from_variable_selector: Optional[list[str]] = None
+    ) -> MessageStreamResponse:
         """
         Message to stream response.
         :param answer: answer
         :param message_id: message id
         :return:
         """
-        return MessageStreamResponse(task_id=self._application_generate_entity.task_id, id=message_id, answer=answer)
+        return MessageStreamResponse(
+            task_id=self._application_generate_entity.task_id,
+            id=message_id,
+            answer=answer,
+            from_variable_selector=from_variable_selector,
+        )
 
     def _message_replace_to_stream_response(self, answer: str) -> MessageReplaceStreamResponse:
         """

api/core/app/task_pipeline/workflow_cycle_manage.py

@@ -383,7 +383,7 @@ class WorkflowCycleManage:
         :param workflow_node_execution: workflow node execution
         :return:
         """
-        if workflow_node_execution.node_type in [NodeType.ITERATION.value, NodeType.LOOP.value]:
+        if workflow_node_execution.node_type in {NodeType.ITERATION.value, NodeType.LOOP.value}:
             return None
 
         response = NodeStartStreamResponse(
@@ -430,7 +430,7 @@ class WorkflowCycleManage:
         :param workflow_node_execution: workflow node execution
         :return:
         """
-        if workflow_node_execution.node_type in [NodeType.ITERATION.value, NodeType.LOOP.value]:
+        if workflow_node_execution.node_type in {NodeType.ITERATION.value, NodeType.LOOP.value}:
             return None
 
         return NodeFinishStreamResponse(

api/core/callback_handler/index_tool_callback_handler.py

@@ -29,7 +29,7 @@ class DatasetIndexToolCallbackHandler:
             source="app",
             source_app_id=self._app_id,
             created_by_role=(
-                "account" if self._invoke_from in [InvokeFrom.EXPLORE, InvokeFrom.DEBUGGER] else "end_user"
+                "account" if self._invoke_from in {InvokeFrom.EXPLORE, InvokeFrom.DEBUGGER} else "end_user"
             ),
             created_by=self._user_id,
         )
@@ -67,7 +67,7 @@ class DatasetIndexToolCallbackHandler:
                     data_source_type=item.get("data_source_type"),
                     segment_id=item.get("segment_id"),
                     score=item.get("score") if "score" in item else None,
-                    hit_count=item.get("hit_count") if "hit_count" else None,
+                    hit_count=item.get("hit_count") if "hit_count" in item else None,
                     word_count=item.get("word_count") if "word_count" in item else None,
                     segment_position=item.get("segment_position") if "segment_position" in item else None,
                     index_node_hash=item.get("index_node_hash") if "index_node_hash" in item else None,

api/core/embedding/cached_embedding.py

@@ -65,7 +65,7 @@ class CacheEmbedding(Embeddings):
                         except IntegrityError:
                             db.session.rollback()
                         except Exception as e:
-                            logging.exception("Failed transform embedding: ", e)
+                            logging.exception("Failed transform embedding: %s", e)
                 cache_embeddings = []
                 try:
                     for i, embedding in zip(embedding_queue_indices, embedding_queue_embeddings):
@@ -85,7 +85,7 @@ class CacheEmbedding(Embeddings):
                     db.session.rollback()
             except Exception as ex:
                 db.session.rollback()
-                logger.error("Failed to embed documents: ", ex)
+                logger.error("Failed to embed documents: %s", ex)
                 raise ex
 
         return text_embeddings
@@ -116,10 +116,7 @@ class CacheEmbedding(Embeddings):
             # Transform to string
             encoded_str = encoded_vector.decode("utf-8")
             redis_client.setex(embedding_cache_key, 600, encoded_str)
-
-        except IntegrityError:
-            db.session.rollback()
-        except:
-            logging.exception("Failed to add embedding to redis")
+        except Exception as ex:
+            logging.exception("Failed to add embedding to redis %s", ex)
 
         return embedding_results

api/core/extension/extensible.py

@@ -3,6 +3,7 @@ import importlib.util
 import json
 import logging
 import os
+from pathlib import Path
 from typing import Any, Optional
 
 from pydantic import BaseModel
@@ -63,8 +64,7 @@ class Extensible:
 
                     builtin_file_path = os.path.join(subdir_path, "__builtin__")
                     if os.path.exists(builtin_file_path):
-                        with open(builtin_file_path, encoding="utf-8") as f:
-                            position = int(f.read().strip())
+                        position = int(Path(builtin_file_path).read_text(encoding="utf-8").strip())
                     position_map[extension_name] = position
 
                 if (extension_name + ".py") not in file_names:

api/core/file/message_file_parser.py

@@ -188,7 +188,8 @@ class MessageFileParser:
     def _check_image_remote_url(self, url):
         try:
             headers = {
-                "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"
+                "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko)"
+                " Chrome/91.0.4472.124 Safari/537.36"
             }
 
             def is_s3_presigned_url(url):

api/core/helper/code_executor/code_executor.py

@@ -16,7 +16,7 @@ from core.helper.code_executor.template_transformer import TemplateTransformer
 logger = logging.getLogger(__name__)
 
 
-class CodeExecutionException(Exception):
+class CodeExecutionError(Exception):
     pass
 
 
@@ -86,15 +86,16 @@ class CodeExecutor:
                 ),
             )
             if response.status_code == 503:
-                raise CodeExecutionException("Code execution service is unavailable")
+                raise CodeExecutionError("Code execution service is unavailable")
             elif response.status_code != 200:
                 raise Exception(
-                    f"Failed to execute code, got status code {response.status_code}, please check if the sandbox service is running"
+                    f"Failed to execute code, got status code {response.status_code},"
+                    f" please check if the sandbox service is running"
                 )
-        except CodeExecutionException as e:
+        except CodeExecutionError as e:
             raise e
         except Exception as e:
-            raise CodeExecutionException(
+            raise CodeExecutionError(
                 "Failed to execute code, which is likely a network issue,"
                 " please check if the sandbox service is running."
                 f" ( Error: {str(e)} )"
@@ -103,15 +104,15 @@ class CodeExecutor:
         try:
             response = response.json()
         except:
-            raise CodeExecutionException("Failed to parse response")
+            raise CodeExecutionError("Failed to parse response")
 
         if (code := response.get("code")) != 0:
-            raise CodeExecutionException(f"Got error code: {code}. Got error msg: {response.get('message')}")
+            raise CodeExecutionError(f"Got error code: {code}. Got error msg: {response.get('message')}")
 
         response = CodeExecutionResponse(**response)
 
         if response.data.error:
-            raise CodeExecutionException(response.data.error)
+            raise CodeExecutionError(response.data.error)
 
         return response.data.stdout or ""
 
@@ -126,13 +127,13 @@ class CodeExecutor:
         """
         template_transformer = cls.code_template_transformers.get(language)
         if not template_transformer:
-            raise CodeExecutionException(f"Unsupported language {language}")
+            raise CodeExecutionError(f"Unsupported language {language}")
 
         runner, preload = template_transformer.transform_caller(code, inputs)
 
         try:
             response = cls.execute_code(language, preload, runner)
-        except CodeExecutionException as e:
+        except CodeExecutionError as e:
             raise e
 
         return template_transformer.transform_response(response)

api/core/helper/tool_parameter_cache.py

@@ -14,7 +14,10 @@ class ToolParameterCache:
     def __init__(
         self, tenant_id: str, provider: str, tool_name: str, cache_type: ToolParameterCacheType, identity_id: str
     ):
-        self.cache_key = f"{cache_type.value}_secret:tenant_id:{tenant_id}:provider:{provider}:tool_name:{tool_name}:identity_id:{identity_id}"
+        self.cache_key = (
+            f"{cache_type.value}_secret:tenant_id:{tenant_id}:provider:{provider}:tool_name:{tool_name}"
+            f":identity_id:{identity_id}"
+        )
 
     def get(self) -> Optional[dict]:
         """

api/core/indexing_runner.py

@@ -78,8 +78,8 @@ class IndexingRunner:
                     dataset_document=dataset_document,
                     documents=documents,
                 )
-            except DocumentIsPausedException:
-                raise DocumentIsPausedException("Document paused, document id: {}".format(dataset_document.id))
+            except DocumentIsPausedError:
+                raise DocumentIsPausedError("Document paused, document id: {}".format(dataset_document.id))
             except ProviderTokenNotInitError as e:
                 dataset_document.indexing_status = "error"
                 dataset_document.error = str(e.description)
@@ -134,8 +134,8 @@ class IndexingRunner:
             self._load(
                 index_processor=index_processor, dataset=dataset, dataset_document=dataset_document, documents=documents
             )
-        except DocumentIsPausedException:
-            raise DocumentIsPausedException("Document paused, document id: {}".format(dataset_document.id))
+        except DocumentIsPausedError:
+            raise DocumentIsPausedError("Document paused, document id: {}".format(dataset_document.id))
         except ProviderTokenNotInitError as e:
             dataset_document.indexing_status = "error"
             dataset_document.error = str(e.description)
@@ -192,8 +192,8 @@ class IndexingRunner:
             self._load(
                 index_processor=index_processor, dataset=dataset, dataset_document=dataset_document, documents=documents
             )
-        except DocumentIsPausedException:
-            raise DocumentIsPausedException("Document paused, document id: {}".format(dataset_document.id))
+        except DocumentIsPausedError:
+            raise DocumentIsPausedError("Document paused, document id: {}".format(dataset_document.id))
         except ProviderTokenNotInitError as e:
             dataset_document.indexing_status = "error"
             dataset_document.error = str(e.description)
@@ -292,7 +292,7 @@ class IndexingRunner:
         self, index_processor: BaseIndexProcessor, dataset_document: DatasetDocument, process_rule: dict
     ) -> list[Document]:
         # load file
-        if dataset_document.data_source_type not in ["upload_file", "notion_import", "website_crawl"]:
+        if dataset_document.data_source_type not in {"upload_file", "notion_import", "website_crawl"}:
             return []
 
         data_source_info = dataset_document.data_source_info_dict
@@ -756,7 +756,7 @@ class IndexingRunner:
         indexing_cache_key = "document_{}_is_paused".format(document_id)
         result = redis_client.get(indexing_cache_key)
         if result:
-            raise DocumentIsPausedException()
+            raise DocumentIsPausedError()
 
     @staticmethod
     def _update_document_index_status(
@@ -767,10 +767,10 @@ class IndexingRunner:
         """
         count = DatasetDocument.query.filter_by(id=document_id, is_paused=True).count()
         if count > 0:
-            raise DocumentIsPausedException()
+            raise DocumentIsPausedError()
         document = DatasetDocument.query.filter_by(id=document_id).first()
         if not document:
-            raise DocumentIsDeletedPausedException()
+            raise DocumentIsDeletedPausedError()
 
         update_params = {DatasetDocument.indexing_status: after_indexing_status}
 
@@ -875,9 +875,9 @@ class IndexingRunner:
         pass
 
 
-class DocumentIsPausedException(Exception):
+class DocumentIsPausedError(Exception):
     pass
 
 
-class DocumentIsDeletedPausedException(Exception):
+class DocumentIsDeletedPausedError(Exception):
     pass

api/core/llm_generator/output_parser/errors.py

@@ -1,2 +1,2 @@
-class OutputParserException(Exception):
+class OutputParserError(Exception):
     pass

api/core/llm_generator/output_parser/rule_config_generator.py

@@ -1,6 +1,6 @@
 from typing import Any
 
-from core.llm_generator.output_parser.errors import OutputParserException
+from core.llm_generator.output_parser.errors import OutputParserError
 from core.llm_generator.prompts import (
     RULE_CONFIG_PARAMETER_GENERATE_TEMPLATE,
     RULE_CONFIG_PROMPT_GENERATE_TEMPLATE,
@@ -29,4 +29,4 @@ class RuleConfigGeneratorOutputParser:
                 raise ValueError("Expected 'opening_statement' to be a str.")
             return parsed
         except Exception as e:
-            raise OutputParserException(f"Parsing text\n{text}\n of rule config generator raised following error:\n{e}")
+            raise OutputParserError(f"Parsing text\n{text}\n of rule config generator raised following error:\n{e}")

api/core/llm_generator/prompts.py

@@ -59,24 +59,27 @@ User Input: yo, 你今天咋样？
 }
 
 User Input: 
-"""
+"""  # noqa: E501
 
 SUGGESTED_QUESTIONS_AFTER_ANSWER_INSTRUCTION_PROMPT = (
     "Please help me predict the three most likely questions that human would ask, "
     "and keeping each question under 20 characters.\n"
-    "MAKE SURE your output is the SAME language as the Assistant's latest response(if the main response is written in Chinese, then the language of your output must be using Chinese.)!\n"
+    "MAKE SURE your output is the SAME language as the Assistant's latest response"
+    "(if the main response is written in Chinese, then the language of your output must be using Chinese.)!\n"
     "The output must be an array in JSON format following the specified schema:\n"
     '["question1","question2","question3"]\n'
 )
 
 GENERATOR_QA_PROMPT = (
-    "<Task> The user will send a long text. Generate a Question and Answer pairs only using the knowledge in the long text. Please think step by step."
+    "<Task> The user will send a long text. Generate a Question and Answer pairs only using the knowledge"
+    " in the long text. Please think step by step."
     "Step 1: Understand and summarize the main content of this text.\n"
     "Step 2: What key information or concepts are mentioned in this text?\n"
     "Step 3: Decompose or combine multiple pieces of information and concepts.\n"
     "Step 4: Generate questions and answers based on these key information and concepts.\n"
     "<Constraints> The questions should be clear and detailed, and the answers should be detailed and complete. "
-    "You must answer in {language}, in a style that is clear and detailed in {language}. No language other than {language} should be used. \n"
+    "You must answer in {language}, in a style that is clear and detailed in {language}."
+    " No language other than {language} should be used. \n"
     "<Format> Use the following format: Q1:\nA1:\nQ2:\nA2:...\n"
     "<QA Pairs>"
 )
@@ -94,7 +97,7 @@ Based on task description, please create a well-structured prompt template that
 - Use the same language as task description. 
 - Output in ``` xml ``` and start with <instruction>
 Please generate the full prompt template with at least 300 words and output only the prompt template.
-"""
+"""  # noqa: E501
 
 RULE_CONFIG_PROMPT_GENERATE_TEMPLATE = """
 Here is a task description for which I would like you to create a high-quality prompt template for:
@@ -109,7 +112,7 @@ Based on task description, please create a well-structured prompt template that
 - Use the same language as task description. 
 - Output in ``` xml ``` and start with <instruction>
 Please generate the full prompt template and output only the prompt template.
-"""
+"""  # noqa: E501
 
 RULE_CONFIG_PARAMETER_GENERATE_TEMPLATE = """
 I need to extract the following information from the input text. The <information to be extracted> tag specifies the 'type', 'description' and 'required' of the information to be extracted. 
@@ -134,7 +137,7 @@ Inside <text></text> XML tags, there is a text that I should extract parameters
 
 ### Answer
 I should always output a valid list. Output nothing other than the list of variable_name. Output an empty list if there is no variable name in input text.
-"""
+"""  # noqa: E501
 
 RULE_CONFIG_STATEMENT_GENERATE_TEMPLATE = """
 <instruction>
@@ -150,4 +153,4 @@ Welcome! I'm here to assist you with any questions or issues you might have with
 Here is the task description: {{INPUT_TEXT}}
 
 You just need to generate the output
-"""
+"""  # noqa: E501

api/core/memory/token_buffer_memory.py

@@ -39,7 +39,7 @@ class TokenBufferMemory:
         )
 
         if message_limit and message_limit > 0:
-            message_limit = message_limit if message_limit <= 500 else 500
+            message_limit = min(message_limit, 500)
         else:
             message_limit = 500
 
@@ -52,7 +52,7 @@ class TokenBufferMemory:
             files = db.session.query(MessageFile).filter(MessageFile.message_id == message.id).all()
             if files:
                 file_extra_config = None
-                if self.conversation.mode not in [AppMode.ADVANCED_CHAT.value, AppMode.WORKFLOW.value]:
+                if self.conversation.mode not in {AppMode.ADVANCED_CHAT.value, AppMode.WORKFLOW.value}:
                     file_extra_config = FileUploadConfigManager.convert(self.conversation.model_config)
                 else:
                     if message.workflow_run_id:

api/core/model_runtime/entities/defaults.py

@@ -8,8 +8,11 @@ PARAMETER_RULE_TEMPLATE: dict[DefaultParameterName, dict] = {
         },
         "type": "float",
         "help": {
-            "en_US": "Controls randomness. Lower temperature results in less random completions. As the temperature approaches zero, the model will become deterministic and repetitive. Higher temperature results in more random completions.",
-            "zh_Hans": "温度控制随机性。较低的温度会导致较少的随机完成。随着温度接近零，模型将变得确定性和重复性。较高的温度会导致更多的随机完成。",
+            "en_US": "Controls randomness. Lower temperature results in less random completions."
+            " As the temperature approaches zero, the model will become deterministic and repetitive."
+            " Higher temperature results in more random completions.",
+            "zh_Hans": "温度控制随机性。较低的温度会导致较少的随机完成。随着温度接近零，模型将变得确定性和重复性。"
+            "较高的温度会导致更多的随机完成。",
         },
         "required": False,
         "default": 0.0,
@@ -24,7 +27,8 @@ PARAMETER_RULE_TEMPLATE: dict[DefaultParameterName, dict] = {
         },
         "type": "float",
         "help": {
-            "en_US": "Controls diversity via nucleus sampling: 0.5 means half of all likelihood-weighted options are considered.",
+            "en_US": "Controls diversity via nucleus sampling: 0.5 means half of all likelihood-weighted options"
+            " are considered.",
             "zh_Hans": "通过核心采样控制多样性：0.5表示考虑了一半的所有可能性加权选项。",
         },
         "required": False,
@@ -88,7 +92,8 @@ PARAMETER_RULE_TEMPLATE: dict[DefaultParameterName, dict] = {
         },
         "type": "int",
         "help": {
-            "en_US": "Specifies the upper limit on the length of generated results. If the generated results are truncated, you can increase this parameter.",
+            "en_US": "Specifies the upper limit on the length of generated results."
+            " If the generated results are truncated, you can increase this parameter.",
             "zh_Hans": "指定生成结果长度的上限。如果生成结果截断，可以调大该参数。",
         },
         "required": False,
@@ -104,7 +109,8 @@ PARAMETER_RULE_TEMPLATE: dict[DefaultParameterName, dict] = {
         },
         "type": "string",
         "help": {
-            "en_US": "Set a response format, ensure the output from llm is a valid code block as possible, such as JSON, XML, etc.",
+            "en_US": "Set a response format, ensure the output from llm is a valid code block as possible,"
+            " such as JSON, XML, etc.",
             "zh_Hans": "设置一个返回格式，确保llm的输出尽可能是有效的代码块，如JSON、XML等",
         },
         "required": False,

api/core/model_runtime/entities/model_entities.py

@@ -27,17 +27,17 @@ class ModelType(Enum):
 
         :return: model type
         """
-        if origin_model_type == "text-generation" or origin_model_type == cls.LLM.value:
+        if origin_model_type in {"text-generation", cls.LLM.value}:
             return cls.LLM
-        elif origin_model_type == "embeddings" or origin_model_type == cls.TEXT_EMBEDDING.value:
+        elif origin_model_type in {"embeddings", cls.TEXT_EMBEDDING.value}:
             return cls.TEXT_EMBEDDING
-        elif origin_model_type == "reranking" or origin_model_type == cls.RERANK.value:
+        elif origin_model_type in {"reranking", cls.RERANK.value}:
             return cls.RERANK
-        elif origin_model_type == "speech2text" or origin_model_type == cls.SPEECH2TEXT.value:
+        elif origin_model_type in {"speech2text", cls.SPEECH2TEXT.value}:
             return cls.SPEECH2TEXT
-        elif origin_model_type == "tts" or origin_model_type == cls.TTS.value:
+        elif origin_model_type in {"tts", cls.TTS.value}:
             return cls.TTS
-        elif origin_model_type == "text2img" or origin_model_type == cls.TEXT2IMG.value:
+        elif origin_model_type in {"text2img", cls.TEXT2IMG.value}:
             return cls.TEXT2IMG
         elif origin_model_type == cls.MODERATION.value:
             return cls.MODERATION

api/core/model_runtime/model_providers/__base/ai_model.py

@@ -72,7 +72,9 @@ class AIModel(ABC):
             if isinstance(error, tuple(model_errors)):
                 if invoke_error == InvokeAuthorizationError:
                     return invoke_error(
-                        description=f"[{provider_name}] Incorrect model credentials provided, please check and try again. "
+                        description=(
+                            f"[{provider_name}] Incorrect model credentials provided, please check and try again."
+                        )
                     )
 
                 return invoke_error(description=f"[{provider_name}] {invoke_error.description}, {str(error)}")
@@ -198,7 +200,7 @@ class AIModel(ABC):
             except Exception as e:
                 model_schema_yaml_file_name = os.path.basename(model_schema_yaml_path).rstrip(".yaml")
                 raise Exception(
-                    f"Invalid model schema for {provider_name}.{model_type}.{model_schema_yaml_file_name}:" f" {str(e)}"
+                    f"Invalid model schema for {provider_name}.{model_type}.{model_schema_yaml_file_name}: {str(e)}"
                 )
 
             # cache model schema

api/core/model_runtime/model_providers/__base/large_language_model.py

@@ -187,7 +187,7 @@ if you are not sure about the structure.
 <instructions>
 {{instructions}}
 </instructions>
-"""
+"""  # noqa: E501
 
         code_block = model_parameters.get("response_format", "")
         if not code_block:
@@ -449,7 +449,7 @@ if you are not sure about the structure.
                 model=real_model,
                 prompt_messages=prompt_messages,
                 message=prompt_message,
-                usage=usage if usage else LLMUsage.empty_usage(),
+                usage=usage or LLMUsage.empty_usage(),
                 system_fingerprint=system_fingerprint,
             ),
             credentials=credentials,
@@ -830,7 +830,8 @@ if you are not sure about the structure.
                     else:
                         if parameter_value != round(parameter_value, parameter_rule.precision):
                             raise ValueError(
-                                f"Model Parameter {parameter_name} should be round to {parameter_rule.precision} decimal places."
+                                f"Model Parameter {parameter_name} should be round to {parameter_rule.precision}"
+                                f" decimal places."
                             )
 
                 # validate parameter value range