opensource/dify
2
0
Fork 0
mirror of https://github.com/langgenius/dify.git synced 2026-05-04 00:18:28 +08:00
Code Issues Packages Projects Releases Wiki Activity

r2

Browse Source
This commit is contained in:
jyong 2025-06-03 15:51:31 +08:00
parent 0486aa3445
commit b8f3b23b1a
2 changed files with 33 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
api/configs/feature/__init__.py
View File

@ -667,7 +667,6 @@ class MailConfig(BaseSettings):
class RagEtlConfig(BaseSettings): class RagEtlConfig(BaseSettings):
""" """
Configuration for RAG ETL processes Configuration for RAG ETL processes
"""
# TODO: This config is not only for rag etl, it is also for file upload, we should move it to file upload config # TODO: This config is not only for rag etl, it is also for file upload, we should move it to file upload config
ETL_TYPE: str = Field( ETL_TYPE: str = Field(

57
api/core/app/apps/pipeline/pipeline_generator.py
View File

@ -55,7 +55,8 @@ class PipelineGenerator(BaseAppGenerator):
streaming: Literal[True], streaming: Literal[True],
call_depth: int, call_depth: int,
workflow_thread_pool_id: Optional[str], workflow_thread_pool_id: Optional[str],
) -> Mapping[str, Any] | Generator[Mapping | str, None, None] | None: ... ) -> Mapping[str, Any] | Generator[Mapping | str, None, None] | None:
...
@overload @overload
def generate( def generate(
@ -69,7 +70,8 @@ class PipelineGenerator(BaseAppGenerator):
streaming: Literal[False], streaming: Literal[False],
call_depth: int, call_depth: int,
workflow_thread_pool_id: Optional[str], workflow_thread_pool_id: Optional[str],
) -> Mapping[str, Any]: ... ) -> Mapping[str, Any]:
...
@overload @overload
def generate( def generate(
@ -83,7 +85,8 @@ class PipelineGenerator(BaseAppGenerator):
streaming: bool, streaming: bool,
call_depth: int, call_depth: int,
workflow_thread_pool_id: Optional[str], workflow_thread_pool_id: Optional[str],
) -> Union[Mapping[str, Any], Generator[Mapping | str, None, None]]: ... ) -> Union[Mapping[str, Any], Generator[Mapping | str, None, None]]:
...
def generate( def generate(
self, self,
@ -184,7 +187,7 @@ class PipelineGenerator(BaseAppGenerator):
) )
if invoke_from == InvokeFrom.DEBUGGER: if invoke_from == InvokeFrom.DEBUGGER:
return self._generate( return self._generate(
flask_app=current_app._get_current_object(),# type: ignore flask_app=current_app._get_current_object(), # type: ignore
context=contextvars.copy_context(), context=contextvars.copy_context(),
pipeline=pipeline, pipeline=pipeline,
workflow_id=workflow.id, workflow_id=workflow.id,
@ -199,6 +202,7 @@ class PipelineGenerator(BaseAppGenerator):
else: else:
# run in child thread # run in child thread
context = contextvars.copy_context() context = contextvars.copy_context()
@copy_current_request_context @copy_current_request_context
def worker_with_context(): def worker_with_context():
# Run the worker within the copied context # Run the worker within the copied context
@ -222,24 +226,25 @@ class PipelineGenerator(BaseAppGenerator):
worker_thread.start() worker_thread.start()
# return batch, dataset, documents # return batch, dataset, documents
return { return {
"batch": batch, "batch": batch,
"dataset": PipelineDataset( "dataset": PipelineDataset(
id=dataset.id, id=dataset.id,
name=dataset.name, name=dataset.name,
description=dataset.description, description=dataset.description,
chunk_structure=dataset.chunk_structure, chunk_structure=dataset.chunk_structure,
).model_dump(), ).model_dump(),
"documents": [PipelineDocument( "documents": [PipelineDocument(
id=document.id, id=document.id,
position=document.position, position=document.position,
data_source_info=json.loads(document.data_source_info) if document.data_source_info else None, data_source_info=json.loads(document.data_source_info) if document.data_source_info else None,
name=document.name, name=document.name,
indexing_status=document.indexing_status, indexing_status=document.indexing_status,
error=document.error, error=document.error,
enabled=document.enabled, enabled=document.enabled,
).model_dump() for document in documents ).model_dump() for document in documents
] ]
} }
def _generate( def _generate(
self, self,
*, *,
@ -268,6 +273,7 @@ class PipelineGenerator(BaseAppGenerator):
:param streaming: is stream :param streaming: is stream
:param workflow_thread_pool_id: workflow thread pool id :param workflow_thread_pool_id: workflow thread pool id
""" """
print("jin ru la 1")
for var, val in context.items(): for var, val in context.items():
var.set(val) var.set(val)
@ -279,6 +285,7 @@ class PipelineGenerator(BaseAppGenerator):
saved_user = g._login_user saved_user = g._login_user
with flask_app.app_context(): with flask_app.app_context():
# Restore user in new app context # Restore user in new app context
print("jin ru la 2")
if saved_user is not None: if saved_user is not None:
from flask import g from flask import g
@ -306,6 +313,7 @@ class PipelineGenerator(BaseAppGenerator):
application_generate_entity=application_generate_entity, application_generate_entity=application_generate_entity,
workflow_thread_pool_id=workflow_thread_pool_id, workflow_thread_pool_id=workflow_thread_pool_id,
) )
# new thread # new thread
worker_thread = threading.Thread( worker_thread = threading.Thread(
target=worker_with_context target=worker_with_context
@ -396,7 +404,7 @@ class PipelineGenerator(BaseAppGenerator):
) )
return self._generate( return self._generate(
flask_app=current_app._get_current_object(),# type: ignore flask_app=current_app._get_current_object(), # type: ignore
pipeline=pipeline, pipeline=pipeline,
workflow=workflow, workflow=workflow,
user=user, user=user,
@ -479,7 +487,7 @@ class PipelineGenerator(BaseAppGenerator):
) )
return self._generate( return self._generate(
flask_app=current_app._get_current_object(),# type: ignore flask_app=current_app._get_current_object(), # type: ignore
pipeline=pipeline, pipeline=pipeline,
workflow=workflow, workflow=workflow,
user=user, user=user,
@ -506,6 +514,7 @@ class PipelineGenerator(BaseAppGenerator):
:param workflow_thread_pool_id: workflow thread pool id :param workflow_thread_pool_id: workflow thread pool id
:return: :return:
""" """
print("jin ru la 3")
for var, val in context.items(): for var, val in context.items():
var.set(val) var.set(val)
from flask import g from flask import g