wylab/nanobot
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

feat: add sub-agent system

Browse Source
This commit is contained in:
Re-bin
2026-02-01 16:28:45 +00:00
parent c8a1190064
commit 051a97fa4e
5 changed files with 417 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
.gitignore vendored Normal file
View File

@@ -0,0 +1,3 @@
.assets
.env
*.pyc

6
nanobot/agent/context.py
View File

@@ -80,6 +80,7 @@ You are nanobot, a helpful AI assistant. You have access to tools that allow you
- Execute shell commands
- Search the web and fetch web pages
- Send messages to users on chat channels
- Spawn subagents for complex background tasks
## Current Time
{now}
@@ -184,10 +185,7 @@ When remembering something, write to {workspace_path}/memory/MEMORY.md"""
Returns:
Updated message list.
"""
msg: dict[str, Any] = {"role": "assistant"}
if content:
msg["content"] = content
msg: dict[str, Any] = {"role": "assistant", "content": content or ""}
if tool_calls:
msg["tool_calls"] = tool_calls

115
nanobot/agent/loop.py
View File

@@ -16,6 +16,8 @@ from nanobot.agent.tools.filesystem import ReadFileTool, WriteFileTool, EditFile
from nanobot.agent.tools.shell import ExecTool
from nanobot.agent.tools.web import WebSearchTool, WebFetchTool
from nanobot.agent.tools.message import MessageTool
from nanobot.agent.tools.spawn import SpawnTool
from nanobot.agent.subagent import SubagentManager
from nanobot.session.manager import SessionManager
@@ -50,6 +52,13 @@ class AgentLoop:
self.context = ContextBuilder(workspace)
self.sessions = SessionManager(workspace)
self.tools = ToolRegistry()
self.subagents = SubagentManager(
provider=provider,
workspace=workspace,
bus=bus,
model=self.model,
brave_api_key=brave_api_key,
)
self._running = False
self._register_default_tools()
@@ -72,6 +81,10 @@ class AgentLoop:
# Message tool
message_tool = MessageTool(send_callback=self.bus.publish_outbound)
self.tools.register(message_tool)
# Spawn tool (for subagents)
spawn_tool = SpawnTool(manager=self.subagents)
self.tools.register(spawn_tool)
async def run(self) -> None:
"""Run the agent loop, processing messages from the bus."""
@@ -117,16 +130,25 @@ class AgentLoop:
Returns:
The response message, or None if no response needed.
"""
# Handle system messages (subagent announces)
# The chat_id contains the original "channel:chat_id" to route back to
if msg.channel == "system":
return await self._process_system_message(msg)
logger.info(f"Processing message from {msg.channel}:{msg.sender_id}")
# Get or create session
session = self.sessions.get_or_create(msg.session_key)
# Update message tool context
# 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)
# Build initial messages (use get_history for LLM-formatted messages)
messages = self.context.build_messages(
history=session.get_history(),
@@ -191,6 +213,97 @@ class AgentLoop:
content=final_content
)
async def _process_system_message(self, msg: InboundMessage) -> OutboundMessage | None:
"""
Process a system message (e.g., subagent announce).
The chat_id field contains "original_channel:original_chat_id" to route
the response back to the correct destination.
"""
logger.info(f"Processing system message from {msg.sender_id}")
# 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]
else:
# Fallback
origin_channel = "cli"
origin_chat_id = msg.chat_id
# Use the origin session for context
session_key = f"{origin_channel}:{origin_chat_id}"
session = self.sessions.get_or_create(session_key)
# Update tool contexts
message_tool = self.tools.get("message")
if isinstance(message_tool, MessageTool):
message_tool.set_context(origin_channel, origin_chat_id)
spawn_tool = self.tools.get("spawn")
if isinstance(spawn_tool, SpawnTool):
spawn_tool.set_context(origin_channel, origin_chat_id)
# Build messages with the announce content
messages = self.context.build_messages(
history=session.get_history(),
current_message=msg.content
)
# Agent loop (limited for announce handling)
iteration = 0
final_content = None
while iteration < self.max_iterations:
iteration += 1
response = await self.provider.chat(
messages=messages,
tools=self.tools.get_definitions(),
model=self.model
)
if response.has_tool_calls:
tool_call_dicts = [
{
"id": tc.id,
"type": "function",
"function": {
"name": tc.name,
"arguments": json.dumps(tc.arguments)
}
}
for tc in response.tool_calls
]
messages = self.context.add_assistant_message(
messages, response.content, tool_call_dicts
)
for tool_call in response.tool_calls:
logger.debug(f"Executing tool: {tool_call.name}")
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
)
else:
final_content = response.content
break
if final_content is None:
final_content = "Background task completed."
# Save to session (mark as system message in history)
session.add_message("user", f"[System: {msg.sender_id}] {msg.content}")
session.add_message("assistant", final_content)
self.sessions.save(session)
return OutboundMessage(
channel=origin_channel,
chat_id=origin_chat_id,
content=final_content
)
async def process_direct(self, content: str, session_key: str = "cli:direct") -> str:
"""
Process a message directly (for CLI usage).

233
nanobot/agent/subagent.py Normal file
View File

@@ -0,0 +1,233 @@
"""Subagent manager for background task execution."""
import asyncio
import json
import uuid
from pathlib import Path
from typing import Any
from loguru import logger
from nanobot.bus.events import InboundMessage
from nanobot.bus.queue import MessageBus
from nanobot.providers.base import LLMProvider
from nanobot.agent.tools.registry import ToolRegistry
from nanobot.agent.tools.filesystem import ReadFileTool, WriteFileTool, ListDirTool
from nanobot.agent.tools.shell import ExecTool
from nanobot.agent.tools.web import WebSearchTool, WebFetchTool
class SubagentManager:
"""
Manages background subagent execution.
Subagents are lightweight agent instances that run in the background
to handle specific tasks. They share the same LLM provider but have
isolated context and a focused system prompt.
"""
def __init__(
self,
provider: LLMProvider,
workspace: Path,
bus: MessageBus,
model: str | None = None,
brave_api_key: str | None = None,
):
self.provider = provider
self.workspace = workspace
self.bus = bus
self.model = model or provider.get_default_model()
self.brave_api_key = brave_api_key
self._running_tasks: dict[str, asyncio.Task[None]] = {}
async def spawn(
self,
task: str,
label: str | None = None,
origin_channel: str = "cli",
origin_chat_id: str = "direct",
) -> str:
"""
Spawn a subagent to execute a task in the background.
Args:
task: The task description for the subagent.
label: Optional human-readable label for the task.
origin_channel: The channel to announce results to.
origin_chat_id: The chat ID to announce results to.
Returns:
Status message indicating the subagent was started.
"""
task_id = str(uuid.uuid4())[:8]
display_label = label or task[:30] + ("..." if len(task) > 30 else "")
origin = {
"channel": origin_channel,
"chat_id": origin_chat_id,
}
# Create background task
bg_task = asyncio.create_task(
self._run_subagent(task_id, task, display_label, origin)
)
self._running_tasks[task_id] = bg_task
# Cleanup when done
bg_task.add_done_callback(lambda _: self._running_tasks.pop(task_id, None))
logger.info(f"Spawned subagent [{task_id}]: {display_label}")
return f"Subagent [{display_label}] started (id: {task_id}). I'll notify you when it completes."
async def _run_subagent(
self,
task_id: str,
task: str,
label: str,
origin: dict[str, str],
) -> None:
"""Execute the subagent task and announce the result."""
logger.info(f"Subagent [{task_id}] starting task: {label}")
try:
# Build subagent tools (no message tool, no spawn tool)
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())
# Build messages with subagent-specific prompt
system_prompt = self._build_subagent_prompt(task)
messages: list[dict[str, Any]] = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": task},
]
# Run agent loop (limited iterations)
max_iterations = 15
iteration = 0
final_result: str | None = None
while iteration < max_iterations:
iteration += 1
response = await self.provider.chat(
messages=messages,
tools=tools.get_definitions(),
model=self.model,
)
if response.has_tool_calls:
# Add assistant message with tool calls
tool_call_dicts = [
{
"id": tc.id,
"type": "function",
"function": {
"name": tc.name,
"arguments": json.dumps(tc.arguments),
},
}
for tc in response.tool_calls
]
messages.append({
"role": "assistant",
"content": response.content or "",
"tool_calls": tool_call_dicts,
})
# Execute tools
for tool_call in response.tool_calls:
logger.debug(f"Subagent [{task_id}] executing: {tool_call.name}")
result = await tools.execute(tool_call.name, tool_call.arguments)
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"name": tool_call.name,
"content": result,
})
else:
final_result = response.content
break
if final_result is None:
final_result = "Task completed but no final response was generated."
logger.info(f"Subagent [{task_id}] completed successfully")
await self._announce_result(task_id, label, task, final_result, origin, "ok")
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")
async def _announce_result(
self,
task_id: str,
label: str,
task: str,
result: str,
origin: dict[str, str],
status: str,
) -> None:
"""Announce the subagent result to the main agent via the message bus."""
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."""
# Inject as system message to trigger main agent
msg = InboundMessage(
channel="system",
sender_id="subagent",
chat_id=f"{origin['channel']}:{origin['chat_id']}",
content=announce_content,
)
await self.bus.publish_inbound(msg)
logger.debug(f"Subagent [{task_id}] announced result to {origin['channel']}:{origin['chat_id']}")
def _build_subagent_prompt(self, task: str) -> str:
"""Build a focused system prompt for the subagent."""
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."""
def get_running_count(self) -> int:
"""Return the number of currently running subagents."""
return len(self._running_tasks)

65
nanobot/agent/tools/spawn.py Normal file
View File

@@ -0,0 +1,65 @@
"""Spawn tool for creating background subagents."""
from typing import Any, TYPE_CHECKING
from nanobot.agent.tools.base import Tool
if TYPE_CHECKING:
from nanobot.agent.subagent import SubagentManager
class SpawnTool(Tool):
"""
Tool to spawn a subagent for background task execution.
The subagent runs asynchronously and announces its result back
to the main agent when complete.
"""
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
@property
def name(self) -> str:
return "spawn"
@property
def description(self) -> str:
return (
"Spawn a subagent to handle a task in the background. "
"Use this for complex or time-consuming tasks that can run independently. "
"The subagent will complete the task and report back when done."
)
@property
def parameters(self) -> dict[str, Any]:
return {
"type": "object",
"properties": {
"task": {
"type": "string",
"description": "The task for the subagent to complete",
},
"label": {
"type": "string",
"description": "Optional short label for the task (for display)",
},
},
"required": ["task"],
}
async def execute(self, task: str, label: str | None = None, **kwargs: Any) -> str:
"""Spawn a subagent to execute the given task."""
return await self._manager.spawn(
task=task,
label=label,
origin_channel=self._origin_channel,
origin_chat_id=self._origin_chat_id,
)