类class agentscope.app.storage

AsyncSQLAlchemyStorage

class AsyncSQLAlchemyStorage(StorageBase)

源码Source agentscope/app/storage/_sql/_storage.py:95–2223

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

Instantiate with any SQLAlchemy async URL (the concrete driver is a caller-installed dependency, not part of the sql extra):

__aenter__ builds the engine and (optionally) provisions the schema; aclose disposes the engine.

示例Example

storage = AsyncSQLAlchemyStorage("sqlite+aiosqlite:///./as.db")
async with storage:
    ...

构造器摘要Constructor Summary

构造器Constructor	说明Description
`def __init__(url: str, *, create_tables: bool = True, auto_migrate: bool = False, engine: 'AsyncEngine \| None' = None,…`	初始化 `AsyncSQLAlchemyStorage` 实例。 Configure the backend; nothing is opened until :meth:`__aenter__`.

方法摘要Method Summary

方法Method	说明Description
`async def __aenter__() -> Self`async	进入异步上下文并返回自身。 Build the engine (or adopt the external one) and optionally provision the schema.
`async def aclose() -> None`async	返回 `aclose` 结果。 Dispose the engine (if owned) and drop the session factory.
`async def upsert_credential(user_id: str, credential_data:` `CredentialBase) -> str`async	执行 `upsert_credential` 方法。 Create or update a credential record for user_id.
`async def list_credentials(user_id: str) -> list[CredentialRecord]`async	返回每个凭证用于用户_标识。 Return every credential for user_id.
`async def get_credential(user_id: str, credential_id: str) ->` `CredentialRecord \| None`async	获取凭证。 Fetch one credential record; enforces owner-scoping.
`async def delete_credential(user_id: str, credential_id: str) -> bool`async	删除凭证。 Delete a credential; owner-scoped.
`async def upsert_mcp(user_id: str, mcp_record:` `MCPRecord) -> str`async	创建或更新已安装的 MCP 配置。 Create or update an installed-MCP record for user_id.
`async def list_mcps(user_id: str) -> list[MCPRecord]`async	列出 `list_mcps`。 Return every installed-MCP record for user_id.
`async def get_mcp(user_id: str, mcp_id: str) ->` `MCPRecord \| None`async	获取MCP。 Fetch one installed-MCP record by id; owner-scoped.
`async def get_mcp_by_name(user_id: str, name: str) ->` `MCPRecord \| None`async	获取MCP 通过名称。 Fetch one installed-MCP record by its user-unique name.
`async def delete_mcp(user_id: str, mcp_id: str) -> bool`async	删除MCP。 Delete an installed-MCP record; owner-scoped.
`async def upsert_skill(user_id: str, skill_record:` `SkillRecord) -> str`async	执行 `upsert_skill` 方法。 Create or update an installed-skill record for user_id.
`async def list_skills(user_id: str) -> list[SkillRecord]`async	列出技能。 Return every installed-skill record for user_id.
`async def get_skill(user_id: str, skill_id: str) ->` `SkillRecord \| None`async	获取技能。 Fetch one installed-skill record by id; owner-scoped.
`async def get_skill_by_name(user_id: str, name: str) ->` `SkillRecord \| None`async	获取技能通过名称。 Fetch one installed-skill record by its user-unique name.
`async def delete_skill(user_id: str, skill_id: str) -> bool`async	删除技能。 Delete an installed-skill record; owner-scoped.
`async def upsert_agent(user_id: str, agent_record:` `AgentRecord) -> str`async	执行 `upsert_agent` 方法。 Persist an agent record (create or overwrite).
`async def list_agents(user_id: str) -> list[AgentRecord]`async	列出智能体。 Return the user's ``source='user'`` agents only.
`async def get_agent(user_id: str, agent_id: str) ->` `AgentRecord \| None`async	获取智能体。 Fetch one agent record; owner-scoped.
`async def delete_agent(user_id: str, agent_id: str) -> bool`async	删除智能体。 Delete an agent + cascade sessions, schedules, team refs.
`async def upsert_sop(user_id: str, record:` `SOPRecord) -> SOPRecord`async	执行 `upsert_sop` 方法。 Persist a procedure (create or overwrite).
`async def get_sop(user_id: str, sop_id: str) ->` `SOPRecord \| None`async	获取SOP。 Fetch one procedure; owner-scoped.
`async def list_sops(user_id: str) -> list[SOPRecord]`async	列出 `list_sops`。 Return the user's procedures.
`async def delete_sop(user_id: str, sop_id: str) -> bool`async	删除SOP。 Delete a procedure, its runs and their sessions atomically.
`async def upsert_sop_run(user_id: str, record:` `SOPRunRecord) -> SOPRunRecord`async	执行 `upsert_sop_run` 方法。 Persist a run (create or overwrite).
`async def get_sop_run(user_id: str, sop_run_id: str) ->` `SOPRunRecord \| None`async	获取SOP 运行。 Fetch one run; owner-scoped.
`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, filtered in the database.
`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 on the payload; raises if absent.
`async def delete_sop_run(user_id: str, sop_run_id: str) -> bool`async	删除SOP 运行。 Delete one run and its sessions in one transaction.
`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 — same shape as the Redis backend.
`async def set_session_team_id(user_id: str, session_id: str, team_id: str \| None) -> None`async	设置会话团队标识。 Single-column UPDATE — atomic on all supported dialects.
`async def update_session_state(user_id: str, agent_id: str, session_id: str, state:` `AgentState) -> None`async	更新会话状态。 Read-modify-write on the payload; raises if absent.
`async def list_sessions(user_id: str, agent_id: str) -> list[SessionRecord]`async	列出会话。 Sessions for a (user, agent) pair — newest first.
`async def get_session(user_id: str, agent_id: str, session_id: str) ->` `SessionRecord \| None`async	获取会话。 One session; owner-scoped.
`async def delete_session(user_id: str, agent_id: str, session_id: str) -> bool`async	删除会话。 Delete a session + cascade (message log, leader-team).
`async def list_sessions_by_schedule(user_id: str, schedule_id: str) -> list[SessionRecord]`async	列出会话通过调度。 Sessions created by schedule_id — newest first.
`async def list_sessions_by_channel(user_id: str, channel_id: str) -> list[SessionRecord]`async	列出会话通过渠道。 Sessions derived from channel_id — newest first.
`async def upsert_schedule(user_id: str, record:` `ScheduleRecord) -> str`async	执行 `upsert_schedule` 方法。 Persist a schedule record (create or overwrite).
`async def get_schedule(user_id: str, schedule_id: str) ->` `ScheduleRecord \| None`async	获取调度。 One schedule; owner-scoped.
`async def list_schedules(user_id: str) -> list[ScheduleRecord]`async	用于用户的所有调度。 All schedules for a user.
`async def delete_schedule(user_id: str, schedule_id: str) -> bool`async	删除调度。 Delete a schedule + cascade its execution sessions.
`async def list_all_schedules() -> list[ScheduleRecord]`async	列出所有调度。 Every schedule across every user (used on startup).
`async def upsert_channel(record:` `ChannelRecord, platform_bot_id: str) -> str`async	执行 `upsert_channel` 方法。 Persist a channel record and its bot-uniqueness column.
`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 user_id.
`async def list_all_channels() -> list[ChannelRecord]`async	列出所有渠道。 Every channel across every user.
`async def delete_channel(channel_id: str, platform_bot_id: str) -> bool`async	删除渠道。 Delete a channel record.
`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 bound to a platform bot, if any.
`async def upsert_message(user_id: str, session_id: str, msg:` `Msg) -> None`async	执行 `upsert_message` 方法。 Insert-or-update by ``(session_id, msg_id)``.
`async def delete_message(user_id: str, session_id: str, message_id: str) -> bool`async	删除消息。 Delete one message by ``(session_id, message_id)``.
`async def get_message(user_id: str, session_id: str, message_id: str) ->` `Msg \| None`async	获取消息。 Fetch a message by ``(session_id, msg_id)``.
`async def list_messages(user_id: str, session_id: str, limit: int = 50, before: str \| None = None, **kwargs: Any) -> tu…`async	列出消息。 Most-recent page of messages with cursor-based pagination.
`async def upsert_team(user_id: str, record:` `TeamRecord) -> TeamRecord`async	执行 `upsert_team` 方法。 Persist a team record.
`async def get_team(user_id: str, team_id: str) ->` `TeamRecord \| None`async	获取团队。 One team; owner-scoped.
`async def list_teams(user_id: str) -> list[TeamRecord]`async	列出 `list_teams`。 All teams for a user.
`async def delete_team(user_id: str, team_id: str) -> bool`async	删除团队。 Delete a team + role-aware member cleanup.
`async def upsert_knowledge_base(user_id: str, record:` `KnowledgeBaseRecord) -> KnowledgeBaseRecord`async	执行 `upsert_knowledge_base` 方法。 Create or update a KB record; enforces ``record.user_id``.
`async def get_knowledge_base(user_id: str, knowledge_base_id: str) ->` `KnowledgeBaseRecord \| None`async	获取知识库。 One KB; owner-scoped.
`async def list_knowledge_bases(user_id: str) -> list[KnowledgeBaseRecord]`async	用于用户的所有 KBs。 All KBs for a user.
`async def delete_knowledge_base(user_id: str, knowledge_base_id: str) -> bool`async	删除知识库。 Delete a KB; its document rows cascade natively via the ``knowledge_documents.knowledge_base_id`` foreign key (``ON DELETE CASCADE``) — one atomic statement, no app loop.
`async def upsert_knowledge_document(user_id: str, record:` `KnowledgeDocumentRecord) -> KnowledgeDocumentRecord`async	执行 `upsert_knowledge_document` 方法。 Create or update a document record; enforces ``record.user_id``.
`async def get_knowledge_document(user_id: str, knowledge_base_id: str, document_id: str) ->` `KnowledgeDocumentRecord \| N…`async	获取知识库文档。 One document; owner + KB scoped.
`async def list_knowledge_documents(user_id: str, knowledge_base_id: str) -> list[KnowledgeDocumentRecord]`async	列出知识库文档。 All documents inside a KB.
`async def delete_knowledge_document(user_id: str, knowledge_base_id: str, document_id: str) -> bool`async	删除知识库文档。 Delete one document.
`async def update_knowledge_document_status(user_id: str, knowledge_base_id: str, document_id: str, status: KnowledgeDoc…`async	更新知识库文档状态。 Fast-path column UPDATE + optional payload rewrite.
`async def acquire_knowledge_document_lease(user_id: str, knowledge_base_id: str, document_id: str, processing_node: str…`async	执行 `acquire_knowledge_document_lease` 方法。 Conditional UPDATE — atomic on every supported dialect.
`async def renew_knowledge_document_lease(user_id: str, knowledge_base_id: str, document_id: str, processing_node: str,…`async	执行 `renew_knowledge_document_lease` 方法。 Conditional UPDATE constrained to the current holder.
`async def release_knowledge_document_lease(user_id: str, knowledge_base_id: str, document_id: str, processing_node: str…`async	执行 `release_knowledge_document_lease` 方法。 Conditional UPDATE constrained to the current holder.
`async def list_knowledge_documents_with_expired_lease(now: datetime \| None = None) -> list[KnowledgeDocumentRecord]`async	列出 `list_knowledge_documents_with_expired_lease`。 Documents past their lease deadline, non-terminal, held.
`async def list_knowledge_documents_pending_since(threshold: datetime) -> list[KnowledgeDocumentRecord]`async	列出 `list_knowledge_documents_pending_since`。 Documents stuck in ``pending`` before threshold.

构造器详情Constructor Detail

AsyncSQLAlchemyStorage

def __init__(url: str, *, create_tables: bool = True, auto_migrate: bool = False, engine: 'AsyncEngine | None' = None, engine_kwargs: dict[str, Any] | None = None) -> None

初始化 AsyncSQLAlchemyStorage 实例。

Configure the backend; nothing is opened until :meth:__aenter__.

参数Parameters

名称Name	类型Type	说明Description
`url`	str	服务地址（类型 `str`）。A SQLAlchemy async URL (e.g. `sqlite+aiosqlite:///./as.db`, `postgresql+asyncpg://user:pw@host/db`). Ignored when `engine` is provided.
`create_tables`kw默认default `True`	bool	创建 tables（类型 `bool`）。When `True`, run `Base.metadata.create_all` on the engine at `__aenter__` — convenient for tests and single-node dev deployments where Alembic overhead is unwanted.
`auto_migrate`kw默认default `False`	bool	auto migrate（类型 `bool`）。When `True`, run `alembic upgrade head` against the packaged migration scripts at `__aenter__`. Convenient for single-node or dev deployments — every process boot brings the schema up to date. Not recommended for multi-replica production: two replicas racing on the same migration is unsafe. In that setup keep this `False` and run `alembic upgrade head` as a discrete deploy step.
`engine`kw默认default `None`	'AsyncEngine \| None'	引擎（类型 `'AsyncEngine \| None'`）。An externally managed engine. When supplied the engine is used as-is and not disposed by :meth:`aclose` — the caller owns its lifecycle. When omitted an engine is constructed from url on :meth:`__aenter__` and disposed on :meth:`aclose`.
`engine_kwargs`kw默认default `None`	dict[str, Any] \| None	引擎关键字参数（类型 `dict[str, Any] \| None`）。Extra keyword arguments forwarded to :func:`sqlalchemy.ext.asyncio.create_async_engine` when the engine is created internally (e.g. `echo=True`, `pool_size=20`). Ignored when engine is supplied.

源码Source agentscope/app/storage/_sql/_storage.py:111–165

方法详情Method Detail

aenter

async def __aenter__() -> Self

async

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

Build the engine (or adopt the external one) and optionally provision the schema.

返回Returns Self

返回自身。Returns self.

源码Source agentscope/app/storage/_sql/_storage.py:167–214

aclose

async def aclose() -> None

async

返回 aclose 结果。

Dispose the engine (if owned) and drop the session factory.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/storage/_sql/_storage.py:243–248

upsert_credential

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

async

执行 upsert_credential 方法。

Same contract as :meth:RedisStorage.upsert_credential — see that method for the id / name handling rules.

参数Parameters

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

返回Returns str

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

源码Source agentscope/app/storage/_sql/_storage.py:712–779

list_credentials

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

async

返回每个凭证用于用户_标识。

Return every credential for user_id.

参数Parameters

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

返回Returns list[CredentialRecord]

返回CredentialRecord列表。Returns a list of credential record.

源码Source agentscope/app/storage/_sql/_storage.py:781–797

get_credential

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

async

获取凭证。

Fetch one credential record; enforces owner-scoping.

参数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/_sql/_storage.py:799–809

delete_credential

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

async

删除凭证。

Delete a credential; owner-scoped.

参数Parameters

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

返回Returns bool

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

源码Source agentscope/app/storage/_sql/_storage.py:811–827

upsert_mcp

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

async

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

Same contract as :meth:RedisStorage.upsert_mcp.

参数Parameters

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

返回Returns str

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

源码Source agentscope/app/storage/_sql/_storage.py:838–851

list_mcps

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

async

列出 list_mcps。

Return every installed-MCP record for user_id.

参数Parameters

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

返回Returns list[MCPRecord]

返回MCPRecord列表。Returns a list of MCP record.

源码Source agentscope/app/storage/_sql/_storage.py:853–867

get_mcp

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

async

获取MCP。

Fetch one installed-MCP record by id; owner-scoped.

参数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/_sql/_storage.py:869–875

get_mcp_by_name

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

async

获取MCP 通过名称。

Fetch one installed-MCP record by its user-unique 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/_sql/_storage.py:877–894

delete_mcp

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

async

删除MCP。

Delete an installed-MCP record; owner-scoped.

参数Parameters

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

返回Returns bool

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

源码Source agentscope/app/storage/_sql/_storage.py:896–908

upsert_skill

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

async

执行 upsert_skill 方法。

Same contract as :meth:RedisStorage.upsert_skill.

参数Parameters

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

返回Returns str

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

源码Source agentscope/app/storage/_sql/_storage.py:910–927

list_skills

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

async

列出技能。

Return every installed-skill record for user_id.

参数Parameters

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

返回Returns list[SkillRecord]

返回SkillRecord列表。Returns a list of skill record.

源码Source agentscope/app/storage/_sql/_storage.py:929–943

get_skill

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

async

获取技能。

Fetch one installed-skill record by id; owner-scoped.

参数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/_sql/_storage.py:945–955

get_skill_by_name

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

async

获取技能通过名称。

Fetch one installed-skill record by its user-unique 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/_sql/_storage.py:957–974

delete_skill

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

async

删除技能。

Delete an installed-skill record; owner-scoped.

参数Parameters

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

返回Returns bool

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

源码Source agentscope/app/storage/_sql/_storage.py:976–992

upsert_agent

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

async

执行 upsert_agent 方法。

Persist an agent record (create or overwrite).

参数Parameters

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

返回Returns str

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

源码Source agentscope/app/storage/_sql/_storage.py:998–1006

list_agents

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

async

列出智能体。

Matches :meth:RedisStorage.list_agents — team-spawned workers (source='team') are addressable by id but are never enumerated in the user's regular agent list.

参数Parameters

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

返回Returns list[AgentRecord]

返回AgentRecord列表。Returns a list of agent record.

源码Source agentscope/app/storage/_sql/_storage.py:1008–1030

get_agent

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

async

获取智能体。

Fetch one agent record; owner-scoped.

参数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/_sql/_storage.py:1032–1042

delete_agent

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

async

删除智能体。

Mirrors the cascade order of :meth:RedisStorage.delete_agent.

参数Parameters

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

返回Returns bool

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

源码Source agentscope/app/storage/_sql/_storage.py:1044–1055

upsert_sop

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

async

执行 upsert_sop 方法。

Persist a procedure (create or overwrite).

参数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/_sql/_storage.py:1061–1064

get_sop

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

async

获取SOP。

Fetch one procedure; owner-scoped.

参数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/_sql/_storage.py:1066–1072

list_sops

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

async

列出 list_sops。

Return the user's procedures.

参数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/_sql/_storage.py:1074–1088

delete_sop

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

async

删除SOP。

Delete a procedure, its runs and their sessions atomically.

参数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/_sql/_storage.py:1090–1114

upsert_sop_run

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

async

执行 upsert_sop_run 方法。

Persist a run (create or overwrite).

参数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/_sql/_storage.py:1135–1142

get_sop_run

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

async

获取SOP 运行。

Fetch one run; owner-scoped.

参数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/_sql/_storage.py:1144–1154

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, filtered in the database.

参数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/_sql/_storage.py:1156–1174

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 on the payload; 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/_sql/_storage.py:1176–1197

delete_sop_run

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

async

删除SOP 运行。

Delete one run and its sessions in one transaction.

参数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/_sql/_storage.py:1199–1207

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 方法。

Create or update a session — same shape as the Redis backend.

参数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/_sql/_storage.py:1213–1260

set_session_team_id

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

async

设置会话团队标识。

Single-column UPDATE — atomic on all supported dialects.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识（类型 `str`）。The user id.
`session_id`	str	会话标识（类型 `str`）。The session id.
`team_id`	str \| None	团队标识（类型 `str \| None`）。The team id.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/storage/_sql/_storage.py:1262–1281

update_session_state

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

async

更新会话状态。

Read-modify-write on the payload; raises if absent.

参数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.

源码Source agentscope/app/storage/_sql/_storage.py:1283–1302

list_sessions

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

async

列出会话。

Sessions for a (user, agent) pair — newest first.

参数Parameters

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

返回Returns list[SessionRecord]

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

源码Source agentscope/app/storage/_sql/_storage.py:1304–1327

get_session

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

async

获取会话。

One session; owner-scoped.

参数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/_sql/_storage.py:1329–1341

delete_session

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

async

删除会话。

Atomic: the session row, its message log, and any team it leads are removed inside one transaction (see :meth:_delete_session_impl).

参数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 bool

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

源码Source agentscope/app/storage/_sql/_storage.py:1343–1363

list_sessions_by_schedule

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

async

列出会话通过调度。

Sessions created by schedule_id — newest first.

参数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/_sql/_storage.py:1365–1388

list_sessions_by_channel

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

async

列出会话通过渠道。

The channel id lives in the JSON payload rather than a column, and in two shapes: nested under source since that became a tagged union, flat for rows written before.

参数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/_sql/_storage.py:1390–1428

upsert_schedule

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

async

执行 upsert_schedule 方法。

Persist a schedule record (create or overwrite).

参数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/_sql/_storage.py:1434–1442

get_schedule

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

async

获取调度。

One schedule; owner-scoped.

参数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/_sql/_storage.py:1444–1454

list_schedules

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

async

用于用户的所有调度。

All schedules for a 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/_sql/_storage.py:1456–1475

delete_schedule

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

async

删除调度。

Atomic: the schedule and every session it spawned are removed in one transaction (see :meth:_delete_schedule_impl).

参数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/_sql/_storage.py:1477–1490

list_all_schedules

async def list_all_schedules() -> list[ScheduleRecord]

async

列出所有调度。

Every schedule across every user (used on startup).

返回Returns list[ScheduleRecord]

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

源码Source agentscope/app/storage/_sql/_storage.py:1492–1498

upsert_channel

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

async

执行 upsert_channel 方法。

Persist a channel record and its bot-uniqueness column.

参数Parameters

名称Name	类型Type	说明Description
`record`	ChannelRecord	记录（类型 `ChannelRecord`）。The channel record to store.
`platform_bot_id`	str	平台 bot 标识（类型 `str`）。The platform-side bot identifier, extracted from the credentials by the caller. Globally unique: a bot drives at most one channel.

返回Returns str

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

异常Raises

可能抛出 `Exception`。`ValueError`:
可能抛出 `Exception`。If another channel already drives this bot. Checked here
可能抛出 `Exception`。rather than left to the UNIQUE constraint, because
可能抛出 `Exception`。MySQL's `ON DUPLICATE KEY UPDATE` fires on *any*
可能抛出 `Exception`。unique-key conflict — the write would silently overwrite
可能抛出 `Exception`。the holder instead of failing, and the same call would
可能抛出 `Exception`。behave differently per dialect.
可能抛出 `Exception`。Neither this check nor `ChannelService.create`'s is
可能抛出 `Exception`。atomic with the write, so two creates racing for one bot
可能抛出 `Exception`。can both pass. The constraint then decides, and how it
可能抛出 `Exception`。surfaces depends on the dialect: Postgres and SQLite
可能抛出 `Exception`。raise `IntegrityError`, while on MySQL the duplicate-key
可能抛出 `Exception`。update lands on the holder's row and the read-back of
可能抛出 `Exception`。the new id then raises `NoResultFound`, rolling the
可能抛出 `Exception`。transaction back. Either way nothing is committed.

源码Source agentscope/app/storage/_sql/_storage.py:1508–1558

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	渠道标识。The channel id.

返回Returns ChannelRecord | None

返回ChannelRecord|None。`ChannelRecord | None`:

源码Source agentscope/app/storage/_sql/_storage.py:1560–1573

list_channels

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

async

列出渠道。

Return all channel records owned by user_id.

参数Parameters

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

返回Returns list[ChannelRecord]

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

源码Source agentscope/app/storage/_sql/_storage.py:1575–1600

list_all_channels

async def list_all_channels() -> list[ChannelRecord]

async

列出所有渠道。

Every channel across every user.

返回Returns list[ChannelRecord]

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

源码Source agentscope/app/storage/_sql/_storage.py:1602–1614

delete_channel

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

async

删除渠道。

Delete a channel record.

参数Parameters

名称Name	类型Type	说明Description
`channel_id`	str	渠道到删除的标识。The id of the channel to delete.
`platform_bot_id`	str	平台 bot 标识（类型 `str`）。Unused, and accepted only for parity with the Redis backend, which keeps the bot index in a separate key. Here the UNIQUE column goes with the row.

返回Returns bool

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

源码Source agentscope/app/storage/_sql/_storage.py:1616–1644

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 bound to a platform bot, if any.

参数Parameters

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

返回Returns str | None

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

源码Source agentscope/app/storage/_sql/_storage.py:1646–1669

upsert_message

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

async

执行 upsert_message 方法。

Insert-or-update by (session_id, msg_id).

参数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/_sql/_storage.py:1675–1704

delete_message

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

async

删除消息。

Delete one message by (session_id, 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/_sql/_storage.py:1706–1724

get_message

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

async

获取消息。

Fetch a message by (session_id, msg_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 Msg | None

返回消息|None。Returns msg none.

源码Source agentscope/app/storage/_sql/_storage.py:1726–1747

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

列出消息。

Mirrors :meth:RedisStorage.list_messages: returns up to limit messages in chronological order together with a has_more flag that is True when older messages exist before the returned page.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识（类型 `str`）。The user id.
`session_id`	str	会话标识（类型 `str`）。The session id.
`limit`默认default `50`	int	数量上限（类型 `int`）。The limit.
`before`默认default `None`	str \| None	之前（类型 `str \| None`）。The before.
`kwargs`	Any	关键字参数（类型 `Any`）。The kwargs.

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

返回列出[消息],bool列表。Returns a list of list msg bool.

源码Source agentscope/app/storage/_sql/_storage.py:1749–1835

upsert_team

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

async

执行 upsert_team 方法。

Persist a team record.

参数Parameters

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

返回Returns TeamRecord

返回TeamRecord。Returns team record.

源码Source agentscope/app/storage/_sql/_storage.py:1841–1849

get_team

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

async

获取团队。

One team; owner-scoped.

参数Parameters

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

返回Returns TeamRecord | None

返回TeamRecord|None。Returns team record none.

源码Source agentscope/app/storage/_sql/_storage.py:1851–1861

list_teams

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

async

列出 list_teams。

All teams for a user.

参数Parameters

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

返回Returns list[TeamRecord]

返回TeamRecord列表。Returns a list of team record.

源码Source agentscope/app/storage/_sql/_storage.py:1863–1877

delete_team

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

async

删除团队。

Atomic: the role-aware member teardown (created members deleted whole, invited members losing only their borrowed session), the leader detach, and the team-record delete all run in one transaction (see :meth:_delete_team_impl).

参数Parameters

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

返回Returns bool

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

源码Source agentscope/app/storage/_sql/_storage.py:1879–1890

upsert_knowledge_base

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

async

执行 upsert_knowledge_base 方法。

Create or update a KB record; enforces record.user_id.

参数Parameters

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

返回Returns KnowledgeBaseRecord

返回KnowledgeBaseRecord。Returns knowledge base record.

源码Source agentscope/app/storage/_sql/_storage.py:1896–1907

get_knowledge_base

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

async

获取知识库。

One KB; owner-scoped.

参数Parameters

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

返回Returns KnowledgeBaseRecord | None

返回KnowledgeBaseRecord|None。Returns knowledge base record none.

源码Source agentscope/app/storage/_sql/_storage.py:1909–1919

list_knowledge_bases

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

async

用于用户的所有 KBs。

All KBs for a user.

参数Parameters

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

返回Returns list[KnowledgeBaseRecord]

返回KnowledgeBaseRecord列表。Returns a list of knowledge base record.

源码Source agentscope/app/storage/_sql/_storage.py:1921–1940

delete_knowledge_base

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

async

删除知识库。

Delete a KB; its document rows cascade natively via the knowledge_documents.knowledge_base_id foreign key (ON DELETE CASCADE) — one atomic statement, no app loop.

参数Parameters

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

返回Returns bool

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

源码Source agentscope/app/storage/_sql/_storage.py:1942–1961

upsert_knowledge_document

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

async

执行 upsert_knowledge_document 方法。

Create or update a document record; enforces record.user_id.

参数Parameters

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

返回Returns KnowledgeDocumentRecord

返回KnowledgeDocumentRecord。Returns knowledge document record.

源码Source agentscope/app/storage/_sql/_storage.py:1967–1978

get_knowledge_document

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

async

获取知识库文档。

One document; owner + KB scoped.

参数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/_sql/_storage.py:1980–1995

list_knowledge_documents

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

async

列出知识库文档。

All documents inside a KB.

参数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/_sql/_storage.py:1997–2019

delete_knowledge_document

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

async

删除知识库文档。

Delete one document.

参数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/_sql/_storage.py:2021–2040

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

更新知识库文档状态。

status is a column so it can always be updated with a single UPDATE; error / chunk_count live inside :attr:~KnowledgeDocumentRecord.data (payload) and therefore require the classic read-modify-write when they change.

参数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/_sql/_storage.py:2042–2077

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 方法。

Conditional UPDATE — atomic on every supported dialect.

参数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/_sql/_storage.py:2079–2115

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 方法。

Conditional UPDATE constrained to the current holder.

参数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/_sql/_storage.py:2117–2145

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 方法。

Conditional UPDATE constrained to the current holder.

参数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/_sql/_storage.py:2147–2174

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。

Documents past their lease deadline, non-terminal, held.

参数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/_sql/_storage.py:2176–2200

list_knowledge_documents_pending_since

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

async

列出 list_knowledge_documents_pending_since。

Documents stuck in pending 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/_sql/_storage.py:2202–2223

继承成员Inherited Members

继承自Inherited from StorageBase

__aexit__