类class agentscope.app.message_bus

RedisMessageBus

class RedisMessageBus(MessageBus)

源码Source agentscope/app/message_bus/_redis_message_bus.py:48–725

Redis 消息总线，把智能体接入外部消息渠道与企业平台。位于 agentscope.app.message_bus 包，继承自 MessageBus。核心成员包括 aclose、get_client、queue_push、queue_drain、queue_delete、log_append。

Mapping of bus modes to Redis primitives:

Mode A (drain queue) uses a Redis Stream per key. XADD appends a payload whose single field payload carries the JSON-serialised dict. queue_drain runs XRANGE and XDEL in one Lua script, so an entry is handed to exactly one consumer however many drain the key concurrently. ttl_secs is enforced via EXPIRE after each push (sliding TTL).
Mode C (replay log) also uses a Redis Stream, but never XDELs on read. Trimming happens via XADD … MAXLEN ~N (approximate, for performance) on append, via the ttl_secs EXPIRE on the whole key, or explicitly via :meth:log_trim. log_read uses XRANGE with an exclusive start id derived from since.
Mode D (transient broadcast) rides Redis Pub/Sub. Wake-ups are best-effort: payloads published before a subscription exists are not delivered.

The bus owns its own connection pool by default; an external pool may be supplied for tests or for sharing a pool across services.

构造器摘要Constructor Summary

构造器Constructor	说明Description
`def __init__(host: str = 'localhost', port: int = 6379, db: int = 0, password: str \| None = None, connection_pool: Conn…`	初始化 `RedisMessageBus` 实例。 Store connection parameters; the actual pool is created in :meth:`__aenter__`.

方法摘要Method Summary

方法Method	说明Description
`async def __aenter__() -> Self`async	进入异步上下文并返回自身。 Create the connection pool and Redis client.
`async def aclose() -> None`async	返回 `aclose` 结果。 Close the connection pool if it was created internally.
`async def __aexit__(exc_type: type[BaseException] \| None, exc_value: BaseException \| None, traceback: Any) -> None`async	退出异步上下文并清理资源。 Exit the async context manager and release resources.
`def get_client() -> Redis`	获取客户端。 Return the underlying Redis client.
`async def queue_push(key: str, payload: dict, *, ttl_secs: int \| None = None) -> str`async	执行队列推送相关操作。 Append ``payload`` to the drain queue at ``key``.
`async def queue_drain(key: str, max_count: int = 100) -> list[tuple[str, dict]]`async	执行 `queue_drain` 方法。 Drain up to ``max_count`` entries from the queue at ``key``.
`async def queue_delete(key: str) -> None`async	设置队列删除。 Delete the drain queue at ``key``.
`async def log_append(key: str, payload: dict, *, ttl_secs: int \| None = None, max_len: int \| None = None) -> str`async	执行日志追加相关操作。 Append ``payload`` to the replay log at ``key``.
`async def log_read(key: str, since: str \| None = None, max_count: int = 100) -> list[tuple[str, dict]]`async	执行日志读取相关操作。 Read up to ``max_count`` entries newer than ``since``.
`async def log_trim(key: str, before_id: str \| None = None) -> None`async	执行日志裁剪相关操作。 Trim the replay log at ``key``.
`async def registry_set(namespace: str, field: str, value: str, *, ttl_secs: int \| None = None) -> None`async	执行注册表设置相关操作。 Set ``field`` in the Redis Hash at ``namespace``.
`async def registry_set_if(namespace: str, field: str, value: str, *, expected: str, ttl_secs: int \| None = None) -> bool`async	执行注册表设置 if相关操作。 Compare-and-set one Hash field via Lua. See base.
`async def registry_pop(namespace: str, field: str) -> str \| None`async	执行 `registry_pop` 方法。 ``HGET`` + ``HDEL`` as one Lua step. See base.
`async def registry_del(namespace: str, field: str) -> None`async	执行 `registry_del` 方法。 Remove ``field`` from the Redis Hash at ``namespace``.
`async def registry_exists(namespace: str, field: str) -> bool`async	执行 `registry_exists` 方法。 Return whether ``field`` exists in the Hash at ``namespace``.
`async def registry_getall(namespace: str) -> dict[str, str]`async	设置该参数。 Return all field-value pairs from the Hash at ``namespace``.
`async def registry_get(namespace: str, field: str) -> str \| None`async	执行注册表读取相关操作。 Return a single field value from the Hash at ``namespace``.
`async def registry_drop(namespace: str) -> None`async	设置该参数。 Delete the entire Hash at ``namespace``.
`async def publish(key: str, payload: dict) -> None`async	执行发布相关操作。 Publish ``payload`` on the broadcast channel ``key``.
`async def subscribe(key: str, *, on_ready: Callable[[], None] \| None = None) -> AsyncGenerator[dict, None]`async	执行订阅相关操作。 Yield broadcast payloads on ``key`` until the consumer closes the generator.
`async def acquire_lock(key: str, *, ttl_secs: int = 600) -> AsyncGenerator[None, None]`asyncasynccontextmanager	执行 `acquire_lock` 方法。 Acquire ``key`` as a distributed mutex.
`async def is_locked(key: str) -> bool`async	判断是否 `is_locked`。 Return whether ``key`` currently holds a lock.
`async def try_lock(key: str, *, ttl_secs: int = 600) -> bool`async	执行 `try_lock` 方法。 Non-blocking claim via ``SET key NX EX``. See base.
`async def unlock(key: str) -> None`async	设置该参数。 Release a ``try_lock`` claim (best-effort ``DEL``).

构造器详情Constructor Detail

RedisMessageBus

def __init__(host: str = 'localhost', port: int = 6379, db: int = 0, password: str | None = None, connection_pool: ConnectionPool | None = None, **kwargs: Any) -> None

初始化 RedisMessageBus 实例。

Store connection parameters; the actual pool is created in :meth:__aenter__.

参数Parameters

名称Name	类型Type	说明Description
`host`默认default `'localhost'`	str	host（类型 `str`）。Redis server host.
`port`默认default `6379`	int	port（类型 `int`）。Redis server port.
`db`默认default `0`	int	数据库（类型 `int`）。Redis logical database index.
`password`默认default `None`	str \| None	密码（类型 `str \| None`）。Redis password if required by the server.
`connection_pool`默认default `None`	ConnectionPool \| None	连接连接池（类型 `ConnectionPool \| None`）。An externally managed connection pool. When provided the pool is used as-is and not closed by :meth:`aclose` — the caller retains ownership of its lifecycle. When omitted a pool is created from host/port/db/password on :meth:`__aenter__` and closed on :meth:`aclose`.
`kwargs`	Any	关键字参数（类型 `Any`）。The kwargs.

源码Source agentscope/app/message_bus/_redis_message_bus.py:73–116

方法详情Method Detail

aenter

async def __aenter__() -> Self

async

进入异步上下文并返回自身。

If an external pool was supplied at construction time it is used directly and its lifecycle remains the caller's responsibility.

返回Returns Self

返回自身。`Self`:

源码Source agentscope/app/message_bus/_redis_message_bus.py:118–153

aclose

async def aclose() -> None

async

返回 aclose 结果。

Externally supplied pools are left open — the caller owns them.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/message_bus/_redis_message_bus.py:155–163

aexit

async def __aexit__(exc_type: type[BaseException] | None, exc_value: BaseException | None, traceback: Any) -> None

async

退出异步上下文并清理资源。

Exit the async context manager and release resources.

参数Parameters

名称Name	类型Type	说明Description
`exc_type`	type[BaseException] \| None	exc 类型（类型 `type[BaseException] \| None`）。Exception class raised inside the context, if any.
`exc_value`	BaseException \| None	exc 值（类型 `BaseException \| None`）。Exception instance raised inside the context, if any.
`traceback`	Any	traceback（类型 `Any`）。Traceback associated with the exception, if any.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/message_bus/_redis_message_bus.py:165–181

get_client

def get_client() -> Redis

获取客户端。

Only valid inside the async context (between :meth:__aenter__ and :meth:aclose).

返回Returns Redis

返回Redis。`Redis`:

源码Source agentscope/app/message_bus/_redis_message_bus.py:183–193

queue_push

async def queue_push(key: str, payload: dict, *, ttl_secs: int | None = None) -> str

async

执行队列推送相关操作。

Append payload to the drain queue at key.

参数Parameters

名称Name	类型Type	说明Description
`key`	str	键（类型 `str`）。Stream key for this drain queue.
`payload`	dict	负载（类型 `dict`）。JSON-serializable dict; encoded into the `payload` field of a Redis Stream entry.
`ttl_secs`kw默认default `None`	int \| None	存活时间 secs（类型 `int \| None`）。If set, refresh the key's expiry on every push (sliding TTL). `None` means no TTL — the queue persists until drained or deleted explicitly.

返回Returns str

返回字符串结果。`str`:

源码Source agentscope/app/message_bus/_redis_message_bus.py:225–255

queue_drain

async def queue_drain(key: str, max_count: int = 100) -> list[tuple[str, dict]]

async

执行 queue_drain 方法。

Implementation: XRANGE and XDEL of the returned ids run as one Lua script, so nothing else observes the entries between the read and their removal.

参数Parameters

名称Name	类型Type	说明Description
`key`	str	键（类型 `str`）。Stream key for the drain queue.
`max_count`默认default `100`	int	最大数量（类型 `int`）。Maximum entries to return in one call.

返回Returns list[tuple[str, dict]]

返回元组[str,dict]列表。`list[tuple[str, dict]]`:

源码Source agentscope/app/message_bus/_redis_message_bus.py:257–291

queue_delete

async def queue_delete(key: str) -> None

async

设置队列删除。

Delete the drain queue at key.

参数Parameters

名称Name	类型Type	说明Description
`key`	str	键（类型 `str`）。Stream key for the drain queue. `DEL` is a no-op when the key does not exist.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/message_bus/_redis_message_bus.py:293–301

log_append

async def log_append(key: str, payload: dict, *, ttl_secs: int | None = None, max_len: int | None = None) -> str

async

执行日志追加相关操作。

Append payload to the replay log at key.

参数Parameters

名称Name	类型Type	说明Description
`key`	str	键（类型 `str`）。Stream key for this replay log.
`payload`	dict	负载（类型 `dict`）。JSON-serializable dict; encoded into the `payload` field of a Redis Stream entry.
`ttl_secs`kw默认default `None`	int \| None	存活时间 secs（类型 `int \| None`）。If set, refresh the key's expiry on every append (sliding TTL). `None` means no TTL.
`max_len`kw默认default `None`	int \| None	最大 len（类型 `int \| None`）。If set, cap the log at approximately this many entries using `XADD MAXLEN ~N`. The cap is approximate so Redis can use its O(1) trim-by-radix-tree-node mode. `None` means no cap.

返回Returns str

返回字符串结果。`str`:

源码Source agentscope/app/message_bus/_redis_message_bus.py:307–348

log_read

async def log_read(key: str, since: str | None = None, max_count: int = 100) -> list[tuple[str, dict]]

async

执行日志读取相关操作。

Read up to max_count entries newer than since.

参数Parameters

名称Name	类型Type	说明Description
`key`	str	键（类型 `str`）。Stream key for the replay log.
`since`默认default `None`	str \| None	since（类型 `str \| None`）。Exclusive cursor — return entries strictly newer than this id (typically the last id from a previous read). `None` reads from the beginning.
`max_count`默认default `100`	int	最大数量（类型 `int`）。Maximum entries to return.

返回Returns list[tuple[str, dict]]

返回元组[str,dict]列表。`list[tuple[str, dict]]`:

源码Source agentscope/app/message_bus/_redis_message_bus.py:350–384

log_trim

async def log_trim(key: str, before_id: str | None = None) -> None

async

执行日志裁剪相关操作。

Trim the replay log at key.

参数Parameters

名称Name	类型Type	说明Description
`key`	str	键（类型 `str`）。Stream key for the replay log.
`before_id`默认default `None`	str \| None	之前标识（类型 `str \| None`）。Drop all entries with id strictly older than this. `None` deletes the entire key.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/message_bus/_redis_message_bus.py:386–404

registry_set

async def registry_set(namespace: str, field: str, value: str, *, ttl_secs: int | None = None) -> None

async

执行注册表设置相关操作。

Set field in the Redis Hash at namespace.

参数Parameters

名称Name	类型Type	说明Description
`namespace`	str	命名空间（类型 `str`）。Hash key.
`field`	str	field（类型 `str`）。Hash field.
`value`	str	值（类型 `str`）。Value to store.
`ttl_secs`kw默认default `None`	int \| None	存活时间 secs（类型 `int \| None`）。Refresh the key's expiry (sliding TTL).

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/message_bus/_redis_message_bus.py:410–432

registry_set_if

async def registry_set_if(namespace: str, field: str, value: str, *, expected: str, ttl_secs: int | None = None) -> bool

async

执行注册表设置 if相关操作。

Compare-and-set one Hash field via Lua.

参数Parameters

名称Name	类型Type	说明Description
`namespace`	str	命名空间（类型 `str`）。The namespace.
`field`	str	field（类型 `str`）。The field.
`value`	str	值（类型 `str`）。The value.
`expected`kw	str	expected（类型 `str`）。The expected.
`ttl_secs`kw默认default `None`	int \| None	存活时间 secs（类型 `int \| None`）。The ttl secs.

返回Returns bool

返回布尔值，表示判断结果。Returns a boolean result.

源码Source agentscope/app/message_bus/_redis_message_bus.py:434–453

registry_pop

async def registry_pop(namespace: str, field: str) -> str | None

async

执行 registry_pop 方法。

HGET + HDEL as one Lua step.

参数Parameters

名称Name	类型Type	说明Description
`namespace`	str	命名空间（类型 `str`）。The namespace.
`field`	str	field（类型 `str`）。The field.

返回Returns str | None

返回str|None。Returns str none.

源码Source agentscope/app/message_bus/_redis_message_bus.py:455–462

registry_del

async def registry_del(namespace: str, field: str) -> None

async

执行 registry_del 方法。

Remove field from the Redis Hash at namespace.

参数Parameters

名称Name	类型Type	说明Description
`namespace`	str	命名空间（类型 `str`）。Hash key.
`field`	str	field（类型 `str`）。Hash field to remove.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/message_bus/_redis_message_bus.py:464–473

registry_exists

async def registry_exists(namespace: str, field: str) -> bool

async

执行 registry_exists 方法。

Return whether field exists in the Hash at namespace.

参数Parameters

名称Name	类型Type	说明Description
`namespace`	str	命名空间（类型 `str`）。Hash key.
`field`	str	field（类型 `str`）。Hash field to check.

返回Returns bool

返回布尔值，表示判断结果。`bool`:

源码Source agentscope/app/message_bus/_redis_message_bus.py:475–488

registry_getall

async def registry_getall(namespace: str) -> dict[str, str]

async

设置该参数。

Return all field-value pairs from the Hash at namespace.

参数Parameters

名称Name	类型Type	说明Description
`namespace`	str	命名空间（类型 `str`）。Hash key.

返回Returns dict[str, str]

返回dict[str,str]。`dict[str, str]`:

源码Source agentscope/app/message_bus/_redis_message_bus.py:490–504

registry_get

async def registry_get(namespace: str, field: str) -> str | None

async

执行注册表读取相关操作。

Return a single field value from the Hash at namespace.

参数Parameters

名称Name	类型Type	说明Description
`namespace`	str	命名空间（类型 `str`）。Hash key.
`field`	str	field（类型 `str`）。Field to retrieve.

返回Returns str | None

返回str|None。`str | None`:

源码Source agentscope/app/message_bus/_redis_message_bus.py:506–523

registry_drop

async def registry_drop(namespace: str) -> None

async

设置该参数。

Delete the entire Hash at namespace.

参数Parameters

名称Name	类型Type	说明Description
`namespace`	str	命名空间（类型 `str`）。Hash key to delete.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/message_bus/_redis_message_bus.py:525–532

publish

async def publish(key: str, payload: dict) -> None

async

执行发布相关操作。

Publish payload on the broadcast channel key.

参数Parameters

名称Name	类型Type	说明Description
`key`	str	键（类型 `str`）。Pub/Sub channel name.
`payload`	dict	负载（类型 `dict`）。JSON-serializable dict; encoded as the channel message body.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/message_bus/_redis_message_bus.py:545–562

async def subscribe(key: str, *, on_ready: Callable[[], None] | None = None) -> AsyncGenerator[dict, None]

async

执行订阅相关操作。

Yield broadcast payloads on key until the consumer closes the generator.

参数Parameters

名称Name	类型Type	说明Description
`key`	str	键（类型 `str`）。Pub/Sub channel name.
`on_ready`kw默认default `None`	Callable[[], None] \| None	on 就绪（类型 `Callable[[], None] \| None`）。Called once after `SUBSCRIBE` has been issued (and before any payload is yielded). Lets callers block on a "subscription is live" event so they can publish immediately after starting a subscriber without losing the first message to a SUBSCRIBE/PUBLISH race.

返回Returns AsyncGenerator[dict, None]

以异步流的形式返回dict,None。`dict`:

源码Source agentscope/app/message_bus/_redis_message_bus.py:564–619

acquire_lock

async def acquire_lock(key: str, *, ttl_secs: int = 600) -> AsyncGenerator[None, None]

asyncasynccontextmanager

执行 acquire_lock 方法。

Implementation: - SET key NX EX ttl_secs to claim the lock atomically.

参数Parameters

名称Name	类型Type	说明Description
`key`	str	键（类型 `str`）。Lock identifier.
`ttl_secs`kw默认default `600`	int	存活时间 secs（类型 `int`）。Lease duration; auto-renewed via heartbeat.

返回Returns AsyncGenerator[None, None]

以异步流的形式返回None,None。`None`: while the lock is held.

源码Source agentscope/app/message_bus/_redis_message_bus.py:629–701

is_locked

async def is_locked(key: str) -> bool

async

判断是否 is_locked。

Return whether key currently holds a lock.

参数Parameters

名称Name	类型Type	说明Description
`key`	str	键（类型 `str`）。Lock identifier.

返回Returns bool

返回布尔值，表示判断结果。`bool`:

源码Source agentscope/app/message_bus/_redis_message_bus.py:703–715

try_lock

async def try_lock(key: str, *, ttl_secs: int = 600) -> bool

async

执行 try_lock 方法。

Non-blocking claim via SET key NX EX.

参数Parameters

名称Name	类型Type	说明Description
`key`	str	键（类型 `str`）。The key.
`ttl_secs`kw默认default `600`	int	存活时间 secs（类型 `int`）。The ttl secs.

返回Returns bool

返回布尔值，表示判断结果。Returns a boolean result.

源码Source agentscope/app/message_bus/_redis_message_bus.py:717–721

unlock

async def unlock(key: str) -> None

async

设置该参数。

Release a try_lock claim (best-effort DEL).

参数Parameters

名称Name	类型Type	说明Description
`key`	str	键（类型 `str`）。The key.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/message_bus/_redis_message_bus.py:723–725

RedisMessageBus

构造器摘要Constructor Summary

方法摘要Method Summary

构造器详情Constructor Detail

RedisMessageBus

方法详情Method Detail

__aenter__

aclose

__aexit__

get_client

queue_push

queue_drain

queue_delete

log_append

log_read

log_trim

registry_set

registry_set_if

registry_pop

registry_del

registry_exists

registry_getall

registry_get

registry_drop

publish

subscribe

acquire_lock

is_locked

try_lock

unlock

继承成员Inherited Members

aenter

aexit