Merge pull request #386 from LlmKira/main

Build Docker

Dockerfile

@@ -19,13 +19,12 @@ RUN apt update && \
     apt install -y ffmpeg && \
     pip install pdm
 
-VOLUME ["/redis", "/rabbitmq", "/mongodb", "/run.log", "/config_dir"]
+VOLUME ["/redis", "/rabbitmq", "/mongodb", "/run.log", ".cache",".montydb",".snapshot"]
 
 WORKDIR /app
 COPY --from=builder /project/.venv /app/.venv
 
 COPY pm2.json ./
-COPY config_dir ./config_dir
 COPY . /app
 
 CMD [ "pm2-runtime", "pm2.json" ]

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/middleware/llm_task.py

@@ -42,6 +42,16 @@ def unique_function(tools: List[Tool]):
     return functions
 
 
+def mock_tool_message(assistant_message: AssistantMessage, mock_content: str):
+    _tool_message = []
+    if assistant_message.tool_calls:
+        for tool_call in assistant_message.tool_calls:
+            _tool_message.append(
+                ToolMessage(content=mock_content, tool_call_id=tool_call.id)
+            )
+    return _tool_message
+
+
 async def validate_mock(messages: List[Message]):
     """
     所有的具有 tool_calls 的 AssistantMessage 后面必须有对应的 ToolMessage 响应，其他消息类型按照原顺序
@@ -90,6 +100,8 @@ def pair_check(_messages):
             else:
                 new_list.append(_messages[i])
         new_list.append(_messages[-1])
+        if isinstance(_messages[-1], AssistantMessage) and _messages[-1].tool_calls:
+            new_list.extend(mock_tool_message(_messages[-1], "[On Queue]"))
         return new_list
 
     final_messages = pair_check(paired_messages)

app/receiver/function.py

@@ -22,6 +22,19 @@
 from llmkira.task.schema import EventMessage, Location, Sign, Snapshot
 from llmkira.task.snapshot import global_snapshot_storage, SnapData
 
+# 记录上次调用时间的字典
+TOOL_CALL_LAST_TIME = {}
+
+
+def has_been_called_recently(userid, n_seconds):
+    current_time = time.time()
+    if userid in TOOL_CALL_LAST_TIME:
+        last_call_time = TOOL_CALL_LAST_TIME[userid]
+        if current_time - last_call_time <= n_seconds:
+            return True
+    TOOL_CALL_LAST_TIME[userid] = current_time
+    return False
+
 
 async def append_snapshot(
     snapshot_credential: Optional[str],
@@ -88,7 +101,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 +168,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 +200,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.")
-            await create_snapshot(
-                task=task,
-                tool_calls_pending_now=pending_task,
-                memory_able=True,
-                channel=task.receiver.platform,
-            )
-        # 运行函数, 传递模型的信息，以及上一条的结果的openai raw信息
+            if has_been_called_recently(userid=task.receiver.uid, n_seconds=5):
+                logger.debug(
+                    "ToolCall run out, resign a new request to request stop sign."
+                )
+                await create_snapshot(
+                    task=task,
+                    tool_calls_pending_now=pending_task,
+                    memory_able=True,
+                    channel=task.receiver.platform,
+                )
+                # 运行函数, 传递模型的信息，以及上一条的结果的openai raw信息
         run_result = await _tool_obj.load(
             task=task,
             receiver=task.receiver,
@@ -220,28 +238,31 @@ async def process_function_call(self, message: AbstractIncomingMessage):
         if os.getenv("STOP_REPLY"):
             logger.warning("🚫 STOP_REPLY is set in env, stop reply message")
             return None
-        task: TaskHeader = TaskHeader.model_validate_json(
-            json_data=message.body.decode("utf-8")
-        )
-        logger.debug(f"[552351] Received A Function Call from {task.receiver.platform}")
-        # Get Function Call
-        pending_task: ToolCall = await task.task_sign.get_pending_tool_call(
-            tool_calls_pending_now=task.task_sign.snapshot_credential,
-            return_default_if_empty=True,
+        logger.debug(
+            f"[552351] Received A Function Call from {message.body.decode('utf-8')}"
         )
-        if not pending_task:
-            return logger.debug("But No ToolCall")
-        logger.debug("Received A ToolCall")
-        try:
-            await self.run_pending_task(task=task, pending_task=pending_task)
-        except Exception as e:
-            await task.task_sign.complete_task(
-                tool_calls=pending_task, success_or_not=False, run_result=str(e)
+        task: TaskHeader = TaskHeader.model_validate_json(message.body.decode("utf-8"))
+        RUN_LIMIT = 6
+        while task.task_sign.tool_calls_pending and RUN_LIMIT > 0:
+            RUN_LIMIT -= 1
+            # Get Function Call
+            pending_task: ToolCall = await task.task_sign.get_pending_tool_call(
+                tool_calls_pending_now=task.task_sign.snapshot_credential,
+                return_default_if_empty=True,
             )
-            logger.error(f"Function Call Error {e}")
-            raise e
-        finally:
-            logger.trace("Function Call Finished")
+            if not pending_task:
+                return logger.debug("But No ToolCall")
+            logger.debug("Received A ToolCall")
+            try:
+                await self.run_pending_task(task=task, pending_task=pending_task)
+            except Exception as e:
+                await task.task_sign.complete_task(
+                    tool_calls=pending_task, success_or_not=False, run_result=str(e)
+                )
+                logger.error(f"Function Call Error {e}")
+                raise e
+            finally:
+                logger.trace("Function Call Finished")
 
     async def on_message(self, message: AbstractIncomingMessage):
         """
@@ -252,7 +273,7 @@ async def on_message(self, message: AbstractIncomingMessage):
         try:
             await self.process_function_call(message=message)
         except Exception as e:
-            logger.exception(f"Function Receiver Error {e}")
+            logger.exception(f"Function Receiver Error:{e}")
             await message.reject(requeue=False)
             raise e
         else: