# api\models\account.py
class Account(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    
    _current_tenant = None
    def set_current_tenant(self, tenant_id: int):
        self._current_tenant = tenant_id
    def get_current_tenant(self) -> Optional[int]:
        return self._current_tenant

# api\core\rag\index_processor\index_processor_base.py
   class BaseIndexProcessor(ABC):
       @abstractmethod
       def extract(self, extract_setting: ExtractSetting, **kwargs) -> list[Document]:
           ...
       @abstractmethod
       def clean(self, dataset: Dataset, node_ids: list[str] | None, with_keywords: bool = True, **kwargs):
           ...
       @abstractmethod
       def index(self, dataset: Dataset, document: DatasetDocument, chunks: Any):
           ...
       @abstractmethod
       def retrieve(
           self,
           retrieval_method: RetrievalMethod,
           query: str,
           dataset: Dataset,
           top_k: int,
           score_threshold: float,
           reranking_model: dict,
       ) -> list[Document]:
           ...

# api\core\app\apps\workflow\app_runner.py
class WorkflowAppRunner(WorkflowBasedAppRunner):
      ...
    @trace_span(WorkflowAppRunnerHandler)
    def run(self):
        system_inputs = SystemVariable(
            files=self.application_generate_entity.files,
            user_id=self._sys_user_id,
            app_id=app_config.app_id,
            workflow_id=app_config.workflow_id,
            workflow_execution_id=self.application_generate_entity.workflow_execution_id,
        )
        variable_pool = VariablePool(
            system_variables=system_inputs,
            user_inputs=self.application_generate_entity.inputs,
            environment_variables=self._workflow.environment_variables,
            conversation_variables=[],
        )
        graph_runtime_state = GraphRuntimeState(variable_pool=variable_pool, start_at=time.perf_counter())
        graph = self._init_graph(
            graph_config=self._workflow.graph_dict,
            graph_runtime_state=graph_runtime_state,
            workflow_id=self._workflow.id,
            tenant_id=self._workflow.tenant_id,
            user_id=self.application_generate_entity.user_id,
            user_from=user_from,
            invoke_from=invoke_from,
            root_node_id=self._root_node_id,
        )
        task_id = self.application_generate_entity.task_id
        channel_key = f"workflow:{task_id}:commands"
        command_channel = RedisChannel(redis_client, channel_key)
        self._queue_manager.graph_runtime_state = graph_runtime_state
        workflow_entry = WorkflowEntry(
            tenant_id=self._workflow.tenant_id,
            app_id=self._workflow.app_id,
            workflow_id=self._workflow.id,
            graph=graph,
            graph_config=self._workflow.graph_dict,
            user_id=self.application_generate_entity.user_id,
            user_from=user_from,
            invoke_from=invoke_from,
            call_depth=self.application_generate_entity.call_depth,
            variable_pool=variable_pool,
            graph_runtime_state=graph_runtime_state,
            command_channel=command_channel,
        )
        persistence_layer = WorkflowPersistenceLayer(...)
        workflow_entry.graph_engine.layer(persistence_layer)
        for layer in self._graph_engine_layers:
            workflow_entry.graph_engine.layer(layer)
        generator = workflow_entry.run()
        for event in generator:
            self._handle_event(workflow_entry, event)

class Plugin:
    def __init__(self, config: dict):
        self.config = config
    def call(self, function_name: str, args: dict) -> Any:
        method = getattr(self, function_name, None)
        if not method:
            raise ValueError(f"Function {function_name} not found")
        return method(**args)
# Example: OrderPlugin
class OrderPlugin(Plugin):
    def get_order_status(self, order_id: str) -> dict:
        return requests.get(f"https://api.example.com/orders/{order_id}").json()

class Role(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(50), nullable=False)  # "admin", "editor", "viewer"
    description = db.Column(db.Text)
    permissions = db.Column(db.JSON)  # JSON array of permission strings
class UserTenantRole(db.Model):
    user_id = db.Column(db.Integer, db.ForeignKey('account.id'), primary_key=True)
    tenant_id = db.Column(db.Integer, db.ForeignKey('tenant.id'), primary_key=True)
    role_id = db.Column(db.Integer, db.ForeignKey('role.id'))

@app.before_request
def load_current_tenant():
    if g.user and hasattr(g.user, '_current_tenant'):
        # 已设置，则跳过
        return
    # 从 JWT token 或 Header 中提取 tenant_id
    tenant_id = request.headers.get('X-Tenant-ID') or get_jwt_claims().get('tenant_id')
    if tenant_id:
        g.current_tenant = tenant_id
        g.user.set_current_tenant(tenant_id)

# api\core\errors\error.py
class LLMError(ValueError):
    """Base class for all LLM exceptions."""
    description: str | None = None
    def __init__(self, description: str | None = None):
        self.description = description
class LLMBadRequestError(LLMError):
    """Raised when the LLM returns bad request."""
    description = "Bad Request"
class ProviderTokenNotInitError(ValueError):
    """
    Custom exception raised when the provider token is not initialized.
    """
    description = "Provider Token Not Init"
    def __init__(self, *args, **kwargs):
        self.description = args[0] if args else self.description
class QuotaExceededError(ValueError):
    """
    Custom exception raised when the quota for a provider has been exceeded.
    """
    description = "Quota Exceeded"
class AppInvokeQuotaExceededError(ValueError):
    """
    Custom exception raised when the quota for an app has been exceeded.
    """
    description = "App Invoke Quota Exceeded"
class ModelCurrentlyNotSupportError(ValueError):
    """
    Custom exception raised when the model not support
    """
    description = "Model Currently Not Support"
class InvokeRateLimitError(ValueError):
    """Raised when the Invoke returns rate limit error."""
    description = "Rate Limit Error"

# api\services\app_generate_service.py
class AppGenerateService:
    @classmethod
    @trace_span(AppGenerateHandler)
    def generate(cls, app_model, user, args, invoke_from, streaming=True, root_node_id=None):
        quota_charge = unlimited()
        if dify_config.BILLING_ENABLED:
            quota_charge = QuotaType.WORKFLOW.consume(app_model.tenant_id)
        max_active_request = cls._get_max_active_requests(app_model)
        rate_limit = RateLimit(app_model.id, max_active_request)
        request_id = RateLimit.gen_request_key()
        try:
            request_id = rate_limit.enter(request_id)
            if app_model.mode == AppMode.COMPLETION:
                return rate_limit.generate(CompletionAppGenerator.convert_to_event_stream(...), request_id)
            elif app_model.mode == AppMode.AGENT_CHAT or app_model.is_agent:
                ...
            elif app_model.mode == AppMode.WORKFLOW:
                ...
        except Exception:
            quota_charge.refund()
            rate_limit.exit(request_id)
            raise
        finally:
            if not streaming:
                rate_limit.exit(request_id)

def run(self) -> Generator[GraphEngineEvent, None, None]:
    self._initialize_layers()
    is_resume = self._graph_execution.started
    ...
    start_event = GraphRunStartedEvent()
    ...
    self._start_execution(resume=is_resume)   # 启动 WorkerPool & Dispatcher
    yield from self._event_manager.emit_events()
    ...
    # 根据 is_paused / aborted / has_error / 部分异常决定 Final Event

with ThreadPoolExecutor(max_workers=dify_config.RETRIEVAL_SERVICE_EXECUTORS) as executor:
    futures = [ ]
    retrieval_service = RetrievalService()
    if query:
        futures.append(executor.submit(retrieval_service._retrieve, ..., attachment_id=None, ...))
    if attachment_ids:
        for attachment_id in attachment_ids:
            futures.append(executor.submit(retrieval_service._retrieve, ..., attachment_id=attachment_id, ...))
    if futures:
        for future in concurrent.futures.as_completed(futures, timeout=3600):
            if exceptions:
                for f in futures:
                    f.cancel()
                break
...
if exceptions:
    raise ValueError(";\n".join(exceptions))
return all_documents