小龙虾 | 架构介绍和运行配置逻辑

AI老马

发布于 2026-05-11 15:13:09

620

文章被收录于专栏：AI前沿技术AI前沿技术

龙虾 Openclaw 其实就是一个运行在用户终端的后台服务。它同时具备消息接入、会话路由、并发队列、短期/长期记忆、工具系统，Agent自主规划，以及一整套故障转移机制。

其核心运行逻辑：调用大模型 API->解析输出->本地执行系统命令>循环->结果回传。

如何搞懂龙虾？

在架构上与过去运行在服务端的 Agent 没有本质上的区别。

Openclaw 源码庞杂，上手门槛较高，为了降低阅读门槛，可以从它的“微缩版”—开源项目 Nanobot入手。

Nanobot 麻雀虽小，五脏俱全。通过拆解其核心的源码，能以极低的认知成本，快速看透 Openclaw 的底层运行逻辑。从架构层面，彻底搞懂”小龙虾“们背后的逻辑。

1，工程架构概览

1.1，目录结构

nanobot/
├── agent/          # 🧠 Core agent logic
│   ├── loop.py     #    Agent loop (LLM ↔ tool execution)
│   ├── context.py  #    Prompt builder
│   ├── memory.py   #    Persistent memory
│   ├── skills.py   #    Skills loader
│   ├── subagent.py #    Background task execution
│   └── tools/      #    Built-in tools (incl. spawn)
├── skills/         # 🎯 Bundled skills (github, weather, tmux...)
├── channels/       # 📱 Chat channel integrations (supports plugins)
├── bus/            # 🚌 Message routing
├── cron/           # ⏰ Scheduled tasks
├── heartbeat/      # 💓 Proactive wake-up
├── providers/      # 🤖 LLM providers (OpenRouter, etc.)
├── session/        # 💬 Conversation sessions
├── config/         # ⚙️ Configuration
└── cli/            # 🖥️ Commands

划重点!

agent 目录下的功能为核心功能，实现了Agent 的主循环，消息的消费和结果生成，并将其他各功能协调在一起。

同级的每一个文件夹都是一个模块，边界清晰，可以快速的定位功能代码。

如何启动?

四个核心命令

• onboard，用于初始化或重置 nanobot的配置文件 (~1.nanobot/config.json) 和本地工作区目录。
• gateway，启动 nanobot 的后台核心服务。它会拉起AI对话逻辑，监听消息，处理定时任务和系统心跳等。
• agent，允许用户在终端直接与AI助手对话。
• status，检查并打印当前工作区路径、配置文件状态以及各LLM 提供商的API Key 配置情况。

1.2，功能架构图

图 2，Nanobot架构图

整体架构可以分为：渠道信息模块，消息总线模块，核心调度模块，命令路由模块，会话管理与上下文模块，工具注册模块，大模型推理模块等。

重点模块功能：

• MessageBus — 消息总线模块解耦渠道和 Agent，渠道只管 push/pop， Agent 只管消费和回写。
• CommandRouter —命令路由层在消息进入 LLM 之前做前置拦截，分priority / exact / prefix / interceptor 四级匹配。命中命令直接短路返回，不消耗LLM token。
• SessionManager + ContextBuilder —会话与上下文负责把“原始对话历史”转换为 LLM需要的 messages 结构，并在每轮结束后通过命令持久化新消息。
• MemoryConsolidator — 记忆压缩层当session 历史超过 context_window_tokens 时自动触发压缩，防止上下文溢出。以后台 Task 异步执行，不阻塞主流程。
• ToolRegistry — 工具注册层统一管理所有工具（文件、Shell、Web、MCP)，提供get_definitions()给LLM 做function calling，提供execute() 执行工具调用结果。
• _run_agent_loop —LLM 推理循环层核心的 ReAct 循环：调用LLM→有 tool_calls 就执行工具并把结果追加到 messages →再次调用LLM一直到无 tool_calls 或达到

实现架构的优势：

• 单一职责，各层可独立替换。

比如换 LLM 提供商只改 LLMProvider，其他层不动。新增渠道只实现渠道层 push/pop，不改 Agent。新增工具，添加数据库查询，只注册新 Tool，不改推理循环。换持久化方式只改 SessionManager。

• 关注点隔离，降低认知负担

每一层只需理解自己的输入输出契约：渠道层不需要知道 LLM 怎么工作。LLM推理层不需要知道消息从哪个平台来。工具层不需要知道会话历史怎么管理

• 异步解耦，提升响应性

run() 主循环通过 create_task 把处理逻辑异步化，使得 /stop 这类优先命令可以在锁外立即响应。主循环不会因为一条慢消息，而卡死后台任务，如记忆压缩不影响主流程。

流式与非流式统一接口，通过回调注入的方式，process message 和 run agent loop 本身不关心是否流式，由上层决定是否传入回调，做到了逻辑与传输方式解耦。

• 错误隔离

每个Task 独立捕获异常，一个用户的消息处理失败不会影响其他用户，也不会崩溃主循环。

2，配置运行

能够运行起来，是迈向成功的第一步！好的项目，仅需要简单的配置操作，就能成功运行。

2.1，项目配置文件

项目的配置分为三层：

第一层，通过Base 继承基础的数据配置类，实现数据格式的统一。

第二层，每个配置继承Base 类进行不同配置项的定义。

第三层，通过 BaseSetting的子类 Config 进行组合，作为聚合根。

2.2，配置文件逻辑实现

• BaseModel

用来做数据模型，定义业务数据，也可校验接口入参数、返回体、数据结构等。

• BaseSettings

用来做项目的配置，自动从环境变量，env 文件，或者命令行读取配置。

BaseSetting 也是继承自 BaseModel，即配置类本质也是一个数据模型，但是多了自动加载配置的功能。

在本项目中具体实现为：

class Config(BaseSettings):
    """Root configuration for nanobot."""
    agents: AgentsConfig = Field(default_factory=AgentsConfig)
    channels: ChannelsConfig = Field(default_factory=ChannelsConfig)
    providers: ProvidersConfig = Field(default_factory=ProvidersConfig)
    api: ApiConfig = Field(default_factory=ApiConfig)
    gateway: GatewayConfig = Field(default_factory=GatewayConfig)
    tools: ToolsConfig = Field(default_factory=ToolsConfig)

在config 中定义了6个一级索引标签。每个一级标签下对应一个默认配置类。详细看下 agents 标签，配置代码。

class AgentsConfig(Base):
    """Agent configuration."""
    defaults: AgentDefaults = Field(default_factory=AgentDefaults)

class AgentDefaults(Base):
    """Default agent configuration."""
    workspace: str = "~/.nanobot/workspace"
    model: str = "anthropic/claude-opus-4-5"
    .....

这样配置代码逻辑和json格式对应关系就很清晰了

{
  "agents":{"defaults":{"workspace":xx,"model":xx}},
  "channels":{},
  "providers":{},
  ....,
  "tools":{}
}

需要注意的是，如果json 文件中没有定义某字段，而默认代码中有相应值，则取默认值。

如何关联使用：

def load_config(config_path: Path | None = None) -> Config:
    path = config_path or get_config_path()
    config = Config()
    if path.exists():
        try:
            with open(path, encoding="utf-8") as f:
                data = json.load(f)
            config = Config.model_validate(data)
        except (json.JSONDecodeError, ValueError, pydantic.ValidationError) as e:
            logger.warning("Using default configuration.")
    return config

传入配置文件json 加载文件后直接用实例化后的 config 对象调用model_validate 函数即可。

如果保存对应的设置到json文件中和写入文件一样。

def save_config(config: Config, config_path: Path | None = None) -> None:
    path = config_path or get_config_path()
    path.parent.mkdir(parents=True, exist_ok=True)
    data = config.model_dump(mode="json", by_alias=True)
    with open(path, "w", encoding="utf-8") as f:
        json.dump(data, f, indent=2, ensure_ascii=False)

以上通过json 文件导入配置，还可以通过环境变量的方式。

from typing importDict, Optional
from pydantic import BaseModel, ConfigDict, Field
from pydantic_settings import BaseSettings

classProviderConfig(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    api_key: Optional[str] = Field(default=None, serialization_alias="apiKey")

classConfig(BaseSettings):
    model_config = ConfigDict(env_prefix="NANOBOT_", env_nested_delimiter="__")
    providers: Dict[str, ProviderConfig] = {}

if __name__ == "__main__":
    config = Config()
    print(config.model_dump(by_alias=True))

# 环境变量自动注入
NANOBOT_PROVIDERS__QIANFAN__API_KEY=xxx
# 等同于 {"providers": {"qianfan": {"apiKey": "xxx"}}}

以上环境变量的加载是自动的，在实例化对象 config = Config() 的时候自动发生，无需调用任何函数。其中 ConfigDict 说明了加载环境变量的规则。

通过了解基本的架构和配置方式，此时就可运行其程序进行对话了。