四千行代码如何实现一个精简OpenClaw?Nanobot核心架构解析

async def _run_agent_loop(self, initial_messages, on_progress=None):    """运行Agent迭代循环"""    messages = initial_messages    iteration = 0    while iteration < self.max_iterations:        iteration += 1        # 第一步：调用LLM获取响应        response = await self.provider.chat(            messages=messages,            tools=self.tools.get_definitions(),            model=self.model,            temperature=self.temperature,            max_tokens=self.max_tokens,        )        # 第二步：如果有工具调用，执行工具        if response.has_tool_calls:            # 添加助手消息（包含工具调用）            messages = self.context.add_assistant_message(...)            # 逐个执行工具调用            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, ...)        else:            # 没有工具调用，说明任务完成            final_content = response.content            break

if msg.content.strip().lower() == "/stop":    await self._handle_stop(msg)

def build_messages(self, history, current_message, media=None, channel=None, chat_id=None):    """构建完整的消息列表"""    # 1. 运行时上下文（时间、渠道等）    runtime_ctx = self._build_runtime_context(channel, chat_id)    # 2. 用户消息（可能包含图片）    user_content = self._build_user_content(current_message, media)    # 3. 合并所有内容    return [        {"role": "system", "content": self.build_system_prompt()},        *history,        {"role": "user", "content": merged},    ]

def build_system_prompt(self, skill_names=None):    parts = [self._get_identity()]      # 核心身份    parts.append(self._load_bootstrap_files())  # 引导文件    parts.append(self.memory.get_memory_context())  # 记忆    parts.append(self.skills.load_skills_for_context(...))  # 常用技能    parts.append(self.skills.build_skills_summary())  # 技能概要    return "\n\n---\n\n".join(parts)

class MemoryStore:    """两层记忆：MEMORY.md（长期事实）+ HISTORY.md（可搜索日志）"""    def __init__(self, workspace):        self.memory_dir = ensure_dir(workspace / "memory")        self.memory_file = self.memory_dir / "MEMORY.md"        self.history_file = self.memory_dir / "HISTORY.md"

async def consolidate(self, session, provider, model, archive_all=False, memory_window=50):    """将旧消息整合到MEMORY.md和HISTORY.md"""    # 1. 提取待整合的消息    old_messages = session.messages[session.last_consolidated:-keep_count]    # 2. 调用LLM进行总结    response = await provider.chat(        messages=[...],        tools=_SAVE_MEMORY_TOOL,  # 专门用于保存记忆的工具    )    # 3. LLM决定：哪些写入MEMORY，哪些写入HISTORY    if entry := args.get("history_entry"):        self.append_history(entry)  # 写入HISTORY.md    if update := args.get("memory_update"):        self.write_long_term(update)  # 更新MEMORY.md

def build_system_prompt(self):    # 1. 常用技能：总是加载完整内容    always_skills = self.skills.get_always_skills()    always_content = self.skills.load_skills_for_context(always_skills)    # 2. 其他技能：只显示摘要    skills_summary = self.skills.build_skills_summary()    # Agent如果需要某个技能，会自己用read_file读取

---name: githubdescription: GitHub操作技能always: false  # 是否总是加载requires:  bins: [gh]        # 需要的命令行工具  env: [GH_TOKEN]   # 需要的环境变量---

async def spawn(self, task, label=None, origin_channel="cli", ...):    """孵化一个子代理来执行后台任务"""    task_id = str(uuid.uuid4())[:8]    # 创建后台任务    bg_task = asyncio.create_task(        self._run_subagent(task_id, task, display_label, origin)    )    # 立即返回，让主Agent可以继续响应用户    return f"Subagent [{display_label}] started (id: {task_id}). I'll notify you when it completes."

async def _announce_result(self, task_id, result, origin, status):    """通过消息总线宣布结果"""    msg = InboundMessage(        channel="system",        sender_id="subagent",        chat_id=f"{origin['channel']}:{origin['chat_id']}",        content=announce_content,    )    await self.bus.publish_inbound(msg)

class ToolRegistry:    def __init__(self):        self._tools: dict[str, Tool] = {}    def register(self, tool: Tool) -> None:        self._tools[tool.name] = tool    async def execute(self, name: str, params: dict) -> str:        tool = self._tools.get(name)        return await tool.execute(**params)

class Tool(ABC):    @property    @abstractmethod    def name(self) -> str:        pass  # 工具名称    @property    @abstractmethod    def description(self) -> str:        pass  # 工具描述（给LLM看）    @property    @abstractmethod    def parameters(self) -> dict:        pass  # JSON Schema参数定义    @abstractmethod    async def execute(self, **kwargs) -> str:        pass  # 执行逻辑

class ReadFileTool(Tool):    @property    def name(self) -> str:        return "read_file"    @property    def description(self) -> str:        return "Read content from a file..."    @property    def parameters(self) -> dict:        return {            "type": "object",            "properties": {                "path": {"type": "string", "description": "File path to read"},            },            "required": ["path"]        }    async def execute(self, path, **kwargs) -> str:        # 实际的文件读取逻辑        content = Path(path).read_text()        return content

class LLMProvider(ABC):    @abstractmethod    async def chat(self, messages, tools=None, model=None, ...):        """发送聊天请求"""        pass    @abstractmethod    def get_default_model(self) -> str:        """获取默认模型"""        pass

# 1. 安装pip install nanobot-ai# 2. 初始化配置nanobot onboard# 3. 启动对话nanobot agent