Nanobot 子代理架构深度分析

bg_task.add_done_callback(lambda _: self._running_tasks.pop(task_id, None))

async def _run_subagent(
    self,
    task_id: str,
    task: str,
    label: str,
    origin: dict[str, str],
) -> None:

tools = ToolRegistry()
tools.register(ReadFileTool())
tools.register(WriteFileTool())
tools.register(ListDirTool())
tools.register(ExecTool(working_dir=str(self.workspace)))
tools.register(WebSearchTool(api_key=self.brave_api_key))
tools.register(WebFetchTool())

system_prompt = self._build_subagent_prompt(task)
messages: list[dict[str, Any]] = [
    {"role": "system", "content": system_prompt},
    {"role": "user", "content": task},
]

max_iterations = 15
iteration = 0
final_result: str | None = None

while iteration < max_iterations:
    iteration += 1
    response = await self.provider.chat(...)
    
    if response.has_tool_calls:
        # 处理工具调用
        ...
    else:
        final_result = response.content
        break

except Exception as e:
    error_msg = f"Error: {str(e)}"
    logger.error(f"Subagent [{task_id}] failed: {e}")
    await self._announce_result(task_id, label, task, error_msg, origin, "error")

def _build_subagent_prompt(self, task: str) -> str:
    return f"""# Subagent

You are a subagent spawned by the main agent to complete a specific task.

## Your Task
{task}

## Rules
1. Stay focused - complete only the assigned task, nothing else
2. Your final response will be reported back to the main agent
3. Do not initiate conversations or take on side tasks
4. Be concise but informative in your findings

## What You Can Do
- Read and write files in the workspace
- Execute shell commands
- Search the web and fetch web pages
- Complete the task thoroughly

## What You Cannot Do
- Send messages directly to users (no message tool available)
- Spawn other subagents
- Access the main agent's conversation history

## Workspace
Your workspace is at: {self.workspace}

When you have completed the task, provide a clear summary of your findings or actions."""

async def _announce_result(
    self,
    task_id: str,
    label: str,
    task: str,
    result: str,
    origin: dict[str, str],
    status: str,
) -> None:
    status_text = "completed successfully" if status == "ok" else "failed"

    announce_content = f"""[Subagent '{label}' {status_text}]

Task: {task}

Result:
{result}

Summarize this naturally for the user. Keep it brief (1-2 sentences). Do not mention technical details like "subagent" or task IDs."""

    msg = InboundMessage(
        channel="system",
        sender_id="subagent",
        chat_id=f"{origin['channel']}:{origin['chat_id']}",
        content=announce_content,
    )

    await self.bus.publish_inbound(msg)

msg = InboundMessage(
    channel="system",              # 特殊通道标识
    sender_id="subagent",         # 发送者标识
    chat_id=f"{origin['channel']}:{origin['chat_id']}",  # 路由信息
    content=announce_content,
)

# Handle system messages (subagent announces)
if msg.channel == "system":
    return await self._process_system_message(msg)

# Parse origin from chat_id (format: "channel:chat_id")
if ":" in msg.chat_id:
    parts = msg.chat_id.split(":", 1)
    origin_channel = parts[0]
    origin_chat_id = parts[1]

def set_context(self, channel: str, chat_id: str) -> None:
    self._origin_channel = channel
    self._origin_chat_id = chat_id

return await self._manager.spawn(
    task=task,
    label=label,
    origin_channel=self._origin_channel,
    origin_chat_id=self._origin_chat_id,
)

origin = {
    "channel": origin_channel,
    "chat_id": origin_chat_id,
}

chat_id=f"{origin['channel']}:{origin['chat_id']}",

return OutboundMessage(
    channel=origin_channel,
    chat_id=origin_chat_id,
    content=final_content
)

messages = self.context.build_messages(
    history=session.get_history(),  # 原始对话历史
    current_message=msg.content     # 公告内容
)

tools = ToolRegistry()
tools.register(ReadFileTool())
tools.register(WriteFileTool())
tools.register(ListDirTool())
tools.register(ExecTool(...))
tools.register(WebSearchTool(...))
tools.register(WebFetchTool())
# 注意：没有 tools.register(MessageTool())
# 注意：没有 tools.register(SpawnTool())

def _register_default_tools(self) -> None:
    # ... 文件和工具 ...
    tools.register(MessageTool(send_callback=...))
    tools.register(SpawnTool(manager=self.subagents))

tools = ToolRegistry()  # 独立实例
tools.register(ReadFileTool())
# ...

self.tools = ToolRegistry()  # 共享实例
self._register_default_tools()  # 注册所有工具

# 子代理的 messages 数组只在子代理内部使用
messages = [
    {"role": "system", "content": system_prompt},
    {"role": "user", "content": task},
    # ... 迭代添加 ...
]
# 从不传递给主代理

bg_task = asyncio.create_task(
    self._run_subagent(task_id, task, display_label, origin)
)

self._running_tasks[task_id] = bg_task

bg_task.add_done_callback(
    lambda _: self._running_tasks.pop(task_id, None)
)

# 主代理可以同时启动多个子代理
await spawn(task="分析日志文件", label="日志分析")
await spawn(task="下载数据集", label="数据获取")
await spawn(task="运行测试", label="测试执行")

# 所有子代理并行运行
assert subagent_manager.get_running_count() == 3

try:
    return await tool.execute(**params)
except Exception as e:
    return f"Error executing {name}: {str(e)}"

except Exception as e:
    error_msg = f"Error: {str(e)}"
    logger.error(f"Subagent [{task_id}] failed: {e}")
    await self._announce_result(task_id, label, task, error_msg, origin, "error")

# 主代理处理消息时设置上下文
message_tool.set_context(msg.channel, msg.chat_id)
spawn_tool.set_context(msg.channel, msg.chat_id)

# spawn 后立即返回，上下文不受子代理影响
result = await self.tools.execute("spawn", {...})

# 子代理1（从 telegram:123 派生）
chat_id="telegram:123"  → 路由到 telegram:123

# 子代理2（从 discord:456 派生）
chat_id="discord:456"  → 路由到 discord:456

async def spawn(self, ...) -> str:
    if self.get_running_count() >= MAX_CONCURRENT_SUBAGENTS:
        return "Too many subagents running. Please wait."
    # ... 原有逻辑 ...