Merge remote-tracking branch 'origin/main' into feat/support-agent-sandbox

# Conflicts:
#	api/.env.example
#	api/uv.lock
#	web/app/components/app/create-app-modal/index.tsx
#	web/app/components/app/create-from-dsl-modal/index.tsx
#	web/app/components/apps/app-card.tsx
#	web/pnpm-lock.yaml

api/core/app/apps/base_app_generate_response_converter.py

@@ -79,6 +79,7 @@ class AppGenerateResponseConverter(ABC):
                         "document_name": resource["document_name"],
                         "score": resource["score"],
                         "content": resource["content"],
+                        "summary": resource.get("summary"),
                     }
                 )
             metadata["retriever_resources"] = updated_resources

api/core/entities/knowledge_entities.py

@@ -3,6 +3,7 @@ from pydantic import BaseModel, Field, field_validator
 
 class PreviewDetail(BaseModel):
     content: str
+    summary: str | None = None
     child_chunks: list[str] | None = None
 
 

api/core/file/file_manager.py

@@ -123,6 +123,8 @@ def download(f: File, /):
     ):
         return _download_file_content(f.storage_key)
     elif f.transfer_method == FileTransferMethod.REMOTE_URL:
+        if f.remote_url is None:
+            raise ValueError("Missing file remote_url")
         response = ssrf_proxy.get(f.remote_url, follow_redirects=True)
         response.raise_for_status()
         return response.content
@@ -153,6 +155,8 @@ def _download_file_content(path: str, /):
 def _get_encoded_string(f: File, /):
     match f.transfer_method:
         case FileTransferMethod.REMOTE_URL:
+            if f.remote_url is None:
+                raise ValueError("Missing file remote_url")
             response = ssrf_proxy.get(f.remote_url, follow_redirects=True)
             response.raise_for_status()
             data = response.content

api/core/helper/ssrf_proxy.py

@@ -4,8 +4,10 @@ Proxy requests to avoid SSRF
 
 import logging
 import time
+from typing import Any, TypeAlias
 
 import httpx
+from pydantic import TypeAdapter, ValidationError
 
 from configs import dify_config
 from core.helper.http_client_pooling import get_pooled_http_client
@@ -18,6 +20,9 @@ SSRF_DEFAULT_MAX_RETRIES = dify_config.SSRF_DEFAULT_MAX_RETRIES
 BACKOFF_FACTOR = 0.5
 STATUS_FORCELIST = [429, 500, 502, 503, 504]
 
+Headers: TypeAlias = dict[str, str]
+_HEADERS_ADAPTER = TypeAdapter(Headers)
+
 _SSL_VERIFIED_POOL_KEY = "ssrf:verified"
 _SSL_UNVERIFIED_POOL_KEY = "ssrf:unverified"
 _SSRF_CLIENT_LIMITS = httpx.Limits(
@@ -76,7 +81,7 @@ def _get_ssrf_client(ssl_verify_enabled: bool) -> httpx.Client:
     )
 
 
-def _get_user_provided_host_header(headers: dict | None) -> str | None:
+def _get_user_provided_host_header(headers: Headers | None) -> str | None:
     """
     Extract the user-provided Host header from the headers dict.
 
@@ -92,7 +97,7 @@ def _get_user_provided_host_header(headers: dict | None) -> str | None:
     return None
 
 
-def _inject_trace_headers(headers: dict | None) -> dict:
+def _inject_trace_headers(headers: Headers | None) -> Headers:
     """
     Inject W3C traceparent header for distributed tracing.
 
@@ -125,7 +130,7 @@ def _inject_trace_headers(headers: dict | None) -> dict:
     return headers
 
 
-def make_request(method, url, max_retries=SSRF_DEFAULT_MAX_RETRIES, **kwargs):
+def make_request(method: str, url: str, max_retries: int = SSRF_DEFAULT_MAX_RETRIES, **kwargs: Any) -> httpx.Response:
     # Convert requests-style allow_redirects to httpx-style follow_redirects
     if "allow_redirects" in kwargs:
         allow_redirects = kwargs.pop("allow_redirects")
@@ -142,10 +147,15 @@ def make_request(method, url, max_retries=SSRF_DEFAULT_MAX_RETRIES, **kwargs):
 
     # prioritize per-call option, which can be switched on and off inside the HTTP node on the web UI
     verify_option = kwargs.pop("ssl_verify", dify_config.HTTP_REQUEST_NODE_SSL_VERIFY)
+    if not isinstance(verify_option, bool):
+        raise ValueError("ssl_verify must be a boolean")
     client = _get_ssrf_client(verify_option)
 
     # Inject traceparent header for distributed tracing (when OTEL is not enabled)
-    headers = kwargs.get("headers") or {}
+    try:
+        headers: Headers = _HEADERS_ADAPTER.validate_python(kwargs.get("headers") or {})
+    except ValidationError as e:
+        raise ValueError("headers must be a mapping of string keys to string values") from e
     headers = _inject_trace_headers(headers)
     kwargs["headers"] = headers
 
@@ -198,25 +208,25 @@ def make_request(method, url, max_retries=SSRF_DEFAULT_MAX_RETRIES, **kwargs):
     raise MaxRetriesExceededError(f"Reached maximum retries ({max_retries}) for URL {url}")
 
 
-def get(url, max_retries=SSRF_DEFAULT_MAX_RETRIES, **kwargs):
+def get(url: str, max_retries: int = SSRF_DEFAULT_MAX_RETRIES, **kwargs: Any) -> httpx.Response:
     return make_request("GET", url, max_retries=max_retries, **kwargs)
 
 
-def post(url, max_retries=SSRF_DEFAULT_MAX_RETRIES, **kwargs):
+def post(url: str, max_retries: int = SSRF_DEFAULT_MAX_RETRIES, **kwargs: Any) -> httpx.Response:
     return make_request("POST", url, max_retries=max_retries, **kwargs)
 
 
-def put(url, max_retries=SSRF_DEFAULT_MAX_RETRIES, **kwargs):
+def put(url: str, max_retries: int = SSRF_DEFAULT_MAX_RETRIES, **kwargs: Any) -> httpx.Response:
     return make_request("PUT", url, max_retries=max_retries, **kwargs)
 
 
-def patch(url, max_retries=SSRF_DEFAULT_MAX_RETRIES, **kwargs):
+def patch(url: str, max_retries: int = SSRF_DEFAULT_MAX_RETRIES, **kwargs: Any) -> httpx.Response:
     return make_request("PATCH", url, max_retries=max_retries, **kwargs)
 
 
-def delete(url, max_retries=SSRF_DEFAULT_MAX_RETRIES, **kwargs):
+def delete(url: str, max_retries: int = SSRF_DEFAULT_MAX_RETRIES, **kwargs: Any) -> httpx.Response:
     return make_request("DELETE", url, max_retries=max_retries, **kwargs)
 
 
-def head(url, max_retries=SSRF_DEFAULT_MAX_RETRIES, **kwargs):
+def head(url: str, max_retries: int = SSRF_DEFAULT_MAX_RETRIES, **kwargs: Any) -> httpx.Response:
     return make_request("HEAD", url, max_retries=max_retries, **kwargs)

api/core/indexing_runner.py

@@ -311,14 +311,18 @@ class IndexingRunner:
         qa_preview_texts: list[QAPreviewDetail] = []
 
         total_segments = 0
+        # doc_form represents the segmentation method (general, parent-child, QA)
         index_type = doc_form
         index_processor = IndexProcessorFactory(index_type).init_index_processor()
+        # one extract_setting is one source document
         for extract_setting in extract_settings:
             # extract
             processing_rule = DatasetProcessRule(
                 mode=tmp_processing_rule["mode"], rules=json.dumps(tmp_processing_rule["rules"])
             )
+            # Extract document content
             text_docs = index_processor.extract(extract_setting, process_rule_mode=tmp_processing_rule["mode"])
+            # Cleaning and segmentation
             documents = index_processor.transform(
                 text_docs,
                 current_user=None,
@@ -361,6 +365,12 @@ class IndexingRunner:
 
         if doc_form and doc_form == "qa_model":
             return IndexingEstimate(total_segments=total_segments * 20, qa_preview=qa_preview_texts, preview=[])
+
+        # Generate summary preview
+        summary_index_setting = tmp_processing_rule.get("summary_index_setting")
+        if summary_index_setting and summary_index_setting.get("enable") and preview_texts:
+            preview_texts = index_processor.generate_summary_preview(tenant_id, preview_texts, summary_index_setting)
+
         return IndexingEstimate(total_segments=total_segments, preview=preview_texts)
 
     def _extract(

api/core/llm_generator/prompts.py

@@ -435,3 +435,20 @@ INSTRUCTION_GENERATE_TEMPLATE_PROMPT = """The output of this prompt is not as ex
 You should edit the prompt according to the IDEAL OUTPUT."""
 
 INSTRUCTION_GENERATE_TEMPLATE_CODE = """Please fix the errors in the {{#error_message#}}."""
+
+DEFAULT_GENERATOR_SUMMARY_PROMPT = (
+    """Summarize the following content. Extract only the key information and main points. """
+    """Remove redundant details.
+
+Requirements:
+1. Write a concise summary in plain text
+2. Use the same language as the input content
+3. Focus on important facts, concepts, and details
+4. If images are included, describe their key information
+5. Do not use words like "好的", "ok", "I understand", "This text discusses", "The content mentions"
+6. Write directly without extra words
+
+Output only the summary text. Start summarizing now:
+
+"""
+)

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

@@ -1,10 +1,11 @@
 import decimal
 import hashlib
-from threading import Lock
+import logging
 
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import BaseModel, ConfigDict, Field, ValidationError
+from redis import RedisError
 
-import contexts
+from configs import dify_config
 from core.model_runtime.entities.common_entities import I18nObject
 from core.model_runtime.entities.defaults import PARAMETER_RULE_TEMPLATE
 from core.model_runtime.entities.model_entities import (
@@ -24,6 +25,9 @@ from core.model_runtime.errors.invoke import (
     InvokeServerUnavailableError,
 )
 from core.plugin.entities.plugin_daemon import PluginModelProviderEntity
+from extensions.ext_redis import redis_client
+
+logger = logging.getLogger(__name__)
 
 
 class AIModel(BaseModel):
@@ -144,34 +148,60 @@ class AIModel(BaseModel):
 
         plugin_model_manager = PluginModelClient()
         cache_key = f"{self.tenant_id}:{self.plugin_id}:{self.provider_name}:{self.model_type.value}:{model}"
-        # sort credentials
         sorted_credentials = sorted(credentials.items()) if credentials else []
         cache_key += ":".join([hashlib.md5(f"{k}:{v}".encode()).hexdigest() for k, v in sorted_credentials])
 
+        cached_schema_json = None
         try:
-            contexts.plugin_model_schemas.get()
-        except LookupError:
-            contexts.plugin_model_schemas.set({})
-            contexts.plugin_model_schema_lock.set(Lock())
-
-        with contexts.plugin_model_schema_lock.get():
-            if cache_key in contexts.plugin_model_schemas.get():
-                return contexts.plugin_model_schemas.get()[cache_key]
-
-            schema = plugin_model_manager.get_model_schema(
-                tenant_id=self.tenant_id,
-                user_id="unknown",
-                plugin_id=self.plugin_id,
-                provider=self.provider_name,
-                model_type=self.model_type.value,
-                model=model,
-                credentials=credentials or {},
+            cached_schema_json = redis_client.get(cache_key)
+        except (RedisError, RuntimeError) as exc:
+            logger.warning(
+                "Failed to read plugin model schema cache for model %s: %s",
+                model,
+                str(exc),
+                exc_info=True,
             )
+        if cached_schema_json:
+            try:
+                return AIModelEntity.model_validate_json(cached_schema_json)
+            except ValidationError:
+                logger.warning(
+                    "Failed to validate cached plugin model schema for model %s",
+                    model,
+                    exc_info=True,
+                )
+                try:
+                    redis_client.delete(cache_key)
+                except (RedisError, RuntimeError) as exc:
+                    logger.warning(
+                        "Failed to delete invalid plugin model schema cache for model %s: %s",
+                        model,
+                        str(exc),
+                        exc_info=True,
+                    )
 
-            if schema:
-                contexts.plugin_model_schemas.get()[cache_key] = schema
+        schema = plugin_model_manager.get_model_schema(
+            tenant_id=self.tenant_id,
+            user_id="unknown",
+            plugin_id=self.plugin_id,
+            provider=self.provider_name,
+            model_type=self.model_type.value,
+            model=model,
+            credentials=credentials or {},
+        )
 
-            return schema
+        if schema:
+            try:
+                redis_client.setex(cache_key, dify_config.PLUGIN_MODEL_SCHEMA_CACHE_TTL, schema.model_dump_json())
+            except (RedisError, RuntimeError) as exc:
+                logger.warning(
+                    "Failed to write plugin model schema cache for model %s: %s",
+                    model,
+                    str(exc),
+                    exc_info=True,
+                )
+
+        return schema
 
     def get_customizable_model_schema_from_credentials(self, model: str, credentials: dict) -> AIModelEntity | None:
         """

api/core/model_runtime/model_providers/model_provider_factory.py

@@ -5,7 +5,11 @@ import logging
 from collections.abc import Sequence
 from threading import Lock
 
+from pydantic import ValidationError
+from redis import RedisError
+
 import contexts
+from configs import dify_config
 from core.model_runtime.entities.model_entities import AIModelEntity, ModelType
 from core.model_runtime.entities.provider_entities import ProviderConfig, ProviderEntity, SimpleProviderEntity
 from core.model_runtime.model_providers.__base.ai_model import AIModel
@@ -18,6 +22,7 @@ from core.model_runtime.model_providers.__base.tts_model import TTSModel
 from core.model_runtime.schema_validators.model_credential_schema_validator import ModelCredentialSchemaValidator
 from core.model_runtime.schema_validators.provider_credential_schema_validator import ProviderCredentialSchemaValidator
 from core.plugin.entities.plugin_daemon import PluginModelProviderEntity
+from extensions.ext_redis import redis_client
 from models.provider_ids import ModelProviderID
 
 logger = logging.getLogger(__name__)
@@ -175,34 +180,60 @@ class ModelProviderFactory:
         """
         plugin_id, provider_name = self.get_plugin_id_and_provider_name_from_provider(provider)
         cache_key = f"{self.tenant_id}:{plugin_id}:{provider_name}:{model_type.value}:{model}"
-        # sort credentials
         sorted_credentials = sorted(credentials.items()) if credentials else []
         cache_key += ":".join([hashlib.md5(f"{k}:{v}".encode()).hexdigest() for k, v in sorted_credentials])
 
+        cached_schema_json = None
         try:
-            contexts.plugin_model_schemas.get()
-        except LookupError:
-            contexts.plugin_model_schemas.set({})
-            contexts.plugin_model_schema_lock.set(Lock())
-
-        with contexts.plugin_model_schema_lock.get():
-            if cache_key in contexts.plugin_model_schemas.get():
-                return contexts.plugin_model_schemas.get()[cache_key]
-
-            schema = self.plugin_model_manager.get_model_schema(
-                tenant_id=self.tenant_id,
-                user_id="unknown",
-                plugin_id=plugin_id,
-                provider=provider_name,
-                model_type=model_type.value,
-                model=model,
-                credentials=credentials or {},
+            cached_schema_json = redis_client.get(cache_key)
+        except (RedisError, RuntimeError) as exc:
+            logger.warning(
+                "Failed to read plugin model schema cache for model %s: %s",
+                model,
+                str(exc),
+                exc_info=True,
             )
+        if cached_schema_json:
+            try:
+                return AIModelEntity.model_validate_json(cached_schema_json)
+            except ValidationError:
+                logger.warning(
+                    "Failed to validate cached plugin model schema for model %s",
+                    model,
+                    exc_info=True,
+                )
+                try:
+                    redis_client.delete(cache_key)
+                except (RedisError, RuntimeError) as exc:
+                    logger.warning(
+                        "Failed to delete invalid plugin model schema cache for model %s: %s",
+                        model,
+                        str(exc),
+                        exc_info=True,
+                    )
 
-            if schema:
-                contexts.plugin_model_schemas.get()[cache_key] = schema
+        schema = self.plugin_model_manager.get_model_schema(
+            tenant_id=self.tenant_id,
+            user_id="unknown",
+            plugin_id=plugin_id,
+            provider=provider_name,
+            model_type=model_type.value,
+            model=model,
+            credentials=credentials or {},
+        )
 
-            return schema
+        if schema:
+            try:
+                redis_client.setex(cache_key, dify_config.PLUGIN_MODEL_SCHEMA_CACHE_TTL, schema.model_dump_json())
+            except (RedisError, RuntimeError) as exc:
+                logger.warning(
+                    "Failed to write plugin model schema cache for model %s: %s",
+                    model,
+                    str(exc),
+                    exc_info=True,
+                )
+
+        return schema
 
     def get_models(
         self,

api/core/rag/datasource/retrieval_service.py

@@ -24,7 +24,13 @@ from core.rag.rerank.rerank_type import RerankMode
 from core.rag.retrieval.retrieval_methods import RetrievalMethod
 from core.tools.signature import sign_upload_file
 from extensions.ext_database import db
-from models.dataset import ChildChunk, Dataset, DocumentSegment, SegmentAttachmentBinding
+from models.dataset import (
+    ChildChunk,
+    Dataset,
+    DocumentSegment,
+    DocumentSegmentSummary,
+    SegmentAttachmentBinding,
+)
 from models.dataset import Document as DatasetDocument
 from models.model import UploadFile
 from services.external_knowledge_service import ExternalDatasetService
@@ -389,15 +395,15 @@ class RetrievalService:
                 .all()
             }
 
-            records = []
-            include_segment_ids = set()
-            segment_child_map = {}
-
             valid_dataset_documents = {}
             image_doc_ids: list[Any] = []
             child_index_node_ids = []
             index_node_ids = []
             doc_to_document_map = {}
+            summary_segment_ids = set()  # Track segments retrieved via summary
+            summary_score_map: dict[str, float] = {}  # Map original_chunk_id to summary score
+
+            # First pass: collect all document IDs and identify summary documents
             for document in documents:
                 document_id = document.metadata.get("document_id")
                 if document_id not in dataset_documents:
@@ -408,16 +414,39 @@ class RetrievalService:
                     continue
                 valid_dataset_documents[document_id] = dataset_document
 
+                doc_id = document.metadata.get("doc_id") or ""
+                doc_to_document_map[doc_id] = document
+
+                # Check if this is a summary document
+                is_summary = document.metadata.get("is_summary", False)
+                if is_summary:
+                    # For summary documents, find the original chunk via original_chunk_id
+                    original_chunk_id = document.metadata.get("original_chunk_id")
+                    if original_chunk_id:
+                        summary_segment_ids.add(original_chunk_id)
+                        # Save summary's score for later use
+                        summary_score = document.metadata.get("score")
+                        if summary_score is not None:
+                            try:
+                                summary_score_float = float(summary_score)
+                                # If the same segment has multiple summary hits, take the highest score
+                                if original_chunk_id not in summary_score_map:
+                                    summary_score_map[original_chunk_id] = summary_score_float
+                                else:
+                                    summary_score_map[original_chunk_id] = max(
+                                        summary_score_map[original_chunk_id], summary_score_float
+                                    )
+                            except (ValueError, TypeError):
+                                # Skip invalid score values
+                                pass
+                    continue  # Skip adding to other lists for summary documents
+
                 if dataset_document.doc_form == IndexStructureType.PARENT_CHILD_INDEX:
-                    doc_id = document.metadata.get("doc_id") or ""
-                    doc_to_document_map[doc_id] = document
                     if document.metadata.get("doc_type") == DocType.IMAGE:
                         image_doc_ids.append(doc_id)
                     else:
                         child_index_node_ids.append(doc_id)
                 else:
-                    doc_id = document.metadata.get("doc_id") or ""
-                    doc_to_document_map[doc_id] = document
                     if document.metadata.get("doc_type") == DocType.IMAGE:
                         image_doc_ids.append(doc_id)
                     else:
@@ -433,6 +462,7 @@ class RetrievalService:
             attachment_map: dict[str, list[dict[str, Any]]] = {}
             child_chunk_map: dict[str, list[ChildChunk]] = {}
             doc_segment_map: dict[str, list[str]] = {}
+            segment_summary_map: dict[str, str] = {}  # Map segment_id to summary content
 
             with session_factory.create_session() as session:
                 attachments = cls.get_segment_attachment_infos(image_doc_ids, session)
@@ -447,6 +477,7 @@ class RetrievalService:
                         doc_segment_map[attachment["segment_id"]].append(attachment["attachment_id"])
                     else:
                         doc_segment_map[attachment["segment_id"]] = [attachment["attachment_id"]]
+
                 child_chunk_stmt = select(ChildChunk).where(ChildChunk.index_node_id.in_(child_index_node_ids))
                 child_index_nodes = session.execute(child_chunk_stmt).scalars().all()
 
@@ -470,6 +501,7 @@ class RetrievalService:
                     index_node_segments = session.execute(document_segment_stmt).scalars().all()  # type: ignore
                     for index_node_segment in index_node_segments:
                         doc_segment_map[index_node_segment.id] = [index_node_segment.index_node_id]
+
                 if segment_ids:
                     document_segment_stmt = select(DocumentSegment).where(
                         DocumentSegment.enabled == True,
@@ -481,6 +513,40 @@ class RetrievalService:
                 if index_node_segments:
                     segments.extend(index_node_segments)
 
+                # Handle summary documents: query segments by original_chunk_id
+                if summary_segment_ids:
+                    summary_segment_ids_list = list(summary_segment_ids)
+                    summary_segment_stmt = select(DocumentSegment).where(
+                        DocumentSegment.enabled == True,
+                        DocumentSegment.status == "completed",
+                        DocumentSegment.id.in_(summary_segment_ids_list),
+                    )
+                    summary_segments = session.execute(summary_segment_stmt).scalars().all()  # type: ignore
+                    segments.extend(summary_segments)
+                    # Add summary segment IDs to segment_ids for summary query
+                    for seg in summary_segments:
+                        if seg.id not in segment_ids:
+                            segment_ids.append(seg.id)
+
+                # Batch query summaries for segments retrieved via summary (only enabled summaries)
+                if summary_segment_ids:
+                    summaries = (
+                        session.query(DocumentSegmentSummary)
+                        .filter(
+                            DocumentSegmentSummary.chunk_id.in_(list(summary_segment_ids)),
+                            DocumentSegmentSummary.status == "completed",
+                            DocumentSegmentSummary.enabled == True,  # Only retrieve enabled summaries
+                        )
+                        .all()
+                    )
+                    for summary in summaries:
+                        if summary.summary_content:
+                            segment_summary_map[summary.chunk_id] = summary.summary_content
+
+            include_segment_ids = set()
+            segment_child_map: dict[str, dict[str, Any]] = {}
+            records: list[dict[str, Any]] = []
+
             for segment in segments:
                 child_chunks: list[ChildChunk] = child_chunk_map.get(segment.id, [])
                 attachment_infos: list[dict[str, Any]] = attachment_map.get(segment.id, [])
@@ -489,30 +555,44 @@ class RetrievalService:
                 if ds_dataset_document and ds_dataset_document.doc_form == IndexStructureType.PARENT_CHILD_INDEX:
                     if segment.id not in include_segment_ids:
                         include_segment_ids.add(segment.id)
+                        # Check if this segment was retrieved via summary
+                        # Use summary score as base score if available, otherwise 0.0
+                        max_score = summary_score_map.get(segment.id, 0.0)
+
                         if child_chunks or attachment_infos:
                             child_chunk_details = []
-                            max_score = 0.0
                             for child_chunk in child_chunks:
-                                document = doc_to_document_map[child_chunk.index_node_id]
+                                child_document: Document | None = doc_to_document_map.get(child_chunk.index_node_id)
+                                if child_document:
+                                    child_score = child_document.metadata.get("score", 0.0)
+                                else:
+                                    child_score = 0.0
                                 child_chunk_detail = {
                                     "id": child_chunk.id,
                                     "content": child_chunk.content,
                                     "position": child_chunk.position,
-                                    "score": document.metadata.get("score", 0.0) if document else 0.0,
+                                    "score": child_score,
                                 }
                                 child_chunk_details.append(child_chunk_detail)
-                                max_score = max(max_score, document.metadata.get("score", 0.0) if document else 0.0)
+                                max_score = max(max_score, child_score)
                             for attachment_info in attachment_infos:
-                                file_document = doc_to_document_map[attachment_info["id"]]
-                                max_score = max(
-                                    max_score, file_document.metadata.get("score", 0.0) if file_document else 0.0
-                                )
+                                file_document = doc_to_document_map.get(attachment_info["id"])
+                                if file_document:
+                                    max_score = max(max_score, file_document.metadata.get("score", 0.0))
 
                             map_detail = {
                                 "max_score": max_score,
                                 "child_chunks": child_chunk_details,
                             }
                             segment_child_map[segment.id] = map_detail
+                        else:
+                            # No child chunks or attachments, use summary score if available
+                            summary_score = summary_score_map.get(segment.id)
+                            if summary_score is not None:
+                                segment_child_map[segment.id] = {
+                                    "max_score": summary_score,
+                                    "child_chunks": [],
+                                }
                         record: dict[str, Any] = {
                             "segment": segment,
                         }
@@ -520,14 +600,23 @@ class RetrievalService:
                 else:
                     if segment.id not in include_segment_ids:
                         include_segment_ids.add(segment.id)
-                        max_score = 0.0
-                        segment_document = doc_to_document_map.get(segment.index_node_id)
-                        if segment_document:
-                            max_score = max(max_score, segment_document.metadata.get("score", 0.0))
+
+                        # Check if this segment was retrieved via summary
+                        # Use summary score if available (summary retrieval takes priority)
+                        max_score = summary_score_map.get(segment.id, 0.0)
+
+                        # If not retrieved via summary, use original segment's score
+                        if segment.id not in summary_score_map:
+                            segment_document = doc_to_document_map.get(segment.index_node_id)
+                            if segment_document:
+                                max_score = max(max_score, segment_document.metadata.get("score", 0.0))
+
+                        # Also consider attachment scores
                         for attachment_info in attachment_infos:
                             file_doc = doc_to_document_map.get(attachment_info["id"])
                             if file_doc:
                                 max_score = max(max_score, file_doc.metadata.get("score", 0.0))
+
                         record = {
                             "segment": segment,
                             "score": max_score,
@@ -576,9 +665,16 @@ class RetrievalService:
                     else None
                 )
 
+                # Extract summary if this segment was retrieved via summary
+                summary_content = segment_summary_map.get(segment.id)
+
                 # Create RetrievalSegments object
                 retrieval_segment = RetrievalSegments(
-                    segment=segment, child_chunks=child_chunks_list, score=score, files=files
+                    segment=segment,
+                    child_chunks=child_chunks_list,
+                    score=score,
+                    files=files,
+                    summary=summary_content,
                 )
                 result.append(retrieval_segment)
 

api/core/rag/embedding/retrieval.py

@@ -20,3 +20,4 @@ class RetrievalSegments(BaseModel):
     child_chunks: list[RetrievalChildChunk] | None = None
     score: float | None = None
     files: list[dict[str, str | int]] | None = None
+    summary: str | None = None  # Summary content if retrieved via summary index

api/core/rag/entities/citation_metadata.py

@@ -22,3 +22,4 @@ class RetrievalSourceMetadata(BaseModel):
     doc_metadata: dict[str, Any] | None = None
     title: str | None = None
     files: list[dict[str, Any]] | None = None
+    summary: str | None = None

api/core/rag/extractor/word_extractor.py

@@ -1,4 +1,7 @@
-"""Abstract interface for document loader implementations."""
+"""Word (.docx) document extractor used for RAG ingestion.
+
+Supports local file paths and remote URLs (downloaded via `core.helper.ssrf_proxy`).
+"""
 
 import logging
 import mimetypes
@@ -8,7 +11,6 @@ import tempfile
 import uuid
 from urllib.parse import urlparse
 
-import httpx
 from docx import Document as DocxDocument
 from docx.oxml.ns import qn
 from docx.text.run import Run
@@ -44,7 +46,7 @@ class WordExtractor(BaseExtractor):
 
         # If the file is a web path, download it to a temporary file, and use that
         if not os.path.isfile(self.file_path) and self._is_valid_url(self.file_path):
-            response = httpx.get(self.file_path, timeout=None)
+            response = ssrf_proxy.get(self.file_path)
 
             if response.status_code != 200:
                 response.close()
@@ -55,6 +57,7 @@ class WordExtractor(BaseExtractor):
             self.temp_file = tempfile.NamedTemporaryFile()  # noqa SIM115
             try:
                 self.temp_file.write(response.content)
+                self.temp_file.flush()
             finally:
                 response.close()
             self.file_path = self.temp_file.name

api/core/rag/index_processor/index_processor_base.py

@@ -13,6 +13,7 @@ from urllib.parse import unquote, urlparse
 import httpx
 
 from configs import dify_config
+from core.entities.knowledge_entities import PreviewDetail
 from core.helper import ssrf_proxy
 from core.rag.extractor.entity.extract_setting import ExtractSetting
 from core.rag.index_processor.constant.doc_type import DocType
@@ -45,6 +46,17 @@ class BaseIndexProcessor(ABC):
     def transform(self, documents: list[Document], current_user: Account | None = None, **kwargs) -> list[Document]:
         raise NotImplementedError
 
+    @abstractmethod
+    def generate_summary_preview(
+        self, tenant_id: str, preview_texts: list[PreviewDetail], summary_index_setting: dict
+    ) -> list[PreviewDetail]:
+        """
+        For each segment in preview_texts, generate a summary using LLM and attach it to the segment.
+        The summary can be stored in a new attribute, e.g., summary.
+        This method should be implemented by subclasses.
+        """
+        raise NotImplementedError
+
     @abstractmethod
     def load(
         self,

api/core/rag/index_processor/processor/paragraph_index_processor.py

@@ -1,9 +1,27 @@
 """Paragraph index processor."""
 
+import logging
+import re
 import uuid
 from collections.abc import Mapping
-from typing import Any
+from typing import Any, cast
 
+logger = logging.getLogger(__name__)
+
+from core.entities.knowledge_entities import PreviewDetail
+from core.file import File, FileTransferMethod, FileType, file_manager
+from core.llm_generator.prompts import DEFAULT_GENERATOR_SUMMARY_PROMPT
+from core.model_manager import ModelInstance
+from core.model_runtime.entities.llm_entities import LLMResult, LLMUsage
+from core.model_runtime.entities.message_entities import (
+    ImagePromptMessageContent,
+    PromptMessage,
+    PromptMessageContentUnionTypes,
+    TextPromptMessageContent,
+    UserPromptMessage,
+)
+from core.model_runtime.entities.model_entities import ModelFeature, ModelType
+from core.provider_manager import ProviderManager
 from core.rag.cleaner.clean_processor import CleanProcessor
 from core.rag.datasource.keyword.keyword_factory import Keyword
 from core.rag.datasource.retrieval_service import RetrievalService
@@ -17,12 +35,17 @@ from core.rag.index_processor.index_processor_base import BaseIndexProcessor
 from core.rag.models.document import AttachmentDocument, Document, MultimodalGeneralStructureChunk
 from core.rag.retrieval.retrieval_methods import RetrievalMethod
 from core.tools.utils.text_processing_utils import remove_leading_symbols
+from core.workflow.nodes.llm import llm_utils
+from extensions.ext_database import db
+from factories.file_factory import build_from_mapping
 from libs import helper
+from models import UploadFile
 from models.account import Account
-from models.dataset import Dataset, DatasetProcessRule
+from models.dataset import Dataset, DatasetProcessRule, DocumentSegment, SegmentAttachmentBinding
 from models.dataset import Document as DatasetDocument
 from services.account_service import AccountService
 from services.entities.knowledge_entities.knowledge_entities import Rule
+from services.summary_index_service import SummaryIndexService
 
 
 class ParagraphIndexProcessor(BaseIndexProcessor):
@@ -108,6 +131,29 @@ class ParagraphIndexProcessor(BaseIndexProcessor):
                 keyword.add_texts(documents)
 
     def clean(self, dataset: Dataset, node_ids: list[str] | None, with_keywords: bool = True, **kwargs):
+        # Note: Summary indexes are now disabled (not deleted) when segments are disabled.
+        # This method is called for actual deletion scenarios (e.g., when segment is deleted).
+        # For disable operations, disable_summaries_for_segments is called directly in the task.
+        # Only delete summaries if explicitly requested (e.g., when segment is actually deleted)
+        delete_summaries = kwargs.get("delete_summaries", False)
+        if delete_summaries:
+            if node_ids:
+                # Find segments by index_node_id
+                segments = (
+                    db.session.query(DocumentSegment)
+                    .filter(
+                        DocumentSegment.dataset_id == dataset.id,
+                        DocumentSegment.index_node_id.in_(node_ids),
+                    )
+                    .all()
+                )
+                segment_ids = [segment.id for segment in segments]
+                if segment_ids:
+                    SummaryIndexService.delete_summaries_for_segments(dataset, segment_ids)
+            else:
+                # Delete all summaries for the dataset
+                SummaryIndexService.delete_summaries_for_segments(dataset, None)
+
         if dataset.indexing_technique == "high_quality":
             vector = Vector(dataset)
             if node_ids:
@@ -227,3 +273,322 @@ class ParagraphIndexProcessor(BaseIndexProcessor):
             }
         else:
             raise ValueError("Chunks is not a list")
+
+    def generate_summary_preview(
+        self, tenant_id: str, preview_texts: list[PreviewDetail], summary_index_setting: dict
+    ) -> list[PreviewDetail]:
+        """
+        For each segment, concurrently call generate_summary to generate a summary
+        and write it to the summary attribute of PreviewDetail.
+        In preview mode (indexing-estimate), if any summary generation fails, the method will raise an exception.
+        """
+        import concurrent.futures
+
+        from flask import current_app
+
+        # Capture Flask app context for worker threads
+        flask_app = None
+        try:
+            flask_app = current_app._get_current_object()  # type: ignore
+        except RuntimeError:
+            logger.warning("No Flask application context available, summary generation may fail")
+
+        def process(preview: PreviewDetail) -> None:
+            """Generate summary for a single preview item."""
+            if flask_app:
+                # Ensure Flask app context in worker thread
+                with flask_app.app_context():
+                    summary, _ = self.generate_summary(tenant_id, preview.content, summary_index_setting)
+                    preview.summary = summary
+            else:
+                # Fallback: try without app context (may fail)
+                summary, _ = self.generate_summary(tenant_id, preview.content, summary_index_setting)
+                preview.summary = summary
+
+        # Generate summaries concurrently using ThreadPoolExecutor
+        # Set a reasonable timeout to prevent hanging (60 seconds per chunk, max 5 minutes total)
+        timeout_seconds = min(300, 60 * len(preview_texts))
+        errors: list[Exception] = []
+
+        with concurrent.futures.ThreadPoolExecutor(max_workers=min(10, len(preview_texts))) as executor:
+            futures = [executor.submit(process, preview) for preview in preview_texts]
+            # Wait for all tasks to complete with timeout
+            done, not_done = concurrent.futures.wait(futures, timeout=timeout_seconds)
+
+            # Cancel tasks that didn't complete in time
+            if not_done:
+                timeout_error_msg = (
+                    f"Summary generation timeout: {len(not_done)} chunks did not complete within {timeout_seconds}s"
+                )
+                logger.warning("%s. Cancelling remaining tasks...", timeout_error_msg)
+                # In preview mode, timeout is also an error
+                errors.append(TimeoutError(timeout_error_msg))
+                for future in not_done:
+                    future.cancel()
+                # Wait a bit for cancellation to take effect
+                concurrent.futures.wait(not_done, timeout=5)
+
+            # Collect exceptions from completed futures
+            for future in done:
+                try:
+                    future.result()  # This will raise any exception that occurred
+                except Exception as e:
+                    logger.exception("Error in summary generation future")
+                    errors.append(e)
+
+        # In preview mode (indexing-estimate), if there are any errors, fail the request
+        if errors:
+            error_messages = [str(e) for e in errors]
+            error_summary = (
+                f"Failed to generate summaries for {len(errors)} chunk(s). "
+                f"Errors: {'; '.join(error_messages[:3])}"  # Show first 3 errors
+            )
+            if len(errors) > 3:
+                error_summary += f" (and {len(errors) - 3} more)"
+            logger.error("Summary generation failed in preview mode: %s", error_summary)
+            raise ValueError(error_summary)
+
+        return preview_texts
+
+    @staticmethod
+    def generate_summary(
+        tenant_id: str,
+        text: str,
+        summary_index_setting: dict | None = None,
+        segment_id: str | None = None,
+    ) -> tuple[str, LLMUsage]:
+        """
+        Generate summary for the given text using ModelInstance.invoke_llm and the default or custom summary prompt,
+        and supports vision models by including images from the segment attachments or text content.
+
+        Args:
+            tenant_id: Tenant ID
+            text: Text content to summarize
+            summary_index_setting: Summary index configuration
+            segment_id: Optional segment ID to fetch attachments from SegmentAttachmentBinding table
+
+        Returns:
+            Tuple of (summary_content, llm_usage) where llm_usage is LLMUsage object
+        """
+        if not summary_index_setting or not summary_index_setting.get("enable"):
+            raise ValueError("summary_index_setting is required and must be enabled to generate summary.")
+
+        model_name = summary_index_setting.get("model_name")
+        model_provider_name = summary_index_setting.get("model_provider_name")
+        summary_prompt = summary_index_setting.get("summary_prompt")
+
+        if not model_name or not model_provider_name:
+            raise ValueError("model_name and model_provider_name are required in summary_index_setting")
+
+        # Import default summary prompt
+        if not summary_prompt:
+            summary_prompt = DEFAULT_GENERATOR_SUMMARY_PROMPT
+
+        provider_manager = ProviderManager()
+        provider_model_bundle = provider_manager.get_provider_model_bundle(
+            tenant_id, model_provider_name, ModelType.LLM
+        )
+        model_instance = ModelInstance(provider_model_bundle, model_name)
+
+        # Get model schema to check if vision is supported
+        model_schema = model_instance.model_type_instance.get_model_schema(model_name, model_instance.credentials)
+        supports_vision = model_schema and model_schema.features and ModelFeature.VISION in model_schema.features
+
+        # Extract images if model supports vision
+        image_files = []
+        if supports_vision:
+            # First, try to get images from SegmentAttachmentBinding (preferred method)
+            if segment_id:
+                image_files = ParagraphIndexProcessor._extract_images_from_segment_attachments(tenant_id, segment_id)
+
+            # If no images from attachments, fall back to extracting from text
+            if not image_files:
+                image_files = ParagraphIndexProcessor._extract_images_from_text(tenant_id, text)
+
+        # Build prompt messages
+        prompt_messages = []
+
+        if image_files:
+            # If we have images, create a UserPromptMessage with both text and images
+            prompt_message_contents: list[PromptMessageContentUnionTypes] = []
+
+            # Add images first
+            for file in image_files:
+                try:
+                    file_content = file_manager.to_prompt_message_content(
+                        file, image_detail_config=ImagePromptMessageContent.DETAIL.LOW
+                    )
+                    prompt_message_contents.append(file_content)
+                except Exception as e:
+                    logger.warning("Failed to convert image file to prompt message content: %s", str(e))
+                    continue
+
+            # Add text content
+            if prompt_message_contents:  # Only add text if we successfully added images
+                prompt_message_contents.append(TextPromptMessageContent(data=f"{summary_prompt}\n{text}"))
+                prompt_messages.append(UserPromptMessage(content=prompt_message_contents))
+            else:
+                # If image conversion failed, fall back to text-only
+                prompt = f"{summary_prompt}\n{text}"
+                prompt_messages.append(UserPromptMessage(content=prompt))
+        else:
+            # No images, use simple text prompt
+            prompt = f"{summary_prompt}\n{text}"
+            prompt_messages.append(UserPromptMessage(content=prompt))
+
+        result = model_instance.invoke_llm(
+            prompt_messages=cast(list[PromptMessage], prompt_messages), model_parameters={}, stream=False
+        )
+
+        # Type assertion: when stream=False, invoke_llm returns LLMResult, not Generator
+        if not isinstance(result, LLMResult):
+            raise ValueError("Expected LLMResult when stream=False")
+
+        summary_content = getattr(result.message, "content", "")
+        usage = result.usage
+
+        # Deduct quota for summary generation (same as workflow nodes)
+        try:
+            llm_utils.deduct_llm_quota(tenant_id=tenant_id, model_instance=model_instance, usage=usage)
+        except Exception as e:
+            # Log but don't fail summary generation if quota deduction fails
+            logger.warning("Failed to deduct quota for summary generation: %s", str(e))
+
+        return summary_content, usage
+
+    @staticmethod
+    def _extract_images_from_text(tenant_id: str, text: str) -> list[File]:
+        """
+        Extract images from markdown text and convert them to File objects.
+
+        Args:
+            tenant_id: Tenant ID
+            text: Text content that may contain markdown image links
+
+        Returns:
+            List of File objects representing images found in the text
+        """
+        # Extract markdown images using regex pattern
+        pattern = r"!\[.*?\]\((.*?)\)"
+        images = re.findall(pattern, text)
+
+        if not images:
+            return []
+
+        upload_file_id_list = []
+
+        for image in images:
+            # For data before v0.10.0
+            pattern = r"/files/([a-f0-9\-]+)/image-preview(?:\?.*?)?"
+            match = re.search(pattern, image)
+            if match:
+                upload_file_id = match.group(1)
+                upload_file_id_list.append(upload_file_id)
+                continue
+
+            # For data after v0.10.0
+            pattern = r"/files/([a-f0-9\-]+)/file-preview(?:\?.*?)?"
+            match = re.search(pattern, image)
+            if match:
+                upload_file_id = match.group(1)
+                upload_file_id_list.append(upload_file_id)
+                continue
+
+            # For tools directory - direct file formats (e.g., .png, .jpg, etc.)
+            pattern = r"/files/tools/([a-f0-9\-]+)\.([a-zA-Z0-9]+)(?:\?[^\s\)\"\']*)?"
+            match = re.search(pattern, image)
+            if match:
+                # Tool files are handled differently, skip for now
+                continue
+
+        if not upload_file_id_list:
+            return []
+
+        # Get unique IDs for database query
+        unique_upload_file_ids = list(set(upload_file_id_list))
+        upload_files = (
+            db.session.query(UploadFile)
+            .where(UploadFile.id.in_(unique_upload_file_ids), UploadFile.tenant_id == tenant_id)
+            .all()
+        )
+
+        # Create File objects from UploadFile records
+        file_objects = []
+        for upload_file in upload_files:
+            # Only process image files
+            if not upload_file.mime_type or "image" not in upload_file.mime_type:
+                continue
+
+            mapping = {
+                "upload_file_id": upload_file.id,
+                "transfer_method": FileTransferMethod.LOCAL_FILE.value,
+                "type": FileType.IMAGE.value,
+            }
+
+            try:
+                file_obj = build_from_mapping(
+                    mapping=mapping,
+                    tenant_id=tenant_id,
+                )
+                file_objects.append(file_obj)
+            except Exception as e:
+                logger.warning("Failed to create File object from UploadFile %s: %s", upload_file.id, str(e))
+                continue
+
+        return file_objects
+
+    @staticmethod
+    def _extract_images_from_segment_attachments(tenant_id: str, segment_id: str) -> list[File]:
+        """
+        Extract images from SegmentAttachmentBinding table (preferred method).
+        This matches how DatasetRetrieval gets segment attachments.
+
+        Args:
+            tenant_id: Tenant ID
+            segment_id: Segment ID to fetch attachments for
+
+        Returns:
+            List of File objects representing images found in segment attachments
+        """
+        from sqlalchemy import select
+
+        # Query attachments from SegmentAttachmentBinding table
+        attachments_with_bindings = db.session.execute(
+            select(SegmentAttachmentBinding, UploadFile)
+            .join(UploadFile, UploadFile.id == SegmentAttachmentBinding.attachment_id)
+            .where(
+                SegmentAttachmentBinding.segment_id == segment_id,
+                SegmentAttachmentBinding.tenant_id == tenant_id,
+            )
+        ).all()
+
+        if not attachments_with_bindings:
+            return []
+
+        file_objects = []
+        for _, upload_file in attachments_with_bindings:
+            # Only process image files
+            if not upload_file.mime_type or "image" not in upload_file.mime_type:
+                continue
+
+            try:
+                # Create File object directly (similar to DatasetRetrieval)
+                file_obj = File(
+                    id=upload_file.id,
+                    filename=upload_file.name,
+                    extension="." + upload_file.extension,
+                    mime_type=upload_file.mime_type,
+                    tenant_id=tenant_id,
+                    type=FileType.IMAGE,
+                    transfer_method=FileTransferMethod.LOCAL_FILE,
+                    remote_url=upload_file.source_url,
+                    related_id=upload_file.id,
+                    size=upload_file.size,
+                    storage_key=upload_file.key,
+                )
+                file_objects.append(file_obj)
+            except Exception as e:
+                logger.warning("Failed to create File object from UploadFile %s: %s", upload_file.id, str(e))
+                continue
+
+        return file_objects

api/core/rag/index_processor/processor/parent_child_index_processor.py

@@ -1,11 +1,14 @@
 """Paragraph index processor."""
 
 import json
+import logging
 import uuid
 from collections.abc import Mapping
 from typing import Any
 
 from configs import dify_config
+from core.db.session_factory import session_factory
+from core.entities.knowledge_entities import PreviewDetail
 from core.model_manager import ModelInstance
 from core.rag.cleaner.clean_processor import CleanProcessor
 from core.rag.datasource.retrieval_service import RetrievalService
@@ -25,6 +28,9 @@ from models.dataset import ChildChunk, Dataset, DatasetProcessRule, DocumentSegm
 from models.dataset import Document as DatasetDocument
 from services.account_service import AccountService
 from services.entities.knowledge_entities.knowledge_entities import ParentMode, Rule
+from services.summary_index_service import SummaryIndexService
+
+logger = logging.getLogger(__name__)
 
 
 class ParentChildIndexProcessor(BaseIndexProcessor):
@@ -135,6 +141,30 @@ class ParentChildIndexProcessor(BaseIndexProcessor):
 
     def clean(self, dataset: Dataset, node_ids: list[str] | None, with_keywords: bool = True, **kwargs):
         # node_ids is segment's node_ids
+        # Note: Summary indexes are now disabled (not deleted) when segments are disabled.
+        # This method is called for actual deletion scenarios (e.g., when segment is deleted).
+        # For disable operations, disable_summaries_for_segments is called directly in the task.
+        # Only delete summaries if explicitly requested (e.g., when segment is actually deleted)
+        delete_summaries = kwargs.get("delete_summaries", False)
+        if delete_summaries:
+            if node_ids:
+                # Find segments by index_node_id
+                with session_factory.create_session() as session:
+                    segments = (
+                        session.query(DocumentSegment)
+                        .filter(
+                            DocumentSegment.dataset_id == dataset.id,
+                            DocumentSegment.index_node_id.in_(node_ids),
+                        )
+                        .all()
+                    )
+                    segment_ids = [segment.id for segment in segments]
+                    if segment_ids:
+                        SummaryIndexService.delete_summaries_for_segments(dataset, segment_ids)
+            else:
+                # Delete all summaries for the dataset
+                SummaryIndexService.delete_summaries_for_segments(dataset, None)
+
         if dataset.indexing_technique == "high_quality":
             delete_child_chunks = kwargs.get("delete_child_chunks") or False
             precomputed_child_node_ids = kwargs.get("precomputed_child_node_ids")
@@ -326,3 +356,91 @@ class ParentChildIndexProcessor(BaseIndexProcessor):
             "preview": preview,
             "total_segments": len(parent_childs.parent_child_chunks),
         }
+
+    def generate_summary_preview(
+        self, tenant_id: str, preview_texts: list[PreviewDetail], summary_index_setting: dict
+    ) -> list[PreviewDetail]:
+        """
+        For each parent chunk in preview_texts, concurrently call generate_summary to generate a summary
+        and write it to the summary attribute of PreviewDetail.
+        In preview mode (indexing-estimate), if any summary generation fails, the method will raise an exception.
+
+        Note: For parent-child structure, we only generate summaries for parent chunks.
+        """
+        import concurrent.futures
+
+        from flask import current_app
+
+        # Capture Flask app context for worker threads
+        flask_app = None
+        try:
+            flask_app = current_app._get_current_object()  # type: ignore
+        except RuntimeError:
+            logger.warning("No Flask application context available, summary generation may fail")
+
+        def process(preview: PreviewDetail) -> None:
+            """Generate summary for a single preview item (parent chunk)."""
+            from core.rag.index_processor.processor.paragraph_index_processor import ParagraphIndexProcessor
+
+            if flask_app:
+                # Ensure Flask app context in worker thread
+                with flask_app.app_context():
+                    summary, _ = ParagraphIndexProcessor.generate_summary(
+                        tenant_id=tenant_id,
+                        text=preview.content,
+                        summary_index_setting=summary_index_setting,
+                    )
+                    preview.summary = summary
+            else:
+                # Fallback: try without app context (may fail)
+                summary, _ = ParagraphIndexProcessor.generate_summary(
+                    tenant_id=tenant_id,
+                    text=preview.content,
+                    summary_index_setting=summary_index_setting,
+                )
+                preview.summary = summary
+
+        # Generate summaries concurrently using ThreadPoolExecutor
+        # Set a reasonable timeout to prevent hanging (60 seconds per chunk, max 5 minutes total)
+        timeout_seconds = min(300, 60 * len(preview_texts))
+        errors: list[Exception] = []
+
+        with concurrent.futures.ThreadPoolExecutor(max_workers=min(10, len(preview_texts))) as executor:
+            futures = [executor.submit(process, preview) for preview in preview_texts]
+            # Wait for all tasks to complete with timeout
+            done, not_done = concurrent.futures.wait(futures, timeout=timeout_seconds)
+
+            # Cancel tasks that didn't complete in time
+            if not_done:
+                timeout_error_msg = (
+                    f"Summary generation timeout: {len(not_done)} chunks did not complete within {timeout_seconds}s"
+                )
+                logger.warning("%s. Cancelling remaining tasks...", timeout_error_msg)
+                # In preview mode, timeout is also an error
+                errors.append(TimeoutError(timeout_error_msg))
+                for future in not_done:
+                    future.cancel()
+                # Wait a bit for cancellation to take effect
+                concurrent.futures.wait(not_done, timeout=5)
+
+            # Collect exceptions from completed futures
+            for future in done:
+                try:
+                    future.result()  # This will raise any exception that occurred
+                except Exception as e:
+                    logger.exception("Error in summary generation future")
+                    errors.append(e)
+
+        # In preview mode (indexing-estimate), if there are any errors, fail the request
+        if errors:
+            error_messages = [str(e) for e in errors]
+            error_summary = (
+                f"Failed to generate summaries for {len(errors)} chunk(s). "
+                f"Errors: {'; '.join(error_messages[:3])}"  # Show first 3 errors
+            )
+            if len(errors) > 3:
+                error_summary += f" (and {len(errors) - 3} more)"
+            logger.error("Summary generation failed in preview mode: %s", error_summary)
+            raise ValueError(error_summary)
+
+        return preview_texts

api/core/rag/index_processor/processor/qa_index_processor.py

@@ -11,6 +11,8 @@ import pandas as pd
 from flask import Flask, current_app
 from werkzeug.datastructures import FileStorage
 
+from core.db.session_factory import session_factory
+from core.entities.knowledge_entities import PreviewDetail
 from core.llm_generator.llm_generator import LLMGenerator
 from core.rag.cleaner.clean_processor import CleanProcessor
 from core.rag.datasource.retrieval_service import RetrievalService
@@ -25,9 +27,10 @@ from core.rag.retrieval.retrieval_methods import RetrievalMethod
 from core.tools.utils.text_processing_utils import remove_leading_symbols
 from libs import helper
 from models.account import Account
-from models.dataset import Dataset
+from models.dataset import Dataset, DocumentSegment
 from models.dataset import Document as DatasetDocument
 from services.entities.knowledge_entities.knowledge_entities import Rule
+from services.summary_index_service import SummaryIndexService
 
 logger = logging.getLogger(__name__)
 
@@ -144,6 +147,31 @@ class QAIndexProcessor(BaseIndexProcessor):
                 vector.create_multimodal(multimodal_documents)
 
     def clean(self, dataset: Dataset, node_ids: list[str] | None, with_keywords: bool = True, **kwargs):
+        # Note: Summary indexes are now disabled (not deleted) when segments are disabled.
+        # This method is called for actual deletion scenarios (e.g., when segment is deleted).
+        # For disable operations, disable_summaries_for_segments is called directly in the task.
+        # Note: qa_model doesn't generate summaries, but we clean them for completeness
+        # Only delete summaries if explicitly requested (e.g., when segment is actually deleted)
+        delete_summaries = kwargs.get("delete_summaries", False)
+        if delete_summaries:
+            if node_ids:
+                # Find segments by index_node_id
+                with session_factory.create_session() as session:
+                    segments = (
+                        session.query(DocumentSegment)
+                        .filter(
+                            DocumentSegment.dataset_id == dataset.id,
+                            DocumentSegment.index_node_id.in_(node_ids),
+                        )
+                        .all()
+                    )
+                    segment_ids = [segment.id for segment in segments]
+                    if segment_ids:
+                        SummaryIndexService.delete_summaries_for_segments(dataset, segment_ids)
+            else:
+                # Delete all summaries for the dataset
+                SummaryIndexService.delete_summaries_for_segments(dataset, None)
+
         vector = Vector(dataset)
         if node_ids:
             vector.delete_by_ids(node_ids)
@@ -212,6 +240,17 @@ class QAIndexProcessor(BaseIndexProcessor):
             "total_segments": len(qa_chunks.qa_chunks),
         }
 
+    def generate_summary_preview(
+        self, tenant_id: str, preview_texts: list[PreviewDetail], summary_index_setting: dict
+    ) -> list[PreviewDetail]:
+        """
+        QA model doesn't generate summaries, so this method returns preview_texts unchanged.
+
+        Note: QA model uses question-answer pairs, which don't require summary generation.
+        """
+        # QA model doesn't generate summaries, return as-is
+        return preview_texts
+
     def _format_qa_document(self, flask_app: Flask, tenant_id: str, document_node, all_qa_documents, document_language):
         format_documents = []
         if document_node.page_content is None or not document_node.page_content.strip():

api/core/rag/retrieval/dataset_retrieval.py

@@ -236,20 +236,24 @@ class DatasetRetrieval:
             if records:
                 for record in records:
                     segment = record.segment
+                    # Build content: if summary exists, add it before the segment content
                     if segment.answer:
-                        document_context_list.append(
-                            DocumentContext(
-                                content=f"question:{segment.get_sign_content()} answer:{segment.answer}",
-                                score=record.score,
-                            )
-                        )
+                        segment_content = f"question:{segment.get_sign_content()} answer:{segment.answer}"
                     else:
-                        document_context_list.append(
-                            DocumentContext(
-                                content=segment.get_sign_content(),
-                                score=record.score,
-                            )
+                        segment_content = segment.get_sign_content()
+
+                    # If summary exists, prepend it to the content
+                    if record.summary:
+                        final_content = f"{record.summary}\n{segment_content}"
+                    else:
+                        final_content = segment_content
+
+                    document_context_list.append(
+                        DocumentContext(
+                            content=final_content,
+                            score=record.score,
                         )
+                    )
                     if vision_enabled:
                         attachments_with_bindings = db.session.execute(
                             select(SegmentAttachmentBinding, UploadFile)
@@ -316,6 +320,9 @@ class DatasetRetrieval:
                                 source.content = f"question:{segment.content} \nanswer:{segment.answer}"
                             else:
                                 source.content = segment.content
+                            # Add summary if this segment was retrieved via summary
+                            if hasattr(record, "summary") and record.summary:
+                                source.summary = record.summary
                             retrieval_resource_list.append(source)
         if hit_callback and retrieval_resource_list:
             retrieval_resource_list = sorted(retrieval_resource_list, key=lambda x: x.score or 0.0, reverse=True)

api/core/tools/utils/dataset_retriever/dataset_retriever_tool.py

@@ -169,20 +169,24 @@ class DatasetRetrieverTool(DatasetRetrieverBaseTool):
                 if records:
                     for record in records:
                         segment = record.segment
+                        # Build content: if summary exists, add it before the segment content
                         if segment.answer:
-                            document_context_list.append(
-                                DocumentContext(
-                                    content=f"question:{segment.get_sign_content()} answer:{segment.answer}",
-                                    score=record.score,
-                                )
-                            )
+                            segment_content = f"question:{segment.get_sign_content()} answer:{segment.answer}"
                         else:
-                            document_context_list.append(
-                                DocumentContext(
-                                    content=segment.get_sign_content(),
-                                    score=record.score,
-                                )
+                            segment_content = segment.get_sign_content()
+
+                        # If summary exists, prepend it to the content
+                        if record.summary:
+                            final_content = f"{record.summary}\n{segment_content}"
+                        else:
+                            final_content = segment_content
+
+                        document_context_list.append(
+                            DocumentContext(
+                                content=final_content,
+                                score=record.score,
                             )
+                        )
 
                     if self.return_resource:
                         for record in records:
@@ -216,6 +220,9 @@ class DatasetRetrieverTool(DatasetRetrieverBaseTool):
                                     source.content = f"question:{segment.content} \nanswer:{segment.answer}"
                                 else:
                                     source.content = segment.content
+                                # Add summary if this segment was retrieved via summary
+                                if hasattr(record, "summary") and record.summary:
+                                    source.summary = record.summary
                                 retrieval_resource_list.append(source)
 
             if self.return_resource and retrieval_resource_list:

api/core/workflow/nodes/knowledge_index/entities.py

@@ -158,3 +158,5 @@ class KnowledgeIndexNodeData(BaseNodeData):
     type: str = "knowledge-index"
     chunk_structure: str
     index_chunk_variable_selector: list[str]
+    indexing_technique: str | None = None
+    summary_index_setting: dict | None = None

api/core/workflow/nodes/knowledge_index/knowledge_index_node.py

@@ -1,9 +1,11 @@
+import concurrent.futures
 import datetime
 import logging
 import time
 from collections.abc import Mapping
 from typing import Any
 
+from flask import current_app
 from sqlalchemy import func, select
 
 from core.app.entities.app_invoke_entities import InvokeFrom
@@ -16,7 +18,9 @@ from core.workflow.nodes.base.node import Node
 from core.workflow.nodes.base.template import Template
 from core.workflow.runtime import VariablePool
 from extensions.ext_database import db
-from models.dataset import Dataset, Document, DocumentSegment
+from models.dataset import Dataset, Document, DocumentSegment, DocumentSegmentSummary
+from services.summary_index_service import SummaryIndexService
+from tasks.generate_summary_index_task import generate_summary_index_task
 
 from .entities import KnowledgeIndexNodeData
 from .exc import (
@@ -67,7 +71,20 @@ class KnowledgeIndexNode(Node[KnowledgeIndexNodeData]):
         # index knowledge
         try:
             if is_preview:
-                outputs = self._get_preview_output(node_data.chunk_structure, chunks)
+                # Preview mode: generate summaries for chunks directly without saving to database
+                # Format preview and generate summaries on-the-fly
+                # Get indexing_technique and summary_index_setting from node_data (workflow graph config)
+                # or fallback to dataset if not available in node_data
+                indexing_technique = node_data.indexing_technique or dataset.indexing_technique
+                summary_index_setting = node_data.summary_index_setting or dataset.summary_index_setting
+
+                outputs = self._get_preview_output_with_summaries(
+                    node_data.chunk_structure,
+                    chunks,
+                    dataset=dataset,
+                    indexing_technique=indexing_technique,
+                    summary_index_setting=summary_index_setting,
+                )
                 return NodeRunResult(
                     status=WorkflowNodeExecutionStatus.SUCCEEDED,
                     inputs=variables,
@@ -148,6 +165,11 @@ class KnowledgeIndexNode(Node[KnowledgeIndexNodeData]):
             )
             .scalar()
         )
+        # Update need_summary based on dataset's summary_index_setting
+        if dataset.summary_index_setting and dataset.summary_index_setting.get("enable") is True:
+            document.need_summary = True
+        else:
+            document.need_summary = False
         db.session.add(document)
         # update document segment status
         db.session.query(DocumentSegment).where(
@@ -163,6 +185,9 @@ class KnowledgeIndexNode(Node[KnowledgeIndexNodeData]):
 
         db.session.commit()
 
+        # Generate summary index if enabled
+        self._handle_summary_index_generation(dataset, document, variable_pool)
+
         return {
             "dataset_id": ds_id_value,
             "dataset_name": dataset_name_value,
@@ -173,9 +198,304 @@ class KnowledgeIndexNode(Node[KnowledgeIndexNodeData]):
             "display_status": "completed",
         }
 
-    def _get_preview_output(self, chunk_structure: str, chunks: Any) -> Mapping[str, Any]:
+    def _handle_summary_index_generation(
+        self,
+        dataset: Dataset,
+        document: Document,
+        variable_pool: VariablePool,
+    ) -> None:
+        """
+        Handle summary index generation based on mode (debug/preview or production).
+
+        Args:
+            dataset: Dataset containing the document
+            document: Document to generate summaries for
+            variable_pool: Variable pool to check invoke_from
+        """
+        # Only generate summary index for high_quality indexing technique
+        if dataset.indexing_technique != "high_quality":
+            return
+
+        # Check if summary index is enabled
+        summary_index_setting = dataset.summary_index_setting
+        if not summary_index_setting or not summary_index_setting.get("enable"):
+            return
+
+        # Skip qa_model documents
+        if document.doc_form == "qa_model":
+            return
+
+        # Determine if in preview/debug mode
+        invoke_from = variable_pool.get(["sys", SystemVariableKey.INVOKE_FROM])
+        is_preview = invoke_from and invoke_from.value == InvokeFrom.DEBUGGER
+
+        if is_preview:
+            try:
+                # Query segments that need summary generation
+                query = db.session.query(DocumentSegment).filter_by(
+                    dataset_id=dataset.id,
+                    document_id=document.id,
+                    status="completed",
+                    enabled=True,
+                )
+                segments = query.all()
+
+                if not segments:
+                    logger.info("No segments found for document %s", document.id)
+                    return
+
+                # Filter segments based on mode
+                segments_to_process = []
+                for segment in segments:
+                    # Skip if summary already exists
+                    existing_summary = (
+                        db.session.query(DocumentSegmentSummary)
+                        .filter_by(chunk_id=segment.id, dataset_id=dataset.id, status="completed")
+                        .first()
+                    )
+                    if existing_summary:
+                        continue
+
+                    # For parent-child mode, all segments are parent chunks, so process all
+                    segments_to_process.append(segment)
+
+                if not segments_to_process:
+                    logger.info("No segments need summary generation for document %s", document.id)
+                    return
+
+                # Use ThreadPoolExecutor for concurrent generation
+                flask_app = current_app._get_current_object()  # type: ignore
+                max_workers = min(10, len(segments_to_process))  # Limit to 10 workers
+
+                def process_segment(segment: DocumentSegment) -> None:
+                    """Process a single segment in a thread with Flask app context."""
+                    with flask_app.app_context():
+                        try:
+                            SummaryIndexService.generate_and_vectorize_summary(segment, dataset, summary_index_setting)
+                        except Exception:
+                            logger.exception(
+                                "Failed to generate summary for segment %s",
+                                segment.id,
+                            )
+                            # Continue processing other segments
+
+                with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
+                    futures = [executor.submit(process_segment, segment) for segment in segments_to_process]
+                    # Wait for all tasks to complete
+                    concurrent.futures.wait(futures)
+
+                logger.info(
+                    "Successfully generated summary index for %s segments in document %s",
+                    len(segments_to_process),
+                    document.id,
+                )
+            except Exception:
+                logger.exception("Failed to generate summary index for document %s", document.id)
+                # Don't fail the entire indexing process if summary generation fails
+        else:
+            # Production mode: asynchronous generation
+            logger.info(
+                "Queuing summary index generation task for document %s (production mode)",
+                document.id,
+            )
+            try:
+                generate_summary_index_task.delay(dataset.id, document.id, None)
+                logger.info("Summary index generation task queued for document %s", document.id)
+            except Exception:
+                logger.exception(
+                    "Failed to queue summary index generation task for document %s",
+                    document.id,
+                )
+                # Don't fail the entire indexing process if task queuing fails
+
+    def _get_preview_output_with_summaries(
+        self,
+        chunk_structure: str,
+        chunks: Any,
+        dataset: Dataset,
+        indexing_technique: str | None = None,
+        summary_index_setting: dict | None = None,
+    ) -> Mapping[str, Any]:
+        """
+        Generate preview output with summaries for chunks in preview mode.
+        This method generates summaries on-the-fly without saving to database.
+
+        Args:
+            chunk_structure: Chunk structure type
+            chunks: Chunks to generate preview for
+            dataset: Dataset object (for tenant_id)
+            indexing_technique: Indexing technique from node config or dataset
+            summary_index_setting: Summary index setting from node config or dataset
+        """
         index_processor = IndexProcessorFactory(chunk_structure).init_index_processor()
-        return index_processor.format_preview(chunks)
+        preview_output = index_processor.format_preview(chunks)
+
+        # Check if summary index is enabled
+        if indexing_technique != "high_quality":
+            return preview_output
+
+        if not summary_index_setting or not summary_index_setting.get("enable"):
+            return preview_output
+
+        # Generate summaries for chunks
+        if "preview" in preview_output and isinstance(preview_output["preview"], list):
+            chunk_count = len(preview_output["preview"])
+            logger.info(
+                "Generating summaries for %s chunks in preview mode (dataset: %s)",
+                chunk_count,
+                dataset.id,
+            )
+            # Use ParagraphIndexProcessor's generate_summary method
+            from core.rag.index_processor.processor.paragraph_index_processor import ParagraphIndexProcessor
+
+            # Get Flask app for application context in worker threads
+            flask_app = None
+            try:
+                flask_app = current_app._get_current_object()  # type: ignore
+            except RuntimeError:
+                logger.warning("No Flask application context available, summary generation may fail")
+
+            def generate_summary_for_chunk(preview_item: dict) -> None:
+                """Generate summary for a single chunk."""
+                if "content" in preview_item:
+                    # Set Flask application context in worker thread
+                    if flask_app:
+                        with flask_app.app_context():
+                            summary, _ = ParagraphIndexProcessor.generate_summary(
+                                tenant_id=dataset.tenant_id,
+                                text=preview_item["content"],
+                                summary_index_setting=summary_index_setting,
+                            )
+                            if summary:
+                                preview_item["summary"] = summary
+                    else:
+                        # Fallback: try without app context (may fail)
+                        summary, _ = ParagraphIndexProcessor.generate_summary(
+                            tenant_id=dataset.tenant_id,
+                            text=preview_item["content"],
+                            summary_index_setting=summary_index_setting,
+                        )
+                        if summary:
+                            preview_item["summary"] = summary
+
+            # Generate summaries concurrently using ThreadPoolExecutor
+            # Set a reasonable timeout to prevent hanging (60 seconds per chunk, max 5 minutes total)
+            timeout_seconds = min(300, 60 * len(preview_output["preview"]))
+            errors: list[Exception] = []
+
+            with concurrent.futures.ThreadPoolExecutor(max_workers=min(10, len(preview_output["preview"]))) as executor:
+                futures = [
+                    executor.submit(generate_summary_for_chunk, preview_item)
+                    for preview_item in preview_output["preview"]
+                ]
+                # Wait for all tasks to complete with timeout
+                done, not_done = concurrent.futures.wait(futures, timeout=timeout_seconds)
+
+                # Cancel tasks that didn't complete in time
+                if not_done:
+                    timeout_error_msg = (
+                        f"Summary generation timeout: {len(not_done)} chunks did not complete within {timeout_seconds}s"
+                    )
+                    logger.warning("%s. Cancelling remaining tasks...", timeout_error_msg)
+                    # In preview mode, timeout is also an error
+                    errors.append(TimeoutError(timeout_error_msg))
+                    for future in not_done:
+                        future.cancel()
+                    # Wait a bit for cancellation to take effect
+                    concurrent.futures.wait(not_done, timeout=5)
+
+                # Collect exceptions from completed futures
+                for future in done:
+                    try:
+                        future.result()  # This will raise any exception that occurred
+                    except Exception as e:
+                        logger.exception("Error in summary generation future")
+                        errors.append(e)
+
+            # In preview mode, if there are any errors, fail the request
+            if errors:
+                error_messages = [str(e) for e in errors]
+                error_summary = (
+                    f"Failed to generate summaries for {len(errors)} chunk(s). "
+                    f"Errors: {'; '.join(error_messages[:3])}"  # Show first 3 errors
+                )
+                if len(errors) > 3:
+                    error_summary += f" (and {len(errors) - 3} more)"
+                logger.error("Summary generation failed in preview mode: %s", error_summary)
+                raise KnowledgeIndexNodeError(error_summary)
+
+            completed_count = sum(1 for item in preview_output["preview"] if item.get("summary") is not None)
+            logger.info(
+                "Completed summary generation for preview chunks: %s/%s succeeded",
+                completed_count,
+                len(preview_output["preview"]),
+            )
+
+        return preview_output
+
+    def _get_preview_output(
+        self,
+        chunk_structure: str,
+        chunks: Any,
+        dataset: Dataset | None = None,
+        variable_pool: VariablePool | None = None,
+    ) -> Mapping[str, Any]:
+        index_processor = IndexProcessorFactory(chunk_structure).init_index_processor()
+        preview_output = index_processor.format_preview(chunks)
+
+        # If dataset is provided, try to enrich preview with summaries
+        if dataset and variable_pool:
+            document_id = variable_pool.get(["sys", SystemVariableKey.DOCUMENT_ID])
+            if document_id:
+                document = db.session.query(Document).filter_by(id=document_id.value).first()
+                if document:
+                    # Query summaries for this document
+                    summaries = (
+                        db.session.query(DocumentSegmentSummary)
+                        .filter_by(
+                            dataset_id=dataset.id,
+                            document_id=document.id,
+                            status="completed",
+                            enabled=True,
+                        )
+                        .all()
+                    )
+
+                    if summaries:
+                        # Create a map of segment content to summary for matching
+                        # Use content matching as chunks in preview might not be indexed yet
+                        summary_by_content = {}
+                        for summary in summaries:
+                            segment = (
+                                db.session.query(DocumentSegment)
+                                .filter_by(id=summary.chunk_id, dataset_id=dataset.id)
+                                .first()
+                            )
+                            if segment:
+                                # Normalize content for matching (strip whitespace)
+                                normalized_content = segment.content.strip()
+                                summary_by_content[normalized_content] = summary.summary_content
+
+                        # Enrich preview with summaries by content matching
+                        if "preview" in preview_output and isinstance(preview_output["preview"], list):
+                            matched_count = 0
+                            for preview_item in preview_output["preview"]:
+                                if "content" in preview_item:
+                                    # Normalize content for matching
+                                    normalized_chunk_content = preview_item["content"].strip()
+                                    if normalized_chunk_content in summary_by_content:
+                                        preview_item["summary"] = summary_by_content[normalized_chunk_content]
+                                        matched_count += 1
+
+                            if matched_count > 0:
+                                logger.info(
+                                    "Enriched preview with %s existing summaries (dataset: %s, document: %s)",
+                                    matched_count,
+                                    dataset.id,
+                                    document.id,
+                                )
+
+        return preview_output
 
     @classmethod
     def version(cls) -> str:

api/core/workflow/nodes/knowledge_retrieval/knowledge_retrieval_node.py

@@ -419,6 +419,9 @@ class KnowledgeRetrievalNode(LLMUsageTrackingMixin, Node[KnowledgeRetrievalNodeD
                             source["content"] = f"question:{segment.get_sign_content()} \nanswer:{segment.answer}"
                         else:
                             source["content"] = segment.get_sign_content()
+                        # Add summary if available
+                        if record.summary:
+                            source["summary"] = record.summary
                         retrieval_resource_list.append(source)
         if retrieval_resource_list:
             retrieval_resource_list = sorted(

api/core/workflow/nodes/llm/node.py

@@ -1093,6 +1093,8 @@ class LLMNode(Node[LLMNodeData]):
                         if "content" not in item:
                             raise InvalidContextStructureError(f"Invalid context structure: {item}")
 
+                        if item.get("summary"):
+                            context_str += item["summary"] + "\n"
                         context_str += item["content"] + "\n"
 
                         retriever_resource = self._convert_to_original_retriever_resource(item)
@@ -1154,6 +1156,7 @@ class LLMNode(Node[LLMNodeData]):
                 page=metadata.get("page"),
                 doc_metadata=metadata.get("doc_metadata"),
                 files=context_dict.get("files"),
+                summary=context_dict.get("summary"),
             )
 
             return source