类class agentscope.app.storage

RedisStorage

class RedisStorage(StorageBase)

源码Source agentscope/app/storage/_redis_storage.py:56–2580

Redis 存储，负责持久化与读取数据。位于 agentscope.app.storage 包，继承自 StorageBase。核心成员包括 aclose、get_client、upsert_credential、list_credentials、get_credential、delete_credential。

The Redis storage implementation. Declared in agentscope.app.storage. Extends StorageBase.

嵌套类摘要Nested Class Summary

类Class	类型Kind	说明Description
`KeyConfig`	类class	键配置，保存相关配置项。 Key templates for all Redis keys used by :class:`RedisStorage`.

构造器摘要Constructor Summary

构造器Constructor	说明Description
`def __init__(host: str = 'localhost', port: int = 6379, db: int = 0, password: str \| None = None, connection_pool: Conn…`	初始化 `RedisStorage` 实例。 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.
`def get_client() -> Redis`	获取客户端。 Get the underlying Redis client instance.
`async def upsert_credential(user_id: str, credential_data:` `CredentialBase) -> str`async	执行 `upsert_credential` 方法。 Create or update a credential record for the given user.
`async def list_credentials(user_id: str) -> list[CredentialRecord]`async	列出凭证。 Return all credential records belonging to the given user.
`async def get_credential(user_id: str, credential_id: str) ->` `CredentialRecord \| None`async	获取凭证。 Fetch a single credential record by id.
`async def delete_credential(user_id: str, credential_id: str) -> bool`async	删除凭证。 Delete a credential record and remove it from the user's index.
`async def upsert_mcp(user_id: str, mcp_record:` `MCPRecord) -> str`async	创建或更新已安装的 MCP 配置。 Create or update an installed-MCP record for the given user.
`async def list_mcps(user_id: str) -> list[MCPRecord]`async	列出 `list_mcps`。 Return every installed-MCP record belonging to the user.
`async def get_mcp(user_id: str, mcp_id: str) ->` `MCPRecord \| None`async	获取MCP。 Fetch a single installed-MCP record by id.
`async def get_mcp_by_name(user_id: str, name: str) ->` `MCPRecord \| None`async	获取MCP 通过名称。 Fetch an installed-MCP record by its MCP name.
`async def delete_mcp(user_id: str, mcp_id: str) -> bool`async	删除MCP。 Delete an installed-MCP record and drop it from both indexes.
`async def upsert_skill(user_id: str, skill_record:` `SkillRecord) -> str`async	执行 `upsert_skill` 方法。 Create or update an installed-skill record for the user.
`async def list_skills(user_id: str) -> list[SkillRecord]`async	列出技能。 Return every installed-skill record belonging to the user.
`async def get_skill(user_id: str, skill_id: str) ->` `SkillRecord \| None`async	获取技能。 Fetch a single installed-skill record by id.
`async def get_skill_by_name(user_id: str, name: str) ->` `SkillRecord \| None`async	获取技能通过名称。 Fetch an installed-skill record by its skill name.
`async def delete_skill(user_id: str, skill_id: str) -> bool`async	删除技能。 Delete an installed-skill record and drop both index entries.
`async def upsert_agent(user_id: str, agent_record:` `AgentRecord) -> str`async	执行 `upsert_agent` 方法。 Persist an agent record and register it in the user's agent index.
`async def list_agents(user_id: str) -> list[AgentRecord]`async	列出智能体。 Return user-facing agent records (``source='user'``).
`async def get_agent(user_id: str, agent_id: str) ->` `AgentRecord \| None`async	获取智能体。 Fetch a single agent record by id.
`async def delete_agent(user_id: str, agent_id: str) -> bool`async	删除智能体。 Delete an agent record and cascade-delete its sessions, schedules, and any team back-references.
`async def upsert_session(user_id: str, agent_id: str, config:` `SessionConfig, state: AgentState \| None = None, session_i…`async	执行 `upsert_session` 方法。 Create or update a session for a (user, agent) pair.
`async def update_session_state(user_id: str, agent_id: str, session_id: str, state:` `AgentState) -> None`async	更新会话状态。 Update only the mutable state of an existing session.
`async def list_sessions(user_id: str, agent_id: str) -> list[SessionRecord]`async	列出会话。 Return all session records for a given (user, agent) pair.
`async def get_session(user_id: str, agent_id: str, session_id: str) ->` `SessionRecord \| None`async	获取会话。 Fetch a single session record by id.
`async def delete_session(user_id: str, agent_id: str, session_id: str) -> bool`async	删除会话。 Delete a session record and cascade clean-up.
`async def list_sessions_by_schedule(user_id: str, schedule_id: str) -> list[SessionRecord]`async	列出会话通过调度。 Return all sessions created by a given schedule.
`async def list_sessions_by_channel(user_id: str, channel_id: str) -> list[SessionRecord]`async	列出会话通过渠道。 Return all sessions derived from a given channel.
`async def upsert_schedule(user_id: str, record:` `ScheduleRecord) -> str`async	执行 `upsert_schedule` 方法。 Persist a cron task record and register it in the user and global indexes.
`async def get_schedule(user_id: str, schedule_id: str) ->` `ScheduleRecord \| None`async	获取调度。 Fetch a single cron task record by id.
`async def list_schedules(user_id: str) -> list[ScheduleRecord]`async	列出调度。 Return all cron task records belonging to the given user.
`async def delete_schedule(user_id: str, schedule_id: str) -> bool`async	删除调度。 Delete a cron task record, cascade-delete its execution sessions, and remove it from the user and global indexes.
`async def list_all_schedules() -> list[ScheduleRecord]`async	列出所有调度。 Return every schedule record across all users.
`async def upsert_channel(record:` `ChannelRecord, platform_bot_id: str) -> str`async	执行 `upsert_channel` 方法。 Persist a channel record and refresh its indexes.
`async def get_channel(channel_id: str) ->` `ChannelRecord \| None`async	获取渠道。 Fetch a channel record by its global id.
`async def list_channels(user_id: str) -> list[ChannelRecord]`async	列出渠道。 Return all channel records owned by the given user.
`async def list_all_channels() -> list[ChannelRecord]`async	列出所有渠道。 Return every channel record across all users.
`async def delete_channel(channel_id: str, platform_bot_id: str) -> bool`async	删除渠道。 Delete a channel record and clean up all indexes.
`async def get_channel_id_by_platform_bot_id(platform_bot_id: str) -> str \| None`async	获取 `get_channel_id_by_platform_bot_id`。 Return the channel id bound to a platform bot, if any.
`async def upsert_message(user_id: str, session_id: str, msg:` `Msg) -> None`async	执行 `upsert_message` 方法。 Persist a message to the session's message list.
`async def delete_message(user_id: str, session_id: str, message_id: str) -> bool`async	删除消息。 Delete the stored message matching ``message_id``.
`async def get_message(user_id: str, session_id: str, message_id: str) ->` `Msg \| None`async	获取消息。 Fetch a single message by id from the session's message list.
`async def list_messages(user_id: str, session_id: str, limit: int = 50, before: str \| None = None, **kwargs: Any) -> tu…`async	列出消息。 Return the most recent messages with cursor-based pagination.
`async def upsert_sop(user_id: str, record:` `SOPRecord) -> SOPRecord`async	执行 `upsert_sop` 方法。 Persist a procedure and index it under the user.
`async def get_sop(user_id: str, sop_id: str) ->` `SOPRecord \| None`async	获取SOP。 Fetch one procedure by id.
`async def list_sops(user_id: str) -> list[SOPRecord]`async	列出 `list_sops`。 Return every procedure in the user's index that still exists.
`async def delete_sop(user_id: str, sop_id: str) -> bool`async	删除SOP。 Delete a procedure and every run of it.
`async def upsert_sop_run(user_id: str, record:` `SOPRunRecord) -> SOPRunRecord`async	执行 `upsert_sop_run` 方法。 Persist a run and index it under the user and its procedure.
`async def get_sop_run(user_id: str, sop_run_id: str) ->` `SOPRunRecord \| None`async	获取SOP 运行。 Fetch one run by id.
`async def list_sop_runs(user_id: str, sop_id: str \| None = None, phase:` `SOPPhase \| None = None) -> list[SOPRunRecord]`async	列出 `list_sop_runs`。 Return the user's runs, newest first. ``phase`` is filtered after reading, since Redis has no index for it.
`async def update_sop_run(user_id: str, sop_run_id: str, state:` `SOPRunState, sessions: dict[str, str] \| None = None) ->…`async	更新SOP 运行。 Read-modify-write the run record; raises if absent.
`async def delete_sop_run(user_id: str, sop_run_id: str) -> bool`async	删除SOP 运行。 Delete one run, its sessions, and both index entries.
`async def upsert_team(user_id: str, record:` `TeamRecord) -> TeamRecord`async	执行 `upsert_team` 方法。 Persist a team record and register it in the user's team index.
`async def get_team(user_id: str, team_id: str) ->` `TeamRecord \| None`async	获取团队。 Fetch a single team record by id.
`async def list_teams(user_id: str) -> list[TeamRecord]`async	列出 `list_teams`。 Return all team records belonging to the given user.
`async def set_session_team_id(user_id: str, session_id: str, team_id: str \| None) -> None`async	设置会话团队标识。 Set or clear ``team_id`` on an existing session record.
`async def delete_team(user_id: str, team_id: str) -> bool`async	删除团队。 Delete a team record and cascade-clean its members by role.
`async def upsert_knowledge_base(user_id: str, record:` `KnowledgeBaseRecord) -> KnowledgeBaseRecord`async	执行 `upsert_knowledge_base` 方法。 Persist a knowledge base record and register it in the user index.
`async def get_knowledge_base(user_id: str, knowledge_base_id: str) ->` `KnowledgeBaseRecord \| None`async	获取知识库。 Fetch a single knowledge base record by id.
`async def list_knowledge_bases(user_id: str) -> list[KnowledgeBaseRecord]`async	列出 `list_knowledge_bases`。 List all knowledge base records belonging to the given user.
`async def delete_knowledge_base(user_id: str, knowledge_base_id: str) -> bool`async	删除知识库。 Delete a knowledge base record and remove it from the index.
`async def upsert_knowledge_document(user_id: str, record:` `KnowledgeDocumentRecord) -> KnowledgeDocumentRecord`async	执行 `upsert_knowledge_document` 方法。 Persist a document record and update its indexes.
`async def get_knowledge_document(user_id: str, knowledge_base_id: str, document_id: str) ->` `KnowledgeDocumentRecord \| N…`async	获取知识库文档。 Fetch a single knowledge document record by id.
`async def list_knowledge_documents(user_id: str, knowledge_base_id: str) -> list[KnowledgeDocumentRecord]`async	列出知识库文档。 List all documents in a knowledge base.
`async def delete_knowledge_document(user_id: str, knowledge_base_id: str, document_id: str) -> bool`async	删除知识库文档。 Delete a document record and remove it from the indexes.
`async def update_knowledge_document_status(user_id: str, knowledge_base_id: str, document_id: str, status: KnowledgeDoc…`async	更新知识库文档状态。 Update only the status-related fields of a document record.
`async def acquire_knowledge_document_lease(user_id: str, knowledge_base_id: str, document_id: str, processing_node: str…`async	执行 `acquire_knowledge_document_lease` 方法。 Compare-and-swap acquisition of the processing lease.
`async def renew_knowledge_document_lease(user_id: str, knowledge_base_id: str, document_id: str, processing_node: str,…`async	执行 `renew_knowledge_document_lease` 方法。 Extend the lease this worker already holds.
`async def release_knowledge_document_lease(user_id: str, knowledge_base_id: str, document_id: str, processing_node: str…`async	执行 `release_knowledge_document_lease` 方法。 Release the lease only if this worker still owns it.
`async def list_knowledge_documents_with_expired_lease(now: datetime \| None = None) -> list[KnowledgeDocumentRecord]`async	列出 `list_knowledge_documents_with_expired_lease`。 Return non-terminal documents whose lease has expired.
`async def list_knowledge_documents_pending_since(threshold: datetime) -> list[KnowledgeDocumentRecord]`async	列出 `list_knowledge_documents_pending_since`。 Return documents stuck in ``pending`` since before ``threshold``.

构造器详情Constructor Detail

RedisStorage

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

初始化 RedisStorage 实例。

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 database index.
`password`默认default `None`	str \| None	密码（类型 `str \| None`）。Redis password if required.
`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`. Extra `**kwargs` (e.g. `max_connections`) are forwarded to the pool constructor only when the pool is created internally.
`key_ttl`默认default `None`	int \| None	键存活时间（类型 `int \| None`）。Expire time in seconds for record keys. Refreshed on every write (sliding TTL). If `None`, keys do not expire.
`key_config`默认default `None`	'RedisStorage.KeyConfig \| None'	键配置（类型 `'RedisStorage.KeyConfig \| None'`）。Key template configuration. Defaults to `RedisStorage.KeyConfig()`.
`kwargs`	Any	关键字参数（类型 `Any`）。The kwargs.

源码Source agentscope/app/storage/_redis_storage.py:152–201

方法详情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

返回自身。Returns self.

源码Source agentscope/app/storage/_redis_storage.py:217–247

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/storage/_redis_storage.py:249–257

aexit

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

async

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

Exit the async context manager.

参数Parameters

名称Name	类型Type	说明Description
`exc_type`	type[BaseException] \| None	exc 类型（类型 `type[BaseException] \| None`）。The exc type.
`exc_value`	BaseException \| None	exc 值（类型 `BaseException \| None`）。The exc value.
`traceback`	Any	traceback（类型 `Any`）。The traceback.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/storage/_redis_storage.py:259–266

get_client

def get_client() -> Redis

获取客户端。

Get the underlying Redis client instance.

返回Returns Redis

返回Redis。Returns redis.

源码Source agentscope/app/storage/_redis_storage.py:268–270

upsert_credential

async def upsert_credential(user_id: str, credential_data: CredentialBase) -> str

async

执行 upsert_credential 方法。

If credential_data.id is set and the record already exists, the existing record's data field is updated in place (preserving created_at).

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`credential_data`	CredentialBase	凭证数据（类型 `CredentialBase`）。Input data containing an optional `id` and the credential `data` dict.

返回Returns str

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

源码Source agentscope/app/storage/_redis_storage.py:303–369

list_credentials

async def list_credentials(user_id: str) -> list[CredentialRecord]

async

列出凭证。

Reads the per-user credential index Set to obtain all ids, then fetches each record individually.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.

返回Returns list[CredentialRecord]

返回CredentialRecord列表。`list[CredentialRecord]`: All credential records for the user.

源码Source agentscope/app/storage/_redis_storage.py:371–400

get_credential

async def get_credential(user_id: str, credential_id: str) -> CredentialRecord | None

async

获取凭证。

Fetch a single credential record by id.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识（类型 `str`）。The user id.
`credential_id`	str	凭证标识（类型 `str`）。The credential id.

返回Returns CredentialRecord | None

返回CredentialRecord|None。Returns credential record none.

源码Source agentscope/app/storage/_redis_storage.py:402–414

delete_credential

async def delete_credential(user_id: str, credential_id: str) -> bool

async

删除凭证。

Delete a credential record and remove it from the user's index.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`credential_id`	str	凭证到删除的标识。The id of the credential to delete.

返回Returns bool

返回布尔值，表示判断结果。`bool`: `True` if the record existed and was deleted,

源码Source agentscope/app/storage/_redis_storage.py:416–442

upsert_mcp

async def upsert_mcp(user_id: str, mcp_record: MCPRecord) -> str

async

创建或更新已安装的 MCP 配置。

Keeps the name index in step with the write: the name is claimed for this record, and a rename releases the old one.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`mcp_record`	MCPRecord	MCP 记录（类型 `MCPRecord`）。The record to write.

返回Returns str

`str`:标识的已创建或已更新。`str`: The id of the created or updated record.

异常Raises

可能抛出 `Exception`。`ValueError`: When another record already uses the name.

源码Source agentscope/app/storage/_redis_storage.py:444–501

list_mcps

async def list_mcps(user_id: str) -> list[MCPRecord]

async

列出 list_mcps。

Return every installed-MCP record belonging to the user.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.

返回Returns list[MCPRecord]

返回MCPRecord列表。`list[MCPRecord]`: All installed-MCP records for the user.

源码Source agentscope/app/storage/_redis_storage.py:503–525

get_mcp

async def get_mcp(user_id: str, mcp_id: str) -> MCPRecord | None

async

获取MCP。

Fetch a single installed-MCP record by id.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识（类型 `str`）。The user id.
`mcp_id`	str	MCP 标识（类型 `str`）。The mcp id.

返回Returns MCPRecord | None

返回MCPRecord|None。Returns MCP record none.

源码Source agentscope/app/storage/_redis_storage.py:527–535

get_mcp_by_name

async def get_mcp_by_name(user_id: str, name: str) -> MCPRecord | None

async

获取MCP 通过名称。

Fetch an installed-MCP record by its MCP name.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识（类型 `str`）。The user id.
`name`	str	名称（类型 `str`）。The name.

返回Returns MCPRecord | None

返回MCPRecord|None。Returns MCP record none.

源码Source agentscope/app/storage/_redis_storage.py:537–547

delete_mcp

async def delete_mcp(user_id: str, mcp_id: str) -> bool

async

删除MCP。

Delete an installed-MCP record and drop it from both indexes.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`mcp_id`	str	MCP 标识（类型 `str`）。The id of the record to delete.

返回Returns bool

返回布尔值，表示判断结果。`bool`: `True` if the record existed and was deleted.

源码Source agentscope/app/storage/_redis_storage.py:549–577

upsert_skill

async def upsert_skill(user_id: str, skill_record: SkillRecord) -> str

async

执行 upsert_skill 方法。

Mirrors :meth:upsert_mcp — see it for the name-index rules.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`skill_record`	SkillRecord	技能记录（类型 `SkillRecord`）。The record to write.

返回Returns str

`str`:标识的已创建或已更新。`str`: The id of the created or updated record.

异常Raises

可能抛出 `Exception`。`ValueError`: When another record already uses the name.

源码Source agentscope/app/storage/_redis_storage.py:579–632

list_skills

async def list_skills(user_id: str) -> list[SkillRecord]

async

列出技能。

Return every installed-skill record belonging to the user.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.

返回Returns list[SkillRecord]

返回SkillRecord列表。`list[SkillRecord]`: All installed-skill records for the user.

源码Source agentscope/app/storage/_redis_storage.py:634–656

get_skill

async def get_skill(user_id: str, skill_id: str) -> SkillRecord | None

async

获取技能。

Fetch a single installed-skill record by id.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识（类型 `str`）。The user id.
`skill_id`	str	技能标识（类型 `str`）。The skill id.

返回Returns SkillRecord | None

返回SkillRecord|None。Returns skill record none.

源码Source agentscope/app/storage/_redis_storage.py:658–670

get_skill_by_name

async def get_skill_by_name(user_id: str, name: str) -> SkillRecord | None

async

获取技能通过名称。

Fetch an installed-skill record by its skill name.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识（类型 `str`）。The user id.
`name`	str	名称（类型 `str`）。The name.

返回Returns SkillRecord | None

返回SkillRecord|None。Returns skill record none.

源码Source agentscope/app/storage/_redis_storage.py:672–684

delete_skill

async def delete_skill(user_id: str, skill_id: str) -> bool

async

删除技能。

Delete an installed-skill record and drop both index entries.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`skill_id`	str	技能标识（类型 `str`）。The id of the record to delete.

返回Returns bool

返回布尔值，表示判断结果。`bool`: `True` if the record existed and was deleted.

源码Source agentscope/app/storage/_redis_storage.py:686–721

upsert_agent

async def upsert_agent(user_id: str, agent_record: AgentRecord) -> str

async

执行 upsert_agent 方法。

The caller is responsible for constructing the full AgentRecord (including its id).

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`agent_record`	AgentRecord	智能体记录（类型 `AgentRecord`）。The fully-populated agent record to store.

返回Returns str

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

源码Source agentscope/app/storage/_redis_storage.py:723–752

list_agents

async def list_agents(user_id: str) -> list[AgentRecord]

async

列出智能体。

Reads the per-user agent index Set to obtain all ids, fetches each record individually, and filters out team-spawned workers (source='team') — those are scoped to a team and only addressable via team detail / direct id lookup, not enumerated as part of the user's regular a…

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.

返回Returns list[AgentRecord]

返回AgentRecord列表。`list[AgentRecord]`:

源码Source agentscope/app/storage/_redis_storage.py:754–788

get_agent

async def get_agent(user_id: str, agent_id: str) -> AgentRecord | None

async

获取智能体。

Fetch a single agent record by id.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识（类型 `str`）。The user id.
`agent_id`	str	智能体标识（类型 `str`）。The agent id.

返回Returns AgentRecord | None

返回AgentRecord|None。Returns agent record none.

源码Source agentscope/app/storage/_redis_storage.py:790–802

delete_agent

async def delete_agent(user_id: str, agent_id: str) -> bool

async

删除智能体。

Cascade order: 1.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`agent_id`	str	智能体到删除的标识。The id of the agent to delete.

返回Returns bool

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

源码Source agentscope/app/storage/_redis_storage.py:804–882

upsert_session

async def upsert_session(user_id: str, agent_id: str, config: SessionConfig, state: AgentState | None = None, session_id: str | None = None, origin: SessionOrigin | None = None, source: str | None = None, source_schedule_id: str | None = None, source_chat_id: str | None = None, source_chat_name: str | None = None, source_channel_id: str | None = None) -> SessionRecord

async

执行 upsert_session 方法。

When session_id is provided the existing session is updated.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识（类型 `str`）。The user id.
`agent_id`	str	智能体标识（类型 `str`）。The agent id.
`config`	SessionConfig	配置对象（类型 `SessionConfig`）。The config.
`state`默认default `None`	AgentState \| None	状态对象（类型 `AgentState \| None`）。The state.
`session_id`默认default `None`	str \| None	会话标识（类型 `str \| None`）。The session id.
`origin`默认default `None`	SessionOrigin \| None	来源（类型 `SessionOrigin \| None`）。The origin.
`source`默认default `None`	str \| None	来源（类型 `str \| None`）。The source.
`source_schedule_id`默认default `None`	str \| None	来源调度标识（类型 `str \| None`）。The source schedule id.
`source_chat_id`默认default `None`	str \| None	来源聊天标识（类型 `str \| None`）。The source chat id.
`source_chat_name`默认default `None`	str \| None	来源聊天名称（类型 `str \| None`）。The source chat name.
`source_channel_id`默认default `None`	str \| None	来源渠道标识（类型 `str \| None`）。The source channel id.

返回Returns SessionRecord

返回SessionRecord。Returns session record.

源码Source agentscope/app/storage/_redis_storage.py:884–967

update_session_state

async def update_session_state(user_id: str, agent_id: str, session_id: str, state: AgentState) -> None

async

更新会话状态。

Update only the mutable state of an existing session.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识（类型 `str`）。The user id.
`agent_id`	str	智能体标识（类型 `str`）。The agent id.
`session_id`	str	会话标识（类型 `str`）。The session id.
`state`	AgentState	状态对象（类型 `AgentState`）。The state.

返回Returns None

无返回值。No value is returned.

异常Raises

KeyError 可能抛出 `KeyError`。If the session does not exist.

源码Source agentscope/app/storage/_redis_storage.py:969–992

list_sessions

async def list_sessions(user_id: str, agent_id: str) -> list[SessionRecord]

async

列出会话。

Reads the per-agent session index Set to obtain all session ids, then fetches each record individually.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`agent_id`	str	智能体标识（类型 `str`）。The agent id whose sessions to list.

返回Returns list[SessionRecord]

返回SessionRecord列表。`list[SessionRecord]`: All session records for the (user, agent)

源码Source agentscope/app/storage/_redis_storage.py:994–1031

get_session

async def get_session(user_id: str, agent_id: str, session_id: str) -> SessionRecord | None

async

获取会话。

Fetch a single session record by id.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识（类型 `str`）。The user id.
`agent_id`	str	智能体标识（类型 `str`）。The agent id.
`session_id`	str	会话标识（类型 `str`）。The session id.

返回Returns SessionRecord | None

返回SessionRecord|None。Returns session record none.

源码Source agentscope/app/storage/_redis_storage.py:1033–1048

delete_session

async def delete_session(user_id: str, agent_id: str, session_id: str) -> bool

async

删除会话。

Cascades: - Existing: per-session message log, schedule-session index entry.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`agent_id`	str	智能体标识（类型 `str`）。The id of the agent that owns the session (used to clean up the per-agent session index).
`session_id`	str	会话到删除的标识。The id of the session to delete.

返回Returns bool

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

源码Source agentscope/app/storage/_redis_storage.py:1050–1143

list_sessions_by_schedule

async def list_sessions_by_schedule(user_id: str, schedule_id: str) -> list[SessionRecord]

async

列出会话通过调度。

Return all sessions created by a given schedule.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识（类型 `str`）。The user id.
`schedule_id`	str	调度标识（类型 `str`）。The schedule id.

返回Returns list[SessionRecord]

返回SessionRecord列表。Returns a list of session record.

源码Source agentscope/app/storage/_redis_storage.py:1145–1169

list_sessions_by_channel

async def list_sessions_by_channel(user_id: str, channel_id: str) -> list[SessionRecord]

async

列出会话通过渠道。

Return all sessions derived from a given channel.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识（类型 `str`）。The user id.
`channel_id`	str	渠道标识（类型 `str`）。The channel id.

返回Returns list[SessionRecord]

返回SessionRecord列表。Returns a list of session record.

源码Source agentscope/app/storage/_redis_storage.py:1171–1195

upsert_schedule

async def upsert_schedule(user_id: str, record: ScheduleRecord) -> str

async

执行 upsert_schedule 方法。

Persist a cron task record and register it in the user and global indexes.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识（类型 `str`）。The user id.
`record`	ScheduleRecord	记录（类型 `ScheduleRecord`）。The record.

返回Returns str

返回字符串结果。Returns a string.

源码Source agentscope/app/storage/_redis_storage.py:1197–1216

get_schedule

async def get_schedule(user_id: str, schedule_id: str) -> ScheduleRecord | None

async

获取调度。

Fetch a single cron task record by id.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识（类型 `str`）。The user id.
`schedule_id`	str	调度标识（类型 `str`）。The schedule id.

返回Returns ScheduleRecord | None

返回ScheduleRecord|None。Returns schedule record none.

源码Source agentscope/app/storage/_redis_storage.py:1218–1232

list_schedules

async def list_schedules(user_id: str) -> list[ScheduleRecord]

async

列出调度。

Return all cron task records belonging to the given user.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识（类型 `str`）。The user id.

返回Returns list[ScheduleRecord]

返回ScheduleRecord列表。Returns a list of schedule record.

源码Source agentscope/app/storage/_redis_storage.py:1234–1252

delete_schedule

async def delete_schedule(user_id: str, schedule_id: str) -> bool

async

删除调度。

Delete a cron task record, cascade-delete its execution sessions, and remove it from the user and global indexes.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识（类型 `str`）。The user id.
`schedule_id`	str	调度标识（类型 `str`）。The schedule id.

返回Returns bool

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

源码Source agentscope/app/storage/_redis_storage.py:1254–1293

list_all_schedules

async def list_all_schedules() -> list[ScheduleRecord]

async

列出所有调度。

Reads the global schedule index (a Redis Set of user_id:schedule_id pairs) and fetches each record individually.

返回Returns list[ScheduleRecord]

返回ScheduleRecord列表。`list[ScheduleRecord]`: All schedule records in the store.

源码Source agentscope/app/storage/_redis_storage.py:1295–1320

upsert_channel

async def upsert_channel(record: ChannelRecord, platform_bot_id: str) -> str

async

执行 upsert_channel 方法。

Persist a channel record and refresh its indexes.

参数Parameters

名称Name	类型Type	说明Description
`record`	ChannelRecord	记录（类型 `ChannelRecord`）。The record.
`platform_bot_id`	str	平台 bot 标识（类型 `str`）。The platform bot id.

返回Returns str

返回字符串结果。Returns a string.

源码Source agentscope/app/storage/_redis_storage.py:1326–1345

get_channel

async def get_channel(channel_id: str) -> ChannelRecord | None

async

获取渠道。

Fetch a channel record by its global id.

参数Parameters

名称Name	类型Type	说明Description
`channel_id`	str	渠道标识（类型 `str`）。The channel id.

返回Returns ChannelRecord | None

返回ChannelRecord|None。Returns channel record none.

源码Source agentscope/app/storage/_redis_storage.py:1347–1357

list_channels

async def list_channels(user_id: str) -> list[ChannelRecord]

async

列出渠道。

Stale entries whose record key has expired are purged from the user index (self-healing).

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识（类型 `str`）。The user id.

返回Returns list[ChannelRecord]

返回ChannelRecord列表。Returns a list of channel record.

源码Source agentscope/app/storage/_redis_storage.py:1359–1369

list_all_channels

async def list_all_channels() -> list[ChannelRecord]

async

列出所有渠道。

Return every channel record across all users.

返回Returns list[ChannelRecord]

返回ChannelRecord列表。Returns a list of channel record.

源码Source agentscope/app/storage/_redis_storage.py:1371–1377

delete_channel

async def delete_channel(channel_id: str, platform_bot_id: str) -> bool

async

删除渠道。

Delete a channel record and clean up all indexes.

参数Parameters

名称Name	类型Type	说明Description
`channel_id`	str	渠道标识（类型 `str`）。The channel id.
`platform_bot_id`	str	平台 bot 标识（类型 `str`）。The platform bot id.

返回Returns bool

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

源码Source agentscope/app/storage/_redis_storage.py:1399–1426

get_channel_id_by_platform_bot_id

async def get_channel_id_by_platform_bot_id(platform_bot_id: str) -> str | None

async

获取 get_channel_id_by_platform_bot_id。

Return the channel id bound to a platform bot, if any.

参数Parameters

名称Name	类型Type	说明Description
`platform_bot_id`	str	平台 bot 标识（类型 `str`）。The platform bot id.

返回Returns str | None

返回str|None。Returns str none.

源码Source agentscope/app/storage/_redis_storage.py:1428–1438

upsert_message

async def upsert_message(user_id: str, session_id: str, msg: Msg) -> None

async

执行 upsert_message 方法。

Persist a message to the session's message list.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识（类型 `str`）。The user id.
`session_id`	str	会话标识（类型 `str`）。The session id.
`msg`	Msg	消息（类型 `Msg`）。The msg.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/storage/_redis_storage.py:1460–1502

delete_message

async def delete_message(user_id: str, session_id: str, message_id: str) -> bool

async

删除消息。

Delete the stored message matching message_id.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识（类型 `str`）。The user id.
`session_id`	str	会话标识（类型 `str`）。The session id.
`message_id`	str	消息标识（类型 `str`）。The message id.

返回Returns bool

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

源码Source agentscope/app/storage/_redis_storage.py:1504–1527

get_message

async def get_message(user_id: str, session_id: str, message_id: str) -> Msg | None

async

获取消息。

Fetch a single message by id from the session's message list.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识（类型 `str`）。The user id.
`session_id`	str	会话标识（类型 `str`）。The session id.
`message_id`	str	消息标识（类型 `str`）。The message id.

返回Returns Msg | None

返回消息|None。Returns msg none.

源码Source agentscope/app/storage/_redis_storage.py:1529–1544

list_messages

async def list_messages(user_id: str, session_id: str, limit: int = 50, before: str | None = None, **kwargs: Any) -> tuple[list[Msg], bool]

async

列出消息。

Messages are returned in chronological order.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`session_id`	str	会话标识。The session id.
`limit`默认default `50`	int	数量上限（类型 `int`）。Maximum number of messages to return. Defaults to 50.
`before`默认default `None`	str \| None	之前（类型 `str \| None`）。A message ID used as the cursor. When provided, returns messages created before this message. Omit to get the latest page.
`kwargs`	Any	关键字参数（类型 `Any`）。The kwargs.

返回Returns tuple[list[Msg], bool]

(消息 in的`tuple[list[Msg], bool]`:元组。`tuple[list[Msg], bool]`: A tuple of (messages in

源码Source agentscope/app/storage/_redis_storage.py:1576–1634

upsert_sop

async def upsert_sop(user_id: str, record: SOPRecord) -> SOPRecord

async

执行 upsert_sop 方法。

Persist a procedure and index it under the user.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识（类型 `str`）。The user id.
`record`	SOPRecord	记录（类型 `SOPRecord`）。The record.

返回Returns SOPRecord

返回SOPRecord。Returns SOP record.

源码Source agentscope/app/storage/_redis_storage.py:1640–1651

get_sop

async def get_sop(user_id: str, sop_id: str) -> SOPRecord | None

async

获取SOP。

Fetch one procedure by id.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识（类型 `str`）。The user id.
`sop_id`	str	SOP 标识（类型 `str`）。The sop id.

返回Returns SOPRecord | None

返回SOPRecord|None。Returns SOP record none.

源码Source agentscope/app/storage/_redis_storage.py:1653–1658

list_sops

async def list_sops(user_id: str) -> list[SOPRecord]

async

列出 list_sops。

Return every procedure in the user's index that still exists.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识（类型 `str`）。The user id.

返回Returns list[SOPRecord]

返回SOPRecord列表。Returns a list of SOP record.

源码Source agentscope/app/storage/_redis_storage.py:1660–1670

delete_sop

async def delete_sop(user_id: str, sop_id: str) -> bool

async

删除SOP。

Delete a procedure and every run of it.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识（类型 `str`）。The user id.
`sop_id`	str	SOP 标识（类型 `str`）。The sop id.

返回Returns bool

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

源码Source agentscope/app/storage/_redis_storage.py:1672–1690

upsert_sop_run

async def upsert_sop_run(user_id: str, record: SOPRunRecord) -> SOPRunRecord

async

执行 upsert_sop_run 方法。

Persist a run and index it under the user and its procedure.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识（类型 `str`）。The user id.
`record`	SOPRunRecord	记录（类型 `SOPRunRecord`）。The record.

返回Returns SOPRunRecord

返回SOPRunRecord。Returns SOP run record.

源码Source agentscope/app/storage/_redis_storage.py:1692–1719

get_sop_run

async def get_sop_run(user_id: str, sop_run_id: str) -> SOPRunRecord | None

async

获取SOP 运行。

Fetch one run by id.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识（类型 `str`）。The user id.
`sop_run_id`	str	SOP 运行标识（类型 `str`）。The sop run id.

返回Returns SOPRunRecord | None

返回SOPRunRecord|None。Returns SOP run record none.

源码Source agentscope/app/storage/_redis_storage.py:1721–1734

list_sop_runs

async def list_sop_runs(user_id: str, sop_id: str | None = None, phase: SOPPhase | None = None) -> list[SOPRunRecord]

async

列出 list_sop_runs。

Return the user's runs, newest first.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识（类型 `str`）。The user id.
`sop_id`默认default `None`	str \| None	SOP 标识（类型 `str \| None`）。The sop id.
`phase`默认default `None`	SOPPhase \| None	阶段（类型 `SOPPhase \| None`）。The phase.

返回Returns list[SOPRunRecord]

返回SOPRunRecord列表。Returns a list of SOP run record.

源码Source agentscope/app/storage/_redis_storage.py:1736–1758

update_sop_run

async def update_sop_run(user_id: str, sop_run_id: str, state: SOPRunState, sessions: dict[str, str] | None = None) -> None

async

更新SOP 运行。

Read-modify-write the run record; raises if absent.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识（类型 `str`）。The user id.
`sop_run_id`	str	SOP 运行标识（类型 `str`）。The sop run id.
`state`	SOPRunState	状态对象（类型 `SOPRunState`）。The state.
`sessions`默认default `None`	dict[str, str] \| None	会话（类型 `dict[str, str] \| None`）。The sessions.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/storage/_redis_storage.py:1760–1774

delete_sop_run

async def delete_sop_run(user_id: str, sop_run_id: str) -> bool

async

删除SOP 运行。

Delete one run, its sessions, and both index entries.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识（类型 `str`）。The user id.
`sop_run_id`	str	SOP 运行标识（类型 `str`）。The sop run id.

返回Returns bool

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

源码Source agentscope/app/storage/_redis_storage.py:1776–1810

upsert_team

async def upsert_team(user_id: str, record: TeamRecord) -> TeamRecord

async

执行 upsert_team 方法。

Persist a team record and register it in the user's team index.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识（类型 `str`）。The owner user id. Used to scope both the record key and the per-user team index.
`record`	TeamRecord	记录（类型 `TeamRecord`）。The team record to persist. Its `id` is used as the primary key; an existing record with the same id is overwritten. `updated_at` is refreshed to `datetime.now()` before writing.

返回Returns TeamRecord

返回TeamRecord。`TeamRecord`:

源码Source agentscope/app/storage/_redis_storage.py:1816–1846

get_team

async def get_team(user_id: str, team_id: str) -> TeamRecord | None

async

获取团队。

Fetch a single team record by id.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`team_id`	str	团队标识（类型 `str`）。The team id to look up.

返回Returns TeamRecord | None

返回TeamRecord|None。`TeamRecord | None`:

源码Source agentscope/app/storage/_redis_storage.py:1848–1874

list_teams

async def list_teams(user_id: str) -> list[TeamRecord]

async

列出 list_teams。

Reads the per-user team index (a Redis Set of team ids) and fetches each record individually.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识（类型 `str`）。The owner user id whose teams to list.

返回Returns list[TeamRecord]

返回TeamRecord列表。`list[TeamRecord]`:

源码Source agentscope/app/storage/_redis_storage.py:1876–1905

set_session_team_id

async def set_session_team_id(user_id: str, session_id: str, team_id: str | None) -> None

async

设置会话团队标识。

Bypasses :meth:upsert_session because that method does not allow writing team_id (which is a relation column the application normally only mutates via team operations).

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`session_id`	str	会话标识（类型 `str`）。The session whose `team_id` should be updated.
`team_id`	str \| None	团队标识（类型 `str \| None`）。The new value. `None` detaches the session from any team (used by :meth:`delete_team` and by the team service when a session leaves a team).

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/storage/_redis_storage.py:1907–1944

delete_team

async def delete_team(user_id: str, team_id: str) -> bool

async

删除团队。

Cascade order (mirrors what SQL's ON DELETE CASCADE would do for the same set of foreign keys, but role-aware to honour the AgentInvite semantics — see :class:TeamMember): 1.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`team_id`	str	团队到删除的标识。The id of the team to delete.

返回Returns bool

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

源码Source agentscope/app/storage/_redis_storage.py:1946–2028

upsert_knowledge_base

async def upsert_knowledge_base(user_id: str, record: KnowledgeBaseRecord) -> KnowledgeBaseRecord

async

执行 upsert_knowledge_base 方法。

If a record with the same id already exists it is overwritten and updated_at is refreshed; created_at is preserved.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`record`	KnowledgeBaseRecord	记录（类型 `KnowledgeBaseRecord`）。The fully-populated record to store.

返回Returns KnowledgeBaseRecord

返回KnowledgeBaseRecord。`KnowledgeBaseRecord`:

源码Source agentscope/app/storage/_redis_storage.py:2034–2076

get_knowledge_base

async def get_knowledge_base(user_id: str, knowledge_base_id: str) -> KnowledgeBaseRecord | None

async

获取知识库。

Fetch a single knowledge base record by id.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`knowledge_base_id`	str	知识库基类标识。The knowledge base id.

返回Returns KnowledgeBaseRecord | None

返回KnowledgeBaseRecord|None。`KnowledgeBaseRecord | None`:

源码Source agentscope/app/storage/_redis_storage.py:2078–2101

list_knowledge_bases

async def list_knowledge_bases(user_id: str) -> list[KnowledgeBaseRecord]

async

列出 list_knowledge_bases。

Reads the per-user knowledge base index Set to obtain all ids, then fetches each record individually.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.

返回Returns list[KnowledgeBaseRecord]

返回KnowledgeBaseRecord列表。`list[KnowledgeBaseRecord]`:

源码Source agentscope/app/storage/_redis_storage.py:2103–2139

delete_knowledge_base

async def delete_knowledge_base(user_id: str, knowledge_base_id: str) -> bool

async

删除知识库。

Cascades the per-knowledge-base document index: every :class:KnowledgeDocumentRecord indexed under the KB is removed from storage so no orphan documents survive the KB deletion.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	所有者用户标识。The owner user id.
`knowledge_base_id`	str	知识库标识（类型 `str`）。The id of the record to delete.

返回Returns bool

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

源码Source agentscope/app/storage/_redis_storage.py:2141–2190

upsert_knowledge_document

async def upsert_knowledge_document(user_id: str, record: KnowledgeDocumentRecord) -> KnowledgeDocumentRecord

async

执行 upsert_knowledge_document 方法。

If a record with the same id already exists it is overwritten and updated_at refreshed; created_at is preserved.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识（类型 `str`）。The owner user id. Must match `record.user_id`.
`record`	KnowledgeDocumentRecord	记录（类型 `KnowledgeDocumentRecord`）。The fully-populated record to persist.

返回Returns KnowledgeDocumentRecord

返回KnowledgeDocumentRecord。`KnowledgeDocumentRecord`:

源码Source agentscope/app/storage/_redis_storage.py:2231–2283

get_knowledge_document

async def get_knowledge_document(user_id: str, knowledge_base_id: str, document_id: str) -> KnowledgeDocumentRecord | None

async

获取知识库文档。

Fetch a single knowledge document record by id.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识（类型 `str`）。The user id.
`knowledge_base_id`	str	知识库标识（类型 `str`）。The knowledge base id.
`document_id`	str	文档标识（类型 `str`）。The document id.

返回Returns KnowledgeDocumentRecord | None

返回KnowledgeDocumentRecord|None。Returns knowledge document record none.

源码Source agentscope/app/storage/_redis_storage.py:2285–2297

list_knowledge_documents

async def list_knowledge_documents(user_id: str, knowledge_base_id: str) -> list[KnowledgeDocumentRecord]

async

列出知识库文档。

Reads the per-KB document index Set and fetches each record.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识（类型 `str`）。The user id.
`knowledge_base_id`	str	知识库标识（类型 `str`）。The knowledge base id.

返回Returns list[KnowledgeDocumentRecord]

返回KnowledgeDocumentRecord列表。Returns a list of knowledge document record.

源码Source agentscope/app/storage/_redis_storage.py:2299–2322

delete_knowledge_document

async def delete_knowledge_document(user_id: str, knowledge_base_id: str, document_id: str) -> bool

async

删除知识库文档。

Delete a document record and remove it from the indexes.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识（类型 `str`）。The user id.
`knowledge_base_id`	str	知识库标识（类型 `str`）。The knowledge base id.
`document_id`	str	文档标识（类型 `str`）。The document id.

返回Returns bool

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

源码Source agentscope/app/storage/_redis_storage.py:2324–2345

update_knowledge_document_status

async def update_knowledge_document_status(user_id: str, knowledge_base_id: str, document_id: str, status: KnowledgeDocumentStatus, error: str | None = None, chunk_count: int | None = None) -> None

async

更新知识库文档状态。

Reads the record, mutates the status fields in memory, and writes it back.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识（类型 `str`）。The user id.
`knowledge_base_id`	str	知识库标识（类型 `str`）。The knowledge base id.
`document_id`	str	文档标识（类型 `str`）。The document id.
`status`	KnowledgeDocumentStatus	状态（类型 `KnowledgeDocumentStatus`）。The status.
`error`默认default `None`	str \| None	错误（类型 `str \| None`）。The error.
`chunk_count`默认default `None`	int \| None	分块数量（类型 `int \| None`）。The chunk count.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/storage/_redis_storage.py:2347–2374

acquire_knowledge_document_lease

async def acquire_knowledge_document_lease(user_id: str, knowledge_base_id: str, document_id: str, processing_node: str, lease_ttl: timedelta, now: datetime | None = None) -> bool

async

执行 acquire_knowledge_document_lease 方法。

Implementation is a read-modify-write under a per-document WATCH so two workers racing on the same document cannot both win.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识（类型 `str`）。The user id.
`knowledge_base_id`	str	知识库标识（类型 `str`）。The knowledge base id.
`document_id`	str	文档标识（类型 `str`）。The document id.
`processing_node`	str	processing 节点（类型 `str`）。The processing node.
`lease_ttl`	timedelta	lease 存活时间（类型 `timedelta`）。The lease ttl.
`now`默认default `None`	datetime \| None	当前时间（类型 `datetime \| None`）。The now.

返回Returns bool

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

源码Source agentscope/app/storage/_redis_storage.py:2376–2433

renew_knowledge_document_lease

async def renew_knowledge_document_lease(user_id: str, knowledge_base_id: str, document_id: str, processing_node: str, lease_ttl: timedelta, now: datetime | None = None) -> bool

async

执行 renew_knowledge_document_lease 方法。

Extend the lease this worker already holds.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识（类型 `str`）。The user id.
`knowledge_base_id`	str	知识库标识（类型 `str`）。The knowledge base id.
`document_id`	str	文档标识（类型 `str`）。The document id.
`processing_node`	str	processing 节点（类型 `str`）。The processing node.
`lease_ttl`	timedelta	lease 存活时间（类型 `timedelta`）。The lease ttl.
`now`默认default `None`	datetime \| None	当前时间（类型 `datetime \| None`）。The now.

返回Returns bool

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

源码Source agentscope/app/storage/_redis_storage.py:2435–2475

release_knowledge_document_lease

async def release_knowledge_document_lease(user_id: str, knowledge_base_id: str, document_id: str, processing_node: str) -> None

async

执行 release_knowledge_document_lease 方法。

Release the lease only if this worker still owns it.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识（类型 `str`）。The user id.
`knowledge_base_id`	str	知识库标识（类型 `str`）。The knowledge base id.
`document_id`	str	文档标识（类型 `str`）。The document id.
`processing_node`	str	processing 节点（类型 `str`）。The processing node.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/storage/_redis_storage.py:2477–2512

list_knowledge_documents_with_expired_lease

async def list_knowledge_documents_with_expired_lease(now: datetime | None = None) -> list[KnowledgeDocumentRecord]

async

列出 list_knowledge_documents_with_expired_lease。

Reads the global document index and filters in-memory — there is no Redis-side filter primitive that knows about our nested data.lease_expires_at field.

参数Parameters

名称Name	类型Type	说明Description
`now`默认default `None`	datetime \| None	当前时间（类型 `datetime \| None`）。The now.

返回Returns list[KnowledgeDocumentRecord]

返回KnowledgeDocumentRecord列表。Returns a list of knowledge document record.

源码Source agentscope/app/storage/_redis_storage.py:2514–2554

list_knowledge_documents_pending_since

async def list_knowledge_documents_pending_since(threshold: datetime) -> list[KnowledgeDocumentRecord]

async

列出 list_knowledge_documents_pending_since。

Return documents stuck in pending since before threshold.

参数Parameters

名称Name	类型Type	说明Description
`threshold`	datetime	阈值（类型 `datetime`）。The threshold.

返回Returns list[KnowledgeDocumentRecord]

返回KnowledgeDocumentRecord列表。Returns a list of knowledge document record.

源码Source agentscope/app/storage/_redis_storage.py:2556–2580

嵌套类详情Nested Class Detail

KeyConfig

class KeyConfig

键配置，保存相关配置项。位于 agentscope.app.storage 包。

Nested on :class:RedisStorage because customising key prefixes is meaningful only for this backend; users tweak them via RedisStorage(key_config=RedisStorage.KeyConfig(...)).