git clone https://github.com/langgenius/dify.git

cd dify/docker
cp middleware.env.example middleware.env
docker compose -f docker-compose.middleware.yaml up -d

conda create --name dify python=3.12
conda activate dify
pip install poetry

# 进入api目录
cd api/
# 拷贝环境变量文件
cp .env.example .env
# 生成随机密钥
openssl rand -base64 42
# 替换环境变量文件中的密钥
sed -i 's/SECRET_KEY=.*/SECRET_KEY=<your_value>/' .env
# 使用poetry安装依赖
poetry install
# 执行数据库迁移到最新版本
flask db upgrade

# 启动dify-api服务 
flask run --host 0.0.0.0 --port=5001 --debug

# 启动dify-worker服务
celery -A app.celery worker -P gevent -c 1 --loglevel INFO -Q dataset,generation,mail,ops_trace

cd web

# 安装依赖
npm install
# 拷贝环境变量文件
cp .env.example .env.local

# 编译
npm run build

# 启动前端服务
npm run start

{
    "version": "0.2.0",
    "configurations": [
        {
            "name": "dify-api",
            "type": "debugpy",
            "request": "launch",
            "module": "flask",
            "env": {
                "FLASK_APP": "app.py",
                "FLASK_ENV": "development",
                "GEVENT_SUPPORT": "true",
            },
            "args": [
                "run",
                "--host=0.0.0.0",
                "--port=5001",
                "--debug"
            ],
            "cwd": "${workspaceFolder}/api",
            "justMyCode": false
        },
        {
            "name": "dify-worker",
            "type": "debugpy",
            "request": "launch",
            "program": "/opt/anaconda3/envs/dify/bin/celery",
            "env": {
                "GEVENT_SUPPORT": "true",
            },
            "args": [
                "-A",
                "app.celery",
                "worker",
                "-P",
                "gevent",
                "-c",
                "1",
                "--loglevel=INFO",
                "-Q",
                "dataset,generation,mail,ops_trace"
            ],
            "cwd": "${workspaceFolder}/api",
            "justMyCode": false,
            "console": "integratedTerminal"
        },
        {
            "name": "dify-web",
            "type": "node",
            "request": "launch",
            "runtimeExecutable": "/opt/homebrew/bin/pnpm",
            "runtimeArgs": [
                "start"
            ],
            "cwd": "${workspaceFolder}/web",
            "console": "integratedTerminal",
            "internalConsoleOptions": "neverOpen",
        }
    ]
}

docker run -it -d -p 3000:3000 -e CONSOLE_URL=http://127.0.0.1:5001 -e APP_URL=http://127.0.0.1:5001 langgenius/dify-web:main

import os
import sys


def is_db_command():
    """
    检查当前是否在执行 Flask 数据库相关命令
    返回值：如果是数据库命令返回 True，否则返回 False
    """
    if len(sys.argv) > 1 and sys.argv[0].endswith("flask") and sys.argv[1] == "db":
        return True
    return False


# 创建应用程序
if is_db_command():
    # 如果是数据库命令，使用专门的工厂函数创建用于数据库迁移的应用实例
    from app_factory import create_migrations_app

    app = create_migrations_app()
else:
    # JetBrains Python 调试器似乎与 gevent 不兼容，
    # 因此在调试模式下需要禁用 gevent。
    # 如果您使用 debugpy 并设置 GEVENT_SUPPORT=True，则可以使用 gevent 进行调试。
    if (flask_debug := os.environ.get("FLASK_DEBUG", "0")) and flask_debug.lower() in {"false", "0", "no"}:
        # 导入并初始化 gevent
        from gevent import monkey  # type: ignore
        monkey.patch_all()  # 使用 gevent 替换标准库中的阻塞操作

        # 初始化 gRPC 的 gevent 支持
        from grpc.experimental import gevent as grpc_gevent  # type: ignore
        grpc_gevent.init_gevent()

        # 为 psycopg（PostgreSQL 驱动）添加 gevent 支持
        import psycogreen.gevent  # type: ignore
        psycogreen.gevent.patch_psycopg()

    # 创建常规应用实例
    from app_factory import create_app
    app = create_app()
    # 获取 Celery 扩展实例
    celery = app.extensions["celery"]

if __name__ == "__main__":
    # 当直接运行此文件时，启动Flask开发服务器
    app.run(host="0.0.0.0", port=5001)

...
  if [[ "${DEBUG}" == "true" ]]; then
    exec flask run --host=${DIFY_BIND_ADDRESS:-0.0.0.0} --port=${DIFY_PORT:-5001} --debug
  else
    exec gunicorn \
      --bind "${DIFY_BIND_ADDRESS:-0.0.0.0}:${DIFY_PORT:-5001}" \
      --workers ${SERVER_WORKER_AMOUNT:-1} \
      --worker-class ${SERVER_WORKER_CLASS:-gevent} \
      --timeout ${GUNICORN_TIMEOUT:-200} \
      --preload \
      app:app
...

class LoginApi(Resource):
    @setup_required
    def post(self):
        # 1. 参数验证
        parser = reqparse.RequestParser()
        parser.add_argument("email", type=email, required=True, location="json")
        parser.add_argument("password", type=valid_password, required=True, location="json")
        parser.add_argument("remember_me", type=bool, required=False, default=False)
        parser.add_argument("invite_token", type=str, required=False)
        parser.add_argument("language", type=str, required=False, default="en-US")
        
        # 2. 账户状态检查
        if dify_config.BILLING_ENABLED and BillingService.is_email_in_freeze(args["email"]):
            raise AccountInFreezeError()
            
        # 3. 登录频率限制
        if AccountService.is_login_error_rate_limit(args["email"]):
            raise EmailPasswordLoginLimitError()
            
        # 4. 邀请码验证
        if invitation:
            invitation = RegisterService.get_invitation_if_token_valid(
                None, args["email"], invitation
            )
            
        # 5. 认证流程
        try:
            account = AccountService.authenticate(args["email"], args["password"])
        except services.errors.account.AccountLoginError:
            raise AccountBannedError()
        except services.errors.account.AccountPasswordError:
            AccountService.add_login_error_rate_limit(args["email"])
            raise EmailOrPasswordMismatchError()
            
        # 6. 工作区检查
        tenants = TenantService.get_join_tenants(account)
        if len(tenants) == 0:
            return {"result": "fail", "data": "workspace not found"}
            
        # 7. 生成令牌
        token_pair = AccountService.login(
            account=account, 
            ip_address=extract_remote_ip(request)
        )
        
        # 8. 重置错误计数
        AccountService.reset_login_error_rate_limit(args["email"])
        return {"result": "success", "data": token_pair.model_dump()}

@setup_required
@login_required
@account_initialization_required
@validate_app_token
@cloud_edition_billing_resource_check
...

def validate_app_token(view: Optional[Callable] = None, *, fetch_user_arg: Optional[FetchUserArg] = None):
    def decorator(view_func):
        @wraps(view_func)
        def decorated_view(*args, **kwargs):
            # 1. 验证并获取 API Token
            api_token = validate_and_get_api_token("app")
            
            # 2. 应用状态检查
            app_model = db.session.query(App).filter(App.id == api_token.app_id).first()
            if not app_model:
                raise Forbidden("The app no longer exists.")
            if app_model.status != "normal":
                raise Forbidden("The app's status is abnormal.")
            if not app_model.enable_api:
                raise Forbidden("The app's API service has been disabled.")
                
            # 3. 租户状态检查
            tenant = db.session.query(Tenant).filter(Tenant.id == app_model.tenant_id).first()
            if tenant.status == TenantStatus.ARCHIVE:
                raise Forbidden("The workspace's status is archived.")
                
            # 4. 用户信息处理
            if fetch_user_arg:
                user_id = get_user_id_from_request(fetch_user_arg)
                kwargs["end_user"] = create_or_update_end_user_for_user_id(
                    app_model, user_id
                )
                
            return view_func(*args, **kwargs)
        return decorated_view
    return decorator if view is None else decorator(view)

# Token验证实现
def validate_and_get_api_token(scope: str | None = None):
    # 1. 验证 Authorization Header
    auth_header = request.headers.get("Authorization")
    if auth_header is None or " " not in auth_header:
        raise Unauthorized("Authorization header must be provided")
        
    # 2. 验证认证方案
    auth_scheme, auth_token = auth_header.split(None, 1)
    if auth_scheme.lower() != "bearer":
        raise Unauthorized("Authorization scheme must be 'Bearer'")
        
    # 3. Token 使用时间更新
    current_time = datetime.now(UTC).replace(tzinfo=None)
    cutoff_time = current_time - timedelta(minutes=1)
    
    with Session(db.engine, expire_on_commit=False) as session:
        # 4. 更新 Token 最后使用时间
        update_stmt = (
            update(ApiToken)
            .where(
                ApiToken.token == auth_token,
                (ApiToken.last_used_at.is_(None) | (ApiToken.last_used_at < cutoff_time)),
                ApiToken.type == scope,
            )
            .values(last_used_at=current_time)
            .returning(ApiToken)
        )
        
        # 5. Token 有效性验证
        api_token = result.scalar_one_or_none()
        if not api_token:
            stmt = select(ApiToken).where(
                ApiToken.token == auth_token, 
                ApiToken.type == scope
            )
            api_token = session.scalar(stmt)
            if not api_token:
                raise Unauthorized("Access token is invalid")
                
        return api_token

class TenantAccountRole(enum.StrEnum):
    OWNER = "owner"      # 租户所有者
    ADMIN = "admin"      # 管理员
    EDITOR = "editor"    # 编辑者
    NORMAL = "normal"    # 普通成员
    DATASET_OPERATOR = "dataset_operator"  # 数据集操作员

class TenantStatus(enum.StrEnum):
    NORMAL = "normal"   # 正常状态
    ARCHIVE = "archive" # 归档状态

class TenantAccountRole(enum.StrEnum):
    OWNER = "owner"
    ADMIN = "admin"
    EDITOR = "editor"
    NORMAL = "normal"
    DATASET_OPERATOR = "dataset_operator"
    # 添加新角色，例如：
    VIEWER = "viewer"  # 只读角色

@staticmethod
def is_valid_role(role: str) -> bool:
    return role in {
        TenantAccountRole.OWNER,
        TenantAccountRole.ADMIN,
        TenantAccountRole.EDITOR,
        TenantAccountRole.NORMAL,
        TenantAccountRole.DATASET_OPERATOR,
        TenantAccountRole.VIEWER,  # 添加新角色
    }

@staticmethod
def is_viewing_role(role: str) -> bool:
    return role in {
        TenantAccountRole.OWNER,
        TenantAccountRole.ADMIN,
        TenantAccountRole.EDITOR,
        TenantAccountRole.NORMAL,
        TenantAccountRole.VIEWER
    }

@property
def is_viewer(self):
    return self._current_tenant.current_role == TenantAccountRole.VIEWER

class TenantAccountJoinRole(enum.Enum):
    OWNER = "owner"
    ADMIN = "admin"
    NORMAL = "normal"
    DATASET_OPERATOR = "dataset_operator"
    VIEWER = "viewer"  # 添加新角色

# 在 TenantAccountJoin 模型中添加角色约束
role = db.Column(db.String(16), 
                nullable=False, 
                server_default="normal",
                db.CheckConstraint("role IN ('owner', 'admin', 'normal', 'dataset_operator', 'viewer')"))

# 基础模型类
class AIModel:
    model_type: ModelType  # 模型类型(LLM/Embedding等)

# LLM基类
class LargeLanguageModel(AIModel):
    model_type = ModelType.LLM
    
    def invoke(self, model: str, credentials: dict, prompt_messages: list[PromptMessage], 
              model_parameters: dict = None, tools: list[PromptMessageTool] = None,
              stop: list[str] = None, stream: bool = True, user: str = None) -> Union[LLMResult, Generator]:
        """调用LLM模型的核心方法"""
        pass
        
    def get_parameter_rules(self, model: str, credentials: dict) -> list[ParameterRule]:
        """获取模型参数规则"""
        pass
        
    def get_model_mode(self, model: str, credentials: dict = None) -> LLMMode:
        """获取模型模式(对话/补全)"""
        pass

class ModelInstance:
    def __init__(self):
        self.tenant_id: str  # 租户ID
        self.model_type: ModelType  # 模型类型
        self.provider: str  # 提供商
        self.model: str  # 模型名称
        self.credentials: dict  # 认证信息
        self.model_type_instance: Union[LargeLanguageModel, TextEmbeddingModel]  # 具体模型实例
        
    def get_llm_num_tokens(self, prompt_messages: list[PromptMessage], 
                          tools: list[PromptMessageTool] = None) -> int:
        """计算token数量"""
        pass
        
    def invoke_text_embedding(self, texts: list[str], user: str = None,
                            input_type: EmbeddingInputType = EmbeddingInputType.DOCUMENT) -> TextEmbeddingResult:
        """调用文本嵌入"""
        pass

class ModelManager:
    def get_model_instance(self, tenant_id: str, model_type: ModelType,
                          provider: str, model: str) -> ModelInstance:
        """获取模型实例"""
        # 1. 获取提供商配置
        provider_configurations = self._get_provider_configurations(tenant_id)
        
        # 2. 验证模型是否存在
        if provider not in provider_configurations:
            raise ValueError(f"Provider {provider} does not exist")
            
        # 3. 创建模型实例
        model_instance = ModelInstance(
            tenant_id=tenant_id,
            model_type=model_type,
            provider=provider,
            model=model
        )
        
        return model_instance
        
    def invoke_llm(self, model_instance: ModelInstance, 
                   prompt_messages: list[PromptMessage],
                   stream: bool = True,
                   user: str = None) -> Union[LLMResult, Generator]:
        """调用LLM模型"""
        pass

# OpenAI实现
class OpenAILargeLanguageModel(LargeLanguageModel):
    def _invoke(self, model: str, credentials: dict,
                prompt_messages: list[PromptMessage],
                model_parameters: dict,
                tools: list[PromptMessageTool] = None,
                stop: list[str] = None,
                stream: bool = True,
                user: str = None) -> Union[LLMResult, Generator]:
        """实现OpenAI模型调用"""
        # 1. 处理认证信息
        client = OpenAI(**credentials)
        
        # 2. 转换消息格式
        messages = self._convert_messages(prompt_messages)
        
        # 3. 调用API
        response = client.chat.completions.create(
            model=model,
            messages=messages,
            stream=stream,
            **model_parameters
        )
        
        # 4. 处理响应
        return self._handle_response(response, stream)
        
    def remote_models(self, credentials: dict) -> list[AIModelEntity]:
        """获取远程可用模型列表"""
        pass

class ModelProperties(BaseModel):
    context_size: int  # 上下文窗口大小
    max_tokens: int    # 最大token数
    mode: LLMMode     # 模型模式

class ModelConfig(BaseModel):
    properties: ModelProperties
    features: list[ModelFeature]  # 模型特性

# 模型配置示例
MODEL_CONFIGS = {
    "gpt-4": ModelConfig(
        properties=ModelProperties(
            context_size=8192,
            max_tokens=4096,
            mode=LLMMode.CHAT
        ),
        features=[
            ModelFeature.AGENT_THOUGHT,
            ModelFeature.TOOL_CALL
        ]
    )
}

class ProviderManager:
    def update_default_model_record(self, tenant_id: str, 
                                  model_type: ModelType,
                                  provider: str, 
                                  model: str) -> TenantDefaultModel:
        """更新默认模型记录"""
        # 1. 验证提供商
        provider_configurations = self.get_configurations(tenant_id)
        if provider not in provider_configurations:
            raise ValueError(f"Provider {provider} does not exist")
            
        # 2. 验证模型
        available_models = provider_configurations.get_models(
            model_type=model_type,
            only_active=True
        )
        
        # 3. 更新数据库记录
        default_model = TenantDefaultModel.query.filter(
            tenant_id=tenant_id,
            model_type=model_type
        ).first()
        
        if default_model:
            default_model.provider_name = provider
            default_model.model_name = model
        else:
            default_model = TenantDefaultModel(
                tenant_id=tenant_id,
                model_type=model_type,
                provider_name=provider,
                model_name=model
            )
            
        db.session.commit()
        return default_model

# 1. 创建模型实例
model_instance = model_manager.get_model_instance(
    tenant_id="tenant_123",
    model_type=ModelType.LLM,
    provider="openai",
    model="gpt-4"
)

# 2. 准备提示消息
messages = [
    SystemPromptMessage(content="You are a helpful assistant"),
    UserPromptMessage(content="Tell me about LLMs")
]

# 3. 设置模型参数
model_parameters = {
    "temperature": 0.7,
    "max_tokens": 1000
}

# 4. 调用模型
try:
    response = model_instance.model_type_instance.invoke(
        model="gpt-4",
        credentials=model_instance.credentials,
        prompt_messages=messages,
        model_parameters=model_parameters,
        stream=True
    )
    
    # 5. 处理流式响应
    for chunk in response:
        yield chunk.delta.content
        
except Exception as e:
    logger.error(f"Model invocation failed: {str(e)}")
    raise

class FileService:
    @staticmethod
    def upload_file(
        filename: str,
        content: bytes,
        mimetype: str,
        user: Union[Account, EndUser, Any],
        source: Literal["datasets"] | None = None,
        source_url: str = "",
    ) -> UploadFile:

class StorageType(StrEnum):
    ALIYUN_OSS = "aliyun-oss"  # 阿里云OSS
    AZURE_BLOB = "azure-blob"  # Azure Blob存储
    BAIDU_OBS = "baidu-obs"    # 百度对象存储
    GOOGLE_STORAGE = "google-storage"  # Google Cloud Storage
    HUAWEI_OBS = "huawei-obs"  # 华为对象存储
    LOCAL = "local"            # 本地存储
    S3 = "s3"                 # AWS S3
    TENCENT_COS = "tencent-cos" # 腾讯云COS
    # ...等

class BaseStorage(ABC):
    @abstractmethod
    def save(self, filename, data):
        """保存文件"""
        
    @abstractmethod  
    def load_once(self, filename: str) -> bytes:
        """一次性加载文件"""
        
    @abstractmethod
    def load_stream(self, filename: str) -> Generator:
        """流式加载文件"""
        
    @abstractmethod
    def download(self, filename, target_filepath):
        """下载文件"""
        
    @abstractmethod
    def exists(self, filename):
        """检查文件是否存在"""
        
    @abstractmethod
    def delete(self, filename):
        """删除文件"""

class AwsS3Storage(BaseStorage):
    def __init__(self):
        self.bucket_name = dify_config.S3_BUCKET_NAME
        self.client = boto3.client(
            "s3",
            aws_secret_access_key=dify_config.S3_SECRET_KEY,
            aws_access_key_id=dify_config.S3_ACCESS_KEY,
            endpoint_url=dify_config.S3_ENDPOINT,
            region_name=dify_config.S3_REGION
        )
        
    def save(self, filename, data):
        self.client.put_object(
            Bucket=self.bucket_name,
            Key=filename,
            Body=data
        )

class FileApi(Resource):
    def get(self):
        return {
            "file_size_limit": dify_config.UPLOAD_FILE_SIZE_LIMIT,
            "batch_count_limit": dify_config.UPLOAD_FILE_BATCH_LIMIT,
            "image_file_size_limit": dify_config.UPLOAD_IMAGE_FILE_SIZE_LIMIT,
            "video_file_size_limit": dify_config.UPLOAD_VIDEO_FILE_SIZE_LIMIT,
            "audio_file_size_limit": dify_config.UPLOAD_AUDIO_FILE_SIZE_LIMIT
        }

class UploadFile(db.Model):
    id: Mapped[str]
    tenant_id: Mapped[str]  # 租户ID
    storage_type: Mapped[str]  # 存储类型
    key: Mapped[str]  # 存储键
    name: Mapped[str]  # 文件名
    size: Mapped[int]  # 文件大小
    extension: Mapped[str]  # 扩展名
    mime_type: Mapped[str]  # MIME类型
    created_by: Mapped[str]  # 创建者
    created_at: Mapped[datetime]  # 创建时间

# 基类定义向量数据库的基本接口
class BaseVector(ABC):
    @abstractmethod
    def create(self, texts, embeddings, **kwargs)  # 创建集合
    @abstractmethod 
    def add_texts(self, documents, embeddings, **kwargs) # 添加文档
    @abstractmethod
    def search_by_vector(self, query_vector, **kwargs) # 向量搜索
    @abstractmethod
    def search_by_full_text(self, query, **kwargs) # 全文搜索
    @abstractmethod
    def delete(self) # 删除集合

def _publish(self, event: AppQueueEvent, pub_from: PublishFrom) -> None:
    message = MessageQueueMessage(
        task_id=self._task_id,
        message_id=self._message_id,
        conversation_id=self._conversation_id,
        app_mode=self._app_mode,
        event=event,
    )
    self._q.put(message)

def generate(
    self,
    app_model: App,
    user: Union[Account, EndUser],
    args: Mapping[str, Any],
    invoke_from: InvokeFrom,
    streaming: bool = True,
):
    # 验证必要参数
    if not args.get("query"):
        raise ValueError("query is required")
    
    # 获取会话信息
    conversation = None
    if args.get("conversation_id"):
        conversation = self._get_conversation_by_user(...)
        
    # 加载应用模型配置
    app_model_config = self._get_app_model_config(...)

def generate(
    self,
    *,
    app_model: App,
    user: Union[Account, EndUser],
    args: Mapping[str, Any],
    invoke_from: InvokeFrom,
    streaming: bool = True,
):
    # 验证流式响应
    if not streaming:
        raise ValueError("Agent Chat App does not support blocking mode")
    
    # 验证必要参数
    if not args.get("query"):
        raise ValueError("query is required")
        
    # 获取会话信息
    conversation = None
    if args.get("conversation_id"):
        conversation = self._get_conversation_by_user(...)

def generate(
    self,
    app_model: App,
    workflow: Workflow,
    user: Union[Account, EndUser],
    args: Mapping[str, Any],
    invoke_from: InvokeFrom,
    streaming: bool = True,
):
    # 验证工作流
    if not workflow:
        raise ValueError("workflow is required")
        
    # 验证输入
    if not args.get("query"):
        raise ValueError("query is required")
        
    # 初始化上下文
    contexts.tenant_id.set(app_model.tenant_id)

def generate(
    self,
    app_model: App,
    workflow_config: dict,
    user: Union[Account, EndUser],
    args: Mapping[str, Any],
    invoke_from: InvokeFrom,
):
    # 验证工作流配置
    if not workflow_config:
        raise ValueError("workflow_config is required")
        
    # 初始化工作流状态
    workflow_state = WorkflowState(workflow_config)

def generate(
    self,
    app_model: App,
    user: Union[Account, EndUser],
    args: Mapping[str, Any],
    invoke_from: InvokeFrom,
):
    # 验证输入文本
    if not args.get("prompt"):
        raise ValueError("prompt is required")
        
    # 加载模型配置
    model_config = self._get_model_config(app_model)