Nanobot 异步化工具执行设计深度分析

class Tool(ABC):
    @abstractmethod
    async def execute(self, **kwargs: Any) -> str:
        """
        Execute the tool with given parameters.

        Returns:
            String result of the tool execution.
        """
        pass

async def execute(self, name: str, params: dict[str, Any]) -> str:
    tool = self._tools.get(name)
    if not tool:
        return f"Error: Tool '{name}' not found"
    
    try:
        return await tool.execute(**params)
    except Exception as e:
        return f"Error executing {name}: {str(e)}"

async def run(self) -> None:
    """Run the agent loop, processing messages from the bus."""
    self._running = True
    logger.info("Agent loop started")
    
    while self._running:
        try:
            msg = await asyncio.wait_for(
                self.bus.consume_inbound(),
                timeout=1.0
            )
            
            response = await self._process_message(msg)
            if response:
                await self.bus.publish_outbound(response)
        except asyncio.TimeoutError:
            continue

async def execute(self, command: str, working_dir: str | None = None, **kwargs: Any) -> str:
    cwd = working_dir or self.working_dir or os.getcwd()
    
    try:
        process = await asyncio.create_subprocess_shell(
            command,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
            cwd=cwd,
        )
        
        try:
            stdout, stderr = await asyncio.wait_for(
                process.communicate(),
                timeout=self.timeout
            )
        except asyncio.TimeoutError:
            process.kill()
            return f"Error: Command timed out after {self.timeout} seconds"
        
        # ... 处理输出 ...

try:
    stdout, stderr = await asyncio.wait_for(
        process.communicate(),
        timeout=self.timeout
    )
except asyncio.TimeoutError:
    process.kill()  # 立即终止进程
    return f"Error: Command timed out after {self.timeout} seconds"

output_parts = []

if stdout:
    output_parts.append(stdout.decode("utf-8", errors="replace"))

if stderr:
    stderr_text = stderr.decode("utf-8", errors="replace")
    if stderr_text.strip():
        output_parts.append(f"STDERR:\n{stderr_text}")

if process.returncode != 0:
    output_parts.append(f"\nExit code: {process.returncode}")

result = "\n".join(output_parts) if output_parts else "(no output)"

# 截断超长输出
max_len = 10000
if len(result) > max_len:
    result = result[:max_len] + f"\n... (truncated, {len(result) - max_len} more chars)"

async def execute(self, query: str, count: int | None = None, **kwargs: Any) -> str:
    if not self.api_key:
        return "Error: BRAVE_API_KEY not configured"
    
    try:
        n = min(max(count or self.max_results, 1), 10)
        async with httpx.AsyncClient() as client:
            r = await client.get(
                "https://api.search.brave.com/res/v1/web/search",
                params={"q": query, "count": n},
                headers={"Accept": "application/json", "X-Subscription-Token": self.api_key},
                timeout=10.0
            )
            r.raise_for_status()
        
        # ... 处理结果 ...

async def execute(self, url: str, extractMode: str = "markdown", maxChars: int | None = None, **kwargs: Any) -> str:
    from readability import Document
    
    max_chars = maxChars or self.max_chars
    
    try:
        async with httpx.AsyncClient() as client:
            r = await client.get(url, headers={"User-Agent": USER_AGENT}, follow_redirects=True, timeout=30.0)
            r.raise_for_status()
        
        # ... 内容提取 ...

async with httpx.AsyncClient() as client:
    r = await client.get(url, ...)

# 工具内部 (filesystem.py:43-46)
except PermissionError:
    return f"Error: Permission denied: {path}"
except Exception as e:
    return f"Error reading file: {str(e)}"

# 注册表层 (registry.py:56-59)
try:
    return await tool.execute(**params)
except Exception as e:
    return f"Error executing {name}: {str(e)}"

# Loop 层 (loop.py:107-108)
except Exception as e:
    logger.error(f"Error processing message: {e}")
    # 发送错误响应给用户

result = await self.tools.execute(tool_call.name, tool_call.arguments)
messages = self.context.add_tool_result(
    messages, tool_call.id, tool_call.name, result
)

for tool_call in response.tool_calls:
    args_str = json.dumps(tool_call.arguments)
    logger.debug(f"Executing tool: {tool_call.name} with arguments: {args_str}")
    result = await self.tools.execute(tool_call.name, tool_call.arguments)
    messages = self.context.add_tool_result(
        messages, tool_call.id, tool_call.name, result
    )

tasks = []
for tool_call in response.tool_calls:
    task = self.tools.execute(tool_call.name, tool_call.arguments)
    tasks.append(task)

results = await asyncio.gather(*tasks)

for tool_call, result in zip(response.tool_calls, results):
    messages = self.context.add_tool_result(
        messages, tool_call.id, tool_call.name, result
    )

from asyncio import Semaphore

async def execute_with_limit(registry: ToolRegistry, name: str, params: dict, semaphore: Semaphore):
    async with semaphore:
        return await registry.execute(name, params)

# 使用方式
semaphore = Semaphore(3)  # 最多 3 个并发
tasks = [
    execute_with_limit(self.tools, tc.name, tc.arguments, semaphore)
    for tc in response.tool_calls
]

class MessageTool(Tool):
    def __init__(
        self, 
        send_callback: Callable[[OutboundMessage], Awaitable[None]] | None = None,
        default_channel: str = "",
        default_chat_id: str = ""
    ):
        self._send_callback = send_callback
        self._default_channel = default_channel
        self._default_chat_id = default_chat_id
    
    def set_context(self, channel: str, chat_id: str) -> None:
        """Set the current message context."""
        self._default_channel = channel
        self._default_chat_id = chat_id

class SpawnTool(Tool):
    def __init__(self, manager: "SubagentManager"):
        self._manager = manager
        self._origin_channel = "cli"
        self._origin_chat_id = "direct"
    
    def set_context(self, channel: str, chat_id: str) -> None:
        """Set the origin context for subagent announcements."""
        self._origin_channel = channel
        self._origin_chat_id = chat_id

# Update tool contexts
message_tool = self.tools.get("message")
if isinstance(message_tool, MessageTool):
    message_tool.set_context(msg.channel, msg.chat_id)

spawn_tool = self.tools.get("spawn")
if isinstance(spawn_tool, SpawnTool):
    spawn_tool.set_context(msg.channel, msg.chat_id)