Merge pull request #383 from LlmKira/dev

dynamic require_auth

README.md

@@ -105,6 +105,8 @@ The model adheres to the Openai Schema, other models are not supported. Please a
 Refer to the [🧀 Deployment Document](https://llmkira.github.io/Docs/) for more information.
 
 ```shell
+# Install Telegram Voice dependencies
+apt install ffmpeg
 # Install RabbitMQ
 docker pull rabbitmq:3.10-management
 docker run -d -p 5672:5672 -p 15672:15672 \
@@ -149,6 +151,9 @@ cp .env.exp .env&&nano .env
 docker-compose -f docker-compose.yml up -d
 ```
 
+The Docker configuration file `docker-compose.yml` contains all databases. In fact, Redis and MongoDB are not required.
+You can remove these databases yourself and use the local file system.
+
 Update image using `docker-compose pull`.
 
 Use `docker exec -it llmbot /bin/bash` to view Shell in Docker, enter `exit` to exit.
@@ -183,6 +188,8 @@ you can enable it by setting `VOICE_REPLY_ME=true` in `.env`.
 /env REECHO_VOICE_KEY=<key in dev.reecho.ai>
 ```
 
+use `/env VOICE_REPLY_ME=NONE` to disable this env.
+
 check the source code in `llmkira/extra/voice_hook.py`, learn to write your own hooks.
 
 ## 🧀 Sponsor

app/receiver/function.py

@@ -88,7 +88,8 @@ async def create_snapshot(
         creator=task_snapshot.receiver.uid,
         expire_at=int(time.time()) + 60 * 2,
     )
-    logger.debug(f"Create a snapshot {task_id}")
+    if snapshot_credential:
+        logger.debug(f"Create a snapshot {task_id}")
     return snapshot_credential
 
 
@@ -154,7 +155,14 @@ async def run_pending_task(task: TaskHeader, pending_task: ToolCall):
         logger.debug(f"Save ToolCall {pending_task.id} to Cache Map")
         # Run Function
         _tool_obj = tool_cls()
-        if _tool_obj.require_auth:
+
+        # Get Env
+        secrets = await EnvManager(user_id=task.receiver.uid).read_env()
+        if not secrets:
+            secrets = {}
+        env_map = {name: secrets.get(name, None) for name in _tool_obj.env_list}
+        # Auth
+        if _tool_obj.require_auth(env_map):
             if task.task_sign.snapshot_credential:
                 # 是携带密钥的函数，是预先构建的可信任务头
                 task.task_sign.snapshot_credential = None
@@ -179,24 +187,21 @@ async def run_pending_task(task: TaskHeader, pending_task: ToolCall):
                 return logger.info(
                     f"[Snapshot Auth] \n--auth-require {pending_task.name} require."
                 )
-        # Get Env
-        env_all = await EnvManager(user_id=task.receiver.uid).read_env()
-        if not env_all:
-            env_all = {}
-        env_map = {}
-        for require in _tool_obj.env_list:
-            env_map[require] = env_all.get(require, None)
 
         # Resign Chain
         if len(task.task_sign.tool_calls_pending) == 1:
             logger.debug("ToolCall run out, resign a new request to request stop sign.")
+            # NOTE:因为 ToolCall 破坏了递归的链式调用，所以这里不再继续调用
+            """
             await create_snapshot(
                 task=task,
                 tool_calls_pending_now=pending_task,
                 memory_able=True,
                 channel=task.receiver.platform,
             )
-        # 运行函数, 传递模型的信息，以及上一条的结果的openai raw信息
+            """
+            pass
+            # 运行函数, 传递模型的信息，以及上一条的结果的openai raw信息
         run_result = await _tool_obj.load(
             task=task,
             receiver=task.receiver,

app/receiver/receiver_client.py

@@ -6,7 +6,6 @@
 #####
 # This file is not a top-level schematic file!
 #####
-
 import os
 import time
 from abc import ABCMeta, abstractmethod
@@ -59,8 +58,8 @@ async def generate_authorization(
 
     tool = tool_object()  # 实例化
     icon = "🌟"
-
-    if tool.require_auth:
+    env_map = {name: secrets.get(name, None) for name in tool.env_list}
+    if tool.require_auth(env_map):
         icon = "🔐"
         auth_key = str(shortuuid.uuid()[0:5].upper())
         authorization_map[auth_key] = tool_invocation
@@ -78,7 +77,8 @@ async def generate_authorization(
         escaped_env_vars = [
             f"`{formatting.escape_markdown(name)}`" for name in missing_env_vars
         ]
-        function_tips.append(f"🦴 Env required:  {','.join(escaped_env_vars)} ")
+        if escaped_env_vars:
+            function_tips.append(f"🦴 Env required:  {','.join(escaped_env_vars)} ")
         help_docs = tool.env_help_docs(missing_env_vars)
         if help_docs:
             function_tips.append(help_docs)
@@ -294,10 +294,11 @@ async def deal_message(self, message) -> Tuple:
         :param message: 消息
         :return: 任务，中间件，路由类型，是否释放函数快照
         """
-        logger.debug("Received MQ Message")
+        logger.debug(f"Received MQ Message 📩{message.message_id}")
         task_head: TaskHeader = TaskHeader.model_validate_json(
             json_data=message.body.decode("utf-8")
         )
+        logger.debug(f"Received Task:{task_head.model_dump_json(indent=2)}")
         router = task_head.task_sign.router
         # Deliver 直接转发
         if router == Router.DELIVER:
@@ -328,7 +329,7 @@ async def deal_message(self, message) -> Tuple:
                 task=task_head,
                 intercept_function=True,
                 disable_tool=True,
-                remember=False,
+                remember=True,
             )
             return (
                 task_head,
@@ -347,8 +348,8 @@ async def deal_message(self, message) -> Tuple:
             await self._flash(
                 task=task_head,
                 llm=llm_middleware,
-                remember=True,
                 intercept_function=True,
+                remember=True,
             )
             return (
                 task_head,
@@ -383,27 +384,38 @@ async def on_message(self, message: AbstractIncomingMessage):
                     data = snap_data.data
                     renew_snap_data = []
                     for task in data:
-                        if not task.snapshot_credential and not task.processed:
-                            if task.expire_at < int(time.time()):
-                                logger.info(
-                                    f"🧀 Expire snapshot {task.snap_uuid} at {router}"
-                                )
-                                continue
-                            try:
-                                await Task.create_and_send(
-                                    queue_name=task.channel, task=task.snapshot_data
-                                )
-                            except Exception as e:
-                                logger.exception(f"Response to snapshot error {e}")
+                        if task.expire_at < int(time.time()):
+                            logger.info(
+                                f"🧀 Expire snapshot {task.snap_uuid} at {router}"
+                            )
+                            # 跳过过期的任何任务
+                            continue
+                        # 不是认证任务
+                        if not task.snapshot_credential:
+                            # 没有被处理
+                            if not task.processed:
+                                try:
+                                    # await asyncio.sleep(10)
+                                    logger.debug(
+                                        f"🧀 Send snapshot {task.snap_uuid} at {router}"
+                                    )
+                                    await Task.create_and_send(
+                                        queue_name=task.channel, task=task.snapshot_data
+                                    )
+                                except Exception as e:
+                                    logger.exception(f"Response to snapshot error {e}")
+                                else:
+                                    logger.info(
+                                        f"🧀 Response to snapshot {task.snap_uuid} at {router}"
+                                    )
+                                finally:
+                                    task.processed_at = int(time.time())
+                                    # renew_snap_data.append(task)
                             else:
-                                logger.info(
-                                    f"🧀 Response to snapshot {task.snap_uuid} at {router}"
-                                )
-                            finally:
-                                task.processed_at = int(time.time())
-                                renew_snap_data.append(task)
+                                # 被处理过的任务。不再处理
+                                pass
                         else:
-                            task.processed_at = None
+                            # 认证任务
                             renew_snap_data.append(task)
                     snap_data.data = renew_snap_data
                     await global_snapshot_storage.write(

app/sender/discord/__init__.py

@@ -336,7 +336,7 @@ async def listen_help_command(ctx: crescent.Context):
         @crescent.command(dm_enabled=True, name="auth", description="auth [credential]")
         async def listen_auth_command(ctx: crescent.Context, credential: str):
             try:
-                await auth_reloader(
+                result = await auth_reloader(
                     snapshot_credential=credential,
                     user_id=f"{ctx.user.id}",
                     platform=__sender__,
@@ -347,7 +347,10 @@ async def listen_auth_command(ctx: crescent.Context, credential: str):
                 )
                 logger.error(f"[270563]auth_reloader failed {exc}")
             else:
-                message = "🪄 Auth Pass"
+                if result:
+                    message = "🪄 Auth Pass"
+                else:
+                    message = "You dont have this snapshot"
             return await ctx.respond(content=message, ephemeral=True)
 
         @client.include

app/sender/discord/event.py

@@ -18,7 +18,7 @@ def help_message():
     `/auth` - activate a task (my power)
     `/login` - login
     `/login_via_url` - login via url
-    `/env` - set environment variable
+    `/env` - set environment variable, split by ; ,  use `/env ENV=NONE` to disable a env.
 
 **Please confirm that that bot instance is secure, some plugins may be dangerous on unsafe instance.**
 """.format(prefix=BotSetting.prefix)

app/sender/kook/__init__.py

@@ -356,19 +356,24 @@ async def listen_tool_command(msg: Message):
         @bot.command(name="auth")
         async def listen_auth_command(msg: Message, credential: str):
             try:
-                await auth_reloader(
+                result = await auth_reloader(
                     snapshot_credential=credential,
                     user_id=f"{msg.author_id}",
                     platform=__sender__,
                 )
             except Exception as e:
-                message = (
+                auth_result = (
                     "❌ Auth failed,You dont have permission or the task do not exist"
                 )
-                logger.error(f"[2753383]auth_reloader failed:{e}")
+                logger.info(f"Auth failed {e}")
             else:
-                message = "🪄 Auth Pass"
-            return await msg.reply(content=message, is_temp=True, type=MessageTypes.KMD)
+                if result:
+                    auth_result = "🪄 Snapshot released"
+                else:
+                    auth_result = "You dont have this snapshot"
+            return await msg.reply(
+                content=auth_result, is_temp=True, type=MessageTypes.KMD
+            )
 
         @bot.command(name="env")
         async def listen_env_command(msg: Message, env_string: str):

app/sender/kook/event.py

@@ -37,7 +37,7 @@ def help_message():
     `/auth` - activate a task (my power)
     `/login` - login openai
     `/login_via_url` - login via provider url
-    `/env` - set environment variable
+    `/env` - set environment variable, split by ; ,  use `/env ENV=NONE` to disable a env.
 
 **Please confirm that that bot instance is secure, some plugins may be dangerous on unsafe instance.**
 """.format(prefix=BotSetting.prefix)

app/sender/slack/__init__.py

@@ -331,7 +331,7 @@ async def listen_tool_command(ack: AsyncAck, respond: AsyncRespond, command):
 
         async def auth_chain(uuid, user_id):
             try:
-                await auth_reloader(
+                result = await auth_reloader(
                     snapshot_credential=uuid,
                     user_id=f"{user_id}",
                     platform=__sender__,
@@ -340,9 +340,12 @@ async def auth_chain(uuid, user_id):
                 auth_result = (
                     "❌ Auth failed,You dont have permission or the task do not exist"
                 )
-                logger.info(f"[3031]auth_reloader failed {e}")
+                logger.info(f"Auth failed {e}")
             else:
-                auth_result = "🪄 Auth Pass"
+                if result:
+                    auth_result = "🪄 Snapshot released"
+                else:
+                    auth_result = "You dont have this snapshot"
             return auth_result
 
         @bot.command(command="/auth")

app/sender/slack/event.py

@@ -24,7 +24,7 @@ def help_message():
 `/clear` - forget...you
 `/auth` - activate a task (my power),but outside the thread
 `/login` - login via url or raw
-`/env` - set environment variable
+`/env` - set environment variable, split by ; ,  use `/env ENV=NONE` to disable a env.
 
 Make sure you invite me before you call me in channel, wink~
 

app/sender/telegram/__init__.py

@@ -341,7 +341,7 @@ async def listen_auth_command(message: types.Message):
             if not _arg:
                 return None
             try:
-                await auth_reloader(
+                result = await auth_reloader(
                     snapshot_credential=_arg,
                     user_id=f"{message.from_user.id}",
                     platform=__sender__,
@@ -352,7 +352,10 @@ async def listen_auth_command(message: types.Message):
                 )
                 logger.info(f"Auth failed {e}")
             else:
-                auth_result = "🪄 Snapshot released"
+                if result:
+                    auth_result = "🪄 Snapshot released"
+                else:
+                    auth_result = "You dont have this snapshot"
             return await bot.reply_to(message, text=convert(auth_result))
 
         @bot.message_handler(

app/sender/telegram/event.py

@@ -17,7 +17,7 @@ def help_message():
 
 Private Chat Only:
     /login - login via url or something
-    /env - 配置变量,use as shell
+    /env - 配置变量 split by ; ,  use `/env ENV=NONE` to disable a env.
 
 !Please confirm that that bot instance is secure, some plugins may be dangerous on unsafe instance.
 """

llmkira/extra/plugins/alarm/__init__.py

@@ -66,10 +66,12 @@ class AlarmTool(BaseTool):
     pattern: Optional[re.Pattern] = re.compile(
         r"(\d+)(分钟|小时|天|周|月|年)后提醒我(.*)"
     )
-    require_auth: bool = True
 
     # env_required: list = ["SCHEDULER", "TIMEZONE"]
 
+    def require_auth(self, env_map: dict) -> bool:
+        return True
+
     def func_message(self, message_text, **kwargs):
         """
         如果合格则返回message，否则返回None，表示不处理