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

@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   # 同层内, 越大越优先

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="*"))

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

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

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

# ---------------------------------------------------------------------------
# 会话键构建
# ---------------------------------------------------------------------------
# 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"