类class agentscope.app.rag

S3BlobStore

class S3BlobStore(BlobStoreBase)

源码Source agentscope/app/rag/blob_store/_s3.py:57–291

S 3 二进制对象存储，负责持久化与读取数据。位于 agentscope.app.rag 包，继承自 BlobStoreBase。核心成员包括 write_stream、open、delete、size、exists。

Store blobs in an S3-compatible bucket. Declared in agentscope.app.rag. Extends BlobStoreBase.

构造器摘要Constructor Summary

构造器Constructor	说明Description
`def __init__(bucket: str, *, region_name: str \| None = None, endpoint_url: str \| None = None, aws_access_key_id: str \|…`	初始化 `S3BlobStore` 实例。 Initialize an S3-compatible bucket.

方法摘要Method Summary

方法Method	说明Description
`async def __aenter__() -> 'S3BlobStore'`async	进入异步上下文并返回自身。 Implements a Python protocol method.
`async def __aexit__(exc_type: type[BaseException] \| None, exc_val: BaseException \| None, exc_tb: Any) -> None`async	退出异步上下文并清理资源。 Implements a Python protocol method.
`async def write_stream(key: str, stream: IO[bytes]) -> str`async	执行写入流相关操作。 Stream-write a blob and return its ``s3://{bucket}/{key}`` URI.
`async def open(uri: str) -> AsyncIterator[AsyncReadable]`asyncasynccontextmanager	设置打开。 Stream-read a blob by URI.
`async def delete(uri: str) -> None`async	删除数据。 Delete the object at ``uri``. Idempotent.
`async def size(uri: str) -> int \| None`async	设置大小。 Return the object's ``ContentLength``, or ``None`` if gone.
`async def exists(uri: str) -> bool`async	判断数据是否存在。 Return whether the object at ``uri`` is present.

构造器详情Constructor Detail

S3BlobStore

def __init__(bucket: str, *, region_name: str | None = None, endpoint_url: str | None = None, aws_access_key_id: str | None = None, aws_secret_access_key: str | None = None, session_token: str | None = None, use_ssl: bool = True, config: Any = None) -> None

初始化 S3BlobStore 实例。

Initialize an S3-compatible bucket.

参数Parameters

名称Name	类型Type	说明Description
`bucket`	str	存储桶（类型 `str`）。Bucket name. The bucket MUST exist; this store will not create it. Production deployments commonly provision the bucket out-of-band via Terraform / CloudFormation so the app's IAM role does not need `s3:CreateBucket`.
`region_name`kw默认default `None`	str \| None	区域名称（类型 `str \| None`）。AWS region for AWS S3. Required for AWS, ignored for MinIO / R2 / other services that locate the bucket via `endpoint_url`.
`endpoint_url`kw默认default `None`	str \| None	端点 URL（类型 `str \| None`）。Full URL of the S3-compatible service. `None` selects real AWS S3. Examples: `http://minio:9000`, `https:// .r2.cloudflarestorage.com`, `https://oss-cn-hangzhou.aliyuncs.com`.
`aws_access_key_id`kw默认default `None`	str \| None	aws 访问键标识（类型 `str \| None`）。Static credential. Prefer leaving this `None` and letting the IAM role / environment chain resolve credentials when running on AWS infrastructure.
`aws_secret_access_key`kw默认default `None`	str \| None	aws 密钥访问键（类型 `str \| None`）。Paired with `aws_access_key_id`.
`session_token`kw默认default `None`	str \| None	会话令牌（类型 `str \| None`）。For STS-issued temporary credentials.
`use_ssl`kw默认default `True`	bool	使用 ssl（类型 `bool`）。Force HTTPS. Production deployments must keep this on; local MinIO with self-signed certs is the only place `False` is reasonable.
`config`kw默认default `None`	Any	配置对象（类型 `Any`）。`aiobotocore.config.AioConfig` instance for users who need to tune timeouts, retry mode, signature version (e.g. `s3v4` for Aliyun OSS), or addressing style. Path-style addressing is needed for MinIO; pass `AioConfig(s3={"addressing_style": "path"})`.

源码Source agentscope/app/rag/blob_store/_s3.py:60–134

方法详情Method Detail

aenter

async def __aenter__() -> 'S3BlobStore'

async

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

Implements a Python protocol method.

返回Returns 'S3BlobStore'

返回'S3BlobStore'。Returns s 3 blob store.

源码Source agentscope/app/rag/blob_store/_s3.py:136–147

aexit

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

async

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

Implements a Python protocol method.

参数Parameters

名称Name	类型Type	说明Description
`exc_type`	type[BaseException] \| None	exc 类型（类型 `type[BaseException] \| None`）。The exc type.
`exc_val`	BaseException \| None	exc val（类型 `BaseException \| None`）。The exc val.
`exc_tb`	Any	exc tb（类型 `Any`）。The exc tb.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/rag/blob_store/_s3.py:149–155

write_stream

async def write_stream(key: str, stream: IO[bytes]) -> str

async

执行写入流相关操作。

Uses upload_fileobj so aioboto3 picks multipart upload automatically for bodies above the multipart threshold (8 MiB by default in botocore).

参数Parameters

名称Name	类型Type	说明Description
`key`	str	键（类型 `str`）。The key.
`stream`	IO[bytes]	流（类型 `IO[bytes]`）。The stream.

返回Returns str

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

源码Source agentscope/app/rag/blob_store/_s3.py:202–212

open

async def open(uri: str) -> AsyncIterator[AsyncReadable]

asyncasynccontextmanager

设置打开。

Yields an :class:AsyncReadable backed by the response Body.

参数Parameters

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

返回Returns AsyncIterator[AsyncReadable]

以异步流的形式返回AsyncReadable。Returns a stream of async readable.

源码Source agentscope/app/rag/blob_store/_s3.py:215–246

delete

async def delete(uri: str) -> None

async

删除数据。

Delete the object at uri.

参数Parameters

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

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/rag/blob_store/_s3.py:248–254

size

async def size(uri: str) -> int | None

async

设置大小。

One HEAD — the same call :meth:exists makes, so a caller that needs both can ask for the size and treat a number as proof of existence.

参数Parameters

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

返回Returns int | None

返回int|None。Returns int none.

源码Source agentscope/app/rag/blob_store/_s3.py:256–269

exists

async def exists(uri: str) -> bool

async

判断数据是否存在。

Return whether the object at uri is present.

参数Parameters

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

返回Returns bool

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

源码Source agentscope/app/rag/blob_store/_s3.py:271–291