OpenClaw源碼解析路由的基本概念詳解

發(fā)布時(shí)間：2026-06-03 10:12:19 作者：昵稱好難啊

今天我們來學(xué)習(xí)路由的概念，通過路由的設(shè)置，我們能夠設(shè)置不同的agent來處理不同層級的消息,本文給大家介紹的非常詳細(xì)，感興趣的朋友一起看看吧

上節(jié)課我們學(xué)習(xí)了通道的概念，openclaw能夠接受來自不同通道（平臺）的消息，然后統(tǒng)一封裝成InboundMessage的格式，然后再交給run_agent_turn進(jìn)行Agent的調(diào)用。今天我們來學(xué)習(xí)路由的概念，通過路由的設(shè)置，我們能夠設(shè)置不同的agent來處理不同層級的消息。

1. 概念解析

1.1. 路由表( BindingTable)

首先是Binding，負(fù)責(zé)綁定入站消息和agent對應(yīng)的規(guī)則，即每條規(guī)則定義了什么情況下由哪位agent響應(yīng)，BingingTable在響應(yīng)入站消息時(shí)，會(huì)按照tier從小到大的順序遍歷規(guī)則。

@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   # 同層內(nèi), 越大越優(yōu)先

參數(shù)說明：
agent_id: 目標(biāo)agent的唯一id

tier: 匹配的層級，數(shù)字是1-5，越小越優(yōu)先

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

match_value: 字段的匹配值

priority: 優(yōu)先級

比如有如下綁定規(guī)則：

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

則有如下解析規(guī)則：

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

如果前面1~4層都沒有任何匹配規(guī)則，則默認(rèn)使用luna作為響應(yīng)agent。

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

通道級別匹配，如果入站消息是來自telegram的，并且前1~3層級沒有更加具體的規(guī)則時(shí)，使用sage作為響應(yīng)agent

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

peer級別的匹配，如果入站消息是用戶id為telegram:admin-001發(fā)出的，那么我們直接使用sage作為響應(yīng)的Agent。

1.2. 路由解析

下面是路由解析的匹配規(guī)則，可以看到，規(guī)則是按照tier進(jìn)行由小到大進(jìn)行匹配的。

def resolve(self, channel: str = "", account_id: str = "",
                guild_id: str = "", peer_id: str = "") -> tuple[str | None, Binding | None]:
    """遍歷第1-5層, 第一個(gè)匹配的獲勝。返回 (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. 會(huì)話隔離

通過指定不同的dm_scope，達(dá)到不同粒度的私聊隔離。

其實(shí)就是創(chuàng)建和加載不同的session.json文件，再序列化成消息格式發(fā)送給llm

# ---------------------------------------------------------------------------
# 會(huì)話鍵構(gòu)建
# ---------------------------------------------------------------------------
# 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"