dify源码分析(1)

[+] Running 10/10 ⠿ Container docker-sandbox-1        Started                                                                        16.7s ⠿ Container docker-ssrf_proxy-1     Started                                                                        16.9s ⠿ Container docker-web-1            Started                                                                        16.7s ⠿ Container docker-worker-1         Started                                                                        50.2s ⠿ Container docker-plugin_daemon-1  Started                                                                        49.7s ⠿ Container docker-api-1            Started                                                                        50.2s ⠿ Container docker-nginx-1          Started                                                                        48.1s ⠿ Container docker-redis-1          Started                                                                         2.0s ⠿ Container docker-db-1             Healthy                                                                        33.8s ⠿ Container docker-weaviate-1       Started 

langgenius/dify-sandbox:0.2.12

ubuntu/squid:latest

uv run flask run --host 0.0.0.0 --port=5001 --debug

uv run celery -A app.celery worker -P threads -c 2 --loglevel INFO -Q dataset,priority_dataset,priority_pipeline,pipeline,mail,ops_trace,app_deletion,plugin,workflow_storage,conversation,workflow,schedule_poller,schedule_executor,triggered_workflow_dispatcher,trigger_refresh_executor,retention

    from app_factory import create_app    app = create_app()    celery = app.extensions["celery"]

if __name__ == "__main__":    app.run(host="0.0.0.0", port=5001)

def create_app() -> DifyApp:    start_time = time.perf_counter()    app = create_flask_app_with_configs()    initialize_extensions(app)    end_time = time.perf_counter()    if dify_config.DEBUG:        logger.info("Finished create_app (%s ms)", round((end_time - start_time) * 1000, 2))    return app

def create_flask_app_with_configs() -> DifyApp:    """    create a raw flask app    with configs loaded from .env file    """    dify_app = DifyApp(__name__)

from flask import Flaskclass DifyApp(Flask):    pass

from flask import Blueprintfrom flask_restx import Namespacefrom libs.external_api import ExternalApibp = Blueprint("service_api", __name__, url_prefix="/v1")api = ExternalApi(    bp,    version="1.0",    title="Service API",    description="API for application services",)service_api_ns = Namespace("service_api", description="Service operations", path="/")api.add_namespace(service_api_ns)

@service_api_ns.route("/conversations")class ConversationApi(Resource):         try:            with Session(db.engine) as session:                return ConversationService.pagination_by_last_id(                    session=session,                    app_model=app_model,                    user=end_user,                    last_id=last_id,                    limit=query_args.limit,                    invoke_from=InvokeFrom.SERVICE_API,                    sort_by=query_args.sort_by,                )

@service_api_ns.route("/completion-messages")class CompletionApi(Resource):    @validate_app_token(fetch_user_arg=FetchUserArg(fetch_from=WhereisUserArg.JSON, required=True))    def post(self, app_model: App, end_user: EndUser):        """Create a completion for the given prompt.        This endpoint generates a completion based on the provided inputs and query.        Supports both blocking and streaming response modes.        """         try:            response = AppGenerateService.generate(                app_model=app_model,                user=end_user,                args=args,                invoke_from=InvokeFrom.SERVICE_API,                streaming=streaming,            )            return helper.compact_generate_response(response)

@service_api_ns.route("/workflows/run/<string:workflow_run_id>")class WorkflowRunDetailApi(Resource):    @service_api_ns.marshal_with(build_workflow_run_model(service_api_ns))    def get(self, app_model: App, workflow_run_id: str):        """Get a workflow task running detail.        Returns detailed information about a specific workflow run.        """         session_maker = sessionmaker(bind=db.engine, expire_on_commit=False)        workflow_run_repo = DifyAPIRepositoryFactory.create_api_workflow_run_repository(session_maker)        workflow_run = workflow_run_repo.get_workflow_run_by_id(            tenant_id=app_model.tenant_id,            app_id=app_model.id,            run_id=workflow_run_id,        )        return workflow_run

@service_api_ns.route("/workflows/run")class WorkflowRunApi(Resource):    @validate_app_token(fetch_user_arg=FetchUserArg(fetch_from=WhereisUserArg.JSON, required=True))    def post(self, app_model: App, end_user: EndUser):        """Execute a workflow.        Runs a workflow with the provided inputs and returns the results.        Supports both blocking and streaming response modes.        """         try:            response = AppGenerateService.generate(                app_model=app_model, user=end_user, args=args, invoke_from=InvokeFrom.SERVICE_API, streaming=streaming            )            return helper.compact_generate_response(response)

class AppGenerateService:    @classmethod    @trace_span(AppGenerateHandler)    def generate(        cls,        app_model: App,        user: Union[Account, EndUser],        args: Mapping[str, Any],        invoke_from: InvokeFrom,        streaming: bool = True,        root_node_id: str | None = None,    ):        """        App Content Generate        :param app_model: app model        :param user: user        :param args: args        :param invoke_from: invoke from        :param streaming: streaming        :return:        """         try:            request_id = rate_limit.enter(request_id)            if app_model.mode == AppMode.COMPLETION:                return rate_limit.generate(                    CompletionAppGenerator.convert_to_event_stream(                        CompletionAppGenerator().generate(                            app_model=app_model, user=user, args=args, invoke_from=invoke_from, streaming=streaming                        ),                    ),                    request_id=request_id,                )            elif app_model.mode == AppMode.AGENT_CHAT or app_model.is_agent:                return rate_limit.generate(                    AgentChatAppGenerator.convert_to_event_stream(                        AgentChatAppGenerator().generate(                            app_model=app_model, user=user, args=args, invoke_from=invoke_from, streaming=streaming                        ),                    ),                    request_id,                )            elif app_model.mode == AppMode.CHAT:                return rate_limit.generate(                    ChatAppGenerator.convert_to_event_stream(                        ChatAppGenerator().generate(                            app_model=app_model, user=user, args=args, invoke_from=invoke_from, streaming=streaming                        ),                    ),                    request_id=request_id,                )            elif app_model.mode == AppMode.ADVANCED_CHAT:                workflow_id = args.get("workflow_id")                workflow = cls._get_workflow(app_model, invoke_from, workflow_id)                return rate_limit.generate(                    AdvancedChatAppGenerator.convert_to_event_stream(                        AdvancedChatAppGenerator().generate(                            app_model=app_model,                            workflow=workflow,                            user=user,                            args=args,                            invoke_from=invoke_from,                            streaming=streaming,                        ),                    ),                    request_id=request_id,                )            elif app_model.mode == AppMode.WORKFLOW:                workflow_id = args.get("workflow_id")                workflow = cls._get_workflow(app_model, invoke_from, workflow_id)                return rate_limit.generate(                    WorkflowAppGenerator.convert_to_event_stream(                        WorkflowAppGenerator().generate(                            app_model=app_model,                            workflow=workflow,                            user=user,                            args=args,                            invoke_from=invoke_from,                            streaming=streaming,                            root_node_id=root_node_id,                            call_depth=0,                        ),                    ),                    request_id,                )            else:                raise ValueError(f"Invalid app mode {app_model.mode}")        except Exception:            quota_charge.refund()            rate_limit.exit(request_id)            raise        finally:            if not streaming:                rate_limit.exit(request_id)

class CompletionAppGenerator(MessageBasedAppGenerator):    @overload    def generate(        self,        app_model: App,        user: Union[Account, EndUser],        args: Mapping[str, Any],        invoke_from: InvokeFrom,        streaming: Literal[True],    ) -> Generator[str | Mapping[str, Any], None, None]: ...    def generate(        self,        app_model: App,        user: Union[Account, EndUser],        args: Mapping[str, Any],        invoke_from: InvokeFrom,        streaming: bool = True,    ) -> Union[Mapping[str, Any], Generator[str | Mapping[str, Any], None, None]]:        """        Generate App response.        :param app_model: App        :param user: account or end user        :param args: request args        :param invoke_from: invoke from source        :param streaming: is stream        """         # init application generate entity        application_generate_entity = CompletionAppGenerateEntity(            task_id=str(uuid.uuid4()),            app_config=app_config,            model_conf=ModelConfigConverter.convert(app_config),            file_upload_config=file_extra_config,            inputs=self._prepare_user_inputs(                user_inputs=inputs, variables=app_config.variables, tenant_id=app_model.tenant_id            ),            query=query,            files=list(file_objs),            user_id=user.id,            stream=streaming,            invoke_from=invoke_from,            extras={},            trace_manager=trace_manager,        )        # new thread with request context        @copy_current_request_context        def worker_with_context():            return self._generate_worker(                flask_app=current_app._get_current_object(),  # type: ignore                application_generate_entity=application_generate_entity,                queue_manager=queue_manager,                message_id=message.id,            )        worker_thread = threading.Thread(target=worker_with_context)        worker_thread.start()        # return response or stream generator        response = self._handle_response(            application_generate_entity=application_generate_entity,            queue_manager=queue_manager,            conversation=conversation,            message=message,            user=user,            stream=streaming,        )  def _generate_worker(        self,        flask_app: Flask,        application_generate_entity: CompletionAppGenerateEntity,        queue_manager: AppQueueManager,        message_id: str,    ):        """        Generate worker in a new thread.        :param flask_app: Flask app        :param application_generate_entity: application generate entity        :param queue_manager: queue manager        :param message_id: message ID        :return:        """        with flask_app.app_context():            try:                # get message                message = self._get_message(message_id)                # chatbot app                runner = CompletionAppRunner()                runner.run(                    application_generate_entity=application_generate_entity,                    queue_manager=queue_manager,                    message=message,                )

class CompletionAppRunner(AppRunner):    """    Completion Application Runner    """    def run(        self, application_generate_entity: CompletionAppGenerateEntity, queue_manager: AppQueueManager, message: Message    ):        """        Run application        :param application_generate_entity: application generate entity        :param queue_manager: application queue manager        :param message: message        :return:        """        try:            # process sensitive_word_avoidance            _, inputs, query = self.moderation_for_inputs(                app_id=app_record.id,                tenant_id=app_config.tenant_id,                app_generate_entity=application_generate_entity,                inputs=inputs,                query=query or "",                message_id=message.id,            )       invoke_result = model_instance.invoke_llm(            prompt_messages=prompt_messages,            model_parameters=application_generate_entity.model_conf.parameters,            stop=stop,            stream=application_generate_entity.stream,            user=application_generate_entity.user_id,        )

class ModelInstance:    """    Model instance class    """    @overload    def invoke_llm(        self,        prompt_messages: Sequence[PromptMessage],        model_parameters: dict | None = None,        tools: Sequence[PromptMessageTool] | None = None,        stop: list[str] | None = None,        stream: Literal[True] = True,        user: str | None = None,        callbacks: list[Callback] | None = None,    ) -> Generator: ...    def invoke_llm(        self,        prompt_messages: Sequence[PromptMessage],        model_parameters: dict | None = None,        tools: Sequence[PromptMessageTool] | None = None,        stop: Sequence[str] | None = None,        stream: bool = True,        user: str | None = None,        callbacks: list[Callback] | None = None,    ) -> Union[LLMResult, Generator]:        """        Invoke large language model        :param prompt_messages: prompt messages        :param model_parameters: model parameters        :param tools: tools for tool calling        :param stop: stop words        :param stream: is stream response        :param user: unique user id        :param callbacks: callbacks        :return: full response or stream response chunk generator result        """         return cast(            Union[LLMResult, Generator],            self._round_robin_invoke(                function=self.model_type_instance.invoke,                model=self.model,                credentials=self.credentials,                prompt_messages=prompt_messages,                model_parameters=model_parameters,                tools=tools,                stop=stop,                stream=stream,                user=user,                callbacks=callbacks,            ),        )

class LargeLanguageModel(AIModel):    """    Model class for large language model.    """    model_type: ModelType = ModelType.LLM    # pydantic configs    model_config = ConfigDict(protected_namespaces=())    def invoke(        self,        model: str,        credentials: dict,        prompt_messages: list[PromptMessage],        model_parameters: dict | None = None,        tools: list[PromptMessageTool] | None = None,        stop: list[str] | None = None,        stream: bool = True,        user: str | None = None,        callbacks: list[Callback] | None = None,    ) -> Union[LLMResult, Generator[LLMResultChunk, None, None]]:        """        Invoke large language model        :param model: model name        :param credentials: model credentials        :param prompt_messages: prompt messages        :param model_parameters: model parameters        :param tools: tools for tool calling        :param stop: stop words        :param stream: is stream response        :param user: unique user id        :param callbacks: callbacks        :return: full response or stream response chunk generator result        """        # validate and filter model parameters        if model_parameters is None:            model_parameters = {}        self.started_at = time.perf_counter()        callbacks = callbacks or []        if dify_config.DEBUG:            callbacks.append(LoggingCallback())        # trigger before invoke callbacks        self._trigger_before_invoke_callbacks(            model=model,            credentials=credentials,            prompt_messages=prompt_messages,            model_parameters=model_parameters,            tools=tools,            stop=stop,            stream=stream,            user=user,            callbacks=callbacks,        )        result: Union[LLMResult, Generator[LLMResultChunk, None, None]]        try:            from core.plugin.impl.model import PluginModelClient            plugin_model_manager = PluginModelClient()            result = plugin_model_manager.invoke_llm(                tenant_id=self.tenant_id,                user_id=user or "unknown",                plugin_id=self.plugin_id,                provider=self.provider_name,                model=model,                credentials=credentials,                model_parameters=model_parameters,                prompt_messages=prompt_messages,                tools=tools,                stop=list(stop) if stop else None,                stream=stream,            )

class PluginModelClient(BasePluginClient):   def invoke_llm(        self,        tenant_id: str,        user_id: str,        plugin_id: str,        provider: str,        model: str,        credentials: dict,        prompt_messages: list[PromptMessage],        model_parameters: dict | None = None,        tools: list[PromptMessageTool] | None = None,        stop: list[str] | None = None,        stream: bool = True,    ) -> Generator[LLMResultChunk, None, None]:        """        Invoke llm        """        response = self._request_with_plugin_daemon_response_stream(            method="POST",            path=f"plugin/{tenant_id}/dispatch/llm/invoke",            type_=LLMResultChunk,            data=jsonable_encoder(                {                    "user_id": user_id,                    "data": {                        "provider": provider,                        "model_type": "llm",                        "model": model,                        "credentials": credentials,                        "prompt_messages": prompt_messages,                        "model_parameters": model_parameters,                        "tools": tools,                        "stop": stop,                        "stream": stream,                    },                }            ),            headers={                "X-Plugin-ID": plugin_id,                "Content-Type": "application/json",            },        )        try:            yield from response