大模型应用编排工具Dify二开之登录Token改造

dify工作室支持在画布上直接编辑业务流程，通过调用开源大模型可以实现特定场景的业务，而且可以迅速更新发布。因此，某些项目要求在产品里面能够直接编辑 dify业务流程，使得现场开发人员能够迅速响应客户需求。另外，方便对 dify进行运维，比如更新开源大模型认证信息。

Daphnis_z

3345人浏览 · 2024-11-28 19:15:00

Daphnis_z · 2024-11-28 19:15:00 发布

1.前言

环境信息：dify-0.8.3, docker-21

2.实现思路分析

常规的思路有两种：

把 dify源码迁移到产品中

代码改造量大、难度高，且会导致后续难以跟随 dify官方进行升级。
把 dify页面直接嵌入到产品中

界面样式不统一，需要改造登录模块，工作量可以接受，且易升级。

根据项目实际情况，这里选择第二种方案，把 dify页面嵌入到产品中。同时，通过观察 dify登录流程和分析相关请求，发现只要在请求页面时带上认证所需的 token即可实现 dify的免登陆，如下：

http://127.0.0.1/apps?console_token=eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VyX2lkIjoiZDhkNDdjMGUtNTY2Yy00ZWQ3LTkxNDgtYmIzYWRjN2YyNDE2IiwiZXhwIjoxNzI3NjA0NTAxLCJpc3MiOiJTRUxGX0hPU1RFRCIsInN1YiI6IkNvbnNvbGUgQVBJIFBhc3Nwb3J0In0.sNjM_1G-856hfd6jYzzYy4Na0w_Yeb-LWKXWJ-SR55

获取认证所需的 token如下图：
在这里插入图片描述

点击工具》F12查看网络请求》打开 tool-labels请求，在 Headers里面可以看到认证 token。

注意：登录 token一般存在有效期，故这里需要修改，见下面的章节。

3.定位登录Token生成代码

通过关键词搜索和添加日志，定位到生成登录所需的 token代码位于 api\services\account_service.py，如下：

    @staticmethod
    def get_account_jwt_token(account, *, exp: timedelta = timedelta(days=30)):
        logger.info(f"------------get_account_jwt_token, account.id: '{account.id}', exp: '{exp}'")

        payload = {
            "user_id": account.id,
            "exp": datetime.now(timezone.utc).replace(tzinfo=None) + exp,
            "iss": dify_config.EDITION,
            "sub": "Console API Passport",
        }

        token = PassportService().issue(payload)
        return token

可以看到上面的代码在生成 token时默认 30天有效期，且 token是和用户ID绑定的。

另外，这里的代码是生成 token的具体函数，需要修改是调用此函数的地方，如下：

@staticmethod
def login(account: Account, *, ip_address: Optional[str] = None):
    if ip_address:
        AccountService.update_last_login(account, ip_address=ip_address)
    exp = timedelta(days=30)
    token = AccountService.get_account_jwt_token(account, exp=exp)
    redis_client.set(_get_login_cache_key(account_id=account.id, token=token), "1", ex=int(exp.total_seconds()))
    return token

4.修改登录Token有效期

通过上一步骤已经定位并分析清楚登录 token的生成和调用逻辑，这里修改 token有效期从 30天到 1095天（3年，一般软件维护期）。

    @staticmethod
    def login(account: Account, *, ip_address: Optional[str] = None):
        if ip_address:
            AccountService.update_last_login(account, ip_address=ip_address)
        # 设置登录token有效期
        exp = timedelta(days=1095)
        token = AccountService.get_account_jwt_token(account, exp=exp)
        redis_client.set(_get_login_cache_key(account_id=account.id, token=token), "1", ex=int(exp.total_seconds()))
        return token