feat: add independent memory

Signed-off-by: -LAN- <laipz8200@outlook.com>

.devcontainer/post_create_command.sh

@@ -2,10 +2,10 @@
 
 npm add -g pnpm@10.8.0
 cd web && pnpm install
-pipx install poetry
+pipx install uv
 
-echo 'alias start-api="cd /workspaces/dify/api && poetry run python -m flask run --host 0.0.0.0 --port=5001 --debug"' >> ~/.bashrc
-echo 'alias start-worker="cd /workspaces/dify/api && poetry run python -m celery -A app.celery worker -P gevent -c 1 --loglevel INFO -Q dataset,generation,mail,ops_trace,app_deletion"' >> ~/.bashrc
+echo 'alias start-api="cd /workspaces/dify/api && uv run python -m flask run --host 0.0.0.0 --port=5001 --debug"' >> ~/.bashrc
+echo 'alias start-worker="cd /workspaces/dify/api && uv run python -m celery -A app.celery worker -P gevent -c 1 --loglevel INFO -Q dataset,generation,mail,ops_trace,app_deletion"' >> ~/.bashrc
 echo 'alias start-web="cd /workspaces/dify/web && pnpm dev"' >> ~/.bashrc
 echo 'alias start-containers="cd /workspaces/dify/docker && docker-compose -f docker-compose.middleware.yaml -p dify --env-file middleware.env up -d"' >> ~/.bashrc
 echo 'alias stop-containers="cd /workspaces/dify/docker && docker-compose -f docker-compose.middleware.yaml -p dify --env-file middleware.env down"' >> ~/.bashrc

.devcontainer/post_start_command.sh

@@ -1,3 +1,3 @@
 #!/bin/bash
 
-cd api && poetry install
+cd api && uv sync

.github/actions/setup-poetry/action.yml

@@ -1,36 +0,0 @@
-name: Setup Poetry and Python
-
-inputs:
-  python-version:
-    description: Python version to use and the Poetry installed with
-    required: true
-    default: '3.11'
-  poetry-version:
-    description: Poetry version to set up
-    required: true
-    default: '2.0.1'
-  poetry-lockfile:
-    description: Path to the Poetry lockfile to restore cache from
-    required: true
-    default: ''
-
-runs:
-  using: composite
-  steps:
-    - name: Set up Python ${{ inputs.python-version }}
-      uses: actions/setup-python@v5
-      with:
-        python-version: ${{ inputs.python-version }}
-        cache: pip
-
-    - name: Install Poetry
-      shell: bash
-      run: pip install poetry==${{ inputs.poetry-version }}
-
-    - name: Restore Poetry cache
-      if: ${{ inputs.poetry-lockfile != '' }}
-      uses: actions/setup-python@v5
-      with:
-        python-version: ${{ inputs.python-version }}
-        cache: poetry
-        cache-dependency-path: ${{ inputs.poetry-lockfile }}

.github/actions/setup-uv/action.yml

@@ -0,0 +1,34 @@
+name: Setup UV and Python
+
+inputs:
+  python-version:
+    description: Python version to use and the UV installed with
+    required: true
+    default: '3.12'
+  uv-version:
+    description: UV version to set up
+    required: true
+    default: '0.6.14'
+  uv-lockfile:
+    description: Path to the UV lockfile to restore cache from
+    required: true
+    default: ''
+  enable-cache:
+    required: true
+    default: true
+
+runs:
+  using: composite
+  steps:
+    - name: Set up Python ${{ inputs.python-version }}
+      uses: actions/setup-python@v5
+      with:
+        python-version: ${{ inputs.python-version }}
+
+    - name: Install uv
+      uses: astral-sh/setup-uv@v5
+      with:
+        version: ${{ inputs.uv-version }}
+        python-version: ${{ inputs.python-version }}
+        enable-cache: ${{ inputs.enable-cache }}
+        cache-dependency-glob: ${{ inputs.uv-lockfile }}

.github/workflows/api-tests.yml

@@ -17,6 +17,9 @@ jobs:
   test:
     name: API Tests
     runs-on: ubuntu-latest
+    defaults:
+      run:
+        shell: bash
     strategy:
       matrix:
         python-version:
@@ -27,40 +30,44 @@ jobs:
       - name: Checkout code
         uses: actions/checkout@v4
         with:
-          fetch-depth: 0
           persist-credentials: false
 
-      - name: Setup Poetry and Python ${{ matrix.python-version }}
-        uses: ./.github/actions/setup-poetry
+      - name: Setup UV and Python
+        uses: ./.github/actions/setup-uv
         with:
           python-version: ${{ matrix.python-version }}
-          poetry-lockfile: api/poetry.lock
+          uv-lockfile: api/uv.lock
 
-      - name: Check Poetry lockfile
-        run: |
-          poetry check -C api --lock
-          poetry show -C api
+      - name: Check UV lockfile
+        run: uv lock --project api --check
 
       - name: Install dependencies
-        run: poetry install -C api --with dev
-
-      - name: Check dependencies in pyproject.toml
-        run: poetry run -P api bash dev/pytest/pytest_artifacts.sh
+        run: uv sync --project api --dev
 
       - name: Run Unit tests
-        run: poetry run -P api bash dev/pytest/pytest_unit_tests.sh
+        run: |
+          uv run --project api bash dev/pytest/pytest_unit_tests.sh
+          # Extract coverage percentage and create a summary
+          TOTAL_COVERAGE=$(python -c 'import json; print(json.load(open("coverage.json"))["totals"]["percent_covered_display"])')
+
+          # Create a detailed coverage summary
+          echo "### Test Coverage Summary :test_tube:" >> $GITHUB_STEP_SUMMARY
+          echo "Total Coverage: ${TOTAL_COVERAGE}%" >> $GITHUB_STEP_SUMMARY
+          echo "\`\`\`" >> $GITHUB_STEP_SUMMARY
+          uv run --project api coverage report >> $GITHUB_STEP_SUMMARY
+          echo "\`\`\`" >> $GITHUB_STEP_SUMMARY
 
       - name: Run dify config tests
-        run: poetry run -P api python dev/pytest/pytest_config_tests.py
+        run: uv run --project api dev/pytest/pytest_config_tests.py
 
-      - name: Cache MyPy
+      - name: MyPy Cache
         uses: actions/cache@v4
         with:
           path: api/.mypy_cache
-          key: mypy-${{ matrix.python-version }}-${{ runner.os }}-${{ hashFiles('api/poetry.lock') }}
+          key: mypy-${{ matrix.python-version }}-${{ runner.os }}-${{ hashFiles('api/uv.lock') }}
 
-      - name: Run mypy
-        run: dev/run-mypy
+      - name: Run MyPy Checks
+        run: dev/mypy-check
 
       - name: Set up dotenvs
         run: |
@@ -80,4 +87,4 @@ jobs:
             ssrf_proxy
 
       - name: Run Workflow
-        run: poetry run -P api bash dev/pytest/pytest_workflow.sh
+        run: uv run --project api bash dev/pytest/pytest_workflow.sh

.github/workflows/db-migration-test.yml

@@ -24,13 +24,13 @@ jobs:
           fetch-depth: 0
           persist-credentials: false
 
-      - name: Setup Poetry and Python
-        uses: ./.github/actions/setup-poetry
+      - name: Setup UV and Python
+        uses: ./.github/actions/setup-uv
         with:
-          poetry-lockfile: api/poetry.lock
+          uv-lockfile: api/uv.lock
 
       - name: Install dependencies
-        run: poetry install -C api
+        run: uv sync --project api
 
       - name: Prepare middleware env
         run: |
@@ -54,6 +54,4 @@ jobs:
       - name: Run DB Migration
         env:
           DEBUG: true
-        run: |
-          cd api
-          poetry run python -m flask upgrade-db
+        run: uv run --directory api flask upgrade-db

.github/workflows/docker-build.yml

@@ -42,6 +42,7 @@ jobs:
         with:
           push: false
           context: "{{defaultContext}}:${{ matrix.context }}"
+          file: "${{ matrix.file }}"
           platforms: ${{ matrix.platform }}
           cache-from: type=gha
           cache-to: type=gha,mode=max

.github/workflows/style.yml

@@ -18,7 +18,6 @@ jobs:
       - name: Checkout code
         uses: actions/checkout@v4
         with:
-          fetch-depth: 0
           persist-credentials: false
 
       - name: Check changed files
@@ -29,24 +28,27 @@ jobs:
             api/**
             .github/workflows/style.yml
 
-      - name: Setup Poetry and Python
+      - name: Setup UV and Python
         if: steps.changed-files.outputs.any_changed == 'true'
-        uses: ./.github/actions/setup-poetry
+        uses: ./.github/actions/setup-uv
+        with:
+          uv-lockfile: api/uv.lock
+          enable-cache: false
 
       - name: Install dependencies
         if: steps.changed-files.outputs.any_changed == 'true'
-        run: poetry install -C api --only lint
+        run: uv sync --project api --dev
 
       - name: Ruff check
         if: steps.changed-files.outputs.any_changed == 'true'
         run: |
-          poetry run -C api ruff --version
-          poetry run -C api ruff check ./
-          poetry run -C api ruff format --check ./
+          uv run --directory api ruff --version
+          uv run --directory api ruff check ./
+          uv run --directory api ruff format --check ./
 
       - name: Dotenv check
         if: steps.changed-files.outputs.any_changed == 'true'
-        run: poetry run -P api dotenv-linter ./api/.env.example ./web/.env.example
+        run: uv run --project api dotenv-linter ./api/.env.example ./web/.env.example
 
       - name: Lint hints
         if: failure()
@@ -63,7 +65,6 @@ jobs:
       - name: Checkout code
         uses: actions/checkout@v4
         with:
-          fetch-depth: 0
           persist-credentials: false
 
       - name: Check changed files
@@ -102,7 +103,6 @@ jobs:
       - name: Checkout code
         uses: actions/checkout@v4
         with:
-          fetch-depth: 0
           persist-credentials: false
 
       - name: Check changed files
@@ -133,7 +133,6 @@ jobs:
       - name: Checkout code
         uses: actions/checkout@v4
         with:
-          fetch-depth: 0
           persist-credentials: false
 
       - name: Check changed files

.github/workflows/tool-test-sdks.yaml

@@ -27,7 +27,6 @@ jobs:
     steps:
       - uses: actions/checkout@v4
         with:
-          fetch-depth: 0
           persist-credentials: false
 
       - name: Use Node.js ${{ matrix.node-version }}

.github/workflows/vdb-tests.yml

@@ -8,7 +8,7 @@ on:
       - api/core/rag/datasource/**
       - docker/**
       - .github/workflows/vdb-tests.yml
-      - api/poetry.lock
+      - api/uv.lock
       - api/pyproject.toml
 
 concurrency:
@@ -29,22 +29,19 @@ jobs:
       - name: Checkout code
         uses: actions/checkout@v4
         with:
-          fetch-depth: 0
           persist-credentials: false
 
-      - name: Setup Poetry and Python ${{ matrix.python-version }}
-        uses: ./.github/actions/setup-poetry
+      - name: Setup UV and Python
+        uses: ./.github/actions/setup-uv
         with:
           python-version: ${{ matrix.python-version }}
-          poetry-lockfile: api/poetry.lock
+          uv-lockfile: api/uv.lock
 
-      - name: Check Poetry lockfile
-        run: |
-          poetry check -C api --lock
-          poetry show -C api
+      - name: Check UV lockfile
+        run: uv lock --project api --check
 
       - name: Install dependencies
-        run: poetry install -C api --with dev
+        run: uv sync --project api --dev
 
       - name: Set up dotenvs
         run: |
@@ -80,7 +77,7 @@ jobs:
             elasticsearch
 
       - name: Check TiDB Ready
-        run: poetry run -P api python api/tests/integration_tests/vdb/tidb_vector/check_tiflash_ready.py
+        run: uv run --project api python api/tests/integration_tests/vdb/tidb_vector/check_tiflash_ready.py
 
       - name: Test Vector Stores
-        run: poetry run -P api bash dev/pytest/pytest_vdb.sh
+        run: uv run --project api bash dev/pytest/pytest_vdb.sh

.github/workflows/web-tests.yml

@@ -23,7 +23,6 @@ jobs:
       - name: Checkout code
         uses: actions/checkout@v4
         with:
-          fetch-depth: 0
           persist-credentials: false
 
       - name: Check changed files

.gitignore

@@ -46,6 +46,7 @@ htmlcov/
 .cache
 nosetests.xml
 coverage.xml
+coverage.json
 *.cover
 *.py,cover
 .hypothesis/

CONTRIBUTING_CN.md

@@ -6,7 +6,7 @@
 
 本指南和 Dify 一样在不断完善中。如果有任何滞后于项目实际情况的地方，恳请谅解，我们也欢迎任何改进建议。
 
-关于许可证，请花一分钟阅读我们简短的[许可和贡献者协议](./LICENSE)。社区同时也遵循[行为准则](https://github.com/langgenius/.github/blob/main/CODE_OF_CONDUCT.md)。
+关于许可证，请花一分钟阅读我们简短的[许可和贡献者协议](./LICENSE)。同时也请遵循社区[行为准则](https://github.com/langgenius/.github/blob/main/CODE_OF_CONDUCT.md)。
 
 ## 开始之前
 

README.md

@@ -8,7 +8,7 @@
   <a href="https://cloud.dify.ai">Dify Cloud</a> ·
   <a href="https://docs.dify.ai/getting-started/install-self-hosted">Self-hosting</a> ·
   <a href="https://docs.dify.ai">Documentation</a> ·
-  <a href="https://udify.app/chat/22L1zSxg6yW1cWQg">Enterprise inquiry</a>
+  <a href="https://dify.ai/pricing">Dify edition overview</a>
 </p>
 
 <p align="center">

README_AR.md

@@ -4,7 +4,7 @@
   <a href="https://cloud.dify.ai">Dify Cloud</a> ·
   <a href="https://docs.dify.ai/getting-started/install-self-hosted">الاستضافة الذاتية</a> ·
   <a href="https://docs.dify.ai">التوثيق</a> ·
-  <a href="https://udify.app/chat/22L1zSxg6yW1cWQg">استفسار الشركات (للإنجليزية فقط)</a>
+  <a href="https://dify.ai/pricing">نظرة عامة على منتجات Dify</a>
 </p>
 
 <p align="center">

README_BN.md

@@ -8,7 +8,7 @@
   <a href="https://cloud.dify.ai">ডিফাই ক্লাউড</a> ·
   <a href="https://docs.dify.ai/getting-started/install-self-hosted">সেল্ফ-হোস্টিং</a> ·
   <a href="https://docs.dify.ai">ডকুমেন্টেশন</a> ·
-  <a href="https://udify.app/chat/22L1zSxg6yW1cWQg">ব্যাবসায়িক অনুসন্ধান</a>
+  <a href="https://dify.ai/pricing">Dify পণ্যের রূপভেদ</a>
 </p>
 
 <p align="center">

README_CN.md

@@ -4,7 +4,7 @@
   <a href="https://cloud.dify.ai">Dify 云服务</a> ·
   <a href="https://docs.dify.ai/getting-started/install-self-hosted">自托管</a> ·
   <a href="https://docs.dify.ai">文档</a> ·
-  <a href="https://udify.app/chat/22L1zSxg6yW1cWQg">（需用英文）常见问题解答 / 联系团队</a>
+  <a href="https://dify.ai/pricing">Dify 产品形态总览</a>
 </div>
 
 <p align="center">

README_DE.md

@@ -8,7 +8,7 @@
   <a href="https://cloud.dify.ai">Dify Cloud</a> ·
   <a href="https://docs.dify.ai/getting-started/install-self-hosted">Selbstgehostetes</a> ·
   <a href="https://docs.dify.ai">Dokumentation</a> ·
-  <a href="https://udify.app/chat/22L1zSxg6yW1cWQg">Anfrage an Unternehmen</a>
+  <a href="https://dify.ai/pricing">Überblick über die Dify-Produkte</a>
 </p>
 
 <p align="center">

README_ES.md

@@ -4,7 +4,7 @@
   <a href="https://cloud.dify.ai">Dify Cloud</a> ·
   <a href="https://docs.dify.ai/getting-started/install-self-hosted">Auto-alojamiento</a> ·
   <a href="https://docs.dify.ai">Documentación</a> ·
-  <a href="https://udify.app/chat/22L1zSxg6yW1cWQg">Consultas empresariales (en inglés)</a>
+  <a href="https://dify.ai/pricing">Resumen de las ediciones de Dify</a>
 </p>
 
 <p align="center">

README_FR.md

@@ -4,7 +4,7 @@
   <a href="https://cloud.dify.ai">Dify Cloud</a> ·
   <a href="https://docs.dify.ai/getting-started/install-self-hosted">Auto-hébergement</a> ·
   <a href="https://docs.dify.ai">Documentation</a> ·
-  <a href="https://udify.app/chat/22L1zSxg6yW1cWQg">Demande d’entreprise (en anglais seulement)</a>
+  <a href="https://dify.ai/pricing">Présentation des différentes offres Dify</a>
 </p>
 
 <p align="center">

README_JA.md

@@ -4,7 +4,7 @@
   <a href="https://cloud.dify.ai">Dify Cloud</a> ·
   <a href="https://docs.dify.ai/getting-started/install-self-hosted">セルフホスティング</a> ·
   <a href="https://docs.dify.ai">ドキュメント</a> ·
-  <a href="https://udify.app/chat/22L1zSxg6yW1cWQg">企業のお問い合わせ（英語のみ）</a>
+  <a href="https://dify.ai/pricing">Difyの各種エディションについて</a>
 </p>
 
 <p align="center">

README_KL.md

@@ -4,7 +4,7 @@
   <a href="https://cloud.dify.ai">Dify Cloud</a> ·
   <a href="https://docs.dify.ai/getting-started/install-self-hosted">Self-hosting</a> ·
   <a href="https://docs.dify.ai">Documentation</a> ·
-  <a href="https://udify.app/chat/22L1zSxg6yW1cWQg">Commercial enquiries</a>
+  <a href="https://dify.ai/pricing">Dify product editions</a>
 </p>
 
 <p align="center">

README_KR.md

@@ -4,7 +4,7 @@
   <a href="https://cloud.dify.ai">Dify 클라우드</a> ·
   <a href="https://docs.dify.ai/getting-started/install-self-hosted">셀프-호스팅</a> ·
   <a href="https://docs.dify.ai">문서</a> ·
-  <a href="https://udify.app/chat/22L1zSxg6yW1cWQg">기업 문의 (영어만 가능)</a>
+  <a href="https://dify.ai/pricing">Dify 제품 에디션 안내</a>
 </p>
 
 <p align="center">

README_PT.md

@@ -8,7 +8,7 @@
   <a href="https://cloud.dify.ai">Dify Cloud</a> ·
   <a href="https://docs.dify.ai/getting-started/install-self-hosted">Auto-hospedagem</a> ·
   <a href="https://docs.dify.ai">Documentação</a> ·
-  <a href="https://udify.app/chat/22L1zSxg6yW1cWQg">Consultas empresariais</a>
+  <a href="https://dify.ai/pricing">Visão geral das edições do Dify</a>
 </p>
 
 <p align="center">

README_SI.md

@@ -8,7 +8,7 @@
   <a href="https://cloud.dify.ai">Dify Cloud</a> ·
   <a href="https://docs.dify.ai/getting-started/install-self-hosted">Samostojno gostovanje</a> ·
   <a href="https://docs.dify.ai">Dokumentacija</a> ·
-  <a href="https://udify.app/chat/22L1zSxg6yW1cWQg">Povpraševanje za podjetja</a>
+  <a href="https://dify.ai/pricing">Pregled ponudb izdelkov Dify</a>
 </p>
 
 <p align="center">

README_TR.md

@@ -4,7 +4,7 @@
   <a href="https://cloud.dify.ai">Dify Bulut</a> ·
   <a href="https://docs.dify.ai/getting-started/install-self-hosted">Kendi Sunucunuzda Barındırma</a> ·
   <a href="https://docs.dify.ai">Dokümantasyon</a> ·
-  <a href="https://udify.app/chat/22L1zSxg6yW1cWQg">Yalnızca İngilizce: Kurumsal Sorgulama</a>
+  <a href="https://dify.ai/pricing">Dify ürün seçeneklerine genel bakış</a>
 </p>
 
 <p align="center">

README_TW.md

@@ -8,7 +8,7 @@
   <a href="https://cloud.dify.ai">Dify 雲端服務</a> ·
   <a href="https://docs.dify.ai/getting-started/install-self-hosted">自行託管</a> ·
   <a href="https://docs.dify.ai">說明文件</a> ·
-  <a href="https://udify.app/chat/22L1zSxg6yW1cWQg">企業諮詢</a>
+  <a href="https://dify.ai/pricing">產品方案概覽</a>
 </p>
 
 <p align="center">

README_VI.md

@@ -4,7 +4,7 @@
   <a href="https://cloud.dify.ai">Dify Cloud</a> ·
   <a href="https://docs.dify.ai/getting-started/install-self-hosted">Tự triển khai</a> ·
   <a href="https://docs.dify.ai">Tài liệu</a> ·
-  <a href="https://udify.app/chat/22L1zSxg6yW1cWQg">Yêu cầu doanh nghiệp</a>
+  <a href="https://dify.ai/pricing">Tổng quan các lựa chọn sản phẩm Dify</a>
 </p>
 
 <p align="center">

api/.env.example

@@ -165,6 +165,7 @@ MILVUS_URI=http://127.0.0.1:19530
 MILVUS_TOKEN=
 MILVUS_USER=root
 MILVUS_PASSWORD=Milvus
+MILVUS_ANALYZER_PARAMS=
 
 # MyScale configuration
 MYSCALE_HOST=127.0.0.1
@@ -326,6 +327,7 @@ UPLOAD_AUDIO_FILE_SIZE_LIMIT=50
 MULTIMODAL_SEND_FORMAT=base64
 PROMPT_GENERATION_MAX_TOKENS=512
 CODE_GENERATION_MAX_TOKENS=1024
+PLUGIN_BASED_TOKEN_COUNTING_ENABLED=false
 
 # Mail configuration, support: resend, smtp
 MAIL_TYPE=
@@ -422,6 +424,12 @@ WORKFLOW_CALL_MAX_DEPTH=5
 WORKFLOW_PARALLEL_DEPTH_LIMIT=3
 MAX_VARIABLE_SIZE=204800
 
+# Workflow storage configuration
+# Options: rdbms, hybrid
+# rdbms: Use only the relational database (default)
+# hybrid: Save new data to object storage, read from both object storage and RDBMS
+WORKFLOW_NODE_EXECUTION_STORAGE=rdbms
+
 # App configuration
 APP_MAX_EXECUTION_TIME=1200
 APP_MAX_ACTIVE_REQUESTS=0
@@ -462,3 +470,19 @@ CREATE_TIDB_SERVICE_JOB_ENABLED=false
 MAX_SUBMIT_COUNT=100
 # Lockout duration in seconds
 LOGIN_LOCKOUT_DURATION=86400
+
+# Enable OpenTelemetry
+ENABLE_OTEL=false
+OTLP_BASE_ENDPOINT=http://localhost:4318
+OTLP_API_KEY=
+OTEL_EXPORTER_TYPE=otlp
+OTEL_SAMPLING_RATE=0.1
+OTEL_BATCH_EXPORT_SCHEDULE_DELAY=5000
+OTEL_MAX_QUEUE_SIZE=2048
+OTEL_MAX_EXPORT_BATCH_SIZE=512
+OTEL_METRIC_EXPORT_INTERVAL=60000
+OTEL_BATCH_EXPORT_TIMEOUT=10000
+OTEL_METRIC_EXPORT_TIMEOUT=30000
+
+# Prevent Clickjacking
+ALLOW_EMBED=false

api/Dockerfile

@@ -3,20 +3,11 @@ FROM python:3.12-slim-bookworm AS base
 
 WORKDIR /app/api
 
-# Install Poetry
-ENV POETRY_VERSION=2.0.1
+# Install uv
+ENV UV_VERSION=0.6.14
 
-# if you located in China, you can use aliyun mirror to speed up
-# RUN pip install --no-cache-dir poetry==${POETRY_VERSION} -i https://mirrors.aliyun.com/pypi/simple/
+RUN pip install --no-cache-dir uv==${UV_VERSION}
 
-RUN pip install --no-cache-dir poetry==${POETRY_VERSION}
-
-# Configure Poetry
-ENV POETRY_CACHE_DIR=/tmp/poetry_cache
-ENV POETRY_NO_INTERACTION=1
-ENV POETRY_VIRTUALENVS_IN_PROJECT=true
-ENV POETRY_VIRTUALENVS_CREATE=true
-ENV POETRY_REQUESTS_TIMEOUT=15
 
 FROM base AS packages
 
@@ -27,8 +18,8 @@ RUN apt-get update \
     && apt-get install -y --no-install-recommends gcc g++ libc-dev libffi-dev libgmp-dev libmpfr-dev libmpc-dev
 
 # Install Python dependencies
-COPY pyproject.toml poetry.lock ./
-RUN poetry install --sync --no-cache --no-root
+COPY pyproject.toml uv.lock ./
+RUN uv sync --locked
 
 # production stage
 FROM base AS production

api/README.md

@@ -3,7 +3,10 @@
 ## Usage
 
 > [!IMPORTANT]
-> In the v0.6.12 release, we deprecated `pip` as the package management tool for Dify API Backend service and replaced it with `poetry`.
+> 
+> In the v1.3.0 release, `poetry` has been replaced with
+> [`uv`](https://docs.astral.sh/uv/) as the package manager
+> for Dify API backend service.
 
 1. Start the docker-compose stack
 
@@ -37,19 +40,19 @@
 
 4. Create environment.
 
-   Dify API service uses [Poetry](https://python-poetry.org/docs/) to manage dependencies. First, you need to add the poetry shell plugin, if you don't have it already, in order to run in a virtual environment. [Note: Poetry shell is no longer a native command so you need to install the poetry plugin beforehand]
+   Dify API service uses [UV](https://docs.astral.sh/uv/) to manage dependencies.
+   First, you need to add the uv package manager, if you don't have it already.
 
    ```bash
-   poetry self add poetry-plugin-shell
+   pip install uv
+   # Or on macOS
+   brew install uv
    ```
-   
-   Then, You can execute `poetry shell` to activate the environment.
 
 5. Install dependencies
 
    ```bash
-   poetry env use 3.12
-   poetry install
+   uv sync --dev
    ```
 
 6. Run migrate
@@ -57,21 +60,21 @@
    Before the first launch, migrate the database to the latest version.
 
    ```bash
-   poetry run python -m flask db upgrade
+   uv run flask db upgrade
    ```
 
 7. Start backend
 
    ```bash
-   poetry run python -m flask run --host 0.0.0.0 --port=5001 --debug
+   uv run flask run --host 0.0.0.0 --port=5001 --debug
    ```
 
 8. Start Dify [web](../web) service.
-9. Setup your application by visiting `http://localhost:3000`...
+9. Setup your application by visiting `http://localhost:3000`.
 10. If you need to handle and debug the async tasks (e.g. dataset importing and documents indexing), please start the worker service.
 
    ```bash
-   poetry run python -m celery -A app.celery worker -P gevent -c 1 --loglevel INFO -Q dataset,generation,mail,ops_trace,app_deletion
+   uv run celery -A app.celery worker -P gevent -c 1 --loglevel INFO -Q dataset,generation,mail,ops_trace,app_deletion
    ```
 
 ## Testing
@@ -79,11 +82,11 @@
 1. Install dependencies for both the backend and the test environment
 
    ```bash
-   poetry install -C api --with dev
+   uv sync --dev
    ```
 
 2. Run the tests locally with mocked system environment variables in `tool.pytest_env` section in `pyproject.toml`
 
    ```bash
-   poetry run -P api bash dev/pytest/pytest_all_tests.sh
+   uv run -P api bash dev/pytest/pytest_all_tests.sh
    ```

api/app_factory.py

@@ -51,8 +51,11 @@ def initialize_extensions(app: DifyApp):
         ext_login,
         ext_mail,
         ext_migrate,
+        ext_otel,
+        ext_otel_patch,
         ext_proxy_fix,
         ext_redis,
+        ext_repositories,
         ext_sentry,
         ext_set_secretkey,
         ext_storage,
@@ -73,6 +76,7 @@ def initialize_extensions(app: DifyApp):
         ext_migrate,
         ext_redis,
         ext_storage,
+        ext_repositories,
         ext_celery,
         ext_login,
         ext_mail,
@@ -81,6 +85,8 @@ def initialize_extensions(app: DifyApp):
         ext_proxy_fix,
         ext_blueprints,
         ext_commands,
+        ext_otel_patch,  # Apply patch before initializing OpenTelemetry
+        ext_otel,
     ]
     for ext in extensions:
         short_name = ext.__name__.split(".")[-1]

api/configs/app_config.py

@@ -9,9 +9,11 @@ from .enterprise import EnterpriseFeatureConfig
 from .extra import ExtraServiceConfig
 from .feature import FeatureConfig
 from .middleware import MiddlewareConfig
+from .observability import ObservabilityConfig
 from .packaging import PackagingInfo
 from .remote_settings_sources import RemoteSettingsSource, RemoteSettingsSourceConfig, RemoteSettingsSourceName
 from .remote_settings_sources.apollo import ApolloSettingsSource
+from .remote_settings_sources.nacos import NacosSettingsSource
 
 logger = logging.getLogger(__name__)
 
@@ -33,6 +35,8 @@ class RemoteSettingsSourceFactory(PydanticBaseSettingsSource):
         match remote_source_name:
             case RemoteSettingsSourceName.APOLLO:
                 remote_source = ApolloSettingsSource(current_state)
+            case RemoteSettingsSourceName.NACOS:
+                remote_source = NacosSettingsSource(current_state)
             case _:
                 logger.warning(f"Unsupported remote source: {remote_source_name}")
                 return {}
@@ -59,6 +63,8 @@ class DifyConfig(
     MiddlewareConfig,
     # Extra service configs
     ExtraServiceConfig,
+    # Observability configs
+    ObservabilityConfig,
     # Remote source configs
     RemoteSettingsSourceConfig,
     # Enterprise feature configs

api/configs/feature/__init__.py

@@ -12,7 +12,7 @@ from pydantic import (
 )
 from pydantic_settings import BaseSettings
 
-from configs.feature.hosted_service import HostedServiceConfig
+from .hosted_service import HostedServiceConfig
 
 
 class SecurityConfig(BaseSettings):
@@ -442,7 +442,7 @@ class LoggingConfig(BaseSettings):
 
 class ModelLoadBalanceConfig(BaseSettings):
     """
-    Configuration for model load balancing
+    Configuration for model load balancing and token counting
     """
 
     MODEL_LB_ENABLED: bool = Field(
@@ -450,6 +450,11 @@ class ModelLoadBalanceConfig(BaseSettings):
         default=False,
     )
 
+    PLUGIN_BASED_TOKEN_COUNTING_ENABLED: bool = Field(
+        description="Enable or disable plugin based token counting. If disabled, token counting will return 0.",
+        default=False,
+    )
+
 
 class BillingConfig(BaseSettings):
     """
@@ -514,6 +519,11 @@ class WorkflowNodeExecutionConfig(BaseSettings):
         default=100,
     )
 
+    WORKFLOW_NODE_EXECUTION_STORAGE: str = Field(
+        default="rdbms",
+        description="Storage backend for WorkflowNodeExecution. Options: 'rdbms', 'hybrid'",
+    )
+
 
 class AuthConfig(BaseSettings):
     """

api/configs/middleware/__init__.py

@@ -22,6 +22,7 @@ from .vdb.baidu_vector_config import BaiduVectorDBConfig
 from .vdb.chroma_config import ChromaConfig
 from .vdb.couchbase_config import CouchbaseConfig
 from .vdb.elasticsearch_config import ElasticsearchConfig
+from .vdb.huawei_cloud_config import HuaweiCloudConfig
 from .vdb.lindorm_config import LindormConfig
 from .vdb.milvus_config import MilvusConfig
 from .vdb.myscale_config import MyScaleConfig
@@ -263,6 +264,7 @@ class MiddlewareConfig(
     VectorStoreConfig,
     AnalyticdbConfig,
     ChromaConfig,
+    HuaweiCloudConfig,
     MilvusConfig,
     MyScaleConfig,
     OpenSearchConfig,

api/configs/middleware/vdb/huawei_cloud_config.py

@@ -0,0 +1,25 @@
+from typing import Optional
+
+from pydantic import Field
+from pydantic_settings import BaseSettings
+
+
+class HuaweiCloudConfig(BaseSettings):
+    """
+    Configuration settings for Huawei cloud search service
+    """
+
+    HUAWEI_CLOUD_HOSTS: Optional[str] = Field(
+        description="Hostname or IP address of the Huawei cloud search service instance",
+        default=None,
+    )
+
+    HUAWEI_CLOUD_USER: Optional[str] = Field(
+        description="Username for authenticating with Huawei cloud search service",
+        default=None,
+    )
+
+    HUAWEI_CLOUD_PASSWORD: Optional[str] = Field(
+        description="Password for authenticating with Huawei cloud search service",
+        default=None,
+    )

api/configs/middleware/vdb/milvus_config.py

@@ -39,3 +39,8 @@ class MilvusConfig(BaseSettings):
         "older versions",
         default=True,
     )
+
+    MILVUS_ANALYZER_PARAMS: Optional[str] = Field(
+        description='Milvus text analyzer parameters, e.g., {"type": "chinese"} for Chinese segmentation support.',
+        default=None,
+    )

api/configs/observability/__init__.py

@@ -0,0 +1,9 @@
+from configs.observability.otel.otel_config import OTelConfig
+
+
+class ObservabilityConfig(OTelConfig):
+    """
+    Observability configuration settings
+    """
+
+    pass

api/configs/observability/otel/otel_config.py

@@ -0,0 +1,44 @@
+from pydantic import Field
+from pydantic_settings import BaseSettings
+
+
+class OTelConfig(BaseSettings):
+    """
+    OpenTelemetry configuration settings
+    """
+
+    ENABLE_OTEL: bool = Field(
+        description="Whether to enable OpenTelemetry",
+        default=False,
+    )
+
+    OTLP_BASE_ENDPOINT: str = Field(
+        description="OTLP base endpoint",
+        default="http://localhost:4318",
+    )
+
+    OTLP_API_KEY: str = Field(
+        description="OTLP API key",
+        default="",
+    )
+
+    OTEL_EXPORTER_TYPE: str = Field(
+        description="OTEL exporter type",
+        default="otlp",
+    )
+
+    OTEL_SAMPLING_RATE: float = Field(default=0.1, description="Sampling rate for traces (0.0 to 1.0)")
+
+    OTEL_BATCH_EXPORT_SCHEDULE_DELAY: int = Field(
+        default=5000, description="Batch export schedule delay in milliseconds"
+    )
+
+    OTEL_MAX_QUEUE_SIZE: int = Field(default=2048, description="Maximum queue size for the batch span processor")
+
+    OTEL_MAX_EXPORT_BATCH_SIZE: int = Field(default=512, description="Maximum export batch size")
+
+    OTEL_METRIC_EXPORT_INTERVAL: int = Field(default=60000, description="Metric export interval in milliseconds")
+
+    OTEL_BATCH_EXPORT_TIMEOUT: int = Field(default=10000, description="Batch export timeout in milliseconds")
+
+    OTEL_METRIC_EXPORT_TIMEOUT: int = Field(default=30000, description="Metric export timeout in milliseconds")

api/configs/packaging/__init__.py

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

api/configs/remote_settings_sources/apollo/client.py

@@ -270,7 +270,7 @@ class ApolloClient:
         while not self._stopping:
             for namespace in self._notification_map:
                 self._do_heart_beat(namespace)
-            time.sleep(60 * 10)  # 10分钟
+            time.sleep(60 * 10)  # 10 minutes
 
     def _do_heart_beat(self, namespace):
         url = "{}/configs/{}/{}/{}?ip={}".format(self.config_url, self.app_id, self.cluster, namespace, self.ip)

api/configs/remote_settings_sources/enums.py

@@ -3,3 +3,4 @@ from enum import StrEnum
 
 class RemoteSettingsSourceName(StrEnum):
     APOLLO = "apollo"
+    NACOS = "nacos"

api/configs/remote_settings_sources/nacos/__init__.py

@@ -0,0 +1,52 @@
+import logging
+import os
+from collections.abc import Mapping
+from typing import Any
+
+from pydantic.fields import FieldInfo
+
+from .http_request import NacosHttpClient
+
+logger = logging.getLogger(__name__)
+
+from configs.remote_settings_sources.base import RemoteSettingsSource
+
+from .utils import _parse_config
+
+
+class NacosSettingsSource(RemoteSettingsSource):
+    def __init__(self, configs: Mapping[str, Any]):
+        self.configs = configs
+        self.remote_configs: dict[str, Any] = {}
+        self.async_init()
+
+    def async_init(self):
+        data_id = os.getenv("DIFY_ENV_NACOS_DATA_ID", "dify-api-env.properties")
+        group = os.getenv("DIFY_ENV_NACOS_GROUP", "nacos-dify")
+        tenant = os.getenv("DIFY_ENV_NACOS_NAMESPACE", "")
+
+        params = {"dataId": data_id, "group": group, "tenant": tenant}
+        try:
+            content = NacosHttpClient().http_request("/nacos/v1/cs/configs", method="GET", headers={}, params=params)
+            self.remote_configs = self._parse_config(content)
+        except Exception as e:
+            logger.exception("[get-access-token] exception occurred")
+            raise
+
+    def _parse_config(self, content: str) -> dict:
+        if not content:
+            return {}
+        try:
+            return _parse_config(self, content)
+        except Exception as e:
+            raise RuntimeError(f"Failed to parse config: {e}")
+
+    def get_field_value(self, field: FieldInfo, field_name: str) -> tuple[Any, str, bool]:
+        if not isinstance(self.remote_configs, dict):
+            raise ValueError(f"remote configs is not dict, but {type(self.remote_configs)}")
+
+        field_value = self.remote_configs.get(field_name)
+        if field_value is None:
+            return None, field_name, False
+
+        return field_value, field_name, False

api/configs/remote_settings_sources/nacos/http_request.py

@@ -0,0 +1,83 @@
+import base64
+import hashlib
+import hmac
+import logging
+import os
+import time
+
+import requests
+
+logger = logging.getLogger(__name__)
+
+
+class NacosHttpClient:
+    def __init__(self):
+        self.username = os.getenv("DIFY_ENV_NACOS_USERNAME")
+        self.password = os.getenv("DIFY_ENV_NACOS_PASSWORD")
+        self.ak = os.getenv("DIFY_ENV_NACOS_ACCESS_KEY")
+        self.sk = os.getenv("DIFY_ENV_NACOS_SECRET_KEY")
+        self.server = os.getenv("DIFY_ENV_NACOS_SERVER_ADDR", "localhost:8848")
+        self.token = None
+        self.token_ttl = 18000
+        self.token_expire_time: float = 0
+
+    def http_request(self, url, method="GET", headers=None, params=None):
+        try:
+            self._inject_auth_info(headers, params)
+            response = requests.request(method, url="http://" + self.server + url, headers=headers, params=params)
+            response.raise_for_status()
+            return response.text
+        except requests.exceptions.RequestException as e:
+            return f"Request to Nacos failed: {e}"
+
+    def _inject_auth_info(self, headers, params, module="config"):
+        headers.update({"User-Agent": "Nacos-Http-Client-In-Dify:v0.0.1"})
+
+        if module == "login":
+            return
+
+        ts = str(int(time.time() * 1000))
+
+        if self.ak and self.sk:
+            sign_str = self.get_sign_str(params["group"], params["tenant"], ts)
+            headers["Spas-AccessKey"] = self.ak
+            headers["Spas-Signature"] = self.__do_sign(sign_str, self.sk)
+            headers["timeStamp"] = ts
+        if self.username and self.password:
+            self.get_access_token(force_refresh=False)
+            params["accessToken"] = self.token
+
+    def __do_sign(self, sign_str, sk):
+        return (
+            base64.encodebytes(hmac.new(sk.encode(), sign_str.encode(), digestmod=hashlib.sha1).digest())
+            .decode()
+            .strip()
+        )
+
+    def get_sign_str(self, group, tenant, ts):
+        sign_str = ""
+        if tenant:
+            sign_str = tenant + "+"
+        if group:
+            sign_str = sign_str + group + "+"
+        if sign_str:
+            sign_str += ts
+        return sign_str
+
+    def get_access_token(self, force_refresh=False):
+        current_time = time.time()
+        if self.token and not force_refresh and self.token_expire_time > current_time:
+            return self.token
+
+        params = {"username": self.username, "password": self.password}
+        url = "http://" + self.server + "/nacos/v1/auth/login"
+        try:
+            resp = requests.request("POST", url, headers=None, params=params)
+            resp.raise_for_status()
+            response_data = resp.json()
+            self.token = response_data.get("accessToken")
+            self.token_ttl = response_data.get("tokenTtl", 18000)
+            self.token_expire_time = current_time + self.token_ttl - 10
+        except Exception as e:
+            logger.exception("[get-access-token] exception occur")
+            raise

api/configs/remote_settings_sources/nacos/utils.py

@@ -0,0 +1,31 @@
+def _parse_config(self, content: str) -> dict[str, str]:
+    config: dict[str, str] = {}
+    if not content:
+        return config
+
+    for line in content.splitlines():
+        cleaned_line = line.strip()
+        if not cleaned_line or cleaned_line.startswith(("#", "!")):
+            continue
+
+        separator_index = -1
+        for i, c in enumerate(cleaned_line):
+            if c in ("=", ":") and (i == 0 or cleaned_line[i - 1] != "\\"):
+                separator_index = i
+                break
+
+        if separator_index == -1:
+            continue
+
+        key = cleaned_line[:separator_index].strip()
+        raw_value = cleaned_line[separator_index + 1 :].strip()
+
+        try:
+            decoded_value = bytes(raw_value, "utf-8").decode("unicode_escape")
+            decoded_value = decoded_value.replace(r"\=", "=").replace(r"\:", ":")
+        except UnicodeDecodeError:
+            decoded_value = raw_value
+
+        config[key] = decoded_value
+
+    return config

api/constants/__init__.py

@@ -3,6 +3,8 @@ from configs import dify_config
 HIDDEN_VALUE = "[__HIDDEN__]"
 UUID_NIL = "00000000-0000-0000-0000-000000000000"
 
+DEFAULT_FILE_NUMBER_LIMITS = 3
+
 IMAGE_EXTENSIONS = ["jpg", "jpeg", "png", "webp", "gif", "svg"]
 IMAGE_EXTENSIONS.extend([ext.upper() for ext in IMAGE_EXTENSIONS])
 

api/controllers/common/helpers.py

@@ -4,8 +4,6 @@ import platform
 import re
 import urllib.parse
 import warnings
-from collections.abc import Mapping
-from typing import Any
 from uuid import uuid4
 
 import httpx
@@ -29,8 +27,6 @@ except ImportError:
 
 from pydantic import BaseModel
 
-from configs import dify_config
-
 
 class FileInfo(BaseModel):
     filename: str
@@ -87,38 +83,3 @@ def guess_file_info_from_response(response: httpx.Response):
         mimetype=mimetype,
         size=int(response.headers.get("Content-Length", -1)),
     )
-
-
-def get_parameters_from_feature_dict(*, features_dict: Mapping[str, Any], user_input_form: list[dict[str, Any]]):
-    return {
-        "opening_statement": features_dict.get("opening_statement"),
-        "suggested_questions": features_dict.get("suggested_questions", []),
-        "suggested_questions_after_answer": features_dict.get("suggested_questions_after_answer", {"enabled": False}),
-        "speech_to_text": features_dict.get("speech_to_text", {"enabled": False}),
-        "text_to_speech": features_dict.get("text_to_speech", {"enabled": False}),
-        "retriever_resource": features_dict.get("retriever_resource", {"enabled": False}),
-        "annotation_reply": features_dict.get("annotation_reply", {"enabled": False}),
-        "more_like_this": features_dict.get("more_like_this", {"enabled": False}),
-        "user_input_form": user_input_form,
-        "sensitive_word_avoidance": features_dict.get(
-            "sensitive_word_avoidance", {"enabled": False, "type": "", "configs": []}
-        ),
-        "file_upload": features_dict.get(
-            "file_upload",
-            {
-                "image": {
-                    "enabled": False,
-                    "number_limits": 3,
-                    "detail": "high",
-                    "transfer_methods": ["remote_url", "local_file"],
-                }
-            },
-        ),
-        "system_parameters": {
-            "image_file_size_limit": dify_config.UPLOAD_IMAGE_FILE_SIZE_LIMIT,
-            "video_file_size_limit": dify_config.UPLOAD_VIDEO_FILE_SIZE_LIMIT,
-            "audio_file_size_limit": dify_config.UPLOAD_AUDIO_FILE_SIZE_LIMIT,
-            "file_size_limit": dify_config.UPLOAD_FILE_SIZE_LIMIT,
-            "workflow_file_upload_limit": dify_config.WORKFLOW_FILE_UPLOAD_LIMIT,
-        },
-    }

api/controllers/console/app/annotation.py

@@ -89,7 +89,7 @@ class AnnotationReplyActionStatusApi(Resource):
         app_annotation_job_key = "{}_app_annotation_job_{}".format(action, str(job_id))
         cache_result = redis_client.get(app_annotation_job_key)
         if cache_result is None:
-            raise ValueError("The job is not exist.")
+            raise ValueError("The job does not exist.")
 
         job_status = cache_result.decode()
         error_msg = ""
@@ -226,7 +226,7 @@ class AnnotationBatchImportStatusApi(Resource):
         indexing_cache_key = "app_annotation_batch_import_{}".format(str(job_id))
         cache_result = redis_client.get(indexing_cache_key)
         if cache_result is None:
-            raise ValueError("The job is not exist.")
+            raise ValueError("The job does not exist.")
         job_status = cache_result.decode()
         error_msg = ""
         if job_status == "error":

api/controllers/console/app/audio.py

@@ -80,8 +80,6 @@ class ChatMessageTextApi(Resource):
     @account_initialization_required
     @get_app_model
     def post(self, app_model: App):
-        from werkzeug.exceptions import InternalServerError
-
         try:
             parser = reqparse.RequestParser()
             parser.add_argument("message_id", type=str, location="json")

api/controllers/console/app/generator.py

@@ -85,5 +85,35 @@ class RuleCodeGenerateApi(Resource):
         return code_result
 
 
+class RuleStructuredOutputGenerateApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self):
+        parser = reqparse.RequestParser()
+        parser.add_argument("instruction", type=str, required=True, nullable=False, location="json")
+        parser.add_argument("model_config", type=dict, required=True, nullable=False, location="json")
+        args = parser.parse_args()
+
+        account = current_user
+        try:
+            structured_output = LLMGenerator.generate_structured_output(
+                tenant_id=account.current_tenant_id,
+                instruction=args["instruction"],
+                model_config=args["model_config"],
+            )
+        except ProviderTokenNotInitError as ex:
+            raise ProviderNotInitializeError(ex.description)
+        except QuotaExceededError:
+            raise ProviderQuotaExceededError()
+        except ModelCurrentlyNotSupportError:
+            raise ProviderModelCurrentlyNotSupportError()
+        except InvokeError as e:
+            raise CompletionRequestError(e.description)
+
+        return structured_output
+
+
 api.add_resource(RuleGenerateApi, "/rule-generate")
 api.add_resource(RuleCodeGenerateApi, "/rule-code-generate")
+api.add_resource(RuleStructuredOutputGenerateApi, "/rule-structured-output-generate")

api/controllers/console/app/workflow_app_log.py

@@ -1,5 +1,4 @@
-from datetime import datetime
-
+from dateutil.parser import isoparse
 from flask_restful import Resource, marshal_with, reqparse  # type: ignore
 from flask_restful.inputs import int_range  # type: ignore
 from sqlalchemy.orm import Session
@@ -41,10 +40,10 @@ class WorkflowAppLogApi(Resource):
 
         args.status = WorkflowRunStatus(args.status) if args.status else None
         if args.created_at__before:
-            args.created_at__before = datetime.fromisoformat(args.created_at__before.replace("Z", "+00:00"))
+            args.created_at__before = isoparse(args.created_at__before)
 
         if args.created_at__after:
-            args.created_at__after = datetime.fromisoformat(args.created_at__after.replace("Z", "+00:00"))
+            args.created_at__after = isoparse(args.created_at__after)
 
         # get paginate workflow app logs
         workflow_app_service = WorkflowAppService()

api/controllers/console/auth/data_source_oauth.py

@@ -74,7 +74,9 @@ class OAuthDataSourceBinding(Resource):
         if not oauth_provider:
             return {"error": "Invalid provider"}, 400
         if "code" in request.args:
-            code = request.args.get("code")
+            code = request.args.get("code", "")
+            if not code:
+                return {"error": "Invalid code"}, 400
             try:
                 oauth_provider.get_access_token(code)
             except requests.exceptions.HTTPError as e:

api/controllers/console/auth/forgot_password.py

@@ -16,7 +16,7 @@ from controllers.console.auth.error import (
     PasswordMismatchError,
 )
 from controllers.console.error import AccountInFreezeError, AccountNotFound, EmailSendIpLimitError
-from controllers.console.wraps import setup_required
+from controllers.console.wraps import email_password_login_enabled, setup_required
 from events.tenant_event import tenant_was_created
 from extensions.ext_database import db
 from libs.helper import email, extract_remote_ip
@@ -30,6 +30,7 @@ from services.feature_service import FeatureService
 
 class ForgotPasswordSendEmailApi(Resource):
     @setup_required
+    @email_password_login_enabled
     def post(self):
         parser = reqparse.RequestParser()
         parser.add_argument("email", type=email, required=True, location="json")
@@ -62,6 +63,7 @@ class ForgotPasswordSendEmailApi(Resource):
 
 class ForgotPasswordCheckApi(Resource):
     @setup_required
+    @email_password_login_enabled
     def post(self):
         parser = reqparse.RequestParser()
         parser.add_argument("email", type=str, required=True, location="json")
@@ -86,12 +88,21 @@ class ForgotPasswordCheckApi(Resource):
             AccountService.add_forgot_password_error_rate_limit(args["email"])
             raise EmailCodeError()
 
+        # Verified, revoke the first token
+        AccountService.revoke_reset_password_token(args["token"])
+
+        # Refresh token data by generating a new token
+        _, new_token = AccountService.generate_reset_password_token(
+            user_email, code=args["code"], additional_data={"phase": "reset"}
+        )
+
         AccountService.reset_forgot_password_error_rate_limit(args["email"])
-        return {"is_valid": True, "email": token_data.get("email")}
+        return {"is_valid": True, "email": token_data.get("email"), "token": new_token}
 
 
 class ForgotPasswordResetApi(Resource):
     @setup_required
+    @email_password_login_enabled
     def post(self):
         parser = reqparse.RequestParser()
         parser.add_argument("token", type=str, required=True, nullable=False, location="json")
@@ -107,6 +118,9 @@ class ForgotPasswordResetApi(Resource):
         reset_data = AccountService.get_reset_password_data(args["token"])
         if not reset_data:
             raise InvalidTokenError()
+        # Must use token in reset phase
+        if reset_data.get("phase", "") != "reset":
+            raise InvalidTokenError()
 
         # Revoke token to prevent reuse
         AccountService.revoke_reset_password_token(args["token"])

api/controllers/console/auth/login.py

@@ -22,7 +22,7 @@ from controllers.console.error import (
     EmailSendIpLimitError,
     NotAllowedCreateWorkspace,
 )
-from controllers.console.wraps import setup_required
+from controllers.console.wraps import email_password_login_enabled, setup_required
 from events.tenant_event import tenant_was_created
 from libs.helper import email, extract_remote_ip
 from libs.password import valid_password
@@ -38,6 +38,7 @@ class LoginApi(Resource):
     """Resource for user login."""
 
     @setup_required
+    @email_password_login_enabled
     def post(self):
         """Authenticate user and login."""
         parser = reqparse.RequestParser()
@@ -110,6 +111,7 @@ class LogoutApi(Resource):
 
 class ResetPasswordSendEmailApi(Resource):
     @setup_required
+    @email_password_login_enabled
     def post(self):
         parser = reqparse.RequestParser()
         parser.add_argument("email", type=email, required=True, location="json")

api/controllers/console/datasets/datasets.py

@@ -664,6 +664,7 @@ class DatasetRetrievalSettingApi(Resource):
                 | VectorType.OPENGAUSS
                 | VectorType.OCEANBASE
                 | VectorType.TABLESTORE
+                | VectorType.HUAWEI_CLOUD
                 | VectorType.TENCENT
             ):
                 return {
@@ -710,6 +711,7 @@ class DatasetRetrievalSettingMockApi(Resource):
                 | VectorType.OCEANBASE
                 | VectorType.TABLESTORE
                 | VectorType.TENCENT
+                | VectorType.HUAWEI_CLOUD
             ):
                 return {
                     "retrieval_method": [

api/controllers/console/datasets/datasets_segments.py

@@ -398,7 +398,7 @@ class DatasetDocumentSegmentBatchImportApi(Resource):
         indexing_cache_key = "segment_batch_import_{}".format(job_id)
         cache_result = redis_client.get(indexing_cache_key)
         if cache_result is None:
-            raise ValueError("The job is not exist.")
+            raise ValueError("The job does not exist.")
 
         return {"job_id": job_id, "job_status": cache_result.decode()}, 200
 

api/controllers/console/datasets/external.py

@@ -21,12 +21,6 @@ def _validate_name(name):
     return name
 
 
-def _validate_description_length(description):
-    if description and len(description) > 400:
-        raise ValueError("Description cannot exceed 400 characters.")
-    return description
-
-
 class ExternalApiTemplateListApi(Resource):
     @setup_required
     @login_required

api/controllers/console/datasets/metadata.py

@@ -14,18 +14,6 @@ from services.entities.knowledge_entities.knowledge_entities import (
 from services.metadata_service import MetadataService
 
 
-def _validate_name(name):
-    if not name or len(name) < 1 or len(name) > 40:
-        raise ValueError("Name must be between 1 to 40 characters.")
-    return name
-
-
-def _validate_description_length(description):
-    if len(description) > 400:
-        raise ValueError("Description cannot exceed 400 characters.")
-    return description
-
-
 class DatasetMetadataCreateApi(Resource):
     @setup_required
     @login_required

api/controllers/console/explore/parameter.py

@@ -1,10 +1,10 @@
 from flask_restful import marshal_with  # type: ignore
 
 from controllers.common import fields
-from controllers.common import helpers as controller_helpers
 from controllers.console import api
 from controllers.console.app.error import AppUnavailableError
 from controllers.console.explore.wraps import InstalledAppResource
+from core.app.app_config.common.parameters_mapping import get_parameters_from_feature_dict
 from models.model import AppMode, InstalledApp
 from services.app_service import AppService
 
@@ -36,9 +36,7 @@ class AppParameterApi(InstalledAppResource):
 
             user_input_form = features_dict.get("user_input_form", [])
 
-        return controller_helpers.get_parameters_from_feature_dict(
-            features_dict=features_dict, user_input_form=user_input_form
-        )
+        return get_parameters_from_feature_dict(features_dict=features_dict, user_input_form=user_input_form)
 
 
 class ExploreAppMetaApi(InstalledAppResource):

api/controllers/console/workspace/account.py

@@ -286,8 +286,6 @@ class AccountDeleteApi(Resource):
 class AccountDeleteUpdateFeedbackApi(Resource):
     @setup_required
     def post(self):
-        account = current_user
-
         parser = reqparse.RequestParser()
         parser.add_argument("email", type=str, required=True, location="json")
         parser.add_argument("feedback", type=str, required=True, location="json")

api/controllers/console/workspace/endpoint.py

@@ -5,6 +5,7 @@ from werkzeug.exceptions import Forbidden
 from controllers.console import api
 from controllers.console.wraps import account_initialization_required, setup_required
 from core.model_runtime.utils.encoders import jsonable_encoder
+from core.plugin.manager.exc import PluginPermissionDeniedError
 from libs.login import login_required
 from services.plugin.endpoint_service import EndpointService
 
@@ -28,15 +29,18 @@ class EndpointCreateApi(Resource):
         settings = args["settings"]
         name = args["name"]
 
-        return {
-            "success": EndpointService.create_endpoint(
-                tenant_id=user.current_tenant_id,
-                user_id=user.id,
-                plugin_unique_identifier=plugin_unique_identifier,
-                name=name,
-                settings=settings,
-            )
-        }
+        try:
+            return {
+                "success": EndpointService.create_endpoint(
+                    tenant_id=user.current_tenant_id,
+                    user_id=user.id,
+                    plugin_unique_identifier=plugin_unique_identifier,
+                    name=name,
+                    settings=settings,
+                )
+            }
+        except PluginPermissionDeniedError as e:
+            raise ValueError(e.description) from e
 
 
 class EndpointListApi(Resource):

api/controllers/console/workspace/plugin.py

@@ -49,6 +49,23 @@ class PluginListApi(Resource):
         return jsonable_encoder({"plugins": plugins})
 
 
+class PluginListLatestVersionsApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self):
+        req = reqparse.RequestParser()
+        req.add_argument("plugin_ids", type=list, required=True, location="json")
+        args = req.parse_args()
+
+        try:
+            versions = PluginService.list_latest_versions(args["plugin_ids"])
+        except PluginDaemonClientSideError as e:
+            raise ValueError(e)
+
+        return jsonable_encoder({"versions": versions})
+
+
 class PluginListInstallationsFromIdsApi(Resource):
     @setup_required
     @login_required
@@ -232,6 +249,31 @@ class PluginInstallFromMarketplaceApi(Resource):
         return jsonable_encoder(response)
 
 
+class PluginFetchMarketplacePkgApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @plugin_permission_required(install_required=True)
+    def get(self):
+        tenant_id = current_user.current_tenant_id
+
+        parser = reqparse.RequestParser()
+        parser.add_argument("plugin_unique_identifier", type=str, required=True, location="args")
+        args = parser.parse_args()
+
+        try:
+            return jsonable_encoder(
+                {
+                    "manifest": PluginService.fetch_marketplace_pkg(
+                        tenant_id,
+                        args["plugin_unique_identifier"],
+                    )
+                }
+            )
+        except PluginDaemonClientSideError as e:
+            raise ValueError(e)
+
+
 class PluginFetchManifestApi(Resource):
     @setup_required
     @login_required
@@ -453,6 +495,7 @@ class PluginFetchPermissionApi(Resource):
 
 api.add_resource(PluginDebuggingKeyApi, "/workspaces/current/plugin/debugging-key")
 api.add_resource(PluginListApi, "/workspaces/current/plugin/list")
+api.add_resource(PluginListLatestVersionsApi, "/workspaces/current/plugin/list/latest-versions")
 api.add_resource(PluginListInstallationsFromIdsApi, "/workspaces/current/plugin/list/installations/ids")
 api.add_resource(PluginIconApi, "/workspaces/current/plugin/icon")
 api.add_resource(PluginUploadFromPkgApi, "/workspaces/current/plugin/upload/pkg")
@@ -470,6 +513,7 @@ api.add_resource(PluginDeleteInstallTaskApi, "/workspaces/current/plugin/tasks/<
 api.add_resource(PluginDeleteAllInstallTaskItemsApi, "/workspaces/current/plugin/tasks/delete_all")
 api.add_resource(PluginDeleteInstallTaskItemApi, "/workspaces/current/plugin/tasks/<task_id>/delete/<path:identifier>")
 api.add_resource(PluginUninstallApi, "/workspaces/current/plugin/uninstall")
+api.add_resource(PluginFetchMarketplacePkgApi, "/workspaces/current/plugin/marketplace/pkg")
 
 api.add_resource(PluginChangePermissionApi, "/workspaces/current/plugin/permission/change")
 api.add_resource(PluginFetchPermissionApi, "/workspaces/current/plugin/permission/fetch")

api/controllers/console/wraps.py

@@ -210,3 +210,16 @@ def enterprise_license_required(view):
         return view(*args, **kwargs)
 
     return decorated
+
+
+def email_password_login_enabled(view):
+    @wraps(view)
+    def decorated(*args, **kwargs):
+        features = FeatureService.get_system_features()
+        if features.enable_email_password_login:
+            return view(*args, **kwargs)
+
+        # otherwise, return 403
+        abort(403)
+
+    return decorated

api/controllers/files/upload.py

@@ -1,3 +1,5 @@
+from mimetypes import guess_extension
+
 from flask import request
 from flask_restful import Resource, marshal_with  # type: ignore
 from werkzeug.exceptions import Forbidden
@@ -9,8 +11,8 @@ from controllers.files.error import UnsupportedFileTypeError
 from controllers.inner_api.plugin.wraps import get_user
 from controllers.service_api.app.error import FileTooLargeError
 from core.file.helpers import verify_plugin_file_signature
+from core.tools.tool_file_manager import ToolFileManager
 from fields.file_fields import file_fields
-from services.file_service import FileService
 
 
 class PluginUploadFileApi(Resource):
@@ -51,19 +53,26 @@ class PluginUploadFileApi(Resource):
             raise Forbidden("Invalid request.")
 
         try:
-            upload_file = FileService.upload_file(
-                filename=filename,
-                content=file.read(),
+            tool_file = ToolFileManager.create_file_by_raw(
+                user_id=user.id,
+                tenant_id=tenant_id,
+                file_binary=file.read(),
                 mimetype=mimetype,
-                user=user,
-                source=None,
+                filename=filename,
+                conversation_id=None,
             )
+
+            extension = guess_extension(tool_file.mimetype) or ".bin"
+            preview_url = ToolFileManager.sign_file(tool_file_id=tool_file.id, extension=extension)
+            tool_file.mime_type = mimetype
+            tool_file.extension = extension
+            tool_file.preview_url = preview_url
         except services.errors.file.FileTooLargeError as file_too_large_error:
             raise FileTooLargeError(file_too_large_error.description)
         except services.errors.file.UnsupportedFileTypeError:
             raise UnsupportedFileTypeError()
 
-        return upload_file, 201
+        return tool_file, 201
 
 
 api.add_resource(PluginUploadFileApi, "/files/upload/for-plugin")

api/controllers/inner_api/plugin/plugin.py

@@ -13,6 +13,7 @@ from core.plugin.backwards_invocation.model import PluginModelBackwardsInvocatio
 from core.plugin.backwards_invocation.node import PluginNodeBackwardsInvocation
 from core.plugin.backwards_invocation.tool import PluginToolBackwardsInvocation
 from core.plugin.entities.request import (
+    RequestFetchAppInfo,
     RequestInvokeApp,
     RequestInvokeEncrypt,
     RequestInvokeLLM,
@@ -278,6 +279,17 @@ class PluginUploadFileRequestApi(Resource):
         return BaseBackwardsInvocationResponse(data={"url": url}).model_dump()
 
 
+class PluginFetchAppInfoApi(Resource):
+    @setup_required
+    @plugin_inner_api_only
+    @get_user_tenant
+    @plugin_data(payload_type=RequestFetchAppInfo)
+    def post(self, user_model: Account | EndUser, tenant_model: Tenant, payload: RequestFetchAppInfo):
+        return BaseBackwardsInvocationResponse(
+            data=PluginAppBackwardsInvocation.fetch_app_info(payload.app_id, tenant_model.id)
+        ).model_dump()
+
+
 api.add_resource(PluginInvokeLLMApi, "/invoke/llm")
 api.add_resource(PluginInvokeTextEmbeddingApi, "/invoke/text-embedding")
 api.add_resource(PluginInvokeRerankApi, "/invoke/rerank")
@@ -291,3 +303,4 @@ api.add_resource(PluginInvokeAppApi, "/invoke/app")
 api.add_resource(PluginInvokeEncryptApi, "/invoke/encrypt")
 api.add_resource(PluginInvokeSummaryApi, "/invoke/summary")
 api.add_resource(PluginUploadFileRequestApi, "/upload/file/request")
+api.add_resource(PluginFetchAppInfoApi, "/fetch/app/info")

api/controllers/service_api/app/app.py

@@ -1,10 +1,10 @@
 from flask_restful import Resource, marshal_with  # type: ignore
 
 from controllers.common import fields
-from controllers.common import helpers as controller_helpers
 from controllers.service_api import api
 from controllers.service_api.app.error import AppUnavailableError
 from controllers.service_api.wraps import validate_app_token
+from core.app.app_config.common.parameters_mapping import get_parameters_from_feature_dict
 from models.model import App, AppMode
 from services.app_service import AppService
 
@@ -32,9 +32,7 @@ class AppParameterApi(Resource):
 
             user_input_form = features_dict.get("user_input_form", [])
 
-        return controller_helpers.get_parameters_from_feature_dict(
-            features_dict=features_dict, user_input_form=user_input_form
-        )
+        return get_parameters_from_feature_dict(features_dict=features_dict, user_input_form=user_input_form)
 
 
 class AppMetaApi(Resource):

api/controllers/service_api/app/message.py

@@ -20,14 +20,6 @@ from services.message_service import MessageService
 
 
 class MessageListApi(Resource):
-    def get_retriever_resources(self):
-        try:
-            if self.message_metadata:
-                return json.loads(self.message_metadata).get("retriever_resources", [])
-            return []
-        except (json.JSONDecodeError, TypeError):
-            return []
-
     message_fields = {
         "id": fields.String,
         "conversation_id": fields.String,
@@ -37,7 +29,11 @@ class MessageListApi(Resource):
         "answer": fields.String(attribute="re_sign_file_url_answer"),
         "message_files": fields.List(fields.Nested(message_file_fields)),
         "feedback": fields.Nested(feedback_fields, attribute="user_feedback", allow_null=True),
-        "retriever_resources": get_retriever_resources,
+        "retriever_resources": fields.Raw(
+            attribute=lambda obj: json.loads(obj.message_metadata).get("retriever_resources", [])
+            if obj.message_metadata
+            else []
+        ),
         "created_at": TimestampField,
         "agent_thoughts": fields.List(fields.Nested(agent_thought_fields)),
         "status": fields.String,

api/controllers/service_api/app/workflow.py

@@ -1,6 +1,6 @@
 import logging
-from datetime import datetime
 
+from dateutil.parser import isoparse
 from flask_restful import Resource, fields, marshal_with, reqparse  # type: ignore
 from flask_restful.inputs import int_range  # type: ignore
 from sqlalchemy.orm import Session
@@ -140,10 +140,10 @@ class WorkflowAppLogApi(Resource):
 
         args.status = WorkflowRunStatus(args.status) if args.status else None
         if args.created_at__before:
-            args.created_at__before = datetime.fromisoformat(args.created_at__before.replace("Z", "+00:00"))
+            args.created_at__before = isoparse(args.created_at__before)
 
         if args.created_at__after:
-            args.created_at__after = datetime.fromisoformat(args.created_at__after.replace("Z", "+00:00"))
+            args.created_at__after = isoparse(args.created_at__after)
 
         # get paginate workflow app logs
         workflow_app_service = WorkflowAppService()

api/controllers/service_api/dataset/dataset.py

@@ -13,6 +13,7 @@ from fields.dataset_fields import dataset_detail_fields
 from libs.login import current_user
 from models.dataset import Dataset, DatasetPermissionEnum
 from services.dataset_service import DatasetPermissionService, DatasetService
+from services.entities.knowledge_entities.knowledge_entities import RetrievalModel
 
 
 def _validate_name(name):
@@ -120,8 +121,11 @@ class DatasetListApi(DatasetApiResource):
             nullable=True,
             required=False,
         )
-        args = parser.parse_args()
+        parser.add_argument("retrieval_model", type=dict, required=False, nullable=True, location="json")
+        parser.add_argument("embedding_model", type=str, required=False, nullable=True, location="json")
+        parser.add_argument("embedding_model_provider", type=str, required=False, nullable=True, location="json")
 
+        args = parser.parse_args()
         try:
             dataset = DatasetService.create_empty_dataset(
                 tenant_id=tenant_id,
@@ -133,6 +137,11 @@ class DatasetListApi(DatasetApiResource):
                 provider=args["provider"],
                 external_knowledge_api_id=args["external_knowledge_api_id"],
                 external_knowledge_id=args["external_knowledge_id"],
+                embedding_model_provider=args["embedding_model_provider"],
+                embedding_model_name=args["embedding_model"],
+                retrieval_model=RetrievalModel(**args["retrieval_model"])
+                if args["retrieval_model"] is not None
+                else None,
             )
         except services.errors.dataset.DatasetNameDuplicateError:
             raise DatasetNameDuplicateError()

api/controllers/service_api/dataset/document.py

@@ -49,7 +49,9 @@ class DocumentAddByTextApi(DatasetApiResource):
         parser.add_argument(
             "indexing_technique", type=str, choices=Dataset.INDEXING_TECHNIQUE_LIST, nullable=False, location="json"
         )
-        parser.add_argument("retrieval_model", type=dict, required=False, nullable=False, location="json")
+        parser.add_argument("retrieval_model", type=dict, required=False, nullable=True, location="json")
+        parser.add_argument("embedding_model", type=str, required=False, nullable=True, location="json")
+        parser.add_argument("embedding_model_provider", type=str, required=False, nullable=True, location="json")
 
         args = parser.parse_args()
         dataset_id = str(dataset_id)
@@ -57,7 +59,7 @@ class DocumentAddByTextApi(DatasetApiResource):
         dataset = db.session.query(Dataset).filter(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id).first()
 
         if not dataset:
-            raise ValueError("Dataset is not exist.")
+            raise ValueError("Dataset does not exist.")
 
         if not dataset.indexing_technique and not args["indexing_technique"]:
             raise ValueError("indexing_technique is required.")
@@ -114,7 +116,7 @@ class DocumentUpdateByTextApi(DatasetApiResource):
         dataset = db.session.query(Dataset).filter(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id).first()
 
         if not dataset:
-            raise ValueError("Dataset is not exist.")
+            raise ValueError("Dataset does not exist.")
 
         # indexing_technique is already set in dataset since this is an update
         args["indexing_technique"] = dataset.indexing_technique
@@ -172,7 +174,7 @@ class DocumentAddByFileApi(DatasetApiResource):
         dataset = db.session.query(Dataset).filter(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id).first()
 
         if not dataset:
-            raise ValueError("Dataset is not exist.")
+            raise ValueError("Dataset does not exist.")
         if not dataset.indexing_technique and not args.get("indexing_technique"):
             raise ValueError("indexing_technique is required.")
 
@@ -239,7 +241,7 @@ class DocumentUpdateByFileApi(DatasetApiResource):
         dataset = db.session.query(Dataset).filter(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id).first()
 
         if not dataset:
-            raise ValueError("Dataset is not exist.")
+            raise ValueError("Dataset does not exist.")
 
         # indexing_technique is already set in dataset since this is an update
         args["indexing_technique"] = dataset.indexing_technique
@@ -303,7 +305,7 @@ class DocumentDeleteApi(DatasetApiResource):
         dataset = db.session.query(Dataset).filter(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id).first()
 
         if not dataset:
-            raise ValueError("Dataset is not exist.")
+            raise ValueError("Dataset does not exist.")
 
         document = DocumentService.get_document(dataset.id, document_id)
 

api/controllers/service_api/dataset/metadata.py

@@ -13,18 +13,6 @@ from services.entities.knowledge_entities.knowledge_entities import (
 from services.metadata_service import MetadataService
 
 
-def _validate_name(name):
-    if not name or len(name) < 1 or len(name) > 40:
-        raise ValueError("Name must be between 1 to 40 characters.")
-    return name
-
-
-def _validate_description_length(description):
-    if len(description) > 400:
-        raise ValueError("Description cannot exceed 400 characters.")
-    return description
-
-
 class DatasetMetadataCreateServiceApi(DatasetApiResource):
     def post(self, tenant_id, dataset_id):
         parser = reqparse.RequestParser()

api/controllers/service_api/dataset/segment.py

@@ -117,14 +117,13 @@ class SegmentApi(DatasetApiResource):
         parser.add_argument("keyword", type=str, default=None, location="args")
         args = parser.parse_args()
 
-        status_list = args["status"]
-        keyword = args["keyword"]
-
         segments, total = SegmentService.get_segments(
             document_id=document_id,
             tenant_id=current_user.current_tenant_id,
             status_list=args["status"],
             keyword=args["keyword"],
+            page=page,
+            limit=limit,
         )
 
         response = {

api/controllers/web/app.py

@@ -1,10 +1,10 @@
 from flask_restful import marshal_with  # type: ignore
 
 from controllers.common import fields
-from controllers.common import helpers as controller_helpers
 from controllers.web import api
 from controllers.web.error import AppUnavailableError
 from controllers.web.wraps import WebApiResource
+from core.app.app_config.common.parameters_mapping import get_parameters_from_feature_dict
 from models.model import App, AppMode
 from services.app_service import AppService
 
@@ -31,9 +31,7 @@ class AppParameterApi(WebApiResource):
 
             user_input_form = features_dict.get("user_input_form", [])
 
-        return controller_helpers.get_parameters_from_feature_dict(
-            features_dict=features_dict, user_input_form=user_input_form
-        )
+        return get_parameters_from_feature_dict(features_dict=features_dict, user_input_form=user_input_form)
 
 
 class AppMeta(WebApiResource):

api/controllers/web/message.py

@@ -46,6 +46,7 @@ class MessageListApi(WebApiResource):
         "retriever_resources": fields.List(fields.Nested(retriever_resource_fields)),
         "created_at": TimestampField,
         "agent_thoughts": fields.List(fields.Nested(agent_thought_fields)),
+        "metadata": fields.Raw(attribute="message_metadata_dict"),
         "status": fields.String,
         "error": fields.String,
     }

api/core/agent/base_agent_runner.py

@@ -21,14 +21,13 @@ from core.model_runtime.entities import (
     AssistantPromptMessage,
     LLMUsage,
     PromptMessage,
-    PromptMessageContent,
     PromptMessageTool,
     SystemPromptMessage,
     TextPromptMessageContent,
     ToolPromptMessage,
     UserPromptMessage,
 )
-from core.model_runtime.entities.message_entities import ImagePromptMessageContent
+from core.model_runtime.entities.message_entities import ImagePromptMessageContent, PromptMessageContentUnionTypes
 from core.model_runtime.entities.model_entities import ModelFeature
 from core.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
 from core.prompt.utils.extract_thread_messages import extract_thread_messages
@@ -501,7 +500,7 @@ class BaseAgentRunner(AppRunner):
         )
         if not file_objs:
             return UserPromptMessage(content=message.query)
-        prompt_message_contents: list[PromptMessageContent] = []
+        prompt_message_contents: list[PromptMessageContentUnionTypes] = []
         prompt_message_contents.append(TextPromptMessageContent(data=message.query))
         for file in file_objs:
             prompt_message_contents.append(

api/core/agent/cot_agent_runner.py

@@ -191,7 +191,7 @@ class CotAgentRunner(BaseAgentRunner, ABC):
                     # action is final answer, return final answer directly
                     try:
                         if isinstance(scratchpad.action.action_input, dict):
-                            final_answer = json.dumps(scratchpad.action.action_input)
+                            final_answer = json.dumps(scratchpad.action.action_input, ensure_ascii=False)
                         elif isinstance(scratchpad.action.action_input, str):
                             final_answer = scratchpad.action.action_input
                         else:

api/core/agent/cot_chat_agent_runner.py

@@ -5,12 +5,11 @@ from core.file import file_manager
 from core.model_runtime.entities import (
     AssistantPromptMessage,
     PromptMessage,
-    PromptMessageContent,
     SystemPromptMessage,
     TextPromptMessageContent,
     UserPromptMessage,
 )
-from core.model_runtime.entities.message_entities import ImagePromptMessageContent
+from core.model_runtime.entities.message_entities import ImagePromptMessageContent, PromptMessageContentUnionTypes
 from core.model_runtime.utils.encoders import jsonable_encoder
 
 
@@ -40,7 +39,7 @@ class CotChatAgentRunner(CotAgentRunner):
         Organize user query
         """
         if self.files:
-            prompt_message_contents: list[PromptMessageContent] = []
+            prompt_message_contents: list[PromptMessageContentUnionTypes] = []
             prompt_message_contents.append(TextPromptMessageContent(data=query))
 
             # get image detail config

api/core/agent/fc_agent_runner.py

@@ -15,14 +15,13 @@ from core.model_runtime.entities import (
     LLMResultChunkDelta,
     LLMUsage,
     PromptMessage,
-    PromptMessageContent,
     PromptMessageContentType,
     SystemPromptMessage,
     TextPromptMessageContent,
     ToolPromptMessage,
     UserPromptMessage,
 )
-from core.model_runtime.entities.message_entities import ImagePromptMessageContent
+from core.model_runtime.entities.message_entities import ImagePromptMessageContent, PromptMessageContentUnionTypes
 from core.prompt.agent_history_prompt_transform import AgentHistoryPromptTransform
 from core.tools.entities.tool_entities import ToolInvokeMeta
 from core.tools.tool_engine import ToolEngine
@@ -395,7 +394,7 @@ class FunctionCallAgentRunner(BaseAgentRunner):
         Organize user query
         """
         if self.files:
-            prompt_message_contents: list[PromptMessageContent] = []
+            prompt_message_contents: list[PromptMessageContentUnionTypes] = []
             prompt_message_contents.append(TextPromptMessageContent(data=query))
 
             # get image detail config

api/core/agent/plugin_entities.py

@@ -52,6 +52,7 @@ class AgentStrategyParameter(PluginParameter):
             return cast_parameter_value(self, value)
 
     type: AgentStrategyParameterType = Field(..., description="The type of the parameter")
+    help: Optional[I18nObject] = None
 
     def init_frontend_parameter(self, value: Any):
         return init_frontend_parameter(self, self.type, value)

api/core/app/app_config/common/parameters_mapping/__init__.py

@@ -0,0 +1,45 @@
+from collections.abc import Mapping
+from typing import Any
+
+from configs import dify_config
+from constants import DEFAULT_FILE_NUMBER_LIMITS
+
+
+def get_parameters_from_feature_dict(
+    *, features_dict: Mapping[str, Any], user_input_form: list[dict[str, Any]]
+) -> Mapping[str, Any]:
+    """
+    Mapping from feature dict to webapp parameters
+    """
+    return {
+        "opening_statement": features_dict.get("opening_statement"),
+        "suggested_questions": features_dict.get("suggested_questions", []),
+        "suggested_questions_after_answer": features_dict.get("suggested_questions_after_answer", {"enabled": False}),
+        "speech_to_text": features_dict.get("speech_to_text", {"enabled": False}),
+        "text_to_speech": features_dict.get("text_to_speech", {"enabled": False}),
+        "retriever_resource": features_dict.get("retriever_resource", {"enabled": False}),
+        "annotation_reply": features_dict.get("annotation_reply", {"enabled": False}),
+        "more_like_this": features_dict.get("more_like_this", {"enabled": False}),
+        "user_input_form": user_input_form,
+        "sensitive_word_avoidance": features_dict.get(
+            "sensitive_word_avoidance", {"enabled": False, "type": "", "configs": []}
+        ),
+        "file_upload": features_dict.get(
+            "file_upload",
+            {
+                "image": {
+                    "enabled": False,
+                    "number_limits": DEFAULT_FILE_NUMBER_LIMITS,
+                    "detail": "high",
+                    "transfer_methods": ["remote_url", "local_file"],
+                }
+            },
+        ),
+        "system_parameters": {
+            "image_file_size_limit": dify_config.UPLOAD_IMAGE_FILE_SIZE_LIMIT,
+            "video_file_size_limit": dify_config.UPLOAD_VIDEO_FILE_SIZE_LIMIT,
+            "audio_file_size_limit": dify_config.UPLOAD_AUDIO_FILE_SIZE_LIMIT,
+            "file_size_limit": dify_config.UPLOAD_FILE_SIZE_LIMIT,
+            "workflow_file_upload_limit": dify_config.WORKFLOW_FILE_UPLOAD_LIMIT,
+        },
+    }

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

@@ -1,6 +1,7 @@
 from collections.abc import Mapping
 from typing import Any
 
+from constants import DEFAULT_FILE_NUMBER_LIMITS
 from core.file import FileUploadConfig
 
 
@@ -18,7 +19,7 @@ class FileUploadConfigManager:
             if file_upload_dict.get("enabled"):
                 transform_methods = file_upload_dict.get("allowed_file_upload_methods", [])
                 file_upload_dict["image_config"] = {
-                    "number_limits": file_upload_dict.get("number_limits", 1),
+                    "number_limits": file_upload_dict.get("number_limits", DEFAULT_FILE_NUMBER_LIMITS),
                     "transfer_methods": transform_methods,
                 }
 

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

@@ -320,10 +320,9 @@ class AdvancedChatAppGenerateTaskPipeline:
                         session=session, workflow_run_id=self._workflow_run_id
                     )
                     workflow_node_execution = self._workflow_cycle_manager._handle_workflow_node_execution_retried(
-                        session=session, workflow_run=workflow_run, event=event
+                        workflow_run=workflow_run, event=event
                     )
                     node_retry_resp = self._workflow_cycle_manager._workflow_node_retry_to_stream_response(
-                        session=session,
                         event=event,
                         task_id=self._application_generate_entity.task_id,
                         workflow_node_execution=workflow_node_execution,
@@ -341,11 +340,10 @@ class AdvancedChatAppGenerateTaskPipeline:
                         session=session, workflow_run_id=self._workflow_run_id
                     )
                     workflow_node_execution = self._workflow_cycle_manager._handle_node_execution_start(
-                        session=session, workflow_run=workflow_run, event=event
+                        workflow_run=workflow_run, event=event
                     )
 
                     node_start_resp = self._workflow_cycle_manager._workflow_node_start_to_stream_response(
-                        session=session,
                         event=event,
                         task_id=self._application_generate_entity.task_id,
                         workflow_node_execution=workflow_node_execution,
@@ -363,11 +361,10 @@ class AdvancedChatAppGenerateTaskPipeline:
 
                 with Session(db.engine, expire_on_commit=False) as session:
                     workflow_node_execution = self._workflow_cycle_manager._handle_workflow_node_execution_success(
-                        session=session, event=event
+                        event=event
                     )
 
                     node_finish_resp = self._workflow_cycle_manager._workflow_node_finish_to_stream_response(
-                        session=session,
                         event=event,
                         task_id=self._application_generate_entity.task_id,
                         workflow_node_execution=workflow_node_execution,
@@ -383,18 +380,15 @@ class AdvancedChatAppGenerateTaskPipeline:
                 | QueueNodeInLoopFailedEvent
                 | QueueNodeExceptionEvent,
             ):
-                with Session(db.engine, expire_on_commit=False) as session:
-                    workflow_node_execution = self._workflow_cycle_manager._handle_workflow_node_execution_failed(
-                        session=session, event=event
-                    )
+                workflow_node_execution = self._workflow_cycle_manager._handle_workflow_node_execution_failed(
+                    event=event
+                )
 
-                    node_finish_resp = self._workflow_cycle_manager._workflow_node_finish_to_stream_response(
-                        session=session,
-                        event=event,
-                        task_id=self._application_generate_entity.task_id,
-                        workflow_node_execution=workflow_node_execution,
-                    )
-                    session.commit()
+                node_finish_resp = self._workflow_cycle_manager._workflow_node_finish_to_stream_response(
+                    event=event,
+                    task_id=self._application_generate_entity.task_id,
+                    workflow_node_execution=workflow_node_execution,
+                )
 
                 if node_finish_resp:
                     yield node_finish_resp

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

@@ -53,20 +53,6 @@ class AgentChatAppRunner(AppRunner):
         query = application_generate_entity.query
         files = application_generate_entity.files
 
-        # Pre-calculate the number of tokens of the prompt messages,
-        # and return the rest number of tokens by model context token size limit and max token size limit.
-        # If the rest number of tokens is not enough, raise exception.
-        # Include: prompt template, inputs, query(optional), files(optional)
-        # Not Include: memory, external data, dataset context
-        self.get_pre_calculate_rest_tokens(
-            app_record=app_record,
-            model_config=application_generate_entity.model_conf,
-            prompt_template_entity=app_config.prompt_template,
-            inputs=dict(inputs),
-            files=list(files),
-            query=query,
-        )
-
         memory = None
         if application_generate_entity.conversation_id:
             # get memory of conversation (read-only)

api/core/app/apps/base_app_generator.py

@@ -17,6 +17,7 @@ class BaseAppGenerator:
         user_inputs: Optional[Mapping[str, Any]],
         variables: Sequence["VariableEntity"],
         tenant_id: str,
+        strict_type_validation: bool = False,
     ) -> Mapping[str, Any]:
         user_inputs = user_inputs or {}
         # Filter input variables from form configuration, handle required fields, default values, and option values
@@ -37,6 +38,7 @@ class BaseAppGenerator:
                     allowed_file_extensions=entity_dictionary[k].allowed_file_extensions,
                     allowed_file_upload_methods=entity_dictionary[k].allowed_file_upload_methods,
                 ),
+                strict_type_validation=strict_type_validation,
             )
             for k, v in user_inputs.items()
             if isinstance(v, dict) and entity_dictionary[k].type == VariableEntityType.FILE

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

@@ -61,20 +61,6 @@ class ChatAppRunner(AppRunner):
         )
         image_detail_config = image_detail_config or ImagePromptMessageContent.DETAIL.LOW
 
-        # Pre-calculate the number of tokens of the prompt messages,
-        # and return the rest number of tokens by model context token size limit and max token size limit.
-        # If the rest number of tokens is not enough, raise exception.
-        # Include: prompt template, inputs, query(optional), files(optional)
-        # Not Include: memory, external data, dataset context
-        self.get_pre_calculate_rest_tokens(
-            app_record=app_record,
-            model_config=application_generate_entity.model_conf,
-            prompt_template_entity=app_config.prompt_template,
-            inputs=inputs,
-            files=files,
-            query=query,
-        )
-
         memory = None
         if application_generate_entity.conversation_id:
             # get memory of conversation (read-only)

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

@@ -54,20 +54,6 @@ class CompletionAppRunner(AppRunner):
         )
         image_detail_config = image_detail_config or ImagePromptMessageContent.DETAIL.LOW
 
-        # Pre-calculate the number of tokens of the prompt messages,
-        # and return the rest number of tokens by model context token size limit and max token size limit.
-        # If the rest number of tokens is not enough, raise exception.
-        # Include: prompt template, inputs, query(optional), files(optional)
-        # Not Include: memory, external data, dataset context
-        self.get_pre_calculate_rest_tokens(
-            app_record=app_record,
-            model_config=application_generate_entity.model_conf,
-            prompt_template_entity=app_config.prompt_template,
-            inputs=inputs,
-            files=files,
-            query=query,
-        )
-
         # organize all inputs and template to prompt messages
         # Include: prompt template, inputs, query(optional), files(optional)
         prompt_messages, stop = self.organize_prompt_messages(

api/core/app/apps/message_based_app_generator.py

@@ -153,6 +153,7 @@ class MessageBasedAppGenerator(BaseAppGenerator):
             query = application_generate_entity.query or "New conversation"
         else:
             query = next(iter(application_generate_entity.inputs.values()), "New conversation")
+        query = query or "New conversation"
         conversation_name = (query[:20] + "…") if len(query) > 20 else query
 
         if not conversation:

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

@@ -92,6 +92,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
             mappings=files,
             tenant_id=app_model.tenant_id,
             config=file_extra_config,
+            strict_type_validation=True if invoke_from == InvokeFrom.SERVICE_API else False,
         )
 
         # convert to app config
@@ -114,7 +115,10 @@ class WorkflowAppGenerator(BaseAppGenerator):
             app_config=app_config,
             file_upload_config=file_extra_config,
             inputs=self._prepare_user_inputs(
-                user_inputs=inputs, variables=app_config.variables, tenant_id=app_model.tenant_id
+                user_inputs=inputs,
+                variables=app_config.variables,
+                tenant_id=app_model.tenant_id,
+                strict_type_validation=True if invoke_from == InvokeFrom.SERVICE_API else False,
             ),
             files=list(system_files),
             user_id=user.id,

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

@@ -279,10 +279,9 @@ class WorkflowAppGenerateTaskPipeline:
                         session=session, workflow_run_id=self._workflow_run_id
                     )
                     workflow_node_execution = self._workflow_cycle_manager._handle_workflow_node_execution_retried(
-                        session=session, workflow_run=workflow_run, event=event
+                        workflow_run=workflow_run, event=event
                     )
                     response = self._workflow_cycle_manager._workflow_node_retry_to_stream_response(
-                        session=session,
                         event=event,
                         task_id=self._application_generate_entity.task_id,
                         workflow_node_execution=workflow_node_execution,
@@ -300,10 +299,9 @@ class WorkflowAppGenerateTaskPipeline:
                         session=session, workflow_run_id=self._workflow_run_id
                     )
                     workflow_node_execution = self._workflow_cycle_manager._handle_node_execution_start(
-                        session=session, workflow_run=workflow_run, event=event
+                        workflow_run=workflow_run, event=event
                     )
                     node_start_response = self._workflow_cycle_manager._workflow_node_start_to_stream_response(
-                        session=session,
                         event=event,
                         task_id=self._application_generate_entity.task_id,
                         workflow_node_execution=workflow_node_execution,
@@ -313,17 +311,14 @@ class WorkflowAppGenerateTaskPipeline:
                 if node_start_response:
                     yield node_start_response
             elif isinstance(event, QueueNodeSucceededEvent):
-                with Session(db.engine, expire_on_commit=False) as session:
-                    workflow_node_execution = self._workflow_cycle_manager._handle_workflow_node_execution_success(
-                        session=session, event=event
-                    )
-                    node_success_response = self._workflow_cycle_manager._workflow_node_finish_to_stream_response(
-                        session=session,
-                        event=event,
-                        task_id=self._application_generate_entity.task_id,
-                        workflow_node_execution=workflow_node_execution,
-                    )
-                    session.commit()
+                workflow_node_execution = self._workflow_cycle_manager._handle_workflow_node_execution_success(
+                    event=event
+                )
+                node_success_response = self._workflow_cycle_manager._workflow_node_finish_to_stream_response(
+                    event=event,
+                    task_id=self._application_generate_entity.task_id,
+                    workflow_node_execution=workflow_node_execution,
+                )
 
                 if node_success_response:
                     yield node_success_response
@@ -334,18 +329,14 @@ class WorkflowAppGenerateTaskPipeline:
                 | QueueNodeInLoopFailedEvent
                 | QueueNodeExceptionEvent,
             ):
-                with Session(db.engine, expire_on_commit=False) as session:
-                    workflow_node_execution = self._workflow_cycle_manager._handle_workflow_node_execution_failed(
-                        session=session,
-                        event=event,
-                    )
-                    node_failed_response = self._workflow_cycle_manager._workflow_node_finish_to_stream_response(
-                        session=session,
-                        event=event,
-                        task_id=self._application_generate_entity.task_id,
-                        workflow_node_execution=workflow_node_execution,
-                    )
-                    session.commit()
+                workflow_node_execution = self._workflow_cycle_manager._handle_workflow_node_execution_failed(
+                    event=event,
+                )
+                node_failed_response = self._workflow_cycle_manager._workflow_node_finish_to_stream_response(
+                    event=event,
+                    task_id=self._application_generate_entity.task_id,
+                    workflow_node_execution=workflow_node_execution,
+                )
 
                 if node_failed_response:
                     yield node_failed_response
@@ -627,6 +618,7 @@ class WorkflowAppGenerateTaskPipeline:
         workflow_app_log.created_by = self._user_id
 
         session.add(workflow_app_log)
+        session.commit()
 
     def _text_chunk_to_stream_response(
         self, text: str, from_variable_selector: Optional[list[str]] = None

api/core/app/task_pipeline/workflow_cycle_manage.py

@@ -6,7 +6,7 @@ from typing import Any, Optional, Union, cast
 from uuid import uuid4
 
 from sqlalchemy import func, select
-from sqlalchemy.orm import Session
+from sqlalchemy.orm import Session, sessionmaker
 
 from core.app.entities.app_invoke_entities import AdvancedChatAppGenerateEntity, InvokeFrom, WorkflowAppGenerateEntity
 from core.app.entities.queue_entities import (
@@ -49,12 +49,14 @@ from core.file import FILE_MODEL_IDENTITY, File
 from core.model_runtime.utils.encoders import jsonable_encoder
 from core.ops.entities.trace_entity import TraceTaskName
 from core.ops.ops_trace_manager import TraceQueueManager, TraceTask
+from core.repository import RepositoryFactory
 from core.tools.tool_manager import ToolManager
 from core.workflow.entities.node_entities import NodeRunMetadataKey
 from core.workflow.enums import SystemVariableKey
 from core.workflow.nodes import NodeType
 from core.workflow.nodes.tool.entities import ToolNodeData
 from core.workflow.workflow_entry import WorkflowEntry
+from extensions.ext_database import db
 from models.account import Account
 from models.enums import CreatedByRole, WorkflowRunTriggeredFrom
 from models.model import EndUser
@@ -80,6 +82,21 @@ class WorkflowCycleManage:
         self._application_generate_entity = application_generate_entity
         self._workflow_system_variables = workflow_system_variables
 
+        # Initialize the session factory and repository
+        # We use the global db engine instead of the session passed to methods
+        # Disable expire_on_commit to avoid the need for merging objects
+        self._session_factory = sessionmaker(bind=db.engine, expire_on_commit=False)
+        self._workflow_node_execution_repository = RepositoryFactory.create_workflow_node_execution_repository(
+            params={
+                "tenant_id": self._application_generate_entity.app_config.tenant_id,
+                "app_id": self._application_generate_entity.app_config.app_id,
+                "session_factory": self._session_factory,
+            }
+        )
+
+        # We'll still keep the cache for backward compatibility and performance
+        # but use the repository for database operations
+
     def _handle_workflow_run_start(
         self,
         *,
@@ -254,19 +271,15 @@ class WorkflowCycleManage:
         workflow_run.finished_at = datetime.now(UTC).replace(tzinfo=None)
         workflow_run.exceptions_count = exceptions_count
 
-        stmt = select(WorkflowNodeExecution.node_execution_id).where(
-            WorkflowNodeExecution.tenant_id == workflow_run.tenant_id,
-            WorkflowNodeExecution.app_id == workflow_run.app_id,
-            WorkflowNodeExecution.workflow_id == workflow_run.workflow_id,
-            WorkflowNodeExecution.triggered_from == WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN.value,
-            WorkflowNodeExecution.workflow_run_id == workflow_run.id,
-            WorkflowNodeExecution.status == WorkflowNodeExecutionStatus.RUNNING.value,
+        # Use the instance repository to find running executions for a workflow run
+        running_workflow_node_executions = self._workflow_node_execution_repository.get_running_executions(
+            workflow_run_id=workflow_run.id
         )
-        ids = session.scalars(stmt).all()
-        # Use self._get_workflow_node_execution here to make sure the cache is updated
-        running_workflow_node_executions = [
-            self._get_workflow_node_execution(session=session, node_execution_id=id) for id in ids if id
-        ]
+
+        # Update the cache with the retrieved executions
+        for execution in running_workflow_node_executions:
+            if execution.node_execution_id:
+                self._workflow_node_executions[execution.node_execution_id] = execution
 
         for workflow_node_execution in running_workflow_node_executions:
             now = datetime.now(UTC).replace(tzinfo=None)
@@ -288,7 +301,7 @@ class WorkflowCycleManage:
         return workflow_run
 
     def _handle_node_execution_start(
-        self, *, session: Session, workflow_run: WorkflowRun, event: QueueNodeStartedEvent
+        self, *, workflow_run: WorkflowRun, event: QueueNodeStartedEvent
     ) -> WorkflowNodeExecution:
         workflow_node_execution = WorkflowNodeExecution()
         workflow_node_execution.id = str(uuid4())
@@ -315,17 +328,14 @@ class WorkflowCycleManage:
         )
         workflow_node_execution.created_at = datetime.now(UTC).replace(tzinfo=None)
 
-        session.add(workflow_node_execution)
+        # Use the instance repository to save the workflow node execution
+        self._workflow_node_execution_repository.save(workflow_node_execution)
 
         self._workflow_node_executions[event.node_execution_id] = workflow_node_execution
         return workflow_node_execution
 
-    def _handle_workflow_node_execution_success(
-        self, *, session: Session, event: QueueNodeSucceededEvent
-    ) -> WorkflowNodeExecution:
-        workflow_node_execution = self._get_workflow_node_execution(
-            session=session, node_execution_id=event.node_execution_id
-        )
+    def _handle_workflow_node_execution_success(self, *, event: QueueNodeSucceededEvent) -> WorkflowNodeExecution:
+        workflow_node_execution = self._get_workflow_node_execution(node_execution_id=event.node_execution_id)
         inputs = WorkflowEntry.handle_special_values(event.inputs)
         process_data = WorkflowEntry.handle_special_values(event.process_data)
         outputs = WorkflowEntry.handle_special_values(event.outputs)
@@ -344,13 +354,13 @@ class WorkflowCycleManage:
         workflow_node_execution.finished_at = finished_at
         workflow_node_execution.elapsed_time = elapsed_time
 
-        workflow_node_execution = session.merge(workflow_node_execution)
+        # Use the instance repository to update the workflow node execution
+        self._workflow_node_execution_repository.update(workflow_node_execution)
         return workflow_node_execution
 
     def _handle_workflow_node_execution_failed(
         self,
         *,
-        session: Session,
         event: QueueNodeFailedEvent
         | QueueNodeInIterationFailedEvent
         | QueueNodeInLoopFailedEvent
@@ -361,9 +371,7 @@ class WorkflowCycleManage:
         :param event: queue node failed event
         :return:
         """
-        workflow_node_execution = self._get_workflow_node_execution(
-            session=session, node_execution_id=event.node_execution_id
-        )
+        workflow_node_execution = self._get_workflow_node_execution(node_execution_id=event.node_execution_id)
 
         inputs = WorkflowEntry.handle_special_values(event.inputs)
         process_data = WorkflowEntry.handle_special_values(event.process_data)
@@ -387,14 +395,14 @@ class WorkflowCycleManage:
         workflow_node_execution.elapsed_time = elapsed_time
         workflow_node_execution.execution_metadata = execution_metadata
 
-        workflow_node_execution = session.merge(workflow_node_execution)
         return workflow_node_execution
 
     def _handle_workflow_node_execution_retried(
-        self, *, session: Session, workflow_run: WorkflowRun, event: QueueNodeRetryEvent
+        self, *, workflow_run: WorkflowRun, event: QueueNodeRetryEvent
     ) -> WorkflowNodeExecution:
         """
         Workflow node execution failed
+        :param workflow_run: workflow run
         :param event: queue node failed event
         :return:
         """
@@ -439,15 +447,12 @@ class WorkflowCycleManage:
         workflow_node_execution.execution_metadata = execution_metadata
         workflow_node_execution.index = event.node_run_index
 
-        session.add(workflow_node_execution)
+        # Use the instance repository to save the workflow node execution
+        self._workflow_node_execution_repository.save(workflow_node_execution)
 
         self._workflow_node_executions[event.node_execution_id] = workflow_node_execution
         return workflow_node_execution
 
-    #################################################
-    #             to stream responses               #
-    #################################################
-
     def _workflow_start_to_stream_response(
         self,
         *,
@@ -455,7 +460,6 @@ class WorkflowCycleManage:
         task_id: str,
         workflow_run: WorkflowRun,
     ) -> WorkflowStartStreamResponse:
-        # receive session to make sure the workflow_run won't be expired, need a more elegant way to handle this
         _ = session
         return WorkflowStartStreamResponse(
             task_id=task_id,
@@ -521,14 +525,10 @@ class WorkflowCycleManage:
     def _workflow_node_start_to_stream_response(
         self,
         *,
-        session: Session,
         event: QueueNodeStartedEvent,
         task_id: str,
         workflow_node_execution: WorkflowNodeExecution,
     ) -> Optional[NodeStartStreamResponse]:
-        # receive session to make sure the workflow_run won't be expired, need a more elegant way to handle this
-        _ = session
-
         if workflow_node_execution.node_type in {NodeType.ITERATION.value, NodeType.LOOP.value}:
             return None
         if not workflow_node_execution.workflow_run_id:
@@ -571,7 +571,6 @@ class WorkflowCycleManage:
     def _workflow_node_finish_to_stream_response(
         self,
         *,
-        session: Session,
         event: QueueNodeSucceededEvent
         | QueueNodeFailedEvent
         | QueueNodeInIterationFailedEvent
@@ -580,8 +579,6 @@ class WorkflowCycleManage:
         task_id: str,
         workflow_node_execution: WorkflowNodeExecution,
     ) -> Optional[NodeFinishStreamResponse]:
-        # receive session to make sure the workflow_run won't be expired, need a more elegant way to handle this
-        _ = session
         if workflow_node_execution.node_type in {NodeType.ITERATION.value, NodeType.LOOP.value}:
             return None
         if not workflow_node_execution.workflow_run_id:
@@ -621,13 +618,10 @@ class WorkflowCycleManage:
     def _workflow_node_retry_to_stream_response(
         self,
         *,
-        session: Session,
         event: QueueNodeRetryEvent,
         task_id: str,
         workflow_node_execution: WorkflowNodeExecution,
     ) -> Optional[Union[NodeRetryStreamResponse, NodeFinishStreamResponse]]:
-        # receive session to make sure the workflow_run won't be expired, need a more elegant way to handle this
-        _ = session
         if workflow_node_execution.node_type in {NodeType.ITERATION.value, NodeType.LOOP.value}:
             return None
         if not workflow_node_execution.workflow_run_id:
@@ -668,7 +662,6 @@ class WorkflowCycleManage:
     def _workflow_parallel_branch_start_to_stream_response(
         self, *, session: Session, task_id: str, workflow_run: WorkflowRun, event: QueueParallelBranchRunStartedEvent
     ) -> ParallelBranchStartStreamResponse:
-        # receive session to make sure the workflow_run won't be expired, need a more elegant way to handle this
         _ = session
         return ParallelBranchStartStreamResponse(
             task_id=task_id,
@@ -692,7 +685,6 @@ class WorkflowCycleManage:
         workflow_run: WorkflowRun,
         event: QueueParallelBranchRunSucceededEvent | QueueParallelBranchRunFailedEvent,
     ) -> ParallelBranchFinishedStreamResponse:
-        # receive session to make sure the workflow_run won't be expired, need a more elegant way to handle this
         _ = session
         return ParallelBranchFinishedStreamResponse(
             task_id=task_id,
@@ -713,7 +705,6 @@ class WorkflowCycleManage:
     def _workflow_iteration_start_to_stream_response(
         self, *, session: Session, task_id: str, workflow_run: WorkflowRun, event: QueueIterationStartEvent
     ) -> IterationNodeStartStreamResponse:
-        # receive session to make sure the workflow_run won't be expired, need a more elegant way to handle this
         _ = session
         return IterationNodeStartStreamResponse(
             task_id=task_id,
@@ -735,7 +726,6 @@ class WorkflowCycleManage:
     def _workflow_iteration_next_to_stream_response(
         self, *, session: Session, task_id: str, workflow_run: WorkflowRun, event: QueueIterationNextEvent
     ) -> IterationNodeNextStreamResponse:
-        # receive session to make sure the workflow_run won't be expired, need a more elegant way to handle this
         _ = session
         return IterationNodeNextStreamResponse(
             task_id=task_id,
@@ -759,7 +749,6 @@ class WorkflowCycleManage:
     def _workflow_iteration_completed_to_stream_response(
         self, *, session: Session, task_id: str, workflow_run: WorkflowRun, event: QueueIterationCompletedEvent
     ) -> IterationNodeCompletedStreamResponse:
-        # receive session to make sure the workflow_run won't be expired, need a more elegant way to handle this
         _ = session
         return IterationNodeCompletedStreamResponse(
             task_id=task_id,
@@ -790,7 +779,6 @@ class WorkflowCycleManage:
     def _workflow_loop_start_to_stream_response(
         self, *, session: Session, task_id: str, workflow_run: WorkflowRun, event: QueueLoopStartEvent
     ) -> LoopNodeStartStreamResponse:
-        # receive session to make sure the workflow_run won't be expired, need a more elegant way to handle this
         _ = session
         return LoopNodeStartStreamResponse(
             task_id=task_id,
@@ -812,7 +800,6 @@ class WorkflowCycleManage:
     def _workflow_loop_next_to_stream_response(
         self, *, session: Session, task_id: str, workflow_run: WorkflowRun, event: QueueLoopNextEvent
     ) -> LoopNodeNextStreamResponse:
-        # receive session to make sure the workflow_run won't be expired, need a more elegant way to handle this
         _ = session
         return LoopNodeNextStreamResponse(
             task_id=task_id,
@@ -836,7 +823,6 @@ class WorkflowCycleManage:
     def _workflow_loop_completed_to_stream_response(
         self, *, session: Session, task_id: str, workflow_run: WorkflowRun, event: QueueLoopCompletedEvent
     ) -> LoopNodeCompletedStreamResponse:
-        # receive session to make sure the workflow_run won't be expired, need a more elegant way to handle this
         _ = session
         return LoopNodeCompletedStreamResponse(
             task_id=task_id,
@@ -934,11 +920,22 @@ class WorkflowCycleManage:
 
         return workflow_run
 
-    def _get_workflow_node_execution(self, session: Session, node_execution_id: str) -> WorkflowNodeExecution:
-        if node_execution_id not in self._workflow_node_executions:
+    def _get_workflow_node_execution(self, node_execution_id: str) -> WorkflowNodeExecution:
+        # First check the cache for performance
+        if node_execution_id in self._workflow_node_executions:
+            cached_execution = self._workflow_node_executions[node_execution_id]
+            # No need to merge with session since expire_on_commit=False
+            return cached_execution
+
+        # If not in cache, use the instance repository to get by node_execution_id
+        execution = self._workflow_node_execution_repository.get_by_node_execution_id(node_execution_id)
+
+        if not execution:
             raise ValueError(f"Workflow node execution not found: {node_execution_id}")
-        cached_workflow_node_execution = self._workflow_node_executions[node_execution_id]
-        return session.merge(cached_workflow_node_execution)
+
+        # Update cache
+        self._workflow_node_executions[node_execution_id] = execution
+        return execution
 
     def _handle_agent_log(self, task_id: str, event: QueueAgentLogEvent) -> AgentLogStreamResponse:
         """

api/core/callback_handler/index_tool_callback_handler.py

@@ -6,7 +6,6 @@ from core.rag.models.document import Document
 from extensions.ext_database import db
 from models.dataset import ChildChunk, DatasetQuery, DocumentSegment
 from models.dataset import Document as DatasetDocument
-from models.model import DatasetRetrieverResource
 
 
 class DatasetIndexToolCallbackHandler:
@@ -71,29 +70,6 @@ class DatasetIndexToolCallbackHandler:
 
     def return_retriever_resource_info(self, resource: list):
         """Handle return_retriever_resource_info."""
-        if resource and len(resource) > 0:
-            for item in resource:
-                dataset_retriever_resource = DatasetRetrieverResource(
-                    message_id=self._message_id,
-                    position=item.get("position") or 0,
-                    dataset_id=item.get("dataset_id"),
-                    dataset_name=item.get("dataset_name"),
-                    document_id=item.get("document_id"),
-                    document_name=item.get("document_name"),
-                    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" 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,
-                    content=item.get("content"),
-                    retriever_from=item.get("retriever_from"),
-                    created_by=self._user_id,
-                )
-                db.session.add(dataset_retriever_resource)
-                db.session.commit()
-
         self._queue_manager.publish(
             QueueRetrieverResourcesEvent(retriever_resources=resource), PublishFrom.APPLICATION_MANAGER
         )

api/core/file/file_manager.py

@@ -7,9 +7,9 @@ from core.model_runtime.entities import (
     AudioPromptMessageContent,
     DocumentPromptMessageContent,
     ImagePromptMessageContent,
-    MultiModalPromptMessageContent,
     VideoPromptMessageContent,
 )
+from core.model_runtime.entities.message_entities import PromptMessageContentUnionTypes
 from extensions.ext_storage import storage
 
 from . import helpers
@@ -43,7 +43,7 @@ def to_prompt_message_content(
     /,
     *,
     image_detail_config: ImagePromptMessageContent.DETAIL | None = None,
-) -> MultiModalPromptMessageContent:
+) -> PromptMessageContentUnionTypes:
     if f.extension is None:
         raise ValueError("Missing file extension")
     if f.mime_type is None:
@@ -58,7 +58,7 @@ def to_prompt_message_content(
     if f.type == FileType.IMAGE:
         params["detail"] = image_detail_config or ImagePromptMessageContent.DETAIL.LOW
 
-    prompt_class_map: Mapping[FileType, type[MultiModalPromptMessageContent]] = {
+    prompt_class_map: Mapping[FileType, type[PromptMessageContentUnionTypes]] = {
         FileType.IMAGE: ImagePromptMessageContent,
         FileType.AUDIO: AudioPromptMessageContent,
         FileType.VIDEO: VideoPromptMessageContent,

api/core/helper/ssrf_proxy.py

@@ -48,25 +48,26 @@ def make_request(method, url, max_retries=SSRF_DEFAULT_MAX_RETRIES, **kwargs):
             write=dify_config.SSRF_DEFAULT_WRITE_TIME_OUT,
         )
 
+    if "ssl_verify" not in kwargs:
+        kwargs["ssl_verify"] = HTTP_REQUEST_NODE_SSL_VERIFY
+
+    ssl_verify = kwargs.pop("ssl_verify")
+
     retries = 0
     while retries <= max_retries:
         try:
             if dify_config.SSRF_PROXY_ALL_URL:
-                with httpx.Client(proxy=dify_config.SSRF_PROXY_ALL_URL, verify=HTTP_REQUEST_NODE_SSL_VERIFY) as client:
+                with httpx.Client(proxy=dify_config.SSRF_PROXY_ALL_URL, verify=ssl_verify) as client:
                     response = client.request(method=method, url=url, **kwargs)
             elif dify_config.SSRF_PROXY_HTTP_URL and dify_config.SSRF_PROXY_HTTPS_URL:
                 proxy_mounts = {
-                    "http://": httpx.HTTPTransport(
-                        proxy=dify_config.SSRF_PROXY_HTTP_URL, verify=HTTP_REQUEST_NODE_SSL_VERIFY
-                    ),
-                    "https://": httpx.HTTPTransport(
-                        proxy=dify_config.SSRF_PROXY_HTTPS_URL, verify=HTTP_REQUEST_NODE_SSL_VERIFY
-                    ),
+                    "http://": httpx.HTTPTransport(proxy=dify_config.SSRF_PROXY_HTTP_URL, verify=ssl_verify),
+                    "https://": httpx.HTTPTransport(proxy=dify_config.SSRF_PROXY_HTTPS_URL, verify=ssl_verify),
                 }
-                with httpx.Client(mounts=proxy_mounts, verify=HTTP_REQUEST_NODE_SSL_VERIFY) as client:
+                with httpx.Client(mounts=proxy_mounts, verify=ssl_verify) as client:
                     response = client.request(method=method, url=url, **kwargs)
             else:
-                with httpx.Client(verify=HTTP_REQUEST_NODE_SSL_VERIFY) as client:
+                with httpx.Client(verify=ssl_verify) as client:
                     response = client.request(method=method, url=url, **kwargs)
 
             if response.status_code not in STATUS_FORCELIST:

api/core/llm_generator/llm_generator.py

@@ -10,6 +10,7 @@ from core.llm_generator.prompts import (
     GENERATOR_QA_PROMPT,
     JAVASCRIPT_CODE_GENERATOR_PROMPT_TEMPLATE,
     PYTHON_CODE_GENERATOR_PROMPT_TEMPLATE,
+    SYSTEM_STRUCTURED_OUTPUT_GENERATE,
     WORKFLOW_RULE_CONFIG_PROMPT_GENERATE_TEMPLATE,
 )
 from core.model_manager import ModelManager
@@ -340,3 +341,37 @@ class LLMGenerator:
 
         answer = cast(str, response.message.content)
         return answer.strip()
+
+    @classmethod
+    def generate_structured_output(cls, tenant_id: str, instruction: str, model_config: dict):
+        model_manager = ModelManager()
+        model_instance = model_manager.get_model_instance(
+            tenant_id=tenant_id,
+            model_type=ModelType.LLM,
+            provider=model_config.get("provider", ""),
+            model=model_config.get("name", ""),
+        )
+
+        prompt_messages = [
+            SystemPromptMessage(content=SYSTEM_STRUCTURED_OUTPUT_GENERATE),
+            UserPromptMessage(content=instruction),
+        ]
+        model_parameters = model_config.get("model_parameters", {})
+
+        try:
+            response = cast(
+                LLMResult,
+                model_instance.invoke_llm(
+                    prompt_messages=list(prompt_messages), model_parameters=model_parameters, stream=False
+                ),
+            )
+
+            generated_json_schema = cast(str, response.message.content)
+            return {"output": generated_json_schema, "error": ""}
+
+        except InvokeError as e:
+            error = str(e)
+            return {"output": "", "error": f"Failed to generate JSON Schema. Error: {error}"}
+        except Exception as e:
+            logging.exception(f"Failed to invoke LLM model, model: {model_config.get('name')}")
+            return {"output": "", "error": f"An unexpected error occurred: {str(e)}"}

api/core/llm_generator/prompts.py

@@ -220,3 +220,110 @@ Here is the task description: {{INPUT_TEXT}}
 
 You just need to generate the output
 """  # noqa: E501
+
+SYSTEM_STRUCTURED_OUTPUT_GENERATE = """
+Your task is to convert simple user descriptions into properly formatted JSON Schema definitions. When a user describes data fields they need, generate a complete, valid JSON Schema that accurately represents those fields with appropriate types and requirements.
+
+## Instructions:
+
+1. Analyze the user's description of their data needs
+2. Identify each property that should be included in the schema
+3. Determine the appropriate data type for each property
+4. Decide which properties should be required
+5. Generate a complete JSON Schema with proper syntax
+6. Include appropriate constraints when specified (min/max values, patterns, formats)
+7. Provide ONLY the JSON Schema without any additional explanations, comments, or markdown formatting.
+8. DO NOT use markdown code blocks (``` or ``` json). Return the raw JSON Schema directly.
+
+## Examples:
+
+### Example 1:
+**User Input:** I need name and age
+**JSON Schema Output:**
+{
+  "type": "object",
+  "properties": {
+    "name": { "type": "string" },
+    "age": { "type": "number" }
+  },
+  "required": ["name", "age"]
+}
+
+### Example 2:
+**User Input:** I want to store information about books including title, author, publication year and optional page count
+**JSON Schema Output:**
+{
+  "type": "object",
+  "properties": {
+    "title": { "type": "string" },
+    "author": { "type": "string" },
+    "publicationYear": { "type": "integer" },
+    "pageCount": { "type": "integer" }
+  },
+  "required": ["title", "author", "publicationYear"]
+}
+
+### Example 3:
+**User Input:** Create a schema for user profiles with email, password, and age (must be at least 18)
+**JSON Schema Output:**
+{
+  "type": "object",
+  "properties": {
+    "email": { 
+      "type": "string",
+      "format": "email"
+    },
+    "password": { 
+      "type": "string",
+      "minLength": 8
+    },
+    "age": { 
+      "type": "integer",
+      "minimum": 18
+    }
+  },
+  "required": ["email", "password", "age"]
+}
+
+### Example 4:
+**User Input:** I need album schema, the ablum has songs, and each song has name, duration, and artist.
+**JSON Schema Output:**
+{
+    "type": "object",
+    "properties": {
+        "properties": {
+            "songs": {
+                "type": "array",
+                "items": {
+                    "type": "object",
+                    "properties": {
+                        "name": {
+                            "type": "string"
+                        },
+                        "id": {
+                            "type": "string"
+                        },
+                        "duration": {
+                            "type": "string"
+                        },
+                        "aritst": {
+                            "type": "string"
+                        }
+                    },
+                    "required": [
+                        "name",
+                        "id",
+                        "duration",
+                        "aritst"
+                    ]
+                }
+            }
+        }
+    },
+    "required": [
+        "songs"
+    ]
+}
+
+Now, generate a JSON Schema based on my description
+"""  # noqa: E501

api/core/memory/base_memory.py

@@ -0,0 +1,64 @@
+from abc import abstractmethod
+from collections.abc import Sequence
+from typing import Optional
+
+from core.model_runtime.entities import (
+    ImagePromptMessageContent,
+    PromptMessage,
+    PromptMessageRole,
+    TextPromptMessageContent,
+)
+
+
+class BaseMemory:
+    @abstractmethod
+    def get_history_prompt_messages(
+        self, max_token_limit: int = 2000, message_limit: Optional[int] = None
+    ) -> Sequence[PromptMessage]:
+        """
+        Get history prompt messages.
+        :param max_token_limit: max token limit
+        :param message_limit: message limit
+        :return:
+        """
+
+    def get_history_prompt_text(
+        self,
+        human_prefix: str = "Human",
+        ai_prefix: str = "Assistant",
+        max_token_limit: int = 2000,
+        message_limit: Optional[int] = None,
+    ) -> str:
+        """
+        Get history prompt text.
+        :param human_prefix: human prefix
+        :param ai_prefix: ai prefix
+        :param max_token_limit: max token limit
+        :param message_limit: message limit
+        :return:
+        """
+        prompt_messages = self.get_history_prompt_messages(max_token_limit=max_token_limit, message_limit=message_limit)
+
+        string_messages = []
+        for m in prompt_messages:
+            if m.role == PromptMessageRole.USER:
+                role = human_prefix
+            elif m.role == PromptMessageRole.ASSISTANT:
+                role = ai_prefix
+            else:
+                continue
+
+            if isinstance(m.content, list):
+                inner_msg = ""
+                for content in m.content:
+                    if isinstance(content, TextPromptMessageContent):
+                        inner_msg += f"{content.data}\n"
+                    elif isinstance(content, ImagePromptMessageContent):
+                        inner_msg += "[image]\n"
+
+                string_messages.append(f"{role}: {inner_msg.strip()}")
+            else:
+                message = f"{role}: {m.content}"
+                string_messages.append(message)
+
+        return "\n".join(string_messages)

api/core/memory/model_context_memory.py

@@ -0,0 +1,200 @@
+import json
+from collections.abc import Sequence
+from typing import Optional, cast
+
+from core.app.app_config.features.file_upload.manager import FileUploadConfigManager
+from core.file import file_manager
+from core.memory.base_memory import BaseMemory
+from core.model_manager import ModelInstance
+from core.model_runtime.entities.message_entities import (
+    AssistantPromptMessage,
+    ImagePromptMessageContent,
+    PromptMessage,
+    PromptMessageContentUnionTypes,
+    TextPromptMessageContent,
+    UserPromptMessage,
+)
+from core.prompt.entities.advanced_prompt_entities import LLMMemoryType
+from core.prompt.utils.extract_thread_messages import extract_thread_messages
+from extensions.ext_database import db
+from factories import file_factory
+from models.model import AppMode, Conversation, Message, MessageFile
+from models.workflow import WorkflowNodeExecution, WorkflowNodeExecutionStatus, WorkflowRun
+
+
+class ModelContextMemory(BaseMemory):
+    def __init__(self, conversation: Conversation, node_id: str, model_instance: ModelInstance) -> None:
+        self.conversation = conversation
+        self.node_id = node_id
+        self.model_instance = model_instance
+
+    def get_history_prompt_messages(
+        self, max_token_limit: int = 2000, message_limit: Optional[int] = None
+    ) -> Sequence[PromptMessage]:
+        """
+        Get history prompt messages.
+        :param max_token_limit: max token limit
+        :param message_limit: message limit
+        """
+        thread_messages = list(reversed(self._fetch_thread_messages(message_limit)))
+        if not thread_messages:
+            return []
+        # Get all required workflow_run_ids
+        workflow_run_ids = [msg.workflow_run_id for msg in thread_messages]
+
+        # Batch query all related WorkflowNodeExecution records
+        node_executions = (
+            db.session.query(WorkflowNodeExecution)
+            .filter(
+                WorkflowNodeExecution.workflow_run_id.in_(workflow_run_ids),
+                WorkflowNodeExecution.node_id == self.node_id,
+                WorkflowNodeExecution.status.in_(
+                    [WorkflowNodeExecutionStatus.SUCCEEDED, WorkflowNodeExecutionStatus.EXCEPTION]
+                ),
+            )
+            .all()
+        )
+
+        # Create mapping from workflow_run_id to node_execution
+        node_execution_map = {ne.workflow_run_id: ne for ne in node_executions}
+
+        # Get the last node_execution
+        last_node_execution = node_execution_map.get(thread_messages[-1].workflow_run_id)
+        prompt_messages = self._get_prompt_messages_in_process_data(last_node_execution)
+
+        # Batch query all message-related files
+        message_ids = [msg.id for msg in thread_messages]
+        all_files = db.session.query(MessageFile).filter(MessageFile.message_id.in_(message_ids)).all()
+
+        # Create mapping from message_id to files
+        files_map = {}
+        for file in all_files:
+            if file.message_id not in files_map:
+                files_map[file.message_id] = []
+            files_map[file.message_id].append(file)
+
+        for message in thread_messages:
+            files = files_map.get(message.id, [])
+            node_execution = node_execution_map.get(message.workflow_run_id)
+            if node_execution and files:
+                file_objs, detail = self._handle_file(message, files)
+                if file_objs:
+                    outputs = node_execution.outputs_dict.get("text", "") if node_execution.outputs_dict else ""
+                    if not outputs:
+                        continue
+                    if outputs not in [prompt.content for prompt in prompt_messages]:
+                        continue
+                    outputs_index = [prompt.content for prompt in prompt_messages].index(outputs)
+                    prompt_index = outputs_index - 1
+                    prompt_message_contents: list[PromptMessageContentUnionTypes] = []
+                    content = cast(str, prompt_messages[prompt_index].content)
+                    prompt_message_contents.append(TextPromptMessageContent(data=content))
+                    for file in file_objs:
+                        prompt_message = file_manager.to_prompt_message_content(
+                            file,
+                            image_detail_config=detail,
+                        )
+                        prompt_message_contents.append(prompt_message)
+                    prompt_messages[prompt_index].content = prompt_message_contents
+        return prompt_messages
+
+    def _get_prompt_messages_in_process_data(
+        self,
+        node_execution: WorkflowNodeExecution,
+    ) -> list[PromptMessage]:
+        """
+        Get prompt messages in process data.
+        :param node_execution: node execution
+        :return: prompt messages
+        """
+        prompt_messages = []
+        if not node_execution.process_data:
+            return []
+
+        try:
+            process_data = json.loads(node_execution.process_data)
+            if process_data.get("memory_type", "") != LLMMemoryType.INDEPENDENT:
+                return []
+            prompts = process_data.get("prompts", [])
+            for prompt in prompts:
+                prompt_content = prompt.get("text", "")
+                if prompt.get("role", "") == "user":
+                    prompt_messages.append(UserPromptMessage(content=prompt_content))
+                elif prompt.get("role", "") == "assistant":
+                    prompt_messages.append(AssistantPromptMessage(content=prompt_content))
+            output = node_execution.outputs_dict.get("text", "") if node_execution.outputs_dict else ""
+            prompt_messages.append(AssistantPromptMessage(content=output))
+        except json.JSONDecodeError:
+            return []
+        return prompt_messages
+
+    def _fetch_thread_messages(self, message_limit: int | None = None) -> list[Message]:
+        """
+        Fetch thread messages.
+        :param message_limit: message limit
+        :return: thread messages
+        """
+        query = (
+            db.session.query(
+                Message.id,
+                Message.query,
+                Message.answer,
+                Message.created_at,
+                Message.workflow_run_id,
+                Message.parent_message_id,
+                Message.answer_tokens,
+            )
+            .filter(
+                Message.conversation_id == self.conversation.id,
+            )
+            .order_by(Message.created_at.desc())
+        )
+
+        if message_limit and message_limit > 0:
+            message_limit = min(message_limit, 500)
+        else:
+            message_limit = 500
+
+        messages = query.limit(message_limit).all()
+
+        # fetch the thread messages
+        thread_messages = extract_thread_messages(messages)
+
+        # for newly created message, its answer is temporarily empty, we don't need to add it to memory
+        if thread_messages and not thread_messages[0].answer and thread_messages[0].answer_tokens == 0:
+            thread_messages.pop(0)
+        if not thread_messages:
+            return []
+        return thread_messages
+
+    def _handle_file(self, message: Message, files: list[MessageFile]):
+        """
+        Handle file for memory.
+        :param message: message
+        :param files: files
+        :return: file objects and detail
+        """
+        file_extra_config = None
+        if self.conversation.mode not in {AppMode.ADVANCED_CHAT, AppMode.WORKFLOW}:
+            file_extra_config = FileUploadConfigManager.convert(self.conversation.model_config)
+        else:
+            if message.workflow_run_id:
+                workflow_run = db.session.query(WorkflowRun).filter(WorkflowRun.id == message.workflow_run_id).first()
+
+                if workflow_run and workflow_run.workflow:
+                    file_extra_config = FileUploadConfigManager.convert(
+                        workflow_run.workflow.features_dict, is_vision=False
+                    )
+
+        detail = ImagePromptMessageContent.DETAIL.LOW
+        app_record = self.conversation.app
+
+        if file_extra_config and app_record:
+            file_objs = file_factory.build_from_message_files(
+                message_files=files, tenant_id=app_record.tenant_id, config=file_extra_config
+            )
+            if file_extra_config.image_config and file_extra_config.image_config.detail:
+                detail = file_extra_config.image_config.detail
+        else:
+            file_objs = []
+        return file_objs, detail

api/core/memory/token_buffer_memory.py

@@ -3,16 +3,16 @@ from typing import Optional
 
 from core.app.app_config.features.file_upload.manager import FileUploadConfigManager
 from core.file import file_manager
+from core.memory.base_memory import BaseMemory
 from core.model_manager import ModelInstance
 from core.model_runtime.entities import (
     AssistantPromptMessage,
     ImagePromptMessageContent,
     PromptMessage,
-    PromptMessageContent,
-    PromptMessageRole,
     TextPromptMessageContent,
     UserPromptMessage,
 )
+from core.model_runtime.entities.message_entities import PromptMessageContentUnionTypes
 from core.prompt.utils.extract_thread_messages import extract_thread_messages
 from extensions.ext_database import db
 from factories import file_factory
@@ -20,7 +20,7 @@ from models.model import AppMode, Conversation, Message, MessageFile
 from models.workflow import WorkflowRun
 
 
-class TokenBufferMemory:
+class TokenBufferMemory(BaseMemory):
     def __init__(self, conversation: Conversation, model_instance: ModelInstance) -> None:
         self.conversation = conversation
         self.model_instance = model_instance
@@ -44,6 +44,7 @@ class TokenBufferMemory:
                 Message.created_at,
                 Message.workflow_run_id,
                 Message.parent_message_id,
+                Message.answer_tokens,
             )
             .filter(
                 Message.conversation_id == self.conversation.id,
@@ -63,7 +64,7 @@ class TokenBufferMemory:
         thread_messages = extract_thread_messages(messages)
 
         # for newly created message, its answer is temporarily empty, we don't need to add it to memory
-        if thread_messages and not thread_messages[0].answer:
+        if thread_messages and not thread_messages[0].answer and thread_messages[0].answer_tokens == 0:
             thread_messages.pop(0)
 
         messages = list(reversed(thread_messages))
@@ -99,7 +100,7 @@ class TokenBufferMemory:
                 if not file_objs:
                     prompt_messages.append(UserPromptMessage(content=message.query))
                 else:
-                    prompt_message_contents: list[PromptMessageContent] = []
+                    prompt_message_contents: list[PromptMessageContentUnionTypes] = []
                     prompt_message_contents.append(TextPromptMessageContent(data=message.query))
                     for file in file_objs:
                         prompt_message = file_manager.to_prompt_message_content(
@@ -128,44 +129,3 @@ class TokenBufferMemory:
                 curr_message_tokens = self.model_instance.get_llm_num_tokens(prompt_messages)
 
         return prompt_messages
-
-    def get_history_prompt_text(
-        self,
-        human_prefix: str = "Human",
-        ai_prefix: str = "Assistant",
-        max_token_limit: int = 2000,
-        message_limit: Optional[int] = None,
-    ) -> str:
-        """
-        Get history prompt text.
-        :param human_prefix: human prefix
-        :param ai_prefix: ai prefix
-        :param max_token_limit: max token limit
-        :param message_limit: message limit
-        :return:
-        """
-        prompt_messages = self.get_history_prompt_messages(max_token_limit=max_token_limit, message_limit=message_limit)
-
-        string_messages = []
-        for m in prompt_messages:
-            if m.role == PromptMessageRole.USER:
-                role = human_prefix
-            elif m.role == PromptMessageRole.ASSISTANT:
-                role = ai_prefix
-            else:
-                continue
-
-            if isinstance(m.content, list):
-                inner_msg = ""
-                for content in m.content:
-                    if isinstance(content, TextPromptMessageContent):
-                        inner_msg += f"{content.data}\n"
-                    elif isinstance(content, ImagePromptMessageContent):
-                        inner_msg += "[image]\n"
-
-                string_messages.append(f"{role}: {inner_msg.strip()}")
-            else:
-                message = f"{role}: {m.content}"
-                string_messages.append(message)
-
-        return "\n".join(string_messages)

api/core/model_manager.py

@@ -177,7 +177,7 @@ class ModelInstance:
         )
 
     def get_llm_num_tokens(
-        self, prompt_messages: list[PromptMessage], tools: Optional[list[PromptMessageTool]] = None
+        self, prompt_messages: Sequence[PromptMessage], tools: Optional[Sequence[PromptMessageTool]] = None
     ) -> int:
         """
         Get number of tokens for llm