Nanobot Agent 核心机制深度分析

# nanobot/agent/loop.py:89-121
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:
            # Wait for next message
            msg = await asyncio.wait_for(
                self.bus.consume_inbound(),
                timeout=1.0
            )

            # Process it
            try:
                response = await self._process_message(msg)
                if response:
                    await self.bus.publish_outbound(response)
            except Exception as e:
                logger.error(f"Error processing message: {e}")
                # Send error response
                await self.bus.publish_outbound(OutboundMessage(
                    channel=msg.channel,
                    chat_id=msg.chat_id,
                    content=f"Sorry, I encountered an error: {str(e)}"
                ))
        except asyncio.TimeoutError:
            continue

# nanobot/agent/loop.py:123-216
async def _process_message(self, msg: InboundMessage) -> OutboundMessage | None:
    # Handle system messages (subagent announces)
    if msg.channel == "system":
        return await self._process_system_message(msg)

    # Get or create session
    session = self.sessions.get_or_create(msg.session_key)

    # Update tool contexts (message_tool, spawn_tool)
    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
    messages = self.context.build_messages(
        history=session.get_history(),
        current_message=msg.content,
        media=msg.media if msg.media else None,
    )

    # Agent loop
    iteration = 0
    final_content = None

    while iteration < self.max_iterations:
        iteration += 1

        # Call LLM
        response = await self.provider.chat(
            messages=messages,
            tools=self.tools.get_definitions(),
            model=self.model
        )

        # Handle tool calls
        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 = self.context.add_assistant_message(
                messages, response.content, tool_call_dicts
            )

            # Execute tools
            for tool_call in response.tool_calls:
                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:
            # No tool calls, we're done
            final_content = response.content
            break

    if final_content is None:
        final_content = "I've completed processing but have no response to give."

    # Save to session
    session.add_message("user", msg.content)
    session.add_message("assistant", final_content)
    self.sessions.save(session)

    return OutboundMessage(
        channel=msg.channel,
        chat_id=msg.chat_id,
        content=final_content
    )

# nanobot/agent/context.py:27-70
def build_system_prompt(self, skill_names: list[str] | None = None) -> str:
    parts = []

    # Core identity
    parts.append(self._get_identity())

    # Bootstrap files
    bootstrap = self._load_bootstrap_files()
    if bootstrap:
        parts.append(bootstrap)

    # Memory context
    memory = self.memory.get_memory_context()
    if memory:
        parts.append(f"# Memory\n\n{memory}")

    # Skills - progressive loading
    # 1. Always-loaded skills: include full content
    always_skills = self.skills.get_always_skills()
    if always_skills:
        always_content = self.skills.load_skills_for_context(always_skills)
        if always_content:
            parts.append(f"# Active Skills\n\n{always_content}")

    # 2. Available skills: only show summary
    skills_summary = self.skills.build_skills_summary()
    if skills_summary:
        parts.append(f"""# Skills

The following skills extend your capabilities. To use a skill, read its SKILL.md file using the read_file tool.
Skills with available="false" need dependencies installed first - you can try installing them with apt/brew.

{skills_summary}""")

    return "\n\n---\n\n".join(parts)

# nanobot/agent/context.py:72-101
def _get_identity(self) -> str:
    """Get the core identity section."""
    from datetime import datetime
    now = datetime.now().strftime("%Y-%m-%d %H:%M (%A)")
    workspace_path = str(self.workspace.expanduser().resolve())

    return f"""# nanobot 🐈

You are nanobot, a helpful AI assistant. You have access to tools that allow you to:
- Read, write, and edit files
- 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}

## Workspace
Your workspace is at: {workspace_path}
- Memory files: {workspace_path}/memory/MEMORY.md
- Daily notes: {workspace_path}/memory/YYYY-MM-DD.md
- Custom skills: {workspace_path}/skills/{{skill-name}}/SKILL.md

IMPORTANT: When responding to direct questions or conversations, reply directly with your text response.
Only use the 'message' tool when you need to send a message to a specific chat channel (like WhatsApp).
For normal conversation, just respond with text - do not call the message tool.

Always be helpful, accurate, and concise. When using tools, explain what you're doing.
When remembering something, write to {workspace_path}/memory/MEMORY.md"""

# nanobot/agent/context.py:103-113
def _load_bootstrap_files(self) -> str:
    """Load all bootstrap files from workspace."""
    parts = []

    for filename in self.BOOTSTRAP_FILES:  # ["AGENTS.md", "SOUL.md", "USER.md", "TOOLS.md", "IDENTITY.md"]
        file_path = self.workspace / filename
        if file_path.exists():
            content = file_path.read_text(encoding="utf-8")
            parts.append(f"## {filename}\n\n{content}")

    return "\n\n".join(parts) if parts else ""

# Always Skills (完整加载)
always_skills = self.skills.get_always_skills()
always_content = self.skills.load_skills_for_context(always_skills)
parts.append(f"# Active Skills\n\n{always_content}")

# Available Skills (仅摘要)
skills_summary = self.skills.build_skills_summary()
parts.append(f"""# Skills

The following skills extend your capabilities. To use a skill, read its SKILL.md file using the read_file tool.
Skills with available="false" need dependencies installed first - you can try installing them with apt/brew.

{skills_summary}""")

# nanobot/agent/subagent.py:44-81
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."""
    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."

# nanobot/agent/subagent.py:83-166
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")

# nanobot/agent/subagent.py:168-198
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']}")

# nanobot/agent/subagent.py:200-229
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."""

# nanobot/agent/tools/base.py (推断)
class Tool(ABC):
    """Base class for all tools."""

    @property
    @abstractmethod
    def name(self) -> str:
        """Tool name."""
        pass

    @property
    @abstractmethod
    def description(self) -> str:
        """Tool description for LLM."""
        pass

    @property
    @abstractmethod
    def parameters(self) -> dict[str, Any]:
        """Tool parameters schema (JSON Schema)."""
        pass

    @abstractmethod
    async def execute(self, **kwargs) -> str:
        """Execute the tool and return result."""
        pass

    def to_schema(self) -> dict[str, Any]:
        """Convert to OpenAI function calling format."""
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": self.parameters,
            }
        }

# nanobot/agent/loop.py:66-87
def _register_default_tools(self) -> None:
    """Register the default set of tools."""
    # File tools
    self.tools.register(ReadFileTool())
    self.tools.register(WriteFileTool())
    self.tools.register(EditFileTool())
    self.tools.register(ListDirTool())

    # Shell tool
    self.tools.register(ExecTool(working_dir=str(self.workspace)))

    # Web tools
    self.tools.register(WebSearchTool(api_key=self.brave_api_key))
    self.tools.register(WebFetchTool())

    # 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)

# nanobot/agent/tools/registry.py:38-59
async def execute(self, name: str, params: dict[str, Any]) -> str:
    """
    Execute a tool by name with given parameters.

    Args:
        name: Tool name.
        params: Tool parameters.

    Returns:
        Tool execution result as string.

    Raises:
        KeyError: If tool not found.
    """
    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)}"

# nanobot/agent/loop.py:163-202
while iteration < self.max_iterations:
    iteration += 1

    # Call LLM
    response = await self.provider.chat(
        messages=messages,
        tools=self.tools.get_definitions(),
        model=self.model
    )

    # Handle tool calls
    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 = self.context.add_assistant_message(
            messages, response.content, tool_call_dicts
        )

        # Execute tools
        for tool_call in response.tool_calls:
            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:
        # No tool calls, we're done
        final_content = response.content
        break

# nanobot/agent/memory.py:16-19
def __init__(self, workspace: Path):
    self.workspace = workspace
    self.memory_dir = ensure_dir(workspace / "memory")
    self.memory_file = self.memory_dir / "MEMORY.md"

# nanobot/agent/memory.py:46-54
def read_long_term(self) -> str:
    """Read long-term memory (MEMORY.md)."""
    if self.memory_file.exists():
        return self.memory_file.read_text(encoding="utf-8")
    return ""

def write_long_term(self, content: str) -> None:
    """Write to long-term memory (MEMORY.md)."""
    self.memory_file.write_text(content, encoding="utf-8")

# nanobot/agent/memory.py:21-44
def get_today_file(self) -> Path:
    """Get path to today's memory file."""
    return self.memory_dir / f"{today_date()}.md"

def read_today(self) -> str:
    """Read today's memory notes."""
    today_file = self.get_today_file()
    if today_file.exists():
        return today_file.read_text(encoding="utf-8")
    return ""

def append_today(self, content: str) -> None:
    """Append content to today's memory notes."""
    today_file = self.get_today_file()

    if today_file.exists():
        existing = today_file.read_text(encoding="utf-8")
        content = existing + "\n" + content
    else:
        # Add header for new day
        header = f"# {today_date()}\n\n"
        content = header + content

    today_file.write_text(content, encoding="utf-8")

# nanobot/agent/memory.py:56-80
def get_recent_memories(self, days: int = 7) -> str:
    """
    Get memories from the last N days.

    Args:
        days: Number of days to look back.

    Returns:
        Combined memory content.
    """
    from datetime import timedelta

    memories = []
    today = datetime.now().date()

    for i in range(days):
        date = today - timedelta(days=i)
        date_str = date.strftime("%Y-%m-%d")
        file_path = self.memory_dir / f"{date_str}.md"

        if file_path.exists():
            content = file_path.read_text(encoding="utf-8")
            memories.append(content)

    return "\n\n---\n\n".join(memories)

# nanobot/agent/memory.py:90-109
def get_memory_context(self) -> str:
    """
    Get memory context for the agent.

    Returns:
        Formatted memory context including long-term and recent memories.
    """
    parts = []

    # Long-term memory
    long_term = self.read_long_term()
    if long_term:
        parts.append("## Long-term Memory\n" + long_term)

    # Today's notes
    today = self.read_today()
    if today:
        parts.append("## Today's Notes\n" + today)

    return "\n\n".join(parts) if parts else ""

# nanobot/agent/skills.py:26-57
def list_skills(self, filter_unavailable: bool = True) -> list[dict[str, str]]:
    """
    List all available skills.

    Args:
        filter_unavailable: If True, filter out skills with unmet requirements.

    Returns:
        List of skill info dicts with 'name', 'path', 'source'.
    """
    skills = []

    # Workspace skills (highest priority)
    if self.workspace_skills.exists():
        for skill_dir in self.workspace_skills.iterdir():
            if skill_dir.is_dir():
                skill_file = skill_dir / "SKILL.md"
                if skill_file.exists():
                    skills.append({"name": skill_dir.name, "path": str(skill_file), "source": "workspace"})

    # Built-in skills
    if self.builtin_skills and self.builtin_skills.exists():
        for skill_dir in self.builtin_skills.iterdir():
            if skill_dir.is_dir():
                skill_file = skill_dir / "SKILL.md"
                if skill_file.exists() and not any(s["name"] == skill_dir.name for s in skills):
                    skills.append({"name": skill_dir.name, "path": str(skill_file), "source": "builtin"})

    # Filter by requirements
    if filter_unavailable:
        return [s for s in skills if self._check_requirements(self._get_skill_meta(s["name"]))]
    return skills

# nanobot/agent/skills.py:193-201
def get_always_skills(self) -> list[str]:
    """Get skills marked as always=true that meet requirements."""
    result = []
    for s in self.list_skills(filter_unavailable=True):
        meta = self.get_skill_metadata(s["name"]) or {}
        skill_meta = self._parse_nanobot_metadata(meta.get("metadata", ""))
        if skill_meta.get("always") or meta.get("always"):
            result.append(s["name"])
    return result

# nanobot/providers/base.py:8-27
@dataclass
class ToolCallRequest:
    """A tool call request from the LLM."""
    id: str
    name: str
    arguments: dict[str, Any]

@dataclass
class LLMResponse:
    """Response from an LLM provider."""
    content: str | None
    tool_calls: list[ToolCallRequest] = field(default_factory=list)
    finish_reason: str = "stop"
    usage: dict[str, int] = field(default_factory=dict)

    @property
    def has_tool_calls(self) -> bool:
        """Check if response contains tool calls."""
        return len(self.tool_calls) > 0

# nanobot/providers/base.py:42-69
class LLMProvider(ABC):
    """Abstract base class for LLM providers."""

    def __init__(self, api_key: str | None = None, api_base: str | None = None):
        self.api_key = api_key
        self.api_base = api_base

    @abstractmethod
    async def chat(
        self,
        messages: list[dict[str, Any]],
        tools: list[dict[str, Any]] | None = None,
        model: str | None = None,
        max_tokens: int = 4096,
        temperature: float = 0.7,
    ) -> LLMResponse:
        """Send a chat completion request."""
        pass

    @abstractmethod
    def get_default_model(self) -> str:
        """Get the default model for this provider."""
        pass

# nanobot/session/manager.py:14-59
@dataclass
class Session:
    """A conversation session."""
    key: str  # channel:chat_id
    messages: list[dict[str, Any]] = field(default_factory=list)
    created_at: datetime = field(default_factory=datetime.now)
    updated_at: datetime = field(default_factory=datetime.now)
    metadata: dict[str, Any] = field(default_factory=dict)

    def add_message(self, role: str, content: str, **kwargs: Any) -> None:
        """Add a message to the session."""
        msg = {
            "role": role,
            "content": content,
            "timestamp": datetime.now().isoformat(),
            **kwargs
        }
        self.messages.append(msg)
        self.updated_at = datetime.now()

    def get_history(self, max_messages: int = 50) -> list[dict[str, Any]]:
        """Get message history for LLM context."""
        recent = self.messages[-max_messages:] if len(self.messages) > max_messages else self.messages
        return [{"role": m["role"], "content": m["content"]} for m in recent]