OpenClaw源码解析路由的基本概念详解

2026-06-03 10:12:19 昵称好难啊

上节课我们学习了通道的概念，openclaw能够接受来自不同通道（平台）的消息，然后统一封装成InboundMessage的格式，然后再交给run_agent_turn进行Agent的调用。今天我们来学习路由的概念，通过路由的设置，我们能够设置不同的agent来处理不同层级的消息。

1. 概念解析

1.1. 路由表( BindingTable)

首先是Binding，负责绑定入站消息和agent对应的规则，即每条规则定义了什么情况下由哪位agent响应，BingingTable在响应入站消息时，会按照tier从小到大的顺序遍历规则。

@dataclass
class Binding:
    agent_id: str
    tier: int           # 1-5, 越小越具体
    match_key: str      # "peer_id" | "guild_id" | "account_id" | "channel" | "default"
    match_value: str    # 例如 "telegram:12345", "discord", "*"
    priority: int = 0   # 同层内, 越大越优先

参数说明：
agent_id: 目标agent的唯一id

tier: 匹配的层级，数字是1-5，越小越优先

match_key: 匹配字段名，从peer_id, channel, default, guild_id和account_id中指定

match_value: 字段的匹配值

priority: 优先级

比如有如下绑定规则：

bt = BindingTable()
bt.add(Binding(agent_id="luna", tier=5, match_key="default", match_value="*"))
bt.add(Binding(agent_id="sage", tier=4, match_key="channel", match_value="telegram"))
bt.add(Binding(agent_id="sage", tier=1, match_key="peer_id",
               match_value="discord:admin-001", priority=10))

则有如下解析规则：

bt.add(Binding(agent_id="luna", tier=5, match_key="default", match_value="*"))

如果前面1~4层都没有任何匹配规则，则默认使用luna作为响应agent。

Binding(agent_id="sage", tier=4, match_key="channel", match_value="telegram")

通道级别匹配，如果入站消息是来自telegram的，并且前1~3层级没有更加具体的规则时，使用sage作为响应agent

Binding(agent_id="sage", tier=1, match_key="peer_id", match_value="telegram:admin-001", priority=10)

peer级别的匹配，如果入站消息是用户id为telegram:admin-001发出的，那么我们直接使用sage作为响应的Agent。

1.2. 路由解析

下面是路由解析的匹配规则，可以看到，规则是按照tier进行由小到大进行匹配的。

def resolve(self, channel: str = "", account_id: str = "",
                guild_id: str = "", peer_id: str = "") -> tuple[str | None, Binding | None]:
    """遍历第1-5层, 第一个匹配的获胜。返回 (agent_id, matched_binding)。"""
    for b in self._bindings:
        if b.tier == 1 and b.match_key == "peer_id":
            if ":" in b.match_value:
                if b.match_value == f"{channel}:{peer_id}":
                    return b.agent_id, b
            elif b.match_value == peer_id:
                return b.agent_id, b
        elif b.tier == 2 and b.match_key == "guild_id" and b.match_value == guild_id:
            return b.agent_id, b
        elif b.tier == 3 and b.match_key == "account_id" and b.match_value == account_id:
            return b.agent_id, b
        elif b.tier == 4 and b.match_key == "channel" and b.match_value == channel:
            return b.agent_id, b
        elif b.tier == 5 and b.match_key == "default":
            return b.agent_id, b
    return None, None

1.3. 会话隔离

通过指定不同的dm_scope，达到不同粒度的私聊隔离。

其实就是创建和加载不同的session.json文件，再序列化成消息格式发送给llm

# ---------------------------------------------------------------------------
# 会话键构建
# ---------------------------------------------------------------------------
# dm_scope 控制私聊隔离粒度:
#   main                      -> agent:{id}:main
#   per-peer                  -> agent:{id}:direct:{peer}
#   per-channel-peer          -> agent:{id}:{ch}:direct:{peer}
#   per-account-channel-peer  -> agent:{id}:{ch}:{acc}:direct:{peer}

def build_session_key(agent_id: str, channel: str = "", account_id: str = "",
                      peer_id: str = "", dm_scope: str = "per-peer") -> str:
    aid = normalize_agent_id(agent_id)
    ch = (channel or "unknown").strip().lower()
    acc = (account_id or "default").strip().lower()
    pid = (peer_id or "").strip().lower()
    if dm_scope == "per-account-channel-peer" and pid:
        return f"agent:{aid}:{ch}:{acc}:direct:{pid}"
    if dm_scope == "per-channel-peer" and pid:
        return f"agent:{aid}:{ch}:direct:{pid}"
    if dm_scope == "per-peer" and pid:
        return f"agent:{aid}:direct:{pid}"
    return f"agent:{aid}:main"

2. 流程设计

下面是代码的流程图，可以看到有两个输入，一个是cli，另一个是websocket客户端。

如果指定了force_agent，那么直接使用该agent进行回答，否则将进行路由表解析。通过查询BindingTable,获得响应的agent，然后使用build_session_key生成sk，决定不同的用户/通道的消息是否共享同一个对话历史。

到此这篇关于OpenClaw源码解析路由的基本概念详解的文章就介绍到这了,更多相关OpenClaw路由内容请搜索脚本之家以前的文章或继续浏览下面的相关文章，希望大家以后多多支持脚本之家！