nanobot 扩展机制深度分析

class SkillsLoader:
    def __init__(self, workspace: Path, builtin_skills_dir: Path | None = None):
        self.workspace = workspace
        self.workspace_skills = workspace / "skills"
        self.builtin_skills = builtin_skills_dir or BUILTIN_SKILLS_DIR

def _check_requirements(self, skill_meta: dict) -> bool:
    requires = skill_meta.get("requires", {})
    for b in requires.get("bins", []):
        if not shutil.which(b):
            return False
    for env in requires.get("env", []):
        if not os.environ.get(env):
            return False
    return True

class CronService:
    def __init__(
        self,
        store_path: Path,
        on_job: Callable[[CronJob], Coroutine[Any, Any, str | None]] | None = None
    ):
        self.store_path = store_path
        self.on_job = on_job  # 回调函数，返回响应文本
        self._store: CronStore | None = None
        self._timer_task: asyncio.Task | None = None
        self._running = False

if schedule.kind == "at":
    return schedule.at_ms if schedule.at_ms and schedule.at_ms > now_ms else None

if schedule.kind == "every":
    if not schedule.every_ms or schedule.every_ms <= 0:
        return None
    return now_ms + schedule.every_ms

if schedule.kind == "cron" and schedule.expr:
    try:
        from croniter import croniter
        cron = croniter(schedule.expr, time.time())
        next_time = cron.get_next()
        return int(next_time * 1000)
    except Exception:
        return None

@dataclass
class CronJob:
    id: str                          # 任务 ID（UUID 前缀）
    name: str                        # 任务名称
    enabled: bool = True              # 是否启用
    schedule: CronSchedule            # 调度配置
    payload: CronPayload             # 执行内容
    state: CronJobState              # 运行状态
    created_at_ms: int = 0           # 创建时间
    updated_at_ms: int = 0           # 更新时间
    delete_after_run: bool = False   # 执行后是否删除

async def _execute_job(self, job: CronJob) -> None:
    start_ms = _now_ms()
    logger.info(f"Cron: executing job '{job.name}' ({job.id})")

    try:
        response = None
        if self.on_job:
            response = await self.on_job(job)

        job.state.last_status = "ok"
        job.state.last_error = None
        logger.info(f"Cron: job '{job.name}' completed")

    except Exception as e:
        job.state.last_status = "error"
        job.state.last_error = str(e)
        logger.error(f"Cron: job '{job.name}' failed: {e}")

    job.state.last_run_at_ms = start_ms
    job.updated_at_ms = _now_ms()

    # 处理一次性任务
    if job.schedule.kind == "at":
        if job.delete_after_run:
            self._store.jobs = [j for j in self._store.jobs if j.id != job.id]
        else:
            job.enabled = False
            job.state.next_run_at_ms = None
    else:
        # 计算下次运行
        job.state.next_run_at_ms = _compute_next_run(job.schedule, _now_ms())

class HeartbeatService:
    """
    Periodic heartbeat service that wakes the agent to check for tasks.

    The agent reads HEARTBEAT.md from the workspace and executes any
    tasks listed there. If nothing needs attention, it replies HEARTBEAT_OK.
    """

def _is_heartbeat_empty(content: str | None) -> bool:
    if not content:
        return True

    # Lines to skip: empty, headers, HTML comments, empty checkboxes
    skip_patterns = {"- [ ]", "* [ ]", "- [x]", "* [x]"}

    for line in content.split("\n"):
        line = line.strip()
        if not line or line.startswith("#") or line.startswith("<!--") or line in skip_patterns:
            continue
        return False  # Found actionable content

    return True

HEARTBEAT_PROMPT = """Read HEARTBEAT.md in your workspace (if it exists).
Follow any instructions or tasks listed there.
If nothing needs attention, reply with just: HEARTBEAT_OK"""

HEARTBEAT_OK_TOKEN = "HEARTBEAT_OK"

def get_api_key(self) -> str | None:
    """Get API key in priority order: OpenRouter > Anthropic > OpenAI > Gemini > Zhipu > Groq > vLLM."""
    return (
        self.providers.openrouter.api_key or
        self.providers.anthropic.api_key or
        self.providers.openai.api_key or
        self.providers.gemini.api_key or
        self.providers.zhipu.api_key or
        self.providers.groq.api_key or
        self.providers.vllm.api_key or
        None
    )

class Config(BaseSettings):
    class Config:
        env_prefix = "NANOBOT_"
        env_nested_delimiter = "__"

from nanobot.channels.base import BaseChannel

class DiscordChannel(BaseChannel):
    name = "discord"

    async def start(self) -> None:
        # 连接到 Discord
        pass

    async def stop(self) -> None:
        # 断开连接
        pass

    async def send(self, msg: OutboundMessage) -> None:
        # 发送消息
        pass

# nanobot/config/schema.py
class DiscordConfig(BaseModel):
    enabled: bool = False
    token: str = ""
    allow_from: list[str] = Field(default_factory=list)

class ChannelsConfig(BaseModel):
    whatsapp: WhatsAppConfig = Field(default_factory=WhatsAppConfig)
    telegram: TelegramConfig = Field(default_factory=TelegramConfig)
    discord: DiscordConfig = Field(default_factory=DiscordConfig)  # 新增

# nanobot/channels/manager.py
def _init_channels(self) -> None:
    # ... 现有渠道 ...

    if self.config.channels.discord.enabled:
        from nanobot.channels.discord import DiscordChannel
        self.channels["discord"] = DiscordChannel(
            self.config.channels.discord, self.bus
        )

from nanobot.agent.tools.base import Tool
from typing import Any

class MyTool(Tool):
    @property
    def name(self) -> str:
        return "my_tool"

    @property
    def description(self) -> str:
        return "Description of what this tool does"

    @property
    def parameters(self) -> dict[str, Any]:
        return {
            "type": "object",
            "properties": {
                "param1": {
                    "type": "string",
                    "description": "Parameter description"
                }
            },
            "required": ["param1"]
        }

    async def execute(self, param1: str, **kwargs: Any) -> str:
        # 执行工具逻辑
        return f"Result: {param1}"

# nanobot/agent/loop.py
def _register_default_tools(self) -> None:
    # ... 现有工具 ...
    self.tools.register(MyTool())

from nanobot.providers.base import LLMProvider, LLMResponse

class MyProvider(LLMProvider):
    def __init__(self, api_key: str | None = None, api_base: str | None = None):
        super().__init__(api_key, api_base)

    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:
        # 调用 API
        response = await self._call_api(messages, model)
        return LLMResponse(content=response["content"])

    def get_default_model(self) -> str:
        return "my-provider/default-model"

# nanobot/config/schema.py
class ProvidersConfig(BaseModel):
    # ... 现有提供商 ...
    my_provider: ProviderConfig = Field(default_factory=ProviderConfig)

def get_api_key(self) -> str | None:
    return (
        # ... 现有优先级 ...
        self.providers.my_provider.api_key or
        None
    )

# 自动检测 OpenRouter
self.is_openrouter = (
    (api_key and api_key.startswith("sk-or-")) or
    (api_base and "openrouter" in api_base)
)

# Zhipu 模型前缀处理
if ("glm" in model.lower() or "zhipu" in model.lower()) and not (
    model.startswith("zhipu/") or
    model.startswith("zai/") or
    model.startswith("openrouter/")
):
    model = f"zhipu/{model}"

# 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_summary}""")

def _arm_timer(self) -> None:
    if self._timer_task:
        self._timer_task.cancel()

    next_wake = self._get_next_wake_ms()
    if not next_wake or not self._running:
        return

    delay_ms = max(0, next_wake - _now_ms())
    delay_s = delay_ms / 1000

    async def tick():
        await asyncio.sleep(delay_s)
        if self._running:
            await self._on_timer()

    self._timer_task = asyncio.create_task(tick())

def _markdown_to_telegram_html(text: str) -> str:
    # 1. Extract and protect code blocks
    code_blocks: list[str] = []
    def save_code_block(m: re.Match) -> str:
        code_blocks.append(m.group(1))
        return f"\x00CB{len(code_blocks) - 1}\x00"

    text = re.sub(r'```[\w]*\n?([\s\S]*?)```', save_code_block, text)

    # 2. Extract and protect inline code
    inline_codes: list[str] = []
    def save_inline_code(m: re.Match) -> str:
        inline_codes.append(m.group(1))
        return f"\x00IC{len(inline_codes) - 1}\x00"

    text = re.sub(r'`([^`]+)`', save_inline_code, text)

    # ... 处理其他 Markdown 语法 ...

    # 11-12. Restore protected content with HTML tags
    for i, code in enumerate(inline_codes):
        escaped = code.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
        text = text.replace(f"\x00IC{i}\x00", f"<code>{escaped}</code>")

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

async def start(self) -> None:
    import websockets

    bridge_url = self.config.bridge_url

    while self._running:
        try:
            async with websockets.connect(bridge_url) as ws:
                self._ws = ws
                self._connected = True

                async for message in ws:
                    try:
                        await self._handle_bridge_message(message)
                    except Exception as e:
                        logger.error(f"Error handling bridge message: {e}")

        except asyncio.CancelledError:
            break
        except Exception as e:
            self._connected = False
            if self._running:
                await asyncio.sleep(5)  # Reconnect delay