diff --git a/astrbot/core/agent/context/compressor.py b/astrbot/core/agent/context/compressor.py index 31a0b0b48..bff40a4de 100644 --- a/astrbot/core/agent/context/compressor.py +++ b/astrbot/core/agent/context/compressor.py @@ -130,7 +130,6 @@ def split_history( # Search backward from split_index to find the first user message # This ensures recent_messages starts with a user message (complete turn) while split_index > 0 and non_system_messages[split_index].role != "user": - # TODO: +=1 or -=1 ? calculate by tokens split_index -= 1 # If we couldn't find a user message, keep all messages as recent diff --git a/astrbot/core/astr_agent_tool_exec.py b/astrbot/core/astr_agent_tool_exec.py index 43bc1b819..602278753 100644 --- a/astrbot/core/astr_agent_tool_exec.py +++ b/astrbot/core/astr_agent_tool_exec.py @@ -2,10 +2,9 @@ import asyncio import inspect import json import traceback -import typing as T import uuid -from collections.abc import Sequence -from collections.abc import Set as AbstractSet +from collections.abc import AsyncGenerator, Awaitable, Callable, Sequence, Set +from typing import Any import mcp @@ -41,14 +40,14 @@ from astrbot.core.utils.string_utils import normalize_and_dedupe_strings class FunctionToolExecutor(BaseFunctionToolExecutor[AstrAgentContext]): @classmethod - def _collect_image_urls_from_args(cls, image_urls_raw: T.Any) -> list[str]: + def _collect_image_urls_from_args(cls, image_urls_raw: Any) -> list[str]: if image_urls_raw is None: return [] if isinstance(image_urls_raw, str): return [image_urls_raw] - if isinstance(image_urls_raw, (Sequence, AbstractSet)) and not isinstance( + if isinstance(image_urls_raw, (Sequence, Set)) and not isinstance( image_urls_raw, (str, bytes, bytearray) ): return [item for item in image_urls_raw if isinstance(item, str)] @@ -88,7 +87,7 @@ class FunctionToolExecutor(BaseFunctionToolExecutor[AstrAgentContext]): async def _collect_handoff_image_urls( cls, run_context: ContextWrapper[AstrAgentContext], - image_urls_raw: T.Any, + image_urls_raw: Any, ) -> list[str]: candidates: list[str] = [] candidates.extend(cls._collect_image_urls_from_args(image_urls_raw)) @@ -308,7 +307,7 @@ class FunctionToolExecutor(BaseFunctionToolExecutor[AstrAgentContext]): run_context: ContextWrapper[AstrAgentContext], *, image_urls_prepared: bool = False, - **tool_args: T.Any, + **tool_args: Any, ): tool_args = dict(tool_args) input_ = tool_args.get("input") @@ -513,10 +512,10 @@ class FunctionToolExecutor(BaseFunctionToolExecutor[AstrAgentContext]): task_id: str, tool_name: str, result_text: str, - tool_args: dict[str, T.Any], + tool_args: dict[str, Any], note: str, summary_name: str, - extra_result_fields: dict[str, T.Any] | None = None, + extra_result_fields: dict[str, Any] | None = None, ) -> None: from astrbot.core.astr_main_agent import ( MainAgentBuildConfig, @@ -667,23 +666,30 @@ class FunctionToolExecutor(BaseFunctionToolExecutor[AstrAgentContext]): yield mcp.types.CallToolResult(content=[text_content]) else: # NOTE: Tool 在这里直接请求发送消息给用户 - # TODO: 是否需要判断 event.get_result() 是否为空? - # 如果为空,则说明没有发送消息给用户,并且返回值为空,将返回一个特殊的 TextContent,其内容如"工具没有返回内容" - if res := run_context.context.event.get_result(): - if res.chain: - try: - await event.send( - MessageChain( - chain=res.chain, - type="tool_direct_result", - ) + res = run_context.context.event.get_result() + if res and res.chain: + try: + await event.send( + MessageChain( + chain=res.chain, + type="tool_direct_result", ) - except Exception as e: - logger.error( - f"Tool 直接发送消息失败: {e}", - exc_info=True, + ) + except Exception as e: + logger.error( + f"Tool 直接发送消息失败: {e}", + exc_info=True, + ) + yield None + else: + yield mcp.types.CallToolResult( + content=[ + mcp.types.TextContent( + type="text", + text="Tool executed successfully with no output.", ) - yield None + ] + ) except asyncio.TimeoutError: raise Exception( f"tool {tool.name} execution timeout after {tool_call_timeout or run_context.tool_call_timeout} seconds.", @@ -706,15 +712,15 @@ class FunctionToolExecutor(BaseFunctionToolExecutor[AstrAgentContext]): async def call_local_llm_tool( context: ContextWrapper[AstrAgentContext], - handler: T.Callable[ + handler: Callable[ ..., - T.Awaitable[MessageEventResult | mcp.types.CallToolResult | str | None] - | T.AsyncGenerator[MessageEventResult | CommandResult | str | None, None], + Awaitable[MessageEventResult | mcp.types.CallToolResult | str | None] + | AsyncGenerator[MessageEventResult | CommandResult | str | None, None], ], method_name: str, *args, **kwargs, -) -> T.AsyncGenerator[T.Any, None]: +) -> AsyncGenerator[Any, None]: """执行本地 LLM 工具的处理函数并处理其返回结果""" ready_to_call = None # 一个协程或者异步生成器 diff --git a/astrbot/core/platform/sources/kook/kook_client.py b/astrbot/core/platform/sources/kook/kook_client.py index 32874f78a..a94600a0c 100644 --- a/astrbot/core/platform/sources/kook/kook_client.py +++ b/astrbot/core/platform/sources/kook/kook_client.py @@ -261,8 +261,6 @@ class KookClient: if code == 0: self.session_id = data.session_id logger.info(f"[KOOK] 握手成功，session_id: {self.session_id}") - # TODO 重置重连延迟 - # self.reconnect_delay = 1 else: logger.error(f"[KOOK] 握手失败，错误码: {code}") if code == 40103: # token过期 diff --git a/astrbot/core/utils/session_waiter.py b/astrbot/core/utils/session_waiter.py index b327a6184..c524aa8f2 100644 --- a/astrbot/core/utils/session_waiter.py +++ b/astrbot/core/utils/session_waiter.py @@ -121,6 +121,9 @@ class SessionWaiter: self._lock = asyncio.Lock() """需要保证一个 session 同时只有一个 trigger""" + self.curr_task: asyncio.Task | None = None + """当前正在执行的处理任务""" + async def register_wait( self, handler: Callable[[SessionController, AstrMessageEvent], Awaitable[Any]], @@ -148,6 +151,10 @@ class SessionWaiter: FILTERS.remove(self.session_filter) except ValueError: pass + + if self.curr_task and not self.curr_task.done(): + self.curr_task.cancel() + self.session_controller.stop(error) @classmethod @@ -163,12 +170,21 @@ class SessionWaiter: session.session_controller.history_chains.append( [copy.deepcopy(comp) for comp in event.get_messages()], ) + + async def _task(): + try: + assert session.handler is not None + await session.handler(session.session_controller, event) + except asyncio.CancelledError: + pass + except Exception as e: + session.session_controller.stop(e) + + session.curr_task = asyncio.create_task(_task()) try: - # TODO: 这里使用 create_task，跟踪 task，防止超时后这里 handler 仍然在执行 - assert session.handler is not None - await session.handler(session.session_controller, event) - except Exception as e: - session.session_controller.stop(e) + await session.curr_task + except asyncio.CancelledError: + pass def session_waiter(timeout: int = 30, record_history_chains: bool = False):